A股量化策略数据源获取实战：从公开API到私有数据仓库的完整解决方案

1次阅读

共计 1875 个字符，预计需要花费 5 分钟才能阅读完成。

在 A 股量化策略开发中，数据获取是一个基础但极具挑战性的环节。以下是开发者常见的痛点：

数据延迟问题：交易所公开行情数据通常有 3 秒以上的延迟，对于高频策略影响显著
字段完整性不足：第三方数据商可能缺失关键字段（如逐笔委托数据中的订单来源标记）
清洗成本高：不同数据源对同一字段的命名和格式差异导致 30% 以上的开发时间消耗在数据清洗
合规风险：爬取非授权数据可能违反《数据安全法》第 21 条规定

数据源类型	延迟	成本	合规性	典型代表
交易所 API	3 秒 +	中	完全合规	上交所 Level-2
商业数据商	500ms	高	需授权	Wind/Choice
开源 API	10 秒 +	免费	需注意条款	Tushare/AkShare
私有爬虫	1 秒内	开发成本高	高风险	自建爬虫系统

实时数据层：采用 WebSocket 协议连接多数据源，实现行情数据低延迟获取
历史数据层：使用列式存储结构组织 T + 1 数据，支持快速回溯测试
质量监控层：通过数据指纹比对实现多源校验

import akshare as ak
from retrying import retry

def fetch_realtime_data(stock_code: str, max_retry: int = 3) -> pd.DataFrame:
    """
    获取单支股票实时行情（自动重试）:param stock_code: 6 位股票代码
    :param max_retry: 最大重试次数
    :return: 包含最新行情的数据框
    """
    @retry(stop_max_attempt_number=max_retry, wait_exponential_multiplier=1000)
    def _fetch():
        return ak.stock_zh_a_spot(symbol=stock_code)

    try:
        return _fetch()
    except Exception as e:
        print(f"获取 {stock_code} 数据失败: {str(e)}")
        return pd.DataFrame()

import polars as pl

def clean_minute_data(raw_df: pl.DataFrame) -> pl.DataFrame:
    """
    分钟级数据清洗流程
    :param raw_df: 原始数据 DataFrame
    :return: 标准化处理后的数据
    """
    return (
        raw_df
        .with_columns(pl.col("timestamp").str.strptime(pl.Datetime, format="%Y-%m-%d %H:%M:%S"),
            (pl.col("volume") / 100).alias("volume_hand")  # 转换为手数
        )
        .drop_nulls()
        .unique(subset=["timestamp", "code"])
    )

-- 创建分钟级行情表
CREATE TABLE minute_bars AS
SELECT 
    code,
    timestamp,
    first(price) as open,
    max(price) as high,
    min(price) as low,
    last(price) as close,
    sum(volume) as volume
FROM raw_ticks
GROUP BY code, date_trunc('minute', timestamp);

异步 IO 处理：使用 aiohttp 实现并发请求，吞吐量提升 5 倍
数据压缩存储：采用 Zstandard 压缩算法，存储空间减少 70%
缓存策略：实现 LRU 缓存机制，热点数据响应时间 <50ms

风险点	控制措施	参数阈值
请求频率	令牌桶算法限流	10 请求 / 秒 /IP
数据存储	敏感字段加密	AES-256 加密
用户协议	定期审查 API 条款变更	每月检查一次