跳轉到

finlab.portfolio

多策略投資組合管理模組,用於將多個策略組合成一個投資組合,並支援雲端同步與自動調倉。

使用情境

  • 分散風險: 組合不同類型策略(動能 + 價值 + 成長),降低單一策略失效的風險
  • 提升穩定性: 單一策略波動大,多策略組合後可有效降低最大回撤
  • 動態調整權重: 根據市場狀況或策略表現,彈性調整各策略比例
  • 實盤同步: 使用 PortfolioSyncManager 自動更新持倉,無需手動管理

投資組合 vs 單策略回測

特性 單策略回測 (sim) 投資組合 (Portfolio)
適用場景 單一策略研究 多策略組合管理
輸入 position DataFrame 多個 Report 物件
權重分配 自動等權 可自訂各策略權重
風險控制 單一策略層級 組合層級再平衡
雲端同步 不支援 支援(PortfolioSyncManager)
複雜度 ⭐ 簡單 ⭐⭐⭐ 中等

快速範例

基礎用法:組合三個策略

from finlab import data
from finlab.backtest import sim
from finlab.portfolio import Portfolio

# 策略 1:動能策略(突破均線)
close = data.get('price:收盤價')
ma20 = close.average(20)
position1 = close > ma20
report1 = sim(position1, resample='M', upload=False)

# 策略 2:價值策略(低本益比)
pb = data.get('price_earning_ratio:股價淨值比')
position2 = pb.is_smallest(100)  # 選最小 100 檔
report2 = sim(position2, resample='M', upload=False)

# 策略 3:營收成長策略
rev_yoy = data.get('monthly_revenue:去年同月增減(%)')
position3 = rev_yoy > 20
report3 = sim(position3, resample='M', upload=False)

# 建立投資組合(權重:40%, 30%, 30%)
portfolio = Portfolio({
    '動能策略': (report1, 0.4),
    '價值策略': (report2, 0.3),
    '營收成長': (report3, 0.3)
})

# 回測組合績效
portfolio_report = portfolio.create_report()
portfolio_report.display()

# 查看組合統計
print(f"組合年化報酬: {portfolio_report.stats['annual_return']:.2%}")
print(f"組合夏普率: {portfolio_report.stats['daily_sharpe']:.2f}")
print(f"組合最大回撤: {portfolio_report.stats['max_drawdown']:.2%}")

詳細教學

參考 多策略組合管理完整流程,了解:

  • 策略權重優化方法(等權、風險平價、最小波動)
  • 動態調整權重策略(根據近期績效)
  • 實盤同步完整流程(PortfolioSyncManager)
  • 避免策略相關性過高的技巧

API Reference

Portfolio

finlab.portfolio.Portfolio

Portfolio(reports)

Bases: Report

建構 Portfolio 物件。

PARAMETER DESCRIPTION
- reports

代表投資組合的字典,key 為資產名稱,value 是回測報告與部位。

TYPE: Dict[str, Tuple[Report, float]]

Example

組合多個策略

from finlab import sim
from finlab.portfolio import Portfolio

# 請參閱 sim 函數的文件以獲取更多信息
# https://doc.finlab.tw/getting-start/

report_strategy1 = sim(...)
report_strategy2 = sim(...)
report_strategy3 = sim(...)

portfolio = Portfolio({
    'strategy1': (report_strategy1, 0.3),
    'strategy2': (report_strategy2, 0.4),
    'strategy3': (report_strategy3, 0.3),
})

權重配置建議

等權配置(適合策略特性相似時):

# 3 個策略各 1/3
portfolio = Portfolio({
    '策略A': (report1, 1/3),
    '策略B': (report2, 1/3),
    '策略C': (report3, 1/3)
})

風險平價(根據策略波動度反向配置):

# 計算各策略波動度
vol1 = report1.get_returns().std()
vol2 = report2.get_returns().std()
vol3 = report3.get_returns().std()

