策略组合（Risk Parity / HRP / CVXPY）

日期: 2026-08-10 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #组合优化 #RiskParity #HRP #CVXPY

---

今日目标

---

一、理论与模型

1.1 Mean-Variance Optimization (Markowitz)

经典 Markowitz： $$ \max_w \quad \mu^T w - \frac{\lambda}{2} w^T \Sigma w $$

s.t. $\sum w_i = 1$, $w_i \ge 0$（long-only）

或最小化方差给定 expected return： $$ \min_w w^T \Sigma w \quad \text{s.t.} \quad \mu^T w \ge r_{target}, \sum w_i = 1 $$

问题：

1. 估计误差敏感（μ 错 1% → 仓位错 50%）
2. 极端权重（concentration）
3. 不稳定（小输入变化大输出变化）

1.2 Risk Parity

每个资产/策略对组合风险贡献相等：

$$ RC_i = w_i \cdot (\Sigma w)_i = \frac{1}{n} \sum_j w_j (\Sigma w)_j $$

数值求解（无 closed-form）： $$ \min_w \sum_i \left( \frac{w_i (\Sigma w)_i}{w^T \Sigma w} - \frac{1}{n} \right)^2 $$

优势：

• 不需要估计 μ（μ 估计误差大）
• 风险均衡分散
• Bridgewater 全天候组合的核心

缺点：

• 高 vol 资产权重低，可能 underweight 高 yield 机会
• 杠杆通常需要补足收益

1.3 Hierarchical Risk Parity (HRP)

López de Prado (2016) 提出，避免 inverse covariance 估计误差。

3 步：

Step 1：Tree Clustering 对 correlation matrix 做 hierarchical clustering，得到 dendrogram。

距离矩阵： $$ d_{ij} = \sqrt{\frac{1 - \rho_{ij}}{2}} $$

Step 2：Quasi-Diagonalization 按 dendrogram 顺序重排 covariance matrix，相似资产相邻。

Step 3：Recursive Bisection 递归对半分配权重：

• 把资产分两半 $V_1, V_2$
• 计算两半的 inverse-variance weighted 方差 $\tilde V_1, \tilde V_2$
• 分配权重： $$ \alpha_1 = 1 - \frac{\tilde V_1}{\tilde V_1 + \tilde V_2}, \quad \alpha_2 = 1 - \alpha_1 $$
• 递归到单资产

HRP 优势：

• 不需 invert covariance（avoid singularity）
• 利用 cluster 结构
• 比 RP 更稳健，特别是高维（资产数 > 50）

1.4 Black-Litterman

把 PM views 与 prior 结合，避免纯 mean-variance 的极端解。

Prior: equilibrium return $\Pi = \lambda \Sigma w_{mkt}$ Views: $P \mu = q + \epsilon, \epsilon \sim N(0, \Omega)$ Posterior: $$ \mu_{BL} = \left[(\tau \Sigma)^{-1} + P^T \Omega^{-1} P\right]^{-1} \left[(\tau \Sigma)^{-1} \Pi + P^T \Omega^{-1} q\right] $$

实战：把"BTC 比 SOL 更好 +5%" 这类 view 注入 prior。

1.5 CVXPY 求解

CVXPY 是 Python 凸优化库，处理 markowitz、RP 等：

• Markowitz 是 QP（quadratic programming），CVXPY 直接求解
• RP 是非凸，需特殊处理
• 约束：sum_to_one、long-only、box constraints、turnover、tracking error

1.6 约束设计

实战常见约束：

• Long-only: $w_i \ge 0$
• Box: $0 \le w_i \le 0.2$（单仓上限）
• Sector: $\sum_{i \in sector} w_i \le 0.4$
• Turnover: $\sum |w_i^{new} - w_i^{old}| \le \tau$（控成本）
• Volatility target: $\sqrt{w^T \Sigma w} \le \sigma_{target}$
• Beta neutrality: $\beta^T w = 0$
• Cardinality: 持仓数量 ≤ K（但是非凸，需 MILP）

---

二、直觉与陷阱

陷阱 1：Markowitz 不稳定

μ 估计 SE = σ/√n，n=252 时 σ=20% → SE = 1.3%。 真实 μ 可能 5%，估计 5±1.3%，仓位差 25%+。 解法：

• Shrinkage（James-Stein, Ledoit-Wolf）
• Robust optimization（不确定性集合）
• 直接用 RP（不需 μ）

陷阱 2：协方差矩阵奇异

