20. 多资产曲面：跨资产类别的统一异常检测框架

说实话，做了这么多年波动率曲面，我最大的感触就是——单一资产的异常检测其实不难。难的是什么呢？是你手里握着股票期权、外汇期权、利率掉期、商品期货，每个市场都有自己的脾气，每个曲面都有自己的扭曲方式。你总不能给每个资产单独写一套检测逻辑吧？那维护成本得炸。

所以这一章，我们来聊聊怎么搭建一个跨资产类别的统一异常检测框架。说白了，就是一套代码，能同时管住股票、外汇、利率、商品这四个大家伙的曲面质量。

为什么需要统一框架？

我2018年在一家自营交易公司干过，当时团队里每个PM都有自己的曲面清洗脚本。有的用Z-score，有的用MAD，有的干脆肉眼扫。结果呢？同一个交易日，不同资产类别的异常标记标准完全不一致。风控那边根本没法做汇总报告。

统一框架的好处很明显：

  • 一致性：所有资产用同一套数学逻辑，避免人为偏差
  • 可维护性：改一个参数，所有资产都生效
  • 可扩展性：新加一个资产类别，只需要注册它的曲面结构
  • 审计友好：风控和合规能看懂你的检测逻辑

核心思路：把异常检测拆成三个层次——点级别（单个报价异常）、曲线级别（整条期限结构异常）、曲面级别（整个波动率面异常）。每个层次用统一的接口，但内部参数可以按资产类别微调。

框架设计：抽象基类 + 资产注册器

我个人习惯用Python的抽象基类（ABC）来定义接口。这样每个资产类别只需要实现几个核心方法，剩下的通用逻辑全部由基类处理。

from abc import ABC, abstractmethod
import numpy as np
import pandas as pd
from scipy.stats import zscore

class VolSurfaceDetector(ABC):
    """波动率曲面异常检测基类"""
    
    def __init__(self, asset_name, z_threshold=3.0, window=20):
        self.asset_name = asset_name
        self.z_threshold = z_threshold
        self.window = window
        self.history = []  # 存储历史曲面快照
        
    @abstractmethod
    def extract_features(self, surface_df):
        """从原始曲面数据中提取检测特征"""
        pass
    
    @abstractmethod
    def get_expected_shape(self, surface_df):
        """返回该资产类别的理论曲面形状（用于形状偏离检测）"""
        pass
    
    def detect_point_outliers(self, surface_df):
        """点级别异常：单个隐含波动率报价异常"""
        features = self.extract_features(surface_df)
        # 使用滚动Z-score检测
        z_scores = np.abs(zscore(features, axis=None))
        outliers = np.where(z_scores > self.z_threshold)
        return outliers
    
    def detect_curve_outliers(self, surface_df):
        """曲线级别异常：整条期限结构或整条执行价切片异常"""
        # 计算相邻期限/执行价之间的变化率
        delta = surface_df.diff(axis=1).abs()
        # 如果某列整体偏离历史均值超过3个标准差，标记
        col_means = delta.mean(axis=0)
        col_stds = delta.std(axis=0)
        abnormal_cols = np.where(
            (col_means - col_means.rolling(self.window).mean()) > 
            3 * col_stds.rolling(self.window).mean()
        )[0]
        return abnormal_cols
    
    def detect_surface_outliers(self, surface_df):
        """曲面级别异常：整个曲面形态异常"""
        # 保存当前曲面到历史
        self.history.append(surface_df.values.flatten())
        if len(self.history) < self.window:
            return False
        
        # 计算当前曲面与历史平均曲面的马氏距离
        hist_array = np.array(self.history[-self.window:])
        mean_surface = hist_array.mean(axis=0)
        cov_surface = np.cov(hist_array.T)
        
        try:
            inv_cov = np.linalg.inv(cov_surface + 1e-6 * np.eye(cov_surface.shape[0]))
            current_flat = surface_df.values.flatten()
            diff = current_flat - mean_surface
            mahalanobis = np.sqrt(diff @ inv_cov @ diff)
            # 卡方分布阈值，自由度=曲面点数
            threshold = np.sqrt(np.percentile(
                np.random.chisquare(len(current_flat), 10000), 95
            ))
            return mahalanobis > threshold
        except np.linalg.LinAlgError:
            # 协方差矩阵奇异时，回退到简单方法
            return False

避坑指南：我曾经在协方差矩阵求逆上栽过跟头。当曲面点数超过历史窗口长度时，协方差矩阵一定是奇异的。所以一定要加正则化项（上面代码里的1e-6 * I），或者用伪逆。

资产注册器：统一管理所有资产

有了基类，接下来就是注册具体的资产类别。我设计了一个简单的注册器模式：

class SurfaceDetectorRegistry:
    """资产检测器注册器"""
    
    _detectors = {}
    
    @classmethod
    def register(cls, asset_type):
        """装饰器：注册资产检测器"""
        def wrapper(detector_class):
            cls._detectors[asset_type] = detector_class
            return detector_class
        return wrapper
    
    @classmethod
    def get_detector(cls, asset_type, **kwargs):
        """获取对应资产的检测器实例"""
        if asset_type not in cls._detectors:
            raise ValueError(f"不支持的资产类型: {asset_type}")
        return cls._detectors[asset_type](**kwargs)

