DeepAlphaStrategy

"""
DeepAlpha Example Strategy for Freqtrade
=========================================

A minimal Freqtrade strategy demonstrating how to use the DeepAlpha
FreqAI model. This strategy:

- Defines technical indicator features for FreqAI
- Uses Triple Barrier labels generated by the DeepAlpha model
- Filters entries using meta-labeling confidence scores
- Implements basic risk management via stop-loss and ROI

Usage:
    freqtrade trade --strategy DeepAlphaStrategy --freqaimodel DeepAlphaModel

Author: DeepAlpha Team
License: MIT
"""

from __future__ import annotations

import logging
from typing import Dict

import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame

from freqtrade.strategy import IStrategy, merge_informative_pair
from freqtrade.strategy.parameters import IntParameter, DecimalParameter

logger = logging.getLogger(__name__)


class DeepAlphaStrategy(IStrategy):
    """
    Freqtrade strategy powered by DeepAlpha's ML pipeline.

    The strategy delegates all signal generation to the FreqAI model.
    Features are defined in `feature_engineering_*` methods and passed
    to the DeepAlpha model, which returns predictions and a confidence
    mask from meta-labeling.
    """

    # ---------------------------------------------------------------
    # Strategy configuration
    # ---------------------------------------------------------------

    INTERFACE_VERSION = 3

    # Minimal ROI table tuned for 1h bars — less aggressive take-profit
    # so winners have room to compound over multiple bars.
    minimal_roi = {
        "0": 0.06,
        "240": 0.04,
        "480": 0.025,
        "1440": 0.015,
    }

    # 1h timeframe: reduces intraday noise that dominated 5m tests and
    # lets Triple Barrier capture real directional moves.
    stoploss = -0.04
    trailing_stop = False
    trailing_stop_positive = 0.015
    trailing_stop_positive_offset = 0.03

    timeframe = "1h"

    # FreqAI configuration
    freqai_info = {
        "enabled": True,
    }

    # Strategy parameters (optimizable).
    # SHORT threshold higher than LONG (0.60 vs 0.55) to compensate for
    # the residual downside skew of Triple Barrier on volatile intraday data.
    buy_prediction_threshold = DecimalParameter(
        0.5, 0.8, default=0.55, decimals=2, space="buy",
        optimize=True, load=True,
    )
    sell_prediction_threshold = DecimalParameter(
        0.5, 0.8, default=0.60, decimals=2, space="sell",
        optimize=True, load=True,
    )
    meta_confidence_threshold = DecimalParameter(
        0.5, 0.8, default=0.55, decimals=2, space="buy",
        optimize=True, load=True,
    )
    # Confidence level that unlocks counter-trend entries (e.g. take a
    # short while EMA24 > EMA96 only if P_short exceeds this).
    counter_trend_override = DecimalParameter(
        0.6, 0.9, default=0.70, decimals=2, space="buy",
        optimize=True, load=True,
    )

    # ---------------------------------------------------------------
    # Feature engineering
    # ---------------------------------------------------------------

    def feature_engineering_expand_all(
        self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        Create features that will be expanded across multiple periods.

        FreqAI automatically generates these features for each period
        defined in the config (e.g., 10, 20, 50 candles).
        """
        # Momentum indicators
        dataframe[f"%-rsi-period_{period}"] = ta.RSI(dataframe, timeperiod=period)
        dataframe[f"%-mfi-period_{period}"] = ta.MFI(dataframe, timeperiod=period)
        dataframe[f"%-adx-period_{period}"] = ta.ADX(dataframe, timeperiod=period)
        dataframe[f"%-cci-period_{period}"] = ta.CCI(dataframe, timeperiod=period)

        # Volatility
        bollinger = ta.BBANDS(dataframe, timeperiod=period, nbdevup=2.0, nbdevdn=2.0)
        dataframe[f"%-bb-width-period_{period}"] = (
            (bollinger["upperband"] - bollinger["lowerband"]) / bollinger["middleband"]
        )
        dataframe[f"%-bb-pct-period_{period}"] = (
            (dataframe["close"] - bollinger["lowerband"])
            / (bollinger["upperband"] - bollinger["lowerband"] + 1e-10)
        )
        dataframe[f"%-atr-period_{period}"] = ta.ATR(dataframe, timeperiod=period)

        # Trend
        dataframe[f"%-sma-period_{period}"] = ta.SMA(dataframe, timeperiod=period)
        dataframe[f"%-ema-period_{period}"] = ta.EMA(dataframe, timeperiod=period)
        dataframe[f"%-close-sma-ratio_{period}"] = (
            dataframe["close"] / (dataframe[f"%-sma-period_{period}"] + 1e-10)
        )

        # Volume
        dataframe[f"%-volume-sma-ratio_{period}"] = (
            dataframe["volume"]
            / (dataframe["volume"].rolling(period).mean() + 1e-10)
        )

        return dataframe

    def feature_engineering_expand_basic(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        Create features that are not expanded across multiple periods.
        """
        # MACD
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe["%-macd"] = macd["macd"]
        dataframe["%-macd-signal"] = macd["macdsignal"]
        dataframe["%-macd-hist"] = macd["macdhist"]

        # Stochastic
        stoch = ta.STOCH(dataframe, fastk_period=14, slowk_period=3, slowd_period=3)
        dataframe["%-slowk"] = stoch["slowk"]
        dataframe["%-slowd"] = stoch["slowd"]

        # Ichimoku
        dataframe["%-tenkan"] = (
            dataframe["high"].rolling(9).max() + dataframe["low"].rolling(9).min()
        ) / 2
        dataframe["%-kijun"] = (
            dataframe["high"].rolling(26).max() + dataframe["low"].rolling(26).min()
        ) / 2

        # Price-based features
        dataframe["%-pct-change"] = dataframe["close"].pct_change()
        dataframe["%-log-return"] = np.log(
            dataframe["close"] / dataframe["close"].shift(1)
        )

        # Candle patterns
        dataframe["%-body-ratio"] = (
            abs(dataframe["close"] - dataframe["open"])
            / (dataframe["high"] - dataframe["low"] + 1e-10)
        )
        dataframe["%-upper-shadow"] = (
            dataframe["high"]
            - dataframe[["close", "open"]].max(axis=1)
        ) / (dataframe["high"] - dataframe["low"] + 1e-10)

        # --- Sequential / memory features (poor-man's LSTM) --------------
        # Rolling log-return statistics over multiple horizons give the
        # tree model temporal context without needing a recurrent network.
        log_ret = dataframe["%-log-return"]
        for win in (4, 12, 24, 48):
            dataframe[f"%-logret-mean-{win}"] = log_ret.rolling(win).mean()
            dataframe[f"%-logret-std-{win}"] = log_ret.rolling(win).std()
            dataframe[f"%-logret-sum-{win}"] = log_ret.rolling(win).sum()
            dataframe[f"%-logret-skew-{win}"] = log_ret.rolling(win).skew()

        # Return z-score: how extreme is the current bar's move vs recent history
        roll_std_24 = log_ret.rolling(24).std()
        dataframe["%-logret-zscore-24"] = log_ret / (roll_std_24 + 1e-10)

        # Volatility-of-volatility: regime-change signal
        dataframe["%-vol-of-vol-48"] = (
            roll_std_24.rolling(48).std() / (roll_std_24.rolling(48).mean() + 1e-10)
        )

        # Momentum-of-momentum: acceleration features
        for lag in (1, 3, 6, 12):
            dataframe[f"%-close-lag-{lag}"] = (
                dataframe["close"] / dataframe["close"].shift(lag) - 1
            )
            dataframe[f"%-vol-lag-{lag}"] = (
                dataframe["volume"] / (dataframe["volume"].shift(lag) + 1e-10)
            )

        # High-Low range momentum
        hl_range = (dataframe["high"] - dataframe["low"]) / dataframe["close"]
        dataframe["%-hl-range-ratio-24"] = hl_range / (hl_range.rolling(24).mean() + 1e-10)

        return dataframe

    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        """
        Features computed after all expansion.  Also defines the target
        label column that the model will train on.
        """
        # Day-of-week and hour features
        dataframe["%-day-of-week"] = dataframe["date"].dt.dayofweek
        dataframe["%-hour"] = dataframe["date"].dt.hour

        # NOTE: Do NOT define a "&-label" column here. Target generation is
        # handled exclusively in set_freqai_targets() with Triple Barrier
        # labels. If both "&-label" and "&-target" exist, FreqAI's
        # BaseClassifierModel.fit uses the first one alphabetically
        # ("&-label"), which would bypass our Triple Barrier labels.

        return dataframe

    def set_freqai_targets(self, dataframe: DataFrame, metadata: Dict, **kwargs) -> DataFrame:
        """
        Generate regime-aware Triple Barrier labels.

        Class mapping (3-class, LightGBM-friendly non-negative ints):
            0 = SHORT (-1), 1 = FLAT (0), 2 = LONG (+1)

        The previous symmetric (1.5 / 1.5 ATR) version produced too many
        SHORT labels in bull markets because intraday pullbacks hit the
        lower barrier first even when the overall drift was positive.
        This version detects the local regime via EMA24 vs EMA96 on the
        training timeframe and adapts the barrier multipliers:

          * BULL    (EMA24 > EMA96 * 1.002): pt=1.2, sl=2.0 (favour LONG)
          * BEAR    (EMA24 < EMA96 * 0.998): pt=2.0, sl=1.2 (favour SHORT)
          * SIDEWAYS: pt=1.5, sl=1.5 (symmetric)

        Horizon raised 12 -> 24 candles (2h on 5m) to let directional
        moves play out before the time-barrier fallback triggers.
        """
        import numpy as np
        closes = dataframe["close"].values
        highs = dataframe["high"].values
        lows = dataframe["low"].values

        # Regime detection via EMA slope
        close_s = pd.Series(closes)
        ema_fast = close_s.ewm(span=24, adjust=False).mean().values
        ema_slow = close_s.ewm(span=96, adjust=False).mean().values
        regime = np.zeros(len(closes), dtype=int)
        regime[ema_fast > ema_slow * 1.002] = 1   # BULL
        regime[ema_fast < ema_slow * 0.998] = -1  # BEAR

        # ATR for adaptive barrier sizing
        atr_period = 14
        tr = np.maximum.reduce([
            highs - lows,
            np.abs(highs - np.roll(closes, 1)),
            np.abs(lows - np.roll(closes, 1))
        ])
        atr = np.zeros(len(closes))
        if len(tr) >= atr_period:
            atr[atr_period:] = np.convolve(
                tr, np.ones(atr_period) / atr_period, mode='valid'
            )[:len(closes) - atr_period]

        horizon = 12  # 12 hours on a 1h timeframe
        labels = np.zeros(len(closes), dtype=int)
        for i in range(len(closes) - horizon):
            if atr[i] <= 0:
                continue
            entry = closes[i]
            atr_pct = atr[i] / entry if entry > 0 else 0
            if atr_pct == 0:
                continue

            # Regime-aware asymmetric barriers
            if regime[i] == 1:
                pt_mult, sl_mult = 1.2, 2.0
            elif regime[i] == -1:
                pt_mult, sl_mult = 2.0, 1.2
            else:
                pt_mult, sl_mult = 1.5, 1.5

            pt_long = entry * (1 + pt_mult * atr_pct)
            sl_long = entry * (1 - sl_mult * atr_pct)
            end = min(i + horizon, len(closes) - 1)

            hit = 0
            for j in range(i + 1, end + 1):
                if highs[j] >= pt_long:
                    hit = 1  # LONG wins
                    break
                if lows[j] <= sl_long:
                    hit = -1  # SHORT wins
                    break
            if hit == 0:
                # Time barrier: use sign of return vs regime-aware dead-zone
                ret = (closes[end] - entry) / entry
                if ret > atr_pct * 0.3:
                    hit = 1
                elif ret < -atr_pct * 0.3:
                    hit = -1
            labels[i] = hit

        target_3class = (labels + 1).astype(int)
        dataframe["&-target"] = target_3class
        return dataframe

    # ---------------------------------------------------------------
    # Signal generation
    # ---------------------------------------------------------------

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        FreqAI handles indicator generation via feature_engineering methods.
        This method is intentionally minimal.
        """
        # FreqAI populates predictions automatically
        dataframe = self.freqai.start(dataframe, metadata, self)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Generate entry signals from DeepAlpha 3-class model predictions,
        gated by a same-timeframe EMA regime filter.

        Class mapping (LightGBM multi-class, non-negative ints):
          "0" -> P(SHORT)
          "1" -> P(FLAT)
          "2" -> P(LONG)

        Entry rules:
          * LONG  if P_long  > long_thr  AND P_long  > P_short
                  AND (uptrend OR P_long  > counter_trend_override)
          * SHORT if P_short > short_thr AND P_short > P_long
                  AND (downtrend OR P_short > counter_trend_override)
        """
        n = len(dataframe)
        p_short = dataframe["0"] if "0" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
        p_long = dataframe["2"] if "2" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
        do_predict = (
            dataframe["do_predict"] if "do_predict" in dataframe.columns
            else pd.Series([1] * n, index=dataframe.index)
        )

        # Regime filter on the trading timeframe (EMA24 vs EMA96 on 5m)
        ema_fast = dataframe["close"].ewm(span=24, adjust=False).mean()
        ema_slow = dataframe["close"].ewm(span=96, adjust=False).mean()
        uptrend = ema_fast > ema_slow
        downtrend = ema_fast < ema_slow

        long_thr = self.buy_prediction_threshold.value
        short_thr = self.sell_prediction_threshold.value
        override = self.counter_trend_override.value

        # Hard regime gate: never counter-trend. Counter-trend shorts in
        # bull markets produced a 3% stop-loss carnage in our Q1 2024
        # tests (112 stop-losses = -36.6% total). The override was too
        # permissive even at 0.70 confidence. Force alignment with
        # same-timeframe EMA regime.
        long_entry = (
            (p_long > long_thr)
            & (p_long > p_short)
            & (do_predict == 1)
            & uptrend
        )
        short_entry = (
            (p_short > short_thr)
            & (p_short > p_long)
            & (do_predict == 1)
            & downtrend
        )
        _ = override  # kept for hyperopt compatibility; currently unused

        dataframe.loc[long_entry, "enter_long"] = 1
        dataframe.loc[long_entry, "enter_tag"] = "DeepAlpha-LONG"
        dataframe.loc[short_entry, "enter_short"] = 1
        dataframe.loc[short_entry, "enter_tag"] = "DeepAlpha-SHORT"

        return dataframe

    # ---------------------------------------------------------------
    # Dynamic position sizing based on prediction confidence
    # ---------------------------------------------------------------

    def custom_stake_amount(
        self,
        pair: str,
        current_time,
        current_rate: float,
        proposed_stake: float,
        min_stake,
        max_stake: float,
        leverage: float,
        entry_tag,
        side: str,
        **kwargs,
    ) -> float:
        """
        Scale the stake linearly with the winning-class probability.

        A P_max of 0.55 (just above the gate) uses 50% of the default
        stake; a P_max of 0.80+ uses 150%. This concentrates risk on
        high-confidence signals and reduces bleed on marginal calls.
        """
        try:
            df, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe)
            if df is None or df.empty:
                return proposed_stake
            last = df.iloc[-1]
            p_short = float(last.get("0", 0.0) or 0.0)
            p_long = float(last.get("2", 0.0) or 0.0)
            p_max = p_long if side == "long" else p_short
            # Linear scaling: 0.55 -> 0.5x, 0.80 -> 1.5x (clipped)
            scale = 0.5 + (p_max - 0.55) * 4.0
            scale = max(0.25, min(scale, 1.75))
            stake = proposed_stake * scale
            if min_stake is not None:
                stake = max(stake, float(min_stake))
            stake = min(stake, float(max_stake))
            return stake
        except Exception as e:
            logger.warning("custom_stake_amount fallback (%s): %s", pair, e)
            return proposed_stake

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        3-class exits: close longs when P_short dominates, close shorts
        when P_long dominates. ROI / stoploss / trailing stop remain the
        primary risk-management layer.
        """
        n = len(dataframe)
        p_short = dataframe["0"] if "0" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)
        p_long = dataframe["2"] if "2" in dataframe.columns else pd.Series([0.0] * n, index=dataframe.index)

        dataframe.loc[(p_short > 0.50) & (p_short > p_long), "exit_long"] = 1
        dataframe.loc[(p_long > 0.50) & (p_long > p_short), "exit_short"] = 1

        return dataframe