首页 / 资讯中心 / 文章详情

东方财富网财报爬取:Requests与Selenium协同实战指南

东方财富网财报爬取:Requests与Selenium协同实战指南 ★ FEATURED ARTICLE
简介本资源是一套面向Python数据采集初学者与金融数据分析爱好者的实战项目聚焦于东方财富网上市公司财务报表的自动化获取解决公开财报数据难以批量下载、格式不统一等实际痛点。压缩包共19个文件含2个核心爬虫脚本eastmoney_crawler.py基于Seleniumeastmoney_crawler2.py基于Requests、10个CSV结构化财报样本覆盖资产负债表、利润表、现金流量表等8类报表、3个备份文件及README.md等工程文档整体37.96MB目录组织清晰便于理解技术选型差异与数据落地路径。已有47人学习下载适合通过对比两种主流爬虫方案自动化驱动 vs HTTP直连深入掌握反爬应对、参数化采集、会计期间控制及CSV标准化输出等关键技能。读者可直接运行命令行程序按年度、季度、报表类型和页码范围定制采集任务并获得符合企业会计准则的本地结构化数据附带完整LICENSE与使用说明兼顾学习合规性与工程可复用性。1. 为什么爬东方财富网财务报表Selenium 和 Requests 必须“双剑合璧”你手头有一份上市公司清单想批量获取近五年资产负债表、利润表、现金流量表的原始数据——不是截图不是 PDF是能直接进 Pandas 做同比环比、进 BI 做可视化、进模型做因子分析的结构化 CSV。但点开东方财富网eastmoney.com财报页面就会发现关键表格是 JavaScript 动态渲染的URL 不带参数、翻页靠 AJAX、数据藏在__NEXT_DATA__或window.varData里甚至部分字段如“应收账款周转率”根本不在 HTML 源码中而是由前端 JS 计算后插入 DOM。这时候只用requests.get()拿到的是一堆空 div只用 Selenium 点击等待提取跑 100 家公司要 4 小时还极易被风控拦截。真正的落地解法不是二选一而是让 Requests 负责稳准快地扛住高频静态请求如公告列表、PDF 下载链接Selenium 只在必须交互的环节如切换报表年份、展开隐藏行、触发 JS 渲染精准出手——这是当前爬取东财财报最省时、最抗干扰、最易维护的工程实践。适合需要稳定获取 A 股上市公司结构化财务数据的量化研究员、信披合规岗、财经数据库建设者以及正在写毕业论文需实证数据的经管类研究生。它不承诺“全自动零维护”但把 90% 的翻车点转化成了可配置、可日志、可重试的确定性步骤。2. 从 URL 解构到数据定位东财财报页面的三层结构与抓取策略东方财富网财报数据并非散落在单一页面而是按“公司 → 报告期 → 报表类型”三级嵌套组织。理解这三层结构是设计 Requests/Selenium 协同逻辑的前提。我们以贵州茅台6005192023 年年报为例手动访问其财报页https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?typewebcodesh600519。注意这个 URL 本身不返回财报表格而是一个 React 应用入口页。真实数据通过后续 XHR 请求加载或由前端 JS 解析全局变量注入。下面分层拆解2.1 第一层公司代码与财报入口页Requests 可全量获取东财对所有 A 股公司使用统一入口模板https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?typewebcode{market}{code}其中{market}是sh沪市或sz深市{code}是 6 位数字代码。该页面 HTML 中虽无表格但包含两个关键静态资源script标签内嵌的__NEXT_DATA__JSON含公司基础信息名称、行业、上市日期页面底部script中的varData对象含“最新报告期”、“报告期列表”等元数据。提示这部分完全可由requests同步获取无需启动浏览器。我一般会先用requests批量拉取 3000 家公司的入口页解析出每家的“最新财报年份”和“可用报告期数组”为后续 Selenium 任务生成精确目标列表避免无效点击。import requests from bs4 import BeautifulSoup import re import json def get_company_meta(code: str, market: str sh) - dict: url fhttps://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?typewebcode{market}{code} headers { 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, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Connection: keep-alive, } try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() soup BeautifulSoup(resp.text, html.parser) # 提取 __NEXT_DATA__ next_data_script soup.find(script, stringre.compile(r__NEXT_DATA__)) if next_data_script: next_data_match re.search(r__NEXT_DATA__\s*\s*({.*?});, next_data_script.string) if next_data_match: next_data json.loads(next_data_match.group(1)) company_name next_data.get(props, {}).get(pageProps, {}).get(companyName, ) # 提取 varData 中的 reportPeriods var_data_script soup.find(script, stringre.compile(rvar\svarData\s*)) if var_data_script: var_data_match re.search(rvar\svarData\s*\s*({.*?});, var_data_script.string, re.DOTALL) if var_data_match: var_data json.loads(var_data_match.group(1)) report_periods var_data.get(reportPeriods, []) latest_year report_periods[0] if report_periods else None return { code: code, name: company_name, latest_report_year: latest_year, all_report_periods: report_periods } except Exception as e: print(f[ERROR] Failed to fetch meta for {code}: {e}) return {code: code, name: , latest_report_year: None, all_report_periods: []} # 示例获取贵州茅台元数据 meta get_company_meta(600519, sh) print(f公司{meta[name]}最新财报年份{meta[latest_report_year]}) # 输出公司贵州茅台最新财报年份2023这段代码的核心价值在于它把“哪家公司有哪几年财报”这个动态决策问题提前转化为静态 JSON 查询彻底规避了 Selenium 在首页反复点击“切换年份下拉框”的低效操作。reportPeriods数组通常形如[2023, 2022, 2021, 2020, 2019]是我们后续 Selenium 任务的明确输入。2.2 第二层报告期切换与报表类型选择Selenium 精准交互点当用户在网页上点击“2023年”或切换“资产负债表”时页面不会整页刷新而是触发一个fetch请求加载对应报表的 HTML 片段或 JSON 数据。这个行为由前端 JS 控制URL 本身不变化。因此Requests 无法直接构造该请求——因为请求体body或请求头headers中往往包含动态 token如X-Requested-With: XMLHttpRequest 时间戳签名且 endpoint 路径藏在 JS 里。Selenium 的作用就在此刻体现它模拟真实用户点击让 JS 自动完成 token 生成、请求发送、DOM 更新全过程。我们只需定位到两个关键元素年份选择器div classreport-year-selector下的span标签文本为“2023”报表类型 Tabdiv classtab-list中的a标签>from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.options import Options def setup_driver(headlessTrue): options Options() if headless: options.add_argument(--headless) options.add_argument(--no-sandbox) options.add_argument(--disable-dev-shm-usage) options.add_argument(--disable-gpu) options.add_argument(--disable-blink-featuresAutomationControlled) options.add_experimental_option(excludeSwitches, [enable-automation]) options.add_experimental_option(useAutomationExtension, False) driver webdriver.Chrome(optionsoptions) driver.execute_cdp_cmd(Page.addScriptToEvaluateOnNewDocument, { source: Object.defineProperty(navigator, webdriver, { get: () undefined }) }) return driver def switch_to_report_period_and_type(driver, year: str, report_type: str): report_type: zcfzb (资产负债表), lrb (利润表), xjllb (现金流量表) try: # 等待年份选择器加载 year_selector WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.CLASS_NAME, report-year-selector)) ) # 点击目标年份如 2023 year_spans year_selector.find_elements(By.TAG_NAME, span) for span in year_spans: if span.text.strip() year: span.click() break else: raise ValueError(fYear {year} not found in selector) # 等待年份切换完成页面出现 loading 动画消失 WebDriverWait(driver, 10).until_not( EC.presence_of_element_located((By.CLASS_NAME, loading-mask)) ) # 切换报表类型 Tab tab_links driver.find_elements(By.CSS_SELECTOR, a[data-type]) for link in tab_links: if link.get_attribute(data-type) report_type: link.click() break else: raise ValueError(fReport type {report_type} not found) # 等待新报表内容加载完成以表格容器出现为标志 table_container WebDriverWait(driver, 15).until( EC.presence_of_element_located((By.CLASS_NAME, table-container)) ) return table_container except Exception as e: print(f[SELENIUM ERROR] Failed to switch to {year}/{report_type}: {e}) raise # 使用示例 driver setup_driver() driver.get(https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?typewebcodesh600519) try: table_container switch_to_report_period_and_type(driver, 2023, zcfzb) print(成功切换至2023年资产负债表) finally: driver.quit()这段代码的关键设计点WebDriverWait的三重等待先等选择器存在再等 loading 掩膜消失最后等表格容器出现。跳过任意一步都可能导致后续find_element报NoSuchElementException>from bs4 import BeautifulSoup import pandas as pd def parse_financial_table_from_html(html: str) - pd.DataFrame: 从财报页面 HTML 中提取主表格返回标准化 DataFrame 表格结构第一行为表头年份列第二行为项目名如货币资金后续行为数值 soup BeautifulSoup(html, html.parser) table soup.find(div, class_table-container).find(table) if not table: raise ValueError(Financial table not found in HTML) # 提取表头年份行 header_row table.find(tr, class_header-row) if not header_row: # 兜底找第一个 tr 作为 header header_row table.find(tr) headers [] for th in header_row.find_all([th, td]): text th.get_text(stripTrue) if text and not text.isdigit(): # 过滤掉纯数字的占位列 headers.append(text) # 提取数据行 data_rows [] for tr in table.find_all(tr)[1:]: # 跳过 header 行 tds tr.find_all([td, th]) if len(tds) 2: continue # 第一列是项目名称其余是数值 item_name tds[0].get_text(stripTrue) if not item_name: continue values [td.get_text(stripTrue) for td in tds[1:]] # 对齐 headers 长度不足补空 values values[:len(headers)] [] * (len(headers) - len(values)) data_rows.append([item_name] values) # 构建 DataFrame df pd.DataFrame(data_rows, columns[项目] headers) return df # 在 Selenium 切换后调用 html driver.page_source df_zcfzb_2023 parse_financial_table_from_html(html) print(df_zcfzb_2023.head())此函数输出的 DataFrame 形如项目2023-12-312022-12-312021-12-31货币资金2,123.451,890.121,654.33交易性金融资产345.67298.45256.78注意东财表格存在“合并单元格”和“隐藏行”问题。例如“应收账款”下方常有“—其中坏账准备”子项它们在 HTML 中是独立tr但视觉上缩进。我们的解析逻辑保留了所有行后续可通过项目列的前缀如—其中做层级归并这是结构化清洗的起点而非爬虫阶段的负担。3. Requests 与 Selenium 的协同流水线如何让两者各司其职、无缝衔接把 Requests 和 Selenium 当成两条流水线工人Requests 负责搬运原材料元数据、静态资源、PDF 链接Selenium 负责精密装配点击、等待、渲染。二者通过明确的“交接物”协同——即结构化的任务清单Task List。这个清单不是硬编码的字符串而是由 Requests 预处理生成的、带上下文的 Python 字典列表。下面展示一个端到端的协同流程3.1 生成任务清单Requests 扫描 规则过滤我们不盲目爬取所有公司所有年份。实际业务中往往只需“近 3 年年报”或“ST 公司全部历史数据”。因此第一步是用 Requests 批量获取元数据并应用业务规则生成最小可行任务集import pandas as pd from concurrent.futures import ThreadPoolExecutor, as_completed def generate_task_list(codes: list, years_back: int 3, report_types: list [zcfzb, lrb, xjllb]) - list: 生成爬取任务列表 codes: 公司代码列表如 [600519, 000001] years_back: 向前追溯年份数 report_types: 报表类型列表 返回: [{code: 600519, name: 贵州茅台, year: 2023, type: zcfzb, url: ...}, ...] tasks [] def process_code(code): # 获取公司元数据 meta get_company_meta(code, sh if code.startswith(6) else sz) if not meta[all_report_periods]: return [] # 取最近 N 年 available_years meta[all_report_periods][:years_back] task_batch [] for year in available_years: for rtype in report_types: task_batch.append({ code: code, name: meta[name], year: year, type: rtype, url: fhttps://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis/index?typewebcode{sh if code.startswith(6) else sz}{code}, market: sh if code.startswith(6) else sz }) return task_batch # 并发获取元数据10 线程足够避免被限频 with ThreadPoolExecutor(max_workers10) as executor: future_to_code {executor.submit(process_code, code): code for code in codes} for future in as_completed(future_to_code): try: batch future.result() tasks.extend(batch) except Exception as e: print(f[TASK GEN ERROR] {e}) print(f生成任务总数{len(tasks)}) return tasks # 示例为贵州茅台和平安银行生成2022-2023年三张报表任务 codes [600519, 000001] task_list generate_task_list(codes, years_back2) # 输出示例 # [ # {code: 600519, name: 贵州茅台, year: 2023, type: zcfzb, ...}, # {code: 600519, name: 贵州茅台, year: 2023, type: lrb, ...}, # ... # ]这个函数的价值在于它把“爬什么”的决策权交给了数据驱动逻辑而非人工 Excel 表。years_back2意味着只爬 2023 和 2022即使某家公司有 2021 年数据也不取——这对降低总耗时、减少风控触发概率至关重要。3.2 执行任务Selenium 复用驱动 Requests 补位下载Selenium 启动浏览器是昂贵操作。若为每个任务新建driver100 个任务就要启停 100 次 Chrome光初始化就耗掉 20 分钟。正确做法是复用单个driver实例按任务顺序执行并在关键节点用 Requests 补位def execute_task_batch(task_list: list, output_dir: str ./data): 批量执行任务复用单个 driver os.makedirs(output_dir, exist_okTrue) driver setup_driver(headlessTrue) # 记录失败任务供重试 failed_tasks [] for i, task in enumerate(task_list): print(f[{i1}/{len(task_list)}] 正在处理{task[name]} {task[year]} {task[type]}) try: # Step 1: 访问公司入口页 driver.get(task[url]) # Step 2: Selenium 切换年份和报表类型 switch_to_report_period_and_type(driver, task[year], task[type]) # Step 3: 获取渲染后 HTML用 Requests/BS4 解析 html driver.page_source df parse_financial_table_from_html(html) # Step 4: 保存为 CSV文件名含公司、年份、类型 filename f{output_dir}/{task[code]}_{task[year]}_{task[type]}.csv df.to_csv(filename, indexFalse, encodingutf-8-sig) print(f✓ 已保存{filename}) # Step 5: 【Requests 补位】下载 PDF 原始文件如有 pdf_url extract_pdf_url_from_page(driver) # 此函数见下文 if pdf_url: download_pdf_with_requests(pdf_url, task[code], task[year], output_dir) except Exception as e: print(f✗ 失败{task[name]} {task[year]} {task[type]}错误{e}) failed_tasks.append(task) # 强制刷新页面避免状态污染下一个任务 driver.refresh() time.sleep(2) driver.quit() # 输出失败任务供人工检查或重试 if failed_tasks: print(f\n共 {len(failed_tasks)} 个任务失败详情见 failed_tasks.json) with open(f{output_dir}/failed_tasks.json, w, encodingutf-8) as f: json.dump(failed_tasks, f, ensure_asciiFalse, indent2) def extract_pdf_url_from_page(driver) - str: 从当前页面提取年报 PDF 下载链接 try: # 东财 PDF 链接通常在 a 标签中href 以 .pdf 结尾且 text 包含PDF或下载 links driver.find_elements(By.TAG_NAME, a) for link in links: href link.get_attribute(href) text link.text.strip() if href and href.lower().endswith(.pdf) and (pdf in text.lower() or 下载 in text): return href # 备用查找页面中隐藏的 PDF URL藏在 script 或>import time import random from functools import wraps def retry_with_backoff(max_retries3, base_delay1, max_delay60): 装饰器指数退避重试 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries 1): try: return func(*args, **kwargs) except Exception as e: if attempt max_retries: raise e # 计算退避时间base * 2^attempt jitter delay min(base_delay * (2 ** attempt), max_delay) jitter random.uniform(0, 0.1 * delay) total_delay delay jitter print(f[RETRY] {func.__name__} 失败{total_delay:.1f}s 后重试第{attempt1}次...) time.sleep(total_delay) return None return wrapper return decorator retry_with_backoff(max_retries2, base_delay2) def get_company_meta_safe(code: str, market: str sh) - dict: 带重试的元数据获取 return get_company_meta(code, market) retry_with_backoff(max_retries1, base_delay5) def switch_to_report_period_and_type_safe(driver, year: str, report_type: str): 带重试的 Selenium 切换 return switch_to_report_period_and_type(driver, year, report_type)血泪经验max_retries2是平衡效率与成功率的黄金值。设为 3 次总耗时可能翻倍设为 0 次则 10% 任务必然失败。base_delay2秒起步配合随机抖动jitter能有效打散请求峰谷显著降低429触发率。4. 避坑指南东财财报爬取中 5 个真实踩过的坑与解法爬取东财财报不是写一次就能永久运行的脚本而是一场与前端迭代、反爬升级、网络波动的持续博弈。以下是我在 37 家券商、12 个量化团队的实际项目中反复验证过的 5 个高频翻车点。每个都附带现象、根因和可立即落地的解法。4.1 现象Selenium 点击年份后表格内容没更新page_source还是旧数据原因东财前端在切换年份后会发起一个POST请求获取新数据但该请求的响应是 JSON前端 JS 再将 JSON 解析为 HTML 插入 DOM。WebDriverWait等待table-container出现但此时 DOM 已更新page_source却未同步——因为page_source返回的是初始 HTML而非当前 DOM 快照。解决改用driver.execute_script(return document.documentElement.outerHTML)获取实时 DOM或直接用driver.find_element(By.CLASS_NAME, table-container).get_attribute(outerHTML)提取目标区域 HTML。后者更快、更精准。# ✅ 正确获取实时 DOM 片段 table_html driver.find_element(By.CLASS_NAME, table-container).get_attribute(outerHTML) df parse_financial_table_from_html(table_html) # ❌ 错误page_source 可能滞后 # html driver.page_source # 不保证是最新渲染结果4.2 现象requests.get()拿到的__NEXT_DATA__为空或varData解析失败原因东财近期将部分公司入口页改为服务端渲染SSR__NEXT_DATA__被移除同时varData的定义方式从var varData {...}改为window.varData {...}正则表达式rvar\svarData\s*失效。解决改用更鲁棒的 JS 变量提取逻辑遍历所有script标签搜索包含reportPeriods和companyName的 JSON 片段def extract_var_data_from_scripts(soup): scripts soup.find_all(script) for script in scripts: if script.string and reportPeriods in script.string and companyName in script.string: # 尝试匹配 { ... } 包裹的 JSON json_match re.search(r({[^}]*?reportPeriods[^}]*?}), script.string, re.DOTALL) if json_match: try: return json.loads(json_match.group(1)) except json.JSONDecodeError: continue return {}4.3 现象爬取过程中突然所有请求返回403 Forbidden且User-Agent未变原因东财的 CDN如阿里云 WAF会基于请求频率、Cookie、TLS 指纹等多维度判定风险。即使User-Agent相同若连续请求来自同一 TLS 握手指纹如 Chrome 120 的默认指纹也会被标记。解决在requests会话中启用requests-toolbelt的SSLAdapter随机化 TLS 版本和扩展对 Selenium使用undetected-chromedriver替代原生 ChromeDriver需 pip install undetected-chromedriver# Requests 端随机 TLS from requests_toolbelt.adapters.ssl import SSLAdapter import ssl session requests.Session() adapter SSLAdapter(ssl_versionrandom.choice([ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_3])) session.mount(https://, adapter) # Selenium 端改用 uc import undetected_chromedriver as uc driver uc.Chrome(headlessTrue, use_subprocessTrue)4.4 现象利润表中“营业收入”数值为--或None但网页上明明显示数字原因东财对部分 ST、*ST 公司或新上市公司的利润表会将“营业收入”等关键字段设为null前端 JS 用--占位。get_text()只能拿到--无法还原原始null。解决不依赖get_text()而用element.get_attribute(textContent)或element.get_attribute(innerText)并结合element.get_attribute(data-value)东财部分单元格会把原始数值存于此属性def safe_get_cell_value(cell): # 优先取>def scroll_to_and_click(element): driver.execute_script(arguments[0].scrollIntoView({block: center});, element) time.sleep(0.5) # 等待滚动动画结束 element.click() # 在 switch_to_report_period_and_type 中替换点击逻辑 for span in year_spans: if span.text.strip() year: scroll_to_and_click(span) # 替代 span.click() break5. 数据质量校验与增量更新让爬取结果真正可用的最后 10%爬下来的数据如果不能直接进模型、进报表、进数据库那前面所有工作都是空中楼阁。我见过太多团队花 3 天写爬虫却用 2 周手工清洗数据——根源在于缺少自动化校验与增量机制。下面是我在线上系统中稳定运行 2 年的校验与更新方案。5.1 三重校验字段完整性、数值合理性、跨报表勾稽东财财报数据不是孤立的三张表之间存在强勾稽关系。例如资产负债表“期末货币资金” ≈ 现金流量表“期末现金及现金等价物余额”利润表“净利润” 资产负债表“年初未分配利润” - “本期分配股利” ≈ “年末未分配利润”。我们构建一个校验器对每个公司每年的三张表做交叉验证def validate_financial_consistency(code: str, year: str, data_dir: str ./financial_data) - dict: 校验单家公司单一年份三张表的勾稽关系 返回: {status: pass/warn/fail, issues: [...]} issues [] # 读取三张表 try: zcfzb pd.read_csv(f{data_dir}/{code}_{year}_zcfzb.csv, dtypestr) lrb pd.read_csv(f{data_dir}/{code}_{year}_lrb.csv, dtypestr) xjllb pd.read_csv(f{data_dir}/{code}_{year}_xjllb.csv, dtypestr) except FileNotFoundError as e: return {status: fail, issues: [f缺失报表文件{e}]} # 提取关键字段需处理千分位逗号、括号负数 def clean_num(x): p a hrefhttps://download.csdn.net/download/2501_91537388/92412100 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p
阅读完成 · 觉得有帮助?
咨询建站