4. 做市商库存风险管理

任务目标

为期权做市商账户建立基于希腊字母的库存风险限额控制：设定Delta、Gamma、Vega的日内限额，实时监控持仓风险暴露，当任一指标超过限额时自动触发对冲指令或收窄报价宽度，将风险拉回安全区间。本案例聚焦限额计算、超限检测与对冲量生成的核心逻辑，不涉及交易所接口对接。

环境与依赖

  • Python 3.9+，仅使用标准库（math, dataclasses, typing）
  • 期权定价采用Black-Scholes模型（本案例内实现），希腊字母通过解析公式计算

核心实现

import math
from dataclasses import dataclass, field
from typing import List, Dict

# ---------- Black-Scholes 定价与希腊字母 ----------
def bs_price(S, K, T, r, sigma, option_type='call'):
    if T <= 0 or sigma <= 0:
        raise ValueError("T and sigma must be positive")
    d1 = (math.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*math.sqrt(T))
    d2 = d1 - sigma*math.sqrt(T)
    if option_type == 'call':
        return S*N(d1) - K*math.exp(-r*T)*N(d2)
    else:
        return K*math.exp(-r*T)*N(-d2) - S*N(-d1)

def N(x):
    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))

def greeks(S, K, T, r, sigma, option_type='call'):
    """返回 delta, gamma, vega(每1%波动), theta(每日)"""
    d1 = (math.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*math.sqrt(T))
    d2 = d1 - sigma*math.sqrt(T)
    pdf_d1 = math.exp(-0.5*d1*d1) / math.sqrt(2*math.pi)
    
    if option_type == 'call':
        delta = N(d1)
    else:
        delta = N(d1) - 1.0
    gamma = pdf_d1 / (S * sigma * math.sqrt(T))
    vega = S * pdf_d1 * math.sqrt(T) / 100.0  # 每1% vega
    theta = -(S*pdf_d1*sigma) / (2*math.sqrt(T)) - r*K*math.exp(-r*T)*N(d2 if option_type=='call' else -d2)
    return delta, gamma, vega, theta/365.0

# ---------- 持仓与限额数据结构 ----------
@dataclass
class Position:
    symbol: str
    option_type: str
    K: float
    T: float
    quantity: int          # 正=多头, 负=空头
    sigma: float = 0.25

@dataclass
class RiskLimits:
    delta_max: float = 500.0
    gamma_max: float = 300.0
    vega_max: float = 2000.0

@dataclass
class RiskState:
    delta: float = 0.0
    gamma: float = 0.0
    vega: float = 0.0
    breaches: List[str] = field(default_factory=list)

# ---------- 组合风险聚合 ----------
def compute_portfolio_greeks(positions: List[Position], S: float, r: float) -> RiskState:
    state = RiskState()
    for p in positions:
        d, g, v, _ = greeks(S, p.K, p.T, r, p.sigma, p.option_type)
        state.delta += d * p.quantity
        state.gamma += g * p.quantity
        state.vega += v * p.quantity
    return state

# ---------- 限额检测与对冲决策 ----------
def check_limits(state: RiskState, limits: RiskLimits) -> RiskState:
    state.breaches.clear()
    if abs(state.delta) > limits.delta_max:
        state.breaches.append(f"delta:{state.delta:.1f}>{limits.delta_max}")
    if abs(state.gamma) > limits.gamma_max:
        state.breaches.append(f"gamma:{state.gamma:.1f}>{limits.gamma_max}")
    if abs(state.vega) > limits.vega_max:
        state.breaches.append(f"vega:{state.vega:.1f}>{limits.vega_max}")
    return state

def generate_hedge_orders(state: RiskState, limits: RiskLimits, S: float) -> Dict[str, float]:
    """返回需要交易的标的与期权数量（正=买入，负=卖出）"""
    orders = {}
    if "delta" in str(state.breaches):
        # 用标的对冲Delta：需要卖出delta份标的
        orders["underlying"] = -state.delta
    if "gamma" in str(state.breaches) or "vega" in str(state.breaches):
        # 用近月平值期权对冲Gamma/Vega（简化：假设对冲期权delta=0.5, gamma=0.05, vega=8）
        hedge_delta = 0.5
        hedge_gamma = 0.05
        hedge_vega = 8.0
        # 优先满足gamma，再检查vega
        qty_gamma = -state.gamma / hedge_gamma
        qty_vega = -state.vega / hedge_vega
        qty = qty_gamma if abs(qty_gamma) > abs(qty_vega) else qty_vega
        orders["hedge_option"] = qty
        # 对冲期权自身delta影响
        orders["underlying"] = orders.get("underlying", 0.0) - qty * hedge_delta
    return orders

# ---------- 报价调整（收窄宽度） ----------
def adjust_quote_spread(breaches: List[str], base_spread: float = 0.10) -> float:
    """超限时收窄报价宽度以降低成交概率"""
    if not breaches:
        return base_spread
    penalty = 0.02 * len(breaches)  # 每个超限指标收窄0.02
    return max(0.02, base_spread - penalty)

# ---------- 主控流程 ----------
def run_risk_control(positions: List[Position], limits: RiskLimits, S: float, r: float = 0.02):
    state = compute_portfolio_greeks(positions, S, r)
    state = check_limits(state, limits)
    
    if state.breaches:
        orders = generate_hedge_orders(state, limits, S)
        spread = adjust_quote_spread(state.breaches)
        return {
            "status": "BREACH",
            "breaches": state.breaches,
            "hedge_orders": orders,
            "adjusted_spread": spread
        }
    else:
        return {"status": "OK", "breaches": [], "hedge_orders": {}, "adjusted_spread": 0.10}

# ---------- 演示入口 ----------
if __name__ == "__main__":
    # 构造持仓：卖出10张call（每张100股），买入5张put
    positions = [
        Position("C1", "call", 100.0, 0.5, -10, 0.20),
        Position("P1", "put", 95.0, 0.5, 5, 0.25),
    ]
    limits = RiskLimits(delta_max=200, gamma_max=150, vega_max=800)
    result = run_risk_control(positions, 100.0, 0.02)
    print(result)

运行方式

python risk_control.py

验证方法

运行后观察输出：

  • 若status为OK，说明组合希腊字母均在限额内，报价宽度保持0.10
  • 若status为BREACH，检查breaches列表是否准确列出超限指标，hedge_orders中的对冲数量符号是否正确（Delta超限卖出标的，Gamma/Vega超限买入对冲期权）
  • 手动修改limits中的限额值（如将delta_max设为50），再次运行应触发Delta超限告警

通过条件：程序能正确计算组合希腊字母暴露，超限时输出明确的对冲指令与收窄后的报价宽度，且对冲方向与超限方向相反（Delta为正则卖出标的，Gamma为正则卖出期权）。

常见问题

  • 若出现ValueError: T and sigma must be positive，检查持仓中T是否大于0（到期时间以年为单位）
  • 对冲期权参数（delta=0.5, gamma=0.05, vega=8）为简化示例，实际使用时应根据真实期权计算
  • 本案例未考虑交易成本与流动性约束，实盘需增加滑点与最小交易单位处理
← 上一章 📖 返回目录 下一章 →