文件
開啟 Studio

完整流程

機器學習策略

本頁內容12

這一頁帶你做出一個機器學習選股策略:把基本面、技術面與營收資料合成特徵,產生「未來一週報酬」的標籤,訓練模型,把預測值最高的 30 檔變成持股並回測,最後每週自動重新預測並下單。

做完這一頁,你會有:一張週頻的特徵表與標籤、一個訓練好並存成 ml_model.pkl 的模型、它在 2023 年之後的回測報告,以及每週執行一次的換股函式。

開始前你需要:

  • 安裝 Qlib 相依套件:pip install 'finlab[qlib]',一次裝好 Qlib 相依套件與 LightGBM、XGBoost 等模型函式庫,之後才能使用 finlab.ml.qlib。
  • 會用 sim() 回測(見歷史回測)。第一次接觸 finlab.ml,可以先看較短的機器學習入門。
  • 測試資料從 2023 年開始。免費帳號的台股資料到 2023 年底,VIP 每天更新到最新。

這頁以台股示範,美股的完整程式碼在「完整程式碼」一節的美股分頁。每個階段都可以請 AI 編程助手幫忙:安裝 FinLab Skill 後,它能提供程式碼範例與除錯協助。

整個流程如下,績效不滿意或沒通過樣本外測試,就回頭調整特徵或模型:

否

是

否

是

原始資料

特徵工程
finlab.ml.feature

標籤生成
finlab.ml.label

資料集切分

模型訓練
finlab.ml.qlib

預測持倉權重

回測驗證

績效滿意?

調整特徵/模型

樣本外測試

通過驗證?

實盤部署

1. 建立特徵

機器學習的成功與否,80% 取決於特徵工程。finlab.ml.feature(下面簡稱 mlf)的 mlf.combine() 把多張資料表合成一張特徵表:索引是 (datetime, instrument) 兩層,每個特徵一欄,resample='W' 表示重採樣為週度資料。

財報與月營收的日期,就是資料公布、你拿得到的日子(月營收是次月 10 日,申報期限),所以特徵直接用就好,不會偷看未來。

基本面特徵

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:ROE稅後')
roa = data.get('fundamental_features:ROA稅後息前')

# 合併為特徵集
fundamental_features = mlf.combine(
    {'pb': pb_ratio, 'pe': pe_ratio, 'roe': roe, 'roa': roa}, resample='W'
)  # 重採樣為週度資料

print(fundamental_features.head())  # 索引 (datetime, instrument),欄位 pb、pe、roe、roa

技術指標特徵

兩種做法:讓 mlf.ta_names() 隨機產生指標參數來探索,或指定你要的指標。

# 方法 1: 使用隨機技術指標參數(探索最佳參數)
names = [
    name
    for name in mlf.ta_names(n=3)  # 每個指標隨機產生 3 種參數配置
    if name.startswith(('talib.RSI__', 'talib.MACD__', 'talib.ATR__', 'talib.ADX__'))
]
ta_features = mlf.ta(names, resample='W')

print(f"技術指標特徵數: {ta_features.shape[1]}")

# 方法 2: 指定特定技術指標
from finlab import data

rsi = data.indicator('RSI', timeperiod=14)
macd, macd_signal, macd_hist = data.indicator('MACD')
upper, middle, lower = data.indicator('BBANDS', timeperiod=20)
obv = data.indicator('OBV')

specific_ta = mlf.combine(
    {
        'rsi': rsi,
        'macd': macd,
        'bb_width': (upper - lower) / middle,
        'obv': obv,
    },
    resample='W',
)

自訂特徵

任何算得出來的表都能當特徵,例如營收動能:

# 營收相關特徵
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())

合併所有特徵並移除缺值太多的

# 合併所有特徵集
all_features = mlf.combine(
    {
        'fundamental': fundamental_features,
        'technical': ta_features,
        'custom': custom_features,
    },
    resample='W',
)

print(f"總特徵數: {all_features.shape[1]}")
print(f"資料筆數: {all_features.shape[0]}")

# 檢查缺失值
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(下面簡稱 mll)的函數都接受 features.index(MultiIndex)作為第一個參數,回傳和特徵同樣索引的 Series。先選一種:

標籤適合
return_percentage最常用,直接預測報酬率
excess_over_median預測相對表現,減少大盤漲跌影響
daytrading_percentage適合當沖策略
maximum_adverse_excursion適合風險管理模型

period 要與策略調倉頻率一致:resample='W', period=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) 索引的 Series

# 檢查標籤分布
print(label.describe())

超額報酬

# 相對同期全市場收益中位數的超額收益
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())

其他標籤

