models/predictor.py

"""
ML predictor for stock direction and price target.

Uses:
  - RandomForest + XGBoost **ensemble** → BUY / HOLD / SELL signal
  - RandomForestRegressor + XGBRegressor → price target 5 trading days ahead
  - Optuna hyperparameter tuning (optional, first train or manual trigger)

Survey-driven upgrade (2026-04-19):
  - XGBoost ensemble: average RF + XGB probabilities → +8-12% accuracy
  - Dynamic buy/sell thresholds based on volatility regime
  - Walk-forward friendly (precomputed_features param)

NOT financial advice — for educational and research purposes only.
"""

import logging
import os
from pathlib import Path
from typing import Optional

import joblib
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.model_selection import train_test_split

# XGBoost (optional — graceful fallback if not installed)
try:
    from xgboost import XGBClassifier, XGBRegressor
    _HAS_XGBOOST = True
except ImportError:
    _HAS_XGBOOST = False

# LightGBM (optional — faster training, better on tabular data)
try:
    from lightgbm import LGBMClassifier, LGBMRegressor
    _HAS_LGBM = True
except ImportError:
    _HAS_LGBM = False

# CatBoost (optional — strong on imbalanced tabular data)
try:
    from catboost import CatBoostClassifier, CatBoostRegressor
    _HAS_CATBOOST = True
except ImportError:
    _HAS_CATBOOST = False

# Optuna (optional — used for one-time hyperparameter tuning)
try:
    import optuna
    optuna.logging.set_verbosity(optuna.logging.WARNING)
    _HAS_OPTUNA = True
except ImportError:
    _HAS_OPTUNA = False
from sklearn.preprocessing import StandardScaler

from models.multi_factor_overlay import apply_multi_factor_overlay
from models.oldwang_trading_strategy import (
    apply_oldwang_strategy_overlay,
    build_oldwang_strategy_context,
)

try:
    from statsmodels.stats.outliers_influence import variance_inflation_factor as _vif
    _HAS_STATSMODELS = True
except ImportError:
    _HAS_STATSMODELS = False

logger = logging.getLogger(__name__)

_CATBOOST_CLF_PARAMS = {
    "iterations": 300,
    "depth": 6,
    "learning_rate": 0.05,
    "loss_function": "MultiClass",
    "eval_metric": "Accuracy",
    "random_seed": 42,
    "verbose": 0,
    "allow_writing_files": False,
    "class_weights": None,  # set dynamically based on class imbalance
}

# ---------------------------------------------------------------------------
# Unified signal thresholds (shared with routers/stock.py)
# ---------------------------------------------------------------------------

THRESHOLD_BUY   = 0.62   # default BUY threshold
THRESHOLD_SELL  = 0.35   # default SELL threshold
CONFIDENCE_HIGH = 0.65
CONFIDENCE_MED  = 0.52

_CONFIDENCE_THRESHOLDS = {"BUY": 0.52, "SELL": 0.58}

OLDWANG_WEIGHT_OVERLAY_ENABLED = os.getenv("ENABLE_OLDWANG_WEIGHT_OVERLAY", "1") != "0"
OLDWANG_WEIGHT_OVERLAY = {
    "buy_bull_weight": 0.04,
    "buy_bear_weight": -0.01,
    "sell_bear_weight": 0.0,
    "sell_bull_weight": 0.0,
    "buy_bear_gate": 3.0,
    "sell_bull_gate": None,
    "hold_bias": 0.0,
    "validation_source": "docs/validation_runs/factor_weight_grid_search_top40_20260522.json",
    "validation_accuracy_delta_pp": 2.9167,
    "validation_buy_precision_delta_pp": 0.8132,
    "validation_stock_count": 40,
}

# Adaptive thresholds by volatility regime
# (vol_low_upper, vol_high_lower) define regime boundaries
VOLATILITY_LOW_UPPER  = 0.015   # daily return std <= 1.5% → low vol
VOLATILITY_HIGH_LOWER = 0.030   # daily return std >= 3.0% → high vol

THRESHOLDS_BY_REGIME = {
    "low":    {"buy": 0.58, "sell": 0.40},   # 低波動：收緊，更容易觸發信號
    "medium": {"buy": 0.62, "sell": 0.35},   # 中波動：維持預設
    "high":   {"buy": 0.68, "sell": 0.28},   # 高波動：放寬，減少假信號
}


def _get_volatility_regime(volatility_20d: float) -> str:
    """Classify volatility into low / medium / high regime."""
    if np.isnan(volatility_20d):
        return "medium"
    if volatility_20d <= VOLATILITY_LOW_UPPER:
        return "low"
    elif volatility_20d >= VOLATILITY_HIGH_LOWER:
        return "high"
    return "medium"


def _adaptive_thresholds(volatility_20d: float) -> tuple[float, float]:
    """Return (buy_threshold, sell_threshold) adapted to volatility regime."""
    regime = _get_volatility_regime(volatility_20d)
    t = THRESHOLDS_BY_REGIME[regime]
    return t["buy"], t["sell"]


def _decide_signal(buy_prob: float, sell_prob: float, buy_threshold: float, sell_threshold: float) -> tuple[str, float]:
    """Map buy/sell probabilities to BUY/SELL/HOLD and its conviction score."""
    if buy_prob >= buy_threshold:
        return "BUY", buy_prob
    if sell_prob >= (1 - sell_threshold):
        return "SELL", sell_prob
    return "HOLD", max(buy_prob, sell_prob)


def _latest_float(frame: pd.DataFrame | pd.Series | None, column: str, default: float = np.nan) -> float:
    """Read a numeric latest value from a DataFrame/Series without raising."""
    try:
        if frame is None:
            return default
        if isinstance(frame, pd.Series):
            value = frame.get(column, default)
        else:
            if column not in frame.columns or frame.empty:
                return default
            value = frame[column].iloc[-1]
        value = float(value)
        return value if np.isfinite(value) else default
    except Exception:
        return default


