第28章 曲面构建的工程化：数据库设计、API服务、性能优化

说实话，很多人在做波动率曲面的时候，都卡在了一个地方——模型跑通了，但没法上线。我见过不少团队，Excel里画出来的曲面漂漂亮亮，一到生产环境就崩。为什么？因为工程化没做好。

这一章，我们就聊聊怎么把波动率曲面从「实验室产品」变成「生产级服务」。说白了，就是数据库怎么存、API怎么暴露、性能怎么压榨。嗯，这里面的坑，我踩过不少。

数据库设计：曲面数据怎么存？

曲面数据有个特点——它是三维的。行权价、到期时间、波动率，三个维度。你想想看，如果每天生成一个曲面，一年就是365个。每个曲面又有几十个行权价和十几个期限。数据量其实不小。

我个人习惯用这样的表结构：

-- 曲面元数据表
CREATE TABLE vol_surface_meta (
    id BIGSERIAL PRIMARY KEY,
    underlying VARCHAR(10) NOT NULL,      -- 标的代码
    surface_date DATE NOT NULL,           -- 曲面日期
    surface_type VARCHAR(20) NOT NULL,    -- 曲面类型: call/put
    model_type VARCHAR(30),               -- 模型类型: SVI/SSVI/SPLINE
    param_json JSONB,                     -- 模型参数（JSON格式）
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE(underlying, surface_date, surface_type)
);

-- 曲面网格数据表
CREATE TABLE vol_surface_grid (
    id BIGSERIAL PRIMARY KEY,
    surface_id BIGINT REFERENCES vol_surface_meta(id),
    strike DECIMAL(12,4) NOT NULL,        -- 行权价
    tenor VARCHAR(10) NOT NULL,           -- 期限: 1M/3M/6M/1Y
    implied_vol DECIMAL(8,4),             -- 隐含波动率
    moneyness DECIMAL(8,4),               -- 虚实值程度
    delta DECIMAL(8,4),                   -- Delta值
    UNIQUE(surface_id, strike, tenor)
);

为什么拆成两张表？我在项目中遇到过一个问题——如果全存成网格数据，查询某个期限的波动率时，要扫描大量行。拆开之后，元数据表存模型参数，网格表存离散点。查询时先定位元数据，再取网格数据，快很多。

API服务设计：怎么暴露曲面数据？

API设计这块，我建议遵循RESTful风格。但要注意，曲面查询的请求参数比较多，用GET请求时URL会很长。我个人习惯用POST + JSON body。

来看一个典型的API设计：

# FastAPI 示例
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, List
import pandas as pd
import numpy as np

app = FastAPI(title="Vol Surface API")

class SurfaceQuery(BaseModel):
    underlying: str
    surface_date: str
    surface_type: str = "call"
    tenors: Optional[List[str]] = None  # 指定期限，None表示全部
    strikes: Optional[List[float]] = None  # 指定行权价

class SurfaceResponse(BaseModel):
    underlying: str
    surface_date: str
    surface_type: str
    data: List[dict]
    model_params: Optional[dict] = None

@app.post("/api/v1/surface/query", response_model=SurfaceResponse)
async def query_surface(query: SurfaceQuery):
    """
    查询波动率曲面数据
    """
    # 1. 查询元数据
    meta = get_surface_meta(
        query.underlying, 
        query.surface_date, 
        query.surface_type
    )
    if not meta:
        raise HTTPException(status_code=404, detail="曲面不存在")
    
    # 2. 查询网格数据
    grid_data = get_surface_grid(
        meta['id'], 
        tenors=query.tenors,
        strikes=query.strikes
    )
    
    # 3. 如果请求了模型参数，返回参数
    model_params = None
    if query.include_params:
        model_params = meta['param_json']
    
    return SurfaceResponse(
        underlying=query.underlying,
        surface_date=query.surface_date,
        surface_type=query.surface_type,
        data=grid_data,
        model_params=model_params
    )

性能优化：曲面计算怎么提速？

曲面构建最耗时的环节是什么？插值计算。尤其是SVI模型，需要拟合5个参数。如果每天要处理几百个标的，计算量就上来了。

我总结了几条优化策略：

• 缓存策略：曲面数据一旦生成，当天基本不变。用Redis缓存，key设计为 vol:surface:{underlying}:{date}:{type}，TTL设到次日凌晨。
• 批量计算：不要一个标的接一个标的算。把同一天的所有标的打包，用numpy向量化计算。我在项目中试过，批量比循环快10倍以上。
• 预计算网格：如果曲面模型参数不变，网格点可以预计算。比如SVI参数确定后，所有行权价和期限的波动率可以一次性算完。
• 异步IO：数据库查询用异步连接池。FastAPI + asyncpg，连接数控制在CPU核心数的2倍左右。

来看一个批量计算的例子：

import numpy as np
from scipy.optimize import minimize
from concurrent.futures import ThreadPoolExecutor

def batch_fit_svi(surfaces_data: List[dict], max_workers=8):
    """
    批量拟合SVI模型
    """
    def fit_single(data):
        # 单个曲面的SVI拟合
        strikes = data['strikes']
        vols = data['vols']
        
        def svi_func(params, k):
            a, b, rho, m, sigma = params
            return a + b * (rho * (k - m) + np.sqrt((k - m)**2 + sigma**2))
        
        def objective(params):
            return np.sum((svi_func(params, strikes) - vols)**2)
        
        result = minimize(objective, [0.1, 0.1, 0.1, 0.0, 0.1], 
                         method='L-BFGS-B')
        return result.x
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        results = list(executor.map(fit_single, surfaces_data))
    
    return results

架构设计：整体流程

下面这张图，是我在实际项目中用的架构。你想想看，从数据采集到API暴露，中间经过了哪些环节：

这个架构看起来复杂，其实核心就三层：计算层、存储层、服务层。缓存夹在中间，用来扛高频查询。监控层兜底，出了问题能第一时间发现。

避坑指南

最后，分享几个我踩过的坑：

• 数据库连接泄漏：我曾经在批量计算时，每个线程都创建了新的数据库连接，但忘了关闭。结果跑了半小时，数据库连接数飙到2000+，直接把库搞挂了。解决方案：用连接池，并且设置合理的 max_connections。
• 缓存穿透：如果查询的曲面日期不存在，每次请求都会穿透到数据库。我后来加了个布隆过滤器，先判断日期是否存在，不存在直接返回空。
• 序列化性能：曲面网格数据转JSON时，如果数据量大，序列化时间可能超过计算时间。我改用MessagePack或者Protobuf，序列化速度快了3倍。
• 时间精度：波动率数据保留4位小数就够了。我之前保留6位，数据量大了30%，但精度对交易决策几乎没有影响。

嗯，工程化这块，说白了就是「把对的事情做对」。模型再漂亮，跑不起来也是白搭。数据库设计、API服务、性能优化，这三件事做好了，你的波动率曲面才能真正落地。