機器學習策略完整流程

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

流程概覽

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 提供多種標籤生成方式。

2.1 分類標籤(預測漲跌方向)

from finlab.ml import label as mll

close = data.get('price:收盤價')

# 產生分類標籤(預測未來 5 日是否上漲)
cls_labels = mll.cls_label(
    close,
    n=5,           # 未來 5 個交易日
    method='rank', # 使用排名法
    resample='W'   # 週度標籤
)

print(cls_labels.head())
# 輸出:
#                                    label
# (2010-01-04 00:00:00, '1101')      1
# (2010-01-04 00:00:00, '1102')      0
# (2010-01-04 00:00:00, '1103')      1

# 檢查標籤分布
print(cls_labels['label'].value_counts())
# 輸出:
# 0    75234  # 下跌或持平
# 1    74766  # 上漲

2.2 回歸標籤(預測報酬率)

# 產生回歸標籤(預測未來 5 日報酬率)
reg_labels = mll.reg_label(
    close,
    n=5,
    resample='W'
)

print(reg_labels.head())
# 輸出:
#                                    label
# (2010-01-04 00:00:00, '1101')    0.032   # 未來 5 日上漲 3.2%
# (2010-01-04 00:00:00, '1102')   -0.015   # 未來 5 日下跌 1.5%

