这一篇是“股票数据API”系列的第10篇我想认认真真聊聊股票最新分时交易数据。K线数据在很多时候被当成刚需但实际做盘中看板、涨停预警、大单异动监控这类需求时真正依赖的反而是分时数据。K线少一根可以下次请求补回来分时数据要是断了一分钟这一分钟就永远找不回来了。我之前用网页抓取的方式做过一版分时采集每天开盘半小时就会因为频率太高被限制后来转用东财股票数据API提供的公共行情接口才把采集频率和稳定性都控制住了。这篇文章会把接口定位、URL参数、返回解析、轮询策略和完整代码一次讲清楚适合想自建轻量分时行情模块的同学直接抄作业。1. 先明确口径分时交易数据拆开来看到底包含哪几个维度1.1 分时数据和1分钟K线有什么本质区别先说一个很容易混淆的地方很多人以为“分时数据”就是“1分钟K线”其实两者的数据口径并不完全一样。分时线更强调“某一分钟结束时那一刻的成交状态”通常包含四个核心维度时间、最新成交价、当日累计均价、分钟累计成交量。以我实际解析过的接口返回为例一条典型的分时数据长这样2025-01-02 09:30,1522.00,1522.00,12345,187654321,-2.00,-0.13,0逐项拆开就是时间、最新价、均价、成交量、成交额、涨跌额、涨跌幅、其它状态位。你看它天然就带了一个“均价”字段这在1分钟K线里是没有的。1分钟K线描述的是“这一分钟内的开高低收”分时描述的是“到这一分钟为止市场累积到了一个什么位置”。这俩谁都能互相替代吗不能。我做分时看板时优先用的是分时接口因为它少算一步均价、少一次K线合成而且响应里的成交量是逐分钟累计值拿来做盘中资金曲线非常直接。反过来做回测策略、算指标形态时我还是会切回标准的分钟K线因为指标计算需要完整的高低价区间。1.2 分时数据在真实业务里的四种常见用法分时数据在开发落地时最常见的四种用途我列一下方便你判断自己属于哪类需求盘中实时监控页面上每60秒刷新一次“最新价、涨跌幅、成交量和均价”这是最轻量也最刚需的场景。触发预警比如价格突破当日均价线、成交量突然放大到过去5分钟均量的数倍这些规则都依赖分时数据的连续序列。盘中回放复盘收盘后把当日每一分钟的数据重新画出来重点看放量节点和均价线支撑这个场景对“时间连续性”要求很高。分钟级策略信号有些短线策略不是用K线而是用分时均价、累计成交量来算拐点这同样需要拿到字段完整的分时数据。明白了这些用法你才会理解为什么后面要花那么大力气做解析稳定性和缓存而不是拿到JSON直接塞给前端就完事。2. 东财接口探查记录URL、参数与secid规则的完整拆解2.1 域名与接口路径的坑push2his和push2的区别东财行情体系里有两类接口我经常搞混实际踩过坑之后才彻底记住push2处理实时的Tick、盘口和最新报价push2his处理历史序列分时数据这个场景要的是“从开盘到现在的全序列”所以理论上应该走历史域名的接口。我实践的完整请求URL是https://push2his.eastmoney.com/api/qt/stock/trends2/get注意有些资料里会把域名写成push2.eastmoney.com那样也能返回数据但响应结构和数据完整度可能会有差异。我建议直接固定用push2his因为分时接口需要后端拼接一整天的时间序列放在历史域名下更合理实测也更稳定。还有一个细节请求头里的Referer最好带上我一般设置成https://quote.eastmoney.com/虽然公开接口不强制校验但遇到限流策略收紧时带Referer和User-Agent的请求被放行的概率明显更高。这个属于“平时感觉没用、真被限制时才知道有用”的保命配置。2.2 secid映射规则和参数表东财接口不认识纯股票代码它只认识secid格式是“市场ID.股票代码”。沪深两市的规则并不复杂沪市市场ID为1例如贵州茅台是1.600519。深市市场ID为0例如平安银行是0.000001。按股票代码首位判断有个常见经验以6、9开头的多半是沪市以0、2、3开头的多半是深市。北交所的代码规则不太一样需要用的时候不要套用这个简单判断先拿真实代码做一次验证再说。完整请求参数我整理成了下面这个表参数说明实测推荐值secid市场ID加股票代码1.600519 或 0.000001ndays拉取最近多少天的分时1表示当天最多建议5iscr是否包含分笔校验数据0即可fields1基础行情字段集合f1,f2,...,f13fields2分时序列字段集合f51,f52,f53,f54,f55,f56,f57,f58fields1我一般直接写成f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13fields2写成f51,f52,f53,f54,f55,f56,f57,f58ndays这个参数值得单独说一句。只做当天盘中监控时老老实实传ndays1返回的序列就是从当天9:30到当前时间。如果传了5接口会返回最近5个交易日的完整分时序列数据量大很多不适合高频轮询。有些同学图省事一直传5结果每分钟多拉了几倍流量其实没必要。2.3 响应结构长什么样请求返回的JSON整体结构大致如下{ rc: 0, rt: 4, svr: s_push2_his_x.eastmoney.com, data: { code: 600519, market: 1, name: 贵州茅台, prePrice: 1523.0, trends: [ 2025-01-02 09:30,1522.00,1522.00,12345,187654321,-2.00,-0.13,0, 2025-01-02 09:31,1523.00,1522.50,4567,68912000,-1.00,-0.07,0 ] } }我拿到的响应里data.trends是一个字符串数组每个元素用逗号分隔。但也有人反馈同接口不同时段会返回对象数组。我的建议是解析时先判断trends[0]是字符串还是字典写一个兼容分支这样不管接口怎么微调都不至于全线崩溃。prePrice是昨日收盘价非常重要。分时里像涨跌额、涨跌幅这类字段如果接口没给准你可以自己拿最新价 - prePrice算一遍起到交叉校验的作用。3. 原始返回的解析细节时间、数字和成交量单位一个都不能错3.1 字符串分割的正确姿势拿到trends数组后第一个动作不是split(,)就完事而是先考虑字段长度和不可信列。我按fields2的声明顺序约定解析结果对应如下索引字段类型说明0time字符串格式为“YYYY-MM-DD HH:mm”1price浮点数该分钟最新成交价2avgPrice浮点数当日均价3volume整数分钟累计成交量单位是手4amount浮点数分钟累计成交额单位是元5change浮点数涨跌额6pct浮点数涨跌幅百分比7extra浮点数/整数不同市场含义不同谨慎使用实际代码可以这样写def parse_trend_line(line: str) - dict: parts line.split(,) if len(parts) 7: raise ValueError(f分时数据字段不足: {line}) return { time: parts[0], price: float(parts[1]), avg_price: float(parts[2]), volume: int(float(parts[3])), amount: float(parts[4]), change: float(parts[5]), pct: float(parts[6]), extra: parts[7] if len(parts) 7 else None, }为什么成交量要用int(float(...))而不是直接int(...)因为有些行情源的数值会带小数点比如12345.0直接int会报错。先用float转换再取整兼容性最好。3.2 时间解析和数字精度问题时间的标准格式是2025-01-02 09:30我建议统一转成datetime对象方便排序和按时间窗口切片from datetime import datetime def parse_time(s: str) - datetime: return datetime.strptime(s, %Y-%m-%d %H:%M)这里有个小坑接口返回的时间是东八区本地时间不包含时区信息。如果你的服务器部署在海外或容器时区不是Asia/Shanghai一定要在系统层面固定时区或者解析后手动tzinfo替换。否则你按时间排序时可能出现“最后一根数据排到中间”的诡异问题。另一个精度问题是价格。A股普通股票价格是两位小数但某些基金、债券可能到三位甚至四位小数。解析时统一用float没问题但展示给用户前不要做无谓的四舍五入尽量保留原始位数。我见过有同事为了“显示统一”把价格全部round(2)结果债券分时全对不上账。3.3 成交量单位一手等于100股分时接口里的volume单位默认是“手”不是“股”。沪深A股市场一手等于100股。做资金流分析时很多人会直接拿volume × price估算成交额算出来和amount对不上原因就在这里——你少乘了一个100。我自己的习惯是做展示时直接用“手”作单位简单直观。做金额换算时用amount字段不自己拿 volume 去乘。确实需要算股数时再手动乘100。这个细节虽然小但影响面很大尤其是盘口预警场景里如果单位没统一触发阈值可能整体偏移两个数量级。4. 稳定运行的核心轮询频率、重试机制、增量缓存与停牌判断4.1 轮询频率到底设置多少秒比较合适分时数据的精度是每分钟一条理论上你每隔60秒拉一次就能拿到全部最新数据。但实际看板为了“刷新更顺滑”很多人会压缩到30秒甚至20秒请求一次。我从稳定性的角度给个参考场景推荐间隔说明单只股票展示30-60秒分时精度足够不会有明显延迟多股票批量轮询60秒降低单IP请求密度预警任务20-30秒可选如果对延迟要求高但要做好限流准备盘后回放不轮询收盘后一次性拉全量即可如果只是做一个自选股列表看板二三十只股票轮询60秒间隔完全够用。低于15秒其实意义不大因为行情源的分钟数据本身也是按分钟维度聚合的你拉得再频繁拿到的还是同一分钟数据。4.2 增量缓存别把全天数据反复拉很多自建行情模块最大的问题是“每次请求都把当天全量序列拉一遍”。早盘时还好到下午两三点一天的分时数据已经有两百多条你再把整个数组反复解析、反复渲染纯属浪费。我的做法是分两层内存层保存当天最后一次拉到的完整序列后续轮询只覆盖更新。存储层盘中把{date: 2025-01-02, trends: [...]}整体缓存起来收盘后用定时任务归档到数据库。对于“只想要最新一根分时”的场景可以本地记录上一次最后一条数据的时间然后new_lines [x for x in trends if x.split(,)[0] last_time]只处理新增的部分旧数据不重复解析。这个优化对CPU和内存都非常友好尤其是你要同时监控几十只股票时。4.3 停牌、开盘前、午间休市怎么判断分时接口在非交易时段并不是报错而是返回不同形态的数据如果你不预判很容易把“正常空数据”当成“接口故障”。我总结过几类情况开盘前trends可能为空或者只有9:30之前的集合竞价数据。午间休市11:30之后、13:00之前接口返回的数据停更在11:30。停牌股票data可能为null或者data.trends为空数组。涨跌停一字板分时序列存在但价格不变成交量很小这是正常现象不要当成解析错误。节假日接口可能返回前一交易日的残留数据建议用当天日期和服务端返回的code校验一次。实际代码里可以加一个简单守卫if not data or not data.get(trends): return [] if not trends: return [] if len(trends) 1 and trends[0].split(,)[0].startswith(2025): # 只有一根数据可能是开盘前或刚开盘 return parse_trend_line(trends[0])5. 一个可以直接拿去用的分时数据采集器Python实现5.1 核心采集类下面这个类我在自己的看板项目里改过几版逻辑很直白但该有的重试、解析、错误保护都有直接复制到项目里就能跑import time import requests from datetime import datetime, timedelta class MinuteTrendFetcher: def __init__(self, timeout10, retries3, sleep_interval0.5): self.timeout timeout self.retries retries self.sleep_interval sleep_interval self.session requests.Session() self.session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64), Referer: https://quote.eastmoney.com/, }) def _build_url(self, secid: str, ndays: int 1) - str: fields1 f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13 fields2 f51,f52,f53,f54,f55,f56,f57,f58 return ( https://push2his.eastmoney.com/api/qt/stock/trends2/get f?secid{secid}ndays{ndays}iscr0 ffields1{fields1}fields2{fields2} ) def fetch(self, secid: str, ndays: int 1) - dict: url self._build_url(secid, ndays) last_exc None for attempt in range(self.retries): try: resp self.session.get(url, timeoutself.timeout) resp.raise_for_status() payload resp.json() if payload.get(rc) ! 0: raise ValueError(f接口返回rc{payload.get(rc)}) data payload.get(data) if not data: return {secid: secid, name: , pre_price: None, trends: []} raw_lines data.get(trends) or [] trends [] for line in raw_lines: parsed self._parse_line(line) if parsed: trends.append(parsed) return { secid: secid, code: data.get(code), name: data.get(name), pre_price: data.get(prePrice), trends: trends, } except Exception as exc: last_exc exc time.sleep(self.sleep_interval * (attempt 1)) raise RuntimeError(f分时数据请求失败: {secid}, {last_exc}) def _parse_line(self, line): if isinstance(line, dict): return { time: line.get(time), price: line.get(price), avg_price: line.get(avgPrice), volume: line.get(volume), amount: line.get(amount), } parts str(line).split(,) if len(parts) 5: return None return { time: parts[0], price: float(parts[1]), avg_price: float(parts[2]), volume: int(float(parts[3])), amount: float(parts[4]), }_build_url里没有把secid做神秘转换因为调用方最好自己掌握规则。5.2 股票代码自动转secid我从项目里抽了一个小函数出来按沪深规则自动拼装def to_secid(symbol: str) - str: symbol symbol.strip() if symbol.startswith((6, 9)): return f1.{symbol} if symbol.startswith((0, 2, 3)): return f0.{symbol} # 北交所和自定义证券代码无法通过首位判断需要自行映射 raise ValueError(f无法自动识别secid: {symbol})这里有个现实问题A股代码首位规则已经挺成熟但科创板688开头在沪市、创业板的300在深市规则都没问题。可一旦遇到北交所83、87、920开头的证券0或1的简单判断就未必可靠。多市场组合时我建议单独维护一张代码到市场的映射表而不是依赖首字母规则。5.3 最小可运行示例如果你只想快速验证一下下面这段代码可以直接跑fetcher MinuteTrendFetcher() result fetcher.fetch(1.600519, ndays1) print(result[name], result[pre_price]) print(result[trends][-1])输出会是类似这样的最新一根分时贵州茅台 1523.0 {time: 2025-01-02 14:59, price: 1522.0, avg_price: 1521.5, volume: 12000, amount: 182400000.0}拿到这个结果后你要展示、写库还是触发预警就都是自己的业务逻辑了。6. 我在生产环境里踩过的坑和最终使用的checklist6.1 坑一用“总成交量”替代“分钟成交量”一开始我把每条记录里的volume当成“到当前时刻的累计成交量”结果画出来的分时成交量图越画越高越看越奇怪。后来和数据源核对才发现接口返回的volume是每分钟新增成交量。累计值需要自己cumsum才算得出来。这不算接口的错但涉及一个认知问题你在做当日量能分析时一定先明确“是从零开始的累计量”还是“每分钟增量”否则展示端会出很尴尬的图表错误。6.2 坑二盘中请求用本地时间判断是否收盘有段时间我在服务器上用本地时间做盘后归档结果发现每天归档时间不固定数据有时多一分钟有时少一分钟。后来才发现服务器时区被改成 UTC 了。分时数据全部是东八区时间但轮询调度用的是服务器本地时间两者没对齐。现在我的统一处理是调度时间用Asia/Shanghai明确指定。数据解析后再加回东八区时区标记。所有日志也统一打印东八区时间避免排查时换算。from zoneinfo import ZoneInfo sh ZoneInfo(Asia/Shanghai) now datetime.now(sh)6.3 坑三多只股票并发时没有做限流启动瞬间被封我最初写批量采集时用线程池同时拉50只股票效果确实快但启动那一瞬间等于50个请求同时打出去很快触发了限流返回的rc不再等于0。后来改成“分批限速”每10只一组组内并行组间间隔1秒整体稳定很多。你也可以简单粗暴地加一个全局互斥锁让每只股票间隔0.2秒以上import threading _lock threading.Lock() def throttled_fetch(fetcher, secid): with _lock: result fetcher.fetch(secid) time.sleep(0.2) return result6.4 我最终使用的checklist把项目跑顺之后我给自己沉淀了一份上线检查清单每次新接入股票或新部署环境都会过一遍确认secid前是否自动判断正确尤其是北交所和特殊代码。确认服务器时区为东八区或用zoneinfo强制指定。确认成交量单位按“手”展示按“股”计算时手动乘100。盘中轮询固定30-60秒不在同一秒并发大量请求。停牌、午间、开盘前三种空数据场景全部做了单独分支。每次请求只更新增量数据不在盘中重复解析全量历史。把rc ! 0、连续请求失败、全天无新增数据三种异常都打上监控日志。最后分享一个我自己的小习惯盘中轮询的日志不要每60秒打一条太吵了。我只打“首次请求成功”和“出现异常”两类日志正常状态用静默计数。这样出问题时不会被海量正常日志淹没排查效率高出一大截。分时数据采集这个事技术难度不大真正拉长战线的永远是这些细枝末节的稳定性处理。对“股票数据API”系列来说把这块做扎实后面的实时监控、盘中预警才好继续往下搭。