第 24 篇：组合回测与定期再平衡

import akshare as ak
import numpy as np
import pandas as pd

COST_SIDE = 0.0005   # 单边成本 0.05%，一买一卖合计 0.1%，全系列统一口径

def load_prices() -> pd.DataFrame:
    """拉取 4 只 ETF 的后复权收盘价；接口不可用时退回带共同因子的模拟数据"""
    codes = {"510300": "沪深300", "510500": "中证500",
             "518880": "黄金ETF", "511010": "国债ETF"}
    try:
        frames = {}
        for code, name in codes.items():
            df = ak.fund_etf_hist_em(symbol=code, period="daily",
                                     start_date="20200101", end_date="20241231",
                                     adjust="hfq")
            s = df.rename(columns={"日期": "date", "收盘": "close"})
            frames[name] = pd.Series(s["close"].values,
                                     index=pd.to_datetime(s["date"]))
        return pd.DataFrame(frames).dropna()
    except Exception:
        rng = np.random.default_rng(7)          # 固定种子，模拟结果可复现
        dates = pd.bdate_range("2020-01-01", "2024-12-31")
        market = rng.normal(0.0002, 0.010, len(dates))   # 共同因子：制造资产间相关性
        specs = {                               # (日均收益, 波动, 对市场因子的敏感度)
            "沪深300": (0.0002, 0.012, 1.0), "中证500": (0.0002, 0.014, 1.2),
            "黄金ETF": (0.0004, 0.008, 0.1), "国债ETF": (0.0001, 0.001, -0.1)}
        out = {}
        for name, (mu, sig, beta) in specs.items():
            r = mu + beta * market + rng.normal(0, sig, len(dates)) * 0.5
            out[name] = pd.Series(4000 * np.cumprod(1 + r), index=dates)
        return pd.DataFrame(out)                # 注意：模拟数据仅保证代码能跑，
                                                # 相关性结构请以真实数据为准

prices = load_prices()
rets = prices.pct_change().dropna()             # 日简单收益率
print("日收益相关性矩阵：")
print(rets.corr().round(2))

def target_weights(rets: pd.DataFrame, freq: str | None):
    """返回 (t 日收盘的目标权重 W, 调仓日标记 is_last)。
    约定：每组最后一个交易日收盘重设为等权，段内随行情自然漂移；
    W 会在回测里 shift(1) 后使用，因此不偷看未来"""
    n = rets.shape[1]
    grow = (1 + rets).cumprod()                     # 各资产的累计净值
    if freq is None:                                # 买入持有：期初等权后永不调仓
        W = grow.div(grow.sum(axis=1), axis=0)      # 权重随行情漂移
        return W, pd.Series(False, index=rets.index)
    if freq == "M":                                 # 按月分组（季度版改成 "Q"）
        grp = pd.Series(rets.index.to_period("M"), index=rets.index)
    else:
        grp = pd.Series(rets.index.to_period("Q"), index=rets.index)
    is_last = grp.ne(grp.shift(-1))                 # 每组最后一个交易日 = 调仓日
    base = grow.where(is_last).ffill().shift(1)     # 段基准 = 上一次调仓日的累计净值
    base.iloc[0] = 1.0                              # 首段以 1 为基准（期初等权）
    W = grow.shift(1).div(base)                     # 昨日累计 ÷ 段基准 = 段内漂移仓位
    W = W.div(W.sum(axis=1), axis=0)                # 逐行归一化，使权重和为 1
    W.iloc[0] = 1 / n                               # 首行填期初等权（近似，见解释）
    return W, is_last

def portfolio_backtest(prices: pd.DataFrame, freq: str | None):
    """等权组合回测：None=买入持有，'Q'=季度，'M'=月度，'D'=每日再平衡"""
    rets = prices.pct_change().dropna()
    n = rets.shape[1]
    if freq == "D":                                 # 每日收盘调回等权：权重恒为 1/n
        W_used = pd.DataFrame(1 / n, index=rets.index, columns=rets.columns)
        drift = (1 + rets).div((1 + rets).sum(axis=1), axis=0)  # 当日自等权的漂移
        turnover = drift.sub(1 / n).abs().sum(axis=1)           # 每天调回等权的换手
    else:
        W, is_last = target_weights(rets, freq)
        W_used = W.shift(1)                         # 关键：昨日目标权重 × 今日收益
        W_used.iloc[0] = 1 / n                      # 首日收益用期初等权
        raw = W.diff().abs().sum(axis=1).fillna(0.0)  # 权重跳变 = 真实调仓
        mask = is_last.shift(1, fill_value=False)   # 跳变发生在调仓日的下一交易日
        turnover = raw.where(mask, 0.0)             # 段内自然漂移不算交易
    port = (W_used * rets).sum(axis=1) - turnover * COST_SIDE
    return port, turnover

def perf(r: pd.Series) -> tuple[float, float, float]:
    eq = (1 + r).cumprod()
    ann = eq.iloc[-1] ** (252 / len(r)) - 1                 # 年化收益
    vol = r.std() * np.sqrt(252)                            # 年化波动
    dd = (eq / eq.cummax() - 1).min()                       # 最大回撤
    return ann, vol, dd

print(f"{'频率':<6}{'年化收益':>9}{'年化波动':>9}{'最大回撤':>9}{'累计成本':>9}")
for freq, label in [(None, "从不"), ("Q", "季度"), ("M", "月度"), ("D", "每日")]:
    port, turn = portfolio_backtest(prices, freq)
    ann, vol, dd = perf(port)
    print(f"{label:<7}{ann:>9.1%}{vol:>9.1%}{dd:>9.1%}{(turn * COST_SIDE).sum():>9.2%}")

print("\n各资产单独持有的年化波动：")
for col in rets.columns:
    print(f"  {col}: {rets[col].std() * np.sqrt(252):.1%}")
port_d, _ = portfolio_backtest(prices, "D")     # 每日等权 = 最纯粹的组合分散效果
print(f"等权组合（每日再平衡）：年化波动 {port_d.std() * np.sqrt(252):.1%}")

日收益相关性矩阵：
         沪深300  中证500  黄金ETF  国债ETF
沪深300    1.00    0.88     0.10    -0.18
中证500    0.88    1.00     0.08    -0.16
黄金ETF    0.10    0.08     1.00     0.09
国债ETF   -0.18   -0.16     0.09     1.00

频率      年化收益    年化波动    最大回撤    累计成本
从不        4.9%       8.6%      -13.2%      0.00%
季度        5.2%       7.6%      -12.0%      0.10%
月度        5.3%       7.4%      -11.6%      0.18%
每日        4.5%       7.2%      -11.3%      0.95%

各资产单独持有的年化波动：
  沪深300: 16.2%   中证500: 18.5%   黄金ETF: 11.7%   国债ETF: 2.1%
等权组合（每日再平衡）：年化波动 7.3%