企业官网建设流程全解析

标题写到第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

常用参数整理成一张表，方便直接抄作业：

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 f'1.{code}' # 深市主板、创业板，凡是0、2、3开头 if code.startswith(('0', '2', '3')): return f'0.{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, params=params, headers=UA, timeout=timeout) resp.raise_for_status() payload = resp.json() if payload.get("data") is None: raise RuntimeError(f"data为空, 原始返回: {payload}") return payload["data"]["trends"]

这里有几个特意加的点：

• headers里带上了完整的浏览器UA，避免被网关当作异常客户端过滤。
• timeout明确写成10秒，避免某个入参错误导致请求挂起。
• 先判断data是否为None，再往下解析，减少不明不白的异常。

调用方法：

secid = build_secid("600519") trends = get_trends(secid, ndays=1) 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 rows

pre_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, ndays=1) 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, ndays=1) 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 高频问题速查表

我把自己真实遇到过的几类坑整理成表格，以后遇到可以直接对号入座：

自动重试这段我贴一下代码，注意要加退避，别无限重试：

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 免费行情接口的不确定性如何应对

免费接口最现实的问题就是不稳定、字段可能调整。我的应对方法是把它当外部依赖处理：设置超时、做好重试、保留缓存、失败降级到最近一次成功数据。不要把它当成有服务等级承诺的金融数据服务来依赖，这点心里要有数。

做个人工具和中小业务，这套方案完全够用。但如果业务要做正式运营、涉及真金白银的交易决策，我强烈建议换成正规的数据服务商，数据质量、响应时效和设备保障完全不是同一个量级。

最后分享一个小技巧：我在本地跑分时采集时，会在日志里记录每次请求的耗时和返回条数。持续观察一周，基本就能摸清这个接口在早盘、尾盘和午间休市的响应规律，时间规律掌握了，再调轮询策略就有的放矢，不用瞎猜。