# 檢查標籤分布
print(reg_labels['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.3 多期預測標籤

# 同時預測 1 週、2 週、1 個月的報酬率
multi_period_labels = mlf.combine({
    'label_1w': mll.reg_label(close, n=5, resample='W'),
    'label_2w': mll.reg_label(close, n=10, resample='W'),
    'label_1m': mll.reg_label(close, n=20, resample='W')
}, resample='W')

print(multi_period_labels.head())

---

階段 3: 資料集準備與切分

3.1 對齊特徵與標籤

import pandas as pd

# 對齊時間與股票代號
aligned_data = all_features.join(cls_labels, how='inner')

print(f"對齊前特徵筆數: {len(all_features)}")
print(f"對齊前標籤筆數: {len(cls_labels)}")
print(f"對齊後總筆數: {len(aligned_data)}")

# 移除包含 NaN 的列
aligned_data = aligned_data.dropna()
print(f"移除 NaN 後: {len(aligned_data)}")

3.2 切分訓練集與測試集

# 方法 1: 時間切分(更符合實務)
train_end_date = '2022-12-31'
test_start_date = '2023-01-01'

train_data = aligned_data[aligned_data.index.get_level_values(0) <= train_end_date]
test_data = aligned_data[aligned_data.index.get_level_values(0) >= test_start_date]

print(f"訓練集: {len(train_data)} 筆 ({train_data.index.get_level_values(0).min()} ~ {train_data.index.get_level_values(0).max()})")
print(f"測試集: {len(test_data)} 筆 ({test_data.index.get_level_values(0).min()} ~ {test_data.index.get_level_values(0).max()})")

# 分離特徵與標籤
X_train = train_data.drop(columns=['label'])
y_train = train_data['label']

X_test = test_data.drop(columns=['label'])
y_test = test_data['label']

print(f"訓練特徵維度: {X_train.shape}")
print(f"測試特徵維度: {X_test.shape}")

---

階段 4: 模型訓練

4.1 使用 LightGBM 訓練分類模型

from finlab.ml.qlib import train

# 訓練 LightGBM 分類器
model, feature_importance = train(
    X_train,
    y_train,
    model='lightgbm',
    objective='binary',      # 二元分類
    num_boost_round=500,     # 迭代次數
    early_stopping_rounds=50,
    verbose_eval=50
)

print("訓練完成!")
print(f"模型類型: {type(model)}")

# 查看特徵重要性 top 20
print("Top 20 重要特徵:")
print(feature_importance.head(20))
# 輸出:
#                  importance
# rev_momentum         856.2
# pb                   742.1
# talib.RSI_14         689.3
# ...

4.2 使用 XGBoost 訓練回歸模型

# 訓練 XGBoost 回歸器(預測報酬率)
model, feature_importance = train(
    X_train,
    y_train,
    model='xgboost',
    objective='reg:squarederror',
    num_boost_round=300,
    early_stopping_rounds=30,
    learning_rate=0.05,
    max_depth=6,
    subsample=0.8,
    colsample_bytree=0.8
)

4.3 模型評估

# 訓練集預測
train_pred = model.predict(X_train)

# 測試集預測
test_pred = model.predict(X_test)

# 評估分類模型
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

print("訓練集表現:")
print(f"  準確率: {accuracy_score(y_train, train_pred > 0.5):.4f}")
print(f"  精確率: {precision_score(y_train, train_pred > 0.5):.4f}")
print(f"  召回率: {recall_score(y_train, train_pred > 0.5):.4f}")
print(f"  F1 分數: {f1_score(y_train, train_pred > 0.5):.4f}")

print("\n測試集表現:")
print(f"  準確率: {accuracy_score(y_test, test_pred > 0.5):.4f}")
print(f"  精確率: {precision_score(y_test, test_pred > 0.5):.4f}")
print(f"  召回率: {recall_score(y_test, test_pred > 0.5):.4f}")
print(f"  F1 分數: {f1_score(y_test, test_pred > 0.5):.4f}")

# 輸出範例:
# 測試集表現:
#   準確率: 0.5623
#   精確率: 0.5712
#   召回率: 0.6234
#   F1 分數: 0.5962

---

階段 5: 預測與持倉權重生成

5.1 產生預測分數

# 對所有資料(包含測試集)進行預測
all_pred = model.predict(aligned_data.drop(columns=['label']))

# 轉換為 DataFrame
pred_df = pd.DataFrame({
    'pred_score': all_pred
}, index=aligned_data.index)

print(pred_df.head())
# 輸出:
#                                    pred_score
# (2010-01-04 00:00:00, '1101')         0.687
# (2010-01-04 00:00:00, '1102')         0.423
# (2010-01-04 00:00:00, '1103')         0.812

# 檢查預測分數分布
print(pred_df['pred_score'].describe())

5.2 轉換為持倉權重

# 方法 1: Top N 選股
def pred_to_position_topn(pred_df, top_n=50):
    """選出預測分數最高的 N 檔股票"""
    position = pred_df.groupby(level=0).apply(
        lambda x: (x['pred_score'].rank(ascending=False) <= top_n).astype(float) / top_n
    )
    return position.unstack()

position_topn = pred_to_position_topn(pred_df, top_n=30)

# 方法 2: 分數加權
def pred_to_position_weighted(pred_df, threshold=0.5):
    """根據預測分數分配權重(分數越高權重越大)"""
    # 只保留分數 > threshold 的股票
    filtered = pred_df[pred_df['pred_score'] > threshold].copy()

    # 每個日期的分數加總
    date_sum = filtered.groupby(level=0)['pred_score'].sum()

    # 計算權重(分數 / 當日總分)
    position = filtered.groupby(level=0).apply(
        lambda x: x['pred_score'] / date_sum[x.index[0][0]]
    )
    return position.unstack()

position_weighted = pred_to_position_weighted(pred_df, threshold=0.6)

print(f"Top 30 策略持股數: {(position_topn > 0).sum(axis=1).mean():.1f}")
print(f"加權策略平均持股數: {(position_weighted > 0).sum(axis=1).mean():.1f}")

---

階段 6: 回測驗證

6.1 執行回測

from finlab.backtest import sim

# 回測 Top 30 策略
report_topn = sim(
    position_topn,
    name="ML Top 30 策略",
    upload=False
)

# 回測加權策略
report_weighted = sim(
    position_weighted,
    name="ML 加權策略",
    upload=False
)

# 顯示績效
report_topn.display()

6.2 績效比較

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)
# 輸出範例:
#             Top 30 策略    加權策略
# 年化報酬率      0.215      0.198
# 夏普率          1.42       1.35
# 最大回撤       -0.287     -0.265
# 勝率            0.587      0.574

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 分析特徵重要性

# 移除重要性 < 10 的特徵
important_features = feature_importance[feature_importance['importance'] > 10].index
X_train_filtered = X_train[important_features]
X_test_filtered = X_test[important_features]

print(f"原始特徵數: {X_train.shape[1]}")
print(f"過濾後特徵數: {X_train_filtered.shape[1]}")

# 重新訓練
model_v2, _ = train(
    X_train_filtered,
    y_train,
    model='lightgbm',
    objective='binary'
)

# 重新預測與回測
test_pred_v2 = model_v2.predict(X_test_filtered)
print(f"V2 測試集 F1 分數: {f1_score(y_test, test_pred_v2 > 0.5):.4f}")

7.2 調整標籤預測期

# 測試不同預測期(5日、10日、20日)
for n_days in [5, 10, 20]:
    labels_n = mll.cls_label(close, n=n_days, method='rank', resample='W')

    # 訓練與評估
    data_n = all_features.join(labels_n, how='inner').dropna()
    X_n = data_n.drop(columns=['label'])
    y_n = data_n['label']

    model_n, _ = train(X_n[:100000], y_n[:100000], model='lightgbm', objective='binary')
    pred_n = model_n.predict(X_n[100000:])

    print(f"\n預測 {n_days} 日 F1 分數: {f1_score(y_n[100000:], pred_n > 0.5):.4f}")

---

階段 8: 實盤部署

8.1 建立實時預測流程

from finlab import data
import pickle

# 1. 儲存模型
with open('ml_model.pkl', 'wb') as f:
    pickle.dump(model, f)

# 2. 建立實時特徵計算函數
def get_latest_features():
    """取得最新的特徵資料"""
    # 重新計算所有特徵(使用最新資料)
    pb_ratio = data.get('price_earning_ratio:股價淨值比')
    pe_ratio = data.get('price_earning_ratio:本益比')
    # ... 其他特徵

    latest_features = mlf.combine({
        'pb': pb_ratio,
        'pe': pe_ratio,
        # ...
    }, resample='W')

    # 取得最後一筆(最新)
    latest = latest_features.iloc[-1]
    return latest

# 3. 載入模型並預測
with open('ml_model.pkl', 'rb') as f:
    loaded_model = pickle.load(f)

latest_features = get_latest_features()
latest_pred = loaded_model.predict(latest_features[important_features])

# 4. 轉換為持倉
latest_position = pred_to_position_topn(
    pd.DataFrame({'pred_score': latest_pred}, index=latest_features.index),
    top_n=30
)

print("最新持股建議:")
print(latest_position[latest_position > 0].sort_values(ascending=False))

8.2 自動交易設定

from finlab.online.sinopac_account import SinopacAccount
from finlab.online.order_executor import OrderExecutor

# 建立定期執行腳本(每週一執行)
def weekly_rebalance():
    # 取得最新特徵
    latest_features = get_latest_features()

    # 預測
    latest_pred = loaded_model.predict(latest_features[important_features])

    # 轉換為持倉
    latest_position = pred_to_position_topn(
        pd.DataFrame({'pred_score': latest_pred}, index=latest_features.index),
        top_n=30
    )

    # 建立虛擬 report(用於下單)
    # 注意:這裡簡化處理,實務上需要更完整的 report 物件
    from finlab.backtest import sim
    report = sim(latest_position.iloc[-1:], 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
from finlab.ml.qlib import train
from finlab.backtest import sim
import pandas as pd

# 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=3))
}, resample='W')

