5. 构建波动率曲面的数据准备：期权链数据获取、数据清洗、异常值处理

做波动率曲面，说白了就是跟数据打交道。我见过不少新手，上来就调模型、画曲面，结果曲面画出来歪歪扭扭的，一看就是数据有问题。嗯，数据准备这一步，其实占了整个曲面构建工作量的60%以上。你想想看，垃圾进垃圾出，这个道理在量化金融里尤其明显。

5.1 期权链数据获取：从哪里来，拿什么

我个人习惯把数据源分成三类：免费公开的、付费商用的、以及交易所直连的。对于做课程和入门研究，免费的就够了。但你要是做实盘交易，我建议还是用付费数据，稳定性和时效性差太多了。

常见的免费数据源包括：

  • Yahoo Finance：用 yfinance 库就能拉，但期权数据只支持近月合约，而且有时会缺行权价
  • Alpha Vantage：有免费额度，但期权数据更新频率低
  • 交易所官网：比如CBOE、上交所，直接下载CSV，但需要自己写解析器

我在项目中遇到过最头疼的事——用Yahoo拉数据，某天突然发现某个行权价的隐含波动率是负数。查了半天，原来是数据源把买一价和卖一价搞反了。所以，不管用哪个源，一定要做交叉验证。

核心原则： 至少用两个独立数据源做交叉比对。如果差异超过5%，直接标记为可疑数据。

下面是我常用的数据获取代码框架：

import yfinance as yf
import pandas as pd
from datetime import datetime, timedelta

def fetch_option_chain(ticker, expiration_date):
    """
    获取期权链数据
    ticker: 股票代码
    expiration_date: 到期日，格式 'YYYY-MM-DD'
    """
    stock = yf.Ticker(ticker)
    
    # 获取所有到期日
    expirations = stock.options
    if expiration_date not in expirations:
        raise ValueError(f"到期日 {expiration_date} 不在可用列表中")
    
    # 拉取看涨和看跌期权
    calls = stock.option_chain(expiration_date).calls
    puts = stock.option_chain(expiration_date).puts
    
    # 添加类型标记
    calls['option_type'] = 'call'
    puts['option_type'] = 'put'
    
    # 合并
    chain = pd.concat([calls, puts], ignore_index=True)
    chain['date'] = pd.Timestamp.today().strftime('%Y-%m-%d')
    chain['expiration'] = expiration_date
    
    return chain

# 示例：获取AAPL最近到期日的期权链
aapl_chain = fetch_option_chain('AAPL', '2024-12-20')
print(f"获取到 {len(aapl_chain)} 条期权记录")

5.2 数据清洗：把脏数据洗干净

数据拉下来之后，你会发现一堆问题。我总结了一下，常见的脏数据大概有这几类：

问题类型 具体表现 处理方式
缺失值 某些行权价的隐含波动率、成交量、持仓量为NaN 删除或插值填充
价格异常 买一价 > 卖一价，或者价格为零 直接删除该行
时间戳错乱 数据日期与当前日期不符 统一标准化为交易日期
重复记录 同一合约出现多次 保留最新一条

我曾经在清洗数据时发现，某只股票的期权链里居然有行权价为0.01的合约。你想想看，这明显是数据录入错误。遇到这种情况，我的做法是：先看成交量，如果成交量为零，直接删；如果有成交量，再查数据源。

小技巧： 清洗时保留原始数据的副本。我习惯在列名上加个 _raw 后缀，比如 strike_raw，这样出了问题还能追溯。

清洗代码示例：

def clean_option_chain(df):
    """
    清洗期权链数据
    """
    # 1. 删除完全缺失的行
    df = df.dropna(subset=['strike', 'lastPrice', 'impliedVolatility'], how='all')
    
    # 2. 删除价格异常：买一价大于卖一价
    mask_bid_ask = df['bid'] > df['ask']
    if mask_bid_ask.any():
        print(f"发现 {mask_bid_ask.sum()} 条买一价大于卖一价的记录，已删除")
        df = df[~mask_bid_ask]
    
    # 3. 删除隐含波动率为零或负数的记录
    mask_iv = (df['impliedVolatility'] <= 0) | (df['impliedVolatility'].isna())
    if mask_iv.any():
        print(f"发现 {mask_iv.sum()} 条IV异常记录，已删除")
        df = df[~mask_iv]
    
    # 4. 删除行权价明显异常（比如小于0.01或大于当前股价的10倍）
    current_price = df['underlyingPrice'].iloc[0] if 'underlyingPrice' in df.columns else None
    if current_price:
        mask_strike = (df['strike'] < 0.01) | (df['strike'] > current_price * 10)
        if mask_strike.any():
            print(f"发现 {mask_strike.sum()} 条行权价异常记录，已删除")
            df = df[~mask_strike]
    
    # 5. 去重：按合约代码保留最新一条
    if 'contractSymbol' in df.columns:
        df = df.sort_values('contractSymbol').drop_duplicates(subset='contractSymbol', keep='last')
    
    return df.reset_index(drop=True)

# 清洗刚才获取的数据
aapl_clean = clean_option_chain(aapl_chain)
print(f"清洗后剩余 {len(aapl_clean)} 条记录")

5.3 异常值处理：别让一个坏点毁了整个曲面

波动率曲面最怕什么？最怕一个异常点把整个插值结果带偏。我记得有一次做SPX的曲面，发现深度虚值看涨期权的IV突然飙到80%，而相邻行权价只有20%。后来一查，原来是那天有人挂了个错单，成交了一笔。

异常值检测，我一般用三种方法：

  1. 基于统计的方法：计算每个到期日下，同一行权价区间的IV均值和标准差，超过3倍标准差的标记为异常
  2. 基于相邻比较：检查相邻行权价的IV变化率，如果超过50%，大概率是异常
  3. 基于市场常识：比如深度实值期权的IV不应该比平值期权低太多，否则就是套利机会
