models/multi_factor_overlay.py

"""Runtime multi-factor probability overlay.

The validation harness learns signed factor weights: positive factor values
raise BUY probability, negative factor values raise SELL probability.  This
module applies the same semantics to the latest prediction row when an audited
config file is explicitly activated.
"""

from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd


ROOT = Path(__file__).resolve().parent.parent
DEFAULT_CONFIG_PATH = ROOT / "config" / "multi_factor_overlay.json"


def _config_path() -> Path:
    value = os.getenv("MULTI_FACTOR_OVERLAY_CONFIG", "").strip()
    return Path(value) if value else DEFAULT_CONFIG_PATH


def _env_enabled() -> bool:
    return os.getenv("ENABLE_MULTI_FACTOR_WEIGHT_OVERLAY", "0").strip().lower() in {"1", "true", "yes", "on"}


def load_multi_factor_overlay_config(path: str | Path | None = None) -> dict[str, Any] | None:
    config_path = Path(path) if path else _config_path()
    try:
        payload = json.loads(config_path.read_text())
    except FileNotFoundError:
        return None
    except Exception as exc:
        return {"active": False, "error": f"config_read_failed: {exc}", "config_path": str(config_path)}
    payload.setdefault("config_path", str(config_path))
    return payload


def _series(frame: pd.DataFrame, name: str, default: float = 0.0) -> pd.Series:
    if name in frame.columns:
        return pd.to_numeric(frame[name], errors="coerce").fillna(default)
    return pd.Series(default, index=frame.index, dtype=float)


def _first_series(frame: pd.DataFrame, names: list[str], default: float = 0.0) -> pd.Series:
    for name in names:
        if name in frame.columns:
            return _series(frame, name, default)
    return pd.Series(default, index=frame.index, dtype=float)


