标题写到第10篇总算轮到分时数据这个大家问得最多的场景了。股票数据API里日线和分钟线都相对好理解分时交易数据却总有人问我能不能直接用接口把当天盘中那条曲线拉下来答案是可以而且东财股票数据API这条通道目前是免费又够稳的。这篇我就把从请求参数、字段解析到盘中刷新的全过程拆开写代码直接可跑适合正在做看盘工具、分时策略回放、交易大屏或者行情监控的开发者参考。1. 分时交易数据到底是在取什么1.1 先从业务视角理解分时看盘软件里那条横跨整屏的曲线通常叫分时线实际上就是当天从开盘到当前时刻每一分钟的最新成交价连起来。A股一天的连续竞价时段是上午09:30到11:30、下午13:00到15:00按60秒一个点计算正常情况下一天大约240个数据点遇到开盘集合竞价、盘中临时停牌再复牌等场景这个数量会变多。所以我在接数据的时候从不把“今天就是240条”写死而是以接口实际返回为准。这条曲线跟K线最大的不同在于K线是切片把每分钟的开盘、收盘、最高、最低明确告诉你分时曲线是连续状态每一分钟都在反映当时市场的实时强弱。很多做短线的用户盯分时不是为了看某一分钟涨了多少而是看这条线整体怎么走、会不会跌破均价、盘口承接力度如何。因此接口返回的字段里除了时间、价格还要带上累计成交量和累计成交额这样才有办法算均价线。1.2 接口返回的常见字段我以自己日常用的东财行情接口为例返回的trends数组里每一条都是逗号分隔的字符串字段顺序与请求时的fields2一一对应。实际返回大概长这样2025-05-12 09:30:00,1705.00,1706.00,1706.98,1702.00,862,146900,1704.56 2025-05-12 09:31:00,1708.66,1706.00,1708.66,1704.50,1245,212300,1705.28按我在项目中验证过的顺序分别是时间、当时价格、当日开盘、最高、最低、累计成交量、累计成交额、当日均价。这里有两个单位要特别注意成交量单位是手1手等于100股成交额单位是元。如果要把量画成底部柱状图建议除以10000用万手显示否则数字太大看图不直观。有个习惯建议保留不管从哪份文档看到字段顺序都要先print前三条原始数据确认一次。数据源偶尔会把字段顺序调整一旦你解析写错位置后面所有价格、均线、涨跌幅都是错的而且错得很隐蔽。1.3 分时、1分钟K线、逐笔成交的关系这几个概念新手容易混我统一说一次。1分钟K线按分钟聚合的OHLC适合做策略回测和指标计算。分时数据当天连续的实线适合看交易节奏、画分时图、做盘口监控。逐笔成交数据每笔真实成交的明细适合深度盘口分析但数据量大、接口压力也大。东财接口里分时数据走trends2/get这类接口逐笔成交走另外的详情接口两者不是一回事。如果只是做一块实时看盘大屏分时数据就够了要做精细化统计才需要再往逐笔层走。这个选型判断能省下很多不必要的接口调用成本和代码复杂度。2. 数据源选型为什么我继续用东财接口2.1 免费行情源那么多为什么选它前面几篇我反复提过自用型量化工具或者个人看盘程序最怕三件事要收钱、要申请繁琐的权限、返回格式不透明。券商提供的行情SDK功能强但通常需要开户资质和专门的运行环境个人开发者想在脚本里快速验证一个想法成本偏高。东财股票数据API的好处是不需要显式申请token接口就是普通的HTTP GET返回标准JSON而且Web端和移动端都在实际使用同一套数据通道说明它是生产环境级别的东西。当然免费就意味着一切都要在合理合规的范围内使用量不要太大、频率不要太高不能拿去做商业化大规模分发这点我们作为使用者心里要有数。我会用它来做个人盯盘工具和策略数据回放不会去碰违规高频抓取。2.2 请求地址与参数说明分时数据我是从历史行情节点取的完整请求地址https://push2his.eastmoney.com/api/qt/stock/trends2/get常用参数整理成一张表方便直接抄作业参数示例值作用secid1.600519行情代码市场前缀加股票代码fields1f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13控制返回的基础信息fields2f51,f52,f53,f54,f55,f56,f57,f58控制trends数组里的字段ndays1取最近几个交易日的分时数据iscr0是否复权0表示不复权fields2是核心我只取f51到f58这8个字段刚好覆盖时间、价格、量额和均价。如果需要筹码分布或者其他衍生数据可以自己在本地计算没必要让远端返回太多字段。2.3 secid的转换规则secid这块经常有人卡住。规则其实很简单A股个股以交易所代码开头判断沪市股票用数字1开头深市股票用数字0开头。比如贵州茅台是600519对应1.600519平安银行是000001对应0.000001。指数也类似上证指数是1.000001深证成指是0.399001。我用一个Python函数来处理def build_secid(code: str) - str: code code.strip() # 沪市主板、科创板凡是6开头 if code.startswith((6, 9)): return f1.{code} # 深市主板、创业板凡是0、2、3开头 if code.startswith((0, 2, 3)): return f0.{code} raise ValueError(f无法识别的代码: {code})这段代码单独放成工具函数以后其他接口取数都能复用。基金、可转债这些代码规则不同不在这一篇展开换场景时稍微注意一下即可。3. 核心实操从组装请求到解析数据3.1 完整取数代码直接上我本地跑通的版本环境是Python 3.10只依赖requests和标准库。import requests import time from datetime import datetime UA { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 } def get_trends(secid: str, ndays: int 1, timeout: int 10) - list: url https://push2his.eastmoney.com/api/qt/stock/trends2/get params { secid: secid, fields1: f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13, fields2: f51,f52,f53,f54,f55,f56,f57,f58, ndays: str(ndays), iscr: 0, } resp requests.get(url, paramsparams, headersUA, timeouttimeout) resp.raise_for_status() payload resp.json() if payload.get(data) is None: raise RuntimeError(fdata为空, 原始返回: {payload}) return payload[data][trends]这里有几个特意加的点headers里带上了完整的浏览器UA避免被网关当作异常客户端过滤。timeout明确写成10秒避免某个入参错误导致请求挂起。先判断data是否为None再往下解析减少不明不白的异常。调用方法secid build_secid(600519) trends get_trends(secid, ndays1) print(len(trends)) for row in trends[:3]: print(row)正常开盘日的下午收盘后当天数据应该有240条上下。开盘集合竞价也有数据所以实际看到的数量可能在241、242这个量级都是正常的。3.2 把字符串拆成结构化数据trends返回的是字符串数组需要把它转成能直接做计算的结构。我的做法是逐条split成8个字段再做类型转换FIELD_NAMES [time, price, open, high, low, volume, amount, avg_price] def parse_trends(trends: list) - list[dict]: rows [] for line in trends: parts line.split(,) if len(parts) len(FIELD_NAMES): continue obj { time: parts[0], price: float(parts[1]), open: float(parts[2]), high: float(parts[3]), low: float(parts[4]), volume: int(parts[5]), # 手 amount: float(parts[6]), # 元 avg_price: float(parts[7]), } rows.append(obj) return rows转完之后我再往dict里补几个派生字段涨跌幅、相对昨收的偏移、累计成交额的单位换算等。这些字段在渲染图表时几乎都用得到提前算好能减少后续重复代码。def enrich_rows(rows: list[dict], pre_close: float) - list[dict]: for row in rows: row[pct] (row[price] - pre_close) / pre_close * 100 row[volume_wan] row[volume] / 10000 return rowspre_close就是昨收价可以从接口返回的基础数据里取。如果懒得额外解析也可以直接用第一根分时的高开、低开情况反推但最稳妥的还是从响应里读昨收字段。3.3 当日数据的几个校验点数据拿到手后我建议先跑三个检查。第一看时间是否落在最近一个交易日的交易时段内。如果你在早上08:30请求当天数据接口很可能只返回空data或昨天的缓存这时候代码要能识别而不是直接把空数组写进缓存。第二看价格是否为零或负数。停牌股、异常数据源偶尔会出现价格字段为0的情况必须过滤。第三看时间序列是否严格递增。接口一般是有序返回但为了下游图表稳定仍然会做一次排序加去重。def validate_rows(rows: list[dict]) - None: if not rows: raise ValueError(分时数据为空) times [r[time] for r in rows] if len(set(times)) ! len(times): raise ValueError(存在重复时间点) if times ! sorted(times): raise ValueError(时间顺序异常)这三行代码不长但在一个跑了很久的数据服务里能拦住大量脏数据。4. 拿到分时数据之后缓存、刷新与渲染4.1 盘中轮询频率怎么设计分时数据是滚动更新的要持续监控就需要周期性去拉。个人自用程序我通常用3到5秒一次的频率既能画出一根大致连续的分时线又不会把免费接口打到限流。具体方案是早盘开盘前拉一次全量缓存盘中每5秒增量拉最新数据午间休市12:00到13:00完全停掉轮询。代码上我会写一个简单的调度循环from datetime import datetime as dt def should_trade() - bool: now dt.now() if now.weekday() 5: return False cur now.strftime(%H:%M) if 09:30 cur 11:30: return True if 13:00 cur 15:00: return True return False def loop_poll(secid: str, interval: int 5): cached [] while True: if not should_trade(): time.sleep(30) continue try: raw get_trends(secid, ndays1) rows parse_trends(raw) if rows: cached rows # 全量覆盖因为分时曲线本身就是累积数据 except Exception as exc: print(f刷新失败: {exc}) time.sleep(interval)这里为什么做全量覆盖而非增量拼接因为分时数据天生是累积型数据前面时间点的价格和成交量已经定案只会追加新的分钟点直接全量覆盖最不容易出错。只需在推送层比对上一个时间戳把新增点推送出去即可。4.2 分时图渲染的基本逻辑如果要做前端展示分时图的核心就四部分价格曲线、均价线、底部成交量柱、左右坐标轴。用前端图表库的话大多数库都有现成股票图表但和我上面的数据结构对齐需要多一点处理。价格曲线直接用price字段均价线用avg_price底部成交量柱用volume或volume_wan颜色则按涨跌决定。我平时用价格相对昨收的涨跌来决定每一根量柱的颜色上涨用红色系下跌用绿色系平盘用灰色。时间轴横坐标直接用解析后的datetime确保11:30到13:00这段没有数据的时间不要硬画成直线前端要把这段间隔跳过去。还有一个容易被忽略的点分时图X轴右侧往往要显示最新价和涨跌幅这些数据直接从rows[-1]取即可不要另外再去单独请求一次实时行情减少一次无谓的接口调用。4.3 增量更新怎么做更稳数据服务端向外推送的时候不要每次把几百KB的完整数据推给前端。我的做法是维护一个全量缓存在内存里每次轮询后对比旧缓存的时间戳集合把新出现的分时点组成一个轻量JSON数组推出去。前端收到之后往本地数组里追加即可。last_key set() def poll_and_publish(secid: str): global last_key raw get_trends(secid, ndays1) rows parse_trends(raw) now_key {r[time] for r in rows} new_rows [r for r in rows if r[time] not in last_key] last_key now_key return new_rows有新点才推送没有就静默。这样对带宽、对前端渲染压力都很友好。这个增量逻辑同样适合接入WebSocket或者SSE后续接实时推送通道的时候几乎不用改。5. 踩坑记录从翻车到稳定的一路心得5.1 高频问题速查表我把自己真实遇到过的几类坑整理成表格以后遇到可以直接对号入座现象可能原因解决办法返回data为Nonesecid写错或者股票停牌打印完整URL核对secid换成活跃股票测试请求一会儿成功一会儿超时免费接口负载波动使用Session复用连接加2到3次自动重试早上9点取不到当天数据接口缓存未生产完加sleep延时重试或先用昨收数据兜底分时数据多出几条集合竞价、临时停牌等正常现象不要写死条数按时间排序去重价格突然全为0数据源异常或停牌校验价格字段异常时保留上一份成功缓存打印时间比本地早或晚几个小时服务器返回UTC未转换解析后统一转为北京时间自动重试这段我贴一下代码注意要加退避别无限重试def get_trends_with_retry(secid: str, retries: int 3): for i in range(retries): try: return get_trends(secid) except Exception: if i retries - 1: raise time.sleep(1 i * 2)退避时间用1秒、3秒、5秒这种递增策略既能给服务端恢复时间也不会因为高频重试把自己送到限流名单里。5.2 我的工程习惯分时采集这个模块我已经重写过两轮踩过的坑不少最后沉淀下来几个固定习惯。第一原始数据一定要落盘。盘中拉到的每一份JSON都按日期单独存一份文本即使解析代码写错也还能从原始文件里恢复字段。第二接入方不要依赖我的解析函数。很多同事喜欢直接拿我解析好的dict来用一旦我升级数据版本他们代码全挂。所以我会保留一个兼容函数专门输出最原始的split结果。第三所有外部行情模块必须独立成服务。这样即使上游接口挂了也不会影响其他核心服务同时方便在故障时单独重启、单独降级。5.3 免费行情接口的不确定性如何应对免费接口最现实的问题就是不稳定、字段可能调整。我的应对方法是把它当外部依赖处理设置超时、做好重试、保留缓存、失败降级到最近一次成功数据。不要把它当成有服务等级承诺的金融数据服务来依赖这点心里要有数。做个人工具和中小业务这套方案完全够用。但如果业务要做正式运营、涉及真金白银的交易决策我强烈建议换成正规的数据服务商数据质量、响应时效和设备保障完全不是同一个量级。最后分享一个小技巧我在本地跑分时采集时会在日志里记录每次请求的耗时和返回条数。持续观察一周基本就能摸清这个接口在早盘、尾盘和午间休市的响应规律时间规律掌握了再调轮询策略就有的放矢不用瞎猜。