企业官网建设流程全解析

这一篇是“股票数据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开头的多半是深市。北交所的代码规则不太一样，需要用的时候不要套用这个简单判断，先拿真实代码做一次验证再说。

完整请求参数我整理成了下面这个表：

fields1我一般直接写成：

f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13

fields2写成：

f51,f52,f53,f54,f55,f56,f57,f58

ndays这个参数值得单独说一句。只做当天盘中监控时，老老实实传ndays=1，返回的序列就是从当天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的声明顺序，约定解析结果对应如下：

实际代码可以这样写：

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秒请求一次。

我从稳定性的角度给个参考：

如果只是做一个自选股列表看板，二三十只股票轮询，60秒间隔完全够用。低于15秒其实意义不大，因为行情源的分钟数据本身也是按分钟维度聚合的，你拉得再频繁，拿到的还是同一分钟数据。

4.2 增量缓存：别把全天数据反复拉

很多自建行情模块最大的问题是“每次请求都把当天全量序列拉一遍”。早盘时还好，到下午两三点，一天的分时数据已经有两百多条，你再把整个数组反复解析、反复渲染，纯属浪费。

我的做法是分两层：

1. 内存层：保存当天最后一次拉到的完整序列，后续轮询只覆盖更新。
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, timeout=10, retries=3, sleep_interval=0.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}&iscr=0" f"&fields1={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, timeout=self.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 f"1.{symbol}" if symbol.startswith(("0", "2", "3")): return f"0.{symbol}" # 北交所和自定义证券代码无法通过首位判断，需要自行映射 raise ValueError(f"无法自动识别secid: {symbol}")

这里有个现实问题：A股代码首位规则已经挺成熟，但科创板688开头在沪市、创业板的300在深市，规则都没问题。可一旦遇到北交所83、87、920开头的证券，0或1的简单判断就未必可靠。多市场组合时，我建议单独维护一张代码到市场的映射表，而不是依赖首字母规则。

5.3 最小可运行示例

如果你只想快速验证一下，下面这段代码可以直接跑：

fetcher = MinuteTrendFetcher() result = fetcher.fetch("1.600519", ndays=1) 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. 我在生产环境里踩过的坑和最终使用的checklist

6.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 result

6.4 我最终使用的checklist

把项目跑顺之后，我给自己沉淀了一份上线检查清单，每次新接入股票或新部署环境都会过一遍：

1. 确认secid前是否自动判断正确，尤其是北交所和特殊代码。
2. 确认服务器时区为东八区，或用zoneinfo强制指定。
3. 确认成交量单位按“手”展示，按“股”计算时手动乘100。
4. 盘中轮询固定30-60秒，不在同一秒并发大量请求。
5. 停牌、午间、开盘前三种空数据场景全部做了单独分支。
6. 每次请求只更新增量数据，不在盘中重复解析全量历史。
7. 把rc != 0、连续请求失败、全天无新增数据三种异常都打上监控日志。

最后分享一个我自己的小习惯：盘中轮询的日志不要每60秒打一条，太吵了。我只打“首次请求成功”和“出现异常”两类日志，正常状态用静默计数。这样出问题时不会被海量正常日志淹没，排查效率高出一大截。分时数据采集这个事，技术难度不大，真正拉长战线的永远是这些细枝末节的稳定性处理。对“股票数据API”系列来说，把这块做扎实，后面的实时监控、盘中预警才好继续往下搭。