# 2. 標籤生成
labels = mll.cls_label(close, n=5, method='rank', resample='W')

# 3. 資料準備
data_ml = features.join(labels, how='inner').dropna()
train_data = data_ml[data_ml.index.get_level_values(0) <= '2022-12-31']
test_data = data_ml[data_ml.index.get_level_values(0) >= '2023-01-01']

X_train = train_data.drop(columns=['label'])
y_train = train_data['label']
X_test = test_data.drop(columns=['label'])
y_test = test_data['label']

# 4. 模型訓練
model, feature_importance = train(
    X_train, y_train,
    model='lightgbm',
    objective='binary',
    num_boost_round=300
)

# 5. 預測與持倉
all_pred = model.predict(data_ml.drop(columns=['label']))
pred_df = pd.DataFrame({'pred_score': all_pred}, index=data_ml.index)

def pred_to_position(pred_df, top_n=30):
    position = pred_df.groupby(level=0).apply(
        lambda x: (x['pred_score'].rank(ascending=False) <= top_n).astype(float) / top_n
    )
    return position.unstack()

position = pred_to_position(pred_df, top_n=30)

# 6. 回測
report = sim(position, name="ML Strategy", upload=False)
report.display()

# 7. 分析
report.run_analysis('LiquidityAnalysis')
report.display_mae_mfe_analysis()

