Strategy based on Dow Theory principles for trend detection.
Timeframe
5m
Direction
Long Only
Stoploss
-10.0%
Trailing Stop
Yes
ROI
0m: 5.0%, 240m: 3.0%, 1440m: 1.0%
Interface Version
3
Startup Candles
N/A
Indicators
5
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement
# flake8: noqa: F401
# isort: skip_file
# --- Do not remove these imports ---
import numpy as np
import pandas as pd
from datetime import datetime, timedelta, timezone
from pandas import DataFrame
from typing import Dict, Optional, Union, Tuple
from freqtrade.strategy import (
IStrategy,
Trade,
Order,
PairLocks,
informative,
BooleanParameter,
CategoricalParameter,
DecimalParameter,
IntParameter,
RealParameter,
timeframe_to_minutes,
timeframe_to_next_date,
timeframe_to_prev_date,
merge_informative_pair,
stoploss_from_absolute,
stoploss_from_open,
)
# --------------------------------
# Add your lib to import here
import talib.abstract as ta
import pandas_ta as pta
from technical import qtpylib
class DowTheoryStrategy(IStrategy):
"""
Strategy based on Dow Theory principles for trend detection.
Key Dow Theory principles implemented:
1. Primary trends identification using higher timeframe
2. Secondary trend corrections using mid timeframe
3. Confirmation through volume
4. Trend continuation until valid reversal signals
The strategy uses multiple timeframes to align with Dow Theory's
concept of primary, secondary, and minor trends.
"""
INTERFACE_VERSION = 3
# Optimal timeframe for the strategy - using 15m as primary trend
timeframe = "5m"
# Can this strategy go short?
can_short: bool = True
# Minimal ROI designed for the strategy
minimal_roi = {
"0": 0.05, # 5% profit is enough to exit
"240": 0.03, # After 240 minutes, exit at 3% profit
"1440": 0.01, # After 24 hours, exit at 1% profit
}
# Optimal stoploss designed for the strategy
stoploss = -0.1
# Trailing stoploss to lock in profits as trend continues
trailing_stop = True
trailing_stop_positive = 0.01
trailing_stop_positive_offset = 0.02
trailing_only_offset_is_reached = True
# Run "populate_indicators()" only for new candle
process_only_new_candles = True
# These values can be overridden in the config
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = False
# Number of candles the strategy requires before producing valid signals
startup_candle_count: int = 50
# Parameters for tuning
trend_strength = IntParameter(20, 50, default=25, space="buy")
volume_threshold = DecimalParameter(1.0, 3.0, default=1.5, space="buy")
reversal_threshold = IntParameter(10, 30, default=15, space="sell")
def informative_pairs(self):
"""
Define additional, informative pair/interval combinations to be cached from the exchange.
We need higher timeframes for primary trend detection and secondary trend confirmations.
"""
pairs = self.dp.current_whitelist()
informative_pairs = []
# 15m timeframe for primary trend detection
for pair in pairs:
informative_pairs.append((pair, "15m"))
# 1h timeframe for major trend confirmation
for pair in pairs:
informative_pairs.append((pair, "1h"))
return informative_pairs
@informative('15m')
def populate_informative_15m(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate indicators for primary trend identification on 15m timeframe
"""
# Calculate primary trend indicators
dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50)
dataframe['ema200'] = ta.EMA(dataframe, timeperiod=200)
# Primary trend determination
dataframe['uptrend_15m'] = (dataframe['ema50'] > dataframe['ema200']) & (dataframe['close'] > dataframe['ema50'])
dataframe['downtrend_15m'] = (dataframe['ema50'] < dataframe['ema200']) & (dataframe['close'] < dataframe['ema50'])
# Relative strength
dataframe['rsi_15m'] = ta.RSI(dataframe, timeperiod=14)
return dataframe
@informative('1h')
def populate_informative_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate indicators for major trend confirmation on 1h timeframe
"""
# Calculate major trend indicators
dataframe['sma50'] = ta.SMA(dataframe, timeperiod=50)
dataframe['sma200'] = ta.SMA(dataframe, timeperiod=200)
# Major trend determination (longer term)
dataframe['major_uptrend'] = dataframe['sma50'] > dataframe['sma200']
dataframe['major_downtrend'] = dataframe['sma50'] < dataframe['sma200']
return dataframe
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Adds indicators for secondary trends and generates buy/sell signals
"""
# Secondary trend indicators
dataframe['ema20'] = ta.EMA(dataframe, timeperiod=20)
dataframe['ema50'] = ta.EMA(dataframe, timeperiod=50)
dataframe['ema100'] = ta.EMA(dataframe, timeperiod=100)
# Momentum and volume indicators
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
# Volume confirmation
dataframe['volume_mean'] = dataframe['volume'].rolling(10).mean()
dataframe['volume_increased'] = dataframe['volume'] > (dataframe['volume_mean'] * self.volume_threshold.value)
# Detect higher highs and higher lows for uptrend
dataframe['higher_high'] = (
(dataframe['high'] > dataframe['high'].shift(1)) &
(dataframe['high'].shift(1) > dataframe['high'].shift(2))
)
dataframe['higher_low'] = (
(dataframe['low'] > dataframe['low'].shift(1)) &
(dataframe['low'].shift(1) > dataframe['low'].shift(2))
)
# Detect lower highs and lower lows for downtrend
dataframe['lower_high'] = (
(dataframe['high'] < dataframe['high'].shift(1)) &
(dataframe['high'].shift(1) < dataframe['high'].shift(2))
)
dataframe['lower_low'] = (
(dataframe['low'] < dataframe['low'].shift(1)) &
(dataframe['low'].shift(1) < dataframe['low'].shift(2))
)
# Identify peak and trough
dataframe['peak'] = ta.MAX(dataframe['high'], timeperiod=10) # Local peak
dataframe['trough'] = ta.MIN(dataframe['low'], timeperiod=10) # Local trough
# Donchian Channels peak and trough detection (using 10-period window)
donchian_upper = dataframe['high'].rolling(window=10).max()
donchian_lower = dataframe['low'].rolling(window=10).min()
dataframe['donchian_peak'] = dataframe['high'] >= donchian_upper
dataframe['donchian_trough'] = dataframe['low'] <= donchian_lower
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on Dow Theory principles, identify entry signals
"""
# Default values
dataframe['enter_long'] = 0
dataframe['enter_short'] = 0
# LONG entries - Primary uptrend confirmed by secondary trend bounce
dataframe.loc[
(
# Primary trend is up (from 15m timeframe)
dataframe['uptrend_15m_15m'] &
# Major trend is up (from 1h timeframe)
dataframe['major_uptrend_1h'] &
# Secondary trend indicators
(dataframe['ema20'] > dataframe['ema50']) &
# Price bounced off support (secondary correction ended)
(dataframe['low'].shift(1) < dataframe['ema50']) &
(dataframe['close'] > dataframe['ema50']) &
# Volume confirms the movement
dataframe['volume_increased'] &
# Pattern of higher lows (accumulation)
dataframe['higher_low'] &
# RSI showing strength but not overbought
(dataframe['rsi'] > self.trend_strength.value) &
(dataframe['rsi'] < 70) &
(dataframe['volume'] > 0) & # Ensure volume is not 0
# Higher peak and higher trough condition
(dataframe['peak'] > dataframe['peak'].shift(11)) &
(dataframe['trough'] > dataframe['trough'].shift(11))
),
'enter_long'] = 1
# SHORT entries - Primary downtrend confirmed by secondary trend bounce
if self.can_short:
dataframe.loc[
(
# Primary trend is down (from 15m timeframe)
dataframe['downtrend_15m_15m'] &
# Major trend is down (from 1h timeframe)
dataframe['major_downtrend_1h'] &
# Secondary trend indicators
(dataframe['ema20'] < dataframe['ema50']) &
# Price bounced off resistance (secondary correction ended)
(dataframe['high'].shift(1) > dataframe['ema50']) &
(dataframe['close'] < dataframe['ema50']) &
# Volume confirms the movement
dataframe['volume_increased'] &
# Pattern of lower highs (distribution)
dataframe['lower_high'] &
# RSI showing weakness but not oversold
(dataframe['rsi'] < (100 - self.trend_strength.value)) &
(dataframe['rsi'] > 30) &
(dataframe['volume'] > 0) # Ensure volume is not 0
# Add conditions based on peak and trough
& (dataframe['peak'] < dataframe['peak'].shift(11)) &
(dataframe['trough'] < dataframe['trough'].shift(11))
),
'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Based on Dow Theory principles, identify exit signals when trends reverse
"""
# Default values
dataframe['exit_long'] = 0
dataframe['exit_short'] = 0
# Exit LONG positions
dataframe.loc[
(
# Reversal signals in primary trend
(dataframe['downtrend_15m_15m']) |
# RSI overbought
(dataframe['rsi'] > 80) |
# Trend reversal on high volume
(
dataframe['lower_high'] &
dataframe['lower_low'] &
dataframe['volume_increased']
) |
# Break below key support
(dataframe['close'] < dataframe['trough'].shift(1)) |
# RSI divergence (falling RSI while price rising)
(
(dataframe['close'] > dataframe['close'].shift(3)) &
(dataframe['rsi'] < dataframe['rsi'].shift(3)) &
(dataframe['rsi'] < self.reversal_threshold.value)
)
),
'exit_long'] = 1
# Exit SHORT positions
if self.can_short:
dataframe.loc[
(
# Reversal signals in primary trend
(dataframe['uptrend_15m_15m']) |
# RSI oversold
(dataframe['rsi'] < 20) |
# Trend reversal on high volume
(
dataframe['higher_high'] &
dataframe['higher_low'] &
dataframe['volume_increased']
) |
# Break above key resistance
(dataframe['close'] > dataframe['peak'].shift(1)) |
# RSI divergence (rising RSI while price falling)
(
(dataframe['close'] < dataframe['close'].shift(3)) &
(dataframe['rsi'] > dataframe['rsi'].shift(3)) &
(dataframe['rsi'] > (100 - self.reversal_threshold.value))
)
),
'exit_short'] = 1
return dataframe
def leverage(self, pair: str, current_time: datetime, current_rate: float,
proposed_leverage: float, max_leverage: float, side: str,
**kwargs) -> float:
"""
Limit leverage to reasonable values when using margin/futures
"""
# Conservative leverage for Dow Theory - trend following is already strong
return min(2.0, max_leverage)