"""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], }