# 反向權重(波動小的配置多)
total_inv_vol = 1/vol1 + 1/vol2 + 1/vol3
w1, w2, w3 = (1/vol1)/total_inv_vol, (1/vol2)/total_inv_vol, (1/vol3)/total_inv_vol

portfolio = Portfolio({
    '策略A': (report1, w1),
    '策略B': (report2, w2),
    '策略C': (report3, w3)
})

目標波動(設定組合目標波動率):

# 先建立等權組合
portfolio = Portfolio({
    '策略A': (report1, 0.5),
    '策略B': (report2, 0.5)
})

# 計算組合波動
port_report = portfolio.create_report()
port_vol = port_report.get_returns().std()

# 調整權重以達成目標波動(如 15%)
target_vol = 0.15
leverage = target_vol / port_vol  # 槓桿倍數

# 重新配置(需確保總權重 <= 1)
if leverage <= 1:
    portfolio = Portfolio({
        '策略A': (report1, 0.5 * leverage),
        '策略B': (report2, 0.5 * leverage)
    })

常見錯誤

1. 權重總和不為 1

# 錯誤:權重總和 = 0.9
portfolio = Portfolio({
    '策略A': (report1, 0.4),
    '策略B': (report2, 0.5)  # ❌ 總和 0.9
})

# 正確:確保權重總和 = 1
portfolio = Portfolio({
    '策略A': (report1, 0.4),
    '策略B': (report2, 0.6)  # ✅ 總和 1.0
})

2. 策略未上傳至雲端(PortfolioSyncManager 需要)

# 錯誤:report 未上傳
report1 = sim(position1, resample='M', upload=False)  # ❌

# 正確:上傳至雲端
report1 = sim(position1, resample='M', upload=True)  # ✅

3. 策略調倉時間不一致

# 錯誤:不同調倉頻率
report1 = sim(position1, resample='M')   # 月線
report2 = sim(position2, resample='W')   # 週線 ❌

# 正確:統一調倉頻率
report1 = sim(position1, resample='M')   # 月線
report2 = sim(position2, resample='M')   # 月線 ✅

create_multi_asset_report()

finlab.portfolio.create_multi_asset_report

create_multi_asset_report(stock_list, **kwargs)

根據提供的股票清單創建多資產報告。 Create a multi-asset report based on the stock list provided.

PARAMETER DESCRIPTION
stock_list

一個以股票代號為 key,權重大小為 value。 A dictionary with stock id as key and weight as value.

TYPE: dict

RETURNS DESCRIPTION
Report

一個包含回測結果的報告對象。A report object with the backtest result.

Example:

>>> from finlab.portfolio import create_multi_asset_report
...
...
>>> report = create_multi_asset_report({'2330': 0.5, '1101': 0.5})

create_report_from_cloud()

finlab.portfolio.create_report_from_cloud

create_report_from_cloud(strategy_id, user_id=None, market=None)

根據提供的用戶ID和策略ID創建在線報告。 Create an online report based on the user id and strategy id provided.

PARAMETER DESCRIPTION
user_id

The user id.

TYPE: str DEFAULT: None

strategy_id

The

TYPE: str

PortfolioSyncManager

finlab.portfolio.PortfolioSyncManager

PortfolioSyncManager(data=None, start_with_empty=False)

投資組合類別,用於設定和獲取投資組合資訊。

ATTRIBUTE DESCRIPTION
path

投資組合資訊的儲存路徑。

TYPE: str

METHOD DESCRIPTION
set

Dict[str, Tuple[int, Report]]): 設定投資組合的函數。

get_position

獲取持倉資訊。

Examples:

from finlab.portfolio import Portfolio, PortfolioSyncManager

# 初始化投資組合
port = Portfolio({'策略A': (report1, 0.3), '策略B': (report2, 0.7)})

# 設定投資組合
# pm = PortfolioSyncManager.from_local() or 
# pm = PortfolioSyncManager.from_cloud()
pm = PortfolioSyncManager() 
pm.update(port, total_balance=1000000)
pm.to_cloud() # pm.to_local()