print("完成!")

---

關鍵要點總結

特徵工程階段

• ✅ 多樣化特徵來源(基本面、技術面、自訂)
• ✅ 檢查並處理缺失值
• ✅ 特徵數量控制(過多會導致過度配適)

標籤生成階段

• ✅ 標籤定義要符合交易邏輯
• ✅ 檢查標籤分布平衡
• ✅ 預測期要合理(太短噪音大,太長難預測)

模型訓練階段

• ✅ 時間切分訓練/測試集(不是隨機切分)
• ✅ 使用 early stopping 避免過度配適
• ✅ 關注測試集表現,不只訓練集

回測驗證階段

• ✅ 樣本外測試必做
• ✅ 執行深度分析(流動性、MAE/MFE)
• ✅ 與傳統策略比較

實盤部署階段

• ✅ 定期重新訓練模型(例如每季)
• ✅ 監控實盤與回測的偏差
• ✅ 設定績效預警機制

---

常見錯誤處理檢查清單

機器學習策略開發過程中，以下是關鍵錯誤檢查點：

階段 1：特徵工程

常見錯誤： - ❌ 特徵與標籤日期範圍不一致 - ❌ 缺失值過多導致訓練資料不足 - ❌ 未來函數（使用未來資料預測過去）

檢查方法：

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

        # 自動填充缺失值
        features = features.ffill()

    # 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：標籤生成

常見錯誤： - ❌ 標籤分布不平衡（全部為 0 或 1） - ❌ 標籤與特徵日期不對齊 - ❌ 預測期設定不合理

檢查方法：

# 生成標籤
labels = mll.cls_label(close, n=5, method='rank', resample='W')

# 檢查標籤分布
label_dist = labels.value_counts()
print("標籤分布:")
print(label_dist)

# 檢查是否平衡
label_ratio = label_dist.min() / label_dist.max()
if label_ratio < 0.3:
    print(f"⚠️  警告：標籤不平衡（比例 {label_ratio:.2f} < 0.3）")
    print("建議：")
    print("1. 使用 method='rank'（排名法）")
    print("2. 調整 n 參數（預測期）")
    print("3. 檢查資料是否完整")

# 檢查標籤缺失值
if labels.isna().sum() > len(labels) * 0.1:
    print("⚠️  警告：標籤缺失值 > 10%")
    print("可能原因：預測期過長，近期資料無標籤")

# 檢查特徵與標籤對齊
data_ml = features.join(labels, how='inner')
if len(data_ml) < len(features) * 0.8:
    print(f"⚠️  警告：特徵與標籤對齊後損失 {(1 - len(data_ml)/len(features)):.1%} 資料")
    print("建議：檢查 resample 參數是否一致")

print(f"✅ 標籤生成完成：{len(labels)} 筆資料")

階段 3：模型訓練

常見錯誤： - ❌ 訓練資料不足（< 1000 筆） - ❌ 訓練集與測試集日期錯誤（測試集在前、訓練集在後） - ❌ 過度配適（測試集表現遠差於訓練集）

檢查方法：

# 切分訓練/測試集
train_data = data_ml[data_ml.index.get_level_values(0) <= '2022-12-31']
test_data = data_ml[data_ml.index.get_level_values(0) >= '2023-01-01']

# 1. 檢查資料量
print(f"訓練集：{len(train_data)} 筆")
print(f"測試集：{len(test_data)} 筆")

if len(train_data) < 1000:
    print("⚠️  警告：訓練資料不足（< 1000 筆）")
    print("建議：")
    print("1. 增加歷史資料範圍")
    print("2. 降低 resample 頻率（如 '1d' 改為 'W'）")

