AdaptiveMLStrategy

"""
AdaptiveML v3 — USDT Futures 5m R2 Short Specialist
====================================================
Currently a single-regime specialist (R2 RANGING, short-only).
Other regimes (R0/R1/R3) are disabled — they produce more
SL losses than ROI wins.

  CORE:
  1. Multi-timeframe 5m + 15m + 1h confirmation
  2. Self-learned ROI from trade duration buckets
  3. Session filter + directional bias
  4. Performance feedback loop (entry_adj/size_adj)
  5. Rule-based regime detection (4 base + 8 sub)

  DISCIPLINE:
  6.  Kelly Criterion position sizing (fractional Kelly)
  7.  MFE-calibrated trailing stops (data-driven)
  8.  Equity curve circuit breaker (pause on drawdown)
  9.  Consecutive loss cooldown (discipline after streaks)
  10. Daily loss limit (hard stop for the day)
  11. Fee-aware minimum edge filter
  12. Trade quality gate (session/direction prior)
  13. Anti-martingale (reduce size on losses)

Honest assessment:
  - Quality model uses 5 features (hour, weekday, is_short,
    regime, leverage) — a session-direction-regime prior,
    not deep trade intelligence.
  - Regime model was removed (iter 14). Regime detection
    is rule-based (ADX/EMA/ATR/BB thresholds).
  - Only R2 short is active. The "adaptive" framing is
    aspirational, not current reality.

Philosophy: "Discipline is the bridge between goals and results."
"""

import hashlib
import hmac
import json
import logging
import pickle  # noqa: S301 — used for quality_model.pkl
from datetime import datetime
from pathlib import Path

import numpy as np
import pandas as pd
import talib.abstract as ta
from freqtrade.strategy import (
    DecimalParameter,
    IntParameter,
    IStrategy,
)
from pandas import DataFrame

# ─── Modular engines (iter 19 P3 architecture split) ────
from regime_engine import (
    detect_regime,
    get_regime_params,
    get_current_session,
    SESSION_HOURS,
)
from discipline_engine import (
    load_discipline_state,
    save_discipline_state,
    check_max_drawdown,
    check_pair_exposure,
    check_correlation_exposure,
    snap_features,
    log_decision,
    log_trade_entry,
    log_trade_exit,
)
from log_config import configure_json_logging, get_structured_logger

logger = logging.getLogger(__name__)

# Path where the ML optimizer writes trained models
MODEL_DIR = Path("/freqtrade/user_data/ml_models")
MODEL_DIR.mkdir(parents=True, exist_ok=True)

# Active model artifacts:
BEST_PARAMS_PATH = MODEL_DIR / "best_params.json"
QUALITY_MODEL_PATH = MODEL_DIR / "quality_model.pkl"
DISCIPLINE_PATH = MODEL_DIR / "discipline_params.json"
ANTI_PATTERN_PATH = MODEL_DIR / "anti_patterns.json"

# Shadow model: evaluated but cannot trade — for A/B validation
SHADOW_DIR = MODEL_DIR / "shadow"
SHADOW_MODEL_PATH = SHADOW_DIR / "quality_model.pkl"

# Decision journal — persists every trade rejection reason
REJECTION_LOG_PATH = MODEL_DIR / "rejection_journal.json"

# Decision Journal v4 — append-only JSONL with every accept+reject
DECISION_JOURNAL_PATH = MODEL_DIR / "decision_journal.jsonl"

# Trade replay — per-trade entry/exit with full context
TRADE_REPLAY_PATH = MODEL_DIR / "trade_replay.json"

# Discipline state persistence — survives container restarts
DISCIPLINE_STATE_PATH = MODEL_DIR / "discipline_state.json"

# Kill-switch: if this file exists, block ALL entries
KILL_SWITCH_PATH = MODEL_DIR / "kill_switch"

# Event freeze windows (UTC hours) — no trading around
# scheduled macro events. Format: {(month, day): [(hour_start, hour_end), ...]}
# Static list; for dynamic FOMC/CPI dates, update periodically.
# These are UTC windows where entry is blocked.
EVENT_FREEZE_HOURS = 3  # freeze for N hours around event

# Fee structure
ROUND_TRIP_FEE = 0.001  # 0.10% (Binance futures maker+taker)

# Risk: strategy-level max drawdown halt (defense in depth)
# This fires INSIDE the strategy even if agent-runner is down.
MAX_DD_HALT = 0.20  # 20% — auto-enable kill switch
MAX_DD_WARN = 0.15  # 15% — log warning, reduce size

# Per-pair position limit: max open trades on one pair
MAX_POSITIONS_PER_PAIR = 2

# Correlated pair groups — simultaneous same-direction entries
# on pairs within the same group are limited.
# Rolling 90d correlation: BTC/ETH ~0.87, SOL/ETH ~0.82
CORRELATION_GROUPS = {
    "layer1": ["ETH/USDT:USDT", "BTC/USDT:USDT", "SOL/USDT:USDT"],
    "altcoin": ["BNB/USDT:USDT", "XRP/USDT:USDT", "DOGE/USDT:USDT"],
}
MAX_CORRELATED_SAME_DIRECTION = 2  # max 2 of 3 in same group+direction

# HMAC key file for model integrity verification
MODEL_HMAC_PATH = MODEL_DIR / "model_hmac.json"