def _add_runtime_factor_inputs(features: pd.DataFrame, df: pd.DataFrame) -> pd.DataFrame:
    out = features.copy()
    market = df.reindex(out.index)
    open_ = _series(market, "open", np.nan)
    high = _series(market, "high", np.nan)
    low = _series(market, "low", np.nan)
    close = _series(market, "close", np.nan)
    volume = _series(market, "volume", 0.0).clip(lower=0.0)

    span = (high - low).abs().replace(0, np.nan)
    upper_shadow = high - np.maximum(open_, close)
    lower_shadow = np.minimum(open_, close) - low
    body = close - open_
    body_abs = body.abs()
    rsi = _series(out, "rsi", 50.0)
    macd_hist_norm = _series(out, "macd_hist_norm", 0.0)
    k_line = _series(out, "k", 50.0)
    d_line = _series(out, "d", 50.0)
    ma10 = close.rolling(10, min_periods=5).mean()
    ma20 = close.rolling(20, min_periods=10).mean()
    high_120 = close.rolling(120, min_periods=30).max()
    low_120 = close.rolling(120, min_periods=30).min()

    out["factor_upper_shadow_ratio"] = (upper_shadow / span).fillna(0.0).clip(0.0, 1.0)
    out["factor_lower_shadow_ratio"] = (lower_shadow / span).fillna(0.0).clip(0.0, 1.0)
    out["factor_body_ratio"] = (body / span).fillna(0.0).clip(-1.0, 1.0)
    out["factor_close_location"] = ((close - low) / span).fillna(0.5).clip(0.0, 1.0)
    body_strength = (body_abs / span).fillna(0.0).clip(0.0, 1.0)
    close_vs_120d_high = ((close / high_120.replace(0, np.nan)) - 1.0).fillna(0.0)
    close_vs_120d_low = ((close / low_120.replace(0, np.nan)) - 1.0).fillna(0.0)
    ma20_ratio = ((close / ma20.replace(0, np.nan)) - 1.0).fillna(0.0)
    high_base_score = pd.Series(0.0, index=out.index)
    high_base_score += ((close_vs_120d_high >= -0.03) & (close.pct_change(20).fillna(0.0) >= 0.18)).astype(float) * 2.0
    high_base_score += (close_vs_120d_low >= 0.60).astype(float)
    high_base_score += (ma20_ratio >= 0.12).astype(float)
    out["factor_close_vs_120d_high"] = close_vs_120d_high.clip(-1.0, 1.0)
    out["factor_close_vs_120d_low"] = close_vs_120d_low.clip(-1.0, 5.0)
    out["factor_high_base_score"] = high_base_score.clip(0.0, 4.0)
    out["factor_short_term_battle"] = (
        (body_strength <= 0.18)
        & (out["factor_upper_shadow_ratio"] >= 0.35)
        & (out["factor_close_location"] <= 0.70)
    ).astype(float)

    rsi_short = rsi.rolling(10, min_periods=5).mean()
    rsi_long = rsi.rolling(20, min_periods=10).mean()
    ma_trend = pd.Series(np.where(ma10 > ma20, 1.0, np.where(ma10 < ma20, -1.0, 0.0)), index=out.index)
    rsi_trend = pd.Series(np.where(rsi_short > rsi_long, 1.0, np.where(rsi_short < rsi_long, -1.0, 0.0)), index=out.index)
    mom15 = close.pct_change(15).fillna(0.0).clip(-0.25, 0.25) * 4.0
    kd_trend = pd.Series(np.where(k_line > d_line, 1.0, np.where(k_line < d_line, -1.0, 0.0)), index=out.index)
    macd_trend = (np.tanh(macd_hist_norm * 200.0) + 0.25 * np.sign(macd_hist_norm.diff().fillna(0.0))).clip(-1.0, 1.0)
    tej_trend_score = (ma_trend + rsi_trend + mom15 + kd_trend + macd_trend) / 5.0
    trend_state = pd.Series(
        np.where((ma_trend > 0) & (rsi_trend > 0), 1.0, np.where((ma_trend < 0) & (rsi_trend < 0), -1.0, 0.0)),
        index=out.index,
    )
    trend_flips = trend_state.ne(trend_state.shift()).astype(float).rolling(20, min_periods=5).sum().fillna(0.0)
    trend_persistence = trend_state.groupby(trend_state.ne(trend_state.shift()).cumsum()).cumcount().add(1).astype(float)
    close_ma20_distance = ((close / ma20.replace(0, np.nan)) - 1.0).abs().fillna(0.0)
    volatility = close.pct_change().rolling(20, min_periods=10).std().fillna(0.0)
    chop_penalty = (
        0.45 * (trend_flips / 8.0).clip(0.0, 1.0)
        + 0.30 * (volatility / 0.04).clip(0.0, 1.0)
        + 0.25 * (1.0 - (close_ma20_distance / 0.04).clip(0.0, 1.0))
    )
    persistent_trend = ((trend_persistence / 8.0).clip(0.0, 1.0) * trend_state).fillna(0.0)
    out["factor_tej_lstm_trend_score"] = tej_trend_score.fillna(0.0).clip(-1.0, 1.0)
    out["factor_tej_lstm_chop_score"] = (persistent_trend - chop_penalty).fillna(0.0).clip(-1.0, 1.0)

    lower_body_ratio = (lower_shadow / body_abs.replace(0, np.nan)).fillna(0.0)
    yin_gao_pao = (
        (body < 0.0)
        & (lower_body_ratio >= 1.0)
        & (out["factor_lower_shadow_ratio"] >= 0.25)
        & (out["factor_close_location"] >= 0.30)
    )
    recent_yin_gao_pao = yin_gao_pao.shift(1).fillna(False).rolling(3, min_periods=1).max().astype(bool)
    recent_yin_low = low.where(yin_gao_pao).shift(1).ffill()
    held_yin_low = recent_yin_gao_pao & recent_yin_low.notna() & (low >= recent_yin_low) & (close >= recent_yin_low)
    broke_yin_low = recent_yin_gao_pao & recent_yin_low.notna() & (low < recent_yin_low)
    out["factor_lower_shadow_body_ratio"] = lower_body_ratio.fillna(0.0).clip(0.0, 5.0)
    out["factor_yin_gao_pao_raw"] = yin_gao_pao.astype(float)
    out["factor_yin_gao_pao_hold"] = held_yin_low.astype(float)
    out["factor_yin_gao_pao_failed"] = broke_yin_low.astype(float)

    bullish_body = body > 0.0
    bearish_body = body < 0.0
    big_bull = bullish_body & (body_strength >= 0.45) & (out["factor_close_location"] >= 0.60)
    big_bear = bearish_body & (body_strength >= 0.45) & (out["factor_close_location"] <= 0.45)
    recent_big_bull = big_bull.shift(1).fillna(False).rolling(5, min_periods=1).max().astype(bool)
    recent_big_bear = big_bear.shift(1).fillna(False).rolling(5, min_periods=1).max().astype(bool)
    prior_big_bull_mid = ((open_ + close) / 2.0).where(big_bull).shift(1).ffill()
    prior_big_bull_low = low.where(big_bull).shift(1).ffill()
    prior_big_bear_high = high.where(big_bear).shift(1).ffill()
    prior_big_bear_mid = ((open_ + close) / 2.0).where(big_bear).shift(1).ffill()
    prior_big_bear_low = low.where(big_bear).shift(1).ffill()

    yin_ya_breakout = recent_big_bear & prior_big_bear_high.notna() & bullish_body & (
        (close > prior_big_bear_high) | (close > prior_big_bear_mid)
    )
    yin_ya_failed = recent_big_bear & prior_big_bear_low.notna() & (close < prior_big_bear_low)
    yi_yang_ding_breakdown = recent_big_bull & prior_big_bull_mid.notna() & (
        (close < prior_big_bull_mid) | (close < prior_big_bull_low)
    )
    out["factor_bt_yi_yin_ya_breakout"] = yin_ya_breakout.astype(float)
    out["factor_bt_yi_yin_ya_failed"] = yin_ya_failed.astype(float)
    out["factor_bt_yi_yang_ding_breakdown"] = yi_yang_ding_breakdown.astype(float)

    recent_upper_shadow_high = high.where(out["factor_upper_shadow_ratio"] >= 0.35).shift(1).rolling(20, min_periods=2).max()
    recent_lower_shadow_low = low.where(out["factor_lower_shadow_ratio"] >= 0.35).shift(1).rolling(20, min_periods=2).min()
    upper_cluster_break = recent_upper_shadow_high.notna() & (close > recent_upper_shadow_high)
    lower_cluster_bounce = (
        recent_lower_shadow_low.notna()
        & (low <= recent_lower_shadow_low * 1.01)
        & (close >= recent_lower_shadow_low)
        & bullish_body
    )
    lower_cluster_break = recent_lower_shadow_low.notna() & (close < recent_lower_shadow_low)
    out["factor_bt_upper_shadow_cluster_break"] = upper_cluster_break.astype(float)
    out["factor_bt_lower_shadow_cluster_bounce"] = lower_cluster_bounce.astype(float)
    out["factor_bt_lower_shadow_cluster_break"] = lower_cluster_break.astype(float)

    yang_gao_pao = (
        bullish_body
        & (body_strength >= 0.35)
        & (out["factor_close_location"] >= 0.90)
        & (((open_ - low) / span).fillna(0.0) <= 0.35)
    )
    out["factor_bt_yang_gao_pao"] = yang_gao_pao.astype(float)

    black_candle = bearish_body & (body_strength >= 0.15) & (out["factor_close_location"] <= 0.55)
    three_black = black_candle & black_candle.shift(1).fillna(False) & black_candle.shift(2).fillna(False)
    ma20_break = ma20.notna() & (close < ma20) & (close.shift(1) >= ma20.shift(1))
    below_ma20 = ma20.notna() & (close < ma20)
    san_sheng_wu_nai = three_black & below_ma20 & (ma20_break | (close < low.shift(1)))
    downside_breakdown = (
        bearish_body
        & (close < ma20)
        & ((close < low.shift(1)) | (close.pct_change().fillna(0.0) <= -0.015))
    )
    recent_breakdown_low = low.where(downside_breakdown).shift(1).ffill()
    recent_breakdown_close = close.where(downside_breakdown).shift(1).ffill()
    recent_breakdown_active = downside_breakdown.shift(1).fillna(False).rolling(5, min_periods=1).max().astype(bool)
    downside_follow_through = recent_breakdown_active & recent_breakdown_low.notna() & (
        (close < recent_breakdown_low) | (close < recent_breakdown_close)
    )
    rebound_from_break_low = ((close / recent_breakdown_low.replace(0, np.nan)) - 1.0).fillna(0.0)
    weak_rebound = (
        recent_breakdown_active
        & recent_breakdown_low.notna()
        & (rebound_from_break_low >= 0.0)
        & (rebound_from_break_low <= 0.025)
        & (close <= ma20)
    )
    out["factor_san_sheng_wu_nai_breakdown"] = san_sheng_wu_nai.astype(float)
    out["factor_downside_follow_through"] = downside_follow_through.astype(float)
    out["factor_weak_rebound_after_breakdown"] = weak_rebound.astype(float)

    prev2_bull = (close.shift(2) > open_.shift(2)) & (((close.shift(2) - open_.shift(2)).abs() / span.shift(2)) >= 0.20)
    prev2_bear = (close.shift(2) < open_.shift(2)) & (((close.shift(2) - open_.shift(2)).abs() / span.shift(2)) >= 0.15)
    prev1_bull = (close.shift(1) > open_.shift(1)) & (((close.shift(1) - open_.shift(1)).abs() / span.shift(1)) >= 0.15)
    prev1_bear = (close.shift(1) < open_.shift(1)) & (((close.shift(1) - open_.shift(1)).abs() / span.shift(1)) >= 0.15)
    current_bull = bullish_body & (body_strength >= 0.20)
    current_bear = (body < 0) & (body_strength >= 0.20)
    engulfs_prev_bear = (open_ <= close.shift(1)) & (close >= open_.shift(1))
    near_engulfs_prev_bear = prev1_bear & current_bull & (open_ <= close.shift(1) * 1.01) & (close >= open_.shift(1) * 0.99)
    engulfs_prev_bull = (open_ >= close.shift(1)) & (close <= open_.shift(1))
    near_engulfs_prev_bull = prev1_bull & current_bear & (open_ >= close.shift(1) * 0.99) & (close <= open_.shift(1) * 1.01)
    trend_context = close.shift(2).pct_change(10).fillna(0.0) > -0.10
    out["factor_bt_double_bull_engulf_one_bear"] = (
        prev2_bull & prev1_bear & current_bull & engulfs_prev_bear & trend_context
    ).astype(float)
    bull_engulf = prev1_bear & current_bull & engulfs_prev_bear
    bull_near_engulf = near_engulfs_prev_bear & ~bull_engulf
    bear_engulf_after_bulls = prev2_bull & prev1_bull & current_bear & engulfs_prev_bull
    bear_near_engulf_after_bulls = prev2_bull & prev1_bull & near_engulfs_prev_bull & ~bear_engulf_after_bulls
    two_bear_one_bull = prev2_bear & prev1_bear & current_bull
    two_bear_then_bull = two_bear_one_bull & (engulfs_prev_bear | near_engulfs_prev_bear)
    two_bull_one_bear = prev2_bull & prev1_bull & current_bear
    out["factor_bt_bull_engulf"] = pd.concat(
        [
            bull_engulf.astype(float),
            bull_near_engulf.astype(float) * 0.5,
            two_bear_one_bull.astype(float) * 0.5,
            two_bear_then_bull.astype(float),
        ],
        axis=1,
    ).max(axis=1)
    out["factor_bt_bull_engulf_bear"] = pd.concat(
        [
            bear_engulf_after_bulls.astype(float),
            bear_near_engulf_after_bulls.astype(float) * 0.5,
            two_bull_one_bear.astype(float) * 0.5,
        ],
        axis=1,
    ).max(axis=1)
    bull_signal = out["factor_bt_bull_engulf"] + out["factor_bt_double_bull_engulf_one_bear"]
    bear_signal = out["factor_bt_bull_engulf_bear"]
    out["factor_bt_bull_bear_engulf_ratio"] = (
        (bull_signal - bear_signal) / (bull_signal + bear_signal).replace(0.0, np.nan)
    ).fillna(0.0).clip(-1.0, 1.0)

    log_volume = np.log1p(volume)
    mean = log_volume.shift(1).rolling(20, min_periods=10).mean()
    std = log_volume.shift(1).rolling(20, min_periods=10).std().mask(lambda item: item <= 1e-9, 1.0)
    volume_z = ((log_volume - mean) / std).fillna(0.0).clip(-5.0, 5.0)
    out["factor_volume_z20_prior"] = volume_z
    volume_base = volume.shift(1).rolling(20, min_periods=10).mean()
    volume_ratio_prior = (volume / volume_base.replace(0, np.nan)).fillna(1.0).clip(0.0, 20.0)
    high_volume_upper_shadow = (
        (volume_ratio_prior >= 2.0).astype(float)
        * out["factor_upper_shadow_ratio"]
        * (volume_ratio_prior / 3.0).clip(0.0, 1.0)
        * (0.5 + 0.5 * (out["factor_high_base_score"] / 4.0).clip(0.0, 1.0))
    )
    out["factor_volume_ratio_prior20"] = volume_ratio_prior
    out["factor_high_volume_upper_shadow"] = high_volume_upper_shadow.fillna(0.0).clip(0.0, 1.0)
    out["factor_high_volume_downside_follow_through"] = (
        (volume_ratio_prior >= 1.5).astype(float)
        * black_candle.astype(float)
        * ((close < low.shift(1)) | downside_follow_through).astype(float)
        * (volume_ratio_prior / 3.0).clip(0.0, 1.0)
    ).fillna(0.0).clip(0.0, 1.0)
    spike = volume_z >= 2.0
    recent_prior_spike = spike.shift(1).fillna(False).rolling(5, min_periods=1).max().astype(bool)
    healthy_fade = recent_prior_spike & (volume_z < 0.0) & (out["factor_close_location"] >= 0.45)
    failed_fade = recent_prior_spike & (volume_z < 0.0) & (out["factor_close_location"] <= 0.30)
    out["factor_volume_contraction_after_spike"] = healthy_fade.astype(float) - failed_fade.astype(float)
    return out


