风险管理（VaR/ES/回撤/Sharpe）

日期: 2026-08-03 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #风险管理 #VaR #ES #Sharpe #Kelly

---

今日目标

---

一、理论与模型

1.1 Value at Risk (VaR)

定义：损失超过 VaR 的概率为 $\alpha$

$$ P(L > \text{VaR}\alpha) = \alpha, \quad \text{VaR}\alpha = -F_L^{-1}(1 - \alpha) $$

三种计算方法

1) Historical VaR：

VaR = -np.percentile(returns, 5)  # 5% VaR

2) Parametric VaR（假设正态）：

$$ \text{VaR}\alpha = -(\mu - z\alpha \sigma) $$

z₀.₀₅ = 1.645, z₀.₀₁ = 2.326

3) Monte Carlo VaR：模拟未来路径，取分位数

1.2 Expected Shortfall (ES / CVaR)

VaR 的尾部条件期望：

$$ \text{ES}\alpha = E[L | L > \text{VaR}\alpha] $$

为什么 ES 优于 VaR：

• VaR 不是一致风险度量（不次可加）：$\text{VaR}(X+Y) > \text{VaR}(X) + \text{VaR}(Y)$ 可能
• ES 满足次可加性，组合 ES ≤ 各资产 ES 之和
• Basel III 已将 ES 列为银行资本计算标准（替代 VaR）

1.3 收益风险比指标

1.4 最大回撤

$$ \text{DD}t = \frac{V_t - \max{\tau \le t} V_\tau}{\max_{\tau \le t} V_\tau} $$

$$ \text{MaxDD} = \min_t \text{DD}_t $$

附加指标：

• Drawdown duration：从峰值到回到峰值的时间
• Underwater curve：累计低于峰值的天数
• Conditional drawdown at risk (CDaR)：尾部回撤期望

1.5 Kelly Criterion

最大化对数效用的最优仓位：

$$ f^* = \frac{\mu}{\sigma^2} $$

（连续时间，对数效用，单期）

多资产 Kelly：

$$ \mathbf{f}^* = \Sigma^{-1} \boldsymbol{\mu} $$

实战调整：

• 半 Kelly（× 0.5）：减少回撤，损失少量 ER
• 1/4 Kelly：高度保守
• Fractional Kelly = target_drawdown / max_loss

加密 Kelly 注意：参数估计误差大，full Kelly 经常爆仓。

1.6 压力测试

历史情景：

• 2018-01-17 BTC -22% 单日
• 2020-03-12 ("黑色星期四") BTC -50% 24h
• 2022-05-12 LUNA 崩盘
• 2022-11-08 FTX 暴雷
• 2023-03-11 USDC 脱钩

测试方法：

1. 历史重放：把策略 weight 应用到历史压力期
2. 假设性场景：BTC 一日 -30% + funding 冲击
3. Sensitivity：每个因子单独移动 ±3σ

---

二、直觉与陷阱

陷阱 1：VaR 的伪安全感

VaR 不告诉你最坏情况。VaR=$10K，可能 99% 时不亏过 $10K，但 1% 时亏 $1M。 解法：必须配合 ES 使用。

陷阱 2：年化 Sharpe 的频率陷阱

sharpe_daily = r.mean() / r.std()
sharpe_annual = sharpe_daily * np.sqrt(252)  # 股票
sharpe_annual = sharpe_daily * np.sqrt(365)  # 加密（24/7）

频率搞错 → Sharpe 偏差 √2 倍。

陷阱 3：Sharpe 在非正态下的失真

加密 fat tail：偏度 -2 到 -5（下行尾部肥），峰度 10+。

• 高 Sharpe 可能掩盖大尾部风险
• LTCM Sharpe = 4.0，1998 年 -90%

解法：用 PSR 校正、配合 CVaR、报告偏度峰度。

陷阱 4：MaxDD 的单一时点偏差

只看 MaxDD 数字 -30% 等不了解：

• 是连续 5 天 -30% 还是 6 个月 -30%？
• 之后多久回到峰值？

解法：报告 underwater duration 和 recovery time。

陷阱 5：Kelly 的参数估计误差

样本估的 $\hat\mu$ 误差 SE = $\sigma / \sqrt{n}$ 远大于实际 $\mu$。 $\hat\mu$ 偏差 +50% 就让 Kelly 仓位翻倍 → 实盘爆仓。 解法：用 1/4 Kelly + 缩水估计 (shrinkage)。

