Python量化数据获取实测:免费股票数据API的坑,我替你们踩了一遍

做量化研究的头一年,我在数据上花的时间比写策略还多。一开始的想法很简单:网上那么多免费股票数据接口,yfinance、akshare、tushare、Alpha Vantage,哪个不能凑合着用?后来发现"凑合"两个字是量化里最贵的词——数据质量、字段口径、接口稳定性,每一个都是回测结果里看不见的变量。这篇把我用过的免费数据源逐个过一遍,说清楚它们各自的坑,以及最后我为什么换成了稳定的统一API。如果你是刚开始接触Python量化数据获取、正在纠结选哪个免费行情接口的开发者,这篇应该能帮你省下不少试错时间。

picture.image

先说结论:免费数据源不是不能用,是看你拿它干什么

折腾一圈下来我的体会是:免费源适合验证想法——写个脚本看个趋势、做个粗糙回测、画个K线图,完全够用。但一旦你的策略开始依赖这些数据做决策,或者要跑长期历史回测,免费源的每一个小毛病都会变成策略里说不清的偏差。免费股票数据API和稳定数据源之间,差的不是钱,是确定性。

逐个说:每个免费源的真实问题

yfinance:方便,但只方便在表层

Yahoo的接口封装,用起来确实方便,yf.download("AAPL") 一行就出数据,很多Python量化教程第一步就是它。优点不用说,说坑:

  • 数据字段缺:拆股、分红调整的历史数据口径跟别的源对不上,同一只股票两个源拉出来的日K,后复权价格能差好几个点,回测里一对比全是脏数据
  • 延迟和限流:频繁请求会被限,尤其盘中拉实时报价,经常拿到的是几分钟前的旧价,做实时行情监控基本不可用
  • 接口不稳定:上游页面结构一变,封装库就要跟着改,有段时间隔三差五拉不到数据,社区issue一片哀嚎

它适合:快速看数据、做演示、验证代码逻辑。不适合:任何要长期稳定跑的东西。

akshare:覆盖广,但"薛定谔的可用"

国产开源库,覆盖的接口特别全,股票、期货、基金、宏观数据都有,这一点确实良心。用久了你会发现:

  • 接口经常"悄悄失效":很多接口是从网页上抓的,上游网站一改版,接口就报错,而且没人通知你,只有跑批的时候才发现数据全是NaN
  • 文档和实际行为对不上:参数名、返回格式经常要自己试错,同一个接口不同版本行为还不一样
  • 速度不稳定:抓网页的接口,慢的时候一个请求要好几秒,批量拉数据能等到怀疑人生

它适合:研究阶段查个数据、找找灵感、做数据探查。不适合:当作生产数据管道的一部分,因为你不知道它哪天就断了。

tushare:规范,但积分制度磨人

老牌国产数据源,社区做得好,文档规范,数据质量在免费源里算上游。免费档的问题在于积分体系:

  • 很多接口有积分门槛:想拿全字段的日线、分钟线,积分不够就得等,或者靠贡献换积分,急用数据的时候只能干瞪眼
  • 频率限制:免费档每分钟调用次数有限,批量拉几百只股票的历史数据要分很多次跑,跑一次全市场要几个小时
  • 数据更新延迟:部分接口数据更新不是实时的,做日内策略会发现数据跟不上行情

它适合:有耐心攒积分、做日频以上研究的开发者。不适合:着急上线、需要分钟级数据的人。

Alpha Vantage免费档:额度算得清清楚楚

国外老牌API,免费档的限制最直接:每天500个请求。听着不少,但你算一笔账:回测里拉一遍标普500成分股的日K,几百个请求就没了,一天的额度直接见底。而且免费档的数据延迟、字段完整性也打了折扣,分钟级数据更是限量供应。

新浪/腾讯的免费行情接口

这是很多"野路子"教程里教的,直接拼URL拉行情。实测下来:

  • 非官方接口,字段不稳定:哪天页面结构改了,你的解析代码就废了,纯纯的黑盒依赖
  • 没有历史K线:基本只有实时报价,回测用不了,做不了任何历史研究
  • 合规风险:数据授权边界模糊,商用基本不用想,自己玩玩可以,别写进正式项目

"免费"的真实成本,算完这笔账你就懂了

把这些源都用过一遍,我的感受是:免费数据源的成本不在钱上,在时间上——

  • 维护成本:每个源都是"薛定谔的可用",你要定期写脚本检查数据源是不是还活着,字段是不是还一样,一套检查脚本写下来,工作量不比写策略小
  • 数据质量成本:免费源的数据错误、缺失、调整口径不一致,回测结果里根本看不出来,等你实盘了才暴露,那时候就不是改代码能解决的问题了
  • 机会成本:免费源的坑花掉的时间,本来可以花在策略本身、因子研究、风控设计上

