Black-Scholes 加密货币期权定价：波动率微笑与波动率倾斜处理技巧

这段代码的核心价值在于：通过 HolySheep AI API 调用大模型，让它帮我们分析波动率曲面特征并给出 SVI 参数的初始猜测。我个人经验是，手动调参 SVI 通常需要 2-3 小时迭代，而借助 AI 的金融知识库，30 分钟内就能得到可用参数。

波动率倾斜的动态校准实现

from datetime import datetime, timedelta
from scipy.optimize import minimize
import json

class VolatilitySkewCalibrator:
    """波动率倾斜动态校准器"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.calibration_history: List[dict] = []
    
    def calibrate_skew(self, market_data: dict, 
                       model: str = "deepseek-chat") -> dict:
        """
        校准波动率倾斜参数
        
        market_data: {
            "btc_spot": 67500.0,
            "options": [
                {"strike": 60000, "expiry": "2026-01-31", "iv": 0.72, "type": "put"},
                {"strike": 65000, "expiry": "2026-01-31", "iv": 0.58, "type": "put"},
                {"strike": 70000, "expiry": "2026-01-31", "iv": 0.52, "type": "call"},
                {"strike": 75000, "expiry": "2026-01-31", "iv": 0.68, "type": "call"},
            ]
        }
        """
        
        # 1. 构建校准优化问题
        skew_model = self._build_skew_model(market_data)
        
        # 2. 使用 AI 分析市场结构
        analysis_prompt = f"""
分析以下 BTC 期权市场的波动率倾斜特征：

当前标的价格：${market_data['btc_spot']}
期权观察数据：{json.dumps(market_data['options'], indent=2)}

请分析：
1. 当前 skew 的斜率和曲率
2. 哪个行权价区域的流动性最好
3. 建议的校准权重分配（用于加权最小二乘）
4. 是否存在流动性折价或溢价

输出 JSON 格式的分析结果。
"""
        
        ai_analysis = self._call_ai_model(analysis_prompt, model)
        
        # 3. 执行数值优化
        optimized_params = self._optimize_skew_params(
            market_data, 
            json.loads(ai_analysis)
        )
        
        # 4. 验证校准结果
        validation_result = self._validate_calibration(
            market_data, 
            optimized_params
        )
        
        self.calibration_history.append({
            "timestamp": datetime.now(),
            "params": optimized_params,
            "validation": validation_result
        })
        
        return optimized_params
    
    def _build_skew_model(self, market_data: dict) -> callable:
        """构建 skew 参数化模型"""
        def skew_function(k: float, params: np.ndarray) -> float:
            """
            参数化 skew 模型：
            - params[0]: skew 斜率
            - params[1]: skew 曲率
            - params[2]: atm_vol（平值波动率）
            """
            slope, curvature, atm_vol = params
            moneyness = np.log(k / market_data['btc_spot'])
            return atm_vol + slope * moneyness + curvature * moneyness ** 2
        return skew_function
    
    def _optimize_skew_params(self, market_data: dict, 
                             ai_weights: dict) -> dict:
        """使用 L-BFGS-B 优化 skew 参数"""
        
        def objective(params: np.ndarray) -> float:
            total_error = 0.0
            for opt in market_data['options']:
                strike = opt['strike']
                observed_iv = opt['iv']
                predicted_iv = self._skew_model(strike, params)
                weight = ai_weights.get('weights', {}).get(str(strike), 1.0)
                total_error += weight * (observed_iv - predicted_iv) ** 2
            return total_error
        
        # 初始猜测
        x0 = np.array([0.0, 0.5, 0.55])
        
        # 边界约束
        bounds = [
            (-1.0, 1.0),    # slope
            (0.0, 2.0),     # curvature  
            (0.2, 1.0)      # atm_vol
        ]
        
        result = minimize(
            objective, 
            x0, 
            method='L-BFGS-B',
            bounds=bounds,
            options={'maxiter': 500, 'disp': True}
        )
        
        return {
            "slope": result.x[0],
            "curvature": result.x[1],
            "atm_vol": result.x[2],
            "optimization_success": result.success,
            "final_error": result.fun
        }
    
    def _skew_model(self, strike: float, params: np.ndarray) -> float:
        """内部 skew 计算"""
        slope, curvature, atm_vol = params
        moneyness = np.log(strike / 67500.0)  # 使用实际标的价格
        return atm_vol + slope * moneyness + curvature * moneyness ** 2
    
    def _call_ai_model(self, prompt: str, model: str) -> str:
        """调用 HolySheep AI"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.1,
            "max_tokens": 1000
        }
        
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=30
        )
        
        if response.status_code != 200:
            raise Exception(f"AI 模型调用失败: {response.text}")
        
        return response.json()["choices"][0]["message"]["content"]
    
    def _validate_calibration(self, market_data: dict, 
                              params: dict) -> dict:
        """验证校准结果"""
        errors = []
        for opt in market_data['options']:
            predicted = self._skew_model(opt['strike'], 
                [params['slope'], params['curvature'], params['atm_vol']])
            error = abs(predicted - opt['iv'])
            errors.append({
                "strike": opt['strike'],
                "observed": opt['iv'],
                "predicted": predicted,
                "error_bps": error * 10000  # 转换为基点
            })
        
        return {
            "errors": errors,
            "max_error_bps": max(e['error_bps'] for e in errors),
            "avg_error_bps": np.mean([e['error_bps'] for e in errors])
        }