陷阱 6：相关性在压力时跳跃

牛市中 BTC/ETH 相关 0.7，但 2022-11 FTX 时跳到 0.95。组合分散化失效。 解法：用 stress-test 相关性（条件相关）做 risk budget。

---

三、代码实现

3.1 完整风险指标库

# risk.py
"""
Risk Management Toolkit
- VaR, ES, drawdown
- Sharpe family
- Kelly sizing
- Stress tests
"""

import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, List, Optional


# ----------------------------- VaR / ES -----------------------------
class VaRCalculator:
    @staticmethod
    def historical(returns: pd.Series, alpha: float = 0.05) -> float:
        return -np.percentile(returns.dropna(), alpha * 100)

    @staticmethod
    def parametric(returns: pd.Series, alpha: float = 0.05) -> float:
        mu = returns.mean()
        sigma = returns.std()
        z = stats.norm.ppf(1 - alpha)
        return -(mu - z * sigma)

    @staticmethod
    def cornish_fisher(returns: pd.Series, alpha: float = 0.05) -> float:
        """考虑偏度峰度的修正 VaR"""
        r = returns.dropna()
        mu = r.mean()
        sigma = r.std()
        skew = r.skew()
        kurt = r.kurtosis()  # excess kurtosis

        z = stats.norm.ppf(1 - alpha)
        z_cf = (z + (z**2 - 1) * skew / 6 +
                (z**3 - 3*z) * kurt / 24 -
                (2*z**3 - 5*z) * skew**2 / 36)
        return -(mu - z_cf * sigma)

    @staticmethod
    def monte_carlo(returns: pd.Series, alpha: float = 0.05,
                     n_sims: int = 10_000, horizon: int = 1) -> float:
        mu = returns.mean()
        sigma = returns.std()
        np.random.seed(42)
        # Multi-step compound (Geometric Brownian Motion approximation)
        sims = np.random.normal(mu, sigma, (n_sims, horizon))
        cum_ret = (1 + sims).prod(axis=1) - 1
        return -np.percentile(cum_ret, alpha * 100)


def expected_shortfall(returns: pd.Series, alpha: float = 0.05) -> float:
    """Historical ES (CVaR)"""
    var = VaRCalculator.historical(returns, alpha)
    tail = returns[returns <= -var]
    return -tail.mean() if len(tail) > 0 else var


# ----------------------------- 收益风险比 -----------------------------
class PerformanceRatios:
    @staticmethod
    def sharpe(returns: pd.Series, rf: float = 0, freq: int = 365) -> float:
        excess = returns - rf / freq
        if excess.std() == 0:
            return 0
        return excess.mean() / excess.std() * np.sqrt(freq)

    @staticmethod
    def sortino(returns: pd.Series, target: float = 0, freq: int = 365) -> float:
        downside = returns[returns < target]
        if len(downside) == 0 or downside.std() == 0:
            return np.inf if returns.mean() > target else 0
        return (returns.mean() - target) / downside.std() * np.sqrt(freq)

    @staticmethod
    def calmar(returns: pd.Series, freq: int = 365) -> float:
        ann_ret = (1 + returns.mean()) ** freq - 1
        equity = (1 + returns).cumprod()
        max_dd = (equity / equity.cummax() - 1).min()
        return ann_ret / abs(max_dd) if max_dd < 0 else np.inf

    @staticmethod
    def omega(returns: pd.Series, threshold: float = 0) -> float:
        gains = (returns - threshold).clip(lower=0).sum()
        losses = (threshold - returns).clip(lower=0).sum()
        return gains / losses if losses > 0 else np.inf

    @staticmethod
    def information_ratio(returns: pd.Series, benchmark: pd.Series,
                           freq: int = 365) -> float:
        excess = returns - benchmark
        if excess.std() == 0:
            return 0
        return excess.mean() / excess.std() * np.sqrt(freq)

    @staticmethod
    def tail_ratio(returns: pd.Series) -> float:
        """右尾 95% / 左尾 5% 比值"""
        return abs(np.percentile(returns, 95)) / abs(np.percentile(returns, 5))