量化里有个说法:垃圾数据进,垃圾策略出。回测跑得再漂亮,数据源不可靠,结果就是自欺欺人。Python量化数据获取这块,选择数据源其实是在选择你的研究可信度。

什么时候必须换稳定数据源

我自己的切换时机是这么判断的,三个信号任中一个就换:

  1. 要跑超过一年的历史回测——数据口径必须统一、完整,免费源的缺失和口径差异会直接污染结论,回测结果再好也是假的
  2. 要接实时行情做监控或盘中策略——免费源的延迟和限流完全扛不住,盘中行情API的要求是毫秒级稳定推送,不是偶尔刷出来一条
  3. 策略进入实盘/半实盘阶段——这个不用多说,数据稳定是底线,实盘里数据断一分钟都是事故

这时候我换成了 iTick 的统一API——一个接口覆盖股票、指数、期货、外汇多个品种,REST和WebSocket都有,token去它官网 itick.org 申请就能拿。下面用股票数据举个实际接入的例子,代码直接能跑。

一个实际的接入例子:历史K线 + 实时报价 + 盘中推送

稳定数据源的用法其实很简单,就是规规矩矩调接口,接口文档怎么写的就怎么调。拉历史日K:

import requests

API_BASE = "https://api.itick.org"
TOKEN = "your_token_here"
headers = {"accept": "application/json", "token": TOKEN}

def get_stock_kline(code="AAPL", region="US", ktype=8, limit=500):
    resp = requests.get(
        f"{API_BASE}/stock/kline",
        headers=headers,
        params={
            "region": region,   # 市场代码:US美股、HK港股、CN A股
            "code": code,       # 股票代码
            "kType": ktype,     # 1=1分钟 2=5分钟 5=1小时 8=日K 9=周K 10=月K
            "limit": limit      # 单次最大500条
        }
    )
    data = resp.json()["data"]
    # 每根K线字段:t时间戳(毫秒)/o开/h高/l低/c收/v量/tu额
    return data

daily = get_stock_kline()
print(daily[0])

实时报价走REST快照,一个请求拿当前全部关键字段:

def get_quote(code="AAPL", region="US"):
    resp = requests.get(
        f"{API_BASE}/stock/quote",
        headers=headers,
        params={"region": region, "code": code}
    )
    q = resp.json()["data"]
    # 字段说明:
    # ld最新价 o开盘 p前收 h最高 l最低
    # t时间戳(毫秒) v成交量 tu成交额
    # ts交易状态(0正常/1停牌/2退市/3熔断) ch涨跌额 chp涨跌幅%
    return q

print(get_quote())

要做盘中实时监控,再开一条WebSocket连接订阅推送:

import json
import threading
import time
import websocket

WS_URL = "wss://api.itick.org/stock"
authenticated = False

def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    # 等鉴权成功再订阅
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "AAPL$US",   # 格式:代码$市场,多标的用逗号分隔
            "types": "quote"       # quote报价/tick逐笔/depth盘口/kline@1分钟K线
        }))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        return

    if payload.get("resAc") == "pong":
        return

    quote = payload.get("data") or {}
    if quote.get("type") == "quote":
        print(quote["s"], quote["ld"], quote["t"])

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(
    WS_URL,
    header=[f"token: {TOKEN}"],
    on_message=on_message
)
ws.run_forever()

数据获取的实践分层:回测、快照、实时各管各的

用稳定源之后,我把数据获取拆成三层,各管各的:

  • 回测层:历史K线一次性批量拉下来存本地(SQLite或Parquet),回测只读本地,不碰线上接口
  • 快照层:每天收盘后或定时拉一次报价快照,做日频监控和汇总
  • 实时层:盘中用WebSocket订阅自己关注的标的,事件驱动更新,不做轮询

这样分层的好处是:回测永远不被网络影响,实时监控不被限流影响,每一层的失败都是独立的,出问题好排查。

踩坑清单速查

  • 免费源的接口随时可能失效,生产代码里不要硬依赖,至少加个数据新鲜度检查
  • 多个数据源的数据口径不一致(复权、字段定义),回测前先统一口径,别混着用
  • 免费档的限流规则要提前算好,别等批量任务跑到一半被限
  • 时间戳单位:这边是毫秒,time.time() 是秒,减之前先换算,不然延迟全是负数
  • 交易状态要判断:ts 字段 0正常/1停牌/2退市/3熔断,别把停牌的旧价当实时价写进策略
  • K线增量更新别全量刷:实时推送场景每来一根新K线就重拉全量,浪费且慢

小结

数据获取这条路上,"免费"两个字最贵。我的建议很朴素:研究阶段随便折腾免费源,但凡是策略要认真跑起来,数据这块别省。分层设计 + 稳定数据源 + 统一口径,这套组合拳打完,你会发现以前花在数据上的时间,全部还给了策略本身。接口参数和返回字段的具体说明,写代码时对照官方文档最靠谱。

GitHub:https://github.com/itick-org/

0
0
0
0
评论
未登录
暂无评论