资产数 N > 时间 T 时 sample cov singular，不可 invert。 解法：

• Shrinkage to identity / market factor
• PCA 降维
• HRP（无需 invert）

陷阱 3：换手成本被低估

Markowitz 优化每月可能 100% 换手。 解法：

• 加 turnover penalty: $\min w^T \Sigma w + \tau |w - w_0|_1$
• Rebalance threshold（变化 < 阈值不动）

陷阱 4：尾部风险忽略

正态假设下 var 是好 risk measure，但加密 fat tail，应用 CVaR。 解法：CVaR optimization（仍是 LP，CVXPY 可解）

陷阱 5：In-sample 优化过拟合

历史数据上 optimize 经常 OOS 失败。 解法：

• Walk-forward 重新优化
• Robust optimization
• Bayesian shrinkage

陷阱 6：相关性不稳定

牛市 BTC/ETH = 0.7，熊市 = 0.95。组合分散化在最需要时失效。 解法：用 stress-test 协方差（max correlation）；或 regime-switching 协方差。

---

三、代码实现

3.1 完整组合优化框架

# portfolio.py
"""
Portfolio Optimization Framework
- Mean-Variance (Markowitz)
- Risk Parity
- Hierarchical Risk Parity
- CVXPY-based with constraints
"""

import numpy as np
import pandas as pd
import cvxpy as cp
from scipy.cluster.hierarchy import linkage, dendrogram, leaves_list
from scipy.spatial.distance import squareform
from typing import Dict, Optional, List


# ----------------------------- Markowitz -----------------------------
def mean_variance_optimize(mu: np.ndarray, cov: np.ndarray,
                            risk_aversion: float = 1.0,
                            max_weight: float = 0.4,
                            long_only: bool = True) -> np.ndarray:
    """
    经典 Markowitz：max μ'w - λ/2 w'Σw
    """
    n = len(mu)
    w = cp.Variable(n)

    constraints = [cp.sum(w) == 1]
    if long_only:
        constraints.append(w >= 0)
    constraints.append(w <= max_weight)
    if not long_only:
        constraints.append(w >= -max_weight)

    objective = cp.Maximize(mu @ w - 0.5 * risk_aversion * cp.quad_form(w, cov))
    prob = cp.Problem(objective, constraints)
    prob.solve()

    if prob.status not in ('optimal', 'optimal_inaccurate'):
        raise RuntimeError(f"Solver status: {prob.status}")
    return w.value


def min_variance_optimize(cov: np.ndarray, target_return: Optional[float] = None,
                            mu: Optional[np.ndarray] = None,
                            max_weight: float = 0.4) -> np.ndarray:
    """最小方差或给定 return 下最小方差"""
    n = cov.shape[0]
    w = cp.Variable(n)
    constraints = [cp.sum(w) == 1, w >= 0, w <= max_weight]
    if target_return is not None and mu is not None:
        constraints.append(mu @ w >= target_return)
    objective = cp.Minimize(cp.quad_form(w, cov))
    prob = cp.Problem(objective, constraints)
    prob.solve()
    return w.value


# ----------------------------- Risk Parity -----------------------------
def risk_parity_optimize(cov: np.ndarray, target_vol: Optional[float] = None,
                          tol: float = 1e-8, max_iter: int = 500) -> np.ndarray:
    """
    Risk Parity via Newton iteration
    """
    n = cov.shape[0]
    w = np.ones(n) / n
    target_rc = 1.0 / n

    for iter in range(max_iter):
        port_var = w @ cov @ w
        if port_var <= 0:
            break
        marginal_rc = cov @ w
        rc = w * marginal_rc / port_var

        # Gradient
        grad = (rc - target_rc) * marginal_rc
        # Newton step (simplified)
        step = 0.01
        w_new = w - step * grad
        w_new = np.maximum(w_new, 0)
        w_new /= w_new.sum()

        if np.max(np.abs(w_new - w)) < tol:
            break
        w = w_new

    if target_vol is not None:
        # Scale leverage
        port_vol = np.sqrt(w @ cov @ w)
        w = w * (target_vol / port_vol)
    return w


def risk_parity_cvxpy(cov: np.ndarray) -> np.ndarray:
    """
    Risk Parity via convex relaxation
    log barrier formulation: min 0.5 w'Σw - 1/n Σ log(w_i)
    """
    n = cov.shape[0]
    y = cp.Variable(n, nonneg=True)
    obj = cp.Minimize(0.5 * cp.quad_form(y, cov) -
                       (1.0 / n) * cp.sum(cp.log(y)))
    prob = cp.Problem(obj)
    prob.solve()
    if y.value is None:
        raise RuntimeError("Solver failed")
    w = y.value / y.value.sum()
    return w


