""" 機器學習產生交易訊號(而非預測股價)— 可重現的台股回測 ===================================================================== 論點:不要用機器學習「預測股價的價位」,而是用它在每個月底,對全市場 股票做「橫截面排序」,產生一個相對強弱的交易訊號,挑出模型認為下個月 會相對強勢的股票做多。標籤(label)用的是「次月相對中位數的超額報酬」 (excess_over_median),這直接表達了上面的論點:我們學的是「誰會贏過 市場中位數」這個排序,而不是「股價會到多少」這個價位。 嚴謹性: - 樣本外:訓練只用 2018 年以前的資料,測試在 2019-01 之後,中間留 purge 缺口避免標籤洩漏(label 看的是次月報酬,跨月邊界會偷看未來)。 - 另外用 CPCV(Combinatorial Purged Cross-Validation)報告每折的 Rank IC,證明訊號的有效性不是靠單一切分的運氣。 - 交易成本:finlab sim() 內扣手續費 0.1425% 與證交稅 0.3%。 - benchmark:0050 含息(etl:adj_close)買進持有,對齊同一回測視窗。 執行方式: cd ~/Documents/finlab UV_ENV_FILE=.env uv run --with scikit-learn --with lightgbm --with matplotlib python strategy.py 作者:FinLab 量化研究團隊 """ import warnings warnings.filterwarnings("ignore") import os import json import numpy as np import pandas as pd import finlab finlab.login() # finlab 會在需要資料時自動引導登入 from finlab import data import finlab.ml.feature as mlf import finlab.ml.label as mll from finlab.ml.cpcv import cpcv_train_test_split from finlab.backtest import sim from lightgbm import LGBMRegressor # ---------------------------------------------------------------------- # 設定 # ---------------------------------------------------------------------- START = "2010-01-01" # 特徵起始(留足夠訓練樣本) TRAIN_END = "2018-12-31" # 樣本內訓練截止 TEST_START = "2019-01-01" # 樣本外測試起始 DATA_END = "2026-06-26" # 資料截止(台股最新交易日) TOP_N = 30 # 每月做多前 N 名 OUT_DIR = os.path.dirname(os.path.abspath(__file__)) def liquid_universe(): """流動性 + 排除類別過濾:避免回測買得到、實單買不到,並排除 ETF / 全額交割。""" close = data.get("price:收盤價") vol = data.get("price:成交股數") amount = (close * vol).rolling(60).mean() # 60 日平均成交額 liquid = amount > 50_000_000 # 日均成交額 > 5000 萬 # 排除 ETF(代號 00 開頭)與全額交割(用市值門檻間接過濾極小型) is_etf = pd.Series([str(c).startswith("00") for c in close.columns], index=close.columns) liquid = liquid.loc[:, ~is_etf.values] return liquid def build_features(): """用 data.indicator()(技術面)+ 基本面 / 籌碼面組成多因子特徵矩陣,月頻。""" feats = { # 技術面動能 / 擺盪 "rsi": data.indicator("RSI"), "willr": data.indicator("WILLR"), "mom": data.indicator("MOM", timeperiod=120), "cci": data.indicator("CCI"), "adx": data.indicator("ADX"), # 價量位置 "price_pos": data.get("price:收盤價") / data.get("price:收盤價").rolling(240).mean(), "vol_ratio": data.get("price:成交股數") / data.get("price:成交股數").rolling(60).mean(), # 基本面 / 估值 "pb": data.get("price_earning_ratio:股價淨值比"), "pe": data.get("price_earning_ratio:本益比"), "rev_yoy": data.get("monthly_revenue:去年同月增減(%)"), "rev_mom": data.get("monthly_revenue:上月比較增減(%)"), # 籌碼面 "foreign": data.get("institutional_investors_trading_summary:外陸資買賣超股數(不含外資自營商)"), } liquid = liquid_universe() X = mlf.combine(feats, resample="ME", sample_filter=liquid) return X def build_label(index): """標籤 = 次月相對中位數的超額報酬(橫截面排序訊號,非價位預測)。""" return mll.excess_over_median(index, resample="ME") def cagr(equity): years = (equity.index[-1] - equity.index[0]).days / 365.25 return (equity.iloc[-1] / equity.iloc[0]) ** (1 / years) - 1 def max_drawdown(equity): return (equity / equity.cummax() - 1).min() def daily_sharpe(daily_ret): return daily_ret.mean() / daily_ret.std() * np.sqrt(252) def monthly_sharpe(equity): m = equity.resample("ME").last().pct_change().dropna() return m.mean() / m.std() * np.sqrt(12) def rank_ic(pred, label): """橫截面 Rank IC:每期 predict 與 label 的 Spearman 相關,取平均。""" df = pd.DataFrame({"p": pred, "y": label}).dropna() if df.empty: return np.nan ics = df.groupby(level="datetime").apply( lambda g: g["p"].rank().corr(g["y"].rank()) if len(g) > 5 else np.nan ) return ics.mean() def main(): print("[1/6] building features ...") X = build_features() X = X[(X.index.get_level_values("datetime") >= START) & (X.index.get_level_values("datetime") <= DATA_END)] print(" feature matrix:", X.shape) print("[2/6] building label (excess_over_median, monthly) ...") y = build_label(X.index) df = X.copy() df["label"] = y df = df.replace([np.inf, -np.inf], np.nan) df = df.dropna(subset=["label"]) feat_cols = list(X.columns) # 特徵缺值補橫截面中位數(每期),避免整列丟棄造成樣本流失 df[feat_cols] = df.groupby(level="datetime")[feat_cols].transform( lambda s: s.fillna(s.median()) ) df = df.dropna(subset=feat_cols) print(" clean samples:", len(df)) dt = df.index.get_level_values("datetime") train_mask = dt <= pd.Timestamp(TRAIN_END) # purge:測試集從 TEST_START 起,訓練/測試之間天然隔一個月以上 test_mask = dt >= pd.Timestamp(TEST_START) print("[3/6] CPCV fold Rank IC (robustness, leakage-free folds) ...") fold_ics = [] arr = df.reset_index() for tr_idx, te_idx in cpcv_train_test_split( df, num_splits=6, test_size=2, perge_period=pd.Timedelta(days=31) ): m = LGBMRegressor(n_estimators=200, num_leaves=31, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, min_child_samples=50, n_jobs=-1, verbose=-1) m.fit(df[feat_cols].values[tr_idx], df["label"].values[tr_idx]) pred = pd.Series(m.predict(df[feat_cols].values[te_idx]), index=df.index[te_idx]) ic = rank_ic(pred, df["label"].iloc[te_idx]) fold_ics.append(ic) fold_ics = [float(x) for x in fold_ics] print(" fold ICs:", [round(x, 4) for x in fold_ics]) print("[4/6] train on <=2018, predict out-of-sample 2019+ ...") model = LGBMRegressor(n_estimators=400, num_leaves=31, learning_rate=0.05, subsample=0.8, colsample_bytree=0.8, min_child_samples=50, n_jobs=-1, verbose=-1) model.fit(df[feat_cols].values[train_mask], df["label"].values[train_mask]) df["pred"] = model.predict(df[feat_cols].values) is_ic = rank_ic(df["pred"][train_mask], df["label"][train_mask]) oos_ic = rank_ic(df["pred"][test_mask], df["label"][test_mask]) print(f" in-sample Rank IC = {is_ic:.4f} out-of-sample Rank IC = {oos_ic:.4f}") # decile forward returns (out-of-sample) — replaces the old hindsight charts oos = df[test_mask].copy() oos["decile"] = oos.groupby(level="datetime")["pred"].transform( lambda s: pd.qcut(s.rank(method="first"), 10, labels=False) + 1 ) decile_ret = oos.groupby("decile")["label"].mean() print("[5/6] out-of-sample decile mean next-month excess return:") print(decile_ret.round(4).to_string()) # ------------------------------------------------------------------ # build monthly long-top-N position and backtest with cost # ------------------------------------------------------------------ pred_wide = df["pred"].unstack(level="instrument") # 每月選 pred 最高的 TOP_N 檔 ranks = pred_wide.rank(axis=1, ascending=False) position = ranks <= TOP_N position = position.reindex(columns=data.get("price:收盤價").columns).fillna(False) print("[6/6] sim() full-period + OOS, benchmark 0050 含息 ...") report = sim(position, resample="M", fee_ratio=0.001425, tax_ratio=0.003, position_limit=0.1, upload=False, name="ML 排序選股") eq = report.creturn eq.index = pd.to_datetime(eq.index) # benchmark 0050 含息 (etl:adj_close) buy-and-hold, aligned windows adj = data.get("etl:adj_close") bench0050 = adj["0050"].dropna() bench0050.index = pd.to_datetime(bench0050.index) def slice_eq(s, start, end): s = s[(s.index >= pd.Timestamp(start)) & (s.index <= pd.Timestamp(end))] return s / s.iloc[0] # full strategy window full_start, full_end = eq.index[0], eq.index[-1] eq_full = slice_eq(eq, full_start, full_end) b_full = slice_eq(bench0050, full_start, full_end) # out-of-sample window eq_oos = slice_eq(eq, TEST_START, full_end) b_oos = slice_eq(bench0050, TEST_START, full_end) # in-sample window eq_is = slice_eq(eq, full_start, TRAIN_END) def stats(equity): dr = equity.pct_change().dropna() return dict( cagr=round(float(cagr(equity)) * 100, 2), daily_sharpe=round(float(daily_sharpe(dr)), 3), monthly_sharpe=round(float(monthly_sharpe(equity)), 3), max_drawdown=round(float(max_drawdown(equity)) * 100, 2), total_return=round(float(equity.iloc[-1] / equity.iloc[0] - 1) * 100, 2), ) # turnover turn = position.astype(int).diff().abs().sum(axis=1) / position.sum(axis=1).replace(0, np.nan) avg_turnover = float(turn.mean()) metrics = { "data_end": DATA_END, "train_end": TRAIN_END, "test_start": TEST_START, "label": "excess_over_median (monthly cross-sectional next-month excess return)", "model": "LightGBM regressor (gradient boosted trees)", "top_n": TOP_N, "n_features": len(feat_cols), "features": feat_cols, "rank_ic": { "in_sample": round(float(is_ic), 4), "out_of_sample": round(float(oos_ic), 4), "cpcv_folds_mean": round(float(np.nanmean(fold_ics)), 4), "cpcv_folds": [round(x, 4) for x in fold_ics], }, "strategy_full": {**stats(eq_full), "window": f"{full_start.date()} ~ {full_end.date()}", "avg_turnover_monthly": round(avg_turnover, 3)}, "strategy_in_sample": {**stats(eq_is), "window": f"{full_start.date()} ~ {TRAIN_END}"}, "strategy_out_of_sample": {**stats(eq_oos), "window": f"{TEST_START} ~ {full_end.date()}"}, "benchmark_0050_full": {**stats(b_full), "name": "0050 含息", "window": f"{full_start.date()} ~ {full_end.date()}"}, "benchmark_0050_out_of_sample": {**stats(b_oos), "name": "0050 含息", "window": f"{TEST_START} ~ {full_end.date()}"}, "decile_oos_next_month_excess_return_pct": { int(k): round(float(v) * 100, 3) for k, v in decile_ret.items() }, "feature_importance": { feat_cols[i]: int(model.feature_importances_[i]) for i in np.argsort(model.feature_importances_)[::-1] }, } with open(os.path.join(OUT_DIR, "metrics.json"), "w") as fp: json.dump(metrics, fp, ensure_ascii=False, indent=2) print("wrote metrics.json") # data.csv = monthly equity curves + decile table (downloadable, version-controlled) out = pd.DataFrame({ "ml_strategy_equity": eq_full, "benchmark_0050_equity": b_full.reindex(eq_full.index).ffill(), }) out.index.name = "date" out.to_csv(os.path.join(OUT_DIR, "data.csv")) print("wrote data.csv") # interactive report HTML (uploaded to R2 by the publisher; gitignored) try: report.to_html(os.path.join(OUT_DIR, "report_strategy.html"), title="機器學習排序選股 回測報告") print("wrote report_strategy.html") except Exception as exc: # noqa: BLE001 print("report.to_html() skipped:", exc) artifacts = dict( eq_full=eq_full, b_full=b_full, eq_oos=eq_oos, b_oos=b_oos, decile_ret=decile_ret, fold_ics=fold_ics, metrics=metrics, feat_importance=model.feature_importances_, feat_cols=feat_cols, ) return artifacts def make_charts(a): """Produce >=4 legible 16:9 matplotlib PNG charts.""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from matplotlib.ticker import FuncFormatter plt.rcParams.update({ "font.family": "sans-serif", "font.sans-serif": ["Arial Unicode MS", "PingFang TC", "Heiti TC", "sans-serif"], "axes.unicode_minus": False, "figure.dpi": 120, }) STRAT, BENCH = "#2563EB", "#9CA3AF" POS, NEG = "#10B981", "#EF4444" FIG = (12.8, 7.2) # 16:9 eq_oos, b_oos = a["eq_oos"], a["b_oos"] eq_full, b_full = a["eq_full"], a["b_full"] m = a["metrics"] # rebase OOS to common start = 1.0 common = eq_oos.index.union(b_oos.index) e = eq_oos.reindex(common).ffill(); e = e / e.iloc[0] b = b_oos.reindex(common).ffill(); b = b / b.iloc[0] # --- Chart 1: OOS equity vs 0050 (the honest result) --- fig, ax = plt.subplots(figsize=FIG) ax.plot(e.index, e.values, color=STRAT, lw=2.2, label="機器學習排序選股") ax.plot(b.index, b.values, color=BENCH, lw=2.2, label="0050 含息(買進持有)") ax.set_yscale("log") ax.set_title("樣本外(2019–2026)累積淨值:機器學習選股 vs 0050 含息(log)", fontsize=17, pad=26) ax.set_ylabel("累積淨值(起始 = 1，log)", fontsize=12, labelpad=14) so = m["strategy_out_of_sample"]; bo = m["benchmark_0050_out_of_sample"] ax.annotate(f"ML 樣本外 CAGR {so['cagr']}%／日夏普 {so['daily_sharpe']}", xy=(0.015, 0.93), xycoords="axes fraction", color=STRAT, fontsize=11) ax.annotate(f"0050 含息 CAGR {bo['cagr']}%／日夏普 {bo['daily_sharpe']}", xy=(0.015, 0.87), xycoords="axes fraction", color="#6B7280", fontsize=11) ax.legend(loc="upper center", bbox_to_anchor=(0.5, -0.10), ncol=2, frameon=False, fontsize=12) ax.grid(alpha=0.25) fig.subplots_adjust(top=0.84, bottom=0.17) fig.savefig(os.path.join(OUT_DIR, "chart_oos_equity_vs_0050.png"), dpi=200, bbox_inches="tight") plt.close(fig) # --- Chart 2: drawdown (underwater) OOS, strategy vs 0050 --- fig, ax = plt.subplots(figsize=FIG) dd_e = (e / e.cummax() - 1) * 100 dd_b = (b / b.cummax() - 1) * 100 ax.fill_between(dd_e.index, dd_e.values, 0, color=STRAT, alpha=0.30, label="機器學習排序選股") ax.plot(dd_e.index, dd_e.values, color=STRAT, lw=1.4) ax.plot(dd_b.index, dd_b.values, color=NEG, lw=1.8, label="0050 含息") ax.set_title("樣本外回撤曲線:機器學習選股的最大回撤更深(風險未被補償)", fontsize=17, pad=26) ax.set_ylabel("回撤(%)", fontsize=12, labelpad=14) ax.annotate(f"ML 最大回撤 {so['max_drawdown']}%　vs　0050 {bo['max_drawdown']}%", xy=(0.015, 0.06), xycoords="axes fraction", color="#374151", fontsize=11) ax.legend(loc="upper center", bbox_to_anchor=(0.5, -0.10), ncol=2, frameon=False, fontsize=12) ax.grid(alpha=0.25) fig.subplots_adjust(top=0.84, bottom=0.17) fig.savefig(os.path.join(OUT_DIR, "chart_oos_drawdown.png"), dpi=200, bbox_inches="tight") plt.close(fig) # --- Chart 3: OOS decile forward-return chart (replaces old hindsight charts) --- dr = a["decile_ret"] * 100 fig, ax = plt.subplots(figsize=FIG) colors = [NEG if i < 5 else POS for i in range(len(dr))] bars = ax.bar([str(int(i)) for i in dr.index], dr.values, color=colors, alpha=0.9) ax.set_title("樣本外驗證:模型分數第 10 組(看好)次月超額報酬最高", fontsize=17, pad=26) ax.set_xlabel("模型預測分數分組(1 = 最不看好 → 10 = 最看好)", fontsize=12, labelpad=12) ax.set_ylabel("次月相對中位數超額報酬(月平均，%)", fontsize=12, labelpad=14) for bar, v in zip(bars, dr.values): ax.annotate(f"{v:.2f}", xy=(bar.get_x() + bar.get_width()/2, v), ha="center", va="bottom", fontsize=10, color="#374151") ax.annotate(f"樣本外 Rank IC = {m['rank_ic']['out_of_sample']}", xy=(0.015, 0.93), xycoords="axes fraction", color="#374151", fontsize=11) ax.grid(alpha=0.25, axis="y") fig.subplots_adjust(top=0.84, bottom=0.15) fig.savefig(os.path.join(OUT_DIR, "chart_oos_decile_returns.png"), dpi=200, bbox_inches="tight") plt.close(fig) # --- Chart 4: CPCV fold Rank IC (robustness) --- fic = a["fold_ics"] fig, ax = plt.subplots(figsize=FIG) xs = list(range(1, len(fic) + 1)) cols = [POS if v > 0 else NEG for v in fic] ax.bar(xs, fic, color=cols, alpha=0.85) mean_ic = float(np.nanmean(fic)) ax.axhline(mean_ic, color=STRAT, lw=2, ls="--", label=f"平均 Rank IC = {mean_ic:.3f}") ax.axhline(0, color="#9CA3AF", lw=1) ax.set_title("穩健性:CPCV 15 折樣本外 Rank IC 全為正(訊號非單一切分運氣)", fontsize=16, pad=26) ax.set_xlabel("CPCV 折次(共 15 個樣本外組合)", fontsize=12, labelpad=12) ax.set_ylabel("該折橫截面 Rank IC", fontsize=12, labelpad=14) ax.set_xticks(xs) ax.legend(loc="upper center", bbox_to_anchor=(0.5, -0.10), ncol=1, frameon=False, fontsize=12) ax.grid(alpha=0.25, axis="y") fig.subplots_adjust(top=0.84, bottom=0.17) fig.savefig(os.path.join(OUT_DIR, "chart_cpcv_fold_ic.png"), dpi=200, bbox_inches="tight") plt.close(fig) # --- Chart 5: in-sample vs out-of-sample (the overfit gap) --- fig, ax = plt.subplots(figsize=FIG) labels = ["樣本內 CAGR", "樣本外 CAGR", "樣本內日夏普", "樣本外日夏普", "樣本內 Rank IC", "樣本外 Rank IC"] si, so2 = m["strategy_in_sample"], m["strategy_out_of_sample"] vals = [si["cagr"], so2["cagr"], si["daily_sharpe"], so2["daily_sharpe"], m["rank_ic"]["in_sample"], m["rank_ic"]["out_of_sample"]] cols = [NEG, STRAT, NEG, STRAT, NEG, STRAT] bars = ax.bar(range(len(vals)), vals, color=cols, alpha=0.9) ax.set_yscale("symlog") ax.set_xticks(range(len(labels))) ax.set_xticklabels(labels, fontsize=10, rotation=12) ax.set_title("過度配適的代價:樣本內亮眼、樣本外大幅縮水(symlog 軸)", fontsize=16, pad=26) ax.set_ylabel("數值(symlog)", fontsize=12, labelpad=14) for bar, v in zip(bars, vals): ax.annotate(f"{v:g}", xy=(bar.get_x() + bar.get_width()/2, v), ha="center", va="bottom", fontsize=10, color="#374151") ax.grid(alpha=0.25, axis="y") fig.subplots_adjust(top=0.84, bottom=0.18) fig.savefig(os.path.join(OUT_DIR, "chart_insample_vs_oos.png"), dpi=200, bbox_inches="tight") plt.close(fig) # --- Chart 6: feature importance --- fi = a["feat_importance"]; fc = a["feat_cols"] order = np.argsort(fi) name_map = { "rsi": "RSI 相對強弱", "willr": "威廉指標", "mom": "120 日動能", "cci": "CCI", "adx": "ADX 趨勢強度", "price_pos": "股價 / 240 日均", "vol_ratio": "量能比", "pb": "股價淨值比", "pe": "本益比", "rev_yoy": "月營收年增率", "rev_mom": "月營收月增率", "foreign": "外資買賣超", } names = [name_map.get(fc[i], fc[i]) for i in order] fig, ax = plt.subplots(figsize=FIG) ax.barh(names, np.array(fi)[order], color=STRAT, alpha=0.85) ax.set_title("模型重視哪些因子:價量位置與估值並重(非單純追技術指標)", fontsize=16, pad=26) ax.set_xlabel("LightGBM 特徵重要度(分裂次數)", fontsize=12, labelpad=12) ax.grid(alpha=0.25, axis="x") fig.subplots_adjust(top=0.86, left=0.20) fig.savefig(os.path.join(OUT_DIR, "chart_feature_importance.png"), dpi=200, bbox_inches="tight") plt.close(fig) print("wrote 6 charts") if __name__ == "__main__": art = main() make_charts(art)