機器學習策略完整流程
本文提供機器學習量化策略的完整開發流程,從特徵工程、標籤生成、模型訓練,到回測驗證與實盤部署。
ML 策略開發推薦搭配 AI 助手
機器學習策略涉及特徵工程、標籤設計、模型訓練等多個環節。安裝 FinLab Skill 後,AI 編程助手可在每個階段提供程式碼範例與除錯協助。
流程概覽
graph TD
A[原始資料] --> B[特徵工程<br/>finlab.ml.feature]
B --> C[標籤生成<br/>finlab.ml.label]
C --> D[資料集切分]
D --> E[模型訓練<br/>finlab.ml.qlib]
E --> F[預測持倉權重]
F --> G[回測驗證]
G --> H{績效滿意?}
H -->|否| I[調整特徵/模型]
I --> B
H -->|是| J[樣本外測試]
J --> K{通過驗證?}
K -->|否| I
K -->|是| L[實盤部署]
階段 1: 特徵工程
機器學習的成功與否,80% 取決於特徵工程。FinLab 提供強大的特徵工程工具。
1.1 載入與合併基本面特徵
from finlab import data
from finlab.ml import feature as mlf
# 載入基本面資料
pb_ratio = data.get('price_earning_ratio:股價淨值比')
pe_ratio = data.get('price_earning_ratio:本益比')
roe = data.get('fundamental_features:股東權益報酬率')
roa = data.get('fundamental_features:資產報酬率')
# 合併為特徵集
fundamental_features = mlf.combine({
'pb': pb_ratio,
'pe': pe_ratio,
'roe': roe,
'roa': roa
}, resample='W') # 重採樣為週度資料
print(fundamental_features.head())
# 輸出:
# pb pe roe roa
# (2010-01-04 00:00:00, '1101') 1.47 18.85 7.80 3.21
# (2010-01-04 00:00:00, '1102') 1.44 14.58 9.87 4.15
1.2 新增技術指標特徵
# 方法 1: 使用隨機技術指標(探索最佳指標)
ta_features = mlf.combine({
'talib': mlf.ta(mlf.ta_names(n=5)) # 每個指標隨機產生 5 種參數配置
}, resample='W')
# 方法 2: 指定特定技術指標
from finlab import data
close = data.get('price:收盤價')
volume = data.get('price:成交股數')
specific_ta = mlf.combine({
'rsi': close.ta.RSI(timeperiod=14),
'macd': close.ta.MACD(),
'bbands': close.ta.BBANDS(timeperiod=20),
'obv': volume.ta.OBV(close)
}, resample='W')
print(f"技術指標特徵數: {ta_features.shape[1]}") # 例如: 450 個特徵
1.3 新增自訂特徵
# 營收相關特徵
rev = data.get('monthly_revenue:當月營收')
rev_yoy = data.get('monthly_revenue:去年同月增減(%)')
custom_features = mlf.combine({
'rev_ma3': rev.average(3), # 3 個月營收移動平均
'rev_ma12': rev.average(12), # 12 個月營收移動平均
'rev_momentum': rev.average(3) / rev.average(12), # 營收動能
'rev_yoy': rev_yoy, # 營收年增率
}, resample='W')
print(custom_features.head())
1.4 合併所有特徵
# 合併所有特徵集
all_features = mlf.combine({
'fundamental': fundamental_features,
'technical': ta_features,
'custom': custom_features
}, resample='W')
print(f"總特徵數: {all_features.shape[1]}") # 例如: 470 個特徵
print(f"資料筆數: {all_features.shape[0]}") # 例如: 150,000 筆
# 檢查缺失值
missing_ratio = all_features.isna().sum() / len(all_features)
print(f"缺失值比例:\n{missing_ratio[missing_ratio > 0.5]}") # 顯示缺失超過 50% 的特徵
# 移除缺失值過多的特徵
all_features = all_features.loc[:, missing_ratio < 0.5]
print(f"過濾後特徵數: {all_features.shape[1]}")
階段 2: 標籤生成
標籤定義了我們的預測目標。finlab.ml.label 提供多種標籤生成函數,所有函數都接受 features.index(MultiIndex)作為第一個參數。
2.1 預測未來報酬率
from finlab.ml import label as mll
# 預測未來 1 週報酬率(最常用)
label = mll.return_percentage(all_features.index, resample='W', period=1)
print(label.head())
# 輸出:
# datetime instrument
# 2010-01-04 00:00:00 1101 0.032
# 1102 -0.015
# 1103 0.021
# dtype: float64
# 檢查標籤分布
print(label.describe())
# 輸出:
# count 150000.00
# mean 0.005
# std 0.087
# min -0.450
# 25% -0.042
# 50% 0.002
# 75% 0.051
# max 0.520
2.2 超額報酬標籤
# 相對同期全市場收益中位數的超額收益
label_excess_median = mll.excess_over_median(all_features.index, resample='W', period=1)
# 相對同期全市場收益均值的超額收益
label_excess_mean = mll.excess_over_mean(all_features.index, resample='W', period=1)
print(label_excess_median.describe())
2.3 其他標籤類型
# 日內交易報酬(開盤到收盤的變化)
label_daytrading = mll.daytrading_percentage(all_features.index)
# 風險指標:最大不利偏移(持有期內最大跌幅)
label_mae = mll.maximum_adverse_excursion(all_features.index, period=5)
# 風險指標:最大有利偏移(持有期內最大漲幅)
label_mfe = mll.maximum_favorable_excursion(all_features.index, period=5)
# 多期預測(預測不同時間跨度)
label_1w = mll.return_percentage(all_features.index, resample='W', period=1)
label_2w = mll.return_percentage(all_features.index, resample='W', period=2)
label_4w = mll.return_percentage(all_features.index, resample='W', period=4)
標籤選擇建議
return_percentage: 最常用,直接預測報酬率excess_over_median: 預測相對表現,減少大盤漲跌影響daytrading_percentage: 適合當沖策略maximum_adverse_excursion: 適合風險管理模型period與策略調倉頻率一致:如resample='W', period=1代表預測一週後的報酬
階段 3: 資料集準備與切分
3.1 對齊特徵與標籤
特徵(all_features)與標籤(label)共用相同的 MultiIndex(datetime, instrument),可直接進行切分。
# 選用標籤
label = mll.return_percentage(all_features.index, resample='W', period=1)
# 檢查對齊狀況
print(f"特徵筆數: {len(all_features)}")
print(f"標籤筆數: {len(label)}")
print(f"標籤 NaN 比例: {label.isna().mean():.2%}")
3.2 切分訓練集與測試集
# 使用時間切分(嚴格避免資料洩露)
is_train = all_features.index.get_level_values('datetime') < '2023-01-01'
X_train = all_features[is_train]
y_train = label[is_train]
X_test = all_features[~is_train]
print(f"訓練集: {len(X_train)} 筆 ({X_train.index.get_level_values(0).min()} ~ {X_train.index.get_level_values(0).max()})")
print(f"測試集: {len(X_test)} 筆 ({X_test.index.get_level_values(0).min()} ~ {X_test.index.get_level_values(0).max()})")
print(f"訓練特徵維度: {X_train.shape}")
時間切分很重要
必須用時間切分,不可用隨機切分。隨機切分會導致資料洩露——模型看到未來的資料來預測過去,導致回測表現虛高。
階段 4: 模型訓練
4.1 使用 LightGBM 模型
import finlab.ml.qlib as q
# 建立並訓練 LightGBM 模型
model = q.LGBModel()
model.fit(X_train, y_train)
print("訓練完成!")
4.2 使用其他模型
# XGBoost
model_xgb = q.XGBModel()
model_xgb.fit(X_train, y_train)
# CatBoost
model_cat = q.CatBoostModel()
model_cat.fit(X_train, y_train)
# 線性模型(快速驗證、不易過擬合)
model_linear = q.LinearModel()
model_linear.fit(X_train, y_train)
# 深度學習
model_dnn = q.DNNModel()
model_dnn.fit(X_train, y_train)
# 列出所有可用模型
models = q.get_models()
print(list(models.keys()))
4.3 多模型比較
import finlab.ml.qlib as q
from finlab.backtest import sim
# 快速比較多個模型
results = {}
for name, ModelClass in [('LightGBM', q.LGBModel), ('XGBoost', q.XGBModel), ('Linear', q.LinearModel)]:
model = ModelClass()
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
position = y_pred.is_largest(30)
report = sim(position, resample='W', name=f"ML {name}", upload=False)
results[name] = report
# 比較績效
for name, report in results.items():
stats = report.get_stats()
print(f"{name}: 年化報酬 {stats['daily_mean']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}")
階段 5: 預測與持倉權重生成
model.predict() 回傳 FinlabDataFrame(index 為日期、columns 為股票代號),可直接使用 FinlabDataFrame 的方法轉換為持倉。
5.1 產生預測
# 對測試集進行預測
y_pred = model.predict(X_test)
print(y_pred.head())
# 輸出(FinlabDataFrame,index=日期,columns=股票代號):
# 1101 1102 1103 1216 2330
# 2023-01-06 0.032 0.015 -0.008 0.045 0.023
# 2023-01-13 0.018 0.027 0.003 0.012 0.041
# 檢查預測分布
print(y_pred.stack().describe())
5.2 轉換為持倉
from finlab.backtest import sim
# 方法 1: Top N 選股(買入預測值最高的 N 檔)
position_topn = y_pred.is_largest(30)
# 方法 2: 前 20% 選股
position_quantile = y_pred > y_pred.quantile(0.8)
# 方法 3: 根據預測值分配權重
position_weighted = y_pred / y_pred.sum()
print(f"Top 30 策略平均持股數: {position_topn.sum(axis=1).mean():.1f}")
print(f"前 20% 策略平均持股數: {position_quantile.sum(axis=1).mean():.1f}")
階段 6: 回測驗證
6.1 執行回測
from finlab.backtest import sim
# 回測 Top 30 策略
report_topn = sim(
position_topn,
resample='W',
name="ML Top 30 策略",
upload=False
)
# 回測加權策略
report_weighted = sim(
position_weighted,
resample='W',
name="ML 加權策略",
upload=False
)
# 顯示績效
report_topn.display()
6.2 績效比較
import pandas as pd
stats_topn = report_topn.get_stats()
stats_weighted = report_weighted.get_stats()
comparison = pd.DataFrame({
'Top 30 策略': [
stats_topn['daily_mean'],
stats_topn['daily_sharpe'],
stats_topn['max_drawdown'],
stats_topn['win_ratio']
],
'加權策略': [
stats_weighted['daily_mean'],
stats_weighted['daily_sharpe'],
stats_weighted['max_drawdown'],
stats_weighted['win_ratio']
]
}, index=['年化報酬率', '夏普率', '最大回撤', '勝率'])
print(comparison)
6.3 深度分析
# 流動性分析
report_topn.run_analysis('LiquidityAnalysis', required_volume=100000)
# MAE/MFE 分析
report_topn.display_mae_mfe_analysis()
# 時期穩定性
report_topn.run_analysis('PeriodStatsAnalysis')
# Alpha/Beta
report_topn.run_analysis('AlphaBetaAnalysis')
階段 7: 特徵工程迭代優化
7.1 減少特徵數量
# 策略 1: 使用較少的技術指標
features_small = mlf.combine({
'pb': pb_ratio,
'pe': pe_ratio,
'roe': roe,
'talib': mlf.ta(mlf.ta_names(n=1)[:20]) # 只取前 20 個指標
}, resample='W')
label_small = mll.return_percentage(features_small.index, resample='W', period=1)
is_train_small = features_small.index.get_level_values('datetime') < '2023-01-01'
model_v2 = q.LGBModel()
model_v2.fit(features_small[is_train_small], label_small[is_train_small])
y_pred_v2 = model_v2.predict(features_small[~is_train_small])
position_v2 = y_pred_v2.is_largest(30)
report_v2 = sim(position_v2, resample='W', name="ML V2 精簡特徵", upload=False)
report_v2.display()
7.2 調整標籤預測期
# 測試不同預測期
for period in [1, 2, 4]:
label_n = mll.return_percentage(all_features.index, resample='W', period=period)
model_n = q.LGBModel()
model_n.fit(X_train, label_n[is_train])
y_pred_n = model_n.predict(X_test)
position_n = y_pred_n.is_largest(30)
report_n = sim(position_n, resample='W', name=f"ML 預測{period}週", upload=False)
stats = report_n.get_stats()
print(f"預測 {period} 週: 年化報酬 {stats['daily_mean']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}")
7.3 嘗試不同標籤類型
# 比較報酬率 vs 超額報酬標籤
label_return = mll.return_percentage(all_features.index, resample='W', period=1)
label_excess = mll.excess_over_median(all_features.index, resample='W', period=1)
for label_name, label_data in [('報酬率', label_return), ('超額報酬', label_excess)]:
model_cmp = q.LGBModel()
model_cmp.fit(X_train, label_data[is_train])
y_pred_cmp = model_cmp.predict(X_test)
position_cmp = y_pred_cmp.is_largest(30)
report_cmp = sim(position_cmp, resample='W', name=f"ML {label_name}", upload=False)
stats = report_cmp.get_stats()
print(f"{label_name}: 年化報酬 {stats['daily_mean']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}")
階段 8: 實盤部署
8.1 建立實時預測流程
from finlab import data
from finlab.ml import feature as mlf, label as mll
import finlab.ml.qlib as q
import pickle
# 1. 訓練完整模型(使用所有歷史資料)
features = mlf.combine({
'pb': data.get('price_earning_ratio:股價淨值比'),
'pe': data.get('price_earning_ratio:本益比'),
'roe': data.get('fundamental_features:股東權益報酬率'),
'talib': mlf.ta(mlf.ta_names(n=1)[:20])
}, resample='W')
label = mll.return_percentage(features.index, resample='W', period=1)
model = q.LGBModel()
model.fit(features, label)
# 2. 儲存模型
with open('ml_model.pkl', 'wb') as f:
pickle.dump(model, f)
# 3. 載入模型並預測最新持倉
with open('ml_model.pkl', 'rb') as f:
loaded_model = pickle.load(f)
y_pred = loaded_model.predict(features)
# 4. 取得最新持倉
position = y_pred.is_largest(30)
latest_position = position.iloc[-1]
latest_position = latest_position[latest_position > 0].sort_values(ascending=False)
print("最新持股建議:")
print(latest_position)
8.2 自動交易設定
from finlab.backtest import sim
from finlab.online.sinopac_account import SinopacAccount
from finlab.online.order_executor import OrderExecutor
# 建立定期執行腳本(每週一執行)
def weekly_rebalance():
# 重新計算特徵
features = mlf.combine({
'pb': data.get('price_earning_ratio:股價淨值比'),
'pe': data.get('price_earning_ratio:本益比'),
'roe': data.get('fundamental_features:股東權益報酬率'),
'talib': mlf.ta(mlf.ta_names(n=1)[:20])
}, resample='W')
# 載入模型並預測
with open('ml_model.pkl', 'rb') as f:
model = pickle.load(f)
y_pred = model.predict(features)
position = y_pred.is_largest(30)
# 用 sim 產生 report
report = sim(position, resample='W', upload=False)
# 執行下單
account = SinopacAccount(simulation=False)
executor = OrderExecutor(report=report, account=account, fund=1000000)
executor.execute()
# 使用 cron 或排程工具定期執行 weekly_rebalance()
完整程式碼彙整
# =============================================================================
# 機器學習策略完整範例
# =============================================================================
from finlab import data
from finlab.ml import feature as mlf
from finlab.ml import label as mll
import finlab.ml.qlib as q
from finlab.backtest import sim
# 1. 特徵工程
close = data.get('price:收盤價')
pb = data.get('price_earning_ratio:股價淨值比')
pe = data.get('price_earning_ratio:本益比')
rev = data.get('monthly_revenue:當月營收')
features = mlf.combine({
'pb': pb,
'pe': pe,
'rev_ma3': rev.average(3),
'rev_ma12': rev.average(12),
'talib': mlf.ta(mlf.ta_names(n=1)[:20])
}, resample='W')
# 2. 標籤生成
label = mll.return_percentage(features.index, resample='W', period=1)
# 3. 資料切分
is_train = features.index.get_level_values('datetime') < '2023-01-01'
X_train = features[is_train]
y_train = label[is_train]
X_test = features[~is_train]
# 4. 模型訓練
model = q.LGBModel()
model.fit(X_train, y_train)
# 5. 預測與持倉
y_pred = model.predict(X_test)
position = y_pred.is_largest(30)
# 6. 回測
report = sim(position, resample='W', name="ML Strategy", upload=False)
report.display()
# 7. 分析
report.run_analysis('LiquidityAnalysis')
report.display_mae_mfe_analysis()
print("完成!")
關鍵要點總結
特徵工程階段
- ✅ 多樣化特徵來源(基本面、技術面、自訂)
- ✅ 使用
mlf.combine()統一合併,確保 MultiIndex 對齊 - ✅ 檢查並處理缺失值
- ✅ 特徵數量控制(過多會導致過度配適)
標籤生成階段
- ✅ 使用
mll.return_percentage()等函數,傳入features.index - ✅
resample參數與特徵一致 - ✅ 預測期(
period)要合理(太短噪音大,太長難預測)
模型訓練階段
- ✅ 使用
q.LGBModel()等包裝類,fit()+predict() - ✅ 時間切分訓練/測試集(不是隨機切分)
- ✅ 先用簡單模型(
LinearModel)建立基準
回測驗證階段
- ✅
predict()回傳 FinlabDataFrame,用is_largest()轉為持倉 - ✅ 樣本外測試必做
- ✅ 執行深度分析(流動性、MAE/MFE)
實盤部署階段
- ✅ 定期重新訓練模型(例如每季)
- ✅ 監控實盤與回測的偏差
- ✅ 設定績效預警機制
常見錯誤處理檢查清單
機器學習策略開發過程中,以下是關鍵錯誤檢查點:
階段 1:特徵工程
常見錯誤:
- ❌ 特徵與標籤的 resample 不一致
- ❌ 缺失值過多導致訓練資料不足
- ❌ 未來函數(使用未來資料預測過去)
檢查方法:
try:
# 1. 建立特徵
features = mlf.combine({
'pb': pb,
'pe': pe,
'rev_ma3': rev.average(3)
}, resample='W')
if features.empty:
raise ValueError("❌ 特徵 DataFrame 為空")
# 2. 檢查缺失值比例
missing_ratio = features.isna().sum() / len(features)
high_missing_cols = missing_ratio[missing_ratio > 0.3].index.tolist()
if high_missing_cols:
print(f"⚠️ 警告:以下特徵缺失值 > 30%:{high_missing_cols}")
print("建議:移除這些特徵或使用 forward fill")
# 3. 檢查日期範圍
print(f"特徵日期範圍:{features.index.get_level_values(0).min()} ~ {features.index.get_level_values(0).max()}")
# 4. 檢查特徵數量
num_features = features.shape[1]
if num_features > 500:
print(f"⚠️ 警告:特徵數量過多({num_features} 個),可能導致過度配適")
print("建議:< 200 個特徵為佳")
print(f"✅ 特徵工程完成:{num_features} 個特徵,{len(features)} 筆資料")
except KeyError as e:
print(f"❌ 資料表名稱錯誤:{e}")
print("請至 https://ai.finlab.tw/database 確認正確名稱")
except ValueError as e:
print(f"❌ 特徵驗證失敗:{e}")
詳細錯誤處理:參考 資料下載錯誤處理
階段 2:標籤生成
常見錯誤:
- ❌ 標籤與特徵的 resample 不一致
- ❌ 傳入錯誤的 index(應傳 features.index)
- ❌ 預測期設定不合理
檢查方法:
# 生成標籤
label = mll.return_percentage(features.index, resample='W', period=1)
# 檢查標籤分布
print("標籤統計:")
print(label.describe())
# 檢查標籤缺失值
nan_ratio = label.isna().mean()
if nan_ratio > 0.1:
print(f"⚠️ 警告:標籤缺失值 {nan_ratio:.1%} > 10%")
print("可能原因:預測期過長,近期資料無標籤")
print(f"✅ 標籤生成完成:{len(label)} 筆資料")
階段 3:模型訓練
常見錯誤: - ❌ 訓練資料不足(< 1000 筆) - ❌ 使用隨機切分而非時間切分 - ❌ 過度配適(測試集表現遠差於訓練集)
檢查方法:
# 切分訓練/測試集
is_train = features.index.get_level_values('datetime') < '2023-01-01'
X_train = features[is_train]
y_train = label[is_train]
X_test = features[~is_train]
# 1. 檢查資料量
print(f"訓練集:{len(X_train)} 筆")
print(f"測試集:{len(X_test)} 筆")
if len(X_train) < 1000:
print("⚠️ 警告:訓練資料不足(< 1000 筆)")
print("建議:增加歷史資料範圍或降低 resample 頻率")
if len(X_test) < 100:
print("⚠️ 警告:測試資料過少(< 100 筆)")
# 2. 檢查日期順序
train_last = X_train.index.get_level_values(0).max()
test_first = X_test.index.get_level_values(0).min()
if train_last >= test_first:
raise ValueError(
f"❌ 訓練集與測試集日期重疊!\n"
f" 訓練集最後日期:{train_last}\n"
f" 測試集第一日期:{test_first}\n"
f" 這會導致資料洩露(data leakage)"
)
print(f"✅ 資料切分正確")
# 3. 模型訓練
try:
model = q.LGBModel()
model.fit(X_train, y_train)
print(f"✅ 模型訓練完成")
except Exception as e:
print(f"❌ 模型訓練失敗:{e}")
print("請檢查:")
print("1. 特徵是否包含 NaN 或 Inf")
print("2. 標籤是否為數值型態")
print("3. 相關套件是否正確安裝(pip install lightgbm / xgboost)")
raise
階段 4:預測與回測
常見錯誤: - ❌ 預測結果全為 NaN - ❌ 持倉 DataFrame 格式錯誤 - ❌ 回測無交易記錄
檢查方法:
# 1. 預測
y_pred = model.predict(X_test)
if y_pred.isna().all().all():
raise ValueError("❌ 預測結果全為 NaN")
# 檢查預測分布
print(f"預測值範圍:{y_pred.min().min():.4f} ~ {y_pred.max().max():.4f}")
print(f"預測平均值:{y_pred.stack().mean():.4f}")
# 2. 產生持倉
position = y_pred.is_largest(30)
if position.empty:
raise ValueError("❌ 持倉 DataFrame 為空")
holding_count = position.sum(axis=1).mean()
if holding_count < 10:
print(f"⚠️ 警告:平均持倉數 {holding_count:.1f} < 10,可能過少")
print(f"✅ 持倉生成成功:平均持有 {holding_count:.1f} 檔")
# 3. 回測
try:
report = sim(position, resample='W', name="ML Strategy", upload=False)
print(f"✅ 回測成功")
stats = report.get_stats()
print(f" 年化報酬:{stats['daily_mean']:.2%}")
print(f" 夏普率:{stats['daily_sharpe']:.2f}")
except Exception as e:
print(f"❌ 回測失敗:{e}")
print("請檢查:")
print("1. position 的 index 是否為 DatetimeIndex")
print("2. position 的 columns 是否為股票代號")
raise
機器學習策略特有風險
相比傳統策略,ML 策略需額外注意:
- ❌ 資料洩露(data leakage)- 使用未來資料預測過去
- ❌ 過度配適(overfitting)- 測試集表現遠差於訓練集
- ❌ 模型衰退(model decay)- 實盤表現隨時間衰退
建議: - ✅ 嚴格使用時間序列切分(不是隨機切分) - ✅ 定期重新訓練模型(每季或每月) - ✅ 監控實盤與回測偏差,設定預警機制