多策略組合

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 個策略完成!")

from finlab.portfolio import Portfolio

# 建立等權重組合(各佔 1/3)
port_equal = Portfolio(
    {
        '動能策略': (report1, 1 / 3),
        '價值策略': (report2, 1 / 3),
        '營收成長策略': (report3, 1 / 3),
    }
)

# 顯示組合績效
port_equal.display()

# 取得各策略夏普率(負值以 0 計,該策略權重為 0,不配置資金)
sharpe1 = max(report1.get_stats()['daily_sharpe'], 0)
sharpe2 = max(report2.get_stats()['daily_sharpe'], 0)
sharpe3 = max(report3.get_stats()['daily_sharpe'], 0)

# 計算權重(夏普率越高權重越大)
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()

# 取得組合績效
stats_equal = port_equal.get_stats()
stats_sharpe = port_sharpe.get_stats()

# 與單一策略比較
import pandas as pd

comparison = pd.DataFrame(
    {
        '動能策略': [
            report1.get_stats()['cagr'],
            report1.get_stats()['daily_sharpe'],
            report1.get_stats()['max_drawdown'],
        ],
        '價值策略': [
            report2.get_stats()['cagr'],
            report2.get_stats()['daily_sharpe'],
            report2.get_stats()['max_drawdown'],
        ],
        '營收成長策略': [
            report3.get_stats()['cagr'],
            report3.get_stats()['daily_sharpe'],
            report3.get_stats()['max_drawdown'],
        ],
        '等權組合': [
            stats_equal['cagr'],
            stats_equal['daily_sharpe'],
            stats_equal['max_drawdown'],
        ],
        '夏普加權組合': [
            stats_sharpe['cagr'],
            stats_sharpe['daily_sharpe'],
            stats_sharpe['max_drawdown'],
        ],
    },
    index=['年化報酬率', '夏普率', '最大回撤'],
)

print(comparison)

# Portfolio 本身就是回測報告
combined_report = port_sharpe

# 流動性分析
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')

# 取得組合的持倉
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}")

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()

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]

# 根據市場環境調整權重
adaptive_w = adaptive_weights([report1, report2, report3], market)

port_adaptive = Portfolio(
    {
        '動能策略': (report1, adaptive_w[0]),
        '價值策略': (report2, adaptive_w[1]),
        '營收成長策略': (report3, adaptive_w[2]),
    }
)

from finlab.portfolio import PortfolioSyncManager

# 第一次建立
pm = PortfolioSyncManager()

# 或從雲端/本地載入已有的部位
# pm = PortfolioSyncManager.from_cloud()
# pm = PortfolioSyncManager.from_local()

# 設定總資金
total_balance = 5000000  # 500 萬

# 更新持股(只在換股日才會換股)
pm.update(port_sharpe, total_balance=total_balance)

# 顯示當前持股
print(pm)

# 零股交易(可買賣零股)
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)

# 建立每日執行腳本
def daily_portfolio_sync():
    """
    每日執行:
    1. 檢查是否為換股日
    2. 檢查停損停利
    3. 更新持股
    """
    from finlab.exceptions import PortfolioError
    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()

    # 更新(只在換股日會實際換股)
    try:
        pm.update(port, total_balance=5000000)
    except PortfolioError as error:
        print(f"資金不足，本次全部停損停利與部位更新尚未執行：{error}")
        # 立即調高 total_balance 並重跑；成功後才儲存或同步。
        raise

    # 同步至雲端
    pm.to_cloud()

    # 如需下單,啟用以下程式碼:
    # account = SinopacAccount()
    # pm.sync(account)

    print("每日同步完成!")

# 使用 cron 或排程工具定期執行
# 例如:每天早上 8:30 執行

# 取得目標部位
target_position = pm.get_position()

# 顯示目標持倉
print(pm.get_total_position())

# 使用 pm.sync 模擬下單,檢查實盤與目標的差異
pm.sync(account, view_only=True)

# 每季重新評估
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()