class AdaptiveMLStrategy(IStrategy):
    """
    USDT Futures 5m R2 Short Specialist (6 pairs)
    Currently runs one active regime (R2 RANGING, short-only).
    Uses Kelly sizing, MFE-calibrated exits, and discipline systems.
    """

    INTERFACE_VERSION = 3
    timeframe = "5m"
    can_short = True  # FIXED: was False, blocking all short signals

    # Dynamic parameters - updated by ML optimizer
    position_c = DecimalParameter(0.1, 1.0, default=0.50, space="buy", optimize=True)
    entry_bias = DecimalParameter(-0.5, 0.5, default=0.00, space="buy", optimize=True)

    # Regime detection windows
    regime_lookback = IntParameter(20, 100, default=50, space="buy", optimize=True)

    # ML confidence threshold
    min_confidence = DecimalParameter(
        0.3, 0.8, default=0.55, space="buy", optimize=True
    )

    # ROI targets: keep high initial target to let winners run
    # 100% WR on ROI exits — don't cut them short
    minimal_roi = {
        "0": 0.015,
        "15": 0.010,
        "30": 0.006,
        "60": 0.003,
        "120": 0.001,
    }

    stoploss = -0.025  # Backstop only; real SL in custom_stoploss()
    use_custom_stoploss = True
    trailing_stop = False

    startup_candle_count = 200

    # Internal state - ML models
    _best_params = None
    _quality_model_data = None  # Quality gate model
    _shadow_model_data = None  # Shadow: candidate model (log-only)
    _discipline_params = None  # Discipline params (cooldown, limits)
    _anti_patterns = None  # Anti-patterns (learned from mistakes)
    _last_model_load = 0
    _MODEL_RELOAD_INTERVAL = 300

    # Discipline tracking (reset on strategy reload)
    _consecutive_losses = 0
    _daily_pnl = 0.0
    _daily_trades = 0
    _last_trade_date = None
    _equity_curve = []  # rolling equity for circuit breaker
    _recent_results = []  # last N trade results
    _MAX_RECENT = 20  # track last 20 trades
    _paused_until = None  # cooldown timestamp
    _rejection_log = []  # decision journal: last 100 rejections
    _MAX_REJECTIONS = 100
    _trade_replay = []  # trade replay: entry+exit decisions
    _MAX_REPLAY = 200

    # Funding rate cache (populated in bot_loop_start for live/dry_run)
    _funding_rates: dict = {}  # {pair: {"rate": float, "ts": str}}
    _FUNDING_RATE_CACHE_TTL = 300  # 5 minutes
    _last_funding_fetch = 0

    # ─── Multi-timeframe informative pairs ───────────────
    def informative_pairs(self):
        """Request 15m and 1h data for higher-TF confirmation."""
        pairs = self.dp.current_whitelist()
        informative = []
        for pair in pairs:
            informative.append((pair, "15m"))
            informative.append((pair, "1h"))
        return informative

    # ─── Funding Rate Fetcher (live/dry_run only) ────────

    def bot_start(self, **kwargs):
        """Called once after strategy load. Configure JSON logging."""
        configure_json_logging()
        logger.info("AdaptiveMLStrategy initialized — R2 short specialist")

    def bot_loop_start(self, current_time, **kwargs):
        """
        Fetch funding rates for all whitelisted pairs every 5 minutes.
        Uses Freqtrade's exchange (ccxt) to call Binance premiumIndex.
        Only runs in live/dry_run — no-op during backtest/hyperopt.
        """
        if self.config.get("runmode", {}).value not in ("live", "dry_run"):
            return
        now = datetime.now().timestamp()
        if now - self._last_funding_fetch < self._FUNDING_RATE_CACHE_TTL:
            return
        self._last_funding_fetch = now
        try:
            exchange = self.dp._exchange._api  # ccxt instance
            pairs = self.dp.current_whitelist()
            for pair in pairs:
                try:
                    fr = exchange.fetch_funding_rate(pair)
                    self._funding_rates[pair] = {
                        "rate": float(fr.get("fundingRate", 0) or 0),
                        "next_ts": fr.get("fundingTimestamp"),
                        "mark_price": float(fr.get("markPrice", 0) or 0),
                    }
                except Exception as e:
                    logger.debug(
                        "Funding rate fetch failed for %s: %s", pair, e
                    )
            if self._funding_rates:
                logger.info(
                    "Funding rates updated: %s",
                    {p: f"{v['rate']:.6f}" for p, v in self._funding_rates.items()},
                )
        except Exception as e:
            logger.debug("Funding rate batch fetch failed: %s", e)

    def get_funding_rates(self) -> dict:
        """Public accessor for cached funding rates (used by dashboard API)."""
        return self._funding_rates

    def _load_models(self):
        """Hot-reload ML models and params from disk."""
        now = datetime.now().timestamp()
        if now - self._last_model_load < self._MODEL_RELOAD_INTERVAL:
            return

        self._last_model_load = now

        if BEST_PARAMS_PATH.exists():
            try:
                with open(BEST_PARAMS_PATH, "r") as f:
                    self._best_params = json.load(f)
                # Staleness check: warn if params are > 7 days old
                age_days = (now - BEST_PARAMS_PATH.stat().st_mtime) / 86400
                if age_days > 7:
                    logger.warning(
                        "best_params.json is %.0f days old — consider retraining "
                        "(./scripts/ml-train.sh or curl -X POST :3001/api/ml/train)",
                        age_days,
                    )
            except Exception as e:
                logger.warning("Failed to load best_params.json: %s", e)
                self._best_params = None

        # Load quality model (with HMAC integrity check)
        if QUALITY_MODEL_PATH.exists():
            try:
                raw = QUALITY_MODEL_PATH.read_bytes()
                if self._verify_model_hmac("quality_model.pkl", raw):
                    self._quality_model_data = pickle.loads(raw)
                else:
                    logger.warning(
                        "quality_model.pkl HMAC mismatch — "
                        "refusing to load (possible tampering)"
                    )
                    self._quality_model_data = None
            except Exception as e:
                logger.warning("Failed to load quality_model.pkl: %s", e)
                self._quality_model_data = None

        # Load discipline params
        if DISCIPLINE_PATH.exists():
            try:
                with open(DISCIPLINE_PATH, "r") as f:
                    self._discipline_params = json.load(f)
            except Exception as e:
                logger.warning("Failed to load discipline_params.json: %s", e)
                self._discipline_params = None

        # Load anti-patterns (learned from mistakes)
        if ANTI_PATTERN_PATH.exists():
            try:
                with open(ANTI_PATTERN_PATH, "r") as f:
                    self._anti_patterns = json.load(f)
            except Exception as e:
                logger.warning("Failed to load anti_patterns.json: %s", e)
                self._anti_patterns = None

        # Shadow model: loaded if present, never affects trading
        if SHADOW_MODEL_PATH.exists():
            try:
                raw = SHADOW_MODEL_PATH.read_bytes()
                if self._verify_model_hmac("shadow/quality_model.pkl", raw):
                    self._shadow_model_data = pickle.loads(raw)
                else:
                    logger.warning(
                        "Shadow model HMAC mismatch — skipping"
                    )
                    self._shadow_model_data = None
            except Exception as e:
                logger.warning("Failed to load shadow model: %s", e)
                self._shadow_model_data = None
        else:
            self._shadow_model_data = None

        # Restore discipline state from disk (survives restarts)
        self._load_discipline_state()

    def _load_discipline_state(self):
        """Restore discipline state from disk — delegates to discipline_engine."""
        state = {
            "consecutive_losses": self._consecutive_losses,
            "daily_pnl": self._daily_pnl,
            "daily_trades": self._daily_trades,
            "recent_results": self._recent_results,
        }
        state = load_discipline_state(DISCIPLINE_STATE_PATH, state)
        self._consecutive_losses = state["consecutive_losses"]
        self._daily_pnl = state["daily_pnl"]
        self._daily_trades = state["daily_trades"]
        self._recent_results = state["recent_results"]

    def _save_discipline_state(self):
        """Persist discipline state to disk — delegates to discipline_engine."""
        state = {
            "consecutive_losses": self._consecutive_losses,
            "daily_pnl": self._daily_pnl,
            "daily_trades": self._daily_trades,
            "recent_results": self._recent_results,
        }
        save_discipline_state(DISCIPLINE_STATE_PATH, state, self._MAX_RECENT)

    # ─── Model Integrity ─────────────────────────────────

    @staticmethod
    def _verify_model_hmac(model_name, raw_bytes):
        """
        Verify HMAC-SHA256 of a model file against stored digest.
        If no HMAC file exists yet, trust-on-first-use: compute
        and store the digest for future verification.
        Returns True if verified or TOFU, False if mismatch.
        """
        try:
            # Key from env or fallback (production should set MODEL_HMAC_KEY)
            import os
            key = os.environ.get(
                "MODEL_HMAC_KEY", "cc-model-integrity-key"
            ).encode()
            digest = hmac.new(key, raw_bytes, hashlib.sha256).hexdigest()

            stored = {}
            if MODEL_HMAC_PATH.exists():
                with open(MODEL_HMAC_PATH, "r") as f:
                    stored = json.load(f)

            if model_name in stored:
                return hmac.compare_digest(stored[model_name], digest)

            # Trust-on-first-use: store digest
            stored[model_name] = digest
            with open(MODEL_HMAC_PATH, "w") as f:
                json.dump(stored, f, indent=1)
            logger.info(
                "TOFU: stored HMAC for %s (first load)", model_name
            )
            return True
        except Exception as e:
            logger.warning("HMAC check failed for %s: %s", model_name, e)
            return True  # fail open on HMAC infra error

    # ─── Portfolio Risk Checks ───────────────────────────

    def _check_max_drawdown(self, current_time):
        """Strategy-level max DD halt — delegates to discipline_engine."""
        return check_max_drawdown(
            self._equity_curve, KILL_SWITCH_PATH,
            current_time, MAX_DD_HALT, MAX_DD_WARN,
        )

    def _check_pair_exposure(self, pair):
        """Per-pair position limit — delegates to discipline_engine."""
        return check_pair_exposure(pair, self.dp, MAX_POSITIONS_PER_PAIR)

    def _check_correlation_exposure(self, pair, side):
        """Cross-pair correlation filter — delegates to discipline_engine."""
        return check_correlation_exposure(
            pair, side, CORRELATION_GROUPS, MAX_CORRELATED_SAME_DIRECTION,
        )

    # ─── Regime Detection ────────────────────────────────

    def _detect_regime(self, dataframe: DataFrame) -> DataFrame:
        """Classify market into regimes — delegates to regime_engine."""
        return detect_regime(dataframe, int(self.regime_lookback.value))

    def _get_regime_params(self, regime: int) -> dict:
        """Get optimal params for regime — delegates to regime_engine."""
        return get_regime_params(regime, self._best_params)

    def _get_current_session(self, current_time=None):
        """Determine current trading session — delegates to regime_engine."""
        return get_current_session(current_time)

    # ─── Multi-Timeframe Indicators ──────────────────────

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # Hot-reload ML models
        self._load_models()

        pair = metadata.get("pair", "")

        # ─── 1H informative indicators ───
        inf_1h = self.dp.get_pair_dataframe(pair, "1h")
        if inf_1h is not None and len(inf_1h) > 0:
            inf_1h["ema50_1h"] = ta.EMA(inf_1h, timeperiod=50)
            inf_1h["ema20_1h"] = ta.EMA(inf_1h, timeperiod=20)
            inf_1h["rsi_1h"] = ta.RSI(inf_1h, timeperiod=14)
            inf_1h["adx_1h"] = ta.ADX(inf_1h, timeperiod=14)
            # 1h trend direction: EMA50 slope
            inf_1h["ema50_slope_1h"] = (
                (inf_1h["ema50_1h"] - inf_1h["ema50_1h"].shift(3))
                / inf_1h["ema50_1h"].shift(3)
                * 100
            )
            # 1h trend strength
            inf_1h["trend_1h"] = np.where(
                inf_1h["ema20_1h"] > inf_1h["ema50_1h"], 1, -1
            )
            # Merge to 5m dataframe
            dataframe = self._merge_informative(
                dataframe,
                inf_1h,
                [
                    "ema50_1h",
                    "ema20_1h",
                    "rsi_1h",
                    "adx_1h",
                    "ema50_slope_1h",
                    "trend_1h",
                ],
            )
        else:
            dataframe["ema50_1h"] = np.nan
            dataframe["ema20_1h"] = np.nan
            dataframe["rsi_1h"] = np.nan
            dataframe["adx_1h"] = np.nan
            dataframe["ema50_slope_1h"] = 0
            dataframe["trend_1h"] = 0

        # ─── 15M informative indicators ───
        inf_15m = self.dp.get_pair_dataframe(pair, "15m")
        if inf_15m is not None and len(inf_15m) > 0:
            inf_15m["rsi_15m"] = ta.RSI(inf_15m, timeperiod=14)
            macd_15m = ta.MACD(inf_15m)
            inf_15m["macd_15m"] = macd_15m["macd"]
            inf_15m["macd_signal_15m"] = macd_15m["macdsignal"]
            inf_15m["macd_hist_15m"] = macd_15m["macdhist"]
            inf_15m["ema12_15m"] = ta.EMA(inf_15m, timeperiod=12)
            inf_15m["ema26_15m"] = ta.EMA(inf_15m, timeperiod=26)
            inf_15m["trend_15m"] = np.where(
                inf_15m["ema12_15m"] > inf_15m["ema26_15m"], 1, -1
            )
            dataframe = self._merge_informative(
                dataframe, inf_15m, ["rsi_15m", "macd_hist_15m", "trend_15m"]
            )
        else:
            dataframe["rsi_15m"] = np.nan
            dataframe["macd_hist_15m"] = 0
            dataframe["trend_15m"] = 0

        # ─── Regime detection ───
        dataframe = self._detect_regime(dataframe)

        # ─── A52-style indicators ───
        dataframe["ema_12"] = ta.EMA(dataframe, timeperiod=12)
        dataframe["ema_26"] = ta.EMA(dataframe, timeperiod=26)
        dataframe["rsi"] = ta.RSI(dataframe, timeperiod=14)
        dataframe["atr"] = ta.ATR(dataframe, timeperiod=14)

        macd = ta.MACD(dataframe)
        dataframe["macd"] = macd["macd"]
        dataframe["macd_signal"] = macd["macdsignal"]
        dataframe["macd_hist"] = macd["macdhist"]

        bb = ta.BBANDS(dataframe, timeperiod=20)
        dataframe["bb_upper"] = bb["upperband"]
        dataframe["bb_lower"] = bb["lowerband"]
        dataframe["bb_mid"] = bb["middleband"]

        # ─── A51-style VWAP ───
        tp = (dataframe["high"] + dataframe["low"] + dataframe["close"]) / 3
        cumvol = dataframe["volume"].rolling(window=288, min_periods=1).sum()
        cumtp = (tp * dataframe["volume"]).rolling(window=288, min_periods=1).sum()
        dataframe["vwap"] = cumtp / cumvol

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

        # ─── A31-style squeeze ───
        kc_mid = ta.EMA(dataframe, timeperiod=20)
        kc_atr = ta.ATR(dataframe, timeperiod=20)
        dataframe["kc_upper"] = kc_mid + 1.5 * kc_atr
        dataframe["kc_lower"] = kc_mid - 1.5 * kc_atr
        dataframe["squeeze_on"] = (
            (dataframe["bb_lower"] > dataframe["kc_lower"])
            & (dataframe["bb_upper"] < dataframe["kc_upper"])
        ).astype(int)
        dataframe["squeeze_fire"] = (
            (dataframe["squeeze_on"].shift(1) == 1) & (dataframe["squeeze_on"] == 0)
        ).astype(int)
        delta = dataframe["close"] - kc_mid
        dataframe["momentum"] = ta.LINEARREG(delta, timeperiod=20)

        # ─── OPT-style SuperTrend ───
        hl2 = (dataframe["high"] + dataframe["low"]) / 2
        st_atr = ta.ATR(dataframe, timeperiod=10)
        dataframe["st_upper"] = hl2 + 3.0 * st_atr
        dataframe["st_lower"] = hl2 - 3.0 * st_atr

        # Volume filter
        dataframe["volume_sma"] = ta.SMA(dataframe["volume"], timeperiod=20)

        # ─── Combined direction score ───
        trend = np.where(dataframe["ema_12"] > dataframe["ema_26"], 1.0, -1.0)
        mom = (dataframe["rsi"] - 50) / 50
        macd_s = np.where(dataframe["macd_hist"] > 0, 0.5, -0.5)

        # Multi-TF confirmation bonus
        tf_bonus = (
            dataframe["trend_1h"].fillna(0) * 0.15
            + dataframe["trend_15m"].fillna(0) * 0.10
        )

        # Dynamic bias from regime (capped to prevent over-filtering)
        regime_bias = dataframe["regime"].map(
            lambda r: max(-0.05, min(0.05, self._get_regime_params(r).get("e", 0.0)))
        )

        dataframe["direction_score"] = (
            trend * 0.35 + mom * 0.25 + macd_s * 0.15 + tf_bonus + regime_bias
        )

        # ─── Multi-TF agreement score (for entry filtering) ───
        dataframe["mtf_agree_long"] = (
            (dataframe["trend_1h"] > 0).astype(int)
            + (dataframe["trend_15m"] > 0).astype(int)
            + (dataframe["ema_12"] > dataframe["ema_26"]).astype(int)
        )
        dataframe["mtf_agree_short"] = (
            (dataframe["trend_1h"] < 0).astype(int)
            + (dataframe["trend_15m"] < 0).astype(int)
            + (dataframe["ema_12"] < dataframe["ema_26"]).astype(int)
        )

        return dataframe

    def _merge_informative(self, dataframe, inf_df, columns):
        """Merge higher-TF informative data into 5m dataframe.
        Uses merge_asof for correct forward-fill alignment."""
        if inf_df is None or len(inf_df) == 0:
            return dataframe

        inf_copy = inf_df[["date"] + columns].copy()
        inf_copy = inf_copy.sort_values("date").reset_index(drop=True)
        dataframe = dataframe.sort_values("date").reset_index(drop=True)

        # merge_asof: for each 5m row, find the latest higher-TF
        # row where date <= 5m date (backward direction)
        merged = pd.merge_asof(
            dataframe[["date"]],
            inf_copy,
            on="date",
            direction="backward",
        )
        for col in columns:
            if col in merged.columns:
                dataframe[col] = merged[col].values

        return dataframe

    # ─── Entry Logic ─────────────────────────────────────

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # ── Only R2 (RANGING) shorts are proven profitable ──
        # R0/R1/R3 all produce more SL losses than ROI wins.
        # Skip computing entry conditions for disabled regimes
        # to save ~120 lines of wasted computation per candle.

        regime_id = 2  # RANGING — the only active regime
        params = self._get_regime_params(regime_id)
        strategy = params.get("strategy", "A52")
        entry_adj = params.get("entry_adj", 1.0)
        direction_bias = params.get("direction_bias", "neutral")
        mask = dataframe["regime"] == regime_id

        # Ranging regime thresholds
        dir_thresh = 0.25 * min(entry_adj, 1.10)
        vol_mult = 1.0 * min(entry_adj, 1.10)

        # Multi-TF: 1-of-3 for ranging
        mtf_short = dataframe["mtf_agree_short"] >= 1

        # Direction bias adjustments
        if direction_bias == "short":
            short_dir_thresh = dir_thresh * 0.85
        else:
            short_dir_thresh = dir_thresh

        if strategy == "OPT":
            short_cond = (
                mask
                & mtf_short
                & (
                    (dataframe["ema_12"] < dataframe["ema_26"])
                    & (dataframe["direction_score"] < -short_dir_thresh)
                    & (dataframe["rsi"] > 30)
                    & (dataframe["rsi"] < 60)
                    & (dataframe["volume"] > dataframe["volume_sma"] * vol_mult)
                    & (dataframe["adx"] > 15)
                )
            )

        elif strategy == "A51":
            a51_vol = max(vol_mult, 1.1)
            short_cond = (
                mask
                & mtf_short
                & (
                    (dataframe["close"] < dataframe["vwap"])
                    & (dataframe["ema_5"] < dataframe["ema_8"])
                    & (dataframe["rsi"] > 38)
                    & (dataframe["rsi"] < 58)
                    & (dataframe["macd_hist"] < 0)
                    & (dataframe["adx"] > 20)
                    & (dataframe["direction_score"] < -0.15)
                    & (dataframe["volume"] > dataframe["volume_sma"] * a51_vol)
                )
            )

        elif strategy == "A31":
            short_cond = (
                mask
                & mtf_short
                & (
                    (dataframe["squeeze_fire"] == 1)
                    & (dataframe["momentum"] < 0)
                    & (dataframe["close"] < dataframe["ema_26"])
                    & (dataframe["direction_score"] < -0.2)
                    & (dataframe["rsi"] > 30)
                    & (dataframe["rsi"] < 60)
                    & (dataframe["adx"] > 15)
                    & (dataframe["volume"] > dataframe["volume_sma"] * 0.8)
                )
            )

        else:  # A52 default
            short_cond = (
                mask
                & mtf_short
                & (
                    (dataframe["direction_score"] < -short_dir_thresh)
                    & (dataframe["close"] < dataframe["ema_12"])
                    & (dataframe["rsi"] > 32)
                    & (dataframe["rsi"] < 60)
                    & (dataframe["macd_hist"] < 0)
                    & (dataframe["volume"] > dataframe["volume_sma"] * vol_mult)
                    & (dataframe["adx"] > 15)
                )
            )

        tag = "ml_{}_r{}".format(strategy.lower(), regime_id)
        dataframe.loc[short_cond, ["enter_short", "enter_tag"]] = (
            1,
            "{}_short".format(tag),
        )

        return dataframe

    # ─── Exit Logic ──────────────────────────────────────

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # ─── Smarter exits: require confirmation, not single triggers ───

        # Overbought exit: RSI extreme AND price above BB
        dataframe.loc[
            (dataframe["rsi"] > 78) & (dataframe["close"] > dataframe["bb_upper"]),
            ["exit_long", "exit_tag"],
        ] = (1, "ml_exit_overbought")

        dataframe.loc[
            (dataframe["rsi"] < 22) & (dataframe["close"] < dataframe["bb_lower"]),
            ["exit_short", "exit_tag"],
        ] = (1, "ml_exit_oversold")

        # Momentum reversal exits DISABLED
        # Data shows 0% WR and -1.12% avg loss.
        # The reversal signal fires too late, after
        # the damage is already done.
        # Keep overbought/oversold (RSI extreme + BB)
        # as these are rare and valid exit signals.

        # MTF trend flip exits DISABLED
        # These exit too late — the loss is already taken.
        # Let ROI + time exits handle it.

        # Vol spike exits DISABLED
        # Data shows 23.5% WR and -11.75 USDT total.
        # The vol spike signal exits at bad levels.

        return dataframe

    # ─── Dynamic Position Sizing (Kelly Criterion) ────────

    def custom_stake_amount(
        self,
        pair,
        current_time,
        current_rate,
        proposed_stake,
        min_stake,
        max_stake,
        leverage,
        entry_tag,
        side,
        **kwargs,
    ):
        """
        Kelly Criterion position sizing with discipline.
        f* = (bp - q) / b, using fractional Kelly (25%).
        Blended with regime confidence and anti-martingale.
        """
        c = float(self.position_c.value)

        if entry_tag and "_r" in entry_tag:
            try:
                regime = int(entry_tag.split("_r")[1][0])
                params = self._get_regime_params(regime)

                # Kelly fraction from optimizer
                kelly = params.get("kelly_fraction", 0)
                regime_c = params.get("c", c)
                if kelly > 0.02:
                    # Blend Kelly with regime c (conservative)
                    c = kelly * 0.4 + regime_c * 0.6
                else:
                    # Weak/zero Kelly = weak edge → use
                    # regime c as-is (no amplification).
                    # Sizing up on uncertain edge is the
                    # single most dangerous bug in a bot.
                    c = regime_c

                # Apply size_adj from performance feedback
                size_adj = params.get("size_adj", 1.0)
                c = c * size_adj

                # Not robust in walk-forward → reduce
                if not params.get("is_robust", True):
                    c *= 0.7  # cautious, not blocked

                # Anti-martingale (reduce after losses)
                if self._consecutive_losses >= 2:
                    reduction = max(0.3, 1.0 - self._consecutive_losses * 0.15)
                    c *= reduction

                # Reduce size in worst session
                worst = params.get("worst_session")
                if worst:
                    current_sess = self._get_current_session(current_time)
                    if current_sess == worst:
                        c *= 0.5

                # R2 short floor handled by 5× multiplier above

            except (ValueError, IndexError):
                pass

        c = max(0.05, min(c, 0.80))
        return proposed_stake * c

    # ─── Trade Entry Confirmation (Discipline Gate) ─────

    def _log_decision(self, pair, side, decision, reason, current_time,
                      rate=None, features=None, edge_score=None,
                      quality_threshold=None):
        """Decision Journal v4 — delegates to discipline_engine."""
        risk_state = {
            "consecutive_losses": self._consecutive_losses,
            "daily_pnl": round(self._daily_pnl, 6),
            "daily_trades": self._daily_trades,
        }
        self._rejection_log = log_decision(
            pair, side, decision, reason, current_time,
            journal_path=DECISION_JOURNAL_PATH,
            rejection_log_path=REJECTION_LOG_PATH,
            rejection_log=self._rejection_log,
            max_rejections=self._MAX_REJECTIONS,
            rate=rate, features=features,
            edge_score=edge_score,
            quality_threshold=quality_threshold,
            risk_state=risk_state,
            model_ts=self._last_model_load,
        )

    def _log_rejection(self, pair, side, reason, current_time,
                       features=None, rate=None, edge_score=None):
        """Backward-compat wrapper → _log_decision(decision='reject')."""
        self._log_decision(
            pair, side, "reject", reason, current_time,
            rate=rate, features=features, edge_score=edge_score,
        )

    def _snap_features(self, row):
        """Extract feature snapshot — delegates to discipline_engine."""
        return snap_features(row)

    def _log_trade_entry(self, pair, side, entry_tag, regime,
                         rate, current_time, features_row,
                         shadow=None):
        """Trade Replay entry — delegates to discipline_engine."""
        params = self._get_regime_params(regime)
        risk_state = {
            "consecutive_losses": self._consecutive_losses,
            "daily_pnl": round(self._daily_pnl, 6),
            "daily_trades": self._daily_trades,
        }
        self._trade_replay = log_trade_entry(
            pair, side, entry_tag, regime, rate, current_time,
            features_row, params, risk_state, self._last_model_load,
            self._trade_replay, TRADE_REPLAY_PATH, self._MAX_REPLAY,
            shadow=shadow,
        )

    def _evaluate_shadow(self, current_time, side, regime, kwargs):
        """
        Evaluate shadow (candidate) model — log-only, never trades.
        Returns dict with shadow decision or None if no shadow model.
        """
        if self._shadow_model_data is None:
            return None
        try:
            sm = self._shadow_model_data
            model = sm.get("model")
            scaler = sm.get("scaler")
            thresholds = sm.get("thresholds", {})
            if not model or not scaler:
                return None
            if not hasattr(current_time, "hour"):
                return None

            leverage = float(kwargs.get("leverage", 1) or 1)
            features = np.array(
                [[
                    current_time.hour,
                    current_time.weekday(),
                    1 if side == "short" else 0,
                    regime,
                    leverage,
                ]]
            )
            scaled = scaler.transform(features)
            quality = float(
                model.predict_proba(scaled)[0][1]
            )
            min_q = thresholds.get("min_quality", 0.5)
            return {
                "quality": round(quality, 4),
                "threshold": min_q,
                "would_allow": quality >= min_q,
            }
        except Exception:
            return None

    def _log_trade_exit(self, pair, trade, exit_reason, profit,
                        rate, current_time):
        """Trade Replay exit — delegates to discipline_engine."""
        risk_state = {
            "consecutive_losses": self._consecutive_losses,
            "daily_pnl": round(self._daily_pnl, 6),
        }
        self._trade_replay = log_trade_exit(
            pair, trade, exit_reason, profit, rate, current_time,
            self.dp, self.timeframe,
            self._trade_replay, TRADE_REPLAY_PATH, self._MAX_REPLAY,
            risk_state=risk_state,
        )

    # _persist_replay removed — now handled by discipline_engine

    def confirm_trade_entry(
        self,
        pair,
        order_type,
        amount,
        rate,
        time_in_force,
        current_time,
        entry_tag,
        side,
        **kwargs,
    ):
        """
        Discipline gate with Decision Journal v4.
        Every decision (accept AND reject) is persisted
        to append-only JSONL so operators can audit
        why the bot did or didn't trade.
        """
        # KILL SWITCH: if file exists, block all entries
        if KILL_SWITCH_PATH.exists():
            self._log_rejection(pair, side, "kill_switch_active", current_time)
            return False

        # GUARD: Only trade futures pairs (must have :USDT suffix)
        if ":USDT" not in pair:
            self._log_rejection(pair, side, "not_futures_pair", current_time)
            return False

        # EVENT FREEZE: block during scheduled macro events
        if hasattr(current_time, "hour"):
            event_file = MODEL_DIR / "event_calendar.json"
            if event_file.exists():
                try:
                    with open(event_file, "r") as ef:
                        events = json.load(ef)
                    # events: [{"date": "2026-04-15", "hour": 14, "name": "CPI"}, ...]
                    ct_str = str(current_time.date())
                    for ev in events:
                        if ev.get("date") == ct_str:
                            ev_hour = ev.get("hour", 12)
                            if abs(current_time.hour - ev_hour) <= EVENT_FREEZE_HOURS:
                                self._log_rejection(
                                    pair, side,
                                    f"event_freeze_{ev.get('name', 'macro')}",
                                    current_time,
                                )
                                return False
                except Exception:
                    pass

        # STRATEGY-LEVEL MAX DD HALT (defense in depth)
        dd_status = self._check_max_drawdown(current_time)
        if dd_status == "halt":
            self._log_rejection(
                pair, side, "max_dd_halt_auto_kill", current_time
            )
            return False

        # PER-PAIR POSITION LIMIT
        if not self._check_pair_exposure(pair):
            self._log_rejection(
                pair, side,
                f"per_pair_limit_{MAX_POSITIONS_PER_PAIR}",
                current_time,
            )
            return False

        # CROSS-PAIR CORRELATION FILTER
        if not self._check_correlation_exposure(pair, side):
            self._log_rejection(
                pair, side, "correlated_pair_limit", current_time,
            )
            return False

        # Reset daily counters
        if hasattr(current_time, "date"):
            today = current_time.date()
        else:
            today = datetime.utcnow().date()

        if self._last_trade_date != today:
            self._daily_pnl = 0.0
            self._daily_trades = 0
            self._last_trade_date = today
            self._save_discipline_state()  # persist reset

        # Cooldown after consecutive losses
        regime = 2  # default
        if entry_tag and "_r" in entry_tag:
            try:
                regime = int(entry_tag.split("_r")[1][0])
            except (ValueError, IndexError):
                pass

        params = self._get_regime_params(regime)
        max_losses = params.get("cooldown_after_losses", 5)

        if self._consecutive_losses >= max_losses:
            self._log_rejection(
                pair,
                side,
                f"cooldown_consecutive_losses_{self._consecutive_losses}",
                current_time,
            )
            return False

        # Daily loss limit
        daily_limit = params.get("daily_loss_limit", 0.02)
        if self._daily_pnl < -daily_limit:
            self._log_rejection(
                pair, side, f"daily_loss_limit_{self._daily_pnl:.4f}", current_time
            )
            return False

        # Equity curve circuit breaker
        if len(self._recent_results) >= 20:
            recent = self._recent_results[-20:]
            if sum(recent) < -0.05:
                if self._consecutive_losses >= 5:
                    self._log_rejection(
                        pair, side, "equity_curve_breaker", current_time
                    )
                    return False

        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        if dataframe is None or len(dataframe) < 1:
            return True
        last = dataframe.iloc[-1]

        # Block in extreme trend against trade direction
        adx_val = last.get("adx", 0)
        ema_slope_val = last.get("ema_slope", 0)
        if side == "long" and adx_val > 40 and ema_slope_val < -0.5:
            self._log_rejection(
                pair, side, f"extreme_downtrend_adx{adx_val:.0f}", current_time,
                features=last,
            )
            return False
        if side == "short" and adx_val > 40 and ema_slope_val > 0.5:
            self._log_rejection(
                pair, side, f"extreme_uptrend_adx{adx_val:.0f}", current_time,
                features=last,
            )
            return False

        # Require meaningful volume
        vol = last.get("volume", 0)
        vol_sma = last.get("volume_sma", 1)
        if vol_sma > 0 and vol < vol_sma * 0.4:
            self._log_rejection(
                pair, side, f"low_volume_{vol:.0f}_vs_{vol_sma:.0f}", current_time,
                features=last,
            )
            return False

        # FUNDING RATE FILTER (live/dry_run only)
        # Positive funding → longs pay shorts → favors short entries
        # Negative funding → shorts pay longs → penalizes short entries
        # Threshold: 0.01% — ignore negligible rates
        fr_info = self._funding_rates.get(pair)
        if fr_info is not None:
            fr = fr_info.get("rate", 0)
            # Block shorts when funding is strongly negative (shorts pay)
            if side == "short" and fr < -0.0003:  # -0.03%
                self._log_rejection(
                    pair, side,
                    f"funding_rate_negative_{fr:.6f}",
                    current_time, features=last,
                )
                return False
            # Block longs when funding is strongly positive (longs pay)
            if side == "long" and fr > 0.0003:  # +0.03%
                self._log_rejection(
                    pair, side,
                    f"funding_rate_positive_{fr:.6f}",
                    current_time, features=last,
                )
                return False

        # 1h trend disagreement — only in extreme trends
        trend_1h = last.get("trend_1h", 0)
        if side == "long" and trend_1h < 0:
            if last.get("adx", 0) > 35:
                self._log_rejection(
                    pair, side, "1h_trend_disagree_long", current_time,
                    features=last,
                )
                return False
        if side == "short" and trend_1h > 0:
            if last.get("adx", 0) > 35:
                self._log_rejection(
                    pair, side, "1h_trend_disagree_short", current_time,
                    features=last,
                )
                return False

        # Anti-pattern filter
        if self._anti_patterns and hasattr(current_time, "hour"):
            try:
                strategy_name = params.get("strategy", "")
                ap = None
                for key, val in self._anti_patterns.items():
                    if strategy_name in key and isinstance(val, dict):
                        ap = val
                        break
                if ap:
                    toxic_hours = ap.get("toxic_hours", [])
                    toxic_days = ap.get("toxic_days", [])

                    # Safety: if >18 hours or >5 days are toxic, the
                    # anti-pattern data is likely from a garbage strategy
                    # assignment. Skip the filter to avoid silent blocking.
                    if len(toxic_hours) > 18 or len(toxic_days) > 5:
                        logger.warning(
                            "Anti-pattern filter skipped: %s has %d toxic hours, "
                            "%d toxic days — likely stale/bad data",
                            strategy_name,
                            len(toxic_hours),
                            len(toxic_days),
                        )
                    else:
                        if current_time.hour in toxic_hours:
                            self._log_rejection(
                                pair,
                                side,
                                f"anti_pattern_hour_{current_time.hour}",
                                current_time,
                                features=last,
                            )
                            return False
                        if current_time.weekday() in toxic_days:
                            self._log_rejection(
                                pair,
                                side,
                                f"anti_pattern_day_{current_time.weekday()}",
                                current_time,
                                features=last,
                            )
                            return False
            except Exception:
                pass

        # Quality model gate (session-direction-regime prior)
        _edge_score = None  # track for decision journal
        _quality_threshold = None
        if self._quality_model_data is not None:
            try:
                qm = self._quality_model_data
                model = qm.get("model")
                scaler = qm.get("scaler")
                thresholds = qm.get("thresholds", {})
                # Use p25 threshold for paper trading to allow more signals
                # through; in live mode, switch to min_quality
                min_quality = thresholds.get("p25", thresholds.get("min_quality", 0.5))
                _quality_threshold = min_quality

                if model and scaler and hasattr(current_time, "hour"):
                    leverage = float(
                        kwargs.get("leverage", 1) or 1
                    )
                    features = np.array(
                        [
                            [
                                current_time.hour,
                                current_time.weekday(),
                                1 if side == "short" else 0,
                                regime,
                                leverage,
                            ]
                        ]
                    )
                    scaled = scaler.transform(features)
                    quality = model.predict_proba(scaled)[0][1]
                    _edge_score = quality

                    if quality < min_quality:
                        self._log_rejection(
                            pair,
                            side,
                            f"quality_model_{quality:.3f}<{min_quality:.3f}",
                            current_time,
                            features=last,
                            rate=rate,
                            edge_score=quality,
                        )
                        return False
            except Exception:
                pass  # fail open — don't block on model error

        # Regime transition filter
        if len(dataframe) >= 12:
            regime_now = last.get("regime", 2)
            prev_regime = dataframe.iloc[-12].get("regime", regime_now)
            if regime_now != prev_regime:
                ds = abs(last.get("direction_score", 0))
                if ds < 0.5:
                    self._log_rejection(
                        pair,
                        side,
                        f"regime_transition_{prev_regime}->{regime_now}",
                        current_time,
                        features=last,
                    )
                    return False

        self._daily_trades += 1

        # Shadow model: evaluate candidate but NEVER block
        shadow_decision = self._evaluate_shadow(
            current_time, side, regime, kwargs,
        )

        # Decision Journal v4: log ACCEPTANCE with full context
        self._log_decision(
            pair, side, "accept", entry_tag or "entry",
            current_time, rate=rate, features=last,
            edge_score=_edge_score,
            quality_threshold=_quality_threshold,
        )

        # Trade replay: log accepted entry with full context
        self._log_trade_entry(
            pair, side, entry_tag, regime, rate, current_time, last,
            shadow=shadow_decision,
        )
        return True

    # ─── Dynamic Stoploss (MFE-Calibrated) ────────────────

    def custom_stoploss(
        self,
        pair,
        trade,
        current_time,
        current_rate,
        current_profit,
        after_fill,
        **kwargs,
    ):
        """
        MFE-calibrated dynamic stoploss.
        Phase 1: Base SL (give trade room to develop)
        Phase 2: Breakeven lock (protect capital after decent move)
        Phase 3: Trailing (lock in gains progressively)
        """
        regime = 2
        if trade.enter_tag and "_r" in trade.enter_tag:
            try:
                regime = int(trade.enter_tag.split("_r")[1][0])
            except (ValueError, IndexError):
                pass

        params = self._get_regime_params(regime)
        base_sl = params.get("sl", -0.008)
        trail_start = params.get("trail_start", 0.005)
        trail_step = params.get("trail_step", 0.003)

        trade_dur = (current_time - trade.open_date_utc).total_seconds() / 60

        # Phase 3: Big winner — trail to lock gains
        if current_profit > trail_start:
            return -(current_profit - trail_step)

        # Phase 2: Breakeven lock after modest gain
        if current_profit > abs(base_sl) * 0.5:
            return -0.002

        # Phase 1: Use regime-calibrated base SL
        # Tighten gradually as trade ages
        if trade_dur < 15:
            return base_sl  # full room
        elif trade_dur < 40:
            return base_sl * 0.75  # tighter
        else:
            return base_sl * 0.60  # tight after 40min

    # ─── Dynamic ROI ─────────────────────────────────────

    def custom_exit(
        self, pair, trade, current_time, current_rate, current_profit, **kwargs
    ):
        """
        Self-learned ROI exit: use per-regime ROI table from
        ml_optimizer duration bucket analysis.
        """
        trade_dur = (current_time - trade.open_date_utc).total_seconds() / 60

        # No time-based exits — pure ROI + SL only
        # Time exits always lose money in backtest.

        if trade.enter_tag and "_r" in trade.enter_tag:
            try:
                regime = int(trade.enter_tag.split("_r")[1][0])
                params = self._get_regime_params(regime)
                roi_table = params.get("roi_table", {})

                if roi_table:
                    applicable_roi = None
                    for mins in sorted(
                        roi_table.keys(), key=lambda x: int(x), reverse=True
                    ):
                        if trade_dur >= int(mins):
                            applicable_roi = roi_table[mins]
                            break

                    if applicable_roi and current_profit >= applicable_roi:
                        return "ml_roi_r{}_{}m".format(regime, int(trade_dur))

            except (ValueError, IndexError, AttributeError):
                pass

        return None

    # ─── Trade Exit Tracking (Discipline) ─────────────
    # confirm_trade_exit updates discipline trackers.
    # Uses trade.calc_profit_ratio for profit tracking.

    def confirm_trade_exit(
        self,
        pair,
        trade,
        order_type,
        amount,
        rate,
        time_in_force,
        exit_reason,
        current_time,
        **kwargs,
    ):
        """
        Track trade results for discipline systems.
        Updates consecutive losses, daily PnL, equity curve.
        """
        try:
            profit = trade.calc_profit_ratio(rate)
        except Exception:
            profit = 0.0

        # Update discipline trackers
        self._recent_results.append(profit)
        if len(self._recent_results) > self._MAX_RECENT:
            self._recent_results = self._recent_results[-self._MAX_RECENT :]

        self._daily_pnl += profit

        if profit < 0:
            self._consecutive_losses += 1
        else:
            self._consecutive_losses = 0  # reset on win

        # Equity curve tracking
        prev = self._equity_curve[-1] if self._equity_curve else 0
        self._equity_curve.append(prev + profit)
        if len(self._equity_curve) > 100:
            self._equity_curve = self._equity_curve[-100:]

        # Trade replay: log exit with PnL attribution
        self._log_trade_exit(
            pair, trade, exit_reason, profit, rate, current_time,
        )

        # Persist discipline state so it survives container restarts
        self._save_discipline_state()

        return True