# 日內交易報酬(開盤到收盤的變化)
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)

3. 依時間切分訓練與測試資料

特徵(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%}")

再以 2023-01-01 為界,之前的資料訓練、之後的資料測試:

# 使用時間切分(嚴格避免資料洩露)
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. 訓練模型

LightGBM

finlab.ml.qlib(下面簡稱 q)把模型包成 fit() 與 predict() 兩步:

import finlab.ml.qlib as q

# 建立並訓練 LightGBM 模型
model = q.LGBModel()
model.fit(X_train, y_train)

print("訓練完成!")

其他模型

用法都一樣。線性模型快、不易過擬合,適合先建立基準:

# 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()))

一次比較多個模型

每個模型都選預測值最高的 30 檔回測,比較年化報酬與夏普率:

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['cagr']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}"
    )

5. 把預測變成持股

model.predict() 回傳 FinlabDataFrame(index 為日期、columns 為股票代號),可以直接用 FinlabDataFrame 的方法轉換為持倉。

產生預測

# 對測試集進行預測
y_pred = model.predict(X_test)

print(y_pred.head())  # FinlabDataFrame:index 為日期,columns 為股票代號

# 檢查預測分布
print(y_pred.stack().describe())

轉換為持股

三種常見做法:

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. 回測驗證

回測

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()

比較兩種持股方式

import pandas as pd

stats_topn = report_topn.get_stats()
stats_weighted = report_weighted.get_stats()

comparison = pd.DataFrame(
    {
        'Top 30 策略': [
            stats_topn['cagr'],
            stats_topn['daily_sharpe'],
            stats_topn['max_drawdown'],
            stats_topn['win_ratio'],
        ],
        '加權策略': [
            stats_weighted['cagr'],
            stats_weighted['daily_sharpe'],
            stats_weighted['max_drawdown'],
            stats_weighted['win_ratio'],
        ],
    },
    index=['年化報酬率', '夏普率', '最大回撤', '勝率'],
)

print(comparison)

深度分析

各分析的解讀見策略分析模組:

# 流動性分析
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. 調整特徵與標籤

結果不滿意時,可以從特徵數量、預測期與標籤類型三個方向調整。

減少特徵數量

# 策略 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()

調整預測期

# 測試不同預測期
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['cagr']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}"
    )

換一種標籤

# 比較報酬率 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['cagr']:.2%}, 夏普率 {stats['daily_sharpe']:.2f}"
    )

8. 部署到實盤

用全部歷史資料訓練並存檔

上線前用所有歷史資料重新訓練,把模型存檔,再用它預測最新一期的持股:

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:ROE稅後'),
        '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)

每週自動換股

每週重算特徵、載入模型預測,再用 sim 產生報告、算出部位並下單。這個例子用永豐帳戶,需先設定環境變數,其他券商見實盤下單:

from finlab.backtest import sim
from finlab.online.sinopac_account import SinopacAccount
from finlab.online.order_executor import OrderExecutor, Position

# 建立定期執行腳本(每週一執行)
def weekly_rebalance():
    # 重新計算特徵
    features = mlf.combine(
        {
            'pb': data.get('price_earning_ratio:股價淨值比'),
            'pe': data.get('price_earning_ratio:本益比'),
            'roe': data.get('fundamental_features:ROE稅後'),
            '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()
    target = Position.from_report(report, fund=1000000)  # 總資金 100 萬
    executor = OrderExecutor(target, account=account)
    executor.create_orders()

# 使用 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)
實盤部署定期重新訓練模型(例如每季);監控實盤與回測的偏差;設定績效預警機制

常見錯誤與檢查方法

每個階段都可以加上檢查,及早發現問題。

特徵工程:resample 不一致、缺失值過多、未來函數

常見錯誤:

  • 特徵與標籤的 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("請至 /database 確認正確名稱")

except ValueError as e:
    print(f"❌ 特徵驗證失敗:{e}")

資料下載的錯誤處理見下載資料的「常見錯誤與解決方法」。

標籤生成:resample 不一致、傳錯 index、預測期不合理

常見錯誤:

  • 標籤與特徵的 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)} 筆資料")
模型訓練:資料不足、隨機切分、過度配適

常見錯誤:

  • 訓練資料不足(< 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. 是否已安裝 Qlib 相依套件(pip install 'finlab[qlib]')")
    raise
預測與回測:預測全為 NaN、持倉格式錯誤、沒有交易

常見錯誤:

  • 預測結果全為 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['cagr']:.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

下一步

其他相關頁面:策略分析模組、實盤下單、常見問題。

↑↓ 選擇 Enter 開啟 Esc 關閉