# 注册股票期权检测器
@SurfaceDetectorRegistry.register('equity')
class EquityVolDetector(VolSurfaceDetector):
    def extract_features(self, surface_df):
        # 股票期权：关注ATM附近的skew和term structure
        atm_idx = surface_df.columns.get_loc(0.5)  # 假设执行价归一化
        return surface_df.iloc[:, atm_idx-2:atm_idx+3].values
    
    def get_expected_shape(self, surface_df):
        # 股票期权预期：微笑形状，短期波动率更高
        return None  # 实际项目中用历史平均

# 注册外汇期权检测器
@SurfaceDetectorRegistry.register('fx')
class FXVolDetector(VolSurfaceDetector):
    def extract_features(self, surface_df):
        # 外汇期权：关注RR和BF组合
        # 实际项目中需要从曲面中提取25-delta RR和BF
        return surface_df.values
    
    def get_expected_shape(self, surface_df):
        # 外汇期权预期：偏斜形状，取决于利率差
        return None

嗯，这里要注意：不同资产类别的特征提取逻辑完全不同。股票期权你盯着ATM附近的微笑形状，外汇期权你得看风险逆转（RR）和蝶式（BF）组合，利率掉期则要关注远期波动率的期限结构。所以extract_features方法必须由子类实现，不能偷懒。

统一检测流程：三步走

实际跑检测的时候，我一般按这个流程走：

  1. 数据清洗：去掉明显的错误报价（负波动率、极端值等）
  2. 分层检测：先跑点级别，再跑曲线级别，最后跑曲面级别
  3. 综合评分：三个层次的异常分数加权汇总，生成最终的风险标签

为什么要分层？你想想看，如果一个报价本身是错的（点级别异常），那它所在的曲线和曲面肯定也被污染了。如果先跑曲面级别检测，可能会把整个曲面标记为异常，但实际上只是那个点的问题。分层检测能帮你精准定位问题源头。

重要提醒：分层检测的顺序不能乱。必须先修点，再修线，最后看面。我曾经在项目中先跑了曲面检测，结果发现整个曲面异常，排查了半天才发现是某个执行价上的一个报价多了一个零。先修点的话，5分钟就能定位。

实战案例：跨资产异常检测对比

为了让你更直观地理解，我拿2023年3月的一个实际交易日数据做了个对比。当时硅谷银行事件导致市场剧烈波动，我们来看看不同资产类别的异常表现：

资产类别 点级别异常数 曲线级别异常 曲面级别异常 综合风险评分
标普500期权 12 3条期限结构 是 0.87（高风险）
欧元/美元期权 5 1条执行价切片 否 0.42（中风险）
美国10年期利率掉期 8 2条远期曲线 是 0.73（高风险）
WTI原油期权 3 无 否 0.21（低风险）

你看，同样是市场剧烈波动，股票和利率市场受到的冲击最大，外汇和商品相对温和。统一框架的好处就是——你一眼就能看出哪个资产类别需要优先处理。

框架的核心逻辑图

下面这张SVG图展示了整个统一检测框架的结构。我画的时候特意把三个检测层次放在中间，左右两边分别是资产注册器和综合评分模块。这样你一眼就能看清数据流向。

跨资产统一异常检测框架 股票期权曲面 外汇期权曲面 利率掉期曲面 商品期货曲面 资产注册器 EquityVolDetector FXVolDetector IRVolDetector CommodityVolDetector 统一接口 extract_features() get_expected_shape() 三层异常检测 点级别检测 Z-score / MAD / IQR 曲线级别检测 期限结构 / 执行价切片 曲面级别检测 马氏距离 / 形状偏离 综合输出 异常点坐标 | 异常曲线索引 | 曲面异常标志 | 综合风险评分 统一格式：{asset_name, timestamp, anomaly_type, severity, details}

实际部署时的一些坑

框架搭好了，但真正上线跑的时候，有几个地方特别容易出问题：

  • 数据频率不一致：股票期权可能每秒钟更新一次，利率掉期可能几分钟才动一次。统一框架里要加一个时间戳对齐模块，不然点级别检测会误报。
  • 参数敏感性：Z-score的阈值设3还是4，对不同资产影响很大。我建议每个资产类别单独做一次回测校准，找到最优参数。
  • 历史窗口长度：window设太短，检测不稳定；设太长，对市场突变反应迟钝。我个人习惯用自适应窗口——市场波动率低时用长窗口，波动率高时用短窗口。

一个小技巧：在注册器里加一个validate_params()方法，每次实例化检测器时自动检查参数是否合理。比如z_threshold不能小于2，window不能小于10。这能避免很多低级错误。

好了，这一章的内容就到这里。统一异常检测框架的核心思想就是抽象接口 + 分层检测 + 统一输出。你只要把基类写好，剩下的就是往注册器里加资产类别。代码量不大，但维护成本能降一个数量级。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →