Week 15 复习 — Strategy Library v1

日期: 2026-08-11 方向: 量化 / 统计套利 / Alpha 阶段: Phase 2 - 统计套利与Alpha Research (Day 89-102) 标签: #量化策略 #StrategyLibrary #整合 #Phase2收官

---

今日目标

---

一、Phase 2 (Day 89-102) 全景回顾

1.1 知识图谱

┌──────────────────────────────────────────────────────┐
│            Phase 2: Statistical Arb & Alpha          │
└──────────────────────────────────────────────────────┘
            │
    ┌───────┴───────┐
    │   Week 14     │   Week 15
    ▼               ▼
[Statistical]    [Arbitrage]
[Methods]        [Strategies]
    │               │
89: Pairs        96: Triangular
90: Mean Rev     97: Funding
91: Momentum     98: Basis
92: Factors      99: Liquidation
93: Backtest    100: DeFi Yield
94: Risk        101: Portfolio Opt
95: Review      102: Library v1

1.2 14 天产出清单

---

二、Strategy Library v1 - 设计

2.1 统一接口

每个策略实现：

class Strategy(ABC):
    name: str
    capacity: float
    typical_sharpe: float

    @abstractmethod
    def evaluate(self, market_data) -> Signal:
        pass

    @abstractmethod
    def size(self, signal, capital) -> Position:
        pass

    @abstractmethod
    def rebalance_freq(self) -> str:  # 'tick', '1h', '8h', '1d'
        pass

    @abstractmethod
    def risk_metrics(self) -> Dict:
        pass

2.2 选择 3 个最佳策略

基于 Sharpe / 容量 / 复杂度 / 当前市场环境：

2.3 Capital Allocation

用 HRP 分配：

---

三、整合代码（核心产出）

