1. 数据源选型的现实与妥协:为什么一个搞Python的绕不开同花顺
聊一个很实际的问题。但凡你用Python做股票相关的数据研究,第一周大概率会把主流的免费数据源挨个试一遍。tushare老版本要攒积分,新版Pro直接按积分档位卡权限,日线入门没问题,可一旦想取资金流、龙虎榜、财务附注这种细颗粒数据,门槛一下子上来。baostock干净、稳定,但覆盖面偏窄,缺的东西太多。akshare一直在更新,确实省事,但它本质上是把各种网页接口包了一层,你依然不知道背后的数据到底是从哪个口子出来的,出了字段变动只能干等库更新。
这种情况下,同花顺系的数据接口几乎是绕不开的选择。原因不复杂:同花顺作为老牌行情交易软件,它的数据链路覆盖行情、财务、资金、板块、资讯,几乎你能想到的日频和日内数据它都有现成的结构。更关键的是,它很多Web端数据接口并不强制要求登录和token,用Python直接请求能拿到相当完整的JSON数据,这对个人开发者来说是巨大的成本优势。
当然我要先说明白:同花顺官方是有正儿八经的商用数据接口的,比如iFinD,那套东西需要机构账号,个人基本不用想。我们平时在社区里讨论的“同花顺全数据接口”,是指它的公开Web行情接口以及配合浏览器调试手段能拿到的数据结构。这套东西适合个人学习研究、本地策略回测,不适合拿去商业化分发。这是边界问题,心里要有数。
我的建议是:与其盲目装一堆第三方库,不如先花半天时间把同花顺Web端的数据链路摸清楚。一旦你理解了它的请求规律,后面不管接口怎么微调,都能快速自我修复。这篇文章就是按这个思路来的——从环境准备开始,带你把行情、资金、财务几类常用数据都跑通,最后再讲稳定运行和排查问题的经验。
2. 环境准备与接口认知:先跑通第一次请求再说
2.1 Python环境与依赖库安装
先说环境。国内网络环境下的Python安装有时候会让人卡住,尤其是下载慢、装完pip还不能用的情况。我的建议是直接走官网下载安装包,版本选3.9到3.11之间都行,别追最新大版本,有些依赖库的预编译包更新没那么快。装的时候记得勾选“Add Python to PATH”,这一步能省掉后面大量“python不是内部或命令”的麻烦。
装完之后用国内镜像源配置pip,这一步是刚需:
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple pip install requests pandas numpy这三个库是先头部队。requests用来发HTTP请求,pandas做数据清洗,numpy做计算。等后面需要批量抓取的时候,再补concurrent.futures这种标准库就行,都不用额外装。
如果你打算从同花顺的某些加密JS参数里抠header字段,可能还需要一个执行JS的环境,比如pyexecjs或py-mini-racer。但我个人经验是,大多数行情接口的请求头可以用固定模板解决,不需要实时执行JS,只有遇到加密签名时才需要。这个后文会细说,现阶段先不用装。
2.2 同花顺Web数据接口的基本结构
很多初学者最大的问题不是不会用Python,而是不知道该请求什么地址。这里分享一个通用方法论:打开同花顺官网或行情中心页面,按F12打开开发者工具,切到Network面板,然后操作页面上的功能——比如翻页、切换个股、切换行情周期——观察哪些XHR请求在返回数据,直接把响应内容复制出来,就是最真实的数据结构。
我长期盯下来,同花顺Web接口的数据返回格式高度统一,绝大多数是JSON,少量是JSONP包裹。响应里基本都有一个状态字段(比如“status_code”或“error_code”),为0时表示成功;还有data节点,里面是字典嵌套列表的结构。请求参数一般是code(证券代码)、type(周期类型)、amount(数量)这类风格,清晰直白。
第一次写请求代码时,建议先带上完整的请求头再试。因为某些接口会校验Referer和User-Agent,缺了直接返回403或空数据。
import requests session = requests.Session() session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://q.10jqka.com.cn/", "Accept": "application/json, text/plain, */*" }) url = "https://q.10jqka.com.cn/api/php/.../xxx/yyy" params = { "code": "600519", "type": "day", } resp = session.get(url, params=params, timeout=10) print(resp.status_code) print(resp.text[:500])注意,我上面给的是一个示意路径,实际接口地址会随版本变化。你真正要掌握的,是用F12去找到当前可用的地址。这个方法比任何插件都可靠。
2.3 先从一只股票的历史K线开始
第一次完整跑通一个接口,建议选历史K线。原因是它结构最标准,验证最容易,而且后续算因子、做回测都用得上。解析逻辑上,不同接口的字段名可能不一样,我建议写成字典映射,不要硬编码索引。
import pandas as pd def parse_kline(raw_list): df = pd.DataFrame(raw_list) df.columns = ["date", "open", "high", "low", "close", "volume", "amount"] for col in ["open", "high", "low", "close", "amount"]: df[col] = pd.to_numeric(df[col], errors="coerce") df["volume"] = pd.to_numeric(df["volume"], errors="coerce") df["date"] = pd.to_datetime(df["date"]) return df.set_index("date").sort_index()第一次跑的时候大概率会遇到两类问题:一类是响应里带了“symbol”之类的多余字段,导致列数对不上;另一类是成交量单位不统一,有的接口返回手,有的返回股。这些都是正常的,说明你正在和真实世界的数据打交道,按实际情况调整解析函数即可。
总结一下这个阶段的重点:不要急着写一个大而全的抓取框架,先把“单个请求→解析→DataFrame”这个最小链路跑通,确认数据准确,再考虑批量。
3. 行情数据实战:实时快照、资金流与板块轮动
3.1 实时行情快照的批量获取逻辑
做日内监控和盘中选股,靠的不是历史K线,而是实时快照。所谓快照,就是某一瞬间的价格、涨跌幅、成交量、五档盘口等数据。同花顺在个股详情页里会周期性请求这些信息,接口返回的JSON通常包含“current_price”、“change_rate”、“volume_ratio”这类字段。
自动盯盘时,不要写死一个品种挨个去查,那样太慢了。同花顺的Web接口基本都支持按“板块内多代码批量查询”,一次请求可以传多个证券代码,用逗号分隔,返回的结果是一个列表,每个元素对应一个代码。这个设计对个人用户非常友好,请求次数直接降一个量级。
我实测下来,一次批量请求控制在50到80个代码以内比较稳妥,超过容易触发超时或畸形响应。如果你要监控全市场5000多只票,正确做法不是一次请求全传,而是按市场板块分批拉。先拉“上证A股”、“深证A股”、“创业板”的成分列表,再按批次取快照。板块成分列表本身也是个接口,这样可以保证数据覆盖完整。
关于频率,纯人工盯盘不需要高并发,一秒一次以内就够。即使做盘中因子计算,5秒一次也差不多了。不要拿爬虫的思路去请求行情,接口被限流之后恢复很麻烦。
3.2 历史K线参数选择的经验
历史K线接口里,type字段的取值每家略有不同,同花顺系的通常有“day”、“week”、“month”三档,部分接口支持分钟级。我的建议是,尽量在本地把日线数据做增量保存,周线和月线由日线自己聚合生成,不要每次都从接口去拉。
为什么?一是周线和月线接口的数据量虽然不大,但重复请求没有任何意义;二是自己聚合的K线,复权逻辑是连贯的,不会出现周线和日线口径不一致的问题。聚合方法用pandas的resample就能实现:
def resample_weekly(df: pd.DataFrame) -> pd.DataFrame: weekly = df.resample("W-FRI").agg({ "open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum", "amount": "sum" }) return weekly.dropna()注意我用了“W-FRI”这个对齐方式,就是把周五作为一周的结束日。A股没有夜盘,周五收盘就是一周定格,这是国内行情数据聚合的基本常识。用默认的“W-SUN”会导致最后一个交易日被算到下一周去,这是新手很容易踩的坑。
3.3 资金博弈指标与资金流的计算口径
这次的热搜词里有一条“同花顺资金博弈指标源码”,说明很多人对资金流数据兴趣很大。同花顺客户端里的“资金博弈”指标,核心思想是把成交单按大小分成主力、散户两个阵营,红线代表主力资金净流入,绿线代表散户资金净流出,本质上是一个基于分笔成交数据的累计模型。
用Python复现的逻辑不复杂。如果拿不到逐笔成交明细,可以用分时成交接口近似处理。思路是:读取当日分钟级别的成交量和成交均价,用价格涨跌方向近似判断资金方向,再按单笔金额大小做阈值切分。
一个简化版的实现思路如下:
def estimate_main_force(df_minute, big_threshold=100000): df = df_minute.copy() df["direction"] = df["close"].diff().fillna(0) df["direction"] = df["direction"].apply(lambda x: 1 if x > 0 else (-1 if x < 0 else 0)) df["amount"] = df["volume"] * df["close"] df["net"] = df["amount"] * df["direction"] df["big_net"] = df["net"].where(df["amount"] >= big_threshold, 0) main_force_net = df["big_net"].sum() return main_force_net注意,这只是一个示意性近似,不是同花顺的精确算法。真正的资金博弈指标会用到委托方向、成交性质等更细的数据,个人能拿到的公开数据很难完全一致。但作为策略因子,这个方向是有效的——它捕捉的是大单驱动的净流入。
3.4 板块行情与同花顺概念分类的抓取
做轮动策略的人,通常不会只看个股,板块概念的热度更重要。同花顺的概念板块分类在Web端有现成的列表接口,返回每个概念板块的涨跌幅、领涨股、总成交额等字段。
我抓板块数据时通常两个维度结合看:一是板块当日涨幅排名,二是板块内个股的普涨程度(上涨家数/总家数)。后者比前者更真实,因为少数权重股拉升也可以把板块指数拉红,但那不代表板块整体强。这个指标可以用Python快速计算:
df["up_ratio"] = df["change_rate"].apply(lambda x: 1 if x > 0 else 0) 板块强度 = df["up_ratio"].mean()结合板块热度排名和内部强度,基本能过滤掉大多数“虚涨”板块。这个思路放到任何一家行情数据源都适用,不局限于同花顺。
4. 基本面数据的获取:F10资料与财务字段的拼接方法
行情数据解决的是“交易”问题,基本面数据解决的是“选股”问题。同花顺的F10资料页里包含公司概况、财务摘要、主营构成、股东信息等,Web端也能找到对应的JSON接口。
4.1 F10财务摘要的接口与解析
F10财务摘要接口的返回结构,通常是一个多级字典,按报告期分组。比如每年的年报和季报,每个报告期下有营收、净利润、ROE、毛利率等字段。解析的时候建议转成“宽表”,即每一行是一个报告期,每一列是一个财务指标。
def flatten_financial_report(meta_list): records = [] for item in meta_list: report_date = item.get("report_date") data = item.get("data", {}) record = {"report_date": report_date} for k, v in data.items(): record[k] = v records.append(record) df = pd.DataFrame(records) return df财务数据单位是个大坑。有些字段单位是万元,有些是元,还有些是百分比的小数形式。我建议在第一次清洗时统一做好单位转换,并把字段名规范成自己习惯的英文命名,后面写因子计算会舒服很多。
4.2 股本变动与除权除息数据的对齐
做财务数据和行情数据拼接的时候,有一个非常容易被忽略的隐藏问题:股本变动。如果一只股票在报告期内实施了送转股或者增发,它的总股本变了,那每股收益、每股净资产这些指标在相邻报告期之间就不可直接比较。
同花顺的F10数据里通常包含“股本结构”或“分红扩股”栏目,可以拿到具体的变动日期和变动原因。使用上,我的经验是:算同比增长率时,注意用“单季度数据”,而不是直接用累计数硬比。比如三季报的累计净利润,同比对比也应该是去年三季报的累计数,而不是去年年报数。这个逻辑写错了,整条选股因子的可信度就崩塌了。
4.3 防止财务数据里的ST和上市不足次新股污染样本
财务数据拼接好之后,还有一道过滤工序不能省。ST股票因为连续亏损,财务指标和股价往往异常,不排除它们会对因子测试产生极端干扰。上市不足60个交易日的次新股,因为没有完整的历史数据,也不适合放进回测样本。
df = df[~df["stock_name"].str.contains("ST")] df = df[df["list_days"] >= 60]这两句话看着简单,但能极大改善回测结果的质量。真实做研究时,脏数据对结论的影响往往比模型本身更大。
5. 数据完整性与口径对齐:最容易翻车的环节
5.1 前复权、后复权和不复权,到底怎么选
在本地保存行情数据时,最让人头疼的问题就是复权。同花顺接口一般提供前复权和后复权两种计算结果。我的建议是:本地原始行情统一保存不复权数据,需要复权的时候再用复权因子自己计算。
为什么这么干?因为前复权数据是动态的——每次发生新的除权除息,历史价格都会被重新调整,之前保存的数据就“过期”了。如果你每天增量保存前复权数据,在除权日之后,你会发现自己数据库里的历史价格和接口返回的对不上,越积越乱。不复权数据则永远是那个原始价格,任何时候重新计算复权因子都能还原。
复权因子的计算逻辑也不复杂。以除权日和除权价为锚点,后复权因子是逐日连乘出来的,前复权只是后复权整体缩放。代码示意:
factor = (1 + 送转比例 + 每股分红 / 除权前收盘价)实际操作中用pandas计算也很快,关键点在于记录每次除权除息的日期和方案。把这些信息存成本地表,每次要复权时重新生成一次就好了。
5.2 停牌日期空缺的填充逻辑
A股市场经常有停牌,短的一天,长的半年。行情接口对停牌日的处理方式不一:有的直接不返回该日数据,有的返回0成交量。如果你用pandas的to_datetime做索引,然后去rolling算均线,停牌空缺会导致窗口错位,必须显式地补全交易日序列。
正确做法是:用同花顺的交易日历数据生成一个完整的交易日序列,然后left join行情数据,缺失的成交量填0,价格用前值填充或直接保留NaN。用NaN表示停牌,比填0更安全,因为很多指标公式里0会被当成真实的价格参与计算。
df = df.reindex(trade_calendar) df["close"] = df["close"].fillna(method="ffill") df["volume"] = df["volume"].fillna(0)5.3 接口返回字段频繁变动的应对策略
Web接口不是稳定不变的REST API,随时可能调整字段名或返回值结构。应对方法不是抱怨,而是写一层适配器。我在实际项目里会定义一个规范的内部数据结构,所有外部数据都先转换成内部结构,再做后续处理。这样外部接口变动时,只需要改适配器这一层,其他业务代码完全不用动。
另外强烈建议做数据缓存。每次拉取K线都直接落到本地SQLite或Parquet文件,下次请求前先查本地时间戳,当天数据就直接用本地。既能缓解接口压力,也能在接口异常时保证研究进程不中断。我用的是SQLite,单文件、无服务、随时查询,非常符合个人量化研究的场景。
6. 批量抓取与性能优化:代码跑得慢不一定是接口的锅
6.1 多线程的正确打开方式
单线程逐个请求,1000只股票每个请求0.3秒,加起来就是5分钟。虽然也不是不能用,但如果你每天要抓全市场日线、财务摘要、资金流三套数据,那就得几十分钟起步了。这时候需要并发。
Python里我习惯用concurrent.futures的ThreadPoolExecutor。它比手动写threading简单,又比multiprocessing省内存。行情接口本身就是IO密集,线程池正合适,把请求函数传给executor.map就行:
from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_stock(code): data = get_kline(code) return code, data with ThreadPoolExecutor(max_workers=8) as pool: futures = [pool.submit(fetch_stock, c) for c in stock_list] for fut in as_completed(futures): code, data = fut.result() save_to_sqlite(code, data)6.2 并发数与频率控制的平衡
并发不是越大越好。接口服务端通常有限流策略,请求太频繁会触发封禁。我实测下来,8个线程、每次请求间隔控制在0.2秒到0.5秒之间,是比较稳的组合。全市场股票半小时内就能抓完,也不会触发异常。
要注意的是,如果同一个接口同时开太多线程,很快会收到签名校验失败或者直接超时的响应。这是服务端的保护策略,不是你的代码写错了。遇到这种情况,退避重试就行,不要硬刚:
for retry in range(3): try: resp = session.get(url, params=params, timeout=10) data = resp.json() break except Exception as e: time.sleep(1 * (retry + 1))6.3 用增量更新替代全量更新
数据抓全一次之后,后续就不要再做全量更新了。每天的增量更新只需要拉最近5个交易日的K线、当日快照和当日资金流,数据量小,请求快,对接口也友好。
增量更新的核心是记录每个代码的本地最新日期,然后从这个日期往后请求。注意在周末或节假日之后,最新日期和当前日期之间可能隔着多个交易日,增量更新时不要把区间写死为“昨天”,最好动态计算。
我看到很多人写爬虫框架时把重心放在并发和反爬上,却忽略了“任务调度”本身。建议把更新任务写成可断点续跑的:每次完成的代码记录到一个状态表里,中途挂掉后重启,只跑未完成的。这个小设计在实盘监控场景中能帮你省很多事。
7. 常见报错排查与长期稳定运行建议
7.1 高频报错场景对照
这里把我在实操中遇到的、以及辅导别人时常见的报错做个汇总,方便你排查时对号入座。
| 现象 | 最可能原因 | 解决方向 |
|---|---|---|
| 返回403或302 | 缺少Referer或Cookie | 补全请求头,必要时先用浏览器登录一次 |
| 返回JSON但data为空列表 | 参数名或证券代码格式错误 | 用F12实际请求观察参数拼写 |
| 部分代码返回成功,部分失败 | 临时停牌或退市整理期个股无数据 | 捕获异常,记录失败的代码到失败清单 |
| 响应里出现签名或加密字样 | 接口启用了JS签名校验 | 降低请求频率,改用符合条件的接口 |
| 请求超时 | 本地网络或接口负载高 | 设置重试机制,超时时间合理放大 |
| 字段key不存在报KeyError | 接口字段变更或数据确实缺失 | 用dict.get方法代替直接索引 |
7.2 用日志记录代替print调试
抓取程序一旦开始跑,控制台print的信息稍纵即逝。强烈建议用logging记录运行状态,包括每次请求的代码、耗时、返回状态。这样出了问题才能定位是哪个代码、哪个步骤导致的。
import logging logging.basicConfig( filename="collector.log", level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" )日志的另外一个作用,是帮你发现“没有报错但数据不对”的隐性bug。比如某天你发现成交量突然少了,翻日志才发现是接口单位变了。这类问题只有日志能帮你还原现场。
7.3 长期维护的几个关键习惯
数据抓取是个长期工程,运行稳定半个月不难,难的是连续跑半年一年不出问题。我总结几个实用习惯:
第一,每两周做一次接口探活。随便取一两个代码,请求一下核心接口,确认返回正常。不正常的接口尽早发现尽早调整。第二,本地数据库定期备份。SQLite数据库直接拷贝文件就行,每天备份一次,保留最近7份,基本就不会丢数据。第三,代码里所有接口地址不要硬编码散落各处,集中放在配置文件里。改接口时只改一处,方便快捷。
关于同花顺“全数据接口”这件事,可行的路径其实是组合式的——行情走行情接口,财务走F10数据源,资金流用分钟数据进行估算。没有哪一个接口能包揽所有,但组合起来,覆盖90%的个人量化研究需求是完全没有问题的。真正有价值的不只是拿到数据,而是拿到干净、连续、口径一致的数据。这一篇里讲的解析规范、复权逻辑、停牌处理、增量更新,本质上都是围绕这个目标。
最后再补充一个很实用的小习惯:每次新增一种数据字段,先在本地写一个最小验证脚本,打印几条样例数据人工核对一遍,再纳入正式的抓取流程。数据采集这种事,处理脏数据的功夫往往比抓数据本身还要大,但这也是一个数据研究者必要的基本功。