""" 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