# strategies/__init__.py
"""
Strategy Library v1
Unified interface for production deployment
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Optional
import numpy as np
import pandas as pd
import statsmodels.api as sm


# ----------------------------- 基类 -----------------------------
@dataclass
class Signal:
    direction: int  # -1, 0, 1
    strength: float  # 0-1
    confidence: float
    metadata: Dict


@dataclass
class Position:
    asset_legs: Dict[str, float]  # {asset: notional_usd}
    total_notional: float
    expected_apr: float


class Strategy(ABC):
    name: str = ""
    capacity_usd: float = 0
    typical_sharpe: float = 0
    rebalance_freq: str = '1d'

    @abstractmethod
    def evaluate(self, market_data: Dict) -> Signal:
        pass

    @abstractmethod
    def size(self, signal: Signal, capital: float) -> Position:
        pass

    def health_check(self) -> bool:
        return True


# ----------------------------- 策略 1: Pairs -----------------------------
class PairsStrategy(Strategy):
    name = "BTC-ETH Pairs Trading"
    capacity_usd = 50_000_000
    typical_sharpe = 1.4
    rebalance_freq = '1d'

    def __init__(self, lookback: int = 60, entry_z: float = 2.0,
                  exit_z: float = 0.5, stop_z: float = 4.0):
        self.lookback = lookback
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.stop_z = stop_z

    def _compute_zscore(self, btc_prices: pd.Series, eth_prices: pd.Series):
        log_btc = np.log(btc_prices.iloc[-self.lookback:])
        log_eth = np.log(eth_prices.iloc[-self.lookback:])
        X = sm.add_constant(log_btc.values)
        beta = np.linalg.lstsq(X, log_eth.values, rcond=None)[0]
        spread = log_eth.values - beta[0] - beta[1] * log_btc.values
        z = (spread[-1] - spread.mean()) / spread.std()
        return z, beta[1]

    def evaluate(self, market_data: Dict) -> Signal:
        btc = market_data['BTC_prices']
        eth = market_data['ETH_prices']
        z, beta = self._compute_zscore(btc, eth)

        if abs(z) > self.stop_z:
            direction = 0
            strength = 0
        elif z > self.entry_z:
            direction = -1  # short spread (short ETH long BTC)
            strength = min(abs(z) / self.entry_z - 1, 1.0)
        elif z < -self.entry_z:
            direction = 1
            strength = min(abs(z) / self.entry_z - 1, 1.0)
        else:
            direction = 0
            strength = 0

        return Signal(
            direction=direction,
            strength=strength,
            confidence=0.7,
            metadata={'z': z, 'beta': beta},
        )

    def size(self, signal: Signal, capital: float) -> Position:
        if signal.direction == 0:
            return Position({}, 0, 0)
        notional = capital * signal.strength * 0.5  # max 50% on one signal
        beta = signal.metadata['beta']
        # Long ETH (signal=1): +ETH, -beta*BTC
        eth_leg = signal.direction * notional
        btc_leg = -signal.direction * notional * beta
        return Position(
            asset_legs={'ETH_spot': eth_leg, 'BTC_spot': btc_leg},
            total_notional=abs(eth_leg) + abs(btc_leg),
            expected_apr=0.15,
        )


# ----------------------------- 策略 2: Funding Carry -----------------------------
class FundingCarryStrategy(Strategy):
    name = "ETH Funding Carry"
    capacity_usd = 200_000_000
    typical_sharpe = 1.5
    rebalance_freq = '8h'

    def __init__(self, min_apr: float = 0.10):
        self.min_apr = min_apr

    def evaluate(self, market_data: Dict) -> Signal:
        funding_8h = market_data['ETH_funding_8h']
        apr = funding_8h * 1095

        if apr > self.min_apr:
            direction = 1
            strength = min((apr - self.min_apr) / 0.20, 1.0)
        elif apr < -0.05:
            # reverse carry potential
            direction = -1
            strength = min(abs(apr) / 0.10, 1.0)
        else:
            direction = 0
            strength = 0

        return Signal(
            direction=direction,
            strength=strength,
            confidence=0.85,
            metadata={'funding_apr': apr},
        )

    def size(self, signal: Signal, capital: float) -> Position:
        if signal.direction == 0:
            return Position({}, 0, 0)
        notional = capital * signal.strength * 0.7
        if signal.direction == 1:
            # spot long + perp short
            return Position(
                asset_legs={'ETH_spot': notional, 'ETH_perp': -notional},
                total_notional=2 * notional,
                expected_apr=signal.metadata['funding_apr'],
            )
        else:
            # spot short (借) + perp long
            return Position(
                asset_legs={'ETH_spot': -notional, 'ETH_perp': notional},
                total_notional=2 * notional,
                expected_apr=-signal.metadata['funding_apr'],  # negative implies receive
            )


# ----------------------------- 策略 3: Delta Neutral (LST) -----------------------------
class DeltaNeutralLSTStrategy(Strategy):
    name = "ETH-stETH Delta Neutral"
    capacity_usd = 50_000_000
    typical_sharpe = 1.5
    rebalance_freq = '1d'

    def __init__(self, min_yield: float = 0.10, max_leverage: float = 2.5):
        self.min_yield = min_yield
        self.max_leverage = max_leverage

    def _compute_yield(self, market_data: Dict) -> Dict:
        stETH_apr = market_data.get('stETH_yield', 0.035)
        borrow_apr = market_data.get('aave_eth_borrow_rate', 0.02)
        funding_apr = market_data.get('ETH_funding_apr', 0.05)
        pendle_implied = market_data.get('pendle_PT_implied', 0.10)

        # Composite expected APR
        L = self.max_leverage
        return {
            'gross': stETH_apr * L - borrow_apr * (L - 1) + funding_apr * 0.3 + pendle_implied * 0.3,
            'leverage': L,
        }

    def evaluate(self, market_data: Dict) -> Signal:
        stETH_discount = market_data.get('stETH_ETH_ratio', 1.0)
        if stETH_discount < 0.98:
            return Signal(0, 0, 0.5, {'reason': 'stETH discount too large'})

        yields = self._compute_yield(market_data)
        if yields['gross'] < self.min_yield:
            return Signal(0, 0, 0.5, {'expected_apr': yields['gross']})

        return Signal(
            direction=1,
            strength=1.0,
            confidence=0.7,
            metadata=yields,
        )

    def size(self, signal: Signal, capital: float) -> Position:
        if signal.direction == 0:
            return Position({}, 0, 0)
        L = signal.metadata['leverage']
        # 简化分配：60% loop, 30% hedge perp, 10% Pendle PT
        return Position(
            asset_legs={
                'stETH_long': capital * L * 0.6,
                'ETH_borrow': -capital * (L - 1) * 0.6,
                'ETH_perp_short': -capital * 0.3,
                'pendle_PT': capital * 0.1,
            },
            total_notional=capital * L,
            expected_apr=signal.metadata['gross'],
        )


# ----------------------------- 组合管理 -----------------------------
@dataclass
class Allocation:
    strategy_name: str
    weight: float
    capital: float


class StrategyLibrary:
    """统一管理所有策略"""

    def __init__(self, total_capital: float = 1_000_000):
        self.strategies: Dict[str, Strategy] = {}
        self.allocations: Dict[str, float] = {}
        self.total_capital = total_capital

    def register(self, strategy: Strategy, allocation: float):
        self.strategies[strategy.name] = strategy
        self.allocations[strategy.name] = allocation

    def evaluate_all(self, market_data: Dict) -> Dict[str, Position]:
        positions = {}
        for name, strat in self.strategies.items():
            capital = self.total_capital * self.allocations[name]
            signal = strat.evaluate(market_data)
            position = strat.size(signal, capital)
            positions[name] = position
        return positions

    def aggregate_positions(self, positions: Dict[str, Position]) -> Dict[str, float]:
        """合并所有策略的资产腿"""
        agg = {}
        for pos in positions.values():
            for asset, notional in pos.asset_legs.items():
                agg[asset] = agg.get(asset, 0) + notional
        return agg

    def expected_portfolio_apr(self, positions: Dict[str, Position]) -> float:
        total_apr = 0
        total_capital = 0
        for name, pos in positions.items():
            cap = self.total_capital * self.allocations[name]
            total_apr += cap * pos.expected_apr
            total_capital += cap
        return total_apr / total_capital if total_capital > 0 else 0


# ----------------------------- 风控 -----------------------------
class PortfolioRiskManager:
    def __init__(self, max_gross: float = 3.0, max_single_protocol: float = 0.30,
                  max_var_pct: float = 0.05):
        self.max_gross = max_gross
        self.max_single_protocol = max_single_protocol
        self.max_var_pct = max_var_pct

    def check(self, positions: Dict[str, Position],
               market_data: Dict, total_capital: float) -> Dict:
        # Gross leverage
        total_notional = sum(abs(notional) for pos in positions.values()
                              for notional in pos.asset_legs.values())
        gross_leverage = total_notional / total_capital

        # Protocol exposure
        protocol_exposure = {
            'aave': abs(positions.get('ETH-stETH Delta Neutral', Position({}, 0, 0)).asset_legs.get('ETH_borrow', 0)),
            'binance_perp': sum(abs(pos.asset_legs.get('ETH_perp_short', 0)) +
                                  abs(pos.asset_legs.get('ETH_perp', 0))
                                  for pos in positions.values()),
            'lido': abs(positions.get('ETH-stETH Delta Neutral', Position({}, 0, 0)).asset_legs.get('stETH_long', 0)),
        }

        return {
            'gross_leverage': gross_leverage,
            'gross_check': gross_leverage <= self.max_gross,
            'protocol_exposure': protocol_exposure,
            'protocol_check': all(e / total_capital <= self.max_single_protocol
                                    for e in protocol_exposure.values()),
        }


# ----------------------------- 主程序 -----------------------------
if __name__ == '__main__':
    # 模拟市场数据
    import requests

    def fetch(s, d=200):
        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']

    btc_p = fetch('BTCUSDT')
    eth_p = fetch('ETHUSDT')

    market_data = {
        'BTC_prices': btc_p,
        'ETH_prices': eth_p,
        'ETH_funding_8h': 0.00012,  # 0.012% / 8h ≈ 13% APR
        'stETH_yield': 0.035,
        'aave_eth_borrow_rate': 0.020,
        'ETH_funding_apr': 0.13,
        'pendle_PT_implied': 0.12,
        'stETH_ETH_ratio': 0.999,
    }

    # 注册
    lib = StrategyLibrary(total_capital=1_000_000)
    lib.register(PairsStrategy(), allocation=0.30)
    lib.register(FundingCarryStrategy(), allocation=0.40)
    lib.register(DeltaNeutralLSTStrategy(), allocation=0.25)

    # 评估
    positions = lib.evaluate_all(market_data)

    print("=" * 60)
    print("Strategy Library v1 - Portfolio Snapshot")
    print("=" * 60)
    for name, pos in positions.items():
        print(f"\n{name}:")
        print(f"  Total notional: ${pos.total_notional:,.0f}")
        print(f"  Expected APR: {pos.expected_apr:.2%}")
        for asset, notional in pos.asset_legs.items():
            print(f"    {asset}: ${notional:,.0f}")

    # 总组合
    print("\n" + "=" * 60)
    print("Aggregated Positions")
    print("=" * 60)
    agg = lib.aggregate_positions(positions)
    for asset, notional in agg.items():
        print(f"  {asset}: ${notional:,.0f}")

    print(f"\nExpected Portfolio APR: {lib.expected_portfolio_apr(positions):.2%}")

    # 风控
    risk_mgr = PortfolioRiskManager()
    risk_check = risk_mgr.check(positions, market_data, 1_000_000)
    print("\n" + "=" * 60)
    print("Risk Checks")
    print("=" * 60)
    for k, v in risk_check.items():
        print(f"  {k}: {v}")

---

四、整合策略表现期望

---

五、Phase 2 关键 Takeaways

5.1 方法论层面

1. 协整 ≠ 相关：配对交易的根基
2. Vol targeting 是 free lunch：30-50% Sharpe 提升
3. HRP > Markowitz：高维下稳定
4. Walk-forward + DSR：避免过拟合的最低门槛
5. Multi-strategy 组合 Sharpe ≈ 单策略 1.5x（uncorrelated 假设下）

5.2 加密特殊性

1. Fat tail：偏度 -2 到 -5，必须 stress test
2. Regime change 频繁：FTX/LUNA/SVB
3. DeFi 原生 alpha：清算、LST、Pendle、AMM 套利
4. MEV 占据简单 alpha：剩余空间在复杂多腿和 niche
5. 24/7 + 全球化：策略容量 / 监控要求高

5.3 实战教训

1. 3AC 教训：杠杆 + 流动性资产是致命组合
2. FTX 教训：单一对手方风险
3. stETH depeg 教训：流动性溢价的 mean revert 假设可能崩
4. funding 反转教训：carry trade 不是 risk-free

---

六、面试准备总结

6.1 核心面试题清单

统计套利类：

• Engle-Granger vs Johansen 协整
• OU 过程在加密的局限性
• HRP vs Markowitz
• TSMOM 怎么防过拟合

套利类：

• DEX 套利为什么不赚钱了
• Funding 反转怎么处理
• Liquidation bot 普通人能做吗

风险类：

• VaR vs ES 区别
• Kelly 公式为什么爆仓
• 加密策略压力测试

组合类：

• 你怎么组合 3 个策略
• 为什么不直接 Markowitz
• 分散化在尾部失效怎么办

---

七、Phase 3 预告

后续课程将进入Day 103-XXX: Phase 3 Algo Trading & MM：

• 高频交易（HFT）
• 做市商策略
• Order book microstructure
• 智能订单路由（SOR）
• 算法执行（VWAP/TWAP/IS）

---

八、产出 Inventory

完成 14 天后已有：

docs/daily/
├── EXPERT-DAY89.md  (Pairs)
├── EXPERT-DAY90.md  (MR)
├── EXPERT-DAY91.md  (Momentum)
├── EXPERT-DAY92.md  (Factors)
├── EXPERT-DAY93.md  (Backtest)
├── EXPERT-DAY94.md  (Risk)
├── EXPERT-DAY95.md  (Review + alpha_report.md)
├── EXPERT-DAY96.md  (Triangular)
├── EXPERT-DAY97.md  (Funding)
├── EXPERT-DAY98.md  (Basis)
├── EXPERT-DAY99.md  (Liquidation)
├── EXPERT-DAY100.md (DeFi Yield + dn_strategy.md)
├── EXPERT-DAY101.md (Portfolio)
└── EXPERT-DAY102.md (Library v1)

代码文件（推荐 git 仓库结构）：

quant_lib/
├── strategies/
│   ├── pairs.py
│   ├── mr.py
│   ├── momentum.py
│   ├── factors.py
│   ├── arb.py
│   ├── funding_arb.py
│   ├── basis.py
│   ├── liq_bot.py
│   └── library.py  (this Day 102)
├── backtest/
│   ├── data.py
│   ├── signal.py
│   ├── execution.py
│   ├── risk.py
│   ├── engine.py
│   ├── analysis.py
│   └── walk_forward.py
├── risk.py        (Day 94)
└── portfolio.py   (Day 101)

---

九、最后一道面试题

Q：你怎么向你的 LP/老板解释你的整套量化体系？

答（30 秒电梯演讲）：

我们的系统包括三个互补 alpha 来源：

1. 统计套利（pairs trading）— 经典价差均值回归，验证 8 年历史 Sharpe 1.4
2. Funding carry — 永续合约资金费率收割，结构性 alpha，容量 $200M
3. Delta-neutral DeFi yield — 利用 LST staking + Pendle PT 锁定收益

三策略相关性 < 0.3，组合 HRP 加权后 Sharpe 约 2.0，目标 APR 15-18%，最大回撤 < 12%。

风控三层：单策略 vol target、组合 VaR limit、协议风险分散（无单协议 > 30%）。 Walk-forward 每月再优化，stress test 包括 LUNA/FTX/USDC 历史压力期。

现在 $1M 容量，扩容到 $50M 估计 6 月，再到 $100M+ 需要扩展策略池或进一步多元化。

完成 Phase 2 14 天 / 收官 Day 102 / 进入 Phase 3 🎯