if len(test_data) < 100:
    print("⚠️  警告：測試資料過少（< 100 筆）")

# 2. 檢查日期順序
train_last_date = train_data.index.get_level_values(0).max()
test_first_date = test_data.index.get_level_values(0).min()

if train_last_date >= test_first_date:
    raise ValueError(
        f"❌ 訓練集與測試集日期重疊！\n"
        f"   訓練集最後日期：{train_last_date}\n"
        f"   測試集第一日期：{test_first_date}\n"
        f"   這會導致資料洩露（data leakage）"
    )

print(f"✅ 資料切分正確")

# 3. 模型訓練與驗證
X_train = train_data.drop(columns=['label'])
y_train = train_data['label']
X_test = test_data.drop(columns=['label'])
y_test = test_data['label']

try:
    model, feature_importance = train(
        X_train, y_train,
        model='lightgbm',
        objective='binary',
        num_boost_round=300
    )

    # 檢查訓練/測試表現
    train_score = model.predict(X_train).mean()
    test_score = model.predict(X_test).mean()

    print(f"訓練集平均預測：{train_score:.4f}")
    print(f"測試集平均預測：{test_score:.4f}")

    # 檢查過度配適
    if abs(train_score - test_score) > 0.1:
        print("⚠️  警告：訓練與測試表現差異過大，可能過度配適")
        print("建議：")
        print("1. 減少特徵數量")
        print("2. 減少 num_boost_round")
        print("3. 增加正則化（lambda_l1, lambda_l2）")

    print(f"✅ 模型訓練完成")

except Exception as e:
    print(f"❌ 模型訓練失敗：{e}")
    print("請檢查：")
    print("1. 特徵是否包含 NaN 或 Inf")
    print("2. 標籤是否為數值型態")
    print("3. qlib 是否正確安裝（pip install pyqlib）")
    raise

階段 4：預測與回測

常見錯誤： - ❌ 預測結果全為 NaN - ❌ 持倉 DataFrame 格式錯誤 - ❌ 回測無交易記錄

檢查方法：

# 1. 檢查預測結果
all_pred = model.predict(data_ml.drop(columns=['label']))

if pd.Series(all_pred).isna().all():
    raise ValueError("❌ 預測結果全為 NaN")

pred_df = pd.DataFrame({'pred_score': all_pred}, index=data_ml.index)

# 檢查預測分布
print(f"預測分數範圍：{pred_df['pred_score'].min():.4f} ~ {pred_df['pred_score'].max():.4f}")
print(f"預測平均值：{pred_df['pred_score'].mean():.4f}")

# 2. 檢查持倉格式
def pred_to_position(pred_df, top_n=30):
    position = pred_df.groupby(level=0).apply(
        lambda x: (x['pred_score'].rank(ascending=False) <= top_n).astype(float) / top_n
    )
    return position.unstack()

position = pred_to_position(pred_df, top_n=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, name="ML Strategy", upload=False)

    trades = report.get_trades()
    if len(trades) == 0:
        raise ValueError("❌ 策略無任何交易記錄")

    print(f"✅ 回測成功：{len(trades)} 筆交易")
    print(f"   年化報酬：{report.get_stats()['annual_return']:.2%}")
    print(f"   夏普率：{report.get_stats()['daily_sharpe']:.2f}")

except Exception as e:
    print(f"❌ 回測失敗：{e}")
    print("請檢查：")
    print("1. position 的 index 是否為 DatetimeIndex")
    print("2. position 的 columns 是否為股票代號")
    print("3. position 的值是否為 0-1 之間的權重")
    raise

機器學習策略特有風險

相比傳統策略，ML 策略需額外注意：

• ❌ 資料洩露（data leakage）- 使用未來資料預測過去
• ❌ 過度配適（overfitting）- 測試集表現遠差於訓練集
• ❌ 模型衰退（model decay）- 實盤表現隨時間衰退

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

---

參考資源

• 機器學習簡易教學
• finlab.ml API Reference
• 完整策略開發流程
• 策略分析模組
• 實盤交易教學
• 風險管理完整指南
• 常見問題 FAQ