print(pm)

# 下單
account = ... # 請參考 Account 產生方式
pm.sync(account) # 平盤價格下單

建構投資組合。

clear

clear()

清除持倉資訊。

RETURNS DESCRIPTION

None

create_order_executor

create_order_executor(account, at='close', consider_margin_as_asset=False, market_name=None, **kwargs)

同步持倉資訊。

PARAMETER DESCRIPTION
account

交易帳戶。

TYPE: Account

consider_margin_as_asset

是否將融資融券視為資產。預設為 True。

TYPE: bool DEFAULT: False

market_name

指定市場名稱。預設為 None,也就是獲取所有市場。

TYPE: str DEFAULT: None

market_order

以類市價盡量即刻成交:所有買單掛漲停價,所有賣單掛跌停價

TYPE: bool

best_price_limit

掛芭樂價:所有買單掛跌停價,所有賣單掛漲停價

TYPE: bool

view_only

預設為 False,會實際下單。若設為 True,不會下單,只會回傳欲執行的委託單資料(dict)

TYPE: bool

extra_bid_pct

以該百分比值乘以價格進行追價下單,如設定為 0.05 時,將以當前價的 +(-)5% 的限價進買入(賣出),也就是更有機會可以成交,但是成交價格可能不理想; 假如設定為 -0.05 時,將以當前價的 -(+)5% 進行買入賣出,也就是限價單將不會立即成交,然而假如成交後,價格比較理想。參數有效範圍為 -0.1 到 0.1 內。

TYPE: float

RETURNS DESCRIPTION

None

from_cloud classmethod

from_cloud(name='default')

從雲端檔案初始化投資組合。

PARAMETER DESCRIPTION
path

雲端檔案的路徑。預設為 'default'。

TYPE: str

RETURNS DESCRIPTION
PortfolioSyncManager

投資組合類別。

from_local classmethod

from_local(name='default')

從本地檔案初始化投資組合。

PARAMETER DESCRIPTION
path

本地檔案的路徑。

TYPE: str

RETURNS DESCRIPTION
PortfolioSyncManager

投資組合類別。

from_path classmethod

from_path(path)

從本地檔案初始化投資組合。

PARAMETER DESCRIPTION
path

本地檔案的路徑。

TYPE: str

RETURNS DESCRIPTION
PortfolioSyncManager

投資組合類別。

get_position

get_position(at='close', market_name=None)

獲取持倉資訊。

PARAMETER DESCRIPTION
market_name

指定市場名稱。預設為 None,也就是獲取所有市場。

TYPE: str DEFAULT: None

RETURNS DESCRIPTION

dict or Position: 若 combined 為 True,則返回合併後的持倉資訊(Position 物件); 若 combined 為 False,則返回原始持倉資訊(dict)。

get_strategy_position

get_strategy_position(strategy_name, at)

獲取策略的開倉部位。

PARAMETER DESCRIPTION
strategy_name

策略名稱。

TYPE: str

RETURNS DESCRIPTION
dict

開倉部位資訊。

get_total_position

get_total_position()

回傳目前持倉的 DataFrame(與 repr 顯示的 df 相同)。

RETURNS DESCRIPTION

pd.DataFrame: 持倉資訊的 DataFrame。

margin_cash_position_combine staticmethod

margin_cash_position_combine(acp, idp)

合併現金與保證金部位。

PARAMETER DESCRIPTION
acp

當前帳戶的股票部位。

TYPE: Position

idp

新的帳戶的股票部位。

TYPE: Position

sync

sync(account, at='close', consider_margin_as_asset=True, market_name=None, **kwargs)

同步持倉資訊。

PARAMETER DESCRIPTION
account

交易帳戶。

TYPE: Account

consider_margin_as_asset

是否將保證金交易視為資產。預設為 True。

