VMM_BTC

"""
VMM — Validated Market Making (DCA) Strategy
=============================================
Freqtrade implementation of the HYPE-USDT regime-conditional DCA market
making strategy, walk-forward validated across three out-of-sample windows
(Sep'25–Jan'26 crash: +28.4%, Jan–Mar'26 recovery: +19.9%,
Mar–Jun'26 breakout: +10.8%; Sharpe ~2, max DD ≤ 12% in each).

Architecture
------------
Base timeframe : 1m   (entry/exit execution)
Informative    : 15m  (regime: ADX14 + EMA50)
Informative    : 1h   (trend gate: EMA200)

LONG  (regime = bull or neutral, AND 1h close > 1h EMA200):
    Entry : limit at prev-candle typical price x (1 - entry_offset)
    DCA1  : +50%  of remaining stake when price <= last fill x (1 - dca_gap)
    DCA2  : +100% of remaining stake when price <= last fill x (1 - dca_gap)
    TP    : take_profit (2%) above average entry (auto-recomputed after DCA)
    SL    : long_sl (3.8%) below average entry, ACTIVE ONLY after DCA2

SHORT (regime = bear, AND 1h close < 1h EMA200): mirrored, plus
    SL    : short_sl (5%) above average entry, ACTIVE ONLY after DCA2

FORCE-EXIT (both sides):
    Opposite 15m regime confirmed for >= flip_confirm consecutive 15m bars
    -> exit at market. This is the crash protection that flipped the
    Sep'25–Jan'26 window from -61% to +28%.

Execution notes
---------------
- Entries are LIMIT orders at the offset price (custom_entry_price).
- TP exits via custom_exit (tag 'tp') — configure exit pricing to limit.
- Force-exits and short SL are effectively market.
- max_open_trades = 1 (sequential slot — by design).
- Designed for FUTURES mode with isolated margin, leverage 1x.

Config requirements (config.json):
    "trading_mode": "futures",
    "margin_mode": "isolated",
    "order_types": {"entry": "limit", "exit": "limit",
                    "force_exit": "market", "stoploss": "market",
                    "stoploss_on_exchange": false}
"""

from datetime import datetime
from typing import Optional, Union

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

from freqtrade.persistence import Trade
from freqtrade.strategy import (
    DecimalParameter,
    IntParameter,
    IStrategy,
    informative,
    stoploss_from_absolute,
)


