""" Walk-forward backtest service. Extracted from routers/stock.py (#21) to separate business logic from the router layer. """ import logging from datetime import datetime from dateutil.relativedelta import relativedelta from data.fetcher import fetch_history, fetch_cross_asset_tw, is_us_ticker from data.institutional_flow import add_institutional_flow from indicators.technical import add_all_indicators, add_cross_asset_tw from models.predictor import ( _build_features, FEATURE_COLUMNS, StockPredictor, _adaptive_thresholds, _as_lgbm_feature_frame, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Trading cost constants (Taiwan stock market) # --------------------------------------------------------------------------- COMMISSION_PCT = 0.1425 # broker commission per side COMMISSION_DISCOUNT = 0.6 # typical online broker discount TAX_PCT = 0.3 # securities transaction tax (sell-side only) BUY_COST = COMMISSION_PCT * COMMISSION_DISCOUNT # 0.0855% SELL_COST = COMMISSION_PCT * COMMISSION_DISCOUNT + TAX_PCT # 0.3855% ROUND_TRIP_COST_PCT = BUY_COST + SELL_COST # 0.471% MIN_TRAIN = 120 # minimum training window (trading days) HORIZON = 5 # prediction horizon (trading days) RETRAIN_EVERY = 5 # retrain every N prediction days EMBARGO = HORIZON # skip HORIZON days before prediction point to prevent look-ahead TRAIN_WINDOW = 250 # sliding training window (1 trading year); use all data if shorter class BacktestError(Exception): """Raised when backtest cannot proceed (e.g. no data).""" pass def compute_net_return(signal: str, actual_return: float, cost: float) -> float: """ Compute net return for a given signal. - BUY → long position: actual_return - cost - SELL → short position: -actual_return - cost - HOLD → no trade: 0 """ if signal == "BUY": return actual_return - cost elif signal == "SELL": return -actual_return - cost else: return 0.0 def walk_forward_backtest(stock_no: str, months: int) -> dict: """ Walk-forward backtest with embargo period and sliding training window. Key improvements: - Embargo: train on data[:i-EMBARGO] instead of data[:i], eliminating the look-ahead bias where the last training target overlaps with the test point. - Dynamic window: use a sliding TRAIN_WINDOW (250 days) instead of the full history, so the model adapts to recent market regimes. - Cross-asset features: TAIEX + USD/TWD merged before feature extraction. Raises BacktestError if no data is found. """ # Fetch extra history for training warmup df = fetch_history(stock_no, months=months + 12) if df.empty: raise BacktestError(f"No data found for {stock_no}") df = add_all_indicators(df) # Cross-asset enrichment (Taiwan stocks only) if not is_us_ticker(stock_no): df = add_institutional_flow(df, stock_no) start = str(df["date"].min()) end = str(df["date"].max()) taiex, usdtwd, sox, tnx = fetch_cross_asset_tw(start, end) df = add_cross_asset_tw(df, taiex, usdtwd, sox_close=sox, tnx_close=tnx) feat = _build_features(df) cutoff = str((datetime.today() - relativedelta(months=months)).date()) results = [] predictor = None last_train_idx = -RETRAIN_EVERY # force first train for i in range(MIN_TRAIN + EMBARGO, len(df) - HORIZON): row_feat = feat.iloc[i] if row_feat[FEATURE_COLUMNS].isna().any(): continue date_str = str(df["date"].iloc[i]) if date_str < cutoff: continue # Embargo: exclude the HORIZON rows immediately before prediction point # so no training target leaks into the test window. train_end = i - EMBARGO train_start = max(0, train_end - TRAIN_WINDOW) if train_end - train_start < MIN_TRAIN: continue # Retrain if needed (every RETRAIN_EVERY prediction days) if i - last_train_idx >= RETRAIN_EVERY or predictor is None: try: predictor = StockPredictor() predictor.train( df.iloc[train_start:train_end], precomputed_features=feat.iloc[train_start:train_end], ) last_train_idx = i except Exception: continue current_close = float(df["close"].iloc[i]) future_close = float(df["close"].iloc[i + HORIZON]) actual_return = (future_close - current_close) / current_close * 100 X = row_feat[FEATURE_COLUMNS].values.reshape(1, -1) X_scaled = predictor.scaler.transform(X) X_scaled_lgbm = _as_lgbm_feature_frame(X_scaled) def _class_one_probability(model, X_input=X_scaled) -> float: prob = model.predict_proba(X_input)[0] return float(prob[1]) if len(prob) > 1 else 0.5 buy_probs = [_class_one_probability(predictor.classifier)] sell_probs = [_class_one_probability(predictor.sell_classifier)] if getattr(predictor, "_use_xgb", False) and predictor.xgb_clf is not None: try: buy_probs.append(_class_one_probability(predictor.xgb_clf)) sell_probs.append(_class_one_probability(predictor.xgb_sell)) except Exception: pass if getattr(predictor, "_use_lgbm", False) and predictor.lgbm_clf is not None: try: buy_probs.append(_class_one_probability(predictor.lgbm_clf, X_scaled_lgbm)) sell_probs.append(_class_one_probability(predictor.lgbm_sell, X_scaled_lgbm)) except Exception: pass buy_prob = sum(buy_probs) / len(buy_probs) sell_prob = sum(sell_probs) / len(sell_probs) vol_20d = float(row_feat.get("volatility_20d", 0) or 0) buy_threshold, sell_threshold = _adaptive_thresholds(vol_20d) if buy_prob >= buy_threshold: signal = "BUY" elif sell_prob >= (1 - sell_threshold): signal = "SELL" else: signal = "HOLD" correct = (signal == "BUY" and actual_return > 0) or \ (signal == "SELL" and actual_return < 0) net_return = compute_net_return(signal, actual_return, ROUND_TRIP_COST_PCT) results.append({ "date": date_str, "signal": signal, "buy_prob": round(buy_prob, 3), "sell_prob": round(sell_prob, 3), "close": round(current_close, 2), "future_close": round(future_close, 2), "actual_return_pct": round(actual_return, 2), "net_return_pct": round(net_return, 2), "correct": correct, }) return _build_stats(stock_no, months, results) def _build_stats(stock_no: str, months: int, results: list[dict]) -> dict: """Aggregate backtest results into summary statistics.""" buys = [r for r in results if r["signal"] == "BUY"] sells = [r for r in results if r["signal"] == "SELL"] holds = [r for r in results if r["signal"] == "HOLD"] non_holds = [r for r in results if r["signal"] != "HOLD"] def win_rate(lst): return round(sum(1 for r in lst if r["correct"]) / len(lst) * 100, 1) if lst else 0 def avg_ret(lst): return round(sum(r["net_return_pct"] for r in lst) / len(lst), 2) if lst else 0 def max_consecutive(lst, correct_val: bool): max_streak = cur_streak = 0 for r in lst: if r["correct"] == correct_val: cur_streak += 1 max_streak = max(max_streak, cur_streak) else: cur_streak = 0 return max_streak trade_net_returns = [r["net_return_pct"] for r in non_holds] if non_holds: signal_accuracy_overall = round( sum(1 for r in non_holds if r["correct"]) / len(non_holds) * 100, 1 ) else: signal_accuracy_overall = 0 return { "stock_no": stock_no, "months": months, "walk_forward": True, "trading_cost_pct": round(ROUND_TRIP_COST_PCT, 4), "cost_breakdown": { "buy_commission": round(BUY_COST, 4), "sell_commission_and_tax": round(SELL_COST, 4), }, "stats": { "total_days": len(results), "buy_signals": len(buys), "sell_signals": len(sells), "hold_signals": len(holds), "buy_win_rate": win_rate(buys), "sell_win_rate": win_rate(sells), "avg_return_on_buy": avg_ret(buys), "avg_return_on_sell": avg_ret(sells), "total_return_if_followed": round(sum(r["net_return_pct"] for r in non_holds), 2), "gross_return": round(sum(r["actual_return_pct"] for r in buys), 2), "max_consecutive_wins": max_consecutive(non_holds, True), "max_consecutive_losses": max_consecutive(non_holds, False), "best_trade_pct": round(max(trade_net_returns), 2) if trade_net_returns else 0, "worst_trade_pct": round(min(trade_net_returns), 2) if trade_net_returns else 0, "signal_accuracy_overall": signal_accuracy_overall, }, "trades": results, }