TYPE: bool DEFAULT: True

market_name

指定市場名稱。預設為 None,也就是獲取所有市場。

TYPE: str DEFAULT: None

market_order

以類市價盡量即刻成交:所有買單掛漲停價,所有賣單掛跌停價

TYPE: bool

best_price_limit

掛芭樂價:所有買單掛跌停價,所有賣單掛漲停價

TYPE: bool

view_only

預設為 False,會實際下單。若設為 True,不會下單,只會回傳欲執行的委託單資料(dict)

TYPE: bool

extra_bid_pct

以該百分比值乘以價格進行追價下單,如設定為 0.05 時,將以當前價的 +(-)5% 的限價進買入(賣出),也就是更有機會可以成交,但是成交價格可能不理想; 假如設定為 -0.05 時,將以當前價的 -(+)5% 進行買入賣出,也就是限價單將不會立即成交,然而假如成交後,價格比較理想。參數有效範圍為 -0.1 到 0.1 內。

TYPE: float

RETURNS DESCRIPTION

None

to_cloud

to_cloud(name='default')

將投資組合資訊存至雲端檔案。

RETURNS DESCRIPTION

None

to_local

to_local(name='default')

將投資組合資訊存至本地檔案。

RETURNS DESCRIPTION

None

to_path

to_path(path)

將投資組合資訊存至本地檔案。

RETURNS DESCRIPTION

None

update

update(portfolio, total_balance=0, rebalance_safety_weight=0.2, smooth_transition=None, force_override_difference=False, custom_position=None, excluded_stock_ids=None, **kwargs)

設定投資組合的函數。

PARAMETER DESCRIPTION
portfolio

包含投資組合資訊的字典。

TYPE: Portfolio

total_balance

總資產。

TYPE: float DEFAULT: 0

rebalance_safety_weight

現金的權重,確保以市價買賣時,新的策略組合價值不超過舊的價值,計算方式為:賣出股票後,有多少比例要變成現金(例如 20%),再買入新的股票。

TYPE: float DEFAULT: 0.2

smooth_transition

是否只在換股日才更新,預設為 None,系統會自行判斷,假如第一次呼叫函示,會是 False,之後會是 True。

TYPE: bool DEFAULT: None

force_override_difference

是否強制覆蓋不同的部位,預設為 False。

TYPE: bool DEFAULT: False

custom_position

自定義部位,預設為 None。當 custom_position 不為 None 時,會將 custom_position 加入到部位中。程式在計算部位時,會將 custom_position 排除在外,不列入。

TYPE: Position or dict DEFAULT: None

excluded_stock_ids

排除的股票代碼列表。預設為 None。

TYPE: List[str] DEFAULT: None

RETURNS DESCRIPTION

None

Examples:

from finlab.backtest import sim
from finlab.portfolio import Portfolio, OnlineReport

# create report 1
report1 = sim(...) # 請參考回測語法

# download online report
report2 = OnlineReport.from_cloud('strategyName') # 下載策略報告

# create portfolio
portfolio = Portfolio({'策略1': (report1, 0.5), '策略2': (report2, 0.5)})

# create PortfolioSyncManager
pm = PortfolioSyncManager.from_cloud()
pm.update(portfolio, total_balance=1000000, cash_weight=0.2) # 投資 100 萬元

# create orders
account = ... # 請參考 Account 產生方式
pm.sync(account) # 平盤價格下單
pm.sync(account, market_order=True) # 市價下單

實盤同步最佳實踐

基本設定:

from finlab.portfolio import PortfolioSyncManager

# 初始化(需要先上傳策略至雲端)
manager = PortfolioSyncManager(
    report_ids=['report_id_1', 'report_id_2'],  # 雲端策略 ID
    weights=[0.5, 0.5],                         # 權重
    total_balance=1000000                        # 總資金 100 萬
)

# 取得當前應持有的股票與張數
current_position = manager.get_position()
print(current_position)

定時更新(建議每日收盤後執行):

