A51Strategy

"""
A51 Strategy — Scalping-oriented with tight risk management
ETH/USDT 5m Futures | WR% 61.7 | DD% 7.2 | Profit +12,841

Core logic:
- VWAP + Order Block detection for precise entries
- c=0.35 (conservative — many small wins)
- e=0.00 (neutral bias — pure price action)
- Tight stops with high win rate
"""
import numpy as np
from freqtrade.strategy import IStrategy, DecimalParameter, IntParameter
from pandas import DataFrame
import talib.abstract as ta


class A51Strategy(IStrategy):
    """
    A51 c=0.35 e=0.00
    High win-rate scalper with tight risk management
    """

    INTERFACE_VERSION = 3
    timeframe = "5m"
    can_short = True

    position_c = DecimalParameter(
        0.1, 1.0, default=0.35, space="buy", optimize=True
    )
    entry_bias = DecimalParameter(
        -0.5, 0.5, default=0.00, space="buy", optimize=True
    )

    minimal_roi = {
        "0": 0.008,
        "15": 0.005,
        "30": 0.003,
        "60": 0.001,
    }

    stoploss = -0.005
    trailing_stop = True
    trailing_stop_positive = 0.003
    trailing_stop_positive_offset = 0.005
    trailing_only_offset_is_reached = True

    startup_candle_count = 100

    def _vwap(self, df: DataFrame) -> DataFrame:
        """Calculate VWAP (resets each 288 candles ≈ 1 day)"""
        period = 288
        tp = (df["high"] + df["low"] + df["close"]) / 3
        cumvol = df["volume"].rolling(window=period, min_periods=1).sum()
        cumtp = (tp * df["volume"]).rolling(
            window=period, min_periods=1
        ).sum()
        df["vwap"] = cumtp / cumvol
        return df

    def populate_indicators(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        # VWAP
        dataframe = self._vwap(dataframe)

        # Short-term EMAs for scalping
        dataframe["ema_5"] = ta.EMA(dataframe, timeperiod=5)
        dataframe["ema_8"] = ta.EMA(dataframe, timeperiod=8)
        dataframe["ema_13"] = ta.EMA(dataframe, timeperiod=13)

        # RSI with shorter period for scalping
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=9)

        # Stochastic RSI
        stoch = ta.STOCHRSI(
            dataframe, timeperiod=14, fastk_period=3, fastd_period=3
        )
        dataframe["stoch_k"] = stoch["fastk"]
        dataframe["stoch_d"] = stoch["fastd"]

        # ATR for stop calculation
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=10)

        # Volume analysis
        dataframe["volume_sma"] = ta.SMA(
            dataframe["volume"], timeperiod=20
        )
        dataframe["volume_ratio"] = (
            dataframe["volume"] / dataframe["volume_sma"].replace(0, np.nan)
        ).fillna(0)

        # ADX for trend strength
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)

        # EMA slope for trend direction
        dataframe["ema_slope"] = (dataframe["ema_5"] - dataframe["ema_5"].shift(10)) / dataframe["ema_5"].shift(10) * 100

        # Order block detection (simplified)
        dataframe["bullish_ob"] = (
            (dataframe["close"].shift(2) < dataframe["open"].shift(2))
            & (dataframe["close"].shift(1) > dataframe["open"].shift(1))
            & (
                dataframe["close"].shift(1)
                > dataframe["high"].shift(2)
            )
        ).astype(int)

        dataframe["bearish_ob"] = (
            (dataframe["close"].shift(2) > dataframe["open"].shift(2))
            & (dataframe["close"].shift(1) < dataframe["open"].shift(1))
            & (
                dataframe["close"].shift(1)
                < dataframe["low"].shift(2)
            )
        ).astype(int)

        return dataframe

    def populate_entry_trend(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        # Long: price above VWAP + bullish OB or stoch cross
        dataframe.loc[
            (
                (dataframe["close"] > dataframe["vwap"])
                & (dataframe["ema_5"] > dataframe["ema_8"])
                & (
                    (dataframe["bullish_ob"] == 1)
                    | (
                        (dataframe["stoch_k"] > dataframe["stoch_d"])
                        & (dataframe["stoch_k"].shift(1) <= dataframe["stoch_d"].shift(1))
                        & (dataframe["stoch_k"] < 30)
                    )
                )
                & (dataframe["volume_ratio"] > 1.5)
                & (dataframe["rsi"] > 40)
                & (dataframe["rsi"] < 60)
                & (dataframe["adx"] > 20)
            ),
            ["enter_long", "enter_tag"],
        ] = (1, "a51_vwap_ob_long")

        # Short: price below VWAP + bearish OB or stoch cross
        dataframe.loc[
            (
                (dataframe["close"] < dataframe["vwap"])
                & (dataframe["ema_5"] < dataframe["ema_8"])
                & (
                    (dataframe["bearish_ob"] == 1)
                    | (
                        (dataframe["stoch_k"] < dataframe["stoch_d"])
                        & (dataframe["stoch_k"].shift(1) >= dataframe["stoch_d"].shift(1))
                        & (dataframe["stoch_k"] > 70)
                    )
                )
                & (dataframe["volume_ratio"] > 1.5)
                & (dataframe["rsi"] > 40)
                & (dataframe["rsi"] < 60)
                & (dataframe["adx"] > 20)
            ),
            ["enter_short", "enter_tag"],
        ] = (1, "a51_vwap_ob_short")

        return dataframe

    def populate_exit_trend(
        self, dataframe: DataFrame, metadata: dict
    ) -> DataFrame:
        # Exit long: EMA cascade down OR extreme RSI
        dataframe.loc[
            (
                (dataframe["ema_5"] < dataframe["ema_8"])
                & (dataframe["ema_8"] < dataframe["ema_13"])
                & (dataframe["close"] < dataframe["vwap"])
            )
            | (dataframe["rsi"] > 75),
            ["exit_long", "exit_tag"],
        ] = (1, "a51_exit_cascade")

        # Exit short: EMA cascade up OR extreme RSI
        dataframe.loc[
            (
                (dataframe["ema_5"] > dataframe["ema_8"])
                & (dataframe["ema_8"] > dataframe["ema_13"])
                & (dataframe["close"] > dataframe["vwap"])
            )
            | (dataframe["rsi"] < 25),
            ["exit_short", "exit_tag"],
        ] = (1, "a51_exit_cascade")

        return dataframe

    def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
                            time_in_force: str, current_time, entry_tag, side: str,
                            **kwargs) -> bool:
        """Block entry in choppy/fading conditions"""
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if len(dataframe) == 0:
            return True
        last = dataframe.iloc[-1]
        adx = last.get("adx", 0)
        slope = last.get("ema_slope", 0)
        if side == "long" and adx > 25 and slope < -0.15:
            return False
        if side == "short" and adx > 25 and slope > 0.15:
            return False
        return True

    def custom_stake_amount(
        self, pair, current_time, current_rate, proposed_stake,
        min_stake, max_stake, leverage, entry_tag, side, **kwargs
    ):
        return proposed_stake * float(self.position_c.value)