3、信息采集与清洗：API数据获取、Web Scraping基础、数据去重与标准化、时间戳对齐、缺失值处理实战

做事件交易，说白了就是跟数据打交道。你策略再牛，模型再炫，拿到的数据是脏的、乱的、对不齐的，那结果就是——垃圾进，垃圾出。我见过太多人花90%的时间调参数，却不愿意花10%的时间把数据洗干净。嗯，这其实是个大坑。

今天这一章，我就带你走一遍数据从采集到清洗的完整流程。全是实战经验，没有废话。

3.1 API数据获取：别当伸手党

我个人习惯，能用API绝不用手动下载。为什么？因为自动化是量化交易的第一条命。你想想看，如果每天开盘前你还在手动点鼠标下载CSV，那你还做什么量化？

常见的金融数据API有几种：

• RESTful API：最主流，通过HTTP请求获取JSON或CSV数据
• WebSocket：实时流数据，适合高频场景
• FTP/SFTP：老牌机构还在用，适合批量下载

这里我以国内某数据平台的REST API为例，展示一个标准的获取流程：

import requests
import pandas as pd
from datetime import datetime, timedelta

def fetch_kline_data(symbol, start_date, end_date, api_key):
    """
    获取K线数据
    """
    url = "https://api.example.com/v1/kline"
    headers = {
        "X-API-Key": api_key,
        "Content-Type": "application/json"
    }
    params = {
        "symbol": symbol,
        "interval": "1d",
        "start": start_date,
        "end": end_date,
        "limit": 1000
    }
    
    try:
        response = requests.get(url, headers=headers, params=params, timeout=10)
        response.raise_for_status()  # 检查HTTP状态码
        data = response.json()
        
        # 解析成DataFrame
        df = pd.DataFrame(data["data"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
        df.set_index("timestamp", inplace=True)
        return df
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None

# 使用示例
df = fetch_kline_data("BTCUSDT", "2024-01-01", "2024-01-31", "your_api_key_here")

3.2 Web Scraping基础：当API不给力时

有些数据源没有公开API，或者API限制太严格。这时候就得靠Web Scraping了。但我要提醒你：爬虫有风险，动手需谨慎。一定要遵守网站的robots.txt和用户协议。

我常用的工具是requests + BeautifulSoup，偶尔用Selenium处理动态加载的页面。来看一个简单的例子：

import requests
from bs4 import BeautifulSoup
import pandas as pd

def scrape_news_headlines(url):
    """
    爬取财经新闻标题
    """
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        soup = BeautifulSoup(response.text, "html.parser")
        
        # 假设新闻标题都在class为"news-title"的a标签里
        headlines = []
        for item in soup.select("a.news-title"):
            headlines.append({
                "title": item.get_text(strip=True),
                "link": item.get("href"),
                "timestamp": item.find("span", class_="time").get_text(strip=True)
            })
        
        return pd.DataFrame(headlines)
    except Exception as e:
        print(f"爬取失败: {e}")
        return None

3.3 数据去重与标准化：脏数据的克星

数据拿到手，第一件事不是分析，而是去重。为什么？因为很多数据源会重复推送，尤其是实时数据。你想想看，如果同一个K线数据被重复计算两次，你的回测结果会变成什么样？

去重其实很简单，核心就是基于唯一标识符去重。对于金融时间序列，唯一标识符通常是时间戳 + 标的代码的组合。

def deduplicate_data(df, key_columns=["timestamp", "symbol"]):
    """
    基于关键列去重，保留最后一条记录
    """
    before = len(df)
    df = df.drop_duplicates(subset=key_columns, keep="last")
    after = len(df)
    print(f"去重前: {before} 条, 去重后: {after} 条, 移除: {before - after} 条")
    return df

标准化呢？说白了就是把不同数据源的数据格式统一起来。比如：

• 日期格式统一为YYYY-MM-DD HH:MM:SS
• 价格统一为浮点数，去除千分位逗号
• 涨跌幅统一为小数形式（0.05 表示 5%）
• 成交量统一为股数或手数，不要混用

我习惯写一个标准化的函数，把所有字段都过一遍：

def standardize_data(df):
    """
    标准化数据格式
    """
    # 日期标准化
    if "timestamp" in df.columns:
        df["timestamp"] = pd.to_datetime(df["timestamp"])
    
    # 价格字段标准化（去除逗号，转浮点）
    price_cols = ["open", "high", "low", "close"]
    for col in price_cols:
        if col in df.columns:
            df[col] = df[col].astype(str).str.replace(",", "").astype(float)
    
    # 涨跌幅标准化（如果是百分比字符串，转小数）
    if "change_pct" in df.columns:
        if df["change_pct"].dtype == object:
            df["change_pct"] = df["change_pct"].str.replace("%", "").astype(float) / 100
    
    return df

3.4 时间戳对齐：让不同数据源说同一种语言

这是事件交易中最头疼的问题之一。不同数据源的时间戳可能不一样：

• A股数据：东八区，精确到秒
• 美股数据：美东时间，精确到毫秒
• 加密货币：UTC时间，精确到毫秒甚至微秒

如果你不做时间戳对齐，直接把数据拼在一起，那结果就是——驴唇不对马嘴。我见过有人把美股数据和A股数据直接合并，回测出来年化收益200%，实盘直接亏到姥姥家。

时间戳对齐的核心原则：统一时区，统一精度。

def align_timestamps(df, target_tz="Asia/Shanghai", freq="1min"):
    """
    时间戳对齐：统一时区 + 重采样到固定频率
    """
    # 确保时间戳是datetime类型
    if not pd.api.types.is_datetime64_any_dtype(df.index):
        df.index = pd.to_datetime(df.index)
    
    # 如果有时区信息，转换为目标时区
    if df.index.tz is not None:
        df.index = df.index.tz_convert(target_tz)
    else:
        df.index = df.index.tz_localize("UTC").tz_convert(target_tz)
    
    # 重采样到固定频率（向前填充）
    df = df.resample(freq).ffill()
    
    return df

3.5 缺失值处理实战：别让NaN毁了你的策略

数据采集过程中，缺失值几乎是不可避免的。网络断了、交易所维护、数据源抽风……各种原因都会导致数据缺失。那怎么处理呢？

我总结了三种常用方法：

方法	适用场景	优点	缺点
向前填充（ffill）	价格数据、成交量	简单、符合市场逻辑	可能掩盖真实波动
线性插值	指数数据、利率	平滑、合理	不适合跳跃性数据
删除缺失行	缺失比例很小（<1%）	干净、无偏差	可能丢失信息

来看一个实战例子：

def handle_missing_values(df, method="ffill", max_gap=5):
    """
    处理缺失值
    method: ffill, linear, drop
    max_gap: 最大连续缺失数，超过则删除
    """
    # 先统计缺失情况
    missing_count = df.isnull().sum()
    print(f"缺失值统计:\n{missing_count}")
    
    if method == "ffill":
        # 向前填充，但限制连续填充次数
        df = df.fillna(method="ffill", limit=max_gap)
    elif method == "linear":
        # 线性插值
        df = df.interpolate(method="linear", limit=max_gap)
    elif method == "drop":
        # 删除包含缺失值的行
        df = df.dropna()
    
    # 如果还有剩余缺失值，删除
    df = df.dropna()
    
    return df

3.6 本章知识体系总览

下面这张图，是我自己梳理的「数据采集与清洗」核心流程。你可以把它当作一个检查清单，每次处理数据时对照着走一遍：

嗯，这一章的内容就到这里。数据采集和清洗看起来琐碎，但它是整个事件交易体系的基石。你想想看，如果地基没打好，上面盖的房子再漂亮，也经不起风吹雨打。

我个人建议，每次拿到新数据源，先花30分钟做一遍完整的清洗流程。养成习惯后，你会发现后面的分析工作顺畅得多。别像我刚开始那样，数据脏了还硬着头皮往上堆模型——那真是自讨苦吃。

好了，去动手试试吧。把今天讲的代码跑一遍，用你自己的数据源。遇到问题很正常，多调试几次就熟了。

---