def build_runtime_signed_factors(features: pd.DataFrame, df: pd.DataFrame, factor_names: list[str]) -> pd.DataFrame:
    features = _add_runtime_factor_inputs(features, df).replace([np.inf, -np.inf], np.nan).fillna(0.0)
    zero = pd.Series(0.0, index=features.index)
    half = pd.Series(0.5, index=features.index)
    rsi = _series(features, "rsi", 50.0)
    bb_pct_b = _series(features, "bb_pct_b", 0.5)
    macd_above_zero = _series(features, "macd_above_zero", 0.5) * 2.0 - 1.0
    macd_norm = _series(features, "macd_hist_norm", 0.0)
    ma_bull = _series(features, "ma_bull_alignment", 0.0)
    ma_bear = _series(features, "ma_bear_alignment", 0.0)
    taiex_return_5d = _series(features, "taiex_return_5d", 0.0)
    taiex_return_20d = _series(features, "taiex_return_20d", 0.0)
    taiex_ma20_ratio = _series(features, "taiex_ma20_ratio", 1.0)
    usdtwd_return_5d = _series(features, "usdtwd_return_5d", 0.0)
    sox_ret_5d = _series(features, "sox_ret_5d", 0.0)
    sox_ma20_ratio = _series(features, "sox_ma20_ratio", 1.0)
    tnx_change_5d = _series(features, "tnx_change_5d", 0.0)
    vix_level = _series(features, "vix_level", 20.0)
    vix_change_5d = _series(features, "vix_change_5d", 0.0)
    volume_z = _series(features, "factor_volume_z20_prior", 0.0)
    volume_ratio = _series(features, "volume_ratio", 0.0)
    volume_spike = _series(features, "oldwang_volume_spike", 0.0)

    def amount_strength(amount: pd.Series) -> pd.Series:
        amount = amount.abs().fillna(0.0)
        normalized_amount = amount.where(amount > 1000.0, amount * 1_000_000.0)
        return (normalized_amount / 5_000_000.0).fillna(0.0).clip(0.0, 1.0)

    big_1m_4m_buy = _first_series(
        features,
        [
            "big_player_buy_1m_4m",
            "big_player_buy_1m_4m_twd",
            "big_player_1m_4m_buy",
            "big_player_1m_4m_buy_twd",
            "big_player_buy_100m_400m",
            "big_player_buy_100m_400m_twd",
        ],
    ).abs()
    big_1m_4m_sell = _first_series(
        features,
        [
            "big_player_sell_1m_4m",
            "big_player_sell_1m_4m_twd",
            "big_player_1m_4m_sell",
            "big_player_1m_4m_sell_twd",
            "big_player_sell_100m_400m",
            "big_player_sell_100m_400m_twd",
        ],
    ).abs()
    big_gt_4m_buy = _first_series(
        features,
        [
            "big_player_buy_gt_4m",
            "big_player_buy_gt_4m_twd",
            "big_player_gt_4m_buy",
            "big_player_gt_4m_buy_twd",
            "big_player_buy_over_4m",
            "big_player_buy_over_4m_twd",
        ],
    ).abs()
    big_gt_4m_sell = _first_series(
        features,
        [
            "big_player_sell_gt_4m",
            "big_player_sell_gt_4m_twd",
            "big_player_gt_4m_sell",
            "big_player_gt_4m_sell_twd",
            "big_player_sell_over_4m",
            "big_player_sell_over_4m_twd",
        ],
    ).abs()
    big_1m_4m_total = (big_1m_4m_buy + big_1m_4m_sell).replace(0.0, np.nan)
    big_gt_4m_total = (big_gt_4m_buy + big_gt_4m_sell).replace(0.0, np.nan)
    big_player_buy = amount_strength(big_1m_4m_buy + big_gt_4m_buy)
    big_player_sell = -amount_strength(big_1m_4m_sell + big_gt_4m_sell)
    big_player_buy_ratio_1m_4m = (big_1m_4m_buy / big_1m_4m_total).fillna(0.0).clip(0.0, 1.0)
    big_player_sell_ratio_1m_4m = -((big_1m_4m_sell / big_1m_4m_total).fillna(0.0).clip(0.0, 1.0))
    big_player_buy_ratio_gt_4m = (big_gt_4m_buy / big_gt_4m_total).fillna(0.0).clip(0.0, 1.0)
    big_player_sell_ratio_gt_4m = -((big_gt_4m_sell / big_gt_4m_total).fillna(0.0).clip(0.0, 1.0))

    lower_shadow_raw = _series(features, "factor_lower_shadow_ratio").clip(0.0, 1.0)
    upper_shadow_raw = _series(features, "factor_upper_shadow_ratio").clip(0.0, 1.0)
    lower_shadow = (
        0.35 * lower_shadow_raw
        + 0.45 * _series(features, "factor_yin_gao_pao_raw")
        + 0.35 * _series(features, "factor_yin_gao_pao_hold")
        - 0.35 * _series(features, "factor_yin_gao_pao_failed")
    ).clip(0.0, 1.0)
    candle_shadow = (lower_shadow - upper_shadow_raw).clip(-1.0, 1.0)
    upper_shadow = -upper_shadow_raw
    bullish_upper_shadow = -((_series(features, "factor_body_ratio") > 0.35).astype(float) * _series(features, "factor_upper_shadow_ratio")).clip(0.0, 1.0)
    high_volume_upper_shadow_legacy = ((volume_z > 1.5).astype(float) * _series(features, "factor_upper_shadow_ratio") * np.maximum(-_series(features, "factor_body_ratio"), 0.25)).clip(0.0, 1.0)
    high_volume_upper_shadow = -np.maximum(
        high_volume_upper_shadow_legacy,
        _series(features, "factor_high_volume_upper_shadow"),
    ).clip(0.0, 1.0)
    high_base = -(_series(features, "factor_high_base_score") / 4.0).clip(0.0, 1.0)
    short_term_battle = -_series(features, "factor_short_term_battle").clip(0.0, 1.0)

    foreign_flow = np.tanh(_series(features, "foreign_net_vol_ratio") * 10.0)
    trust_flow = np.tanh(_series(features, "trust_net_vol_ratio") * 10.0)
    dealer_flow = np.tanh(_series(features, "dealer_net_vol_ratio") * 10.0)
    institutional_flow = (
        _series(features, "foreign_trust_alignment")
        + np.tanh(_series(features, "institutional_streak") / 5.0)
        + np.tanh(_series(features, "institutional_20d_zscore") / 2.0)
    ) / 3.0
    margin_risk = (-_series(features, "margin_5d_change_pct") - _series(features, "short_balance_5d_change") + (_series(features, "margin_buy_pressure", 0.5) - 0.5)).clip(-1.0, 1.0)
    short_pressure = -(
        np.tanh(_series(features, "short_margin_ratio") / 0.30)
        + np.tanh(_series(features, "sbl_balance_5d_change"))
        + np.tanh(_series(features, "sbl_balance_ratio") / 0.10)
    ) / 3.0
    volatility_risk = -(np.tanh(_series(features, "volatility_20d") / 0.04) + np.tanh(_series(features, "atr_ratio") / 0.05)) / 2.0
    momentum_60d = _series(features, "return_60d").clip(-0.5, 0.5) * 2.0
    momentum_120d = _series(features, "return_120d").clip(-0.5, 0.5) * 2.0
    tej_macro_risk_proxy = (
        0.28 * np.tanh((taiex_ma20_ratio - 1.0) / 0.03)
        + 0.18 * np.tanh(taiex_return_5d / 0.04)
        + 0.14 * np.tanh(taiex_return_20d / 0.08)
        + 0.16 * np.tanh(sox_ret_5d / 0.06)
        + 0.10 * np.tanh((sox_ma20_ratio - 1.0) / 0.04)
        - 0.06 * np.tanh(usdtwd_return_5d / 0.02)
        - 0.04 * np.tanh(tnx_change_5d / 0.30)
        - 0.04 * np.tanh((vix_level - 20.0) / 10.0)
        - 0.04 * np.tanh(vix_change_5d / 6.0)
    )

    available = {
        "oldwang_trend": _series(features, "oldwang_triple_bull") - _series(features, "oldwang_triple_bear"),
        "oldwang_guard": (
            _series(features, "oldwang_ma5_hold")
            + _series(features, "oldwang_trust_ma10_guard")
            + _series(features, "oldwang_foreign_ma20_guard")
            - _series(features, "oldwang_trust_ma10_broken")
            - _series(features, "oldwang_foreign_ma20_broken")
        ),
        "oldwang_volume": _series(features, "oldwang_volume_high_break") - _series(features, "oldwang_volume_low_break"),
        "san_sheng_wu_nai_breakdown": -_series(features, "factor_san_sheng_wu_nai_breakdown").clip(0.0, 1.0),
        "downside_follow_through": -_series(features, "factor_downside_follow_through").clip(0.0, 1.0),
        "weak_rebound_after_breakdown": -_series(features, "factor_weak_rebound_after_breakdown").clip(0.0, 1.0),
        "high_volume_downside_follow_through": -_series(
            features, "factor_high_volume_downside_follow_through"
        ).clip(0.0, 1.0),
        "intraday_60k_volume_low_guard": _series(features, "factor_intraday_60k_volume_low_guard").clip(0.0, 1.0),
        "intraday_60k_volume_low_break": -_series(features, "factor_intraday_60k_volume_low_break").clip(0.0, 1.0),
        "big_player_buy": big_player_buy,
        "big_player_sell": big_player_sell,
        # Legacy ratio/signed fields remain supported for old reports/seeds.
        "big_player_buy_ratio_1m_4m": big_player_buy_ratio_1m_4m,
        "big_player_sell_ratio_1m_4m": big_player_sell_ratio_1m_4m,
        "big_player_buy_ratio_gt_4m": big_player_buy_ratio_gt_4m,
        "big_player_sell_ratio_gt_4m": big_player_sell_ratio_gt_4m,
        "big_player_power_1m_4m": _series(features, "big_player_power_1m_4m").clip(-1.0, 1.0),
        "big_player_power_gt_4m": _series(features, "big_player_power_gt_4m").clip(-1.0, 1.0),
        "oldwang_gap": _series(features, "oldwang_gap_guard") - _series(features, "oldwang_gap_filled"),
        "volume_expansion": np.tanh((volume_ratio - 1.0).clip(-2.0, 4.0) / 2.0).clip(-1.0, 1.0),
        "volume_spike": volume_spike.clip(0.0, 1.0),
        "volume_surge": np.tanh(volume_z / 2.0).clip(-1.0, 1.0),
        "volume_contraction_after_spike": _series(features, "factor_volume_contraction_after_spike").clip(-1.0, 1.0),
        "candle_shadow": candle_shadow,
        "upper_shadow": upper_shadow,
        "lower_shadow": lower_shadow,
        "bullish_upper_shadow": bullish_upper_shadow,
        "high_base": high_base,
        "high_volume_upper_shadow": high_volume_upper_shadow,
        "short_term_battle": short_term_battle,
        "doji_battle": short_term_battle,
        "foreign_flow": foreign_flow.clip(-1.0, 1.0),
        "trust_flow": trust_flow.clip(-1.0, 1.0),
        "dealer_flow": dealer_flow.clip(-1.0, 1.0),
        "institutional_flow": institutional_flow.clip(-1.0, 1.0),
        "chip_score": _series(features, "chip_score"),
        "margin_risk": margin_risk,
        "short_pressure": short_pressure.clip(-1.0, 0.0),
        "macd_momentum": (np.tanh(macd_norm * 200.0) + 0.25 * macd_above_zero).clip(-1.0, 1.0),
        "rsi_trend": ((rsi - 50.0) / 50.0).clip(-1.0, 1.0),
        "rsi_reversal": pd.Series(
            np.where(rsi < 30.0, (30.0 - rsi) / 30.0, np.where(rsi > 70.0, -(rsi - 70.0) / 30.0, 0.0)),
            index=features.index,
        ).clip(-1.0, 1.0),
        "bb_reversion": pd.Series(
            np.where(bb_pct_b < 0.15, 1.0 - bb_pct_b / 0.15, np.where(bb_pct_b > 0.85, -(bb_pct_b - 0.85) / 0.15, 0.0)),
            index=features.index,
        ).clip(-1.0, 1.0),
        "ma_alignment": (ma_bull - ma_bear).clip(-1.0, 1.0),
        "momentum_20d": _series(features, "return_20d").clip(-0.25, 0.25) * 4.0,
        "momentum_60d": momentum_60d.clip(-1.0, 1.0),
        "momentum_120d": momentum_120d.clip(-1.0, 1.0),
        "volatility_risk": volatility_risk.clip(-1.0, 0.0),
        "liquidity_quality": -np.tanh(_series(features, "amihud_zscore") / 2.0),
        "bt_yi_yin_ya_breakout": (
            _series(features, "factor_bt_yi_yin_ya_breakout") - _series(features, "factor_bt_yi_yin_ya_failed")
        ).clip(-1.0, 1.0),
        "bt_yi_yang_ding_breakdown": -_series(features, "factor_bt_yi_yang_ding_breakdown").clip(0.0, 1.0),
        "bt_shadow_cluster_support_resistance": (
            _series(features, "factor_bt_upper_shadow_cluster_break")
            + _series(features, "factor_bt_lower_shadow_cluster_bounce")
            - _series(features, "factor_bt_lower_shadow_cluster_break")
        ).clip(-1.0, 1.0),
        "bt_yang_gao_pao": _series(features, "factor_bt_yang_gao_pao").clip(0.0, 1.0),
        "bt_bull_engulf": _series(features, "factor_bt_bull_engulf").clip(0.0, 1.0),
        "bt_bull_engulf_bear": -_series(features, "factor_bt_bull_engulf_bear").clip(0.0, 1.0),
        "bt_bull_bear_engulf_ratio": _series(features, "factor_bt_bull_bear_engulf_ratio").clip(-1.0, 1.0),
        "bt_double_bull_engulf_one_bear": _series(features, "factor_bt_double_bull_engulf_one_bear").clip(0.0, 1.0),
        "tej_lstm_trend_signal": _series(features, "factor_tej_lstm_trend_score").clip(-1.0, 1.0),
        "tej_lstm_chop_filter": _series(features, "factor_tej_lstm_chop_score").clip(-1.0, 1.0),
        "tej_macro_risk_proxy": tej_macro_risk_proxy.clip(-1.0, 1.0),
    }
    missing = [name for name in factor_names if name not in available]
    if missing:
        raise ValueError(f"unsupported runtime multi-factor names: {', '.join(missing)}")
    return pd.DataFrame({name: available[name].fillna(0.0).astype(float) for name in factor_names})