# ----------------------------- 回撤分析 -----------------------------
def drawdown_series(returns: pd.Series) -> pd.DataFrame:
    equity = (1 + returns).cumprod()
    peak = equity.cummax()
    dd = equity / peak - 1
    underwater = (dd < 0).astype(int)
    return pd.DataFrame({
        'equity': equity,
        'peak': peak,
        'drawdown': dd,
        'underwater': underwater,
    })


def drawdown_analysis(returns: pd.Series) -> Dict:
    dd_df = drawdown_series(returns)
    max_dd = dd_df['drawdown'].min()
    max_dd_date = dd_df['drawdown'].idxmin()

    # Recovery
    peak_before = dd_df.loc[:max_dd_date, 'peak'].iloc[-1]
    after_max_dd = dd_df.loc[max_dd_date:]
    recovered = after_max_dd[after_max_dd['equity'] >= peak_before]
    recovery_date = recovered.index[0] if len(recovered) > 0 else None
    recovery_days = (recovery_date - max_dd_date).days if recovery_date else None

    # Underwater duration
    underwater_streaks = []
    current_streak = 0
    for u in dd_df['underwater']:
        if u:
            current_streak += 1
        else:
            if current_streak > 0:
                underwater_streaks.append(current_streak)
            current_streak = 0
    if current_streak > 0:
        underwater_streaks.append(current_streak)

    return {
        'max_drawdown': max_dd,
        'max_dd_date': max_dd_date,
        'recovery_date': recovery_date,
        'recovery_days': recovery_days,
        'avg_underwater_duration': np.mean(underwater_streaks) if underwater_streaks else 0,
        'max_underwater_duration': max(underwater_streaks) if underwater_streaks else 0,
        'pct_time_underwater': dd_df['underwater'].mean(),
    }


# ----------------------------- Kelly Sizing -----------------------------
class KellySizer:
    @staticmethod
    def single_asset(returns: pd.Series, fraction: float = 0.5) -> float:
        """单资产 Kelly: f* = μ / σ² × fraction"""
        mu = returns.mean()
        var = returns.var()
        if var == 0:
            return 0
        return (mu / var) * fraction

    @staticmethod
    def multi_asset(returns_df: pd.DataFrame, fraction: float = 0.5,
                     shrinkage: float = 0.5) -> pd.Series:
        """
        多资产 Kelly: f* = Σ^-1 μ × fraction
        shrinkage: 协方差矩阵收缩（减少估计误差）
        """
        mu = returns_df.mean()
        cov = returns_df.cov()
        # Ledoit-Wolf 收缩到对角阵
        target = np.diag(np.diag(cov.values))
        cov_shrunk = (1 - shrinkage) * cov.values + shrinkage * target
        try:
            inv_cov = np.linalg.inv(cov_shrunk)
        except np.linalg.LinAlgError:
            inv_cov = np.linalg.pinv(cov_shrunk)
        kelly = pd.Series(inv_cov @ mu.values, index=returns_df.columns)
        return kelly * fraction


# ----------------------------- 压力测试 -----------------------------
class StressTest:
    HISTORICAL_SCENARIOS = {
        'BTC_2018_crash': {'BTC': -0.22, 'ETH': -0.25, 'SOL': -0.30},
        'COVID_2020': {'BTC': -0.50, 'ETH': -0.55, 'SOL': -0.60},
        'LUNA_collapse_2022': {'BTC': -0.18, 'ETH': -0.22, 'SOL': -0.30,
                                'LUNA': -0.99, 'UST': -0.95},
        'FTX_2022': {'BTC': -0.25, 'ETH': -0.27, 'SOL': -0.55, 'FTT': -0.92},
        'USDC_depeg_2023': {'USDC': -0.13, 'USDT': 0.005},
    }

    @staticmethod
    def historical_scenario(weights: Dict[str, float],
                             scenario: str) -> float:
        shocks = StressTest.HISTORICAL_SCENARIOS.get(scenario, {})
        loss = sum(weights.get(asset, 0) * shock
                    for asset, shock in shocks.items())
        return loss

    @staticmethod
    def custom_shock(weights: Dict[str, float],
                      shocks: Dict[str, float]) -> float:
        return sum(weights.get(asset, 0) * shock
                    for asset, shock in shocks.items())

    @staticmethod
    def correlation_breakdown(weights: pd.Series, returns_df: pd.DataFrame,
                                stress_corr: float = 0.95) -> float:
        """
        模拟相关性突破到 stress_corr，单资产 vol 不变
        计算压力下的组合 vol
        """
        vols = returns_df.std()
        n = len(weights)
        corr_matrix = np.full((n, n), stress_corr)
        np.fill_diagonal(corr_matrix, 1.0)
        cov_matrix = corr_matrix * np.outer(vols.values, vols.values)
        port_var = weights.values @ cov_matrix @ weights.values
        return np.sqrt(port_var)


