ARTICLE DETAIL

资讯详情

深耕商务建站与企业官网运营的一线实战洞察。

A股行情数据获取实战:实时与历史API选型、复权处理及量化回测落地指南

A股行情数据获取实战:实时与历史API选型、复权处理及量化回测落地指南 先把话放在前面做量化、做数据分析或者只是想把自己关注的一篮子A股做成自动盯盘小工具的朋友十有八九都会在行情数据获取这一步卡住。我在做A股实时行情和历史行情抓取的时候最开始也是到处翻网页、找接口、看论坛帖子后来才慢慢理清头条路实时行情靠什么API、历史K线去哪拉、分钟数据怎么落地、回测和初筛怎么用。这篇就把我折腾出来的整套方案和踩过的坑一起写出来给正在跟行情API较劲的你做个参考。我不打算只贴几个代码片段就完事因为这类接口最大的问题从来不是“能不能调”而是“能不能稳定调、调完能不能直接用”。比如数据缺字段、复权口径不对、分钟线缺失、盘中数据和收盘数据不一致这些小问题放到回测里都会变成大事故。下面我会从方案选型、环境准备、实时接口、历史接口到常见问题和实战场景整条链路拆开讲一遍。1. 项目概述与方案选型先想清楚你要什么数据1.1 核心需求拆解实时行情和历史行情其实是两套事很多人一开始说“我要A股行情API”但真正落地的需求往往是分层的。我习惯把它拆成四类数据类型典型用途更新频率典型来源实时快照盯盘、自选股监控、盘中预警秒级到分钟级新浪/腾讯公开接口、Tushare快照分时/逐笔数据盘口分析、短周期回测秒级数据商付费接口为主历史日线选股、策略回测、业绩统计每日收盘后更新Tushare、Baostock、AKShare历史分钟线日内策略、精细化回测长时间跨度难拿Tushare Pro、AKShare、Baostock这个拆分很重要。因为实时行情和历史行情的获取难度、接口形态、数据成本完全不一样。实时行情讲究的是低延迟和持续轮询历史行情讲究的是覆盖度、复权准确性和存储方式。如果你用抓实时行情的思路去拉历史数据或者反过来用拉历史数据的方式做实时盯盘都会很别扭。1.2 数据源选型免费接口、开放平台和公开HTTP接口怎么挑我实际用过并且现在还在用的主要有四个方向各有各的脾气。这里直接给结论性的对比数据源是否免费Token/认证历史日线分钟线稳定程度Tushare Pro部分免费需要Token和积分很全需要高积分高Baostock免费无需Token全5、15、30、60分钟高但接口风格老AKShare免费开源无需Token全1、5、15、30、60分钟中上游页面一变就挂新浪/腾讯公开HTTP接口免费无不提供长期历史不提供适合实时快照如果你只是想快速验证一个策略或者做学习Demo我建议先用AKShare一行pip install就能拿到大量数据代码写起来也顺手。如果你打算把数据落地、长期维护、跑正经回测那Tushare Pro和Baostock会更稳。实时盘中监控我目前的主力是新浪和腾讯的公开HTTP接口响应速度快字段也够用。这里有一个我特别想强调的判断不要为了省事去网上找那些“打包好的历史数据下载包”。我踩过这个坑——网上流传的所谓A股历史分钟数据打包下载下来之后要么字段缺失要么复权口径说不清楚最要命的是你根本不知道它最后更新时间是哪一天。自己拉数据落库虽然前期麻烦但后续的稳定性和可追溯性完全值得。1.3 选型背后的“为什么”为什么不是只推荐一个数据源因为没有任何一个免费渠道能同时包办实时和历史两类需求。Tushare的分钟线和财务数据很全但它的实时快照能力一般而且部分接口需要积分AkShare覆盖广但它的本质是把公开页面结构化上游接口一旦调整你的脚本可能第二天就报错Baostock胜在接口稳定但只覆盖K线和基础数据衍生指标基本都要自己算。所以我的长期方案是实时快照走新浪/腾讯公开HTTP接口历史日线和分钟线走Tushare和Baostock双备份AKShare作为临时补数工具。不是哪个最好而是每个渠道都有自己的强项组合起来才最稳。这种多数据源冗余的做法在真实项目里能帮你省掉很多半夜跑批时间。2. 环境准备与实时行情API的接入细节2.1 环境准备和基础依赖安装无论你是Windows、macOS还是Linux先把Python环境弄干净。我推荐用Python 3.8以上单独的虚拟环境避免把系统Python搞乱。python -m venv market_env source market_env/bin/activate # Windows下用 market_env\Scripts\activate pip install pandas requests akshare tushare baostock有些接口返回的数据是GBK编码如果你在Windows下跑建议顺便把编码相关的习惯养成拿到字节先判断编码再解码成字符串。下面所有示例我都默认在UTF-8环境下运行但如果遇到中文乱码优先检查requests返回的response.encoding。2.2 通过新浪公开HTTP接口获取实时行情快照先来一个最直接的方案新浪的行情快照接口。它的URL格式很简单把股票代码拆成市场和代码两部分拼接。规则我整理了一下沪市6开头、科创板688开头用sh深市0和3开头、创业板300开头用sz指数类用对应的sh000001、sz399001这种方式。import requests def get_sina_quote(codes): code_str ,.join(codes) url fhttps://hq.sinajs.cn/list{code_str} headers { Referer: https://finance.sina.com.cn, User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) } resp requests.get(url, headersheaders, timeout5) resp.encoding gbk return resp.text lines get_sina_quote([sh600000, sz000001, sz399001]) print(lines)这里有两个细节必须提醒第一新浪接口直接裸请求会返回403需要在请求头里带上正常的浏览器Referer否则服务端会拒绝响应。第二返回内容默认是GBK编码如果你不指定resp.encoding gbk长字符串里拆出来的第一个字段“股票名称”十有八九是乱码。2.3 腾讯公开接口作为备选方案新浪接口偶尔会出现空值或者断连这时我一般切到腾讯的行情接口作为备选。它的地址是https://qt.gtimg.cn/q代码参数格式保持一致只是返回字段顺序跟新浪不太一样。import requests def get_tencent_quote(codes): code_str ,.join(codes) url fhttps://qt.gtimg.cn/q{code_str} resp requests.get(url, timeout5) resp.encoding gbk return resp.text腾讯接口返回的字段比新浪更丰富除了最新价、涨跌幅、成交量之外还有买卖五档、成交笔数、市净率这些。我自己的经验是新浪和腾讯都会出现偶发超时比较好的做法是写一个简单的故障切换先请求新浪失败或返回空就用腾讯间隔500毫秒重试一次。不要在同一秒内疯狂请求同一个接口公开接口不是专门给你家程序用的频率太高很容易被临时封禁。2.4 用Tushare获取当日交易数据如果你已经注册了Tushare Pro也可以用它拿当日交易数据。Tushare的优势是数据结构规范字段都有官方说明不会像新浪腾讯那种用逗号分隔的字符串还得自己数下标。下面是拉取某一交易日全市场日线数据的示例import tushare as ts ts.set_token(你的token) pro ts.pro_api() df pro.daily(trade_date20240115) print(df.head())这个接口返回的字段包含ts_code、trade_date、open、high、low、close、pre_close、change、pct_chg、volume、amount字段命名规范清晰适合做量化入库。需要注意这里的volume单位是手amount单位是千元如果后面要自己算成交额记得先做单位换算。Tushare的积分体系决定了你能调用哪些接口普通注册用户拉基础日线没问题但分钟线、财务数据这类高级接口需要更多积分。2.5 实时轮询的频率控制和多标的数据组装实时行情没有免费推送基本靠轮询。轮询频率怎么定我实践下来的一个平衡点自选股数量在30只以内每5秒轮询一次没问题股票数量超过50只建议放宽到10秒以上。频繁轮询不只是浪费带宽更关键的是容易被服务端限流导致拿回来的数据出现整段缺失。import time import pandas as pd codes [sh600000, sz000001, sz300750, sh688981] all_rows [] while True: raw get_sina_quote(codes) for line in raw.strip().split(\n): parts line.split()[1].strip().split(,) if len(parts) 32: continue all_rows.append({ code: line.split()[0].replace(hq_str_, ), name: parts[0], open: float(parts[1]), pre_close: float(parts[2]), price: float(parts[3]), high: float(parts[4]), low: float(parts[5]), volume: float(parts[8]), amount: float(parts[9]), timestamp: time.strftime(%Y-%m-%d %H:%M:%S) }) df pd.DataFrame(all_rows) print(df) time.sleep(5)这里有一个单位细节新浪返回的volume单位是股amount单位是元跟Tushare的手和千元不同。你如果是多数据源混用一定要在入库前统一单位我建议统一成“股”和“元”因为这是大部分策略库默认的数据口径。3. 历史行情API的获取、清洗与存储3.1 日线历史数据三条路径历史日线是选股和回测的基础。我先给三个数据源的调用示例直接可跑。Tushare版本import tushare as ts pro ts.pro_api() df pro.daily(ts_code600000.SH, start_date20200101, end_date20231231) df df.sort_values(trade_date).reset_index(dropTrue) print(df.head())Baostock版本import baostock as bs import pandas as pd bs.login() rs bs.query_history_k_data_plus( sh.600000, date,code,open,high,low,close,volume,amount, start_date2020-01-01, end_date2023-12-31, frequencyd, adjustflag2 ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) bs.logout() df pd.DataFrame(rows, columnsrs.fields) print(df.head())AKShare版本import akshare as ak df ak.stock_zh_a_hist( symbol600000, perioddaily, start_date20200101, end_date20231231, adjustqfq ) print(df.head())三种方式里Tushare的ts_code要用“600000.SH”这种带交易所后缀的格式Baostock要用“sh.600000”格式AKShare只要6位数字代码就行。我最早就是在这上面反复报错其实不是代码逻辑问题纯粹是代码格式不同。3.2 分钟级数据怎么拉才划算分钟级数据的获取要更谨慎因为数据量会随着时间跨度快速膨胀。拿一只股票的一年5分钟数据来算每天48根K线4小时交易时间每小时12根一年大约240个交易日就是11520根几百只股票就是百万级以上的记录。先看Tushare的分钟线接口需要足够的积分import tushare as ts pro ts.pro_api() df ts.pro_bar(ts_code600000.SH, freq5min, start_date20240101, end_date20240131) print(df.head())Baostock的分钟线相对亲民支持5、15、30、60分钟但我提醒一句它没有1分钟数据想做高频策略的话需要另找渠道。AKShare的分钟接口用的是东方财富的公开接口可以拿1分钟数据import akshare as ak df ak.stock_zh_a_hist_min_em( symbol600000, start_date2024-01-01 09:30:00, end_date2024-01-31 15:00:00, period5, adjust ) print(df.head())分钟数据的拉取我强烈建议分批按月份循环每次拿一个月中间加time.sleep(1)避免连续高频请求。尤其是AKShare这种底层依赖公开页面的工具打得太快会被上游限制甚至导致接口短期失效。3.3 复权处理回测数据必须过的一道关复权这个话题值得单独说因为很多新手直接在原始K线上跑回测结果跟真实收益差了十万八千里。先解释什么是复权上市公司分红送股之后股价会变低比如一只100块的股票每股分红5块除息日股价直接少了5块。表现在K线图上就是一个向下的跳空缺口。如果不复权这个跳空会被均线、MACD、OBV这些指标误判成“暴跌”但实际你的账户价值并没有变。复权分为前复权和后复权。前复权是把历史价格统一调整到当前价格水平图形直观但历史价格会随着每次除权除息不断变化后复权是以最早价格为基准向后调整当前价格会显示成一个很大的数字不太直观但优点是历史数据一旦生成就不变适合长期回测对比。我自己的习惯是原始不复权数据必须留一份存档回测时按需计算前复权或后复权千万不要把复权数据覆盖到原始库里不然后面想重新核对都无从下手。三个数据源的复权参数也不一致Tushare的ts.pro_bar有adj参数qfq是前复权hfq是后复权Baostock的adjustflag中1是后复权2是前复权3是不复权AKShare的adjust参数里qfq和hfq分别是前复权和后复权。这些参数我建议每个人都自己验证一次拿一只分红大户的股票分别拉不复权和前复权对比除权日的跳空有没有消失比看文档印象更深刻。3.4 数据增量更新与本地落库历史数据不是拉一次就完事A股每个交易日都会产生新数据所以增量更新能力才是数据链路里的核心竞争力。我推荐的存储方式是SQLite轻量、单文件、方便备份对行情数据这种结构化的低并发数据完全够用。下面是我常用的建表语句CREATE TABLE IF NOT EXISTS daily_kline ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, pre_close REAL, volume REAL, amount REAL, PRIMARY KEY (ts_code, trade_date) );增量更新的逻辑很简单先查询本地表里某只股票的最大trade_date然后从那个日期的下一天开始拉数据最后用INSERT OR REPLACE写回表里。这样即使某天重复执行也不会产生重复数据主键能天然过滤掉已存在的记录。import sqlite3 conn sqlite3.connect(market.db) cur conn.cursor() sql SELECT MAX(trade_date) FROM daily_kline WHERE ts_code ? cur.execute(sql, (600000.SH,)) last_date cur.fetchone()[0] print(本地最新日期:, last_date)这个思路同样可以扩展到分钟表分钟表的唯一键改成(ts_code, trade_time)就行。增量更新看起来简单但它能保证你以后任何时候补数据都是可重复的这种规范化的数据管理习惯最后都会体现在策略开发的效率上。4. 常见报错和数据问题的排查实录4.1 权限类报错Token、积分和接口权限如果你在用Tushare最常遇到的是“抱歉您没有访问该接口的权限”或者“积分不足”。这类问题不用慌先去官网看接口文档对应的积分要求。基础日线一般注册就有权限但分钟线、高频财务指标这些需要更高积分。解决方法没什么捷径主账号积累积分或完成认证但实践中有个更聪明的办法把高频需求拆到AKShare或Baostock上去完成别为了一个接口在一棵树上吊死。4.2 频率限制明明好好的突然返回空或报错频率限制是所有免费API都会遇到的隐性墙。我的经验是每次拉完数据固定sleep(1)批量下载时把任务拆到多个时间段执行。曾经我写过一个循环脚本一次性下载几百只股票的历史数据结果跑到第30只就开始被限流报错信息还不是直接说限流而是返回空DataFrame特别误导人。后来我把循环改成每下载10只股票就暂停30秒再也没出现过这种问题。4.3 代码格式和返回字段变化接口切换时第一坑就是股票代码前缀前面也提到了。第二种坑是字段顺序。新浪行情接口返回40多个字段中间可能有空字符串如果你直接按下标取指数和个股的字段长度还不一样很容易错位。我建议拿到原始文本后先看len(parts)按长度做分支处理或者只取自己需要的几个固定下标然后在循环里加一个try/except解析失败就打印原始行方便定位。4.4 停牌、新股和缺失数据历史数据里最烦人的不是代码写错而是股票本身状态导致的缺数据。停牌股票的区间自然没有K线新股上市之前也没有历史数据还有退市股的代码后面会加挂牌日期后缀比如600002.SH这种格式可能最后变成带日期的样子。处理这类问题的核心思路是接口返回多少就存多少不要自己强行补零填充。回测时再根据策略需要决定是否填充比如用NaN标记停牌让回测引擎自行跳过这比硬补价格更符合交易事实。4.5 常见问题速查表为了方便你快速定位我把遇到过的典型现象整理成一张表现象可能原因处理方法新浪接口返回403缺少Referer请求头带上https://finance.sina.com.cn的Referer返回中文乱码接口是GBK编码resp.encoding gbkTushare提示无权限积分不足查看接口积分门槛或换数据源AKShare返回空DataFrame上游接口变动或限流加sleep重试检查代码格式分钟数据缺第一根部分源不含9:31第一分钟按交易时段规则自行对齐回测收益明显偏高用了未来函数或未复权检查指标计算是否用了当日future数据4.6 盘中数据和收盘数据不一致这个话题我必须放在这里提醒盘中拿到的实时快照和收盘后通过历史接口拿到的日线数据经常不是完全一致的。原因很简单历史日线里的收盘价是最终确认价而盘中接口会包含临时成交、尾盘集合竞价阶段的变动部分券商还会在收盘后做异常交易处理。做监控可以但如果你把盘中收到的最后一笔价格直接当成收盘价入库第二天回测可能对不上账。我的做法是盘中数据单独存到一张“盘中快照”表收盘后统一用历史日线接口覆盖当天的正式数据两边互不干扰。5. 行情数据的实战应用与扩展方向5.1 把行情数据喂给backtrader做回测拿到干净的历史K线之后最常见的使用场景就是量化回测。Backtrader虽然用得人少了但它的架构清晰适合个人投研尤其是配合PandasData导入非常顺手。喂数据之前先确认DataFrame的字段和顺序import pandas as pd import backtrader as bt df pd.read_sql_query( SELECT trade_date, open, high, low, close, volume FROM daily_kline WHERE ts_code600000.SH ORDER BY trade_date, conn ) df[trade_date] pd.to_datetime(df[trade_date]) df.set_index(trade_date, inplaceTrue) class PandasData(bt.feeds.PandasData): params ( (open, open), (high, high), (low, low), (close, close), (volume, volume), ) data PandasData(datanamedf) cerebro bt.Cerebro() cerebro.adddata(data)这里有两个容易踩的细节。第一DataFrame的索引必须是DatetimeIndex而且时间顺序要升序不然Backtrader会认为数据没有时间轴。第二Backtrader的PandasData默认字段名是open、high、low、close、volume、openinterest如果你的字段叫trade_date、vol这些需要通过params显式映射。多股回测时数据对齐是最麻烦的。我的经验是先把所有股票的交易日并集拿出来然后对每只股票做reindex缺失的日期填NaN这样Backtrader的adddata加入多只股票后系统内部的日期对齐才不会错乱。5.2 用实时行情做异动股初筛很多人会在各个论坛找所谓“妖股选股公式源码”其实拆开看本质就是一组量价异动条件的组合。我用行情API自己搭过一个初筛脚本用OBV和量比做过滤逻辑很清楚这里分享出来供你参考。OBV能量潮的核心思想是把成交量按涨跌方向累计上涨日的成交量算正贡献下跌日算负贡献等于把所有量能粘合在一条线上。代码实现很短import numpy as np def calc_obv(close, volume): direction np.sign(close.diff()).fillna(0) return (volume * direction).cumsum()配合实时快照做异动初筛的思路是当日涨幅在2%到9%之间既保证活跃又排除一字涨停板量比大于2说明当日成交量显著超过过去5日均量OBV创出近60日新高说明量能正在持续推动趋势换手率在5%到20%之间太高可能已经过热每次盘中轮询时跑一遍这个规则命中的标的单独存一个观察列表。这里必须重复强调一句这只是数据筛选规则不是荐股建议所有规则能否赚钱要用历史数据严格回测之后才能知道。5.3 构建自己的分钟数据仓库最近总有人问那种“A股历史分钟数据打包下载”靠不靠谱。我的意见很直接如果是研究历史行情与其花时间找别人打包好的旧数据不如自己维护一份干净的数据仓库。分钟数据的建设和日线类似只是表换成五分钟或一分钟K线主键换成(ts_code, trade_time)。批量拉取时我的脚本框架是这样import time code_list [600000, 000001, 300750] month_list [202401, 202402, 202403] for code in code_list: for month in month_list: df ak.stock_zh_a_hist_min_em( symbolcode, start_datef{month}-01 09:30:00, end_datef{month}-31 15:00:00, period5, adjustqfq ) save_to_sqlite(df, code) time.sleep(1)整个仓库建设最耗时间的不是写代码而是等待数据一点点落库。如果你需要很大的时间跨度建议每周跑一次增量别一次性全部拉取。我自己第一次建一分钟数据仓库的时候一次性拉了几十只股票半年的数据跑了快一个晚上才完成中途断网一次还要重新校验断点。另外如果你需要复权后的分钟数据尽量用接口的adjust参数直接生成不要自己手动在前复权日线上插值。分钟级别的复权计算涉及跳空处理自己做容易产生边界误差。最后再分享一个小经验无论你选哪个数据源都要给自己留一条“原始数据”的底仓。我在实际操作中体会最深的一点就是数据清洗和处理可以反复优化但原始不复权数据一旦被污染后面所有策略结果都变得不可信。所以数据入库时我永远分两张表一张原始表只存接口原始返回一张计算表专门放复权后的数据。这样即使中途发现复权逻辑写错了也能重新算不用重新拉一遍历史数据。搞A股行情数据这件事表面上是API调用真正拉开差距的其实是数据管理和调试的细致程度。希望这篇文章能让你少走几段弯路。
返回列表
PREV
查看更多资讯
NEXT
返回资讯列表