ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

Python财务数据采集实战:Selenium与Requests混合爬取东方财富报表

Python财务数据采集实战:Selenium与Requests混合爬取东方财富报表 简介资源围绕东方财富网上市公司财务报表数据采集展开提供Selenium与Requests两种技术方案的完整实现代码适合网络爬虫初学者及金融数据分析人员参考学习。方案A基于Selenium需预先配置浏览器驱动采集速率相对较低方案B利用Requests直接发送HTTP请求数据处理效率提升两个数量级被确定为主要实施方案。程序支持配置会计年度2018-2023、财季标识、报表分类及采集页码区间自动执行数据采集并输出CSV结构化文件至指定目录覆盖资产负债表、利润表、现金流量表、业绩报表等多类财务数据。包内共19个文件以Python脚本、CSV数据文件为主附README说明文档及许可证整体约37.96MB。已有46人学习使用适合希望快速掌握企业财报抓取思路并动手实践的开发者。1. 为什么把 Selenium 和 Requests 混着用而不是只选一个手头放着东方财富网这个目标很多第一次做财务报表爬虫的人会纠结一个问题到底是无脑上 Selenium 模拟浏览器还是直接用 Requests 怼 HTTP 接口我的答案是两个都要而且分工必须明确。东方财富网的数据分布有两种典型形态一种是 F10 页面里的“主营构成”“公司资料”“核心题材”这些数据由 JavaScript 动态渲染直接 Requests 拿 HTML 只能看到空壳另一种是“业绩报表”“业绩快报”“业绩预告”这类数据页面底层的异步接口返回的是 JSON只要模拟好请求头Requests 一把就能拿干净。还有一个现实问题Selenium 跑起来吃内存、吃带宽一个 Chrome 实例轻松占掉几百 MB你要是对几百家公司逐个开浏览器机器基本废了。Requests 的单线程开销可以忽略但对反爬的容忍度低频率一高就撞上 429 too many requests。所以常见做法是“能用 Requests 拿的绝不开浏览器必须用 Selenium 的才开而且最小化使用”。这个标题里的“实践”二字本质就是在这两条路径之间找平衡点。本文覆盖的读者是已经会写 Python 基础脚本、但没正经做过金融数据采集的工程人员也照顾到被 429 和动态渲染卡住的老手。2. 分清东方财富网的两类数据确定 Requests 的 URL 与参数2.1 从页面请求里找到真正的数据接口东方财富网的财务数据页面绝大多数情况下地址栏里的 URL 并不是数据真正的来源。常见做法是打开浏览器开发者工具切到 Network 选项卡然后点击页面里的“下一页”或者切换报告期观察哪些请求返回了 JSON 而不是 HTML。你会看到类似https://datacenter-web.eastmoney.com/api/data/v1/get这样的接口地址。这个地址就是 Requests 要打的对象参数里通常带着sortColumns、sortTypes、pageSize、pageNumber、reportName等键值。以业绩报表为例请求参数里reportNameRPT_LICO_FN_CPD这个值表示“业绩报表”这个数据集的编号columnsALL表示取该数据集全部字段filter(SECURITY_CODE600519)这种写法是对股票代码做筛选pageNumber1和pageSize50控制分页。这里最关键的认知是Requests 脚本不是去模拟页面点击而是直接模拟这个 GET 请求。import requests session requests.Session() session.headers.update({ User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Referer: https://data.eastmoney.com/, }) api_url https://datacenter-web.eastmoney.com/api/data/v1/get params { reportName: RPT_LICO_FN_CPD, columns: ALL, filter: (SECURITY_CODE600519), pageNumber: 1, pageSize: 50, sortColumns: UPDATE_DATE, sortTypes: -1, source: WEB, client: WEB, } resp session.get(api_url, paramsparams, timeout10) data resp.json() print(data[result][pages]) # 先看看总页数这里的逻辑核心是params字典直接传给 Requests由它完成 URL 编码不需要手动拼接查询字符串。sortColumns与sortTypes的组合影响去重逻辑建议固定按更新时间倒序保证最新一份报告排在前面。sourceWEB和clientWEB是东方财富接口识别调用来源的标志缺失时有概率拿不到数据。返回的 JSON 结构里result.data是具体行记录result.pages是总页数先读出总页数再决定要不要翻页。2.2 设置分页循环、随机延时与指数退避一次只拿 50 条肯定不够A 股五千多家上市公司加上每个报告期的多条记录翻页是必须的。但无脑 for 循环翻页大概率会在第 20 页左右触发风控。核心参数是请求间隔。我一般会在每页请求之间随机睡 1.5 到 3.5 秒同时对result.data做空值判断——很多接口在页数超出实际数据时会返回空 list而不是报错。import time import random from typing import List, Dict def fetch_all_pages(session: requests.Session, params: Dict, max_retries: int 5) - List[Dict]: all_rows [] cur_params params.copy() total_pages 1 for page in range(1, total_pages 1): cur_params[pageNumber] page for attempt in range(max_retries): try: resp session.get(api_url, paramscur_params, timeout10) if resp.status_code 200: break if resp.status_code 429: wait_time 2 ** attempt random.uniform(0, 1) print(f触发 429等待 {wait_time:.2f} 秒后重试) time.sleep(wait_time) except requests.exceptions.RequestException: time.sleep(2) else: print(f第 {page} 页重试耗尽跳过) continue payload resp.json() rows (payload.get(result) or {}).get(data) or [] all_rows.extend(rows) print(f第 {page} 页累计 {len(all_rows)} 条) return all_rows注意for...else的写法for循环正常结束且没有break时else块执行。这里的语义是“重试全部失败则跳过当前页”。2 ** attempt是指数退避第一次重试等 2 秒第二次 4 秒第三次 8 秒对 429 这种限流比固定间隔有效得多。更稳妥的做法是解析响应头里的Retry-After字段如果服务器返回了明确的等待秒数以它为准。2.3 用 POST 还是 GET东方财富接口的差别上面用的是 GET但东方财富的数据中心部分接口也支持 POST。区别在于GET 的查询参数走 URL适合简单筛选POST 的 body 可以承载更复杂的 JSON 条件。财务报表场景里绝大多数公开接口 GET 就够了POST 多用于带权限的后台接口。这里深挖的目的是避免读者照抄代码后遇到接口返回 500 就懵——你可以在开发者工具里看请求方法脚本里保持一致即可。一个小坑是部分接口对Accept头敏感你需要把它设为application/json, text/plain, */*否则网关可能返回一段 HTML 而不是 JSON。如果resp.json()报错先打印resp.text[:200]看一眼是不是被重定向到了登录页。出现这种情况基本是 Cookie 缺失或过期Selenium 那套可以帮你拿到新鲜的 Cookie这块放在下一章讲。3. Selenium 接管动态渲染页面从 HTML 里取 F10 数据3.1 为什么 F10 必须走 SeleniumRequests 拿不到的典型页面是东方财富 F10 里的“主营构成”和“公司资料”。这些页面的数据由前端 JavaScript 异步加载初始 HTML 里只有模板骨架真实数据在XMLHttpRequest回调里才填充进 DOM。你要么去逆向那一堆混淆过的 JS 找出加密参数要么直接让浏览器渲染完再抓。对财务报表数据采集这种需要稳而不是快的任务Selenium 是更划算的选择。Selenium 的本质是 WebDriver 协议它通过驱动浏览器执行真实操作所以页面里任何你能看到的数据它都能拿到。3.2 只开一次浏览器循环遍历股票代码爬 F10 的时候最多见的错误是每只股票都重新webdriver.Chrome()一次然后再quit()。这个操作的耗时在 3 到 5 秒之间五百只股票就是半小时起步。正确做法是启动一个 driver 实例循环内只做get()和页面交互。from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.common.by import By import time options Options() options.add_argument(--headlessnew) # 无头模式不弹浏览器窗口 options.add_argument(--disable-gpu) options.add_argument(--no-sandbox) options.add_argument(--window-size1920,1080) options.add_experimental_option(excludeSwitches, [enable-automation]) driver webdriver.Chrome(optionsoptions) stock_codes [600519, 000858, 300750] base_url https://emweb.securities.eastmoney.com/PC_HSF10/OperationsRequired/Index?typewebcodeSH600519 for code in stock_codes: # 注意不同市场前缀不同SH代表沪市SZ代表深市 prefix SH if code.startswith(6) else SZ url fhttps://emweb.securities.eastmoney.com/PC_HSF10/BusinessAnalysis/Index?typewebcode{prefix}{code} driver.get(url) try: WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.CSS_SELECTOR, table)) ) except Exception: print(f{code} 表格未渲染) continue time.sleep(1) # 等待JS数据填充稳定 html driver.page_source # 后续解析交给 pandas.read_html 或 BeautifulSoupexcludeSwitches里的enable-automation是去掉 Chrome 顶部“正在受到自动测试软件控制”的提示这个开关本身不解决反爬但对稳定性有帮助。headlessnew是 Chrome 新无头模式比旧版的兼容性好能正确渲染更多 Canvas 和 WebGL 内容。presence_of_element_located只检查元素存在不保证数据渲染完所以后面跟一个sleep(1)是务实写法——你可以用更精确的条件比如等某个包含特定文字的元素出现但代价是选择器更脆。3.3 表格解析别用正则硬抠拿到page_source之后建议直接用pandas.read_html把table转成 DataFrame。这个方法内部依赖lxml解析 HTML对标准表格的识别率很高。F10 页面的表格结构相对规整read_html一次能解析出多个表格返回一个 DataFrame 列表你需要根据列名筛选出来想要的那一个。import pandas as pd tables pd.read_html(html) for i, table in enumerate(tables): first_col str(table.iloc[0, 0]) if 主营构成 in table.columns.tolist() or 营业收入 in table.values.tolist()[0]: table.to_csv(ff10_{code}.csv, indexFalse, encodingutf-8-sig) break用utf-8-sig编码是避免 Excel 打开 CSV 时中文乱码。table.values.tolist()[0]拿第一行数据检测关键词比只查列名更稳因为部分表格的第一列名称在数据行而不是表头里。3.4 定位动态下拉框或 TabSelenium 的常见卡点东方财富 F10 页面上常有报告期选择下拉框或 Tab 切换。用 Selenium 操作时最常见的坑是下拉框不是原生select而是divulli自定义组件当你试图Select(element).select_by_visible_text()时会直接报错。正确做法是模拟点击后再按文本匹配。from selenium.webdriver.common.by import By dropdown driver.find_element(By.CSS_SELECTOR, .datacommon-select) dropdown.click() options driver.find_elements(By.CSS_SELECTOR, .datacommon-select-option) for opt in options: if opt.text.strip() 2024-06-30: opt.click() break这里的思路是先点击打开列表再用text属性精确匹配。定位策略用 CSS 类名.datacommon-select是东方财富前端组件通用的命名实际开发时建议先用 XPath 或 CSS 在开发者工具里验证。如果元素是懒加载的点击后需要加WebDriverWait等待选项出现不要直接find_elements否则拿到的可能是空列表。4. 数据清洗合并把 API 和 Selenium 两路的 DataFrame 对齐4.1 统一报告期字段的格式差异财务数据的坑往往不是爬不到而是爬到之后对不上。业绩报表 API 返回的REPORTDATE字段值是类似2024-06-30 00:00:00的时间戳字符串而 F10 页面表格里展示的是2024-06-30。合并之前必须统一格式。df[REPORTDATE] pd.to_datetime(df[REPORTDATE]).dt.strftime(%Y-%m-%d)如果你的 DataFrame 里既有 API 数据又有 Selenium 数据建议各自先清洗再合并而不是合并后再清洗——出了问题好定位是哪一路的数据格式异常。除此之外数字列也有坑API 返回的数字是 float但 F10 页面展示的可能是“1.23亿”这种带单位的字符串需要手动转换。def parse_yi(s): if isinstance(s, (int, float)): return float(s) if 亿 in str(s): return float(str(s).replace(亿, )) * 1e8 if 万 in str(s): return float(str(s).replace(万, )) * 1e4 return float(s)REPORTDATE统一成字符串后后续merge才能对上。数据处理顺序是“先格式统一、后去重、再合并”只要是财务数据这个顺序基本不变。4.2 股票代码对齐注意 6 开头是沪市还是深市上面 Selenium 的 URL 拼接里已经涉及SH600519和SZ000858的区别合并数据时同样要处理。业绩报表 API 的SECURITY_CODE字段只返回纯六位数字但 F10 页面的某个表头或字段可能带交易所前缀。常见做法是新增一列market从股票代码推导。df[market] df[SECURITY_CODE].apply(lambda x: SH if str(x).startswith(6) else SZ) df[sec_code] df[market] df[SECURITY_CODE]注意科创板688 开头和创业板300 开头都包含在内各自的交易所归属不同但前缀规则是稳定的。startswith(6)在沪深两市范围内足够可靠不必过度设计。4.3 合并后按主键去重同一份财报API 返回一次、Selenium 页面里也可能出现合并后大概率产生重复行。去重主键建议选SECURITY_CODEREPORTDATEREPORT_TYPE的组合。REPORT_TYPE如果有的话代表是“年报”“一季报”“中报”“三季报”中的哪一个。df_merged df_merged.drop_duplicates( subset[SECURITY_CODE, REPORTDATE, REPORT_TYPE], keeplast )keeplast意味着两条重复记录里保留最后一条也就是后爬到的覆盖先爬到的。如果你担心 F10 页面里人工修正过的数据被 API 原始数据覆盖可以反过来用keepfirst。这个参数的取舍没有标准答案取决于你信任哪条数据源。4.4 写入 SQLite 而不是 CSV爬取五百家公司的三张报表CSV 文件数量会迅速膨胀到难管理的地步文件名、编码、追加写入都会成为负累。我一般在这种体量下直接写 SQLite标准库sqlite3就能处理不需要额外装 MySQL。import sqlite3 conn sqlite3.connect(eastmoney_financials.db) df_merged.to_sql(financial_report, conn, if_existsappend, indexFalse) conn.close()用if_existsappend支持断点续爬——爬完一批数据入库下次重新运行时不会覆盖旧数据而是追加新数据。入库后每次查询就是一条 SQL 的事不用每次重新读一堆 CSV 再 concat。5. 应对 429 限流与 Cookie 失效的验证手法5.1 用“先探后爬”测出接口的容忍度一段代码拿到手别直接全量跑先拿单个股票跑通再用三只股票跑一轮观察是否出现 429。429 响应本身带着明确的语义请求频率超限。“exceeded retry limit, last status: 429 too many requests” 这类日志之所以常见就是因为脚本缺少退避机制。我的验证标准是一分钟不超过 20 次请求单次请求间隔不低于 3 秒。这个阈值不是东方财富公开的官方值而是从实际踩坑里总结的经验值同域名的不同接口容忍度不同。5.2 Cookie 时效与 Selenium 取 Cookie 注入 RequestsRequests 会话在某些场景下会突然开始返回 403 或登录跳转页面原因是服务端下发的 Cookie 过期了。这时不必重开 Requests 流程直接从已有的 Selenium driver 里导出 Cookie再注入 Requests 的 session。from selenium import webdriver # 假设 driver 已登录并访问过目标页面 cookies driver.get_cookies() session requests.Session() for cookie in cookies: session.cookies.set(cookie[name], cookie[value])这种方式特别适合“先 Selenium 过验证、后 Requests 批量拉数据”的混合架构。注意 Cookie 携带域名信息Requests 请求的 URL 域名必须与 Cookie 所属域名一致否则 Cookie 不会生效。5.3 落盘后的数据完整性校验爬完不能直接收工还得验证有没有缺页、缺行。最直接的办法是把已入库数据的数量与接口返回的result.count做对比。result.count是接口返回的数据总量len(df_merged)是本地实际落盘数量两者应该相等。expected payload[result][count] actual len(df_merged) if actual ! expected: print(f数量不一致期望 {expected}实际 {actual})另一个验证维度是抽样对比单只股票的财报关键字段比如营业收入和净利润与东方财富页面展示值做人工核对。自动化脚本只能保证“爬到了”不能保证“爬对了”。字段错位、单位混淆这类逻辑错误只有抽样能发现。本文还有配套的精品资源点击获取
返回列表