3. 数据源与数据采集

因子库的底层质量直接取决于数据源的广度、精度与采集链路的可靠性。本章系统梳理常见数据源类型、采集策略、API 对接规范以及数据质量校验与清洗流程，帮助你在构建因子库的第一步就建立坚实的数据基座。

3.1 常见数据源分类

量化数据源按属性可分为三大类：行情数据、财务数据与另类数据。下表对比了各类数据源的核心特征与典型用途。

3.2 数据采集策略

数据采集策略需根据数据源类型、更新频率与延迟要求进行设计。以下是三种主流采集模式：

• 全量快照 + 增量更新：适用于日频行情与财务数据。每日收盘后拉取全量快照，后续仅拉取增量变更，降低带宽与存储压力。
• 实时流式采集：适用于 Tick 级行情与高频因子。通过 WebSocket 或 TCP 协议订阅实时数据流，本地缓存并落盘。
• 定时批量拉取：适用于另类数据与低频指标。通过 Cron 任务定时调用 API，批量拉取后写入数据库。

3.3 API 对接规范

API 对接是数据采集的核心环节。以下是一个标准化的 RESTful API 对接流程示例：

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

def fetch_daily_quotes(api_key, symbol, start_date, end_date):
    """
    从数据供应商获取日频行情数据
    """
    url = "https://api.dataprovider.com/v1/quote/daily"
    params = {
        "symbol": symbol,
        "start": start_date,
        "end": end_date,
        "apikey": api_key,
        "format": "json"
    }
    response = requests.get(url, params=params, timeout=30)
    response.raise_for_status()
    data = response.json()
    df = pd.DataFrame(data["records"])
    df["date"] = pd.to_datetime(df["date"])
    df.set_index("date", inplace=True)
    return df

# 使用示例
df_quotes = fetch_daily_quotes(
    api_key="your_api_key_here",
    symbol="000300.SH",
    start_date="2024-01-01",
    end_date="2024-12-31"
)
print(df_quotes.head())

3.4 数据质量校验与清洗

数据质量是因子有效性的前提。以下是一套标准的数据质量校验流程：

1. 完整性校验：检查时间序列是否存在缺失日期、空值或断点。
2. 一致性校验：验证同一数据在不同字段或不同数据源之间是否逻辑自洽（如开盘价 ≤ 最高价 ≥ 最低价 ≤ 收盘价）。
3. 异常值检测：使用 Z-Score、IQR 或 MAD 方法识别价格突变、成交量异常等离群点。
4. 去重与对齐：去除重复记录，按统一时间戳对齐不同数据源。
5. 填充与插值：对缺失值采用前向填充、线性插值或行业均值填充。

def clean_quote_data(df):
    """
    清洗日频行情数据：去重、排序、异常值标记、缺失填充
    """
    # 去重
    df = df[~df.index.duplicated(keep='first')]
    # 排序
    df.sort_index(inplace=True)
    # 一致性校验：标记违反 OHLC 逻辑的记录
    mask = (df["high"] >= df["low"]) & \
           (df["high"] >= df["close"]) & \
           (df["low"] <= df["open"])
    df["quality_flag"] = mask.astype(int)
    # 异常值检测：成交量超过 5 倍滚动标准差
    df["volume_zscore"] = (df["volume"] - df["volume"].rolling(20).mean()) / \
                          df["volume"].rolling(20).std()
    df.loc[df["volume_zscore"].abs() > 5, "quality_flag"] = -1
    # 缺失值前向填充
    df.fillna(method='ffill', inplace=True)
    return df

df_clean = clean_quote_data(df_quotes)
print(df_clean["quality_flag"].value_counts())

3.5 数据采集与清洗流程总览

下图展示了从数据源到因子库的完整数据采集与清洗链路，帮助你建立全局视角。

3.6 数据质量监控看板

建立持续的数据质量监控机制，是因子库长期稳定运行的保障。下表列出了核心监控指标与告警阈值：