def _apply_side_policy(factors: pd.DataFrame, policy: dict[str, Any] | None) -> pd.DataFrame:
    if not policy:
        return factors

    keep_positive = {str(name) for name in policy.get("keep_positive", [])}
    keep_negative = {str(name) for name in policy.get("keep_negative", [])}
    if not keep_positive and not keep_negative:
        return factors

    output = factors.copy()
    for name in output.columns:
        values = pd.to_numeric(output[name], errors="coerce").fillna(0.0)
        if name not in keep_positive:
            values = values.where(values <= 0.0, 0.0)
        if name not in keep_negative:
            values = values.where(values >= 0.0, 0.0)
        output[name] = values
    return output


def apply_multi_factor_overlay(
    *,
    buy_prob: float,
    sell_prob: float,
    features: pd.DataFrame,
    df: pd.DataFrame,
    config: dict[str, Any] | None = None,
    enabled: bool | None = None,
) -> tuple[float, float, dict[str, Any]]:
    payload = config if config is not None else load_multi_factor_overlay_config()
    if not payload:
        return buy_prob, sell_prob, {"applied": False, "reason": "config_missing"}
    if payload.get("error"):
        return buy_prob, sell_prob, {"applied": False, "reason": payload["error"]}

    should_apply = _env_enabled() if enabled is None else bool(enabled)
    should_apply = should_apply and bool(payload.get("active"))
    selected = payload.get("selected_config") or payload.get("config") or {}
    weights = selected.get("weights") or {}
    if not should_apply:
        return buy_prob, sell_prob, {
            "applied": False,
            "reason": "disabled",
            "active": bool(payload.get("active")),
            "policy": payload.get("policy"),
            "config_path": payload.get("config_path"),
        }
    if not weights:
        return buy_prob, sell_prob, {"applied": False, "reason": "weights_missing"}

    factor_names = list(weights)
    side_policy = selected.get("side_policy") or payload.get("side_policy")
    factor_frame = _apply_side_policy(build_runtime_signed_factors(features, df, factor_names), side_policy)
    latest = factor_frame.iloc[-1]
    buy_adjustment = 0.0
    sell_adjustment = 0.0
    contributions: dict[str, dict[str, float]] = {}
    for name, weight in weights.items():
        value = float(latest.get(name, 0.0))
        weight = float(weight)
        buy_delta = weight * max(value, 0.0)
        sell_delta = weight * max(-value, 0.0)
        buy_adjustment += buy_delta
        sell_adjustment += sell_delta
        contributions[name] = {
            "value": round(value, 4),
            "weight": round(weight, 4),
            "buy_delta": round(buy_delta, 4),
            "sell_delta": round(sell_delta, 4),
        }

    adjusted_buy = float(np.clip(buy_prob + buy_adjustment, 0.0, 1.0))
    adjusted_sell = float(np.clip(sell_prob + sell_adjustment, 0.0, 1.0))
    return adjusted_buy, adjusted_sell, {
        "applied": True,
        "policy": payload.get("policy"),
        "scope": payload.get("scope"),
        "source_report": payload.get("source_report"),
        "selected_name": payload.get("selected_name"),
        "side_policy": side_policy,
        "config_path": payload.get("config_path"),
        "generated_at": payload.get("generated_at"),
        "runtime_applied_at": datetime.now(timezone.utc).isoformat(),
        "validation": payload.get("validation"),
        "gate": payload.get("gate"),
        "factor_count": len(weights),
        "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),
        "buy_adjustment": round(buy_adjustment, 4),
        "sell_adjustment": round(sell_adjustment, 4),
        "top_contributions": sorted(
            contributions.items(),
            key=lambda item: max(abs(item[1]["buy_delta"]), abs(item[1]["sell_delta"])),
            reverse=True,
        )[:8],
    }