def _rolling_or_column(df: pd.DataFrame, column: str, window: int) -> pd.Series:
    if column in df.columns:
        series = pd.to_numeric(df[column], errors="coerce")
        if not series.dropna().empty:
            return series
    return pd.to_numeric(df["close"], errors="coerce").rolling(window, min_periods=max(2, window // 2)).mean()


def build_trend_conflict_snapshot(
    df: pd.DataFrame,
    features: pd.DataFrame | pd.Series | None = None,
) -> dict:
    """Summarize short/mid-term trend risk for post-model signal gating."""
    if df is None or df.empty or "close" not in df.columns:
        return {"available": False, "reason": "missing_price_history"}

    frame = df.copy()
    close_series = pd.to_numeric(frame["close"], errors="coerce")
    if close_series.dropna().empty:
        return {"available": False, "reason": "missing_close"}

    close = float(close_series.iloc[-1])
    ma5 = _latest_float(frame.assign(_ma5=_rolling_or_column(frame, "ma5", 5)), "_ma5")
    ma20 = _latest_float(frame.assign(_ma20=_rolling_or_column(frame, "ma20", 20)), "_ma20")
    ma60 = _latest_float(frame.assign(_ma60=_rolling_or_column(frame, "ma60", 60)), "_ma60")

    feat_latest = None
    if isinstance(features, pd.DataFrame) and not features.dropna(how="all").empty:
        feat_latest = features.dropna(how="all").iloc[-1]
    elif isinstance(features, pd.Series):
        feat_latest = features

    rsi = _latest_float(feat_latest, "rsi", _latest_float(frame, "rsi"))
    macd_hist = _latest_float(feat_latest, "macd_hist", _latest_float(frame, "macd_hist"))

    def pct_change(periods: int) -> float:
        if len(close_series) <= periods:
            return np.nan
        ref = float(close_series.iloc[-periods - 1])
        if not np.isfinite(ref) or ref == 0:
            return np.nan
        return close / ref - 1.0

    return_5d = _latest_float(feat_latest, "return_5d", pct_change(5))
    return_20d = _latest_float(feat_latest, "return_20d", pct_change(20))
    high_20d = float(pd.to_numeric(frame.get("high", close_series), errors="coerce").tail(20).max())
    low_20d = float(pd.to_numeric(frame.get("low", close_series), errors="coerce").tail(20).min())

    bearish_flags = {
        "close_below_ma5": np.isfinite(ma5) and close < ma5,
        "close_below_ma20": np.isfinite(ma20) and close < ma20,
        "close_below_ma60": np.isfinite(ma60) and close < ma60,
        "ma5_below_ma20": np.isfinite(ma5) and np.isfinite(ma20) and ma5 < ma20,
        "rsi_below_45": np.isfinite(rsi) and rsi < 45,
        "macd_hist_negative": np.isfinite(macd_hist) and macd_hist < 0,
        "near_20d_low": np.isfinite(low_20d) and low_20d > 0 and close <= low_20d * 1.02,
        "sharp_5d_drop": np.isfinite(return_5d) and return_5d <= -0.08,
        "sharp_20d_drop": np.isfinite(return_20d) and return_20d <= -0.10,
    }
    bullish_flags = {
        "close_above_ma5": np.isfinite(ma5) and close > ma5,
        "close_above_ma20": np.isfinite(ma20) and close > ma20,
        "close_above_ma60": np.isfinite(ma60) and close > ma60,
        "ma5_above_ma20": np.isfinite(ma5) and np.isfinite(ma20) and ma5 > ma20,
        "rsi_above_55": np.isfinite(rsi) and rsi > 55,
        "macd_hist_positive": np.isfinite(macd_hist) and macd_hist > 0,
        "near_20d_high": np.isfinite(high_20d) and high_20d > 0 and close >= high_20d * 0.98,
        "sharp_5d_rise": np.isfinite(return_5d) and return_5d >= 0.08,
        "sharp_20d_rise": np.isfinite(return_20d) and return_20d >= 0.10,
    }
    bearish_count = int(sum(bool(v) for v in bearish_flags.values()))
    bullish_count = int(sum(bool(v) for v in bullish_flags.values()))
    severe_bearish = bearish_count >= 6 and bearish_flags["close_below_ma5"] and (
        bearish_flags["sharp_5d_drop"]
        or bearish_flags["sharp_20d_drop"]
        or (
            bearish_flags["near_20d_low"]
            and (np.isfinite(return_5d) and return_5d < -0.04)
        )
    )
    severe_bullish = bullish_count >= 6 and bullish_flags["close_above_ma5"] and (
        bullish_flags["sharp_5d_rise"]
        or bullish_flags["sharp_20d_rise"]
        or (
            bullish_flags["near_20d_high"]
            and (np.isfinite(return_5d) and return_5d > 0.04)
        )
    )

    def _round_or_none(value: float, digits: int = 4):
        return round(float(value), digits) if np.isfinite(value) else None

    return {
        "available": True,
        "close": _round_or_none(close, 4),
        "ma5": _round_or_none(ma5, 4),
        "ma20": _round_or_none(ma20, 4),
        "ma60": _round_or_none(ma60, 4),
        "return_5d": _round_or_none(return_5d, 4),
        "return_20d": _round_or_none(return_20d, 4),
        "rsi": _round_or_none(rsi, 2),
        "macd_hist": _round_or_none(macd_hist, 4),
        "high_20d": _round_or_none(high_20d, 4),
        "low_20d": _round_or_none(low_20d, 4),
        "bearish_count": bearish_count,
        "bullish_count": bullish_count,
        "bearish_flags": bearish_flags,
        "bullish_flags": bullish_flags,
        "severe_bearish": bool(severe_bearish),
        "severe_bullish": bool(severe_bullish),
    }


def apply_trend_conflict_guard(
    *,
    signal: str,
    buy_prob: float,
    sell_prob: float,
    df: pd.DataFrame,
    features: pd.DataFrame | pd.Series | None = None,
) -> tuple[str, dict]:
    """Block high-risk contrarian BUY/SELL labels when recent trend is severe."""
    original = str(signal or "HOLD").upper()
    snapshot = build_trend_conflict_snapshot(df, features)
    guard = {
        "applied": False,
        "reason": None,
        "original_signal": original,
        "final_signal": original,
        "buy_probability": round(float(buy_prob), 4),
        "sell_probability": round(float(sell_prob), 4),
        "snapshot": snapshot,
    }
    if not snapshot.get("available"):
        return original, guard

    final_signal = original
    if original == "BUY" and snapshot.get("severe_bearish"):
        final_signal = "HOLD"
        guard["reason"] = "severe_bearish_trend_blocks_buy"
    elif original == "SELL" and snapshot.get("severe_bullish"):
        final_signal = "HOLD"
        guard["reason"] = "severe_bullish_trend_blocks_sell"

    guard["applied"] = final_signal != original
    guard["final_signal"] = final_signal
    return final_signal, guard


def _apply_oldwang_weight_overlay(
    *,
    buy_prob: float,
    sell_prob: float,
    oldwang_context: dict,
    enabled: bool = OLDWANG_WEIGHT_OVERLAY_ENABLED,
    config: dict | None = None,
) -> tuple[float, float, dict]:
    """Apply validated Old Wang bull/bear score weights to probabilities."""
    cfg = config or OLDWANG_WEIGHT_OVERLAY
    if not enabled:
        return buy_prob, sell_prob, {"applied": False, "reason": "disabled"}

    try:
        bull_score = float(oldwang_context.get("bull_score", 0.0) or 0.0)
        bear_score = float(oldwang_context.get("bear_score", 0.0) or 0.0)
    except (TypeError, ValueError):
        bull_score = 0.0
        bear_score = 0.0

    adjusted_buy = (
        buy_prob
        + float(cfg.get("buy_bull_weight", 0.0) or 0.0) * bull_score
        + float(cfg.get("buy_bear_weight", 0.0) or 0.0) * bear_score
    )
    adjusted_sell = (
        sell_prob
        + float(cfg.get("sell_bear_weight", 0.0) or 0.0) * bear_score
        + float(cfg.get("sell_bull_weight", 0.0) or 0.0) * bull_score
    )

    buy_bear_gate = cfg.get("buy_bear_gate")
    if buy_bear_gate is not None and bear_score >= float(buy_bear_gate):
        adjusted_buy = 0.0
    sell_bull_gate = cfg.get("sell_bull_gate")
    if sell_bull_gate is not None and bull_score >= float(sell_bull_gate):
        adjusted_sell = 0.0

    adjusted_buy = float(np.clip(adjusted_buy, 0.0, 1.0))
    adjusted_sell = float(np.clip(adjusted_sell, 0.0, 1.0))
    return adjusted_buy, adjusted_sell, {
        "applied": True,
        "config": {k: v for k, v in cfg.items() if k.endswith("_weight") or k.endswith("_gate") or k == "hold_bias"},
        "bull_score": round(bull_score, 4),
        "bear_score": round(bear_score, 4),
        "buy_probability_before": round(float(buy_prob), 4),
        "buy_probability_after": round(adjusted_buy, 4),
        "sell_probability_before": round(float(sell_prob), 4),
        "sell_probability_after": round(adjusted_sell, 4),
        "validation_source": cfg.get("validation_source"),
        "validation_accuracy_delta_pp": cfg.get("validation_accuracy_delta_pp"),
        "validation_buy_precision_delta_pp": cfg.get("validation_buy_precision_delta_pp"),
        "validation_stock_count": cfg.get("validation_stock_count"),
    }


def _as_lgbm_feature_frame(X_scaled: np.ndarray) -> pd.DataFrame:
    """Keep stable feature names for LightGBM after scaling."""
    return pd.DataFrame(X_scaled, columns=FEATURE_COLUMNS)


def _detect_regime(df: pd.DataFrame) -> pd.DataFrame:
    """
    Fit 3-state GaussianHMM on cross-asset features to tag market regime.
    Adds: hmm_regime (0=bear,1=chop,2=bull), hmm_regime_bull, hmm_regime_bear.
    Gracefully falls back to zeros if hmmlearn is unavailable or data is thin.
    """
    try:
        from hmmlearn import hmm as _hmm
        obs_cols = [c for c in ["taiex_return_5d", "volatility_20d", "taiex_ma20_ratio"]
                    if c in df.columns]
        if len(obs_cols) < 2:
            raise ValueError("insufficient obs columns")
        X = df[obs_cols].fillna(0).values.astype(float)
        model = _hmm.GaussianHMM(n_components=3, covariance_type="diag",
                                  n_iter=100, random_state=42)
        model.fit(X)
        raw = model.predict(X)
        means = [X[raw == s, 0].mean() if (raw == s).sum() > 0 else 0.0 for s in range(3)]
        order = np.argsort(means)
        remap = {order[0]: 0, order[1]: 1, order[2]: 2}
        labeled = np.array([remap[r] for r in raw])
        df = df.copy()
        df["hmm_regime"]      = labeled.astype(float)
        df["hmm_regime_bull"] = (labeled == 2).astype(float)
        df["hmm_regime_bear"] = (labeled == 0).astype(float)
    except Exception:
        df = df.copy()
        for col in ("hmm_regime", "hmm_regime_bull", "hmm_regime_bear"):
            df[col] = 0.0
    return df

# ---------------------------------------------------------------------------
# Feature list (must stay in sync between train and predict)
# ---------------------------------------------------------------------------

FEATURE_VERSION = "institutional-macd-oldwang-v2"

FEATURE_COLUMNS = [
    # Momentum / returns
    "return_1d",
    "return_5d",
    "return_10d",
    "return_20d",
    # MA ratios
    "close_ma5_ratio",
    "close_ma20_ratio",
    "ma5_ma20_ratio",
    "ma20_ma60_ratio",
    # RSI
    "rsi",
    # MACD
    "macd_hist",
    "macd_signal_ratio",
    "macd_hist_norm",
    "macd_hist_delta_1d",
    "macd_hist_slope_3d",
    # macd_cross_up / macd_cross_down removed: SHAP < 0.001 across all 5 stocks
    "macd_above_zero",
    # Bollinger
    "bb_pct_b",
    # Stochastic
    "k",
    "d",
    # Volume
    "volume_ratio",
    # Volatility / range
    "atr_ratio",
    "high_low_ratio",
    # OBV momentum
    "obv_trend",
    # Extended MA ratio (半年線)
    "close_ma60_ratio",
    # Volume (log-scaled)
    "log_volume_ratio",
    # Sentiment proxy
    "volume_zscore",
    # price_volume_div removed: SHAP < 0.001 across all 5 stocks
    # Volatility regime
    "volatility_20d",
    # Cross-asset (TAIEX + USD/TWD)
    "taiex_return_5d",
    "taiex_ma20_ratio",
    "usdtwd_return_5d",
    # Old Wang style technical/rule context (C21 dry-run: aggregate accuracy non-decrease)
    "oldwang_triple_bull",
    "oldwang_triple_bear",
    "oldwang_ma5_hold",
    "oldwang_trust_ma10_guard",
    "oldwang_trust_ma10_broken",
    "oldwang_foreign_ma20_guard",
    "oldwang_foreign_ma20_broken",
    "oldwang_volume_spike",
    "oldwang_volume_high_break",
    "oldwang_volume_low_guard",
    "oldwang_volume_low_break",
    "oldwang_gap_guard",
    "oldwang_gap_filled",
    "oldwang_bull_score",
    "oldwang_bear_score",
    # Institutional flow features removed: SHAP < 0.001 across all 5 stocks
    # (foreign_net_vol_ratio, trust_net_vol_ratio, dealer_net_vol_ratio,
    #  institutional_net_vol_ratio, institutional_5d_net_vol_ratio,
    #  institutional_20d_zscore, foreign_trust_alignment, institutional_streak)
    # Sector peer return features (C30)
    "peer_ret_1d",
    "peer_ret_5d",
]

# 跨資產特徵(可選,有資料就加,沒有就用 NaN → impute)
CROSS_ASSET_COLUMNS = [
    "vix_level",
    "vix_change_5d",
]

# 台股專用跨資產
TW_CROSS_ASSET_COLUMNS = [
    "taiex_return_5d",
    "taiex_ma20_ratio",
    "usdtwd_return_5d",
]

# 美股專用跨資產
US_CROSS_ASSET_COLUMNS = [
    "tnx_level",
    "tnx_change_5d",
    "dxy_return_5d",
]


VIF_THRESHOLD = 10.0  # VIF > 10 indicates high multicollinearity


def _compute_vif(X: np.ndarray, feature_names: list[str]) -> list[tuple[str, float]]:
    """Compute Variance Inflation Factor for each feature column."""
    if not _HAS_STATSMODELS:
        return []
    # Add constant column for intercept
    X_with_const = np.column_stack([np.ones(X.shape[0]), X])
    vif_values = []
    for i in range(X.shape[1]):
        # offset by 1 because column 0 is the constant
        vif_val = _vif(X_with_const, i + 1)
        vif_values.append((feature_names[i], round(float(vif_val), 2)))
    return vif_values


def _log_vif_warnings(X: np.ndarray, feature_names: list[str]) -> None:
    """Log features with VIF above threshold."""
    vif_values = _compute_vif(X, feature_names)
    if not vif_values:
        return
    high_vif = [(name, val) for name, val in vif_values if val > VIF_THRESHOLD]
    if high_vif:
        logger.warning(
            "High VIF (multicollinearity) detected: %s",
            high_vif[:5],
        )


def _rolling_recent_flag(flag: pd.Series, window: int) -> pd.Series:
    """Return 1.0 when a boolean event happened inside the recent window."""
    return flag.astype(float).rolling(window, min_periods=1).max().fillna(0.0)


def _last_event_level(level: pd.Series, event: pd.Series) -> pd.Series:
    """Forward-fill the prior level from the latest completed event candle."""
    return level.where(event).shift(1).ffill()


def _true_range(df: pd.DataFrame) -> pd.Series:
    high = df["high"].astype(float)
    low = df["low"].astype(float)
    close = df["close"].astype(float)
    prev_close = close.shift(1)
    return pd.concat(
        [
            high - low,
            (high - prev_close).abs(),
            (low - prev_close).abs(),
        ],
        axis=1,
    ).max(axis=1)


def _zscore_prior(series: pd.Series, window: int, min_periods: int) -> pd.Series:
    """Rolling z-score using only prior rows to avoid lookahead."""
    mean = series.rolling(window, min_periods=min_periods).mean().shift(1)
    std = series.rolling(window, min_periods=min_periods).std().shift(1).replace(0, np.nan)
    return ((series - mean) / std).replace([np.inf, -np.inf], np.nan).fillna(0.0)


def _round_level(value) -> float | None:
    if value is None or pd.isna(value):
        return None
    return round(float(value), 2)


def _build_oldwang_context(df: pd.DataFrame, features: pd.DataFrame) -> dict:
    """Build a compact strategy explanation from the latest Old Wang features."""
    if features.empty:
        return {
            "bull_score": 0.0,
            "bear_score": 0.0,
            "bull_reasons": [],
            "risk_reasons": [],
            "key_levels": {},
        }

    latest = features.iloc[-1]

    def active(name: str) -> bool:
        return float(latest.get(name, 0.0) or 0.0) >= 0.5

    bull_reasons: list[str] = []
    risk_reasons: list[str] = []

    if active("oldwang_triple_bull"):
        bull_reasons.append("三陽開泰：收盤站上 5/10/20MA 且均線上揚")
    if active("oldwang_ma5_hold"):
        bull_reasons.append("站穩 5 日線")
    if active("oldwang_trust_ma10_guard"):
        bull_reasons.append("投信 5 日買超且守住 10 日線")
    if active("oldwang_foreign_ma20_guard"):
        bull_reasons.append("外資 5 日買超且守住 20 日線")
    if active("oldwang_volume_high_break"):
        bull_reasons.append("突破爆大量 K 棒高點")
    if active("oldwang_gap_guard"):
        bull_reasons.append("跳空缺口守住")

    if active("oldwang_triple_bear"):
        risk_reasons.append("三聲無奈：收盤跌破 5/10/20MA 且均線下彎")
    if active("oldwang_trust_ma10_broken"):
        risk_reasons.append("投信買超但跌破 10 日線")
    if active("oldwang_foreign_ma20_broken"):
        risk_reasons.append("外資買超但跌破 20 日線")
    if active("oldwang_volume_low_break"):
        risk_reasons.append("跌破爆大量 K 棒低點")
    if active("oldwang_gap_filled"):
        risk_reasons.append("跳空缺口回補")

    close = df["close"].astype(float)
    high = df.get("high", close).astype(float)
    low = df.get("low", close).astype(float)
    open_ = df.get("open", close).astype(float)
    volume = df.get("volume", pd.Series(0.0, index=df.index)).astype(float)

    log_volume = np.log1p(volume.clip(lower=0.0))
    volume_spike = _zscore_prior(log_volume, 20, 10) >= 2.0
    volume_spike_high = _last_event_level(high, volume_spike)
    volume_spike_low = _last_event_level(low, volume_spike)

    tr = _true_range(df)
    atr = df.get("atr", tr.rolling(14, min_periods=5).mean()).astype(float)
    gap_up = open_ > high.shift(1)
    gap_support = _last_event_level(high.shift(1), gap_up)

    key_levels = {
        "ma5": _round_level(df.get("ma5", close.rolling(5, min_periods=3).mean()).iloc[-1]),
        "ma10": _round_level(df.get("ma10", close.rolling(10, min_periods=5).mean()).iloc[-1]),
        "ma20": _round_level(df.get("ma20", close.rolling(20, min_periods=10).mean()).iloc[-1]),
        "volume_spike_high": _round_level(volume_spike_high.iloc[-1]),
        "volume_spike_low": _round_level(volume_spike_low.iloc[-1]),
        "gap_support": _round_level(gap_support.iloc[-1]),
        "atr": _round_level(atr.iloc[-1]),
    }

    return {
        "bull_score": round(float(latest.get("oldwang_bull_score", 0.0) or 0.0), 2),
        "bear_score": round(float(latest.get("oldwang_bear_score", 0.0) or 0.0), 2),
        "bull_reasons": bull_reasons,
        "risk_reasons": risk_reasons,
        "key_levels": key_levels,
    }


def _build_features(df: pd.DataFrame) -> pd.DataFrame:
    """
    Compute ML feature columns from an indicator-enriched OHLCV DataFrame.
    Returns a new DataFrame aligned with `df` index.
    """
    feat = pd.DataFrame(index=df.index)

    # Return features
    feat["return_1d"] = df["close"].pct_change(1)
    feat["return_5d"] = df["close"].pct_change(5)
    feat["return_10d"] = df["close"].pct_change(10)
    feat["return_20d"] = df["close"].pct_change(20)

    # MA ratios (avoid division by zero)
    # When ma60 is all-NaN (< 60 rows), fall back to ma20 to avoid empty features
    ma60 = df.get("ma60", pd.Series(np.nan, index=df.index))
    if ma60.isna().all():
        ma60 = df.get("ma20", pd.Series(np.nan, index=df.index))
    feat["close_ma5_ratio"] = df["close"] / df["ma5"].replace(0, np.nan)
    feat["close_ma20_ratio"] = df["close"] / df["ma20"].replace(0, np.nan)
    feat["ma5_ma20_ratio"] = df["ma5"] / df["ma20"].replace(0, np.nan)
    feat["ma20_ma60_ratio"] = df["ma20"] / ma60.replace(0, np.nan)

    # RSI
    feat["rsi"] = df.get("rsi", pd.Series(np.nan, index=df.index))

    # MACD
    macd = df.get("macd", pd.Series(np.nan, index=df.index))
    macd_signal = df.get("macd_signal", pd.Series(np.nan, index=df.index))
    macd_hist = df.get("macd_hist", pd.Series(np.nan, index=df.index))
    feat["macd_hist"] = macd_hist
    feat["macd_signal_ratio"] = macd / macd_signal.replace(0, np.nan)
    feat["macd_hist_norm"] = macd_hist / df["close"].replace(0, np.nan)
    feat["macd_hist_delta_1d"] = macd_hist.diff(1)
    feat["macd_hist_slope_3d"] = macd_hist.diff(3) / 3
    prev_macd = macd.shift(1)
    prev_signal = macd_signal.shift(1)
    feat["macd_cross_up"] = ((macd > macd_signal) & (prev_macd <= prev_signal)).astype(float)
    feat["macd_cross_down"] = ((macd < macd_signal) & (prev_macd >= prev_signal)).astype(float)
    feat["macd_above_zero"] = (macd > 0).astype(float)

    # Bollinger
    feat["bb_pct_b"] = df.get("bb_pct_b", pd.Series(np.nan, index=df.index))

    # Stochastic
    feat["k"] = df.get("k", pd.Series(np.nan, index=df.index))
    feat["d"] = df.get("d", pd.Series(np.nan, index=df.index))

    # Volume
    feat["volume_ratio"] = df.get("volume_ratio", pd.Series(np.nan, index=df.index))

    # Volatility / range
    feat["atr_ratio"] = df.get("atr_ratio", pd.Series(np.nan, index=df.index))
    feat["high_low_ratio"] = (df["high"] - df["low"]) / df["close"].replace(0, np.nan)

    # OBV momentum
    feat["obv_trend"] = df.get("obv_trend", pd.Series(np.nan, index=df.index))

    # Extended MA ratios (close vs 60/120/240-day MA; fallback to ma20/ma60 if insufficient data)
    feat["close_ma60_ratio"] = df["close"] / ma60.replace(0, np.nan)
    ma120 = df.get("ma120", pd.Series(np.nan, index=df.index))
    if ma120.isna().all():
        ma120 = ma60
    feat["close_ma120_ratio"] = df["close"] / ma120.replace(0, np.nan)
    ma240 = df.get("ma240", pd.Series(np.nan, index=df.index))
    if ma240.isna().all():
        ma240 = ma120
    feat["close_ma240_ratio"] = df["close"] / ma240.replace(0, np.nan)

    # MA cross signals (golden/death cross — explicit binary event feature)
    nan_col = pd.Series(np.nan, index=df.index)
    feat["golden_cross_5_20"] = df.get("golden_cross_5_20", nan_col).fillna(0.0)
    feat["death_cross_5_20"] = df.get("death_cross_5_20", nan_col).fillna(0.0)
    feat["golden_cross_20_60"] = df.get("golden_cross_20_60", nan_col).fillna(0.0)
    feat["death_cross_20_60"] = df.get("death_cross_20_60", nan_col).fillna(0.0)

    # Log-scaled volume ratio (reduces skew)
    vr = df.get("volume_ratio", pd.Series(np.nan, index=df.index))
    feat["log_volume_ratio"] = np.log1p(vr.clip(lower=0))

    # ---- Sentiment proxy(不需 API,純 price/volume 計算)----
    # volume z-score:20日滾動 z-score,> 2 表示異常放量
    vol = df.get("volume", pd.Series(0, index=df.index)).astype(float)
    vol_mean = vol.rolling(20).mean()
    vol_std = vol.rolling(20).std().replace(0, np.nan)
    feat["volume_zscore"] = (vol - vol_mean) / vol_std

    # 價量背離:5 日 return 方向 vs 5 日 volume 變化方向
    # +1 = 同向(正常), -1 = 背離(異常,可能反轉)
    price_dir = np.sign(df["close"].pct_change(5))
    vol_dir = np.sign(vol.pct_change(5))
    feat["price_volume_div"] = price_dir * vol_dir  # -1 = divergence

    # Rolling volatility regime flag
    feat["volatility_20d"] = df.get("volatility_20d", pd.Series(np.nan, index=df.index))

    # Cross-asset features (Taiwan: TAIEX + USD/TWD + SOX + TNX)
    # Forward-fill then back-fill to handle early NaN from rolling.
    # If column absent entirely, fall back to neutral value.
    _CROSS_ASSET_NEUTRAL = {
        "taiex_return_5d": 0.0,
        "taiex_ma20_ratio": 1.0,
        "usdtwd_return_5d": 0.0,
    }
    for col, neutral in _CROSS_ASSET_NEUTRAL.items():
        series = df.get(col, pd.Series(np.nan, index=df.index))
        feat[col] = series.ffill().bfill().fillna(neutral)

    for col in ("sox_ret_1d", "sox_ret_5d", "sox_ma20_ratio", "tnx_level", "tnx_change_5d"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # Institutional investor flow features (Taiwan only). Public data is EOD;
    # missing rows are neutral-imputed upstream, with the latest row optionally
    # carrying forward the most recent published trading day.
    volume = df.get("volume", pd.Series(np.nan, index=df.index)).astype(float).replace(0, np.nan)
    foreign_net = df.get("foreign_net", pd.Series(0, index=df.index)).astype(float)
    trust_net = df.get("trust_net", pd.Series(0, index=df.index)).astype(float)
    dealer_net = df.get("dealer_net", pd.Series(0, index=df.index)).astype(float)
    institutional_net = df.get("institutional_net", pd.Series(0, index=df.index)).astype(float)

    feat["foreign_net_vol_ratio"] = (foreign_net / volume).fillna(0.0)
    feat["trust_net_vol_ratio"] = (trust_net / volume).fillna(0.0)
    feat["dealer_net_vol_ratio"] = (dealer_net / volume).fillna(0.0)
    inst_ratio = (institutional_net / volume).fillna(0.0)
    feat["institutional_net_vol_ratio"] = inst_ratio

    rolling_net = institutional_net.rolling(5, min_periods=1).sum()
    rolling_volume = volume.rolling(5, min_periods=1).sum().replace(0, np.nan)
    feat["institutional_5d_net_vol_ratio"] = (rolling_net / rolling_volume).fillna(0.0)

    inst_mean = inst_ratio.rolling(20, min_periods=10).mean()
    inst_std = inst_ratio.rolling(20, min_periods=10).std().replace(0, np.nan)
    feat["institutional_20d_zscore"] = ((inst_ratio - inst_mean) / inst_std).fillna(0.0)

    foreign_sign = np.sign(foreign_net)
    trust_sign = np.sign(trust_net)
    feat["foreign_trust_alignment"] = np.where(
        (foreign_sign == trust_sign) & (foreign_sign != 0),
        foreign_sign,
        0,
    ).astype(float)

    inst_sign = np.sign(institutional_net)
    streak_group = inst_sign.ne(inst_sign.shift()).cumsum()
    streak = inst_sign.groupby(streak_group).cumcount().add(1).astype(float) * inst_sign
    feat["institutional_streak"] = streak.where(inst_sign != 0, 0.0)

    # Old Wang style rules as learnable context features only. They are not hard
    # buy/sell gates because C21 showed aggregate improvement, not universal per-stock uplift.
    open_ = df.get("open", df["close"]).astype(float)
    high = df.get("high", df["close"]).astype(float)
    low = df.get("low", df["close"]).astype(float)
    close = df["close"].astype(float)
    ma5 = df.get("ma5", close.rolling(5, min_periods=3).mean()).astype(float)
    ma10 = df.get("ma10", close.rolling(10, min_periods=5).mean()).astype(float)
    ma20 = df.get("ma20", close.rolling(20, min_periods=10).mean()).astype(float)
    ma5_slope = ma5.pct_change(3).fillna(0.0)
    ma10_slope = ma10.pct_change(3).fillna(0.0)
    ma20_slope = ma20.pct_change(3).fillna(0.0)

    oldwang_triple_bull = (
        (close > ma5)
        & (ma5 > ma10)
        & (ma10 > ma20)
        & (ma5_slope > 0)
        & (ma10_slope > 0)
        & (ma20_slope > 0)
    )
    oldwang_triple_bear = (
        (close < ma5)
        & (ma5 < ma10)
        & (ma10 < ma20)
        & (ma5_slope < 0)
        & (ma10_slope < 0)
        & (ma20_slope < 0)
    )
    feat["oldwang_triple_bull"] = oldwang_triple_bull.astype(float)
    feat["oldwang_triple_bear"] = oldwang_triple_bear.astype(float)
    feat["oldwang_ma5_hold"] = (close >= ma5).astype(float)

    trust_5d = trust_net.rolling(5, min_periods=1).sum()
    foreign_5d = foreign_net.rolling(5, min_periods=1).sum()
    feat["oldwang_trust_ma10_guard"] = ((trust_5d > 0) & (close >= ma10)).astype(float)
    feat["oldwang_trust_ma10_broken"] = ((trust_5d > 0) & (close < ma10)).astype(float)
    feat["oldwang_foreign_ma20_guard"] = ((foreign_5d > 0) & (close >= ma20)).astype(float)
    feat["oldwang_foreign_ma20_broken"] = ((foreign_5d > 0) & (close < ma20)).astype(float)

    raw_volume = df.get("volume", pd.Series(0.0, index=df.index)).astype(float)
    log_volume = np.log1p(raw_volume.clip(lower=0.0))
    oldwang_volume_z = _zscore_prior(log_volume, 20, 10)
    oldwang_volume_spike = oldwang_volume_z >= 2.0
    spike_high = _last_event_level(high, oldwang_volume_spike)
    spike_low = _last_event_level(low, oldwang_volume_spike)
    feat["oldwang_volume_spike"] = oldwang_volume_spike.astype(float)
    feat["oldwang_volume_high_break"] = ((close > spike_high) & spike_high.notna()).astype(float)
    feat["oldwang_volume_low_guard"] = ((close >= spike_low) & spike_low.notna()).astype(float)
    feat["oldwang_volume_low_break"] = ((close < spike_low) & spike_low.notna()).astype(float)

    tr = _true_range(df)
    atr = df.get("atr", tr.rolling(14, min_periods=5).mean()).astype(float)
    atr_prior = atr.shift(1).replace(0, np.nan).bfill().fillna(close * 0.02)
    gap_up = open_ > high.shift(1)
    gap_support = _last_event_level(high.shift(1), gap_up)
    recent_gap_up = _rolling_recent_flag(gap_up, 5) > 0
    feat["oldwang_gap_guard"] = (
        recent_gap_up & (low >= gap_support - 0.1 * atr_prior)
    ).astype(float)
    feat["oldwang_gap_filled"] = (recent_gap_up & (close < gap_support)).astype(float)

    feat["oldwang_bull_score"] = (
        feat["oldwang_triple_bull"]
        + feat["oldwang_ma5_hold"]
        + feat["oldwang_trust_ma10_guard"]
        + feat["oldwang_foreign_ma20_guard"]
        + feat["oldwang_volume_high_break"]
        + feat["oldwang_gap_guard"]
    ).fillna(0.0)
    feat["oldwang_bear_score"] = (
        feat["oldwang_triple_bear"]
        + feat["oldwang_trust_ma10_broken"]
        + feat["oldwang_foreign_ma20_broken"]
        + feat["oldwang_volume_low_break"]
        + feat["oldwang_gap_filled"]
    ).fillna(0.0)

    # Margin trading features (融資融券 — 老王's 籌碼 signals)
    margin_bal = df.get("margin_balance", pd.Series(0, index=df.index)).astype(float)
    short_bal  = df.get("short_balance",  pd.Series(0, index=df.index)).astype(float)
    margin_buy  = df.get("margin_buy",  pd.Series(0, index=df.index)).astype(float)
    margin_sell = df.get("margin_sell", pd.Series(0, index=df.index)).astype(float)

    # margin_ratio: leverage fraction relative to daily turnover (units: 1000 shares vs shares)
    feat["margin_ratio"] = (margin_bal * 1000 / volume).fillna(0.0).clip(upper=50)

    # 券資比: short / margin balance
    feat["short_margin_ratio"] = (short_bal / margin_bal.replace(0, np.nan)).fillna(0.0).clip(upper=5)

    # 5-day % change in margin balance (retail momentum)
    margin_prev5 = margin_bal.shift(5).replace(0, np.nan)
    feat["margin_5d_change_pct"] = ((margin_bal - margin_prev5) / margin_prev5).fillna(0.0).clip(-1, 1)

    # Direction of margin flow (1.0 = all buying, 0.5 = balanced, 0.0 = all selling)
    margin_total = (margin_buy + margin_sell).replace(0, np.nan)
    feat["margin_buy_pressure"] = (margin_buy / margin_total).fillna(0.5)

    # Squeeze signal: institutions net-buying AND retail adding margin leverage
    inst_buying = (foreign_net > 0).astype(float)
    margin_rising = (feat["margin_5d_change_pct"] > 0.05).astype(float)
    feat["chip_squeeze_signal"] = (inst_buying * margin_rising)

    # Composite chip concentration score (老王's 籌碼集中度 synthesis)
    # Weights: streak 40% | foreign/trust alignment 30% | z-score 20% | margin health 10%
    streak_norm = np.tanh(feat["institutional_streak"] / 5.0)
    zscore_norm = np.tanh(feat["institutional_20d_zscore"] / 2.0)
    margin_health = (-feat["margin_5d_change_pct"]).clip(-1.0, 1.0)
    feat["chip_score"] = (
        0.40 * streak_norm
        + 0.30 * feat["foreign_trust_alignment"]
        + 0.20 * zscore_norm
        + 0.10 * margin_health
    ).clip(-1.0, 1.0)

    # MA alignment (from technical.py — already in df if add_ma_bull_alignment was called)
    for col in ("ma_bull_alignment", "ma_bear_alignment"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0).astype(float)
    feat["ma_alignment_days"] = df.get("ma_alignment_days", pd.Series(0.0, index=df.index)).fillna(0.0).clip(-20, 20)

    # BIAS rates
    for col in ("bias_20", "bias_60"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0).clip(-0.3, 0.3)

    # Feature interactions
    macd = feat.get("macd_hist", pd.Series(0.0, index=feat.index)).fillna(0.0)
    rsi  = feat.get("rsi", pd.Series(50.0, index=feat.index)).fillna(50.0)
    feat["macd_rsi_interact"] = (macd * ((rsi - 50) / 50)).clip(-1, 1)
    vol_z = feat.get("volume_zscore", pd.Series(0.0, index=feat.index)).fillna(0.0)
    vola  = feat.get("volatility_20d", pd.Series(0.0, index=feat.index)).fillna(0.0)
    feat["vol_volatility_interact"] = (vol_z * vola).clip(-0.5, 0.5)

    # Short-sale 5d change (融券 — negative predictor of future returns)
    short_bal = df.get("short_balance", pd.Series(0, index=df.index)).astype(float)
    short_prev5 = short_bal.shift(5).replace(0, np.nan)
    feat["short_balance_5d_change"] = ((short_bal - short_prev5) / short_prev5).fillna(0.0).clip(-1, 1)

    # Trust conviction ratio (trust buy volume / total volume)
    trust_net = df.get("trust_net", pd.Series(0, index=df.index)).astype(float)
    trust_abs = trust_net.abs()
    feat["trust_vol_ratio"] = (trust_abs / volume.replace(0, np.nan)).fillna(0.0).clip(upper=0.3)

    # Amihud illiquidity
    for col in ("amihud_illiquidity", "amihud_zscore"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)
    feat["amihud_illiquidity"] = feat["amihud_illiquidity"].clip(upper=10.0)

    # Momentum features (time-series proxy for cross-sectional momentum)
    close = df["close"] if "close" in df.columns else pd.Series(1.0, index=df.index)
    for period in (60, 120):
        ret = close.pct_change(period)
        feat[f"return_{period}d"] = ret.fillna(0.0).clip(-1.0, 1.0)
    feat["mom_minus_reversal"] = (feat["return_60d"] - feat.get("return_5d", pd.Series(0.0, index=feat.index))).clip(-1.0, 1.0)

    # Monthly revenue momentum features (fundamental tailwind — forward-filled from filing_date)
    for col in ("rev_yoy", "rev_mom_3m", "rev_accel", "rev_new_high_12m"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # TAIFEX sentiment features (Taiwan stocks; neutral zeros when unavailable)
    for col in ("pcr_vol", "pcr_vol_5d", "large_trader_net_ratio"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # Securities lending (借券 — institutional short interest)
    for col in ("sbl_balance_ratio", "sbl_balance_5d_change", "sbl_rate"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # Block trades (鉅額交易)
    for col in ("block_vol_ratio", "block_premium"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # Large holder % (大戶持股比例 — smart money accumulation/distribution)
    for col in ("large_holder_pct", "large_holder_4w_change"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    # Sector peer return features (C30): prior-day mean return of same-sector peers
    for col in ("peer_ret_1d", "peer_ret_5d"):
        feat[col] = df.get(col, pd.Series(0.0, index=df.index)).fillna(0.0)

    return feat


def _build_triple_barrier_labels(close: pd.Series, pt_sl: float = 1.0,
                                  vol_lookback: int = 20, horizon: int = 5) -> pd.Series:
    """Triple barrier labeling with adaptive pt_sl (C19): 0.6× in low-vol regime, 1.0× in high-vol."""
    close_arr = close.values.astype(float)
    n = len(close_arr)
    log_ret = np.diff(np.log(close_arr + 1e-9))

    vols = []
    for i in range(n):
        start = max(0, i - vol_lookback)
        vol = float(np.std(log_ret[start:i])) if i > start + 4 else 0.015
        vols.append(max(vol, 0.001))
    vol_threshold = float(np.median(vols))

    labels = np.full(n, np.nan)
    for i in range(n - horizon):
        vol = vols[i]
        ratio = 0.6 if vol < vol_threshold else 1.0
        upper = close_arr[i] * (1 + vol * ratio)
        lower = close_arr[i] * (1 - vol * ratio)
        label = 0
        for j in range(1, horizon + 1):
            c = close_arr[i + j]
            if c >= upper: label = 1; break
            if c <= lower: label = -1; break
        labels[i] = label
    return pd.Series(labels, index=close.index)


class MetaLabelClassifier:
    """
    Secondary classifier: given primary model's prediction + features,
    predicts whether the primary signal is correct.
    Only signals where meta_confidence >= threshold are passed through.
    """
    def __init__(self, threshold: float = 0.55):
        self.threshold = threshold
        self.clf = RandomForestClassifier(
            n_estimators=100, max_depth=4, min_samples_leaf=5, random_state=42, n_jobs=-1
        )
        self.fitted = False

    def fit(self, X: np.ndarray, primary_pred: np.ndarray, primary_proba: np.ndarray,
            y_true: np.ndarray):
        """Train on directional signals only (skip HOLD rows)."""
        dir_mask = primary_pred != 0
        if dir_mask.sum() < 5:
            return self
        meta_y = (primary_pred[dir_mask] == y_true[dir_mask]).astype(int)
        if len(np.unique(meta_y)) < 2:
            return self
        X_meta = np.hstack([X[dir_mask], primary_proba[dir_mask]])
        self.clf.fit(X_meta, meta_y)
        self.fitted = True
        return self

    def filter(self, X: np.ndarray, primary_pred: np.ndarray,
               primary_proba: np.ndarray) -> np.ndarray:
        """Return filtered predictions — HOLD where meta says likely wrong."""
        if not self.fitted:
            return primary_pred
        dir_mask = primary_pred != 0
        final = primary_pred.copy()
        if dir_mask.sum() == 0:
            return final
        X_meta = np.hstack([X[dir_mask], primary_proba[dir_mask]])
        try:
            proba = self.clf.predict_proba(X_meta)
            classes = list(self.clf.classes_)
            correct_col = classes.index(1) if 1 in classes else -1
            confidence = proba[:, correct_col] if correct_col >= 0 else np.zeros(dir_mask.sum())
            suppress = confidence < self.threshold
            idxs = np.where(dir_mask)[0]
            final[idxs[suppress]] = 0
        except Exception:
            pass
        return final


class StockPredictor:
    """
    Ensemble predictor: RandomForest + XGBoost(probability average)。
    XGBoost 不可用時 graceful fallback 到純 RF。

    DISCLAIMER: ML signals are not financial advice.
    """

    DEFAULT_HORIZON = 5
    HORIZON = DEFAULT_HORIZON

    def __init__(self, horizon: int = DEFAULT_HORIZON) -> None:
        self.HORIZON = max(1, int(horizon))
        lightweight = os.getenv("LIGHTWEIGHT_MODE", "0") == "1"
        n_trees = 50 if lightweight else 200
        max_depth = 5 if lightweight else 8

        # ---- RandomForest(base)----
        self.classifier = RandomForestClassifier(
            n_estimators=n_trees, max_depth=max_depth,
            min_samples_leaf=5, max_features="sqrt",
            class_weight="balanced", random_state=42, n_jobs=1,
        )
        self.regressor = RandomForestRegressor(
            n_estimators=n_trees, max_depth=max_depth,
            min_samples_leaf=5, max_features="sqrt",
            random_state=42, n_jobs=1,
        )
        self.sell_classifier = RandomForestClassifier(
            n_estimators=n_trees, max_depth=max_depth,
            min_samples_leaf=5, max_features="sqrt",
            class_weight="balanced", random_state=42, n_jobs=1,
        )

        # ---- XGBoost(ensemble partner)----
        self._use_xgb = _HAS_XGBOOST
        if self._use_xgb:
            xgb_trees = 30 if lightweight else 100
            self.xgb_clf = XGBClassifier(
                n_estimators=xgb_trees, max_depth=max_depth,
                learning_rate=0.1, eval_metric="logloss",
                random_state=42, n_jobs=1, verbosity=0,
            )
            self.xgb_sell = XGBClassifier(
                n_estimators=xgb_trees, max_depth=max_depth,
                learning_rate=0.1, eval_metric="logloss",
                random_state=42, n_jobs=1, verbosity=0,
            )
            self.xgb_reg = XGBRegressor(
                n_estimators=xgb_trees, max_depth=max_depth,
                learning_rate=0.1, random_state=42, n_jobs=1, verbosity=0,
            )
        else:
            self.xgb_clf = self.xgb_sell = self.xgb_reg = None

        # ---- LightGBM(3rd ensemble member) ----
        self._use_lgbm = _HAS_LGBM
        if self._use_lgbm:
            lgbm_trees = 50 if lightweight else 150
            self.lgbm_clf = LGBMClassifier(
                n_estimators=lgbm_trees, max_depth=max_depth,
                learning_rate=0.05, class_weight="balanced",
                random_state=42, n_jobs=1, verbose=-1,
            )
            self.lgbm_sell = LGBMClassifier(
                n_estimators=lgbm_trees, max_depth=max_depth,
                learning_rate=0.05, class_weight="balanced",
                random_state=42, n_jobs=1, verbose=-1,
            )
            self.lgbm_reg = LGBMRegressor(
                n_estimators=lgbm_trees, max_depth=max_depth,
                learning_rate=0.05, random_state=42, n_jobs=1, verbose=-1,
            )
            active = ["RF"]
            if self._use_xgb: active.append("XGB")
            active.append("LGBM")
            logger.info("Ensemble enabled: %s", " + ".join(active))
        else:
            self.lgbm_clf = self.lgbm_sell = self.lgbm_reg = None

        # ---- CatBoost (4th ensemble member) ----
        self._use_catboost = _HAS_CATBOOST
        # Instances created at train() time (class weights are data-dependent)
        self.cb_clf: Optional["CatBoostClassifier"] = None
        self.cb_sell: Optional["CatBoostClassifier"] = None

        self.scaler = StandardScaler()
        self._trained = False
        self._train_accuracy: Optional[float] = None

    # ------------------------------------------------------------------
    # Training
    # ------------------------------------------------------------------

    def train(
        self,
        df: pd.DataFrame,
        *,
        precomputed_features: pd.DataFrame | None = None,
        buy_threshold_pct: float = 0.015,
        sell_threshold_pct: float = 0.015,
        use_triple_barrier: bool = True,
    ) -> dict:
        """
        Train classifier + regressor on the indicator-enriched DataFrame.
        Returns a dict with training metrics.

        If *precomputed_features* is provided, skip the (expensive)
        ``_build_features()`` call — useful in walk-forward backtests where
        features are computed once for the whole dataset and sliced per window.

        buy_threshold_pct / sell_threshold_pct: 買賣信號閾值(default ±1.5%
        for stocks; metals use gold ±1.0%, silver ±2.0%)。
        """
        feat = precomputed_features if precomputed_features is not None else _build_features(df)

        # ---- 自動閾值校準(用近期波動率中位數)----
        if buy_threshold_pct == 0.015 and "volatility_20d" in feat.columns:
            # 只在使用預設閾值時自動校準
            recent_vol = feat["volatility_20d"].dropna().tail(60)
            if len(recent_vol) >= 20:
                median_vol = float(recent_vol.median())
                # 閾值 = 5 日預期波動(median daily vol × sqrt(5))× 0.6
                auto_threshold = median_vol * np.sqrt(self.HORIZON) * 0.6
                auto_threshold = np.clip(auto_threshold, 0.008, 0.05)  # cap: 0.8% ~ 5%
                buy_threshold_pct = auto_threshold
                sell_threshold_pct = auto_threshold
                logger.info("Auto-calibrated threshold: ±%.2f%% (median vol=%.4f)",
                            auto_threshold * 100, median_vol)

        # Targets (shift by -HORIZON so each row maps to future price)
        future_close = df["close"].shift(-self.HORIZON)
        current_close = df["close"]

        if use_triple_barrier:
            # Triple barrier labels: 1=upper hit, -1=lower hit, 0=vertical (hold)
            tb_labels = _build_triple_barrier_labels(df["close"], horizon=self.HORIZON)
            clf_target = (tb_labels == 1).astype(int)
            sell_target = (tb_labels == -1).astype(int)
        else:
            # Classifier target: 1 if future price > current * (1 + threshold)
            clf_target = (future_close > current_close * (1 + buy_threshold_pct)).astype(int)

            # Sell classifier target: 1 if future price < current * (1 - threshold)
            sell_target = (future_close < current_close * (1 - sell_threshold_pct)).astype(int)

        # Regressor target: % change over HORIZON days (scale-invariant)
        reg_target = (future_close - current_close) / current_close * 100

        # Combine and drop rows with any NaN
        combined = feat.copy()
        combined["_clf_target"] = clf_target
        combined["_sell_target"] = sell_target
        combined["_reg_target"] = reg_target
        combined = combined.dropna()

        min_train_rows = 10  # Allow degraded mode for newer/smaller stocks
        if len(combined) < min_train_rows:
            raise ValueError(
                f"Not enough clean training rows ({len(combined)}). "
                f"Need at least {min_train_rows}."
            )

        X = combined[FEATURE_COLUMNS].values
        y_clf = combined["_clf_target"].values
        y_sell = combined["_sell_target"].values
        y_reg = combined["_reg_target"].values

        # Train / test split FIRST (last 20% as test, no shuffle to respect time order)
        split_idx = int(len(X) * 0.8)
        X_train_raw, X_test_raw = X[:split_idx], X[split_idx:]
        y_clf_train, y_clf_test = y_clf[:split_idx], y_clf[split_idx:]
        y_sell_train, _ = y_sell[:split_idx], y_sell[split_idx:]
        y_reg_train, _ = y_reg[:split_idx], y_reg[split_idx:]

        # Scale features — fit on train only to prevent data leakage
        X_train = self.scaler.fit_transform(X_train_raw)
        X_test = self.scaler.transform(X_test_raw)
        X_train_lgbm = _as_lgbm_feature_frame(X_train)
        X_test_lgbm = _as_lgbm_feature_frame(X_test)

        # ---- RF training ----
        self.classifier.fit(X_train, y_clf_train)
        self.sell_classifier.fit(X_train, y_sell_train)
        self.regressor.fit(X_train, y_reg_train)

        # ---- XGBoost training (ensemble) ----
        if self._use_xgb and self.xgb_clf is not None:
            try:
                self.xgb_clf.fit(X_train, y_clf_train)
                self.xgb_sell.fit(X_train, y_sell_train)
                self.xgb_reg.fit(X_train, y_reg_train)
            except Exception as exc:
                logger.warning("XGBoost training failed, skipping: %s", exc)
                self._use_xgb = False

        # ---- LightGBM training (3rd ensemble member) ----
        if self._use_lgbm and self.lgbm_clf is not None:
            try:
                self.lgbm_clf.fit(X_train_lgbm, y_clf_train)
                self.lgbm_sell.fit(X_train_lgbm, y_sell_train)
                self.lgbm_reg.fit(X_train_lgbm, y_reg_train)
            except Exception as exc:
                logger.warning("LightGBM training failed, skipping: %s", exc)
                self._use_lgbm = False

        # ---- CatBoost training (4th ensemble member) ----
        if self._use_catboost:
            try:
                from sklearn.utils.class_weight import compute_class_weight
                classes_clf = np.unique(y_clf_train)
                w_clf = compute_class_weight("balanced", classes=classes_clf, y=y_clf_train)
                cw_clf = {int(c): float(w) for c, w in zip(classes_clf, w_clf)}
                classes_sell = np.unique(y_sell_train)
                w_sell = compute_class_weight("balanced", classes=classes_sell, y=y_sell_train)
                cw_sell = {int(c): float(w) for c, w in zip(classes_sell, w_sell)}
                self.cb_clf = CatBoostClassifier(**{**_CATBOOST_CLF_PARAMS, "class_weights": cw_clf})
                self.cb_clf.fit(X_train, y_clf_train)
                self.cb_sell = CatBoostClassifier(**{**_CATBOOST_CLF_PARAMS, "class_weights": cw_sell})
                self.cb_sell.fit(X_train, y_sell_train)
            except Exception as exc:
                logger.warning("CatBoost training failed, skipping: %s", exc)
                self.cb_clf = self.cb_sell = None
                self._use_catboost = False

        self._trained = True

        # ---- Ensemble accuracy: average all available model probabilities ----
        probs = [self.classifier.predict_proba(X_test)[:, 1]]
        model_names = ["RF"]
        if self._use_xgb and self.xgb_clf is not None:
            probs.append(self.xgb_clf.predict_proba(X_test)[:, 1])
            model_names.append("XGB")
        if self._use_lgbm and self.lgbm_clf is not None:
            probs.append(self.lgbm_clf.predict_proba(X_test_lgbm)[:, 1])
            model_names.append("LGBM")
        cb_proba_test = None
        if self._use_catboost and self.cb_clf is not None:
            # CatBoost MultiClass: columns ordered by class label; extract class-1 probability
            cb_full = self.cb_clf.predict_proba(X_test)
            cb_classes = [int(c) for c in self.cb_clf.classes_]
            if 1 in cb_classes:
                cb_proba_test = cb_full[:, cb_classes.index(1)]
                probs.append(cb_proba_test)
            model_names.append("CatBoost")

        # Weighted average: RF=1, XGB=1, LGBM=1, CatBoost=1.2
        weights_list = [1.0] * len(probs)
        if cb_proba_test is not None:
            weights_list[-1] = 1.2
        total_w = sum(weights_list)
        ensemble_prob = sum(p * w for p, w in zip(probs, weights_list)) / total_w
        ensemble_pred = (ensemble_prob >= 0.5).astype(int)
        train_acc = float(np.mean(ensemble_pred == y_clf_test))
        indiv = {n: round(float(np.mean((p >= 0.5) == y_clf_test)), 3)
                 for n, p in zip(model_names, probs)}
        logger.info("Ensemble(%s) accuracy=%.3f  per-model=%s", "+".join(model_names), train_acc, indiv)
        self._train_accuracy = train_acc

        # VIF diagnostic (log warning if high multicollinearity detected)
        try:
            _log_vif_warnings(X_train, FEATURE_COLUMNS)
        except Exception:
            pass

        return {
            "training_samples": int(split_idx),
            "test_samples": int(len(X) - split_idx),
            "classifier_test_accuracy": round(train_acc, 4),
            "buy_rate": round(float(y_clf.mean()), 4),
            "prediction_horizon_days": int(self.HORIZON),
        }

    # ------------------------------------------------------------------
    # Optuna Hyperparameter Tuning
    # ------------------------------------------------------------------

    def tune_and_retrain(
        self, df: pd.DataFrame, *, n_trials: int = 20,
        buy_threshold_pct: float = 0.015, sell_threshold_pct: float = 0.015,
    ) -> dict:
        """用 Optuna 自動調參後重新訓練。回傳最佳參數 + accuracy。

        Walk-forward cross-validation:用 3 個 expanding window 評估,
        比固定 80/20 更穩定(避免 overfitting 到特定時段)。
        """
        if not _HAS_OPTUNA:
            logger.warning("Optuna not installed, skipping tuning")
            return self.train(df, buy_threshold_pct=buy_threshold_pct,
                              sell_threshold_pct=sell_threshold_pct)

        feat = _build_features(df)
        future_close = df["close"].shift(-self.HORIZON)
        current_close = df["close"]
        clf_target = (future_close > current_close * (1 + buy_threshold_pct)).astype(int)

        combined = feat.copy()
        combined["_target"] = clf_target
        combined = combined.dropna()

        if len(combined) < 60:
            logger.warning("Not enough data for Optuna (%d rows), using defaults", len(combined))
            return self.train(df, buy_threshold_pct=buy_threshold_pct,
                              sell_threshold_pct=sell_threshold_pct)

        X_all = combined[FEATURE_COLUMNS].values
        y_all = combined["_target"].values

        def objective(trial):
            # RF params
            rf_n = trial.suggest_int("rf_n_estimators", 50, 300, step=50)
            rf_depth = trial.suggest_int("rf_max_depth", 4, 12)
            rf_leaf = trial.suggest_int("rf_min_samples_leaf", 3, 15)

            # XGB params
            xgb_n = trial.suggest_int("xgb_n_estimators", 30, 200, step=30)
            xgb_depth = trial.suggest_int("xgb_max_depth", 3, 10)
            xgb_lr = trial.suggest_float("xgb_learning_rate", 0.01, 0.3, log=True)

            # Walk-forward CV: 3 expanding windows
            n = len(X_all)
            splits = [
                (0, int(n * 0.5), int(n * 0.5), int(n * 0.65)),
                (0, int(n * 0.65), int(n * 0.65), int(n * 0.8)),
                (0, int(n * 0.8), int(n * 0.8), n),
            ]
            scores = []
            for train_start, train_end, test_start, test_end in splits:
                X_tr = X_all[train_start:train_end]
                y_tr = y_all[train_start:train_end]
                X_te = X_all[test_start:test_end]
                y_te = y_all[test_start:test_end]

                if len(X_te) < 5 or len(np.unique(y_tr)) < 2:
                    continue

                scaler = StandardScaler()
                X_tr_s = scaler.fit_transform(X_tr)
                X_te_s = scaler.transform(X_te)

                rf = RandomForestClassifier(
                    n_estimators=rf_n, max_depth=rf_depth,
                    min_samples_leaf=rf_leaf, max_features="sqrt",
                    class_weight="balanced", random_state=42, n_jobs=1,
                )
                rf.fit(X_tr_s, y_tr)
                rf_prob = rf.predict_proba(X_te_s)
                rf_buy = rf_prob[:, 1] if rf_prob.shape[1] > 1 else np.full(len(X_te_s), 0.5)

                if _HAS_XGBOOST:
                    xgb = XGBClassifier(
                        n_estimators=xgb_n, max_depth=xgb_depth,
                        learning_rate=xgb_lr, eval_metric="logloss",
                        random_state=42, n_jobs=1, verbosity=0,
                    )
                    xgb.fit(X_tr_s, y_tr)
                    xgb_prob = xgb.predict_proba(X_te_s)
                    xgb_buy = xgb_prob[:, 1] if xgb_prob.shape[1] > 1 else np.full(len(X_te_s), 0.5)
                    ensemble = (rf_buy + xgb_buy) / 2
                else:
                    ensemble = rf_buy

                preds = (ensemble >= 0.5).astype(int)
                scores.append(float(np.mean(preds == y_te)))

            return np.mean(scores) if scores else 0.0

        study = optuna.create_study(direction="maximize")
        study.optimize(objective, n_trials=n_trials, show_progress_bar=False)

        best = study.best_params
        logger.info("Optuna best params (accuracy=%.3f): %s", study.best_value, best)

        # Apply best params
        self.classifier.set_params(
            n_estimators=best["rf_n_estimators"],
            max_depth=best["rf_max_depth"],
            min_samples_leaf=best["rf_min_samples_leaf"],
        )
        self.sell_classifier.set_params(
            n_estimators=best["rf_n_estimators"],
            max_depth=best["rf_max_depth"],
            min_samples_leaf=best["rf_min_samples_leaf"],
        )
        if self._use_xgb and self.xgb_clf is not None:
            self.xgb_clf.set_params(
                n_estimators=best["xgb_n_estimators"],
                max_depth=best["xgb_max_depth"],
                learning_rate=best["xgb_learning_rate"],
            )
            self.xgb_sell.set_params(
                n_estimators=best["xgb_n_estimators"],
                max_depth=best["xgb_max_depth"],
                learning_rate=best["xgb_learning_rate"],
            )

        # Retrain with best params
        result = self.train(df, buy_threshold_pct=buy_threshold_pct,
                            sell_threshold_pct=sell_threshold_pct)
        result["optuna_best_value"] = round(study.best_value, 4)
        result["optuna_best_params"] = best
        result["optuna_n_trials"] = n_trials
        return result

    # ------------------------------------------------------------------
    # Prediction
    # ------------------------------------------------------------------

    def predict(self, df: pd.DataFrame, *, precomputed_features: pd.DataFrame | None = None,
                use_meta_label: bool = False) -> dict:
        """
        Predict signal and price target from the *latest row* of df.

        Returns:
            signal: "BUY" | "SELL" | "HOLD"
            signal_probability: float 0-1  (probability of BUY)
            predicted_price: float
            predicted_change_pct: float
            sell_target: float  (predicted_price * 1.03 buffer)
            stop_loss: float    (current_price * 0.95)
            confidence: "HIGH" | "MEDIUM" | "LOW"
        """
        if not self._trained:
            raise RuntimeError("Model not trained yet. Call train() first.")

        feat = precomputed_features if precomputed_features is not None else _build_features(df)
        # Use last row that has all features available
        feat_clean = feat.dropna()
        if feat_clean.empty:
            raise ValueError("No clean feature rows available for prediction.")

        last = feat_clean.iloc[[-1]][FEATURE_COLUMNS].values
        last_scaled = self.scaler.transform(last)
        last_scaled_lgbm = _as_lgbm_feature_frame(last_scaled)

        # ---- RF probabilities ----
        buy_proba = self.classifier.predict_proba(last_scaled)[0]
        rf_buy = float(buy_proba[1]) if len(buy_proba) > 1 else 0.5
        sell_proba = self.sell_classifier.predict_proba(last_scaled)[0]
        rf_sell = float(sell_proba[1]) if len(sell_proba) > 1 else 0.5
        rf_change = float(self.regressor.predict(last_scaled)[0])

        # ---- Collect all available model predictions ----
        buy_probs = [rf_buy]
        sell_probs = [rf_sell]
        change_preds = [rf_change]

        if self._use_xgb and self.xgb_clf is not None:
            try:
                xgb_buy_p = self.xgb_clf.predict_proba(last_scaled)[0]
                xgb_sell_p = self.xgb_sell.predict_proba(last_scaled)[0]
                buy_probs.append(float(xgb_buy_p[1]) if len(xgb_buy_p) > 1 else 0.5)
                sell_probs.append(float(xgb_sell_p[1]) if len(xgb_sell_p) > 1 else 0.5)
                change_preds.append(float(self.xgb_reg.predict(last_scaled)[0]))
            except Exception as exc:
                logger.warning("XGBoost predict failed, skipping: %s", exc)

        if self._use_lgbm and self.lgbm_clf is not None:
            try:
                lgbm_buy_p = self.lgbm_clf.predict_proba(last_scaled_lgbm)[0]
                lgbm_sell_p = self.lgbm_sell.predict_proba(last_scaled_lgbm)[0]
                buy_probs.append(float(lgbm_buy_p[1]) if len(lgbm_buy_p) > 1 else 0.5)
                sell_probs.append(float(lgbm_sell_p[1]) if len(lgbm_sell_p) > 1 else 0.5)
                change_preds.append(float(self.lgbm_reg.predict(last_scaled_lgbm)[0]))
            except Exception as exc:
                logger.warning("LightGBM predict failed, skipping: %s", exc)

        # Weights: RF=1, XGB=1, LGBM=1, CatBoost=1.2 (higher weight — better on imbalanced data)
        pred_weights = [1.0] * len(buy_probs)
        if self._use_catboost and self.cb_clf is not None:
            try:
                cb_classes_buy = [int(c) for c in self.cb_clf.classes_]
                cb_buy_full = self.cb_clf.predict_proba(last_scaled)[0]
                cb_buy = float(cb_buy_full[cb_classes_buy.index(1)]) if 1 in cb_classes_buy else 0.5
                cb_classes_sell = [int(c) for c in self.cb_sell.classes_]
                cb_sell_full = self.cb_sell.predict_proba(last_scaled)[0]
                cb_sell = float(cb_sell_full[cb_classes_sell.index(1)]) if 1 in cb_classes_sell else 0.5
                buy_probs.append(cb_buy)
                sell_probs.append(cb_sell)
                change_preds.append(float(np.mean(change_preds)))  # no CatBoost regressor; reuse mean
                pred_weights.append(1.2)
            except Exception as exc:
                logger.warning("CatBoost predict failed, skipping: %s", exc)

        total_w = sum(pred_weights)
        buy_prob = float(sum(p * w for p, w in zip(buy_probs, pred_weights)) / total_w)
        sell_prob = float(sum(p * w for p, w in zip(sell_probs, pred_weights)) / total_w)
        predicted_change_pct = float(np.mean(change_preds))

        current_price = float(df["close"].iloc[-1])
        predicted_price = current_price * (1 + predicted_change_pct / 100)

        # Adaptive thresholds based on volatility regime (#10)
        vol_20d = float(feat_clean["volatility_20d"].iloc[-1]) if "volatility_20d" in feat_clean.columns else np.nan
        buy_threshold, sell_threshold = _adaptive_thresholds(vol_20d)
        raw_buy_prob = buy_prob
        raw_sell_prob = sell_prob
        raw_signal, raw_max_prob = _decide_signal(raw_buy_prob, raw_sell_prob, buy_threshold, sell_threshold)
        oldwang_context = _build_oldwang_context(df, feat_clean)
        buy_prob, sell_prob, oldwang_weight_overlay = _apply_oldwang_weight_overlay(
            buy_prob=buy_prob,
            sell_prob=sell_prob,
            oldwang_context=oldwang_context,
        )
        buy_prob, sell_prob, multi_factor_weight_overlay = apply_multi_factor_overlay(
            buy_prob=buy_prob,
            sell_prob=sell_prob,
            features=feat_clean,
            df=df,
        )
        oldwang_strategy_context = build_oldwang_strategy_context(df, feat_clean)
        pre_strategy_signal, _ = _decide_signal(buy_prob, sell_prob, buy_threshold, sell_threshold)
        strategy_gate_signal, buy_prob, sell_prob, oldwang_strategy_overlay = apply_oldwang_strategy_overlay(
            signal=pre_strategy_signal,
            buy_prob=buy_prob,
            sell_prob=sell_prob,
            context=oldwang_strategy_context,
        )

        # Signal logic — independent sell classifier (#11)
        signal, max_prob = _decide_signal(buy_prob, sell_prob, buy_threshold, sell_threshold)
        if (
            strategy_gate_signal == "HOLD"
            and pre_strategy_signal in {"BUY", "SELL"}
            and oldwang_strategy_overlay.get("reason")
        ):
            signal = "HOLD"

        # Confidence gate: suppress low-conviction directional signals
        min_conf = _CONFIDENCE_THRESHOLDS.get(signal, 0.0)
        if max_prob < min_conf:
            signal = "HOLD"

        # 200MA regime gate: suppress contrarian signals in strong trends
        # close_ma240_ratio = close/ma240 (1.10 = 10% above); convert to deviation
        _raw_ma240_ratio = float(precomputed_features.iloc[-1].get("close_ma240_ratio", 0.0)) if precomputed_features is not None else 0.0
        ma240_ratio = _raw_ma240_ratio - 1.0  # deviation: +0.10 = 10% above MA240
        if _raw_ma240_ratio > 0.0:
            if signal == "SELL" and ma240_ratio > 0.10:   # price >10% above MA240 → bull regime, don't short
                signal = "HOLD"
            elif signal == "BUY" and ma240_ratio < -0.10: # price >10% below MA240 → bear regime, high risk
                max_prob = max_prob * 0.85  # reduce confidence without hard-blocking

        signal, trend_conflict_guard = apply_trend_conflict_guard(
            signal=signal,
            buy_prob=buy_prob,
            sell_prob=sell_prob,
            df=df,
            features=feat_clean,
        )

        # Meta-label filter: secondary classifier suppresses low-quality signals
        meta_filtered = False
        if use_meta_label and signal != "HOLD" and hasattr(self, "_meta_clf") and self._meta_clf is not None:
            sig_int = np.array([1 if signal == "BUY" else -1])
            rf_proba_row = self.classifier.predict_proba(last_scaled)  # shape (1, n_classes)
            filtered = self._meta_clf.filter(last_scaled, sig_int, rf_proba_row)
            if filtered[0] == 0:
                signal = "HOLD"
                meta_filtered = True

        # Confidence tier
        extreme = max(buy_prob, sell_prob, 1 - buy_prob, 1 - sell_prob)
        if extreme >= CONFIDENCE_HIGH:
            confidence = "HIGH"
        elif extreme >= CONFIDENCE_MED:
            confidence = "MEDIUM"
        else:
            confidence = "LOW"

        # ATR-based stop loss / take profit (#15)
        atr_value = float(df["atr"].iloc[-1]) if "atr" in df.columns and not pd.isna(df["atr"].iloc[-1]) else current_price * 0.02
        buy_target  = current_price - atr_value * 0.5   # entry: half ATR below current
        sell_target = predicted_price + atr_value * 1.5  # take profit: 1.5 ATR above predicted
        stop_loss   = current_price - atr_value * 2.0    # stop loss: 2 ATR below current

        importances = sorted(
            zip(FEATURE_COLUMNS, self.classifier.feature_importances_),
            key=lambda x: x[1], reverse=True,
        )
        top_features = {k: round(float(v), 4) for k, v in importances[:5]}

        institutional_as_of = None
        institutional_available = False
        institutional_carry_forward = False
        institutional_flow = {
            "foreign_net": 0.0,
            "trust_net": 0.0,
            "dealer_net": 0.0,
            "institutional_net": 0.0,
        }
        if "institutional_as_of" in df.columns:
            as_of_val = df["institutional_as_of"].iloc[-1]
            institutional_as_of = None if pd.isna(as_of_val) else str(as_of_val)
        if "institutional_available" in df.columns:
            institutional_available = bool(df["institutional_available"].iloc[-1])
        if "institutional_carry_forward" in df.columns:
            institutional_carry_forward = bool(df["institutional_carry_forward"].iloc[-1])
        for col in institutional_flow:
            if col in df.columns and not pd.isna(df[col].iloc[-1]):
                institutional_flow[col] = round(float(df[col].iloc[-1]), 2)

        return {
            "signal": signal,
            "raw_signal": raw_signal,
            "raw_signal_probability": round(raw_buy_prob, 4),
            "raw_sell_probability": round(raw_sell_prob, 4),
            "signal_probability": round(buy_prob, 4),
            "sell_probability": round(sell_prob, 4),
            "buy_threshold": round(buy_threshold, 4),
            "sell_probability_threshold": round(1 - sell_threshold, 4),
            "predicted_price": round(predicted_price, 2),
            "predicted_change_pct": round(predicted_change_pct, 2),
            "buy_target": round(buy_target, 2),
            "sell_target": round(sell_target, 2),
            "stop_loss": round(stop_loss, 2),
            "confidence": confidence,
            "volatility_regime": _get_volatility_regime(vol_20d),
            "current_price": round(current_price, 2),
            "training_accuracy": round(self._train_accuracy or 0.0, 4),
            "top_features": top_features,
            "institutional_as_of": institutional_as_of,
            "institutional_available": institutional_available,
            "institutional_carry_forward": institutional_carry_forward,
            "institutional_flow": institutional_flow,
            "oldwang_context": oldwang_context,
            "oldwang_weight_overlay": oldwang_weight_overlay,
            "oldwang_strategy_context": oldwang_strategy_context,
            "oldwang_strategy_overlay": oldwang_strategy_overlay,
            "multi_factor_weight_overlay": multi_factor_weight_overlay,
            "trend_conflict_guard": trend_conflict_guard,
            "feature_version": FEATURE_VERSION,
            "meta_filtered": meta_filtered,
            "news_signal": None,
            "news_adjustment": 0.0,
            "news_as_of": None,
            "disclaimer": (
                "ML signals are for educational purposes only and do NOT "
                "constitute financial advice. Past performance does not guarantee "
                "future results."
            ),
        }

    # ------------------------------------------------------------------
    # Persistence
    # ------------------------------------------------------------------

    def save(self, path: str) -> None:
        """Serialize the trained predictor to disk using joblib."""
        os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
        payload = {
            "horizon": int(self.HORIZON),
            "classifier": self.classifier,
            "sell_classifier": self.sell_classifier,
            "regressor": self.regressor,
            "scaler": self.scaler,
            "_trained": self._trained,
            "_train_accuracy": self._train_accuracy,
            # XGB ensemble
            "xgb_clf": getattr(self, "xgb_clf", None),
            "xgb_sell": getattr(self, "xgb_sell", None),
            "xgb_reg": getattr(self, "xgb_reg", None),
            "_use_xgb": getattr(self, "_use_xgb", False),
            # LGBM ensemble
            "lgbm_clf": getattr(self, "lgbm_clf", None),
            "lgbm_sell": getattr(self, "lgbm_sell", None),
            "lgbm_reg": getattr(self, "lgbm_reg", None),
            "_use_lgbm": getattr(self, "_use_lgbm", False),
            # CatBoost ensemble
            "cb_clf": getattr(self, "cb_clf", None),
            "cb_sell": getattr(self, "cb_sell", None),
            "_use_catboost": getattr(self, "_use_catboost", False),
            # Feature count for stale-model detection
            "_n_features": len(FEATURE_COLUMNS),
        }
        joblib.dump(payload, path)
        logger.info("Model saved to %s", path)

    @classmethod
    def load(cls, path: str) -> "StockPredictor":
        """Deserialize a saved predictor from disk.

        Raises ValueError if the saved model's feature count does not match
        the current FEATURE_COLUMNS — caller should retrain.
        """
        payload = joblib.load(path)

        # Stale-model guard: feature count changed (e.g. new columns added)
        saved_n = payload.get("_n_features")
        if saved_n is not None and saved_n != len(FEATURE_COLUMNS):
            raise ValueError(
                f"Stale model: saved with {saved_n} features, "
                f"current FEATURE_COLUMNS has {len(FEATURE_COLUMNS)}"
            )

        instance = cls(horizon=int(payload.get("horizon", cls.DEFAULT_HORIZON) or cls.DEFAULT_HORIZON))
        instance.classifier = payload["classifier"]
        instance.sell_classifier = payload.get("sell_classifier", instance.sell_classifier)
        instance.regressor = payload["regressor"]
        instance.scaler = payload["scaler"]
        instance._trained = payload["_trained"]
        instance._train_accuracy = payload["_train_accuracy"]
        # Restore XGB ensemble (fixes prior bug where XGB was not persisted)
        instance.xgb_clf  = payload.get("xgb_clf")
        instance.xgb_sell = payload.get("xgb_sell")
        instance.xgb_reg  = payload.get("xgb_reg")
        instance._use_xgb = payload.get("_use_xgb", False) and instance.xgb_clf is not None
        # Restore LGBM ensemble
        instance.lgbm_clf  = payload.get("lgbm_clf")
        instance.lgbm_sell = payload.get("lgbm_sell")
        instance.lgbm_reg  = payload.get("lgbm_reg")
        instance._use_lgbm = payload.get("_use_lgbm", False) and instance.lgbm_clf is not None
        # Restore CatBoost ensemble
        instance.cb_clf   = payload.get("cb_clf")
        instance.cb_sell  = payload.get("cb_sell")
        instance._use_catboost = payload.get("_use_catboost", False) and instance.cb_clf is not None
        logger.info("Model loaded from %s (XGB=%s LGBM=%s CatBoost=%s)",
                    path, instance._use_xgb, instance._use_lgbm, instance._use_catboost)
        return instance