# ❌ 错误写法
headers = {
    "Authorization": "Bearer YOUR_HOLYSHEEP_API_KEY"  # 硬编码 Key
}

✅ 正确写法 - 从环境变量读取
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("请设置 HOLYSHEEP_API_KEY 环境变量")

headers = {
    "Authorization": f"Bearer {api_key}"
}

验证 Key 格式
if not api_key.startswith("sk-"):
    raise ValueError(f"API Key 格式错误: {api_key[:10]}***")

# ❌ 隐含波动率求解失败时的错误处理
def get_implied_volatility_buggy(self, params: OptionParams, market_price: float):
    sigma = 0.3
    for i in range(100):
        # 缺少边界检查和收敛判断
        sigma = self._newton_step(sigma, params, market_price)
    return sigma  # 可能返回完全错误的值

✅ 带完整保护的实现
def get_implied_volatility_safe(self, params: OptionParams, 
                                market_price: float) -> dict:
    sigma_bounds = (0.01, 5.0)
    sigma = 0.3
    history = []
    
    for i in range(100):
        calculated_price = self.black_scholes_price(params)
        vega = self._calculate_vega(params)
        
        if abs(vega) < 1e-10:
            return {"status": "vega_zero", "sigma": sigma, "iterations": i}
        
        diff = market_price - calculated_price
        if abs(diff) < 1e-8:
            return {"status": "converged", "sigma": sigma, "iterations": i}
        
        sigma_new = sigma + diff / vega
        sigma = max(sigma_bounds[0], min(sigma_new, sigma_bounds[1]))
        params.sigma = sigma
        history.append({"sigma": sigma, "error": abs(diff)})
        
        if len(history) > 5:
            # 检查是否在局部最优震荡
            recent_errors = [h["error"] for h in history[-5:]]
            if max(recent_errors) / min(recent_errors) < 1.01:
                return {"status": "local_optimum", "sigma": sigma, 
                       "iterations": i, "error": recent_errors[-1]}
    
    return {"status": "max_iterations", "sigma": sigma, 
            "iterations": 100, "error": history[-1]["error"] if history else None}

# ❌ 无约束的 SVI 参数直接使用
def svi_vol_buggy(k: float, a: float, b: float, rho: float, 
                  m: float, sigma: float, T: float) -> float:
    # SVI 公式无约束，某些参数组合会产生负方差
    inner = 0.5 * (b * sigma * (k - m) + 
                    np.sqrt((b * sigma * (k - m))**2 + 
                           4 * sigma**2 * (T + b * rho * (k - m) + a)))
    return np.sqrt(inner / T)

✅ 带完整约束检查的 SVI 实现
def validate_svi_params(params: SVIParameters) -> tuple[bool, str]:
    """验证 SVI 参数是否满足约束条件"""
    checks = []
    
    # 约束 1: b >= 0
    if params.b < 0:
        return False, f"b 必须 >= 0，当前值 {params.b}"
    
    # 约束 2: sigma > 0
    if params.sigma <= 0:
        return False, f"sigma 必须 > 0，当前值 {params.sigma}"
    
    # 约束 3: |rho| < 1
    if abs(params.rho) >= 1:
        return False, f"|rho| 必须 < 1，当前值 {params.rho}"
    
    # 约束 4: a + b * sigma * (1 - |rho|) >= 0
    lower_bound = params.a + params.b * params.sigma * (1 - abs(params.rho))
    if lower_bound < 0:
        return False, f"a + b * sigma * (1 - |rho|) = {lower_bound} < 0"
    
    return True, "参数验证通过"

def svi_vol_safe(k: float, params: SVIParameters, T: float) -> float:
    valid, msg = validate_svi_params(params)
    if not valid:
        raise ValueError(f"SVI 参数约束违反: {msg}")
    
    a, b, rho, m, sigma = params.a, params.b, params.rho, params.m, params.sigma
    
    inner = 0.5 * (b * sigma * (k - m) + 
                    np.sqrt(max(0, (b * sigma * (k - m))**2 + 
                              4 * sigma**2 * (T + b * rho * (k - m) + a))))
    
    vol_squared = inner / T if T > 0 else b * sigma
    return np.sqrt(max(0, vol_squared))