第三章 数据获取与清洗：期权链数据的预处理艺术

说实话，做量化交易这么多年，我最大的体会就是：数据质量决定了策略的生死。你策略再牛，模型再炫，喂进去一堆脏数据，结果就是垃圾进垃圾出。今天咱们就来聊聊期权链数据的获取与清洗，这是波动率套利的第一步，也是最容易被忽视的一步。

3.1 使用yfinance获取期权链数据

yfinance这个库，我用得最多。它免费、简单，而且数据质量还不错。不过要注意，它获取的是美式期权数据，对于欧式期权需要额外处理。

先看个基础示例：

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

# 获取标的资产
ticker = yf.Ticker("SPY")

# 获取所有到期日
expirations = ticker.options
print(f"可用到期日: {expirations[:5]}")  # 只看前5个

嗯，这里有个坑。yfinance返回的到期日列表是按字符串排序的，不是按日期排序。我刚开始做的时候没注意，结果数据顺序全乱了。建议你拿到后先转成datetime再排序：

# 排序到期日
exp_dates = sorted([datetime.strptime(d, "%Y-%m-%d") for d in expirations])
exp_dates_str = [d.strftime("%Y-%m-%d") for d in exp_dates]

获取单个到期日的期权链：

# 获取最近到期日的期权链
target_date = exp_dates_str[0]
opt_chain = ticker.option_chain(target_date)

# 分离看涨和看跌
calls = opt_chain.calls
puts = opt_chain.puts

print(f"看涨期权数量: {len(calls)}")
print(f"看跌期权数量: {len(puts)}")

你想想看，一个到期日就有几百个期权合约。如果我们要做全期限分析，数据量会非常大。所以我建议你一次性获取多个到期日的数据：

def get_option_chain(ticker_symbol, exp_dates):
    """批量获取期权链数据"""
    ticker = yf.Ticker(ticker_symbol)
    all_data = []
    
    for exp_date in exp_dates:
        try:
            chain = ticker.option_chain(exp_date)
            calls = chain.calls
            calls['option_type'] = 'call'
            calls['expiration'] = exp_date
            
            puts = chain.puts
            puts['option_type'] = 'put'
            puts['expiration'] = exp_date
            
            all_data.append(calls)
            all_data.append(puts)
        except Exception as e:
            print(f"获取 {exp_date} 数据失败: {e}")
            continue
    
    return pd.concat(all_data, ignore_index=True)

# 获取5个到期日的数据
target_dates = exp_dates_str[:5]
option_data = get_option_chain("SPY", target_dates)
print(f"总数据量: {len(option_data)}")

3.2 数据清洗与预处理

数据拿到手了，但别急着用。yfinance返回的数据里有很多字段，我们真正需要的其实就几个：行权价、最后交易价、隐含波动率、希腊值等。先看看原始数据长什么样：

# 查看数据列
print(option_data.columns.tolist())
# ['contractSymbol', 'strike', 'currency', 'lastPrice', 'change', 
#  'percentChange', 'volume', 'openInterest', 'bid', 'ask', 
#  'expiration', 'option_type', ...]

我个人习惯只保留核心字段，其他的删掉。为什么？因为数据越干净，后续处理越不容易出错。我曾经因为保留了多余的列，导致merge时出现重复数据，排查了半天。

def clean_option_data(df):
    """清洗期权数据"""
    # 选择核心字段
    core_cols = ['strike', 'lastPrice', 'bid', 'ask', 'volume', 
                 'openInterest', 'impliedVolatility', 'expiration', 
                 'option_type']
    
    df_clean = df[core_cols].copy()
    
    # 重命名列
    df_clean.columns = ['strike', 'last_price', 'bid', 'ask', 
                        'volume', 'open_interest', 'iv', 
                        'expiration', 'option_type']
    
    # 删除缺失值
    df_clean = df_clean.dropna(subset=['strike', 'last_price', 'iv'])
    
    # 过滤异常数据
    df_clean = df_clean[df_clean['strike'] > 0]
    df_clean = df_clean[df_clean['last_price'] > 0]
    df_clean = df_clean[df_clean['iv'] > 0]
    
    # 添加时间戳
    df_clean['fetch_time'] = datetime.now()
    
    return df_clean

cleaned_data = clean_option_data(option_data)
print(f"清洗后数据量: {len(cleaned_data)}")

这里有个关键点：隐含波动率（iv）。yfinance返回的iv是年化值，但有时候会出现异常值。我遇到过iv大于10的情况，这明显不合理。所以我会加一个过滤：

# 过滤异常IV值
df_clean = df_clean[(df_clean['iv'] > 0.05) & (df_clean['iv'] < 2.0)]
# 保留IV在5%到200%之间的数据

3.3 存储到本地数据库

数据清洗完了，接下来就是存储。我个人推荐用SQLite，轻量、无需安装、适合个人研究。当然，如果你做生产环境，可以考虑PostgreSQL。

先创建数据库和表结构：

import sqlite3
from sqlite3 import Error

def create_database(db_path="options_data.db"):
    """创建期权数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 创建期权数据表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS option_chain (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ticker TEXT NOT NULL,
            strike REAL NOT NULL,
            last_price REAL,
            bid REAL,
            ask REAL,
            volume INTEGER,
            open_interest INTEGER,
            iv REAL,
            expiration TEXT NOT NULL,
            option_type TEXT NOT NULL,
            fetch_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    # 创建索引，加速查询
    cursor.execute('''
        CREATE INDEX IF NOT EXISTS idx_ticker_exp 
        ON option_chain(ticker, expiration)
    ''')
    
    conn.commit()
    conn.close()
    print("数据库创建成功")

create_database()

然后写个批量插入函数：

def save_to_database(df, ticker, db_path="options_data.db"):
    """保存数据到数据库"""
    conn = sqlite3.connect(db_path)
    
    # 添加ticker列
    df['ticker'] = ticker
    
    # 选择要保存的列
    save_cols = ['ticker', 'strike', 'last_price', 'bid', 'ask', 
                 'volume', 'open_interest', 'iv', 'expiration', 
                 'option_type']
    
    df_to_save = df[save_cols].copy()
    
    # 批量插入
    df_to_save.to_sql('option_chain', conn, if_exists='append', index=False)
    
    conn.close()
    print(f"成功保存 {len(df_to_save)} 条数据")

# 执行保存
save_to_database(cleaned_data, "SPY")

最后，写个查询函数方便后续使用：

def query_option_data(ticker, expiration=None, option_type=None, db_path="options_data.db"):
    """查询期权数据"""
    conn = sqlite3.connect(db_path)
    
    query = "SELECT * FROM option_chain WHERE ticker = ?"
    params = [ticker]
    
    if expiration:
        query += " AND expiration = ?"
        params.append(expiration)
    
    if option_type:
        query += " AND option_type = ?"
        params.append(option_type)
    
    df = pd.read_sql_query(query, conn, params=params)
    conn.close()
    
    return df

# 示例查询
spy_calls = query_option_data("SPY", option_type="call")
print(f"SPY看涨期权数据: {len(spy_calls)} 条")

3.4 完整流程与SVG框架图

好了，整个流程串起来就是这样。我画了个流程图，帮你理清思路：

这个流程看起来简单，但每一步都有细节。我刚开始做的时候，光数据清洗就折腾了两周。后来总结出一套标准流程，现在每次获取数据都按这个来，基本没出过问题。

数据准备好了，下一步就是计算隐含波动率。不过那是下一章的事。今天先把数据基础打牢，后面才能走得稳。