import schedule
import time

def sync_portfolio():
    manager = PortfolioSyncManager(
        report_ids=['xxx', 'yyy'],
        weights=[0.6, 0.4],
        total_balance=1000000
    )
    position = manager.get_position()
    # 執行下單邏輯(參考 finlab.online)
    print(f"更新時間: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(position)

# 每日收盤後執行(假設 15:00)
schedule.every().day.at("15:00").do(sync_portfolio)

while True:
    schedule.run_pending()
    time.sleep(60)

雲端同步注意事項

  1. 確保策略已上傳: 使用 report.upload()sim(..., upload=True)
  2. 檢查持倉衝突: 多策略可能同時持有/放空同一股票
    # Portfolio 會自動處理衝突:
    # - 策略 A: 持有 2330 (20% 權重,策略配置 40%)
    # - 策略 B: 持有 2330 (15% 權重,策略配置 60%)
    # → 組合後:2330 總權重 = 0.4 * 20% + 0.6 * 15% = 17%
    
  3. 定期檢查同步狀態: 確保雲端資料與本地一致

常見問題

Q: 如何處理策略之間的持倉衝突?

Portfolio 會自動處理持倉衝突:

# 假設兩個策略同時持有 2330
# 策略 A (權重 40%): 2330 佔 20%
# 策略 B (權重 60%): 2330 佔 15%

portfolio = Portfolio({
    '策略A': (report1, 0.4),
    '策略B': (report2, 0.6)
})

# 組合後 2330 的總權重 = 0.4 * 0.20 + 0.6 * 0.15 = 0.17 (17%)

Q: 如何設定最大持股數限制?

在各策略回測時使用 is_largest() 限制:

# 策略 1: 最多 30 檔
position1 = (close > ma20).is_largest(30)
report1 = sim(position1, resample='M')

# 策略 2: 最多 20 檔
position2 = (pb < 1.5).is_largest(20)
report2 = sim(position2, resample='M')

# 組合後最多約 50 檔(可能有重複持股)
portfolio = Portfolio({
    '策略1': (report1, 0.5),
    '策略2': (report2, 0.5)
})

Q: 如何動態調整策略權重?

# 方法 1: 根據近期績效調整
def dynamic_weights(reports, window=60):
    """根據近 60 日表現調整權重"""
    recent_returns = []
    for report in reports:
        ret = report.get_returns().tail(window).mean()  # 近期平均報酬
        recent_returns.append(max(ret, 0))  # 負報酬設為 0

    # 歸一化
    total = sum(recent_returns)
    if total == 0:
        return [1/len(reports)] * len(reports)  # 等權

    return [r / total for r in recent_returns]

# 使用動態權重
weights = dynamic_weights([report1, report2, report3], window=90)
portfolio = Portfolio({
    '策略1': (report1, weights[0]),
    '策略2': (report2, weights[1]),
    '策略3': (report3, weights[2])
})

Q: 組合後績效反而變差?

可能原因與檢查方法:

  1. 策略相關性過高:

    # 檢查策略相關性
    returns1 = report1.get_returns()
    returns2 = report2.get_returns()
    correlation = returns1.corr(returns2)
    print(f"相關性: {correlation:.2f}")  # > 0.8 表示高度相關
    
    # 解決:選擇相關性低的策略(< 0.5)
    

  2. 權重配置不當:

    # 檢查各策略夏普率
    print(f"策略 1 夏普: {report1.stats['daily_sharpe']:.2f}")
    print(f"策略 2 夏普: {report2.stats['daily_sharpe']:.2f}")
    
    # 給夏普率高的策略更多權重
    

  3. 回測期間不一致:

    # 確保所有策略使用相同的回測期間
    import finlab
    finlab.truncate_start = '2018-01-01'
    
    report1 = sim(position1, resample='M')
    report2 = sim(position2, resample='M')
    # 確保兩個 report 的起始日期一致
    

參考資源