不想一上来就聊技术细节先说说背景。很多做量化研究或者金融数据分析的朋友起步阶段最头疼的不是策略怎么写、模型怎么跑而是数据从哪来。市面上商业数据库贵得离谱手工复制粘贴又效率太低这时候Python爬虫几乎成了唯一现实的选择。但金融数据又有点特殊一方面涉及合规问题不能乱抓另一方面数据质量直接决定分析结果宁可慢一点也要干净。这篇文章聚焦两个主力数据源Tushare和东方财富公开接口。前者适合拿规范的结构化金融数据后者适合做实时行情、资金流向、公告信息等公开数据采集。我把从账号准备、接口调用到数据入库的完整链路串一遍重点讲清楚每个环节为什么这样做以及踩过的坑。1. 金融数据获取的双轨思路1.1 为什么Tushare是合规首选先说Tushare。做量化入门或者学术研究Tushare Pro几乎绕不开。它提供股票行情、财务报表、宏观经济、行业分类、基金净值、期货数据等数百个接口数据经过清洗和复权处理直接拉过来就能用。最关键的原因是合规。Tushare要求用户注册、实名认证、申请token数据使用有明确的用户协议和频率限制。相比直接爬取其他网站这种模式在合规层面要稳得多。尤其是打算写论文、做正式报告甚至商业项目用Tushare的数据源在合规风险上要小很多。另外一个优势是标准化。Tushare返回的是规整的DataFrame格式字段命名统一日期格式规范不需要做太多预处理。比如日线行情接口返回字段包括ts_code、trade_date、open、high、low、close、vol、amount等几乎就是量化模型需要的原始格式。相比之下网页上抓来的数据字段名五花八门清洗成本不低。1.2 东方财富公开数据的不可替代性那为什么还要爬东方财富因为Tushare的很多接口有积分门槛。实时盘中数据、资金流向、龙虎榜明细、股票弹幕舆情这些高频数据要么积分要求很高要么干脆没有。而东方财富作为公开财经网站提供了大量公开的浏览器API接口这些接口不需要登录、不需要付费数据更新还非常快。比如个股资金流向数据东方财富的数据接口返回当天的主力净流入、超大单净流入、中单净流入、小单净流入等维度。这种数据对短线交易和资金面分析非常有用Tushare免费的积分档位不一定能拿到。再比如盘中实时分时数据东方财富的接口几乎是无延迟的做盘中监控非常顺手。还有一个优势是覆盖面。东方财富的接口涵盖了股票、基金、债券、期货、外汇、宏观指标等众多类型一个接口格式摸清之后套用参数规则就能扩展到整个站点体系。所谓一次分析、全家复用这个特点在实际操作中非常节省时间。所以我的习惯是Tushare负责基础数据和低频数据的合规获取东方财富公开接口负责高频、实时、网站特有的数据补充。两者结合基本能覆盖从宏观到个股的绝大多数分析场景。2. Tushare从注册到数据落地的完整流程2.1 注册与token管理Tushare Pro的注册流程不复杂。打开官网用手机号注册然后在个人主页找到接口TOKEN页面复制属于自己的token字符串。这里有个细节token是用户身份的唯一凭证所有的API调用都需要带上这个token一旦泄露别人就能消耗你的权限和积分。我的做法是把它写到环境变量里不要硬编码在代码中尤其是准备发到GitHub上的项目。setup如下import os import tushare as ts from dotenv import load_dotenv load_dotenv() # 加载.env文件 TOKEN os.getenv(TUSHARE_TOKEN) ts.set_token(TOKEN) pro ts.pro_api()有人图省事直接写token明文我强烈不建议。换成环境变量或配置文件多花不了半分钟安全等级提升一大截。2.2 核心接口的调用与参数细节Tushare的接口体系比较统一核心是pro.query()或者直接调用pro.xxx()方法。以股票日线行情为例import tushare as ts import datetime pro ts.pro_api() # 获取某个交易日所有股票的日线行情 df pro.daily( trade_date20250110, fieldsts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount ) print(df.head())这个接口的参数设计有个细节要注意trade_date和ts_code至少要传一个。传trade_date会返回当天全市场的行情数据传ts_code返回单只股票的历史行情。两个都不传会报错两个都传则以股票为准。另外Tushare的很多接口有积分门槛。日线行情接口的基础积分很低基本注册之后就能用但像moneyflow资金流向、cyq_perf筹码分布这类接口需要更高的积分。积分不够怎么办两个途径一是完善个人资料、使用平台功能攒积分二是充值获得高权限账号。不要想着用抓包绕过平台的服务端校验做得比较到位破解成本远高于充值成本。2.3 交易日历与日期边界问题一个实际使用中很容易踩的坑节假日和停牌日。比如元旦、春节前后你以为当天有数据结果接口返回空DataFrame。这就是典型的数据缺失陷阱——不是代码出错了而是当天根本不是交易日。解决方法是维护一张交易日历表。Tushare提供了trade_cal接口返回每个交易所的日历数据import tushare as ts pro ts.pro_api() cal pro.trade_cal( exchangeSSE, start_date20250101, end_date20251231 ) # is_open 1 表示交易日 trade_days cal[cal[is_open] 1][cal_date].tolist() print(trade_days[:10])有了这张表在写遍历逻辑时就能动态跳过非交易日。否则你写个循环从1月1日跑到12月31日遇到节假日接口返回空数据你的代码要么报错、要么把空数据写进数据库后续分析全乱套。我个人强烈建议第一次建库时先把全年的交易日历抓下来存进数据库后续所有定时任务都先查日历表再决定是否运行。这个设计在后续维护中能省掉大量莫名其妙的报错。2.4 限频与并发控制Tushare对每分钟的调用次数有明确限制不同积分等级对应不同的每分钟最大请求数。免费用户大概每分钟几次到几十次不等如果写个循环批量拉历史数据很容易触发抱歉您每分钟最多访问该接口X次的错误。我踩过的坑有一次拉全市场5年的日线数据用了个for循环一个接一个请求跑了几百次之后突然连续报错。排查半天是触发了接口的分钟级频率限制。后面我改成带time.sleep()的限速版本才稳定下来。import time # 批量获取历史行情注意控制频率 all_dfs [] for date in trade_days: df pro.daily(trade_datedate, fieldsts_code,trade_date,open,high,low,close,pre_close,pct_chg,vol,amount) if df is not None and not df.empty: all_dfs.append(df) time.sleep(0.6) # 保守等待避免触发限频并发也不是越大越好。Tushare服务端有尽量幂等的设计但请求过多会拖慢响应甚至返回错误。数据抓取这个场景稳定性远大于速度宁可多等两分钟也别让自己被封。3. 东方财富公开接口的抓包分析与实战3.1 如何找到接口地址东方财富的数据并非直接渲染在HTML里而是通过JS脚本异步请求后端API获取JSON数据。所以要爬东方财富核心不在于写HTML解析器而在于找到这些公开API地址。打开浏览器开发者工具F12切到Network网络面板然后在页面上点击查看个股行情或资金流向就能看到大量XHR请求。过滤掉图片、CSS、字体文件剩下的就是JSON数据接口。以股票资金流向为例列表页请求的地址长这样https://push2.eastmoney.com/api/qt/stock/fflow/kline/get?lmt0klt101fields1f1,f2,f3,f7fields2f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65secid1.600519secid20.600519这个接口返回的是某只股票的历史资金流向数据数据结构是JSON字段非常规整。f51是日期f52~f56是不同类型资金的净流入额f57~f61是净流入占比等。解析起来比网页HTML简单太多。3.2 secid参数的两段式结构东方财富接口中几乎所有行情接口都要带secid参数它决定了请求的是哪个市场、哪只证券。这个参数的结构是市场代码.证券代码。证券市场代码有一定的约定上海A股是1深圳A股是0北京A股是0还是0指数是1或0按市场区分。例如贵州茅台600519在上交所secid1.600519平安银行000001在深交所secid0.000001。第二个参数secid2在部分接口中会出现通常用于指定对比对象。比如资金流向接口中secid20.600519可能指的是深市前缀映射的同一代码具体含义要看不同接口的约定。实操时最稳的办法用浏览器开发者工具直接复制完整的请求URL改代码值测试确认接口能返回预期数据后再封装。这里有个容易搞混的地方沪深市场的代码前缀规则和我们在Tushare里看到的ts_code不同。Tushare的600519.SH是字符串后缀东方财富的1.600519是整数点整数。两个体系之间转换时要写个映射函数别搞混了。def tushare_ts_to_eastmoney_secid(ts_code: str) - str: code, market ts_code.split(.) if market SH: return f1.{code} elif market SZ: return f0.{code} elif market BJ: return f0.{code} else: raise ValueError(fUnknown market: {market})这个映射函数我在实际项目中一直在用非常省心。3.3 行情快照接口与分时数据除了资金流向东方财富的行情快照接口也很常用。比如获取个股的最新价、涨跌幅、成交量、换手率、市盈率等实时数据请求这个地址https://push2.eastmoney.com/api/qt/stock/get?secid1.600519fieldsf43,f44,f45,f46,f47,f48,f50,f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65这些字段代码的含义各不相同f43是最新价f44是最高f45是最低f46是开盘f47是成交量f48是成交额f50是量比f51是涨停价f52是跌停价f53是静态市盈率f54是动态市盈率等等。这里有个细节f43等价格类字段返回的是整数比如最新价105900对应的实际价格是105.90不对——东方财富很多接口的价格字段单位是分需要除以100换算成元。但这个规则并不统一有的字段本来就是元有的字段是分。我在实际项目中总结了一套校验方法拿接口返回的f43值和你看到的页面价格对比如果差100倍就整除100如果一致就保持原样。这个校准步骤必须做一次不然数据分析全是错的。比如600519当日实时价格import requests import json params { secid: 1.600519, fields: f43,f44,f45,f46,f47,f48,f50,f57,f58,f59, } resp requests.get( https://push2.eastmoney.com/api/qt/stock/get, paramsparams, headers{User-Agent: Mozilla/5.0} ) data resp.json()[data] if data: print(json.dumps(data, ensure_asciiFalse, indent2))返回结果大致长这样实际值可能有变化{f43: 152601, f44: 153800, f45: 152000, f46: 153200, ...}此时f43 152601如果页面显示1526.01元那说明单位确实是分需要/100。我拿过贵州茅台的实时数据做过对比确认这个数值需要除以100。3.4 公开接口中的字段映射与数据清洗东方财富接口的字段名都是f51这种纯数字代号可读性很差。在写存储逻辑时建议建立一个字段映射表把数字代号映射成语义化的字段名。EASTMONEY_FIELD_MAP { f51: trade_date, f52: main_net_inflow, f53: small_net_inflow, f54: medium_net_inflow, f55: large_net_inflow, f56: super_large_net_inflow, # ... 按需补充 }有了映射表后续分析代码直接读列名不用去记每个数字的含义。这个映射关系也建议以JSON或Python常量形式独立存放接口变动时只改映射表不改业务逻辑。4. 数据持久化SQLAlchemy构架数据存储4.1 为什么选择SQLAlchemy爬下来的数据总要存起来。早期我图简单直接存CSV文件但很快发现三个问题一是文件越来越多难以管理二是增量数据要全量重写效率极低三是后续用SQL查询时得加载全部文件再过滤内存吃不消。后来切换到SQLite单文件数据库轻量方便但并发写入时容易锁库。再后来用PostgreSQL功能强大但本地部署需要额外维护一个服务。如果你只是在单机做数据分析SQLite搭配SQLAlchemy是一个非常平衡的方案如果要把数据提供给多个服务共享建议用MySQL或者PostgreSQL。代码层面用SQLAlchemy抽象之后切换数据库几乎不用改业务代码。选择SQLAlchemy的理由很直接ORM映射可以把DataFrame的行列结构直接对应到数据表字段不用手写拼接SQL内置连接池管理不用自己处理连接释放支持自动建表、字段变更迁移。以日线行情为例定义模型from sqlalchemy import Column, Integer, String, Float, Date from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class StockDaily(Base): __tablename__ stock_daily id Column(Integer, primary_keyTrue, autoincrementTrue) ts_code Column(String(16), indexTrue) trade_date Column(Date, indexTrue) open Column(Float) high Column(Float) low Column(Float) close Column(Float) pre_close Column(Float) change Column(Float) pct_chg Column(Float) vol Column(Float) amount Column(Float) def __repr__(self): return fStockDaily ts_code{self.ts_code} date{self.trade_date}4.2 DataFrame批量写入的高效方式Tushare返回的是DataFrameSQLAlchemy ORM要逐条session.add再commit数据量一大就非常慢。更高效的方式是使用df.to_sql()配合SQLAlchemy的engine一次性批量写入。import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///stock_data.db) # 将DataFrame写入数据库 df.to_sql( namestock_daily, conengine, if_existsappend, indexFalse )to_sql内部执行的是批量插入速度远快于逐条add。但有几个细节要注意第一if_existsappend表示数据追加前提是表结构已经存在。第一次写入时可以先if_existsreplace建好表之后改成append。第二to_sql默认对datetime列要求是Python的datetime对象不要传字符串否则可能报错。Tushare返回的trade_date本身是字符串格式如20250110最好先转成日期类型df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d).dt.date第三to_sql不支持UPSERT存在则更新不存在则插入逻辑重复运行会插入重复数据。解决方法是配合数据库唯一索引或者每次写入前做去重。我一般是在ts_code trade_date上建一个唯一索引再配合批量插入前后的DELETE语句实现幂等写入。对于东方财富接口返回的数据同样可以先标准化字段名、转换数据类型再走to_sql入库。4.3 增量更新策略爬虫类数据更新最好的策略是增量而非全量。核心逻辑先查询数据库中最大的日期再从这个日期往后拉取数据。这样既节省请求时间也减少数据库冗余写入。from sqlalchemy import create_engine, text engine create_engine(sqlite:///stock_data.db) with engine.connect() as conn: result conn.execute(text( SELECT MAX(trade_date) FROM stock_daily WHERE ts_code :code ), {code: 600519.SH}) max_date result.scalar_one_or_none() if max_date is None: start_date 20200101 else: start_date max_date.strftime(%Y%m%d) print(f增量拉取起始日期{start_date})这个逻辑在定时任务里至关重要。每天晚上跑一次增量脚本把当天新增数据写入数据库持久化为历史库分析时直接查库即可不必每次都重新爬一遍。5. 实战中的常见问题与反爬策略5.1 请求频率礼貌爬虫的生存法则东方财富的公开接口虽然没有显式的登录限制但对请求频率仍然有隐形的管控。短时间内高并发请求很容易触发IP封禁或者请求返回乱码。我踩过一次比较大的坑写了一个多线程加速脚本同时开20个线程拉取全市场实时行情跑了几分钟后接口返回的数据开始错乱部分请求直接超时。原因就是单位时间内的请求量超过服务端的QPS阈值。解决思路是单线程合理延时把并发控制在一个安全的水平import time import requests from concurrent.futures import ThreadPoolExecutor def fetch_one(secid): params {secid: secid, fields: f43,f44,f45,f46,f47,f48} resp requests.get(https://push2.eastmoney.com/api/qt/stock/get, paramsparams, timeout5) time.sleep(0.3) # 每个请求之后等待0.3秒 return resp.json() secids [1.600519, 0.000001, 1.601318] with ThreadPoolExecutor(max_workers3) as executor: results list(executor.map(fetch_one, secids))注意我用的是3个线程而不是10个或20个。实测下来东方财富接口在低并发合理延时的模式下响应最稳定数据完整率接近100%。如果追求极致速度控制在0.2秒延时左右也没问题但超过10个并发就很容易出问题。5.2 数据字段缺失与单位混淆这是新手最容易翻车的地方。东方财富的接口字段数量非常多但并非所有字段在每只股票上都返回有效值。比如新股上市初期部分指标可能为空停牌股票的价格类字段可能返回-或者0。这些异常值如果直接入库会对后续分析造成污染。我的处理方法是在入库前加一个数据质量校验层def clean_row(raw): 对接口JSON原始字段做数据清洗 1. 将字符串类型的-转为None 2. 将数值型字段转换为float失败则置为None 3. 价格类字段按比例换算为元 cleaned {} for key, value in raw.items(): if value - or value is None: cleaned[key] None continue try: float_value float(value) if key in (f43, f44, f45, f46): # 价格字段以分为单位 cleaned[key] round(float_value / 100, 2) else: cleaned[key] float_value except (ValueError, TypeError): cleaned[key] None return cleaned有了清洗层入库数据基本可以保证类型统一、无脏值后续做技术分析时不用反复处理异常数据。另一个容易忽略的细节接口返回的涨跌幅字段单位是百分比数值直接返回比如pct_chg1.25代表1.25%不要额外除100。如果你在清洗层里统一对所有字段除以100就会得到一个错误的结果。所以我每次对接一个新接口第一件事是输出一批样本数据手工核对特殊字段确认单位规则后再写清洗逻辑。5.3 接口变动与失效的应急预案公开的接口终究不是正式开放的开发者接口它的参数和路径可能会调整。我遇到过几次接口地址或字段含义变动的情况排查过程并不轻松。有几个经验第一接口返回的JSON里通常包含错误提示或状态码先把这些信息打日志。不要只记录response.status_code而是把返回的JSON原文都存下来排错时能少走很多弯路。第二定期用一个小脚本监控关键接口的响应结构。比如每周跑一次检查f43字段是否还在、返回数据条数是否正常。一旦出现异常马上发通知提醒人工介入。第三代码中把接口URL和字段映射独立配置。接口变动时只需要改配置文件和映射表不需要动整个项目。这看起来是个很基础的设计但很多项目因为图方便把URL写死在多个模块里最终自食其果。比如东方财富的push2.eastmoney.com域名偶尔会出现波动一些备用域名也可以替换使用。配置化之后切换环境非常方便。5.4 本地调试与线上部署的差异同一个爬虫脚本在本地跑得很欢放到服务器上可能各种报错。最大的差异在于网络环境和Python环境。网络层面服务器如果不在国内访问东方财富接口可能超时或者返回异常。解决方法是部署前先测试连通性必要时使用国内云服务器。另外服务器上的DNS解析也偶有问题可以在requests的params中直接指定Host头或者使用IP直连方案但要小心合规风险建议统一通过域名访问。环境层面线上环境经常没有图形界面也不一定有selenium需要的浏览器。所以我在爬虫设计上一开始就避免依赖浏览器引擎全部通过requests直接请求HTTP接口。这样最大化地降低了部署难度pip install requests就能解决全部依赖。如果你非要使用selenium渲染网页也得考虑服务器上的chromium安装与驱动配置复杂度会高一个量级。能用HTTP接口解决的就不要用浏览器。6. 进阶让数据管线自动化运行6.1 每日定时增量任务完成基础爬取和入库之后很自然会想到让脚本每天自动跑一遍。Linux下用crontab是最成熟稳妥的方案。假设脚本文件位于/home/user/finance_etl/daily_update.py可以把执行日志输出到指定文件中30 18 * * 1-5 cd /home/user/finance_etl /usr/bin/python3 daily_update.py logs/etl.log 21这里解释一下时间设置工作日18:30运行是为了在收盘后拉取当天数据。东方财富的数据通常在收盘之后很快就完整了18:30再跑基本不会缺数据。而Tushare的接口可能需要等当日数据复核完成一般晚上也能拿到完整数据。日志很重要。我比较推荐把关键步骤都打上日志包括拉取了多少条数据、耗时多久、是否有接口报错。排查问题时没有日志等于大海捞针。6.2 数据校验与异常重试自动化的另一个重点是失败重试。网络请求本质上是不可靠的偶发超时、连接重置、响应内容为空都是正常现象。脚本必须具备失败自动重试的能力。写一个简单的重试装饰器import time import functools import requests def retry(max_tries3, delay2): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_tries): try: return func(*args, **kwargs) except (requests.RequestException, KeyError, ValueError) as e: if attempt max_tries - 1: raise time.sleep(delay * (attempt 1)) # 指数退避 return None return wrapper return decorator retry(max_tries3, delay2) def fetch_eastmoney_json(url, params): resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() data resp.json() if data.get(data) is None: raise ValueError(Empty data field, retry later) return data重试间隔用delay * (attempt 1)也就是第一次失败后等2秒第二次失败后等4秒相当于一个轻量的指数退避。这个策略比固定间隔要好既能快速恢复又能避免加重服务端压力。6.3 结果监控与报警数据采集自动化之后无人值守状态下最怕的就是静默失败。脚本跑了但数据没更新过了两周你才发现那整个分析结论可能都是基于过期数据的。最朴素的方案是在每日任务最后发送一条通知消息。邮件、企业微信机器人、钉钉机器人或者Server酱都可以。我习惯用企业微信机器人配置简单支持Markdown消息。比如在脚本最后增加def send_notify(text): # 企业微信机器人webhook webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key payload { msgtype: text, text: {content: text} } requests.post(webhook_url, jsonpayload) send_notify(f每日数据更新完成新增记录{new_count}条) } 这样每天定时任务跑完你手机上就能收到一条更新通知。如果某天没收到通知大概率就是脚本出问题了可以及时排查。 ## 7. 选择适合自己的数据路径 最后说说我对这两个数据源搭配使用的个人体会。 Tushare适合作为数据基建层。它数据结构规整、接口稳定、字段含义明确尤其适合需要长期历史数据的回测研究。你不需要每次重新拉一遍几年的数据一次性入库之后所有策略代码都从数据库读效率高而且可复现。 东方财富公开接口适合作为日内补充层。它的实时性非常好盘中行情、资金流向、板块热度这类数据是Tushare免费档位很难提供的。但因为是公开接口稳定性相对弱一些必须做好重试和数据校验并且要接受接口随时可能调整的现实。 两者结合基本能应对个人投资者的数据需求。我在实际项目中Tushare负责建设历史行情库和基础财务库东方财富负责实时行情监控和资金流分析。每天盘后跑一次增量任务把两者数据汇总到统一数据库第二天直接在此基础上做分析和策略回测。 如果你刚开始做金融数据的爬虫项目不建议一上来就把架构搞得特别复杂。先从Tushare的日线行情入手跑通注册—调用接口—入库这条链路再逐步加东方财富的实时数据源。先跑起来再优化这是最务实的路线。
Tushare与东方财富接口实战：从Python爬虫到金融数据入库全流程
不想一上来就聊技术细节先说说背景。很多做量化研究或者金融数据分析的朋友起步阶段最头疼的不是策略怎么写、模型怎么跑而是数据从哪来。市面上商业数据库贵得离谱手工复制粘贴又效率太低这时候Python爬虫几乎成了唯一现实的选择。但金融数据又有点特殊一方面涉及合规问题不能乱抓另一方面数据质量直接决定分析结果宁可慢一点也要干净。这篇文章聚焦两个主力数据源Tushare和东方财富公开接口。前者适合拿规范的结构化金融数据后者适合做实时行情、资金流向、公告信息等公开数据采集。我把从账号准备、接口调用到数据入库的完整链路串一遍重点讲清楚每个环节为什么这样做以及踩过的坑。1. 金融数据获取的双轨思路1.1 为什么Tushare是合规首选先说Tushare。做量化入门或者学术研究Tushare Pro几乎绕不开。它提供股票行情、财务报表、宏观经济、行业分类、基金净值、期货数据等数百个接口数据经过清洗和复权处理直接拉过来就能用。最关键的原因是合规。Tushare要求用户注册、实名认证、申请token数据使用有明确的用户协议和频率限制。相比直接爬取其他网站这种模式在合规层面要稳得多。尤其是打算写论文、做正式报告甚至商业项目用Tushare的数据源在合规风险上要小很多。另外一个优势是标准化。Tushare返回的是规整的DataFrame格式字段命名统一日期格式规范不需要做太多预处理。比如日线行情接口返回字段包括ts_code、trade_date、open、high、low、close、vol、amount等几乎就是量化模型需要的原始格式。相比之下网页上抓来的数据字段名五花八门清洗成本不低。1.2 东方财富公开数据的不可替代性那为什么还要爬东方财富因为Tushare的很多接口有积分门槛。实时盘中数据、资金流向、龙虎榜明细、股票弹幕舆情这些高频数据要么积分要求很高要么干脆没有。而东方财富作为公开财经网站提供了大量公开的浏览器API接口这些接口不需要登录、不需要付费数据更新还非常快。比如个股资金流向数据东方财富的数据接口返回当天的主力净流入、超大单净流入、中单净流入、小单净流入等维度。这种数据对短线交易和资金面分析非常有用Tushare免费的积分档位不一定能拿到。再比如盘中实时分时数据东方财富的接口几乎是无延迟的做盘中监控非常顺手。还有一个优势是覆盖面。东方财富的接口涵盖了股票、基金、债券、期货、外汇、宏观指标等众多类型一个接口格式摸清之后套用参数规则就能扩展到整个站点体系。所谓一次分析、全家复用这个特点在实际操作中非常节省时间。所以我的习惯是Tushare负责基础数据和低频数据的合规获取东方财富公开接口负责高频、实时、网站特有的数据补充。两者结合基本能覆盖从宏观到个股的绝大多数分析场景。2. Tushare从注册到数据落地的完整流程2.1 注册与token管理Tushare Pro的注册流程不复杂。打开官网用手机号注册然后在个人主页找到接口TOKEN页面复制属于自己的token字符串。这里有个细节token是用户身份的唯一凭证所有的API调用都需要带上这个token一旦泄露别人就能消耗你的权限和积分。我的做法是把它写到环境变量里不要硬编码在代码中尤其是准备发到GitHub上的项目。setup如下import os import tushare as ts from dotenv import load_dotenv load_dotenv() # 加载.env文件 TOKEN os.getenv(TUSHARE_TOKEN) ts.set_token(TOKEN) pro ts.pro_api()有人图省事直接写token明文我强烈不建议。换成环境变量或配置文件多花不了半分钟安全等级提升一大截。2.2 核心接口的调用与参数细节Tushare的接口体系比较统一核心是pro.query()或者直接调用pro.xxx()方法。以股票日线行情为例import tushare as ts import datetime pro ts.pro_api() # 获取某个交易日所有股票的日线行情 df pro.daily( trade_date20250110, fieldsts_code,trade_date,open,high,low,close,pre_close,change,pct_chg,vol,amount ) print(df.head())这个接口的参数设计有个细节要注意trade_date和ts_code至少要传一个。传trade_date会返回当天全市场的行情数据传ts_code返回单只股票的历史行情。两个都不传会报错两个都传则以股票为准。另外Tushare的很多接口有积分门槛。日线行情接口的基础积分很低基本注册之后就能用但像moneyflow资金流向、cyq_perf筹码分布这类接口需要更高的积分。积分不够怎么办两个途径一是完善个人资料、使用平台功能攒积分二是充值获得高权限账号。不要想着用抓包绕过平台的服务端校验做得比较到位破解成本远高于充值成本。2.3 交易日历与日期边界问题一个实际使用中很容易踩的坑节假日和停牌日。比如元旦、春节前后你以为当天有数据结果接口返回空DataFrame。这就是典型的数据缺失陷阱——不是代码出错了而是当天根本不是交易日。解决方法是维护一张交易日历表。Tushare提供了trade_cal接口返回每个交易所的日历数据import tushare as ts pro ts.pro_api() cal pro.trade_cal( exchangeSSE, start_date20250101, end_date20251231 ) # is_open 1 表示交易日 trade_days cal[cal[is_open] 1][cal_date].tolist() print(trade_days[:10])有了这张表在写遍历逻辑时就能动态跳过非交易日。否则你写个循环从1月1日跑到12月31日遇到节假日接口返回空数据你的代码要么报错、要么把空数据写进数据库后续分析全乱套。我个人强烈建议第一次建库时先把全年的交易日历抓下来存进数据库后续所有定时任务都先查日历表再决定是否运行。这个设计在后续维护中能省掉大量莫名其妙的报错。2.4 限频与并发控制Tushare对每分钟的调用次数有明确限制不同积分等级对应不同的每分钟最大请求数。免费用户大概每分钟几次到几十次不等如果写个循环批量拉历史数据很容易触发抱歉您每分钟最多访问该接口X次的错误。我踩过的坑有一次拉全市场5年的日线数据用了个for循环一个接一个请求跑了几百次之后突然连续报错。排查半天是触发了接口的分钟级频率限制。后面我改成带time.sleep()的限速版本才稳定下来。import time # 批量获取历史行情注意控制频率 all_dfs [] for date in trade_days: df pro.daily(trade_datedate, fieldsts_code,trade_date,open,high,low,close,pre_close,pct_chg,vol,amount) if df is not None and not df.empty: all_dfs.append(df) time.sleep(0.6) # 保守等待避免触发限频并发也不是越大越好。Tushare服务端有尽量幂等的设计但请求过多会拖慢响应甚至返回错误。数据抓取这个场景稳定性远大于速度宁可多等两分钟也别让自己被封。3. 东方财富公开接口的抓包分析与实战3.1 如何找到接口地址东方财富的数据并非直接渲染在HTML里而是通过JS脚本异步请求后端API获取JSON数据。所以要爬东方财富核心不在于写HTML解析器而在于找到这些公开API地址。打开浏览器开发者工具F12切到Network网络面板然后在页面上点击查看个股行情或资金流向就能看到大量XHR请求。过滤掉图片、CSS、字体文件剩下的就是JSON数据接口。以股票资金流向为例列表页请求的地址长这样https://push2.eastmoney.com/api/qt/stock/fflow/kline/get?lmt0klt101fields1f1,f2,f3,f7fields2f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65secid1.600519secid20.600519这个接口返回的是某只股票的历史资金流向数据数据结构是JSON字段非常规整。f51是日期f52~f56是不同类型资金的净流入额f57~f61是净流入占比等。解析起来比网页HTML简单太多。3.2 secid参数的两段式结构东方财富接口中几乎所有行情接口都要带secid参数它决定了请求的是哪个市场、哪只证券。这个参数的结构是市场代码.证券代码。证券市场代码有一定的约定上海A股是1深圳A股是0北京A股是0还是0指数是1或0按市场区分。例如贵州茅台600519在上交所secid1.600519平安银行000001在深交所secid0.000001。第二个参数secid2在部分接口中会出现通常用于指定对比对象。比如资金流向接口中secid20.600519可能指的是深市前缀映射的同一代码具体含义要看不同接口的约定。实操时最稳的办法用浏览器开发者工具直接复制完整的请求URL改代码值测试确认接口能返回预期数据后再封装。这里有个容易搞混的地方沪深市场的代码前缀规则和我们在Tushare里看到的ts_code不同。Tushare的600519.SH是字符串后缀东方财富的1.600519是整数点整数。两个体系之间转换时要写个映射函数别搞混了。def tushare_ts_to_eastmoney_secid(ts_code: str) - str: code, market ts_code.split(.) if market SH: return f1.{code} elif market SZ: return f0.{code} elif market BJ: return f0.{code} else: raise ValueError(fUnknown market: {market})这个映射函数我在实际项目中一直在用非常省心。3.3 行情快照接口与分时数据除了资金流向东方财富的行情快照接口也很常用。比如获取个股的最新价、涨跌幅、成交量、换手率、市盈率等实时数据请求这个地址https://push2.eastmoney.com/api/qt/stock/get?secid1.600519fieldsf43,f44,f45,f46,f47,f48,f50,f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65这些字段代码的含义各不相同f43是最新价f44是最高f45是最低f46是开盘f47是成交量f48是成交额f50是量比f51是涨停价f52是跌停价f53是静态市盈率f54是动态市盈率等等。这里有个细节f43等价格类字段返回的是整数比如最新价105900对应的实际价格是105.90不对——东方财富很多接口的价格字段单位是分需要除以100换算成元。但这个规则并不统一有的字段本来就是元有的字段是分。我在实际项目中总结了一套校验方法拿接口返回的f43值和你看到的页面价格对比如果差100倍就整除100如果一致就保持原样。这个校准步骤必须做一次不然数据分析全是错的。比如600519当日实时价格import requests import json params { secid: 1.600519, fields: f43,f44,f45,f46,f47,f48,f50,f57,f58,f59, } resp requests.get( https://push2.eastmoney.com/api/qt/stock/get, paramsparams, headers{User-Agent: Mozilla/5.0} ) data resp.json()[data] if data: print(json.dumps(data, ensure_asciiFalse, indent2))返回结果大致长这样实际值可能有变化{f43: 152601, f44: 153800, f45: 152000, f46: 153200, ...}此时f43 152601如果页面显示1526.01元那说明单位确实是分需要/100。我拿过贵州茅台的实时数据做过对比确认这个数值需要除以100。3.4 公开接口中的字段映射与数据清洗东方财富接口的字段名都是f51这种纯数字代号可读性很差。在写存储逻辑时建议建立一个字段映射表把数字代号映射成语义化的字段名。EASTMONEY_FIELD_MAP { f51: trade_date, f52: main_net_inflow, f53: small_net_inflow, f54: medium_net_inflow, f55: large_net_inflow, f56: super_large_net_inflow, # ... 按需补充 }有了映射表后续分析代码直接读列名不用去记每个数字的含义。这个映射关系也建议以JSON或Python常量形式独立存放接口变动时只改映射表不改业务逻辑。4. 数据持久化SQLAlchemy构架数据存储4.1 为什么选择SQLAlchemy爬下来的数据总要存起来。早期我图简单直接存CSV文件但很快发现三个问题一是文件越来越多难以管理二是增量数据要全量重写效率极低三是后续用SQL查询时得加载全部文件再过滤内存吃不消。后来切换到SQLite单文件数据库轻量方便但并发写入时容易锁库。再后来用PostgreSQL功能强大但本地部署需要额外维护一个服务。如果你只是在单机做数据分析SQLite搭配SQLAlchemy是一个非常平衡的方案如果要把数据提供给多个服务共享建议用MySQL或者PostgreSQL。代码层面用SQLAlchemy抽象之后切换数据库几乎不用改业务代码。选择SQLAlchemy的理由很直接ORM映射可以把DataFrame的行列结构直接对应到数据表字段不用手写拼接SQL内置连接池管理不用自己处理连接释放支持自动建表、字段变更迁移。以日线行情为例定义模型from sqlalchemy import Column, Integer, String, Float, Date from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class StockDaily(Base): __tablename__ stock_daily id Column(Integer, primary_keyTrue, autoincrementTrue) ts_code Column(String(16), indexTrue) trade_date Column(Date, indexTrue) open Column(Float) high Column(Float) low Column(Float) close Column(Float) pre_close Column(Float) change Column(Float) pct_chg Column(Float) vol Column(Float) amount Column(Float) def __repr__(self): return fStockDaily ts_code{self.ts_code} date{self.trade_date}4.2 DataFrame批量写入的高效方式Tushare返回的是DataFrameSQLAlchemy ORM要逐条session.add再commit数据量一大就非常慢。更高效的方式是使用df.to_sql()配合SQLAlchemy的engine一次性批量写入。import pandas as pd from sqlalchemy import create_engine engine create_engine(sqlite:///stock_data.db) # 将DataFrame写入数据库 df.to_sql( namestock_daily, conengine, if_existsappend, indexFalse )to_sql内部执行的是批量插入速度远快于逐条add。但有几个细节要注意第一if_existsappend表示数据追加前提是表结构已经存在。第一次写入时可以先if_existsreplace建好表之后改成append。第二to_sql默认对datetime列要求是Python的datetime对象不要传字符串否则可能报错。Tushare返回的trade_date本身是字符串格式如20250110最好先转成日期类型df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d).dt.date第三to_sql不支持UPSERT存在则更新不存在则插入逻辑重复运行会插入重复数据。解决方法是配合数据库唯一索引或者每次写入前做去重。我一般是在ts_code trade_date上建一个唯一索引再配合批量插入前后的DELETE语句实现幂等写入。对于东方财富接口返回的数据同样可以先标准化字段名、转换数据类型再走to_sql入库。4.3 增量更新策略爬虫类数据更新最好的策略是增量而非全量。核心逻辑先查询数据库中最大的日期再从这个日期往后拉取数据。这样既节省请求时间也减少数据库冗余写入。from sqlalchemy import create_engine, text engine create_engine(sqlite:///stock_data.db) with engine.connect() as conn: result conn.execute(text( SELECT MAX(trade_date) FROM stock_daily WHERE ts_code :code ), {code: 600519.SH}) max_date result.scalar_one_or_none() if max_date is None: start_date 20200101 else: start_date max_date.strftime(%Y%m%d) print(f增量拉取起始日期{start_date})这个逻辑在定时任务里至关重要。每天晚上跑一次增量脚本把当天新增数据写入数据库持久化为历史库分析时直接查库即可不必每次都重新爬一遍。5. 实战中的常见问题与反爬策略5.1 请求频率礼貌爬虫的生存法则东方财富的公开接口虽然没有显式的登录限制但对请求频率仍然有隐形的管控。短时间内高并发请求很容易触发IP封禁或者请求返回乱码。我踩过一次比较大的坑写了一个多线程加速脚本同时开20个线程拉取全市场实时行情跑了几分钟后接口返回的数据开始错乱部分请求直接超时。原因就是单位时间内的请求量超过服务端的QPS阈值。解决思路是单线程合理延时把并发控制在一个安全的水平import time import requests from concurrent.futures import ThreadPoolExecutor def fetch_one(secid): params {secid: secid, fields: f43,f44,f45,f46,f47,f48} resp requests.get(https://push2.eastmoney.com/api/qt/stock/get, paramsparams, timeout5) time.sleep(0.3) # 每个请求之后等待0.3秒 return resp.json() secids [1.600519, 0.000001, 1.601318] with ThreadPoolExecutor(max_workers3) as executor: results list(executor.map(fetch_one, secids))注意我用的是3个线程而不是10个或20个。实测下来东方财富接口在低并发合理延时的模式下响应最稳定数据完整率接近100%。如果追求极致速度控制在0.2秒延时左右也没问题但超过10个并发就很容易出问题。5.2 数据字段缺失与单位混淆这是新手最容易翻车的地方。东方财富的接口字段数量非常多但并非所有字段在每只股票上都返回有效值。比如新股上市初期部分指标可能为空停牌股票的价格类字段可能返回-或者0。这些异常值如果直接入库会对后续分析造成污染。我的处理方法是在入库前加一个数据质量校验层def clean_row(raw): 对接口JSON原始字段做数据清洗 1. 将字符串类型的-转为None 2. 将数值型字段转换为float失败则置为None 3. 价格类字段按比例换算为元 cleaned {} for key, value in raw.items(): if value - or value is None: cleaned[key] None continue try: float_value float(value) if key in (f43, f44, f45, f46): # 价格字段以分为单位 cleaned[key] round(float_value / 100, 2) else: cleaned[key] float_value except (ValueError, TypeError): cleaned[key] None return cleaned有了清洗层入库数据基本可以保证类型统一、无脏值后续做技术分析时不用反复处理异常数据。另一个容易忽略的细节接口返回的涨跌幅字段单位是百分比数值直接返回比如pct_chg1.25代表1.25%不要额外除100。如果你在清洗层里统一对所有字段除以100就会得到一个错误的结果。所以我每次对接一个新接口第一件事是输出一批样本数据手工核对特殊字段确认单位规则后再写清洗逻辑。5.3 接口变动与失效的应急预案公开的接口终究不是正式开放的开发者接口它的参数和路径可能会调整。我遇到过几次接口地址或字段含义变动的情况排查过程并不轻松。有几个经验第一接口返回的JSON里通常包含错误提示或状态码先把这些信息打日志。不要只记录response.status_code而是把返回的JSON原文都存下来排错时能少走很多弯路。第二定期用一个小脚本监控关键接口的响应结构。比如每周跑一次检查f43字段是否还在、返回数据条数是否正常。一旦出现异常马上发通知提醒人工介入。第三代码中把接口URL和字段映射独立配置。接口变动时只需要改配置文件和映射表不需要动整个项目。这看起来是个很基础的设计但很多项目因为图方便把URL写死在多个模块里最终自食其果。比如东方财富的push2.eastmoney.com域名偶尔会出现波动一些备用域名也可以替换使用。配置化之后切换环境非常方便。5.4 本地调试与线上部署的差异同一个爬虫脚本在本地跑得很欢放到服务器上可能各种报错。最大的差异在于网络环境和Python环境。网络层面服务器如果不在国内访问东方财富接口可能超时或者返回异常。解决方法是部署前先测试连通性必要时使用国内云服务器。另外服务器上的DNS解析也偶有问题可以在requests的params中直接指定Host头或者使用IP直连方案但要小心合规风险建议统一通过域名访问。环境层面线上环境经常没有图形界面也不一定有selenium需要的浏览器。所以我在爬虫设计上一开始就避免依赖浏览器引擎全部通过requests直接请求HTTP接口。这样最大化地降低了部署难度pip install requests就能解决全部依赖。如果你非要使用selenium渲染网页也得考虑服务器上的chromium安装与驱动配置复杂度会高一个量级。能用HTTP接口解决的就不要用浏览器。6. 进阶让数据管线自动化运行6.1 每日定时增量任务完成基础爬取和入库之后很自然会想到让脚本每天自动跑一遍。Linux下用crontab是最成熟稳妥的方案。假设脚本文件位于/home/user/finance_etl/daily_update.py可以把执行日志输出到指定文件中30 18 * * 1-5 cd /home/user/finance_etl /usr/bin/python3 daily_update.py logs/etl.log 21这里解释一下时间设置工作日18:30运行是为了在收盘后拉取当天数据。东方财富的数据通常在收盘之后很快就完整了18:30再跑基本不会缺数据。而Tushare的接口可能需要等当日数据复核完成一般晚上也能拿到完整数据。日志很重要。我比较推荐把关键步骤都打上日志包括拉取了多少条数据、耗时多久、是否有接口报错。排查问题时没有日志等于大海捞针。6.2 数据校验与异常重试自动化的另一个重点是失败重试。网络请求本质上是不可靠的偶发超时、连接重置、响应内容为空都是正常现象。脚本必须具备失败自动重试的能力。写一个简单的重试装饰器import time import functools import requests def retry(max_tries3, delay2): def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_tries): try: return func(*args, **kwargs) except (requests.RequestException, KeyError, ValueError) as e: if attempt max_tries - 1: raise time.sleep(delay * (attempt 1)) # 指数退避 return None return wrapper return decorator retry(max_tries3, delay2) def fetch_eastmoney_json(url, params): resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() data resp.json() if data.get(data) is None: raise ValueError(Empty data field, retry later) return data重试间隔用delay * (attempt 1)也就是第一次失败后等2秒第二次失败后等4秒相当于一个轻量的指数退避。这个策略比固定间隔要好既能快速恢复又能避免加重服务端压力。6.3 结果监控与报警数据采集自动化之后无人值守状态下最怕的就是静默失败。脚本跑了但数据没更新过了两周你才发现那整个分析结论可能都是基于过期数据的。最朴素的方案是在每日任务最后发送一条通知消息。邮件、企业微信机器人、钉钉机器人或者Server酱都可以。我习惯用企业微信机器人配置简单支持Markdown消息。比如在脚本最后增加def send_notify(text): # 企业微信机器人webhook webhook_url https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key payload { msgtype: text, text: {content: text} } requests.post(webhook_url, jsonpayload) send_notify(f每日数据更新完成新增记录{new_count}条) } 这样每天定时任务跑完你手机上就能收到一条更新通知。如果某天没收到通知大概率就是脚本出问题了可以及时排查。 ## 7. 选择适合自己的数据路径 最后说说我对这两个数据源搭配使用的个人体会。 Tushare适合作为数据基建层。它数据结构规整、接口稳定、字段含义明确尤其适合需要长期历史数据的回测研究。你不需要每次重新拉一遍几年的数据一次性入库之后所有策略代码都从数据库读效率高而且可复现。 东方财富公开接口适合作为日内补充层。它的实时性非常好盘中行情、资金流向、板块热度这类数据是Tushare免费档位很难提供的。但因为是公开接口稳定性相对弱一些必须做好重试和数据校验并且要接受接口随时可能调整的现实。 两者结合基本能应对个人投资者的数据需求。我在实际项目中Tushare负责建设历史行情库和基础财务库东方财富负责实时行情监控和资金流分析。每天盘后跑一次增量任务把两者数据汇总到统一数据库第二天直接在此基础上做分析和策略回测。 如果你刚开始做金融数据的爬虫项目不建议一上来就把架构搞得特别复杂。先从Tushare的日线行情入手跑通注册—调用接口—入库这条链路再逐步加东方财富的实时数据源。先跑起来再优化这是最务实的路线。