class VMM_BTC(IStrategy):
    """Validated Market Making — regime-conditional DCA MM for crypto perps."""

    INTERFACE_VERSION = 3

    # ── Core architecture ────────────────────────────────────────────
    timeframe = "1m"
    can_short = True
    process_only_new_candles = True
    use_exit_signal = True
    exit_profit_only = False
    ignore_roi_if_entry_signal = False

    # Sequential single-slot architecture (by design)
    max_open_trades = 1

    # DCA
    position_adjustment_enable = True
    max_entry_position_adjustment = 2  # DCA1 + DCA2

    # Startup: 1h EMA200 needs 200 x 1h candles; 250 covers all informatives
    startup_candle_count: int = 250

    # ROI disabled — TP handled in custom_exit so it stays hyperopt-able
    minimal_roi = {"0": 100}

    # Hard stoploss effectively disabled; real stops via custom_stoploss
    stoploss = -0.99
    use_custom_stoploss = True

    # ── Strategy parameters (walk-forward consensus values) ──────────
    # x: entry offset from prev-candle typical price
    entry_offset = DecimalParameter(
        0.001, 0.009, default=0.0086, decimals=4, space="buy", optimize=True
    )
    # y: take-profit from average entry (min 0.8% enforced below). default=0.0159
    take_profit = DecimalParameter(
        0.008, 0.025, default=0.0084, decimals=4, space="sell", optimize=True
    )
    # z: short stop-loss after DCA2
    short_sl = DecimalParameter(
        0.02, 0.08, default=0.02, decimals=3, space="sell", optimize=True
    )
    # Long stop-loss after DCA2
    long_sl = DecimalParameter(
        0.02, 0.08, default=0.036, decimals=3, space="sell", optimize=True
    )
    # DCA trigger: adverse move from last fill
    dca_gap = DecimalParameter(
        0.02, 0.05, default=0.03, decimals=3, space="buy", optimize=False
    )
    # Regime-flip confirmation (consecutive 15m bars of opposite regime)
    flip_confirm = IntParameter(2, 6, default=5, space="sell", optimize=True)
    # ADX threshold for trending regime
    adx_threshold = IntParameter(15, 30, default=20, space="buy", optimize=False)

    # Fraction of available stake used at initial entry
    initial_stake_pct = 0.30

    # ── Plotting ─────────────────────────────────────────────────────
    plot_config = {
        "main_plot": {
            "ema50_15m": {"color": "orange"},
            "ema200_1h": {"color": "purple"},
        },
        "subplots": {
            "Regime": {
                "adx_15m": {"color": "blue"},
                "bear_streak_15m": {"color": "red"},
                "bull_streak_15m": {"color": "green"},
            },
        },
    }

    # ══════════════════════════════════════════════════════════════════
    # INFORMATIVE TIMEFRAMES
    # ══════════════════════════════════════════════════════════════════

    @informative("15m")
    def populate_indicators_15m(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """15m regime: ADX(14) + EMA(50) -> bull / bear / neutral + streaks."""
        dataframe["adx"] = ta.ADX(dataframe, timeperiod=14)
        dataframe["ema50"] = ta.EMA(dataframe, timeperiod=50)

        adx_ok = dataframe["adx"] > self.adx_threshold.value
        bull = (dataframe["close"] > dataframe["ema50"]) & adx_ok
        bear = (dataframe["close"] < dataframe["ema50"]) & adx_ok

        dataframe["regime"] = np.select([bull, bear], [1, -1], default=0)

        # Consecutive opposite-regime streaks (for flip-exit confirmation)
        is_bear = (dataframe["regime"] == -1).astype(int)
        grp_bear = (is_bear != is_bear.shift()).cumsum()
        dataframe["bear_streak"] = is_bear.groupby(grp_bear).cumsum() * is_bear

        is_bull = (dataframe["regime"] == 1).astype(int)
        grp_bull = (is_bull != is_bull.shift()).cumsum()
        dataframe["bull_streak"] = is_bull.groupby(grp_bull).cumsum() * is_bull

        return dataframe

    @informative("1h")
    def populate_indicators_1h(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """1h EMA200 trend gate — blocks counter-trend entries."""
        dataframe["ema200"] = ta.EMA(dataframe, timeperiod=200)
        dataframe["above_ema200"] = (dataframe["close"] > dataframe["ema200"]).astype(int)
        return dataframe

    # ══════════════════════════════════════════════════════════════════
    # BASE TIMEFRAME
    # ══════════════════════════════════════════════════════════════════

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Mid price proxy: typical price (OHLC/4) of the PREVIOUS candle
        dataframe["mid_price"] = (
            (dataframe["open"] + dataframe["high"] + dataframe["low"] + dataframe["close"])
            / 4
        ).shift(1)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        Entry signals. The actual fill price is set in custom_entry_price
        (limit at mid x (1 +/- entry_offset)); the signal only gates WHEN
        an order may rest in the book.
        """
        dataframe.loc[:, "enter_long"] = 0
        dataframe.loc[:, "enter_short"] = 0

        # LONG: regime bull(1) or neutral(0)  +  1h gate up
        dataframe.loc[
            (dataframe["regime_15m"] >= 0)
            & (dataframe["above_ema200_1h"] == 1)
            & (dataframe["volume"] > 0),
            ["enter_long", "enter_tag"],
        ] = (1, "vmm_long")

        # SHORT: regime bear(-1)  +  1h gate down
        dataframe.loc[
            (dataframe["regime_15m"] == -1)
            & (dataframe["above_ema200_1h"] == 0)
            & (dataframe["volume"] > 0),
            ["enter_short", "enter_tag"],
        ] = (1, "vmm_short")

        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """All exits handled by custom_exit / custom_stoploss."""
        dataframe.loc[:, "exit_long"] = 0
        dataframe.loc[:, "exit_short"] = 0
        return dataframe

    # ══════════════════════════════════════════════════════════════════
    # PRICING & SIZING
    # ══════════════════════════════════════════════════════════════════

    def custom_entry_price(
        self,
        pair: str,
        trade: Optional[Trade],
        current_time: datetime,
        proposed_rate: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> float:
        """Limit entry at prev-candle typical price x (1 -/+ entry_offset)."""
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return proposed_rate
        mid = dataframe["mid_price"].iat[-1]
        if pd.isna(mid):
            return proposed_rate
        if side == "long":
            return mid * (1.0 - self.entry_offset.value)
        return mid * (1.0 + self.entry_offset.value)

    def custom_stake_amount(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_stake: float,
        min_stake: Optional[float],
        max_stake: float,
        leverage: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> float:
        """Initial entry = 30% of total stake (keeps 70% for DCA1+DCA2)."""
        available = self.wallets.get_total_stake_amount()
        stake = available * self.initial_stake_pct
        return max(min_stake or 0.0, min(stake, max_stake))

    def adjust_trade_position(
        self,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        min_stake: Optional[float],
        max_stake: float,
        current_entry_rate: float,
        current_exit_rate: float,
        current_entry_profit: float,
        current_exit_profit: float,
        **kwargs,
    ) -> Union[Optional[float], tuple[Optional[float], Optional[str]]]:
        """
        DCA ladder:
          DCA1 (2nd fill) : 50%  of remaining available stake
          DCA2 (3rd fill) : 100% of remaining available stake
        Triggered on an adverse move >= dca_gap from the LAST FILL price.
        """
        n_entries = trade.nr_of_successful_entries
        if n_entries >= 3:  # entry + 2 DCAs already done
            return None

        filled = trade.select_filled_orders(trade.entry_side)
        if not filled:
            return None
        last_fill = filled[-1].safe_price
        gap = self.dca_gap.value

        if trade.is_short:
            triggered = current_rate >= last_fill * (1.0 + gap)
        else:
            triggered = current_rate <= last_fill * (1.0 - gap)
        if not triggered:
            return None

        available = self.wallets.get_available_stake_amount()
        if available <= (min_stake or 0.0):
            return None

        if n_entries == 1:        # -> DCA1
            stake = available * 0.50
            tag = "dca1"
        else:                      # n_entries == 2 -> DCA2
            stake = available * 1.00
            tag = "dca2"

        stake = min(stake, max_stake)
        if min_stake and stake < min_stake:
            return None
        return stake, tag

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> float:
        return 2.0

    # ══════════════════════════════════════════════════════════════════
    # EXITS
    # ══════════════════════════════════════════════════════════════════

    def custom_exit(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        **kwargs,
    ) -> Optional[Union[str, bool]]:
        """
        1) Take profit: current_profit >= take_profit (>= 0.8% floor).
           current_profit is computed from the trade's AVERAGE open rate,
           so the TP level auto-updates after every DCA fill — exactly the
           'update TP to avg +/- y%' rule from the spec.
        2) Regime-flip force-exit: opposite 15m regime confirmed for
           flip_confirm consecutive bars.
        """
        # ── 1) Take profit ──
        tp = max(self.take_profit.value, 0.008)
        if current_profit >= tp:
            return "tp"

        # ── 2) Regime-flip force-exit ──
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return None
        last = dataframe.iloc[-1]

        if not trade.is_short:
            if last.get("bear_streak_15m", 0) >= self.flip_confirm.value:
                return "regime_flip"
        else:
            if last.get("bull_streak_15m", 0) >= self.flip_confirm.value:
                return "regime_flip"

        return None

    def custom_stoploss(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        after_fill: bool,
        **kwargs,
    ) -> Optional[float]:
        """
        Longs : after DCA2 (3rd fill), hard stop at avg entry x (1 - long_sl).
        Shorts: after DCA2 (3rd fill), hard stop at avg entry x (1 + short_sl).
        """
        if trade.nr_of_successful_entries < 3:
            return None

        if trade.is_short:
            stop_price = trade.open_rate * (1.0 + self.short_sl.value)
            return stoploss_from_absolute(
                stop_price, current_rate, is_short=True, leverage=trade.leverage
            )
        else:
            stop_price = trade.open_rate * (1.0 - self.long_sl.value)
            return stoploss_from_absolute(
                stop_price, current_rate, is_short=False, leverage=trade.leverage
            )

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time: datetime,
        entry_tag: Optional[str],
        side: str,
        **kwargs,
    ) -> bool:
        """Final sanity gate: never enter with stale informative data."""
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe.empty:
            return False
        last = dataframe.iloc[-1]
        if pd.isna(last.get("regime_15m")) or pd.isna(last.get("above_ema200_1h")):
            return False
        return True