從研究到實盤:完整策略開發流程
本文提供一個完整的端到端範例,展示如何從策略構想開始,經過回測、優化、驗證,最終部署至實盤交易,涵蓋整個量化交易策略的生命週期。
流程概覽
graph TD
A[策略構想] --> B[資料探索]
B --> C[策略開發]
C --> D[策略優化]
D --> E[深度分析]
E --> F[樣本外測試]
F --> G{通過驗證?}
G -->|是| H[部署前準備]
G -->|否| C
H --> I[實盤交易]
I --> J[績效追蹤]
J --> K{需要調整?}
K -->|是| C
K -->|否| I
階段 1: 策略構想與初步研究
策略假設
我們要開發一個「營收成長 + 技術突破」的組合策略:
- 基本面: 營收近期加速成長(月均線上揚)
- 技術面: 股價突破均線,確認動能
- 風控: 跌破均線出場,設定 10% 停損
資料探索
首先檢查所需資料是否完整:
from finlab import data
import pandas as pd
# 載入所需資料
close = data.get('price:收盤價')
rev = data.get('monthly_revenue:當月營收')
# 檢查資料範圍
print(f"收盤價資料: {close.index[0]} ~ {close.index[-1]}")
print(f"營收資料: {rev.index[0]} ~ {rev.index[-1]}")
# 檢查缺失值比例
print(f"收盤價缺失值: {close.isna().sum().sum() / close.size * 100:.2f}%")
print(f"營收缺失值: {rev.isna().sum().sum() / rev.size * 100:.2f}%")
# 輸出:
# 收盤價資料: 2007-04-23 ~ 2024-12-31
# 營收資料: 2000-01-01 ~ 2024-12-01
# 收盤價缺失值: 2.15%
# 營收缺失值: 8.73%
階段 2: 策略開發與回測
撰寫策略邏輯
from finlab import data
from finlab.backtest import sim
# 1. 資料載入
close = data.get('price:收盤價')
rev = data.get('monthly_revenue:當月營收')
# 2. 計算技術指標
ma20 = close.average(20)
ma60 = close.average(60)
# 3. 計算營收指標
rev_ma3 = rev.average(3) # 近 3 個月平均營收
rev_ma12 = rev.average(12) # 近 12 個月平均營收
# 4. 定義進場條件
entry_tech = (close > ma20) & (close > ma60) # 技術面:突破均線
entry_fund = (rev_ma3 / rev_ma12) > 1.1 # 基本面:營收加速成長
entry_signal = entry_tech & entry_fund
# 5. 定義出場條件
exit_signal = close < ma20 # 跌破 20 日均線
# 6. 組合進出場訊號(含停損)
position = entry_signal.hold_until(
exit=exit_signal,
stop_loss=0.1 # 停損 10%
)
# 7. 回測
report = sim(
position,
resample='M', # 每月調整
position_limit=0.1, # 單一持股上限 10%
upload=False,
name="營收成長突破策略 v1.0"
)
# 8. 顯示績效
report.display()
初步績效評估
stats = report.get_stats()
print(f"年化報酬率: {stats['daily_mean']*100:.2f}%")
print(f"夏普率: {stats['daily_sharpe']:.2f}")
print(f"最大回撤: {stats['max_drawdown']*100:.2f}%")
print(f"勝率: {stats['win_ratio']*100:.2f}%")
# 輸出範例:
# 年化報酬率: 15.23%
# 夏普率: 0.87
# 最大回撤: -28.45%
# 勝率: 52.3%
初步判斷
- 年化報酬率 15% 不錯,但夏普率 0.87 偏低
- 最大回撤 -28% 可接受
- 接下來需要優化策略,提升風險調整後報酬
階段 3: 策略優化
3.1 使用 sim_conditions() 測試條件組合
我們新增更多候選條件,找出最佳組合:
from finlab.optimize.combinations import sim_conditions
# 原有條件
c1 = (close > ma20) & (close > ma60) # 均線多頭
c2 = rev_ma3 / rev_ma12 > 1.1 # 營收加速
# 新增條件
pe = data.get('price_earning_ratio:本益比')
c3 = pe < 15 # 低本益比
c4 = close == close.rolling(20).max() # 創 20 日新高
c5 = rev / rev.shift(1) > 0.9 # 營收月增率 > -10%
# 組合字典
conditions = {
'c1': c1, # 均線多頭
'c2': c2, # 營收加速
'c3': c3, # 低本益比
'c4': c4, # 創新高
'c5': c5 # 營收月增率
}
# 測試所有組合(31 種)
report_collection = sim_conditions(
conditions=conditions,
hold_until={'exit': exit_signal, 'stop_loss': 0.1},
resample='M',
position_limit=0.1,
upload=False
)
# 視覺化比較
report_collection.plot_stats('heatmap')
3.2 分析優化結果
# 取得績效指標表
stats_df = report_collection.get_stats()
# 找出 top 3 組合
top3 = stats_df.T.sort_values('daily_sharpe', ascending=False).head(3)
print(top3[['daily_mean', 'daily_sharpe', 'max_drawdown', 'win_ratio']])
# 輸出範例:
# daily_mean daily_sharpe max_drawdown win_ratio
# c1 & c2 & c3 0.182 1.35 -0.245 0.561
# c1 & c2 & c5 0.165 1.28 -0.267 0.548
# c1 & c2 0.152 0.87 -0.285 0.523
優化成果
c1 & c2 & c3(均線多頭 + 營收加速 + 低本益比)組合的夏普率從 0.87 提升至 1.35,大幅改善!
3.3 採用最佳組合
# 使用最佳組合重新定義策略
best_entry = c1 & c2 & c3
position_optimized = best_entry.hold_until(
exit=exit_signal,
stop_loss=0.1
)
report_opt = sim(
position_optimized,
resample='M',
position_limit=0.1,
upload=False,
name="營收成長突破策略 v2.0 (優化後)"
)
3.4 優化停損停利
透過 MAE/MFE 分析找出最佳停損停利點:
# 顯示波動分析
report_opt.display_mae_mfe_analysis()
# 取得交易記錄
trades = report_opt.get_trades()
# 分析 MAE/MFE 分位數
mae_q75 = abs(trades['mae'].quantile(0.75))
gmfe_q75 = trades['gmfe'].quantile(0.75)
print(f"MAE Q75: {mae_q75*100:.2f}%") # 例如 8.5%
print(f"GMFE Q75: {gmfe_q75*100:.2f}%") # 例如 18.3%
# 調整停損停利
position_final = best_entry.hold_until(
exit=exit_signal,
stop_loss=mae_q75 * 1.2, # 停損設為 MAE Q75 * 1.2 = 10.2%
stop_profit=gmfe_q75 * 0.8 # 停利設為 GMFE Q75 * 0.8 = 14.6%
)
report_final = sim(
position_final,
resample='M',
position_limit=0.1,
upload=False,
name="營收成長突破策略 v3.0 (最終版)"
)
階段 4: 深度分析
4.1 流動性風險檢測
檢查策略是否適合大資金:
# 流動性分析(假設資金 1000 萬,單筆進出場至少需要 10 萬股)
report_final.run_analysis('LiquidityAnalysis', required_volume=100000)
# 輸出範例:
# buy_high sell_low low_volume_stocks 處置股
# entry 0.032 0.008 0.087 0.012
# exit 0.015 0.045 0.092 0.008
流動性檢視
- 低成交量股票比例 8.7%,大資金需注意
- 處置股比例 1.2%,可接受
- 建議加入成交量篩選條件
4.2 年度穩定性分析
檢視策略在不同年度的表現:
4.3 Alpha/Beta 分析
衡量策略的超額報酬:
report_final.run_analysis('AlphaBetaAnalysis')
# 輸出範例:
# Alpha: 0.082 (8.2% 年化超額報酬)
# Beta: 0.65 (市場上漲 1%,策略預期上漲 0.65%)
Alpha/Beta 評估
- Alpha 8.2% 優秀,代表策略有真實的超額報酬
- Beta 0.65 代表策略波動小於大盤,風險較低
階段 5: 樣本外測試
將資料切分為訓練集與測試集,驗證策略是否過度配適:
# 定義切分日期
train_end = '2022-12-31'
test_start = '2023-01-01'
# 訓練集回測(2018-2022)
position_train = position_final[position_final.index <= train_end]
report_train = sim(
position_train,
resample='M',
position_limit=0.1,
upload=False,
name="訓練集(2018-2022)"
)
# 測試集回測(2023-2024)
position_test = position_final[position_final.index >= test_start]
report_test = sim(
position_test,
resample='M',
position_limit=0.1,
upload=False,
name="測試集(2023-2024)"
)
# 績效對比
stats_train = report_train.get_stats()
stats_test = report_test.get_stats()
comparison = pd.DataFrame({
'訓練集': [
stats_train['daily_mean'],
stats_train['daily_sharpe'],
stats_train['max_drawdown'],
stats_train['win_ratio']
],
'測試集': [
stats_test['daily_mean'],
stats_test['daily_sharpe'],
stats_test['max_drawdown'],
stats_test['win_ratio']
]
}, index=['年化報酬率', '夏普率', '最大回撤', '勝率'])
print(comparison)
# 輸出範例:
# 訓練集 測試集
# 年化報酬率 0.182 0.165
# 夏普率 1.35 1.21
# 最大回撤 -0.245 -0.267
# 勝率 0.561 0.542
樣本外判斷標準
| 指標 | 訓練集 vs 測試集 | 判斷 |
|---|---|---|
| 年化報酬率 | 18.2% vs 16.5% | ✅ 衰退合理(<20%) |
| 夏普率 | 1.35 vs 1.21 | ✅ 略降但仍 > 1 |
| 最大回撤 | -24.5% vs -26.7% | ✅ 差異不大 |
| 勝率 | 56.1% vs 54.2% | ✅ 穩定 |
樣本外測試通過
測試集績效略降但仍保持優秀,策略沒有過度配適,可進入部署階段!
階段 6: 部署前準備
6.1 上傳策略至雲端
# 上傳最終版本至 FinLab 平台
report_final.upload(name="營收成長突破策略 v3.0")
# 取得持股清單
position_info = report_final.position_info()
print(f"下期換股日: {position_info['next_trading_date']}")
print(f"當前持股數: {len(position_info['stocks'])}")
print("持股清單:")
print(position_info['stocks'])
# 輸出範例:
# 下期換股日: 2025-01-31
# 當前持股數: 8
# 持股清單:
# stock_id position
# 0 2330 0.12
# 1 2317 0.11
# 2 2454 0.10
# ...
6.2 準備實盤資金配置
# 假設實盤資金 100 萬
capital = 1000000
# 計算每檔股票應投入的金額
stocks = position_info['stocks']
stocks['amount'] = stocks['position'] * capital
print(stocks[['stock_id', 'position', 'amount']])
# 輸出:
# stock_id position amount
# 0 2330 0.12 120000
# 1 2317 0.11 110000
# ...
階段 7: 實盤交易
7.1 設定券商帳戶(以永豐證券為例)
from finlab.online.sinopac_account import SinopacAccount
# 設定券商帳戶
account = SinopacAccount(
simulation=False, # False 為實盤,True 為模擬單
)
實盤交易注意事項
- 首次使用請先設定
simulation=True測試 - 確認券商 API 金鑰正確
- 確認有足夠的資金與額度
- 建議先小資金測試
7.2 計算目標部位
from finlab.online.order_executor import Position
# 計算目標部位(整股)
capital = 1000000 # 實盤資金 100 萬
position = Position.from_report(report_final, capital)
# 或使用零股交易
position = Position.from_report(report_final, capital, odd_lot=True)
print(position)
# 輸出: [{'stock_id': '2330', 'quantity': 5, 'order_condition': <OrderCondition.CASH: 1>}, ...]
7.3 執行下單
from finlab.online.order_executor import OrderExecutor
# 建立下單執行器
order_executor = OrderExecutor(position, account=account)
# 下單檢查(瀏覽模式,不會真的下單)
order_executor.create_orders(view_only=True)
# 執行下單(會真的下單,初次使用建議收市時測試)
order_executor.create_orders()
# 更新限價(將最後一筆成交價當成新的限價)
order_executor.update_order_price()
# 刪除所有委託單
order_executor.cancel_orders()
階段 8: 績效追蹤
8.1 取得實盤交易記錄
# 取得實盤部位
account_positions = account.get_stocks()
print(account_positions)
# 取得帳戶總市值與現金
account_value = account.get_total_balance()
print(f"帳戶總值: {account_value:,.0f}")
8.2 監控策略表現
# 定期檢查帳戶狀態(建議每日收盤後執行)
import pandas as pd
# 記錄每日帳戶淨值
daily_value = {
'date': pd.Timestamp.today(),
'account_value': account.get_total_balance(),
'positions': len(account.get_stocks())
}
print(f"日期: {daily_value['date']}")
print(f"帳戶淨值: {daily_value['account_value']:,.0f}")
print(f"持股檔數: {daily_value['positions']}")
# 與回測比較
backtest_return = report_final.get_stats()['daily_mean']
print(f"回測年化報酬率: {backtest_return * 100:.2f}%")
print(f"實盤 vs 回測差異: {(realized_return - backtest_return) * 100:.2f}%")
8.3 定期檢視
建議每月檢視以下項目:
# 1. 持股是否按計劃執行
current_stocks = set(account_positions['stock_id'])
planned_stocks = set(position_info['stocks']['stock_id'])
print(f"計劃持股: {planned_stocks}")
print(f"實際持股: {current_stocks}")
print(f"誤差股票: {planned_stocks.symmetric_difference(current_stocks)}")
# 2. 報酬率是否符合預期
print(f"當月報酬率: {account.get_monthly_return()*100:.2f}%")
# 3. 是否觸發停損
for stock_id, pos in account_positions.items():
entry_price = pos['avg_price']
current_price = pos['current_price']
unrealized_pnl = (current_price / entry_price - 1)
if unrealized_pnl < -0.10: # 停損 10%
print(f"警告: {stock_id} 已觸發停損,未實現損益: {unrealized_pnl*100:.2f}%")
完整程式碼彙整
以下是完整的可執行程式碼:
# =============================================================================
# 從研究到實盤:營收成長突破策略完整範例
# =============================================================================
from finlab import data
from finlab.backtest import sim
from finlab.optimize.combinations import sim_conditions
# -----------------------------------------------------------------------------
# 階段 1 & 2: 資料載入與策略開發
# -----------------------------------------------------------------------------
# 載入資料
close = data.get('price:收盤價')
rev = data.get('monthly_revenue:當月營收')
pe = data.get('price_earning_ratio:本益比')
# 計算指標
ma20 = close.average(20)
ma60 = close.average(60)
rev_ma3 = rev.average(3)
rev_ma12 = rev.average(12)
# 定義條件
c1 = (close > ma20) & (close > ma60) # 均線多頭
c2 = rev_ma3 / rev_ma12 > 1.1 # 營收加速
c3 = pe < 15 # 低本益比
# 出場訊號
exit_signal = close < ma20
# -----------------------------------------------------------------------------
# 階段 3: 策略優化
# -----------------------------------------------------------------------------
# 測試條件組合
conditions = {'c1': c1, 'c2': c2, 'c3': c3}
report_collection = sim_conditions(
conditions=conditions,
hold_until={'exit': exit_signal, 'stop_loss': 0.1},
resample='M',
position_limit=0.1,
upload=False
)
# 找出最佳組合
stats_df = report_collection.get_stats()
top1 = stats_df.T.sort_values('daily_sharpe', ascending=False).index[0]
print(f"最佳組合: {top1}")
# 使用最佳組合
best_entry = c1 & c2 & c3
# 優化停損停利
position_temp = best_entry.hold_until(exit=exit_signal, stop_loss=0.1)
report_temp = sim(position_temp, resample='M', position_limit=0.1, upload=False)
trades = report_temp.get_trades()
mae_q75 = abs(trades['mae'].quantile(0.75))
gmfe_q75 = trades['gmfe'].quantile(0.75)
# 最終策略
position_final = best_entry.hold_until(
exit=exit_signal,
stop_loss=mae_q75 * 1.2,
stop_profit=gmfe_q75 * 0.8
)
report_final = sim(
position_final,
resample='M',
position_limit=0.1,
upload=False,
name="營收成長突破策略 v3.0"
)
# -----------------------------------------------------------------------------
# 階段 4: 深度分析
# -----------------------------------------------------------------------------
# 流動性分析
report_final.run_analysis('LiquidityAnalysis', required_volume=100000)
# 波動分析
report_final.display_mae_mfe_analysis()
# 時期穩定性
report_final.run_analysis('PeriodStatsAnalysis')
# Alpha/Beta
report_final.run_analysis('AlphaBetaAnalysis')
# -----------------------------------------------------------------------------
# 階段 5: 樣本外測試
# -----------------------------------------------------------------------------
train_end = '2022-12-31'
test_start = '2023-01-01'
position_train = position_final[position_final.index <= train_end]
position_test = position_final[position_final.index >= test_start]
report_train = sim(position_train, resample='M', position_limit=0.1, upload=False)
report_test = sim(position_test, resample='M', position_limit=0.1, upload=False)
print("訓練集績效:", report_train.get_stats()['daily_sharpe'])
print("測試集績效:", report_test.get_stats()['daily_sharpe'])
# -----------------------------------------------------------------------------
# 階段 6: 部署前準備
# -----------------------------------------------------------------------------
# 上傳至雲端
report_final.upload(name="營收成長突破策略 v3.0")
# 取得持股清單
position_info = report_final.position_info()
print("下期持股:")
print(position_info['stocks'])
# -----------------------------------------------------------------------------
# 階段 7: 實盤交易(需先設定券商帳戶)
# -----------------------------------------------------------------------------
# from finlab.online.sinopac_account import SinopacAccount
# from finlab.online.order_executor import OrderExecutor
#
# account = SinopacAccount(simulation=True) # 先用模擬單測試
# executor = OrderExecutor(
# report=report_final,
# account=account,
# fund=1000000
# )
# executor.execute()
# -----------------------------------------------------------------------------
# 階段 8: 績效追蹤
# -----------------------------------------------------------------------------
# 定期執行以下程式碼檢視實盤狀況:
# account_value = account.get_total_value()
# print(f"帳戶總值: {account_value}")
關鍵要點總結
開發階段檢查清單
- [ ] 策略構想明確: 有清晰的邏輯假設(技術面/基本面/籌碼面)
- [ ] 資料完整性檢查: 確認所需資料範圍足夠,缺失值比例合理
- [ ] 初步回測績效可接受: 年化報酬 > 10%,夏普率 > 0.5
- [ ] 條件組合優化: 使用
sim_conditions()找出最佳組合 - [ ] 停損停利優化: 使用 MAE/MFE 分析設定合理的停損停利點
驗證階段檢查清單
- [ ] 流動性風險: 低成交量比例 < 20%,處置股比例 < 5%
- [ ] 年度穩定性: 沒有連續多年表現極差
- [ ] Alpha > 0: 策略有超額報酬
- [ ] 樣本外測試通過: 測試集績效衰退 < 30%,夏普率仍 > 1
部署階段檢查清單
- [ ] 策略已上傳雲端: 可隨時檢視持股清單
- [ ] 券商帳戶設定完成: API 金鑰正確,權限開通
- [ ] 先用模擬單測試: 確認下單邏輯正確
- [ ] 小資金測試: 先用小額資金測試 1-2 個月
- [ ] 定期檢視機制: 設定每月檢視實盤績效
後續學習建議
完成本流程後,可進一步學習:
常見錯誤處理檢查清單
在執行完整流程時,務必在每個階段進行錯誤檢查。以下是關鍵檢查點與解決方案:
階段 1-2:資料載入與回測
常見錯誤:
- ❌ KeyError: 'price:收盤價' - API token 未設定或資料表名稱錯誤
- ❌ 資料為空或缺失值過多
- ❌ 回測無交易記錄(進場條件過嚴)
檢查方法:
try:
# 1. 檢查資料載入
close = data.get('price:收盤價')
if close.empty:
raise ValueError("❌ 資料為空,請檢查 API token")
missing_ratio = close.isna().sum().sum() / close.size
if missing_ratio > 0.1:
print(f"⚠️ 警告:缺失值比例 {missing_ratio:.1%} > 10%")
# 2. 檢查進場訊號
entry_count = entry_signal.sum(axis=1).mean()
if entry_count < 1:
print("⚠️ 警告:進場條件過嚴,平均每日 < 1 檔股票")
print("建議:放寬條件或使用 .is_largest(N) 確保固定檔數")
# 3. 檢查回測結果
report = sim(position, resample='M', position_limit=0.1)
trades = report.get_trades()
if len(trades) == 0:
raise ValueError("❌ 策略無任何交易記錄,請檢查進場條件")
print(f"✅ 回測成功:{len(trades)} 筆交易")
except KeyError as e:
print(f"❌ 找不到資料表:{e}")
print("請至 https://ai.finlab.tw/database 確認資料表名稱")
print("或檢查 API token 是否正確設定")
except ValueError as e:
print(f"❌ 資料驗證失敗:{e}")
詳細錯誤處理:參考 資料下載錯誤處理
階段 3:策略優化
常見錯誤:
- ❌ sim_conditions() 組合數量過多導致執行時間過長
- ❌ 條件定義錯誤(日期範圍不一致)
檢查方法:
# 限制組合數量(< 50 個組合)
conditions = {
'c1': c1,
'c2': c2,
'c3': c3,
# 最多 5 個條件(2^5 = 32 種組合)
}
# 估算執行時間
num_combinations = 2**len(conditions) - 1
print(f"將測試 {num_combinations} 種組合,預估時間:約 {num_combinations * 30 / 60:.1f} 分鐘")
if num_combinations > 100:
print("⚠️ 警告:組合數量過多,建議先測試部分條件")
詳細錯誤處理:參考 策略優化錯誤處理
階段 5:樣本外測試
常見錯誤: - ❌ 訓練集與測試集日期範圍重疊 - ❌ 測試集績效與訓練集差異過大(過度配適)
檢查方法:
# 檢查日期範圍
train_end = '2022-12-31'
test_start = '2023-01-01'
position_train = position_final[position_final.index <= train_end]
position_test = position_final[position_final.index >= test_start]
# 確認無重疊
if position_train.index[-1] >= position_test.index[0]:
raise ValueError("❌ 訓練集與測試集日期重疊!")
# 檢查績效差異
sharpe_train = report_train.get_stats()['daily_sharpe']
sharpe_test = report_test.get_stats()['daily_sharpe']
degradation = (sharpe_train - sharpe_test) / sharpe_train
if degradation > 0.5:
print(f"⚠️ 警告:夏普率衰退 {degradation:.1%} > 50%,可能過度配適")
print("建議:")
print("1. 簡化策略條件")
print("2. 增加訓練集資料範圍")
print("3. 使用交叉驗證")
詳細錯誤處理:參考 回測錯誤處理
階段 7:實盤交易
常見錯誤: - ❌ 券商帳戶連線失敗(API key 錯誤、憑證過期) - ❌ 資金不足或持倉過多 - ❌ 訂單被拒絕(處置股未圈存、漲跌停鎖死)
安全檢查清單:
# 實盤下單前必須檢查
def pre_trading_checks(position, account):
"""實盤下單前安全檢查"""
print("=== 實盤下單安全檢查 ===\n")
# 1. 檢查持倉數量
if len(position) > 30:
raise ValueError(f"❌ 持倉過多({len(position)} 檔),建議 < 30 檔")
# 2. 檢查可用資金
available = account.get_balance()
required = position.total_value # 假設此方法存在
if required > available * 0.9:
raise ValueError(f"❌ 資金不足!需要 {required:,.0f},可用 {available:,.0f}")
# 3. 檢查帳戶連線
try:
account.get_position()
except Exception as e:
raise ConnectionError(f"❌ 券商帳戶連線失敗:{e}")
print("✅ 所有檢查通過\n")
# 執行檢查
try:
pre_trading_checks(position_info, account)
# 先預覽(不會真的下單)
executor.create_orders(view_only=True)
# 確認後執行(取消註解)
# executor.create_orders()
except (ValueError, ConnectionError) as e:
print(f"\n{e}")
print("請修正問題後再執行下單")
詳細錯誤處理:參考 實盤下單錯誤處理
實盤交易最後提醒
下單前務必確認:
- ✅ 策略已通過樣本外測試
- ✅ 先用模擬帳戶測試至少 1 週
- ✅ 初期資金不超過總資金的 10-20%
- ✅ 設定停損機制,避免單次巨大虧損
- ✅ 每日監控持倉與訂單狀態
錯誤可能導致財務損失!請務必謹慎!