跳轉到

機器學習策略完整流程

本文提供機器學習量化策略的完整開發流程,從特徵工程、標籤生成、模型訓練,到回測驗證與實盤部署。

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)- 實盤表現隨時間衰退

建議: - ✅ 嚴格使用時間序列切分(不是隨機切分) - ✅ 定期重新訓練模型(每季或每月) - ✅ 監控實盤與回測偏差,設定預警機制


參考資源