# ----------------------------- 综合评估 -----------------------------
def full_risk_report(returns: pd.Series, freq: int = 365) -> Dict:
    """完整风险报告"""
    r = returns.dropna()
    return {
        'mean_daily': r.mean(),
        'std_daily': r.std(),
        'ann_return': (1 + r.mean()) ** freq - 1,
        'ann_vol': r.std() * np.sqrt(freq),
        'skew': r.skew(),
        'kurtosis': r.kurtosis(),
        'sharpe': PerformanceRatios.sharpe(r, freq=freq),
        'sortino': PerformanceRatios.sortino(r, freq=freq),
        'calmar': PerformanceRatios.calmar(r, freq=freq),
        'omega': PerformanceRatios.omega(r),
        'tail_ratio': PerformanceRatios.tail_ratio(r),
        'var_95_hist': VaRCalculator.historical(r, 0.05),
        'var_99_hist': VaRCalculator.historical(r, 0.01),
        'var_95_param': VaRCalculator.parametric(r, 0.05),
        'var_95_cf': VaRCalculator.cornish_fisher(r, 0.05),
        'es_95': expected_shortfall(r, 0.05),
        'es_99': expected_shortfall(r, 0.01),
        **drawdown_analysis(r),
    }


# ----------------------------- 主程序 -----------------------------
if __name__ == '__main__':
    import requests
    r = requests.get('https://api.binance.com/api/v3/klines',
                      params={'symbol': 'BTCUSDT', 'interval': '1d', 'limit': 1000})
    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)
    btc_ret = df.set_index('ct')['c'].pct_change().dropna()

    report = full_risk_report(btc_ret)
    print("BTC Risk Report")
    print("=" * 60)
    for k, v in report.items():
        if isinstance(v, float):
            print(f"  {k}: {v:.4f}")
        else:
            print(f"  {k}: {v}")

    # Kelly
    print("\nKelly Sizing")
    f_full = KellySizer.single_asset(btc_ret, fraction=1.0)
    f_half = KellySizer.single_asset(btc_ret, fraction=0.5)
    f_quarter = KellySizer.single_asset(btc_ret, fraction=0.25)
    print(f"  Full Kelly: {f_full:.2%}")
    print(f"  Half Kelly: {f_half:.2%}")
    print(f"  Quarter Kelly: {f_quarter:.2%}")

    # Stress test
    print("\nStress Test (50% BTC + 50% ETH portfolio)")
    weights = {'BTC': 0.5, 'ETH': 0.5}
    for sc in ['COVID_2020', 'LUNA_collapse_2022', 'FTX_2022']:
        loss = StressTest.historical_scenario(weights, sc)
        print(f"  {sc}: {loss:.2%}")

---

四、真实数据/案例

案例 1：3AC 的风险管理失败

3AC AUM $10B → 0 (2022)：

• Full Kelly + 杠杆 5x 在 GBTC、stETH、LUNA 多个仓位
• VaR 模型用历史 95% 置信，但 LUNA 是 100σ 事件
• 没做 stress test 包括 stablecoin 脱钩
• 教训：Tail risk + 流动性风险被低估

案例 2：FTX 客户资金风险案例

FTX 用 FTT 作担保，FTT 历史 vol 2x 高，但 11 月 8 日 FTT 一日 -90%。

• Parametric VaR 完全失效（tail event）
• 流动性风险：抛售 FTT 推低自身价格
• 教训：用单一代币做担保有 reflexivity 风险

案例 3：LTCM 启示对加密的意义

LTCM (1998)：

• 25 PhD + 诺贝尔奖团队
• Sharpe = 4.0
• 杠杆 25x
• 1998 俄罗斯违约，回撤 90%

加密对照：

• 高 Sharpe 不代表无尾部风险
• 流动性消失时杠杆爆炸性放大损失
• "100 年一次"事件每 5 年发生一次

