完整流程
風險管理
本頁內容8
在量化交易中,不虧錢比賺錢更重要。良好的風險管理可以避免單次巨大虧損、降低整體投資組合波動、提升長期穩定報酬。這一頁帶你替一個策略依序加上四層控制:限制部位、停損停利、決定投入資金、控制回撤,最後用程式檢查風險。
做完這一頁,你會有:一份加了風控的回測報告、risk_alert() 與 generate_risk_report() 兩個檢查函式,以及回測、實盤與緊急狀況的檢查清單。
開始前,你需要會用 sim() 回測(見歷史回測),並知道怎麼用 hold_until 設定進出場(見寫選股條件)。
1. 限制每檔持股的大小
設定單一持股上限
position_limit 限制單一持股最多佔總資金的比例:
from finlab import data
from finlab.backtest import sim
close = data.get('price:收盤價')
position = close > close.average(20)
# 單一持股上限 10%
report = sim(
position,
resample='M',
position_limit=0.1, # 單一持股最多佔總資金 10%
)from finlab import data
from finlab.backtest import sim
data.set_market('us')
with data.us_universe(index='SP500'):
close = data.get('price:adj_close')
position = close > close.average(20)
# 單一持股上限 10%
report = sim(
position,
resample='M',
position_limit=0.1, # 單一持股最多佔總資金 10%
)這一頁後面的程式都沿用這裡的 close、position 與 report。
限制持股檔數
# 方法 1: 使用 is_largest 限制持股數
top_n_position = (close > close.average(20)).is_largest(30)
# 方法 2: 在回測時限制
report = sim(
position,
resample='M',
position_limit=0.1,
# 最多持有 30 檔 (每檔 10%,最多 30 檔 = 300% 槓桿)
)依波動度調整部位
波動度越高,持倉越小:
# 計算波動度
returns = close.pct_change()
volatility = returns.rolling(60).std()
# 波動度越高,持倉越小
inverse_vol_weight = 1 / volatility
normalized_weight = inverse_vol_weight / inverse_vol_weight.sum(axis=1)
# 與選股訊號結合
position_with_vol = (close > close.average(20)) * normalized_weight2. 設定停損停利
固定百分比
停損停利寫在 hold_until 裡:
# 站上 20 日均線進場,跌破 60 日均線出場
entry_signal = close > close.average(20)
exit_signal = close < close.average(60)
# 停損 10%,停利 20%
position = entry_signal.hold_until(exit=exit_signal, stop_loss=0.10, take_profit=0.20)
report = sim(position, resample='M')用 MAE/MFE 找停損停利點
先跑一次不設停損停利的回測,從每筆交易的最大不利幅度(MAE)與最大有利幅度(GMFE)決定停損停利。這裡停損設為 MAE 第 75 百分位數的 1.2 倍,停利設為 GMFE 第 75 百分位數的 0.8 倍:
# 先執行初步回測(不設停損停利)
position_without_stops = entry_signal.hold_until(exit=exit_signal)
report_initial = sim(position_without_stops, resample='M', upload=False)
# 查看 MAE/MFE
trades = report_initial.get_trades()
mae_q75 = abs(trades['mae'].quantile(0.75))
gmfe_q75 = trades['gmfe'].quantile(0.75)
print(f"建議停損: {mae_q75 * 1.2 * 100:.1f}%") # MAE Q75 * 1.2
print(f"建議停利: {gmfe_q75 * 0.8 * 100:.1f}%") # GMFE Q75 * 0.8
# 使用優化後的停損停利
position_optimized = entry_signal.hold_until(
exit=exit_signal, stop_loss=mae_q75 * 1.2, take_profit=gmfe_q75 * 0.8
)MAE/MFE 的圖表與解讀見策略分析模組。
移動停利
移動停利需透過 sim() 的 trail_stop 參數設定,不是寫在 hold_until 裡:
# 從最高報酬回落超過 8 個百分點時產生出場訊號
position = entry_signal.hold_until(exit=exit_signal, stop_loss=0.10)
report = sim(position, resample='M', trail_stop=0.08)時間停損
連續持有超過 30 個交易日就自動出場:
# 連續持有超過 30 個交易日自動出場
def add_time_stop(position, max_hold_days=30):
"""新增時間停損"""
time_exit = (position > 0).sustain(max_hold_days + 1)
return position & ~time_exit
position_with_time_stop = add_time_stop(position, max_hold_days=30)3. 決定每次投入多少資金
凱利公式
用回測的勝率、平均獲利率與平均虧損率,算出建議投入比例:
def kelly_criterion(win_rate, avg_win, avg_loss):
"""
計算凱利比例
Args:
win_rate: 勝率 (0-1)
avg_win: 平均獲利率
avg_loss: 平均虧損率 (正值)
Returns:
float: 建議投入比例
"""
if avg_loss == 0:
return 0
kelly = (win_rate * avg_win - (1 - win_rate) * avg_loss) / avg_win
return max(0, min(kelly, 1)) # 限制在 0-1 之間
# 使用範例
stats = report.get_stats()
trades = report.get_trades()
kelly_ratio = kelly_criterion(
win_rate=stats['win_ratio'],
avg_win=trades[trades['return'] > 0]['return'].mean(),
avg_loss=abs(trades[trades['return'] < 0]['return'].mean()),
)
print(f"凱利建議投入比例: {kelly_ratio * 100:.1f}%")
# 實務上使用 0.5 * Kelly (半凱利) 較保守
print(f"半凱利建議: {kelly_ratio * 0.5 * 100:.1f}%")固定分數法
每次只承擔總資金固定比例的風險,再依停損幅度推算部位大小:
# 每次只投入總資金的固定比例
RISK_PER_TRADE = 0.02 # 單次風險 2%
# 根據停損幅度計算部位大小
stop_loss = 0.10 # 停損 10%
position_size = RISK_PER_TRADE / stop_loss # = 0.2 = 20%
print(f"建議部位大小: {position_size * 100:.0f}%")4. 控制回撤
回撤超過上限就暫停交易
def check_drawdown_limit(report, max_drawdown=0.20):
"""檢查是否超過最大回撤限制"""
stats = report.get_stats()
current_dd = stats['max_drawdown']
if abs(current_dd) > max_drawdown:
print(f"警告:回撤 {current_dd * 100:.1f}% 超過限制 {max_drawdown * 100:.0f}%")
print("建議暫停交易,檢視策略!")
return False
return True
# 使用
if not check_drawdown_limit(report, max_drawdown=0.25):
# 暫停交易邏輯
pass回撤越大、部位越小
回撤到 -20% 時部位減半,調整係數限制在 0.5 到 1.0 之間:
def adjust_position_by_drawdown(position, creturn, base_size=1.0):
"""根據當前回撤動態調整部位
回撤越大,部位越小
"""
# 計算當前回撤
peak = creturn.expanding().max()
drawdown = (creturn - peak) / peak
# 調整係數 (回撤 -20% 時,部位減半)
adj_factor = 1 + drawdown / 0.20
adj_factor = adj_factor.clip(0.5, 1.0) # 限制在 0.5-1.0
# 調整部位
adjusted_position = position * adj_factor.reindex(position.index, method='ffill')
return adjusted_position
# 使用
position_adjusted = adjust_position_by_drawdown(position, report.creturn, base_size=1.0)5. 檢查風險並產生報告
風險預警
risk_alert() 檢查四件事,任一項不通過就列出警報,全部通過時回傳 True:
| 檢查 | 警報條件 |
|---|---|
| 夏普率 | 低於 0.5 |
| 最大回撤 | 超過 30% |
| 勝率 | 低於 40% |
| 平均獲利/平均虧損 | 低於 1.5 |
def risk_alert(report):
"""風險預警檢查"""
stats = report.get_stats()
alerts = []
# 檢查 1: 夏普率過低
if stats['daily_sharpe'] < 0.5:
alerts.append(f"⚠️ 夏普率過低: {stats['daily_sharpe']:.2f}")
# 檢查 2: 最大回撤過大
if abs(stats['max_drawdown']) > 0.30:
alerts.append(f"⚠️ 最大回撤過大: {stats['max_drawdown'] * 100:.1f}%")
# 檢查 3: 勝率過低
if stats['win_ratio'] < 0.40:
alerts.append(f"⚠️ 勝率過低: {stats['win_ratio'] * 100:.1f}%")
# 檢查 4: 平均獲利/虧損比過低
trades = report.get_trades()
avg_win = trades[trades['return'] > 0]['return'].mean()
avg_loss = abs(trades[trades['return'] < 0]['return'].mean())
win_loss_ratio = avg_win / avg_loss if avg_loss > 0 else 0
if win_loss_ratio < 1.5:
alerts.append(f"⚠️ 獲利/虧損比過低: {win_loss_ratio:.2f}")
# 輸出警報
if alerts:
print("=" * 50)
print("風險警報:")
for alert in alerts:
print(alert)
print("=" * 50)
else:
print("✅ 風險檢查通過")
return len(alerts) == 0
# 使用
risk_alert(report)定期風險報告
generate_risk_report() 印出年化報酬率、年化波動率、夏普率、最大回撤、勝率、平均持有天數與總交易次數:
def generate_risk_report(report):
"""產生風險報告"""
stats = report.get_stats()
trades = report.get_trades()
risk_metrics = {
'年化報酬率': f"{stats['cagr'] * 100:.2f}%",
'年化波動率': f"{stats['daily_vol'] * 100:.2f}%",
'夏普率': f"{stats['daily_sharpe']:.2f}",
'最大回撤': f"{stats['max_drawdown'] * 100:.2f}%",
'勝率': f"{stats['win_ratio'] * 100:.1f}%",
'平均持有天數': f"{trades['period'].mean():.1f} 天",
'總交易次數': len(trades),
}
print("\n" + "=" * 50)
print("風險報告")
print("=" * 50)
for key, value in risk_metrics.items():
print(f"{key:12s}: {value}")
print("=" * 50 + "\n")
# 使用
generate_risk_report(report)把風控組合起來
下面這段把停損、停利、單一持股上限與移動停利一起用上,再跑風險檢查與風險報告。risk_alert() 與 generate_risk_report() 要先執行上一節的定義:
from finlab import data
from finlab.backtest import sim
# 載入資料
close = data.get('price:收盤價')
# 策略邏輯
entry = close > close.average(20)
exit_signal = close < close.average(10)
# 加入風險控制
position = entry.hold_until(
exit=exit_signal,
stop_loss=0.10, # 停損 10%
take_profit=0.20, # 停利 20%
)
# 回測(單一持股上限 10%,移動停利 8%)
report = sim(position, resample='M', position_limit=0.1, trail_stop=0.08, upload=False)
# 風險檢查
if risk_alert(report):
print("策略通過風險檢查,可考慮實盤")
else:
print("策略未通過風險檢查,需要優化")
# 產生風險報告
generate_risk_report(report)from finlab import data
from finlab.backtest import sim
# 載入資料
data.set_market('us')
with data.us_universe(index='SP500'):
close = data.get('price:adj_close')
# 策略邏輯
entry = close > close.average(20)
exit_signal = close < close.average(10)
# 加入風險控制
position = entry.hold_until(
exit=exit_signal,
stop_loss=0.10, # 停損 10%
take_profit=0.20, # 停利 20%
)
# 回測(單一持股上限 10%,移動停利 8%)
report = sim(position, resample='M', position_limit=0.1, trail_stop=0.08, upload=False)
# 風險檢查
if risk_alert(report):
print("策略通過風險檢查,可考慮實盤")
else:
print("策略未通過風險檢查,需要優化")
# 產生風險報告
generate_risk_report(report)檢查清單
回測階段
- 單一持股權重 ≤ 10%
- 設定合理的停損(8-12%)與停利(15-25%)
- 最大回撤 < 30%
- 夏普率 > 1.0
- 勝率 > 45% 或 平均獲利/虧損比 > 1.5
實盤階段
- 使用模擬單測試至少 1 個月
- 初期資金 ≤ 總資金的 20%
- 每日監控回撤與持倉
- 每週檢視實盤 vs 回測的偏差
- 每月產生風險報告
緊急狀況
- 回撤 > 20%:暫停新增部位
- 回撤 > 30%:平倉 50%
- 回撤 > 40%:全部平倉,檢討策略
- 連續虧損 > 5 筆:暫停交易,檢視策略
下一步
這一頁有幫到你嗎?