DowTheoryStrategy

# 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)