多策略組合管理完整流程
本文介紹如何建立、管理與部署多策略投資組合,從策略池建立到實盤執行的完整流程。
為什麼需要多策略組合?
單一策略的風險:
- 市場環境變化: 策略在特定市場環境表現好,但環境改變時可能失效
- 回撤風險: 單一策略的最大回撤可能很大
- 收益不穩定: 策略可能長時間表現不佳
多策略組合的優勢:
- 分散風險: 不同策略在不同市場環境下互補
- 降低回撤: 組合的最大回撤通常小於單一策略
- 穩定收益: 平滑單一策略的波動
流程概覽
graph TB
A[建立策略池] --> B{有多個策略?}
B -->|是| C[使用 Portfolio 組合]
B -->|否| A
C --> D[回測組合績效]
D --> E{績效滿意?}
E -->|否| F[調整權重或策略]
F --> C
E -->|是| G[使用 PortfolioSyncManager]
G --> H[計算持股張數]
H --> I[實盤執行]
I --> J[定期調整]
J --> K{需要優化?}
K -->|是| F
K -->|否| I階段 1: 策略池建立
1.1 開發多個不同類型的策略
建議組合包含不同風格的策略:
from finlab import data
from finlab.backtest import sim
# 策略 1: 技術面動能策略(短期)
close = data.get('price:收盤價')
volume = data.get('price:成交股數')
entry1 = (close == close.rolling(20).max()) & (volume > volume.average(20) * 1.5)
exit1 = close < close.average(10)
position1 = entry1.hold_until(exit=exit1, stop_loss=0.08)
report1 = sim(position1, resample='W', name="動能策略", upload=False)
# 策略 2: 基本面價值策略(中期)
pb = data.get('price_earning_ratio:股價淨值比')
roe = data.get('fundamental_features:股東權益報酬率')
entry2 = (pb < pb.quantile(0.3, axis=1)) & (roe > 0.1)
exit2 = pb > pb.quantile(0.7, axis=1)
position2 = entry2.hold_until(exit=exit2)
report2 = sim(position2, resample='M', name="價值策略", upload=False)
# 策略 3: 營收成長策略(中長期)
rev = data.get('monthly_revenue:當月營收')
rev_ma3 = rev.average(3)
rev_ma12 = rev.average(12)
entry3 = (rev_ma3 / rev_ma12 > 1.15) & (close > close.average(60))
exit3 = rev_ma3 / rev_ma12 < 1.0
position3 = entry3.hold_until(exit=exit3, stop_loss=0.1)
report3 = sim(position3, resample='M', name="營收成長策略", upload=False)
print("策略池建立完成!")
print(f"策略 1 年化報酬: {report1.get_stats()['daily_mean']*100:.2f}%")
print(f"策略 2 年化報酬: {report2.get_stats()['daily_mean']*100:.2f}%")
print(f"策略 3 年化報酬: {report3.get_stats()['daily_mean']*100:.2f}%")
1.2 從雲端載入已開發的策略
from finlab.portfolio import create_report_from_cloud
# 從 FinLab 平台載入已上傳的策略
report1 = create_report_from_cloud('我的動能策略')
report2 = create_report_from_cloud('我的價值策略')
report3 = create_report_from_cloud('我的營收成長策略')
print("從雲端載入 3 個策略完成!")
階段 2: 使用 Portfolio 模組組合策略
2.1 基礎組合:等權重
from finlab.portfolio import Portfolio
# 建立等權重組合(各佔 1/3)
port_equal = Portfolio({
'動能策略': (report1, 1/3),
'價值策略': (report2, 1/3),
'營收成長策略': (report3, 1/3)
})
# 顯示組合績效
port_equal.display()
2.2 進階組合:根據夏普率分配權重
# 取得各策略夏普率
sharpe1 = report1.get_stats()['daily_sharpe']
sharpe2 = report2.get_stats()['daily_sharpe']
sharpe3 = report3.get_stats()['daily_sharpe']
# 計算權重(夏普率越高權重越大)
total_sharpe = sharpe1 + sharpe2 + sharpe3
w1 = sharpe1 / total_sharpe
w2 = sharpe2 / total_sharpe
w3 = sharpe3 / total_sharpe
print(f"動能策略權重: {w1*100:.1f}%")
print(f"價值策略權重: {w2*100:.1f}%")
print(f"營收成長策略權重: {w3*100:.1f}%")
# 建立夏普率加權組合
port_sharpe = Portfolio({
'動能策略': (report1, w1),
'價值策略': (report2, w2),
'營收成長策略': (report3, w3)
})
port_sharpe.display()
2.3 組合績效評估
# 取得組合績效
stats_equal = port_equal.report.get_stats()
stats_sharpe = port_sharpe.report.get_stats()
# 與單一策略比較
import pandas as pd
comparison = pd.DataFrame({
'動能策略': [
report1.get_stats()['daily_mean'],
report1.get_stats()['daily_sharpe'],
report1.get_stats()['max_drawdown']
],
'價值策略': [
report2.get_stats()['daily_mean'],
report2.get_stats()['daily_sharpe'],
report2.get_stats()['max_drawdown']
],
'營收成長策略': [
report3.get_stats()['daily_mean'],
report3.get_stats()['daily_sharpe'],
report3.get_stats()['max_drawdown']
],
'等權組合': [
stats_equal['daily_mean'],
stats_equal['daily_sharpe'],
stats_equal['max_drawdown']
],
'夏普加權組合': [
stats_sharpe['daily_mean'],
stats_sharpe['daily_sharpe'],
stats_sharpe['max_drawdown']
]
}, index=['年化報酬率', '夏普率', '最大回撤'])
print(comparison)
# 輸出範例:
# 動能策略 價值策略 營收成長策略 等權組合 夏普加權組合
# 年化報酬率 0.185 0.142 0.168 0.165 0.172
# 夏普率 1.23 0.98 1.15 1.35 1.42
# 最大回撤 -0.312 -0.285 -0.298 -0.245 -0.238
組合優勢
觀察到組合的夏普率(1.35, 1.42)高於所有單一策略,且最大回撤(-0.245, -0.238)也較小,證實了分散效果!
階段 3: 回測多策略組合
3.1 深度分析組合
# 取得組合的回測報告
combined_report = port_sharpe.report
# 流動性分析
combined_report.run_analysis('LiquidityAnalysis', required_volume=100000)
# MAE/MFE 分析
combined_report.display_mae_mfe_analysis()
# 時期穩定性
combined_report.run_analysis('PeriodStatsAnalysis')
# Alpha/Beta
combined_report.run_analysis('AlphaBetaAnalysis')
3.2 檢查組合持股分散度
# 取得組合的持倉
position_combined = combined_report.position
# 計算平均持股數
avg_holdings = (position_combined > 0).sum(axis=1).mean()
print(f"平均持股數: {avg_holdings:.1f}")
# 計算最大單一持股權重
max_weight = position_combined.max(axis=1).mean()
print(f"平均最大單一持股權重: {max_weight*100:.1f}%")
# 檢查三個策略的重疊度
pos1 = report1.position.iloc[-1]
pos2 = report2.position.iloc[-1]
pos3 = report3.position.iloc[-1]
overlap_12 = len(set(pos1[pos1>0].index) & set(pos2[pos2>0].index))
overlap_13 = len(set(pos1[pos1>0].index) & set(pos3[pos3>0].index))
overlap_23 = len(set(pos2[pos2>0].index) & set(pos3[pos3>0].index))
print(f"動能 & 價值 重疊股數: {overlap_12}")
print(f"動能 & 營收成長 重疊股數: {overlap_13}")
print(f"價值 & 營收成長 重疊股數: {overlap_23}")
階段 4: 動態調整權重
4.1 滾動視窗重新計算權重
def optimize_weights(reports, lookback_days=252):
"""
根據過去一年的表現,動態調整權重
Args:
reports: list of Report 物件
lookback_days: 回望天數(預設 252 交易日 ≈ 1 年)
Returns:
dict: 最優權重
"""
import numpy as np
from scipy.optimize import minimize
# 取得過去一年的日報酬率
returns = []
for report in reports:
daily_return = report.daily_creturn.pct_change().iloc[-lookback_days:]
returns.append(daily_return)
returns_df = pd.concat(returns, axis=1).dropna()
# 計算協方差矩陣
cov_matrix = returns_df.cov()
# 最小化變異數(風險平價)
def portfolio_variance(weights):
return np.dot(weights, np.dot(cov_matrix, weights))
# 約束:權重總和 = 1,每個權重 >= 0
constraints = {'type': 'eq', 'fun': lambda w: np.sum(w) - 1}
bounds = tuple((0, 1) for _ in range(len(reports)))
# 初始權重
init_weights = np.array([1/len(reports)] * len(reports))
# 優化
result = minimize(
portfolio_variance,
init_weights,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
return result.x
# 計算最優權重
optimal_weights = optimize_weights([report1, report2, report3])
print("最優權重:")
for i, w in enumerate(optimal_weights, 1):
print(f" 策略 {i}: {w*100:.1f}%")
# 建立最優權重組合
port_optimal = Portfolio({
'動能策略': (report1, optimal_weights[0]),
'價值策略': (report2, optimal_weights[1]),
'營收成長策略': (report3, optimal_weights[2])
})
port_optimal.display()
4.2 根據市場環境動態切換
def adaptive_weights(reports, market_condition='bull'):
"""
根據市場環境調整權重
Args:
reports: list of (name, Report) tuples
market_condition: 'bull', 'bear', 或 'sideways'
Returns:
dict: 調整後權重
"""
if market_condition == 'bull':
# 多頭:增加動能策略權重
return [0.5, 0.25, 0.25]
elif market_condition == 'bear':
# 空頭:增加價值策略權重
return [0.2, 0.5, 0.3]
else: # sideways
# 盤整:均衡配置
return [0.33, 0.33, 0.34]
# 判斷市場環境(簡化範例)
from finlab import data
taiex = data.get('benchmark_return:發行量加權股價報酬指數').squeeze()
taiex_ma50 = taiex.rolling(50).mean()
taiex_ma200 = taiex.rolling(200).mean()
if taiex.iloc[-1] > taiex_ma50.iloc[-1] > taiex_ma200.iloc[-1]:
market = 'bull'
print("市場環境:多頭")
elif taiex.iloc[-1] < taiex_ma50.iloc[-1] < taiex_ma200.iloc[-1]:
market = 'bear'
print("市場環境:空頭")
else:
market = 'sideways'
print("市場環境:盤整")
# 根據市場環境調整權重
adaptive_w = adaptive_weights([report1, report2, report3], market)
port_adaptive = Portfolio({
'動能策略': (report1, adaptive_w[0]),
'價值策略': (report2, adaptive_w[1]),
'營收成長策略': (report3, adaptive_w[2])
})
階段 5: 實盤執行
5.1 使用 PortfolioSyncManager
from finlab.portfolio import PortfolioSyncManager
# 第一次建立
pm = PortfolioSyncManager()
# 或從雲端/本地載入已有的部位
# pm = PortfolioSyncManager.from_cloud()
# pm = PortfolioSyncManager.from_local()
5.2 更新持股部位
# 設定總資金
total_balance = 5000000 # 500 萬
# 更新持股(只在換股日才會換股)
pm.update(port_sharpe, total_balance=total_balance)
# 顯示當前持股
print(pm)
# 輸出範例:
# Estimate value: 5,123,450
#
# quantity price weight close_price volume strategy type
# stock_id
# 2330 50.0 520.0 0.052 520.0 1250 策略1,策略3 STOCK
# 2317 100.0 85.0 0.017 85.0 850 策略2 STOCK
# 2454 80.0 120.0 0.019 120.0 960 策略1 STOCK
# ...
5.3 啟用零股或融資交易
# 零股交易(可買賣零股)
pm.update(port_sharpe, total_balance=total_balance, odd_lot=True)
# 融資交易(可使用融資融券)
pm.update(port_sharpe, total_balance=total_balance, margin_trading=True)
# 組合使用
pm.update(
port_sharpe,
total_balance=total_balance,
odd_lot=True,
margin_trading=True
)
5.4 實盤下單
from finlab.online.sinopac_account import SinopacAccount
from finlab.online.order_executor import OrderExecutor
# 設定券商帳戶
account = SinopacAccount(simulation=False)
# 建立下單執行器
executor = OrderExecutor(
positions=pm.generate_positions_amount(), # 從 PortfolioSyncManager 取得部位
account=account,
market_order=False, # 使用限價單
price_discount=0.01, # 限價單折價 1%
)
# 執行下單
executor.create_orders()
# 輸出:
# 2024-12-20 09:05:23 [INFO] 開始執行換股
# 2024-12-20 09:05:25 [INFO] 賣出 2303 100 股 @ 10.5
# 2024-12-20 09:05:27 [INFO] 買入 2330 50 股 @ 519.5
# ...
5.5 定期執行腳本
# 建立每日執行腳本
def daily_portfolio_sync():
"""
每日執行:
1. 檢查是否為換股日
2. 檢查停損停利
3. 更新持股
"""
from finlab.portfolio import PortfolioSyncManager, create_report_from_cloud
# 載入策略
report1 = create_report_from_cloud('動能策略')
report2 = create_report_from_cloud('價值策略')
report3 = create_report_from_cloud('營收成長策略')
# 建立組合
port = Portfolio({
'動能策略': (report1, 0.4),
'價值策略': (report2, 0.3),
'營收成長策略': (report3, 0.3)
})
# 載入現有部位
pm = PortfolioSyncManager.from_cloud()
# 更新(只在換股日會實際換股)
pm.update(port, total_balance=5000000)
# 同步至雲端
pm.sync_to_cloud()
# 如需下單,啟用以下程式碼:
# account = SinopacAccount(simulation=False)
# executor = OrderExecutor(
# positions=pm.generate_positions_amount(),
# account=account
# )
# executor.create_orders()
print("每日同步完成!")
# 使用 cron 或排程工具定期執行
# 例如:每天早上 8:30 執行
階段 6: 績效追蹤與調整
6.1 監控實盤與回測的差異
# 取得實盤部位
current_positions = pm.get_positions()
# 取得應有部位(回測建議)
target_positions = pm.generate_positions_amount()
# 比較差異
diff = {}
for stock_id in set(current_positions.keys()) | set(target_positions.keys()):
current = current_positions.get(stock_id, 0)
target = target_positions.get(stock_id, 0)
if abs(current - target) > 0.01: # 差異 > 1%
diff[stock_id] = {
'current': current,
'target': target,
'diff': current - target
}
if diff:
print("部位差異:")
for stock_id, info in diff.items():
print(f" {stock_id}: 實盤 {info['current']:.2%}, 目標 {info['target']:.2%}, 差異 {info['diff']:.2%}")
6.2 定期重新評估策略權重
# 每季重新評估
def quarterly_rebalance():
"""
每季執行:
1. 重新評估各策略表現
2. 調整權重
3. 移除表現差的策略
4. 新增表現好的策略
"""
# 取得最新 3 個月績效
sharpe1 = report1.get_stats()['daily_sharpe']
sharpe2 = report2.get_stats()['daily_sharpe']
sharpe3 = report3.get_stats()['daily_sharpe']
print(f"動能策略 3M 夏普率: {sharpe1:.2f}")
print(f"價值策略 3M 夏普率: {sharpe2:.2f}")
print(f"營收成長策略 3M 夏普率: {sharpe3:.2f}")
# 移除夏普率 < 0.5 的策略
active_strategies = {}
if sharpe1 >= 0.5:
active_strategies['動能策略'] = (report1, sharpe1)
if sharpe2 >= 0.5:
active_strategies['價值策略'] = (report2, sharpe2)
if sharpe3 >= 0.5:
active_strategies['營收成長策略'] = (report3, sharpe3)
# 重新分配權重
total_sharpe = sum(s for _, s in active_strategies.values())
new_port = Portfolio({
name: (report, sharpe/total_sharpe)
for name, (report, sharpe) in active_strategies.items()
})
print(f"\n保留 {len(active_strategies)} 個策略,重新調整權重")
return new_port
# 每季執行
new_port = quarterly_rebalance()
完整程式碼彙整
# =============================================================================
# 多策略組合管理完整範例
# =============================================================================
from finlab import data
from finlab.backtest import sim
from finlab.portfolio import Portfolio, PortfolioSyncManager
# 1. 建立策略池
close = data.get('price:收盤價')
volume = data.get('price:成交股數')
pb = data.get('price_earning_ratio:股價淨值比')
roe = data.get('fundamental_features:股東權益報酬率')
rev = data.get('monthly_revenue:當月營收')
# 策略 1: 動能
position1 = ((close == close.rolling(20).max()) &
(volume > volume.average(20) * 1.5)).hold_until(
exit=close < close.average(10),
stop_loss=0.08
)
report1 = sim(position1, resample='W', name="動能策略", upload=False)
# 策略 2: 價值
position2 = ((pb < pb.quantile(0.3, axis=1)) &
(roe > 0.1)).hold_until(
exit=pb > pb.quantile(0.7, axis=1)
)
report2 = sim(position2, resample='M', name="價值策略", upload=False)
# 策略 3: 營收成長
rev_momentum = rev.average(3) / rev.average(12)
position3 = ((rev_momentum > 1.15) &
(close > close.average(60))).hold_until(
exit=rev_momentum < 1.0,
stop_loss=0.1
)
report3 = sim(position3, resample='M', name="營收成長策略", upload=False)
# 2. 建立組合
port = Portfolio({
'動能策略': (report1, 0.4),
'價值策略': (report2, 0.3),
'營收成長策略': (report3, 0.3)
})
# 3. 評估組合績效
port.display()
print(port.report.get_stats())
# 4. 實盤執行
pm = PortfolioSyncManager()
pm.update(port, total_balance=5000000)
print(pm)
# 5. 同步至雲端
pm.sync_to_cloud()
print("多策略組合建立完成!")
關鍵要點總結
策略池建立階段
- ✅ 選擇風格不同的策略(技術面、基本面、籌碼面)
- ✅ 策略間相關性低,分散效果好
- ✅ 每個策略都要經過完整回測驗證
組合權重階段
- ✅ 等權重是最簡單的起點
- ✅ 根據夏普率加權可提升整體風險調整後報酬
- ✅ 定期(每季)重新評估權重
回測驗證階段
- ✅ 組合的夏普率應高於單一策略
- ✅ 組合的最大回撤應小於單一策略
- ✅ 檢查持股分散度與策略重疊度
實盤執行階段
- ✅ 使用 PortfolioSyncManager 自動管理部位
- ✅ 定期(每日)同步,確保部位一致
- ✅ 監控實盤與回測的偏差
持續優化階段
- ✅ 每季重新評估策略表現
- ✅ 移除夏普率 < 0.5 的策略
- ✅ 根據市場環境動態調整權重
常見問題
Q1: 組合多少個策略最好?
建議 3-5 個。太少無法有效分散,太多會稀釋收益且管理複雜。
Q2: 如何判斷策略是否應該移除?
觀察 3-6 個月的滾動夏普率,若持續 < 0.5 考慮移除。
Q3: 可以動態增減策略嗎?
可以!使用 quarterly_rebalance() 定期評估,新增表現好的策略,移除表現差的。
Q4: PortfolioSyncManager 如何處理衝突?
當多個策略同時持有同一檔股票時,會自動加總權重。例如策略 A 持有 2330 5%,策略 B 持有 2330 3%,最終組合持有 8%。
Q5: 如何避免過度交易?
- 使用較長的 resample 週期(M 或 Q)
- 設定最小換股幅度(例如權重變化 < 2% 不調整)
- 優先選擇換股頻率較低的策略