最近有做个股研究的朋友问我,能不能写一份直接从东方财富抓上市公司纯利润(也就是净利润)的 Python 代码。他不是程序员,只是想要一份能自动跑的数据底稿,每天别手工复制粘贴。这个需求其实特别典型,量化交易、基本面筛选、财报跟踪都能用上。我把整套实现思路、关键接口参数、完整代码和踩过的坑一起整理出来,既方便他按需取用,也给想折腾爬虫的同学一份可以直接抄作业的参考。
先说结论:东财自己有一整套对外的数据接口,返回的是干净的 JSON,根本不需要去解析网页 HTML。只要知道接口地址、参数怎么拼、字段对应关系,就能批量拿到净利润数据。这篇文章我会从零把方案设计讲清楚,然后把能直接运行的代码贴出来,最后集中说明那些文档里不会写的翻车记录。
1. 项目全貌与方案选型
1.1 这个需求到底在解决什么问题
“从东方财富网站爬取纯利润代码”,听起来很普通,但仔细拆一下,其实包含了三层诉求。
第一层是数据获取:上市公司财报里披露的净利润,散落在各个页面里,人工翻很累。第二层是结构化:拿到的数据应该是一张规规矩矩的表,包含股票代码、股票名称、报告期、净利润、扣非净利润、营收等字段,而不是浏览器里那种花里胡哨的展示。第三层是可持续更新:年度报告、季度报告都在固定时间发布,代码要能反复跑,新增数据只需增量更新,不能每次全量重抓。
所以这不是一个简单的爬虫脚本,而是一个半自动化的数据采集方案。目标是用最小成本、最稳定的方式,把东财的财报数据转成本地文件或数据库,方便后续做筛选、画图、回测。
1.2 为什么我不推荐直接抓网页 HTML
很多人第一次想到爬虫,第一反应是打开东财某个财报页面,用 requests 拉 HTML,然后用正则表达式或者 BeautifulSoup 去剥页面里的数字。这条路我不是没走过,实测下来问题很多:
- 页面是 JavaScript 动态渲染的,requests 拿到的 HTML 里根本没有净利润数字,全是 JS 变量和初始化函数;
- 如果用 Selenium 去模拟浏览器,虽然能拿到渲染后的页面,但速度慢、内存占用高,几十只股票还好,全市场五千只股票根本跑不动;
- 页面模板改版是家常便饭,改一个 class 名或标签结构,你的正则是就作废,维护成本极高。
我踩过几次动态渲染的坑之后,彻底改变了思路:与其逆向 HTML,不如直接找页面数据是从哪个接口请求来的。
浏览器里看到的数据,背后一定是某个 JSON 接口在给前端喂数据。只要打开开发者工具,切到 Network 面板,刷新页面就能找到这些数据接口。东财的接口返回结构非常规整,几乎是开箱即用,爬取效率和稳定性都远超 HTML 方案。
1.3 选定两条路:F10 财报接口 vs 数据中心批量接口
东财体系里有两种拿净利润的常见途径,我分别测过。
一种是 F10 资料页背后的接口,地址类似emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/Index,它返回的是经过包装的财务指标数据。但这个接口返回的数据结构嵌套较多,而且不同股票页面请求参数略有差异,处理起来有点脏。
另一种是东财数据中心用的报表接口,也就是data.eastmoney.com页面背后那套datacenter-web.eastmoney.com接口。这套接口最爽的地方在于:它就是一个标准 REST API,直接传报表名、筛选条件、分页参数,返回干净的 JSON 数组。
我最终选定数据中心接口。理由很简单:
- 支持
filter参数做条件过滤,可以直接按股票代码、报告期筛选; - 支持分页,一次最多拿 500 条,全市场业绩报表也能轻松翻页;
- 返回字段是字典格式,
json.loads之后直接通过 key 取值,顺手丢进 pandas 即可; - 不需要登录、不需要 cookie,请求头带一个 User-Agent 和 Referer 就能跑。
2. 东方财富数据接口的核心机制
2.1 接口地址和返回格式
我实际使用的接口地址是:
https://datacenter-web.eastmoney.com/api/data/v1/get这个接口本身不关心你要什么报表,而是靠参数里的reportName来指定报告类型。例如业绩快报用RPT_LICO_FN_CPD,业绩报表用RPT_LICO_FN_CPD,有时候不同的页面背后对应不同的 name。我抓纯利润时用的是业绩报表 ReportName,返回结构大概是这样的:
{ "version": 1, "result": { "pages": 3, "count": 132, "data": [ { "SECURITY_CODE": "600519", "SECURITY_NAME_ABBR": "贵州茅台", "REPORT_DATE": "2024-09-30T00:00:00", "TOTAL_OPERATE_INCOME": 123456789.0, "PARENT_NETPROFIT": 567890.0, "WEIGHTAVG_ROE": 12.34 } ] } }result.data就是数据行数组,result.pages是总页数,result.count是总条数。有了总数和页数,写分页循环就非常方便了。
2.2 参数与字段解析:filter、reportName、sortColumns 的含义
这个接口的请求参数有一堆,但真正必须掌握的就几个。我用一张表说明。
| 参数名 | 作用 | 示例 |
|---|---|---|
reportName | 指定报表类型 | RPT_LICO_FN_CPD(业绩报表) |
columns | 返回哪些字段,全字段用ALL | ALL |
filter | 筛选条件,格式类似 SQL where | (SECURITY_CODE="600519") |
sortColumns | 排序字段 | REPORT_DATE |
sortTypes | 排序方式,-1 表示倒序 | -1 |
pageNumber | 页码 | 1 |
pageSize | 每页条数,最大 500 | 50 |
特别值得说的是filter参数。它的写法是(字段="值"),多条件用AND、OR连接,单值必须加双引号。比如我想查贵州茅台 2024 年以后的数据,filter 可以写成:
(SECURITY_CODE="600519")(REPORT_DATE>='2024-01-01')在东财接口里,多个括号条件实际上是 AND 关系,不需要再写 AND 关键字。这个细节我一开始不知道,导致返回 result 为空,排查了很久。
另一个坑是REPORT_DATE的类型。接口返回的是带T00:00:00的日期字符串,如果直接拿它做排序或者比较,没问题;但如果要存数据库,记得先转成YYYY-MM-DD格式,否则后续 SQL 查询会出幺蛾子。
2.3 净利润字段的单位陷阱
用户要的是“纯利润”,财报上通常叫“净利润”。东财数据中心接口里,净利润有好几个近亲字段,千万不能搞混:
NETPROFIT:净利润整体,包含少数股东损益;PARENT_NETPROFIT:归属于母公司所有者的净利润,也就是经常听说的“归母净利润”;DEDUCT_PARENT_NETPROFIT:扣除非经常性损益后的归母净利润,俗称“扣非净利润”。
这三个字段在业绩报表RPT_LICO_FN_CPD里面都出现。做个股基本面分析,最常用的其实是PARENT_NETPROFIT,因为它是真正属于上市公司股东的利润。如果标题里的“纯利润”指的是扣非净利润,那就选DEDUCT_PARENT_NETPROFIT。
更坑的是单位。数据中心接口金额字段的单位是“元”。但你在网页上看到的数字经常被格式化成“亿”。如果你直接把接口返回的值当成页面展示值,会发现整整差了一亿倍。我头一次拿数据做回测,算出的收益率离谱,查了半天才发现是把 350 亿看成了 350 元。
所以代码里要做一步处理:要么统一除以 100000000 转成“亿”,要么保留元精确值,后续计算时统一口径。我个人建议数据落地时保留“元”为单位的原始值,展示时再转亿,避免精度损失。
3. 完整实操:从单只股票到全市场扫描
3.1 第一步:用 Python 请求东财财务接口
先解决最简单的场景:给定一个股票代码,返回它的历年净利润数据。
import requests import pandas as pd headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://data.eastmoney.com/", "Accept": "application/json, text/plain, */*", } def get_profit_by_code(code: str) -> pd.DataFrame: url = "https://datacenter-web.eastmoney.com/api/data/v1/get" params = { "sortColumns": "REPORT_DATE", "sortTypes": "-1", "pageSize": "50", "pageNumber": "1", "reportName": "RPT_LICO_FN_CPD", "columns": "ALL", "filter": f'(SECURITY_CODE="{code}")', } resp = requests.get(url, params=params, headers=headers, timeout=10) resp.raise_for_status() payload = resp.json() if payload.get("result") is None: return pd.DataFrame() rows = payload["result"]["data"] if not rows: return pd.DataFrame() df = pd.DataFrame(rows) # 统一只留下常用列,减少噪音 keep_cols = [ "SECURITY_CODE", "SECURITY_NAME_ABBR", "REPORT_DATE", "TOTAL_OPERATE_INCOME", "PARENT_NETPROFIT", "DEDUCT_PARENT_NETPROFIT", "WEIGHTAVG_ROE", ] df = df[[c for c in keep_cols if c in df.columns]] df["REPORT_DATE"] = pd.to_datetime(df["REPORT_DATE"]).dt.strftime("%Y-%m-%d") df["PARENT_NETPROFIT_YI"] = df["PARENT_NETPROFIT"] / 1e8 return df # 测试:获取贵州茅台的净利润数据 profit_df = get_profit_by_code("600519") print(profit_df.head(10))这段代码跑起来很快,单只股票几十条季度数据一秒内就能返回。注意我把单位是元的PARENT_NETPROFIT转成了亿,新增一列PARENT_NETPROFIT_YI,方便直接看。
3.2 第二步:解析 JSON 并转成干净的表格
上面的代码里其实已经包含了 JSON 解析过程。resp.json()会直接返回 Python 字典,然后我通过payload["result"]["data"]取数据行。
有几个细节我需要单独拿出来说。
第一,接口偶尔会返回result为 null。出现这种情况通常不是被反爬,而是你的filter条件没匹配到任何数据,比如代码写错了,或者该股票没有披露对应报告期的业绩。如果不对result做一次判空,直接访问["data"]会抛 TypeError。我在代码里加了if payload.get("result") is None,就是为了避免程序中途崩溃。
第二,字段不一定全都有。新股、停牌股、数据不完整的股票,可能缺少WEIGHTAVG_ROE或DEDUCT_PARENT_NETPROFIT。如果我用固定列索引去取值,容易 KeyError。我的做法是用[c for c in keep_cols if c in df.columns]动态过滤列名,只留下存在的字段。丢了几个字段没关系,核心的净利润和日期字段在绝大多数情况下都在。
第三,报表的日期时间是带时区格式的,直接存 CSV 没问题,但存数据库前必须pd.to_datetime规范化。上面代码里已经做了处理。
3.3 第三步:批量抓取全部 A 股净利润
单只股票能跑通,接下来就是批量扫描。要把全市场五千多只股票的净利润都抓下来,先得拿到一份股票代码清单。
东财行情接口也可以直接给代码清单,我写了一个获取全部 A 股代码的函数:
def get_stock_list() -> list: url = "https://push2.eastmoney.com/api/qt/clist/get" params = { "pn": 1, "pz": 500, "po": 1, "np": 1, "fltt": 2, "invt": 2, "fid": "f3", "fs": "m:0 t:6,m:0 t:80,m:1 t:2,m:1 t:23", "fields": "f12,f14", } resp = requests.get(url, params=params, headers=headers, timeout=10) rows = resp.json()["data"]["diff"] return [(row["f12"], row["f14"]) for row in rows]fs参数含义是市场范围,m:0 t:6代表深市主板,m:0 t:80代表创业板,m:1 t:2代表沪市主板,m:1 t:23代表科创板,分开定义的好处是想去掉某类股票时可以直接删对应分组。这里返回的f12是六位股票代码,f14是股票名称。
拿到代码清单后,用单线程循环去抓是可行的,但速度太慢。五千只股票,每只停个 0.5 秒都要四十分钟。我建议用线程池控制并发数量,同时控制请求频率。
from concurrent.futures import ThreadPoolExecutor import time def fetch_one(code_name): code, name = code_name df = get_profit_by_code(code) if df.empty: return None df.insert(0, "SECURITY_NAME", name) time.sleep(0.3) return df stock_list = get_stock_list() all_frames = [] with ThreadPoolExecutor(max_workers=4) as pool: results = pool.map(fetch_one, stock_list) for frame in results: if frame is not None: all_frames.append(frame) final_df = pd.concat(all_frames, ignore_index=True) final_df.to_csv("a_share_profit.csv", index=False, encoding="utf-8-sig")线程数控制在 4 到 6 就够了。我曾经开 20 个线程跑,三分钟就被限流,之后请求开始超时,所以这个参数宁可保守一点。
3.4 第四步:增量更新与入库策略
全量抓取完之后,“可持续更新”就是下一个目标。财报不是每天变,季度报告只在 1 月、4 月、8 月、10 月集中披露,所以增量更新逻辑可以简化为两步:
- 本地保存一份
last_fetch_date.txt,记录上次抓取的日期; - 每次运行只抓取
REPORT_DATE大于这个日期的数据,或者干脆只抓最新一个报告期的数据。
如果抓的是RPT_LICO_FN_CPD,半年报和年报更新时间相对固定,年报集中在次年 4 月底之前披露完毕。更稳妥的做法是定期全量抓一遍,然后drop_duplicates根据 ReportDate 去重。原因很简单:东财接口并不会有历史数据修改通知,全量重抓后去重,逻辑最简单,不容易漏数据。
至于存储,小数据量直接 CSV 最顺手。需要做复杂查询时,再考虑把结果写进 SQLite:
import sqlite3 conn = sqlite3.connect("stock_profit.db") final_df.to_sql("profit", conn, if_exists="replace", index=False) conn.close()以后查询只用SELECT * FROM profit WHERE SECURITY_CODE='600519',比每次重新爬网页快得多。
4. 爬取过程中的翻车记录与排查心得
4.1 问题一:接口返回 result 为 null
这是我碰到最多的一个问题。明明浏览器里能看到数据,代码请求却拿不到。排查思路如下:
- 先打印完整的请求 URL,复制到浏览器打开,看浏览器能不能返回 JSON;
- 检查
filter格式,必须是(SECURITY_CODE="600519")这种括号加引号的写法,不能用单引号; - 检查报告期,新股在没有发布过季报之前,业绩报表中确实没有数据;
- 检查
reportName,不同页面背后的报表名不一样,如果页面类型选错,同样样 null。
我最初犯的错误是拿业绩快报的 reportName 去查业绩报表的数据,字段对不上,result 直接为空。后来我用东财页面 Network 面板逐个对比,才确定要找的是RPT_LICO_FN_CPD。
4.2 问题二:净利润字段数值异常偏大或偏小
这个问题的根因基本逃不开单位。接口返回单位是元,网页展示单位往往是亿,差别是一亿倍。
建议在数据处理阶段就统一单位,不要等到分析时才想起换算。我在第三部分代码里专门增加了PARENT_NETPROFIT_YI列,转成以亿为单位的格式,输出时好分辨。
另外还要注意:净利润是负值时,东财接口照样返回负数。比如亏损企业的PARENT_NETPROFIT是-321000000.0,这代表亏损 3.21 亿。如果你的筛选条件里有“净利润大于某值”,负数会被正确地过滤掉,不用特殊处理。
4.3 问题三:请求被限流,连接超时
东财接口整体算友好,但短时间高并发访问还是会被限制。我做全市场扫描时,线程池开到 20 个,跑了几分钟,突然大量请求卡住,返回 504 或者直接 Connection Timeout。
解决办法有几个:
- 把线程数降为 4;
- 每次请求之间
time.sleep(0.3)到time.sleep(1); - 在请求函数内部套一层重试逻辑,失败两次后自动跳过;
- 严格控制每天请求总量,不要全天候轰炸。
我踩过限流的坑后,现在写法是:先小批量测试,确认接口、参数都没问题,再启动全市场扫描。如果中途有股票失败,记录下来,全部跑完之后单独重试失败列表。
4.4 实操心得小结
这套方案我用下来稳定性很高,但有几个习惯必须养成:
- 抓下来的数据第一时间另存原始文件,不要直接覆盖原表。财报数据有时候会被东财修正,留一个原始备份有利于回溯;
- 写代码时一定加
timeout,requests 默认不设超时,一旦网络抖动,线程会一直挂住,整个程序像死机了一样; - 数据落地后用
df.info()看看每列空值数量,净利润列如果大量为空,先别急着分析,回头检查是不是报表名选错了; - 每次跑完脚本,把日期的最大值和实际报告期对齐,比如 4 月底抓年报数据,就可能存在部分公司还没披露的情况,数据缺失是正常的,不是 bug。
如果你只是关注几只自选股的净利润,用get_profit_by_code就够了,改一下股票代码,几秒钟出结果。如果你要搭建自己的量化选股底稿,把全市场扫描和 SQLite 存储结合起来,后面再做条件筛选、横截面对比都会顺手很多。这套代码同样可以扩展去抓营收、ROE、资产负债率等字段,核心逻辑不用换,改一下reportName和保留字段名称就行。