第5章 機器學習在交易中的應用

# 第5章 機器學習在交易中的應用 本章聚焦於**機器學習**（ML）在量化交易中的實際運用。\n 從監督式學習（回歸、分類、特徵工程）到無監督式學習（聚類、PCA），再到強化學習（RL）與自動交易系統，我們將逐步剖析理論基礎、實務實作與風險控制，並提供完整的程式碼範例與部署建議。 --- ## 5.1 監督式學習 監督式學習的核心是「有標籤」資料：\n- **特徵（X）**：代表市場狀態的數值或向量。\n- **目標（y）**：預測的值（回歸）或類別（分類）。\n ### 5.1.1 回歸 | 目的 | 典型模型 | 評估指標 | |------|----------|----------| | 價格預測 | Linear Regression, RandomForestRegressor, XGBoostRegressor | MAE / RMSE | | 波動率預測 | ARIMA + ML, GARCH + ML | RMSE, MAPE | #### 範例：使用 XGBoost 進行日內價格回歸 python import pandas as pd import xgboost as xgb from sklearn.model_selection import TimeSeriesSplit, cross_val_score # 讀取 OHLCV 資料 price_df = pd.read_csv('data/stock_A_daily.csv', parse_dates=['date'], index_col='date') # 特徵工程：技術指標 price_df['MA_5'] = price_df['close'].rolling(5).mean() price_df['MA_20'] = price_df['close'].rolling(20).mean() price_df['RSI'] = ta.momentum.RSIIndicator(price_df['close'], window=14).rsi() price_df['vol_change'] = price_df['volume'].pct_change() price_df.dropna(inplace=True) X = price_df[['MA_5', 'MA_20', 'RSI', 'vol_change']] y = price_df['close'].shift(-1).dropna() X = X.iloc[:-1] model = xgb.XGBRegressor(n_estimators=500, learning_rate=0.05, max_depth=4) tscv = TimeSeriesSplit(n_splits=5) rmse_scores = -cross_val_score(model, X, y, cv=tscv, scoring='neg_root_mean_squared_error') print('RMSE (10% CV):', rmse_scores.mean()) ### 5.1.2 分類 | 目的 | 典型模型 | 評估指標 | |------|----------|----------| | 方向預測 | Logistic Regression, SVM, Gradient Boosting, LSTM | Accuracy, F1, ROC‑AUC | | 事件預測 | RandomForestClassifier, CatBoost, XGBoost | Precision, Recall | #### 範例：方向預測（漲跌） python import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report # 讀取資料 df = pd.read_csv('data/stock_A_daily.csv', parse_dates=['date'], index_col='date') # 產生目標：1 為漲，0 為跌 prev_close = df['close'].shift(1) df['target'] = (df['close'] > prev_close).astype(int) # 特徵：技術指標 for window in [5, 10, 20]: df[f'MA_{window}'] = df['close'].rolling(window).mean() df[f'RSI_{window}'] = ta.momentum.RSIIndicator(df['close'], window=window).rsi() df[f'Stdev_{window}'] = df['close'].rolling(window).std() # 填補缺失 df.dropna(inplace=True) X = df[[f'MA_{w}' for w in [5,10,20]] + [f'RSI_{w}' for w in [5,10,20]] + [f'Stdev_{w}' for w in [5,10,20]]] y = df['target'] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, shuffle=False) scaler = StandardScaler() X_train = scaler.fit_transform(X_train) X_test = scaler.transform(X_test) clf = RandomForestClassifier(n_estimators=300, max_depth=8, random_state=42) clf.fit(X_train, y_train) pred = clf.predict(X_test) print(classification_report(y_test, pred)) ### 5.1.3 特徵工程 | 步驟 | 目的 | 常用技術 | |------|------|-----------| | 缺失處理 | 防止模型崩潰 | 均值/中位數/插值 | | 標準化 | 促進收斂 | StandardScaler, MinMaxScaler | | 交互項 | 捕捉非線性關係 | PolynomialFeatures, InteractionTerm | | 變數選擇 | 降維 + 提升穩定性 | Lasso, Ridge, Tree‑based feature importance | | 時間序列特徵 | 捕捉趨勢 | Lag features, rolling aggregates | > **實務提示**：對於時間序列，**不要**在訓練資料中使用未來資訊（Look‑ahead Bias）。 --- ## 5.2 無監督式學習 無監督式學習不需要目標變數，常用於 **市場分段、異常偵測** 或 **特徵降維**。 ### 5.2.1 聚類（Clustering） | 方法 | 適用場景 | 優點 | |------|----------|------| | K‑Means | 股票群組、資產分級 | 直觀、快速 | | DBSCAN | 异常偵測、密度聚類 | 免需指定 k | | GMM | 需求概率分布 | 可輸出成分隱含概率 | #### 範例：使用 K‑Means 進行股票風格分類 python import pandas as pd from sklearn.preprocessing import StandardScaler from sklearn.cluster import KMeans # 讀取多支股票的因子資料 factors = pd.read_csv('data/factors.csv') # 因子：市值、ROE、PE、PB、Beta X = factors[['market_cap', 'roe', 'pe', 'pb', 'beta']] scaler = StandardScaler() X_scaled = scaler.fit_transform(X) kmeans = KMeans(n_clusters=4, random_state=42) labels = kmeans.fit_predict(X_scaled) factors['style_cluster'] = labels print(factors.groupby('style_cluster').mean()) ### 5.2.2 主成分分析（PCA） | 目的 | 優點 | |------|------| | 降維 | 降低維度、消除多重共線 | | 特徵提取 | 抽取最具信息量的組合 | | 噪聲抑制 | 去除高頻噪音 | #### 範例：對日內行情進行 PCA python import pandas as pd from sklearn.decomposition import PCA from sklearn.preprocessing import StandardScaler # 讀取日內 5 分鐘資料 df = pd.read_csv('data/stock_A_intraday.csv') # 取收盤價、成交量、波動幅度 X = df[['close', 'volume', 'high_low_diff']] scaler = StandardScaler() X_scaled = scaler.fit_transform(X) pca = PCA(n_components=2) principal = pca.fit_transform(X_scaled) print('Explained variance:', pca.explained_variance_ratio_) --- ## 5.3 強化學習與自動交易系統 強化學習（RL）適合**動態決策**問題：在不斷變化的市場中學習最佳行動策略。\n典型架構： - **Agent**：策略模型（如 DQN、PPO）。\n- **Environment**：市場模擬器（帶有交易成本、滑點）。\n- **Reward**：日常收益、風險調整後的獎勵。\n ### 5.3.1 RL 基礎 | 參數 | 含義 | |------|------| | State | 當前市場狀態（價格、技術指標、持倉）。 | | Action | 交易指令（買、賣、持有或多頭、空頭幅度）。 | | Reward | 交易收益（扣除手續費、滑點）。 | | Policy | 取樣行動的機率分佈（π）。 | | Value | 估計未來獲得獎勵的價值（V）。 | #### 範例：簡易 DQN 訓練流程（伪代码） python import numpy as np import gym import torch import torch.nn as nn import torch.optim as optim # 1. 建立環境 class StockEnv(gym.Env): def __init__(self, data, window_size): self.data = data self.window_size = window_size self.action_space = gym.spaces.Discrete(3) # 0: sell, 1: hold, 2: buy self.observation_space = gym.spaces.Box(low=-np.inf, high=np.inf, shape=(window_size, 5)) self.reset() def reset(self): self.idx = self.window_size return self._get_state() def step(self, action): reward = self._calculate_reward(action) self.idx += 1 done = self.idx >= len(self.data) - 1 return self._get_state(), reward, done, {} # ... (implement _get_state, _calculate_reward) ... # 2. 建立 Q‑network class QNet(nn.Module): def __init__(self, state_dim, action_dim): super().__init__() self.fc = nn.Sequential( nn.Linear(state_dim, 128), nn.ReLU(), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, action_dim) ) def forward(self, x): return self.fc(x) # 3. 訓練迴圈 env = StockEnv(data, window_size=20) q_net = QNet(state_dim=20*5, action_dim=3) optimizer = optim.Adam(q_net.parameters(), lr=1e-3) criterion = nn.MSELoss() for episode in range(200): state = env.reset() total_reward = 0 done = False while not done: state_tensor = torch.FloatTensor(state).unsqueeze(0) q_values = q_net(state_tensor) action = torch.argmax(q_values).item() # ε‑greedy 可自行加入 next_state, reward, done, _ = env.step(action) # 計算 TD 目標 next_q = q_net(torch.FloatTensor(next_state).unsqueeze(0)).max(1)[0].detach() target = reward + 0.99 * next_q loss = criterion(q_values[0, action], target) optimizer.zero_grad(); loss.backward(); optimizer.step() state = next_state total_reward += reward print(f'Episode {episode}, Total Reward: {total_reward}') > **部署建議**：將已訓練模型封裝成 REST API，利用 **Docker** + **Kubernetes** 進行水平擴展，確保實時低延遲。\n> **風險控制**：在策略部署前，務必將策略塞進「沙盒」環境，進行「walk‑forward」回測與「風險因子測試」。 ### 5.3.2 自動交易系統框架 | 組件 | 作用 | |------|------| | 資料源 | 市場行情、基本面、新聞情緒 | | 資料倉儲 | Time‑Series DB（如 kdb+, InfluxDB） | | 計算層 | 風險模型、策略回測、模型評估 | | 執行層 | 交易接口（FIX、REST、WebSocket） | | 監控層 | 風險監控、性能監控、合規報告 | #### 案例：使用 Zipline + Catalyst 進行自動化交易 python # zipline_backtest.py from zipline.api import (order_target_percent, record, symbol, get_datetime) import pandas as pd def initialize(context): context.security = symbol('AAPL') context.target = 0.5 def handle_data(context, data): price = data.current(context.security, 'price') # 簡單均線交叉策略 sma_short = data.history(context.security, 'price', bar_count=20, frequency='1d').mean() sma_long = data.history(context.security, 'price', bar_count=50, frequency='1d').mean() if sma_short > sma_long: order_target_percent(context.security, context.target) else: order_target_percent(context.security, -context.target) record(price=price) if __name__ == '__main__': start = pd.Timestamp('2015-01-01', tz='utc') end = pd.Timestamp('2020-12-31', tz='utc') from zipline import run_algorithm run_algorithm(start=start, end=end, initialize=initialize, capital_base=100000, handle_data=handle_data) --- ## 5.4 風險與效能考量 | 觀點 | 建議 | |------|------| | **過擬合** | 使用時間序列 CV、早停、參數正則化 | | **資料偏差** | 避免回測期間的資料泄露，加入交易成本、滑點 | | **計算成本** | 將模型分層：特徵工程（CPU）、模型推論（GPU） | | **監控** | 監控策略收益、持倉、滑點、回撤；實時報警 | | **合規** | 日誌完整、交易行為可審計；遵守交易所規範 | --- ## 5.5 小結 - **監督式學習** 主要解決「價格/波動率預測」與「方向判斷」；重點是高品質特徵與嚴謹的評估。\n- **無監督式學習** 可協助市場分段、異常偵測與特徵降維，減少模型輸入維度。\n- **強化學習** 適用於需動態調整頭寸的策略，關鍵是構建真實且公平的環境、設計合理的 reward。\n- **部署** 必須考慮 **延遲、可擴展性** 與 **風險控制**；使用 Docker、K8s、API Gateway 可實現高可用。\n- **風險與效能** 的持續監控是長期成功的關鍵，策略須在真實交易前經過多層驗證與風險限額檢查。 > **提示**：在實際交易前，先在「沙盒」環境進行「walk‑forward」回測，再逐步擴大投資規模；任何機器學習模型都可能隨市場演化而失效，持續更新與重訓是必要的。