# ----------------------------- HRP -----------------------------
def hrp_optimize(returns_df: pd.DataFrame) -> pd.Series:
    """Hierarchical Risk Parity (Lopez de Prado)"""
    cov = returns_df.cov().values
    corr = returns_df.corr().values

    # Step 1: distance + linkage
    dist = np.sqrt((1 - corr) / 2)
    np.fill_diagonal(dist, 0)
    condensed = squareform(dist, checks=False)
    link = linkage(condensed, method='single')

    # Step 2: quasi-diagonalize
    sort_ix = leaves_list(link)
    sorted_assets = returns_df.columns[sort_ix].tolist()

    # Step 3: recursive bisection
    weights = pd.Series(1.0, index=sorted_assets)
    cluster_items = [sorted_assets]

    while len(cluster_items) > 0:
        # 拆分每个 cluster
        new_items = []
        for items in cluster_items:
            if len(items) <= 1:
                continue
            mid = len(items) // 2
            left = items[:mid]
            right = items[mid:]

            # Compute inverse-variance weighted variance per cluster
            def cluster_var(asset_list):
                ix = [returns_df.columns.get_loc(a) for a in asset_list]
                sub_cov = cov[np.ix_(ix, ix)]
                ivp = 1 / np.diag(sub_cov)
                ivp /= ivp.sum()
                return ivp @ sub_cov @ ivp

            v_left = cluster_var(left)
            v_right = cluster_var(right)
            alpha = 1 - v_left / (v_left + v_right)

            for asset in left:
                weights[asset] *= alpha
            for asset in right:
                weights[asset] *= (1 - alpha)

            new_items.append(left)
            new_items.append(right)
        cluster_items = new_items

    return weights / weights.sum()


# ----------------------------- 约束扩展 -----------------------------
def constrained_optimize(mu: np.ndarray, cov: np.ndarray,
                          risk_aversion: float = 1.0,
                          max_weight: float = 0.3,
                          sector_limits: Optional[Dict] = None,
                          asset_to_sector: Optional[Dict] = None,
                          turnover_limit: Optional[float] = None,
                          w_prev: Optional[np.ndarray] = None,
                          beta_target: Optional[float] = None,
                          beta_vec: Optional[np.ndarray] = None) -> np.ndarray:
    """完整约束下 Markowitz"""
    n = len(mu)
    w = cp.Variable(n)

    constraints = [cp.sum(w) == 1, w >= 0, w <= max_weight]

    # Sector limits
    if sector_limits and asset_to_sector:
        for sector, limit in sector_limits.items():
            indices = [i for i in range(n) if asset_to_sector.get(i) == sector]
            if indices:
                constraints.append(cp.sum(w[indices]) <= limit)

    # Turnover
    if turnover_limit is not None and w_prev is not None:
        constraints.append(cp.norm(w - w_prev, 1) <= turnover_limit)

    # Beta
    if beta_target is not None and beta_vec is not None:
        constraints.append(beta_vec @ w == beta_target)

    objective = cp.Maximize(mu @ w - 0.5 * risk_aversion * cp.quad_form(w, cov))
    prob = cp.Problem(objective, constraints)
    prob.solve()
    return w.value


# ----------------------------- CVaR Optimization -----------------------------
def cvar_optimize(returns: np.ndarray, alpha: float = 0.05,
                   target_return: Optional[float] = None,
                   max_weight: float = 0.3) -> np.ndarray:
    """
    最小化 CVaR (Expected Shortfall)
    Rockafellar-Uryasev formulation (LP)
    """
    T, n = returns.shape
    w = cp.Variable(n)
    z = cp.Variable(T, nonneg=True)
    var = cp.Variable()

    portfolio_loss = -returns @ w
    constraints = [
        cp.sum(w) == 1,
        w >= 0,
        w <= max_weight,
        z >= portfolio_loss - var,
    ]
    if target_return is not None:
        constraints.append(returns.mean(axis=0) @ w >= target_return)

    cvar = var + (1.0 / (alpha * T)) * cp.sum(z)
    prob = cp.Problem(cp.Minimize(cvar), constraints)
    prob.solve()
    return w.value