案例 4：CVaR 监管化（Basel III）

银行业 2019 起用 Expected Shortfall 替代 VaR：

• VaR 不能反映尾部
• ES 在压力期更准确
• 加密机构（Galaxy/Coinbase Institutional）已 follow

---

五、CEX vs DEX 策略差异

DeFi 风险量化挑战：

1. 协议风险：用历史漏洞频率（Code4rena 数据）估计
2. Oracle 风险：检验 oracle 多源 + TWAP
3. Governance 风险：top 10 holder 占比 > 50% 警报

---

六、风险管理

6.1 风控金字塔

       [策略]
        ↓
       [仓位]  ← Kelly / vol target
        ↓
       [组合]  ← 单仓上限 / 行业上限
        ↓
       [基金]  ← VaR limit / Margin limit
        ↓
       [机构]  ← Capital limit / Stress test

6.2 风险预算（Risk Budget）

# 等风险贡献（Risk Parity）
def equal_risk_contribution(cov_matrix, target_vol):
    n = cov_matrix.shape[0]
    # 每个资产贡献 1/n 总风险
    weights = 1 / np.sqrt(np.diag(cov_matrix))
    weights /= weights.sum()
    return weights * target_vol / np.sqrt(weights @ cov_matrix @ weights)

6.3 风控限额示例

---

七、关键速查

风险指标速记

Kelly 实战

---

八、面试题

Q1：VaR 和 ES 的区别？为什么 Basel III 改用 ES？

答：

• VaR：损失超过阈值的概率 = α，但不告诉超过多少
• ES：α 尾部的条件期望，反映极端损失大小
• VaR 不是一致风险度量（缺次可加性），合并组合可能 VaR 上升 → 不合理
• ES 满足次可加性，符合分散化原则
• 2008 危机暴露 VaR 缺陷（VaR 看似可控但 tail loss 巨大），Basel III FRTB 改用 97.5% ES

Q2：Sharpe = 2.0 的策略你买吗？

答：取决于：

1. 样本量：n=100 的 Sharpe=2 不可信，n=1000+ 可信
2. 偏度：负偏度（fat left tail）则 Sharpe 高估真实风险
3. 峰度：高峰度 + 高 Sharpe 可能掩盖崩溃风险
4. PSR/DSR：考虑 N 次试错的真实显著性
5. 回撤：MaxDD < 15% 才合格
6. 容量：可投 $1M？$100M？容量影响实际可用性
7. 稳定性：2010-2015、2015-2020、2020-2025 各段 SR 都 > 1？

Q3：Kelly 公式在加密为什么经常爆？

答：

• 估计 μ 误差很大（小样本）：SE(μ̂) = σ/√n，相对误差经常 > 100%
• σ 估计不稳定（vol clustering）
• 假设的 normal 分布严重违反
• 真实 fat tail 让 full Kelly 在尾部事件爆仓
• 解：1/4 Kelly + bayesian shrinkage + 频繁 recalibrate

Q4：怎么做加密策略的压力测试？

答：

1. 历史情景重放：把当前 weight 应用到历史压力期（COVID/LUNA/FTX）
2. 假设性单边冲击：BTC -30% 一日 + funding +0.5%/8h
3. 相关性突破：所有相关性跳到 0.95
4. 流动性枯竭：spread × 5、order book 深度 / 10
5. 协议风险（DeFi）：top 3 仓位有 1 个 100% 损失
6. 稳定币脱钩：USDC -15%, USDT -10%
7. 执行报告：列每种压力的损失，超出限额的需调整组合

Q5：怎么评价两个策略哪个更好？

答： 不只是 Sharpe，要看：

• Sharpe + Sortino：下行风险敏感
• Calmar：回撤约束敏感
• Tail risk：偏度、峰度、ES
• 稳定性：滚动 Sharpe 一致性
• 容量：单位 AUM 的 alpha
• 与现有组合相关性：低相关性的"分散化 alpha" 更值钱
• 可解释性：能讲清楚 alpha 来源（行为/微观结构/不对称信息）
• 样本外验证：out-of-sample Sharpe / in-sample > 0.5

---

明日预告

Day 95: Week 14 复习 — 整合 Day 89-94 的方法论，写一份完整的"加密 alpha research 报告"（pairs/MR/momentum/factor/backtest/risk 全栈）。