注意： 不要机械地删除所有异常值。有些异常其实是市场真实的信号，比如财报前IV突然飙升。我的做法是：先标记，再人工复核。

异常值检测代码：

def detect_outliers_by_iv(df, z_threshold=3):
    """
    基于隐含波动率的Z-score检测异常值
    """
    # 按到期日和期权类型分组
    grouped = df.groupby(['expiration', 'option_type'])
    
    outlier_mask = pd.Series(False, index=df.index)
    
    for name, group in grouped:
        if len(group) < 5:  # 样本太少不检测
            continue
        
        mean_iv = group['impliedVolatility'].mean()
        std_iv = group['impliedVolatility'].std()
        
        if std_iv == 0:
            continue
        
        z_scores = (group['impliedVolatility'] - mean_iv) / std_iv
        group_outliers = z_scores.abs() > z_threshold
        
        outlier_mask.loc[group.index[group_outliers]] = True
    
    return outlier_mask

# 检测异常
outliers = detect_outliers_by_iv(aapl_clean)
print(f"检测到 {outliers.sum()} 个异常值")

# 查看异常值详情
if outliers.any():
    print(aapl_clean[outliers][['strike', 'option_type', 'impliedVolatility', 'lastPrice']])

5.4 数据标准化：让不同合约能放在一起比较

数据清洗干净后，还需要做标准化。为什么？因为不同到期日、不同行权价的合约，它们的价格和波动率不能直接比较。我们需要把它们映射到统一的坐标系里。

标准化的核心步骤：

  • 计算货币性（Moneyness）：行权价 / 标的资产价格，或者用对数货币性 ln(K/S)
  • 计算剩余期限：用年化表示，比如30天就是 30/365
  • 计算隐含波动率：用BS公式反推，注意要用正确的利率和股息率

我个人习惯用 moneyness = strike / spot 这个定义，简单直观。但做学术研究时，更常用 log_moneyness = ln(strike / spot)，因为它在数学上更对称。

def standardize_chain(df, spot_price, risk_free_rate=0.05, dividend_yield=0.0):
    """
    标准化期权链数据
    """
    from datetime import datetime
    import numpy as np
    
    # 计算货币性
    df['moneyness'] = df['strike'] / spot_price
    df['log_moneyness'] = np.log(df['strike'] / spot_price)
    
    # 计算剩余期限（年化）
    current_date = pd.Timestamp.today()
    df['expiration_date'] = pd.to_datetime(df['expiration'])
    df['ttm'] = (df['expiration_date'] - current_date).dt.days / 365.0
    
    # 过滤掉已经到期的合约
    df = df[df['ttm'] > 0]
    
    # 计算隐含波动率（如果数据源没有提供）
    # 这里假设数据源已经提供了 impliedVolatility 列
    # 如果没有，需要用BS公式反推
    
    return df

# 假设AAPL当前价格是190
aapl_standard = standardize_chain(aapl_clean, spot_price=190)
print(aapl_standard[['strike', 'moneyness', 'ttm', 'impliedVolatility']].head())

5.5 构建数据管道：自动化你的准备工作

手动一步步操作太慢了。我建议把整个流程封装成一个数据管道，每天定时跑一遍。这样你每天早上打开电脑，看到的都是清洗好的、标准化的数据。

下面是一个完整的管道示例：

def build_option_data_pipeline(ticker, expirations=None):
    """
    完整的期权数据准备管道
    """
    all_data = []
    
    # 如果没有指定到期日，获取所有可用到期日
    if expirations is None:
        stock = yf.Ticker(ticker)
        expirations = stock.options[:10]  # 取最近10个到期日
    
    for exp in expirations:
        try:
            # 1. 获取数据
            chain = fetch_option_chain(ticker, exp)
            
            # 2. 清洗
            chain = clean_option_chain(chain)
            
            # 3. 检测异常
            outliers = detect_outliers_by_iv(chain)
            chain['is_outlier'] = outliers
            
            # 4. 标准化
            spot = yf.Ticker(ticker).history(period='1d')['Close'].iloc[-1]
            chain = standardize_chain(chain, spot_price=spot)
            
            all_data.append(chain)
            print(f"成功处理到期日 {exp}")
            
        except Exception as e:
            print(f"处理到期日 {exp} 时出错: {e}")
            continue
    
    # 合并所有数据
    final_df = pd.concat(all_data, ignore_index=True)
    
    # 保存到本地
    final_df.to_parquet(f'{ticker}_option_data.parquet')
    
    return final_df

# 运行管道
result = build_option_data_pipeline('AAPL')
print(f"最终数据包含 {len(result)} 条记录，{result['expiration'].nunique()} 个到期日")
避坑指南： 我曾经在管道里忘记处理除息日，结果某天所有期权的IV都跳变了。后来我加了个日历，在除息日前一天自动调整股息率参数。这种细节，不做实盘你根本不会注意到。

5.6 本章小结

数据准备这件事，说白了就是「三分技术，七分经验」。技术上的代码你照着写就能跑，但那些坑——比如数据源突然改格式、某个交易所的期权代码规则变了——这些只能靠时间积累。

我给你的建议是：先跑通一个最小可行版本，然后每天盯着数据看。看多了，你自然就知道哪些数据是正常的，哪些是异常的。嗯，这就是所谓的「数据直觉」。

下一章，我们会用这些清洗好的数据，开始构建真正的波动率曲面。到时候你会发现，前面这些脏活累活，全都值了。

公众号:蓝海资料掘金营，微信deep3321
← 上一章 📖 返回目录 下一章 →