# ----------------------------- 完整 Demo -----------------------------
if __name__ == '__main__':
    import requests

    def fetch(s, d=500):
        r = requests.get('https://api.binance.com/api/v3/klines',
                          params={'symbol': s, 'interval': '1d', 'limit': d})
        df = pd.DataFrame(r.json(), columns=['ot','o','h','l','c','v','ct',
                                              'qav','t','tb','tq','i'])
        df['ct'] = pd.to_datetime(df['ct'], unit='ms')
        df['c'] = df['c'].astype(float)
        return df.set_index('ct')['c']

    syms = ['BTCUSDT', 'ETHUSDT', 'SOLUSDT', 'BNBUSDT', 'AVAXUSDT', 'MATICUSDT']
    prices = pd.DataFrame({s: fetch(s) for s in syms}).fillna(method='ffill')
    returns = prices.pct_change().dropna()

    mu = returns.mean().values * 365
    cov = returns.cov().values * 365

    print("Asset Universe:", syms)
    print(f"\nMean Annual Returns: {dict(zip(syms, np.round(mu, 3)))}")

    print("\n=== Markowitz (λ=2, max=40%) ===")
    w_mv = mean_variance_optimize(mu, cov, risk_aversion=2.0, max_weight=0.40)
    for s, w_ in zip(syms, w_mv):
        print(f"  {s}: {w_:.3f}")
    port_ret = mu @ w_mv
    port_vol = np.sqrt(w_mv @ cov @ w_mv)
    print(f"  Expected Return: {port_ret:.2%}")
    print(f"  Volatility: {port_vol:.2%}")
    print(f"  Sharpe: {port_ret / port_vol:.2f}")

    print("\n=== Risk Parity ===")
    w_rp = risk_parity_optimize(cov)
    for s, w_ in zip(syms, w_rp):
        print(f"  {s}: {w_:.3f}")
    port_ret_rp = mu @ w_rp
    port_vol_rp = np.sqrt(w_rp @ cov @ w_rp)
    print(f"  Expected Return: {port_ret_rp:.2%}")
    print(f"  Volatility: {port_vol_rp:.2%}")
    print(f"  Sharpe: {port_ret_rp / port_vol_rp:.2f}")

    print("\n=== HRP ===")
    w_hrp = hrp_optimize(returns)
    print(w_hrp.round(3))
    port_ret_hrp = mu @ w_hrp.values
    port_vol_hrp = np.sqrt(w_hrp.values @ cov @ w_hrp.values)
    print(f"  Expected Return: {port_ret_hrp:.2%}")
    print(f"  Volatility: {port_vol_hrp:.2%}")

    print("\n=== CVaR Optimization (α=5%) ===")
    w_cvar = cvar_optimize(returns.values, alpha=0.05, max_weight=0.4)
    for s, w_ in zip(syms, w_cvar):
        print(f"  {s}: {w_:.3f}")

---

四、真实数据/案例

案例 1：Bridgewater All Weather

桥水全天候是 Risk Parity 经典实施：

• 4 大资产：股票、债券、商品、抗通胀
• 每个资产对组合贡献 25% 风险
• 杠杆补足收益（2-3x）
• 1996 至今年化 8-10%, Sharpe ~ 1.0
• 2022 大幅亏损（股债同跌，相关性突破）

案例 2：Lopez de Prado HRP 论文

2016 年原始论文：

• 实证：HRP 比 Markowitz、Min-Var 在 OOS 表现更好
• 在 100+ 资产时优势显著
• 不需 invert covariance（避免估计噪声放大）
• 已被 BlackRock、JPMorgan 等机构采用

案例 3：加密多策略组合

某基金组合（公开示例）：

• 30% TSMOM (BTC/ETH)
• 25% Pairs trading
• 20% Funding carry
• 15% Mean reversion
• 10% Cash buffer

各策略相关性：

• TSMOM vs Pairs: 0.15
• TSMOM vs Funding: 0.30
• Funding vs Pairs: 0.10

组合 Sharpe ≈ 2.0（vs 单策略平均 1.0）

案例 4：组合 Sharpe 改善示例

3 策略，每个 Sharpe = 1.0，相关性 = 0.3：

• 等权组合 Sharpe ≈ 1.5（Sharpe^2 加和但相关性减分散效应）
• HRP 组合 Sharpe ≈ 1.7（更好的风险均衡）

公式： $$ SR_{port}^2 \approx \frac{\sum SR_i^2}{1 + (n-1)\bar\rho \cdot \text{(rough)}} $$

案例 5：2022 加密熊市的组合表现

某基金（公开报告）：

• TSMOM 单策略 -45%
• Pairs trading +5%
• Funding carry -10%（funding 反转）
• Yield strategy +15%
• 组合（HRP）：-12%（vs BTC -65%）
• 教训：分散化在熊市保留资本

---

五、CEX vs DEX 策略差异

DeFi 组合特殊考虑：

1. 协议风险加权：单协议 < 30%
2. 链分散：单链 < 50%
3. Stablecoin 分散：USDC + USDT + DAI

---

六、风险管理

6.1 组合级风控

# 每日风控检查
def portfolio_risk_check(weights, cov, returns, limits):
    port_vol = np.sqrt(weights @ cov @ weights) * np.sqrt(365)
    var_95 = -np.percentile(returns @ weights, 5)
    es_95 = -returns[returns @ weights <= -var_95].mean()
    max_dd = ...

    return {
        'vol_check': port_vol <= limits['vol'],
        'var_check': var_95 <= limits['var'],
        'concentration': max(weights) <= limits['max_pos'],
    }

6.2 动态再平衡

触发条件：
- |w_actual - w_target|.max() > 5%
- 总 vol drift > 20%
- Monthly schedule

避免过度交易：

• min change threshold 1%
• max trades per day 5%

6.3 黑天鹅 reserve

总组合留 5-10% cash 应对：

• Liquidity 紧张
• 突发机会
• Margin call

---

七、关键速查

三种组合方法对比

CVXPY 模板

w = cp.Variable(n)
constraints = [cp.sum(w) == 1, w >= 0]
prob = cp.Problem(cp.Minimize(cp.quad_form(w, cov)), constraints)
prob.solve()

推荐组合配置（加密）

---

八、面试题

Q1：为什么不直接用 Markowitz？

答：Markowitz 理论最优但实操问题：

1. μ 估计敏感：参数误差被解放大
2. 极端权重：经常 100% 单资产
3. 不稳定：小输入大输出
4. 协方差矩阵：高维时 singular 改进：

• Robust optimization（不确定性集合）
• Bayesian (Black-Litterman, shrinkage)
• Risk Parity / HRP（无需 μ）
• 实战：约束下 Markowitz 也能用，但需大量约束（max_weight、sector、turnover）

Q2：HRP 相比 Risk Parity 的优势？

答：

1. 稳定性：不需 invert covariance，避免估计误差被放大
2. 聚类感知：利用 correlation 结构，相似资产作为 group 分配
3. 高维友好：100+ 资产时 RP 经常 singular，HRP 仍稳定
4. 解释性：dendrogram 直观
5. OOS 性能：Lopez de Prado 论文实证 HRP > MinVar > RP 何时用 RP：低维（< 20 资产），相关性结构简单

Q3：组合优化怎么避免在 in-sample 过拟合？

答：

1. Walk-forward optimize：每月 rolling 重训
2. Robust optimization：用 uncertainty set 而非 point estimate
3. Shrinkage：Ledoit-Wolf shrinkage of covariance matrix
4. Resampled efficient frontier (Michaud)：bootstrap optimize 多次取均值
5. Constraint-rich：约束多，自由度低，过拟合风险小
6. OOS validation：留 30% 数据做严格验证

Q4：为什么 2022 桥水全天候表现差？

答：

• 全天候依赖 stocks 与 bonds 负相关（典型 risk-off bonds 涨）
• 2022 通胀 + 紧缩，stocks 跌 -19%, bonds 跌 -15%（两资产同跌）
• 相关性突破（regime change）
• Risk Parity 在多资产同跌时分散化失效
• 教训：相关性是 backward-looking，stress-test 必须包括"correlation = 1" 极端

Q5：你怎么把 view 注入组合？

答：

• Black-Litterman：把 view 作为 prior 修正，得到 posterior μ
• Tilting：在 RP/HRP 基础上调整权重（向看好资产 +5%）
• CVXPY 约束：加 view 不等式（w_BTC ≥ w_ETH）
• Bayesian shrinkage：减少 sample μ，倾向先验 view
• 实战：直接在 Markowitz μ 上手动 +5%（"我看好 BTC 比模型多 5%"）

---

明日预告

Day 102: Week 15 复习 — 整合 Day 89-101 的所有策略，构建 Strategy Library v1，整合最好 3 个策略到统一框架。Phase 2 收官。