简介本资源是一套面向人工智能与数据分析初学者的实战型网络爬虫项目聚焦于从东方财富网批量获取上市公司结构化财务数据解决金融数据采集难、动态页面解析复杂等实际问题。压缩包共12个文件含10个CSV格式的财务报表涵盖资产负债表、利润表、现金流量表、业绩预告及预约披露时间表等和2个核心爬虫脚本Selenium版与Requests高效版总大小19.88MB目录按数据类型分层组织便于理解财报维度与字段映射关系。已有351人学习下载适合高校课程实践、量化分析入门或求职项目储备。用户可直接运行eastmoney_crawler2.py实现百倍提速的批量抓取支持按年份、季度、报表类型及页码范围灵活配置输出结果自动归档为标准CSV附带字段说明逻辑严谨杜绝虚构填充具备完整工程闭环与生产级可复用性。1. 项目概述从数据需求到爬虫实现的完整链路最近在做一个金融数据分析相关的项目核心需求是批量获取A股上市公司的财务报表数据。市面上虽然有Wind、Choice这样的专业金融终端但对于个人开发者、学生团队或者初创的量化研究小组来说成本是个大问题。这时候公开的财经网站就成了重要的数据源东方财富网因其数据相对全面、结构清晰成为了很多人的首选。这个项目就是围绕如何用Python爬虫稳定、高效地从东方财富网抓取上市公司的三大报表资产负债表、利润表、现金流量表数据并将其整理成结构化的数据集。听起来好像就是写个爬虫抓网页但实际做下来你会发现这里面的门道不少。首先东方财富网的页面结构并不总是那么简单直接尤其是涉及到财务报表这种分页、分年度、分报告期的数据。其次网站的反爬策略也在不断升级直接裸奔requests库很容易被拦截。再者财务数据本身字段多、关联性强如何设计存储结构确保数据完整、便于后续分析也是一个需要仔细考量的问题。这个项目实践不仅涉及基础的HTTP请求、HTML解析还考验你对反爬机制的应对、数据清洗的功底以及工程化思维——如何让这个爬虫能长期、稳定地运行。如果你正面临类似的数据获取难题无论是为了课程作业、研究课题还是想自己搭建一个简单的量化分析数据源这篇内容应该能给你提供一条清晰的路径。我会从工具选型、页面分析、核心代码实现到反爬策略、数据存储和错误处理一步步拆解整个流程并分享我在这个过程中踩过的坑和总结的经验。2. 核心工具选型与前期环境准备工欲善其事必先利其器。在开始编码之前选择合适的工具链能事半功倍。这个项目我们主要使用Python因为它有极其丰富的爬虫和数据处理生态。2.1 网络请求库的选择Requests 与 Selenium 的权衡对于东方财富网的财务报表页面我的首选是requests库。它轻量、高效是处理HTTP请求的瑞士军刀。大部分静态页面内容requests配合headers伪装就能搞定。我会用到requests.Session()来维持会话这样能更好地模拟浏览器行为比如保持登录状态如果需要的话和处理一些cookie。但是财经网站的数据展示越来越依赖JavaScript动态渲染。如果你发现用requests获取的HTML里找不到表格数据那很可能数据是通过AJAX接口异步加载的。这时候直接分析页面网络请求找到真正的数据API接口是最高效的方法。东方财富网的财务报表数据通常有对应的JSON接口我们可以直接调用这些接口这比渲染整个页面要快得多也稳定得多。什么情况下会用Selenium或Playwright呢当目标数据必须通过复杂的交互如点击选项卡、输入查询条件才能触发加载且这些交互逻辑难以通过模拟API请求复现时。对于东方财富网经过测试其核心财务数据接口是暴露的因此本项目以requests为主Selenium作为备用方案用于应对接口变动或特别复杂的场景。这样可以保证爬虫的效率。2.2 数据解析与处理库拿到HTML或者JSON数据后需要解析和提取。BeautifulSoup4和lxml是解析HTML的黄金搭档。BeautifulSoup写起来更简单直观适合快速原型开发lxml的解析速度更快适合处理大量页面。本项目两者都会涉及根据场景选用。对于JSON数据Python内置的json库就足够了。而数据清洗和整理绝对离不开pandas。它强大的DataFrame结构非常适合处理表格型数据无论是数据清洗、转换还是最终输出为Excel或CSVpandas都能优雅地完成。2.3 反爬虫策略的应对工具箱直接爬取很容易被屏蔽我们需要一些“伪装”技巧。User-Agent轮换准备一个包含多个常见浏览器UA的列表每次请求随机选取一个。IP代理池对于大规模、高频的爬取使用代理IP是必须的可以避免单个IP被封锁。可以考虑一些付费的代理服务或者自建代理池。请求头Headers完善除了User-Agent还应合理设置Referer、Accept-Language、Accept-Encoding等字段让请求看起来更像来自真实浏览器。请求频率控制在请求间加入随机延时例如time.sleep(random.uniform(1, 3))避免对服务器造成瞬间高压这也是基本的网络礼仪。2.4 环境搭建步骤假设你已经安装了Python3.7及以上版本我们通过pip安装必要的库。建议使用虚拟环境如venv或conda来管理依赖。# 创建并激活虚拟环境以venv为例 python -m venv ef_crawler_env source ef_crawler_env/bin/activate # Linux/Mac # ef_crawler_env\Scripts\activate # Windows # 安装核心库 pip install requests beautifulsoup4 lxml pandas # 可选如果需要用到动态渲染作为备用方案 pip install selenium # 同时需要下载对应的浏览器驱动如ChromeDriver环境准备好后我们可以新建一个项目目录比如eastmoney_crawler在里面开始我们的代码编写。3. 目标网站分析与数据接口定位这是爬虫最核心、也最需要耐心的一步。盲目写代码只会事倍功半我们必须先搞清楚数据在哪、怎么来的。3.1 分析东方财富网财务报表页面结构首先手动打开东方财富网找到任意一家上市公司的“财务分析”页面。例如搜索“贵州茅台(600519)”进入其股票详情页找到“财务分析”或“资产负债表”等标签页。打开浏览器的开发者工具F12切换到“网络”(Network)选项卡。刷新页面你会看到浏览器发出的所有请求。我们的目标是找到真正携带财务表格数据的请求。通常表格数据不会直接嵌在初始的HTML文档里而是通过XHRAJAX请求获取的。在“网络”选项卡中筛选XHR或Fetch类型的请求。一边点击页面上的不同报告期如“年报”、“季报”或不同报表类型一边观察哪个请求的响应里包含了表格数据。以资产负债表为例你可能会发现一个类似这样的请求URLhttps://datacenter.eastmoney.com/securities/api/get/v1?reportNameRPT_F10_FN_BALANCEfilter(SECURITY_CODE%3D%22600519%22)pageNum1pageSize10这个URL就是关键的数据接口。其中包含了几个重要参数reportName: 报告名称如RPT_F10_FN_BALANCE资产负债表、RPT_F10_FN_INCOME利润表、RPT_F10_FN_CASHFLOW现金流量表。filter: 过滤条件这里指定了股票代码SECURITY_CODE“600519”。pageNum和pageSize: 分页参数。通过修改这些参数我们就能获取不同公司、不同报表、不同分页的数据。响应格式通常是JSON结构清晰包含了data字段里面就是表格的行数据以及result字段可能包含总页数等信息。注意东方财富网的接口参数和URL结构可能会随时间更新。以上URL和参数仅为示例实际操作时务必使用开发者工具捕获最新的、有效的接口地址和参数名。这是爬虫项目中最易变的部分。3.2 接口参数逆向与规律总结捕获到接口后需要系统性地测试参数找出规律。我们需要爬取所有上市公司所以关键是如何构造filter参数以及如何获取完整的股票代码列表。股票代码列表获取东方财富网有专门的接口或页面提供A股所有股票的列表。我们可以找一个相对稳定的列表页或者从其他数据源如交易所官网获取基础股票代码列表作为我们爬虫的输入。报表类型与报告期通过修改reportName可以切换三大报表。报告期年报、季报通常也通过接口参数控制可能是reportDate或type如Q1表示一季报。需要仔细测试不同值对应的结果。分页逻辑财务报表数据可能有多页。接口返回的JSON里通常会包含总记录数total和每页大小pageSize。我们需要计算总页数total_pages ceil(total / pageSize)然后循环请求每一页。这个分析过程需要反复试验和记录。我建议单独写一个api_explorer.py脚本用不同的参数组合去请求接口并将响应结果打印出来直观地理解数据结构。4. 核心爬虫代码实现与模块化设计有了清晰的数据接口认识我们就可以开始编写爬虫了。一个好的爬虫应该是模块化、可配置、易维护的。4.1 请求模块封装处理Headers、代理与异常首先我们封装一个稳健的请求函数集成反爬策略和错误重试机制。import requests import time import random from typing import Optional, Dict, Any import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class EastMoneyFetcher: def __init__(self, use_proxyFalse, proxy_poolNone): self.session requests.Session() self.use_proxy use_proxy self.proxy_pool proxy_pool or [] # 初始化一个常见的User-Agent列表 self.user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ..., # ... 添加更多UA ] # 设置基础请求头 self.base_headers { Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, } self.session.headers.update(self.base_headers) def _get_random_headers(self) - Dict[str, str]: 生成随机的请求头 headers self.base_headers.copy() headers[User-Agent] random.choice(self.user_agents) # 可以随机添加或修改其他头信息 return headers def _get_proxy(self) - Optional[Dict[str, str]]: 从代理池中随机获取一个代理 if self.use_proxy and self.proxy_pool: return {http: random.choice(self.proxy_pool), https: random.choice(self.proxy_pool)} return None def fetch_json(self, url: str, params: Optional[Dict] None, max_retries: int 3) - Optional[Dict[str, Any]]: 获取JSON数据包含重试逻辑 for attempt in range(max_retries): try: headers self._get_random_headers() proxies self._get_proxy() # 添加随机延时避免请求过快 time.sleep(random.uniform(1, 2.5)) response self.session.get( url, paramsparams, headersheaders, proxiesproxies, timeout15 ) response.raise_for_status() # 检查HTTP错误 # 东方财富网接口成功时通常返回一个包含result或data字段的JSON json_data response.json() # 这里可以根据接口返回的具体成功标志进行判断例如 json_data.get(success) if json_data and isinstance(json_data, dict): return json_data else: logger.warning(f请求成功但返回数据格式异常: {url}) return None except requests.exceptions.RequestException as e: logger.error(f请求失败 (尝试 {attempt 1}/{max_retries}): {url}, 错误: {e}) if attempt max_retries - 1: wait_time (attempt 1) * 5 # 指数退避 time.sleep(wait_time) else: logger.error(f已达到最大重试次数放弃请求: {url}) return None except ValueError as e: # JSON解码错误 logger.error(fJSON解析失败: {url}, 错误: {e}) return None return None这个EastMoneyFetcher类封装了请求的核心逻辑包括随机UA、代理支持、错误重试和延时控制是爬虫稳定运行的基石。4.2 数据解析模块从JSON到结构化DataFrame接口返回的JSON数据需要被解析成我们需要的表格格式。不同报表的字段不同我们需要为每种报表编写对应的解析函数。import pandas as pd from typing import List, Dict, Any def parse_balance_sheet(json_data: Dict[str, Any]) - pd.DataFrame: 解析资产负债表JSON数据。 假设接口返回的JSON结构为: {result: {data: [...], total: ...}} 其中 data 是一个列表每个元素是一条记录一个报告期的数据。 if not json_data or result not in json_data: return pd.DataFrame() data_list json_data[result].get(data, []) if not data_list: return pd.DataFrame() # 将数据列表转换为DataFrame df pd.DataFrame(data_list) # 关键步骤字段重命名和筛选 # 原始数据字段名可能是英文或缩写我们需要映射成易懂的中文名 column_mapping { SECURITY_CODE: 股票代码, SECURITY_NAME_ABBR: 股票名称, REPORT_DATE: 报告日期, TOTAL_ASSETS: 总资产, TOTAL_LIABILITIES: 总负债, TOTAL_EQUITY: 所有者权益合计, # ... 根据实际接口返回字段添加所有需要的映射 } # 只保留我们映射过的、感兴趣的列 existing_columns [col for col in column_mapping.keys() if col in df.columns] df df[existing_columns].rename(columnscolumn_mapping) # 数据类型转换例如将字符串数字转换为数值型 numeric_columns [总资产, 总负债, 所有者权益合计] for col in numeric_columns: if col in df.columns: # 处理可能存在的逗号分隔符和空值 df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ), errorscoerce) # 处理报告日期转换为datetime格式 if 报告日期 in df.columns: df[报告日期] pd.to_datetime(df[报告日期], errorscoerce) return df # 类似地编写 parse_income_statement 和 parse_cash_flow 函数解析函数的核心任务是数据清洗和标准化选择需要的字段、重命名、类型转换、处理缺失值。这能保证后续分析时数据是干净、可用的。4.3 主控流程与任务调度现在我们需要一个主程序将股票列表、报表类型、日期范围等参数组织起来协调请求和解析模块并保存结果。import os from tqdm import tqdm # 用于显示进度条 def crawl_financial_reports(stock_codes: List[str], report_type: str, start_date: str, end_date: str, output_dir: str): 主爬取函数。 :param stock_codes: 股票代码列表 :param report_type: 报表类型如 balance, income, cashflow :param start_date: 开始日期格式 2022-01-01 :param end_date: 结束日期 :param output_dir: 输出目录 os.makedirs(output_dir, exist_okTrue) # 根据报表类型选择对应的接口URL模板和解析函数 api_config { balance: { url: https://datacenter.eastmoney.com/securities/api/get/v1, params_template: { reportName: RPT_F10_FN_BALANCE, pageSize: 50, # 每页条数可根据接口限制调整 }, parser: parse_balance_sheet }, # ... 配置利润表和现金流量表 } if report_type not in api_config: logger.error(f不支持的报表类型: {report_type}) return config api_config[report_type] fetcher EastMoneyFetcher(use_proxyFalse) # 根据需求开启代理 all_data_frames [] for stock_code in tqdm(stock_codes, descf爬取{report_type}数据): # 构造请求参数 params config[params_template].copy() # 构造filter参数这里假设接口filter格式为 (SECURITY_CODE“代码”) params[filter] f(SECURITY_CODE{stock_code}) # 可以添加报告日期过滤如果接口支持 # params[filter] f(SECURITY_CODE{stock_code})(REPORT_DATE{start_date})(REPORT_DATE{end_date}) page_num 1 while True: params[pageNum] page_num json_data fetcher.fetch_json(config[url], paramsparams) if not json_data: logger.warning(f股票 {stock_code} 第{page_num}页数据获取失败跳过。) break # 解析当前页数据 df_page config[parser](json_data) if df_page.empty: break # 当前页无数据可能已到最后一页 all_data_frames.append(df_page) # 判断是否还有下一页 result_info json_data.get(result, {}) total result_info.get(total, 0) current_count page_num * params[pageSize] if current_count total: break page_num 1 # 页间延时避免请求过快 time.sleep(random.uniform(0.5, 1.5)) # 股票间延时 time.sleep(random.uniform(2, 4)) # 合并所有数据并保存 if all_data_frames: final_df pd.concat(all_data_frames, ignore_indexTrue) # 按股票代码和报告日期排序 final_df.sort_values(by[股票代码, 报告日期], inplaceTrue) output_file os.path.join(output_dir, f{report_type}_sheet_{start_date}_to_{end_date}.csv) final_df.to_csv(output_file, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 logger.info(f数据已保存至: {output_file}, 共 {len(final_df)} 条记录。) else: logger.warning(未获取到任何数据。)这个主控流程实现了分页爬取、错误跳过、进度显示和结果合并。你可以通过调整stock_codes列表来控制爬取范围。5. 数据存储、增量更新与工程化考量一次性爬取全量数据后如何存储和后续更新是另一个关键问题。5.1 存储方案选择CSV vs. 数据库CSV/Excel文件适合数据量不大例如只爬取几百家公司最近几年的数据、一次性分析的项目。优点是简单直观无需搭建数据库环境。本项目初步输出结果可以使用CSV。数据库SQLite/MySQL/PostgreSQL如果数据量庞大全市场历史数据或需要频繁查询、关联分析数据库是更好的选择。SQLite是一个轻量级的选择无需服务器单个文件即可管理非常适合个人项目。使用pandas的to_sql方法和sqlalchemy库可以轻松地将DataFrame写入数据库。使用SQLite进行存储的示例import sqlite3 from sqlalchemy import create_engine # 创建数据库连接 engine create_engine(sqlite:///eastmoney_finance.db) # 将DataFrame写入数据库表 final_df.to_sql(namebalance_sheet, conengine, if_existsappend, indexFalse)if_existsappend参数允许我们进行增量更新。但需要注意直接追加可能导致重复数据需要在插入前根据“股票代码”和“报告日期”做去重判断。5.2 实现增量更新机制一个健壮的爬虫应该支持增量更新只爬取新的数据避免重复劳动和请求浪费。记录爬取状态在数据库或一个状态文件中记录每只股票最后一次成功爬取到的报告日期。爬取前检查每次运行爬虫时先读取状态只请求该股票在最后记录日期之后的数据。更新状态成功爬取并存储新数据后更新该股票的“最后爬取日期”状态。这需要修改主控流程增加状态管理的逻辑。一个简单的实现是使用一个JSON文件或一个专门的status数据库表来存储{stock_code: last_crawled_date}。5.3 日志记录与监控完善的日志对于排查问题至关重要。我们之前已经使用了Python的logging模块。应该将日志分级INFO, WARNING, ERROR输出到文件和控制台。# 在项目初始化时配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log, encodingutf-8), logging.StreamHandler() ] )定期检查crawler.log文件可以了解爬虫的运行状况、错误频率及时发现IP被封或接口变更等问题。6. 常见问题排查与实战经验分享在实际操作中你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的应对方法。6.1 请求被拒绝或返回空数据症状requests返回403、404或者返回的HTML/JSON里没有目标数据。排查步骤检查Headers特别是User-Agent和Referer。用浏览器开发者工具复制一次成功请求的所有Headers尽量模拟得一模一样。有时Cookie也很关键。检查IP短时间内请求过多IP可能被临时封禁。这就是为什么需要控制频率和使用代理。如果你在本地运行可以尝试重启路由器更换公网IP或者使用代理。验证接口有效性东方财富网的接口地址和参数名可能发生变化。务必用最新的开发者工具抓包确认当前的接口URL和参数格式。尝试动态渲染如果上述方法都无效数据可能是由非常复杂的JS生成的。这时可以祭出备用方案Selenium或Playwright直接模拟浏览器操作获取渲染后的页面源码再用BeautifulSoup解析。但这会慢很多。6.2 数据解析错误或字段缺失症状KeyError找不到字典键或者解析出来的DataFrame列名不对、数据为空。解决方案打印原始响应在解析函数里先把json_data完整地打印或保存下来仔细查看其结构。接口返回的字段名可能和你预期的不一样。动态构建字段映射不要硬编码字段映射字典。可以先写一个函数打印出第一次成功请求返回的JSON的所有键名然后根据这个列表来创建或更新你的column_mapping。处理嵌套结构有时数据可能嵌套在多层字典或列表里。你需要使用.get()方法安全地访问并做好空值判断。6.3 爬取速度慢与效率优化问题A股有几千家上市公司每家又有多年多期报表串行爬取会非常慢。优化方案并发/异步请求使用concurrent.futures的ThreadPoolExecutor可以实现多线程并发请求显著提升速度。但必须谨慎过高的并发会导致IP被迅速封禁。建议控制线程数如3-5个并为每个线程配置独立的代理IP。调整请求参数如果接口支持尽量在一次请求中获取更多数据如增大pageSize减少请求次数。分离IO和计算将网络请求IO密集型和数据分析CPU密集型分开。可以用队列queue模式生产者线程负责请求和下载原始数据消费者线程负责解析和存储。重要经验对于公开网站保持礼貌的爬取节奏是第一原则。在速度和数据完整性之间优先选择后者。设置合理的延时使用代理池并准备好当爬虫因各种原因中断后能从中断点恢复的机制比如记录已成功爬取的股票代码。6.4 法律与伦理风险规避遵守robots.txt检查https://www.eastmoney.com/robots.txt尊重网站规定的不允许爬取的目录。限制爬取频率这是最重要的伦理准则。我们的代码中已经加入了随机延时避免对目标服务器造成压力。数据用途爬取的数据应用于个人学习、研究或内部分析。切勿用于商业用途、公开大量分发或从事任何可能干扰网站正常服务的行为。标识自己在请求头中可以考虑添加一个自定义头如From: your-emailexample.com表明身份和善意方便网站管理员联系。这个项目从分析到实现涵盖了网络爬虫在真实场景下的主要挑战和解决方案。它不仅仅是一个脚本更是一个需要考虑健壮性、可维护性和伦理规范的小型数据工程。当你成功运行起这个爬虫并看到规整的财务数据源源不断地保存下来时那种成就感是实实在在的。剩下的就是基于这些高质量的数据去施展你在金融分析或人工智能算法上的才华了。本文还有配套的精品资源点击获取
Python爬虫实战：高效抓取东方财富网财务报表数据
简介本资源是一套面向人工智能与数据分析初学者的实战型网络爬虫项目聚焦于从东方财富网批量获取上市公司结构化财务数据解决金融数据采集难、动态页面解析复杂等实际问题。压缩包共12个文件含10个CSV格式的财务报表涵盖资产负债表、利润表、现金流量表、业绩预告及预约披露时间表等和2个核心爬虫脚本Selenium版与Requests高效版总大小19.88MB目录按数据类型分层组织便于理解财报维度与字段映射关系。已有351人学习下载适合高校课程实践、量化分析入门或求职项目储备。用户可直接运行eastmoney_crawler2.py实现百倍提速的批量抓取支持按年份、季度、报表类型及页码范围灵活配置输出结果自动归档为标准CSV附带字段说明逻辑严谨杜绝虚构填充具备完整工程闭环与生产级可复用性。1. 项目概述从数据需求到爬虫实现的完整链路最近在做一个金融数据分析相关的项目核心需求是批量获取A股上市公司的财务报表数据。市面上虽然有Wind、Choice这样的专业金融终端但对于个人开发者、学生团队或者初创的量化研究小组来说成本是个大问题。这时候公开的财经网站就成了重要的数据源东方财富网因其数据相对全面、结构清晰成为了很多人的首选。这个项目就是围绕如何用Python爬虫稳定、高效地从东方财富网抓取上市公司的三大报表资产负债表、利润表、现金流量表数据并将其整理成结构化的数据集。听起来好像就是写个爬虫抓网页但实际做下来你会发现这里面的门道不少。首先东方财富网的页面结构并不总是那么简单直接尤其是涉及到财务报表这种分页、分年度、分报告期的数据。其次网站的反爬策略也在不断升级直接裸奔requests库很容易被拦截。再者财务数据本身字段多、关联性强如何设计存储结构确保数据完整、便于后续分析也是一个需要仔细考量的问题。这个项目实践不仅涉及基础的HTTP请求、HTML解析还考验你对反爬机制的应对、数据清洗的功底以及工程化思维——如何让这个爬虫能长期、稳定地运行。如果你正面临类似的数据获取难题无论是为了课程作业、研究课题还是想自己搭建一个简单的量化分析数据源这篇内容应该能给你提供一条清晰的路径。我会从工具选型、页面分析、核心代码实现到反爬策略、数据存储和错误处理一步步拆解整个流程并分享我在这个过程中踩过的坑和总结的经验。2. 核心工具选型与前期环境准备工欲善其事必先利其器。在开始编码之前选择合适的工具链能事半功倍。这个项目我们主要使用Python因为它有极其丰富的爬虫和数据处理生态。2.1 网络请求库的选择Requests 与 Selenium 的权衡对于东方财富网的财务报表页面我的首选是requests库。它轻量、高效是处理HTTP请求的瑞士军刀。大部分静态页面内容requests配合headers伪装就能搞定。我会用到requests.Session()来维持会话这样能更好地模拟浏览器行为比如保持登录状态如果需要的话和处理一些cookie。但是财经网站的数据展示越来越依赖JavaScript动态渲染。如果你发现用requests获取的HTML里找不到表格数据那很可能数据是通过AJAX接口异步加载的。这时候直接分析页面网络请求找到真正的数据API接口是最高效的方法。东方财富网的财务报表数据通常有对应的JSON接口我们可以直接调用这些接口这比渲染整个页面要快得多也稳定得多。什么情况下会用Selenium或Playwright呢当目标数据必须通过复杂的交互如点击选项卡、输入查询条件才能触发加载且这些交互逻辑难以通过模拟API请求复现时。对于东方财富网经过测试其核心财务数据接口是暴露的因此本项目以requests为主Selenium作为备用方案用于应对接口变动或特别复杂的场景。这样可以保证爬虫的效率。2.2 数据解析与处理库拿到HTML或者JSON数据后需要解析和提取。BeautifulSoup4和lxml是解析HTML的黄金搭档。BeautifulSoup写起来更简单直观适合快速原型开发lxml的解析速度更快适合处理大量页面。本项目两者都会涉及根据场景选用。对于JSON数据Python内置的json库就足够了。而数据清洗和整理绝对离不开pandas。它强大的DataFrame结构非常适合处理表格型数据无论是数据清洗、转换还是最终输出为Excel或CSVpandas都能优雅地完成。2.3 反爬虫策略的应对工具箱直接爬取很容易被屏蔽我们需要一些“伪装”技巧。User-Agent轮换准备一个包含多个常见浏览器UA的列表每次请求随机选取一个。IP代理池对于大规模、高频的爬取使用代理IP是必须的可以避免单个IP被封锁。可以考虑一些付费的代理服务或者自建代理池。请求头Headers完善除了User-Agent还应合理设置Referer、Accept-Language、Accept-Encoding等字段让请求看起来更像来自真实浏览器。请求频率控制在请求间加入随机延时例如time.sleep(random.uniform(1, 3))避免对服务器造成瞬间高压这也是基本的网络礼仪。2.4 环境搭建步骤假设你已经安装了Python3.7及以上版本我们通过pip安装必要的库。建议使用虚拟环境如venv或conda来管理依赖。# 创建并激活虚拟环境以venv为例 python -m venv ef_crawler_env source ef_crawler_env/bin/activate # Linux/Mac # ef_crawler_env\Scripts\activate # Windows # 安装核心库 pip install requests beautifulsoup4 lxml pandas # 可选如果需要用到动态渲染作为备用方案 pip install selenium # 同时需要下载对应的浏览器驱动如ChromeDriver环境准备好后我们可以新建一个项目目录比如eastmoney_crawler在里面开始我们的代码编写。3. 目标网站分析与数据接口定位这是爬虫最核心、也最需要耐心的一步。盲目写代码只会事倍功半我们必须先搞清楚数据在哪、怎么来的。3.1 分析东方财富网财务报表页面结构首先手动打开东方财富网找到任意一家上市公司的“财务分析”页面。例如搜索“贵州茅台(600519)”进入其股票详情页找到“财务分析”或“资产负债表”等标签页。打开浏览器的开发者工具F12切换到“网络”(Network)选项卡。刷新页面你会看到浏览器发出的所有请求。我们的目标是找到真正携带财务表格数据的请求。通常表格数据不会直接嵌在初始的HTML文档里而是通过XHRAJAX请求获取的。在“网络”选项卡中筛选XHR或Fetch类型的请求。一边点击页面上的不同报告期如“年报”、“季报”或不同报表类型一边观察哪个请求的响应里包含了表格数据。以资产负债表为例你可能会发现一个类似这样的请求URLhttps://datacenter.eastmoney.com/securities/api/get/v1?reportNameRPT_F10_FN_BALANCEfilter(SECURITY_CODE%3D%22600519%22)pageNum1pageSize10这个URL就是关键的数据接口。其中包含了几个重要参数reportName: 报告名称如RPT_F10_FN_BALANCE资产负债表、RPT_F10_FN_INCOME利润表、RPT_F10_FN_CASHFLOW现金流量表。filter: 过滤条件这里指定了股票代码SECURITY_CODE“600519”。pageNum和pageSize: 分页参数。通过修改这些参数我们就能获取不同公司、不同报表、不同分页的数据。响应格式通常是JSON结构清晰包含了data字段里面就是表格的行数据以及result字段可能包含总页数等信息。注意东方财富网的接口参数和URL结构可能会随时间更新。以上URL和参数仅为示例实际操作时务必使用开发者工具捕获最新的、有效的接口地址和参数名。这是爬虫项目中最易变的部分。3.2 接口参数逆向与规律总结捕获到接口后需要系统性地测试参数找出规律。我们需要爬取所有上市公司所以关键是如何构造filter参数以及如何获取完整的股票代码列表。股票代码列表获取东方财富网有专门的接口或页面提供A股所有股票的列表。我们可以找一个相对稳定的列表页或者从其他数据源如交易所官网获取基础股票代码列表作为我们爬虫的输入。报表类型与报告期通过修改reportName可以切换三大报表。报告期年报、季报通常也通过接口参数控制可能是reportDate或type如Q1表示一季报。需要仔细测试不同值对应的结果。分页逻辑财务报表数据可能有多页。接口返回的JSON里通常会包含总记录数total和每页大小pageSize。我们需要计算总页数total_pages ceil(total / pageSize)然后循环请求每一页。这个分析过程需要反复试验和记录。我建议单独写一个api_explorer.py脚本用不同的参数组合去请求接口并将响应结果打印出来直观地理解数据结构。4. 核心爬虫代码实现与模块化设计有了清晰的数据接口认识我们就可以开始编写爬虫了。一个好的爬虫应该是模块化、可配置、易维护的。4.1 请求模块封装处理Headers、代理与异常首先我们封装一个稳健的请求函数集成反爬策略和错误重试机制。import requests import time import random from typing import Optional, Dict, Any import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class EastMoneyFetcher: def __init__(self, use_proxyFalse, proxy_poolNone): self.session requests.Session() self.use_proxy use_proxy self.proxy_pool proxy_pool or [] # 初始化一个常见的User-Agent列表 self.user_agents [ Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..., Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 ..., # ... 添加更多UA ] # 设置基础请求头 self.base_headers { Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive, } self.session.headers.update(self.base_headers) def _get_random_headers(self) - Dict[str, str]: 生成随机的请求头 headers self.base_headers.copy() headers[User-Agent] random.choice(self.user_agents) # 可以随机添加或修改其他头信息 return headers def _get_proxy(self) - Optional[Dict[str, str]]: 从代理池中随机获取一个代理 if self.use_proxy and self.proxy_pool: return {http: random.choice(self.proxy_pool), https: random.choice(self.proxy_pool)} return None def fetch_json(self, url: str, params: Optional[Dict] None, max_retries: int 3) - Optional[Dict[str, Any]]: 获取JSON数据包含重试逻辑 for attempt in range(max_retries): try: headers self._get_random_headers() proxies self._get_proxy() # 添加随机延时避免请求过快 time.sleep(random.uniform(1, 2.5)) response self.session.get( url, paramsparams, headersheaders, proxiesproxies, timeout15 ) response.raise_for_status() # 检查HTTP错误 # 东方财富网接口成功时通常返回一个包含result或data字段的JSON json_data response.json() # 这里可以根据接口返回的具体成功标志进行判断例如 json_data.get(success) if json_data and isinstance(json_data, dict): return json_data else: logger.warning(f请求成功但返回数据格式异常: {url}) return None except requests.exceptions.RequestException as e: logger.error(f请求失败 (尝试 {attempt 1}/{max_retries}): {url}, 错误: {e}) if attempt max_retries - 1: wait_time (attempt 1) * 5 # 指数退避 time.sleep(wait_time) else: logger.error(f已达到最大重试次数放弃请求: {url}) return None except ValueError as e: # JSON解码错误 logger.error(fJSON解析失败: {url}, 错误: {e}) return None return None这个EastMoneyFetcher类封装了请求的核心逻辑包括随机UA、代理支持、错误重试和延时控制是爬虫稳定运行的基石。4.2 数据解析模块从JSON到结构化DataFrame接口返回的JSON数据需要被解析成我们需要的表格格式。不同报表的字段不同我们需要为每种报表编写对应的解析函数。import pandas as pd from typing import List, Dict, Any def parse_balance_sheet(json_data: Dict[str, Any]) - pd.DataFrame: 解析资产负债表JSON数据。 假设接口返回的JSON结构为: {result: {data: [...], total: ...}} 其中 data 是一个列表每个元素是一条记录一个报告期的数据。 if not json_data or result not in json_data: return pd.DataFrame() data_list json_data[result].get(data, []) if not data_list: return pd.DataFrame() # 将数据列表转换为DataFrame df pd.DataFrame(data_list) # 关键步骤字段重命名和筛选 # 原始数据字段名可能是英文或缩写我们需要映射成易懂的中文名 column_mapping { SECURITY_CODE: 股票代码, SECURITY_NAME_ABBR: 股票名称, REPORT_DATE: 报告日期, TOTAL_ASSETS: 总资产, TOTAL_LIABILITIES: 总负债, TOTAL_EQUITY: 所有者权益合计, # ... 根据实际接口返回字段添加所有需要的映射 } # 只保留我们映射过的、感兴趣的列 existing_columns [col for col in column_mapping.keys() if col in df.columns] df df[existing_columns].rename(columnscolumn_mapping) # 数据类型转换例如将字符串数字转换为数值型 numeric_columns [总资产, 总负债, 所有者权益合计] for col in numeric_columns: if col in df.columns: # 处理可能存在的逗号分隔符和空值 df[col] pd.to_numeric(df[col].astype(str).str.replace(,, ), errorscoerce) # 处理报告日期转换为datetime格式 if 报告日期 in df.columns: df[报告日期] pd.to_datetime(df[报告日期], errorscoerce) return df # 类似地编写 parse_income_statement 和 parse_cash_flow 函数解析函数的核心任务是数据清洗和标准化选择需要的字段、重命名、类型转换、处理缺失值。这能保证后续分析时数据是干净、可用的。4.3 主控流程与任务调度现在我们需要一个主程序将股票列表、报表类型、日期范围等参数组织起来协调请求和解析模块并保存结果。import os from tqdm import tqdm # 用于显示进度条 def crawl_financial_reports(stock_codes: List[str], report_type: str, start_date: str, end_date: str, output_dir: str): 主爬取函数。 :param stock_codes: 股票代码列表 :param report_type: 报表类型如 balance, income, cashflow :param start_date: 开始日期格式 2022-01-01 :param end_date: 结束日期 :param output_dir: 输出目录 os.makedirs(output_dir, exist_okTrue) # 根据报表类型选择对应的接口URL模板和解析函数 api_config { balance: { url: https://datacenter.eastmoney.com/securities/api/get/v1, params_template: { reportName: RPT_F10_FN_BALANCE, pageSize: 50, # 每页条数可根据接口限制调整 }, parser: parse_balance_sheet }, # ... 配置利润表和现金流量表 } if report_type not in api_config: logger.error(f不支持的报表类型: {report_type}) return config api_config[report_type] fetcher EastMoneyFetcher(use_proxyFalse) # 根据需求开启代理 all_data_frames [] for stock_code in tqdm(stock_codes, descf爬取{report_type}数据): # 构造请求参数 params config[params_template].copy() # 构造filter参数这里假设接口filter格式为 (SECURITY_CODE“代码”) params[filter] f(SECURITY_CODE{stock_code}) # 可以添加报告日期过滤如果接口支持 # params[filter] f(SECURITY_CODE{stock_code})(REPORT_DATE{start_date})(REPORT_DATE{end_date}) page_num 1 while True: params[pageNum] page_num json_data fetcher.fetch_json(config[url], paramsparams) if not json_data: logger.warning(f股票 {stock_code} 第{page_num}页数据获取失败跳过。) break # 解析当前页数据 df_page config[parser](json_data) if df_page.empty: break # 当前页无数据可能已到最后一页 all_data_frames.append(df_page) # 判断是否还有下一页 result_info json_data.get(result, {}) total result_info.get(total, 0) current_count page_num * params[pageSize] if current_count total: break page_num 1 # 页间延时避免请求过快 time.sleep(random.uniform(0.5, 1.5)) # 股票间延时 time.sleep(random.uniform(2, 4)) # 合并所有数据并保存 if all_data_frames: final_df pd.concat(all_data_frames, ignore_indexTrue) # 按股票代码和报告日期排序 final_df.sort_values(by[股票代码, 报告日期], inplaceTrue) output_file os.path.join(output_dir, f{report_type}_sheet_{start_date}_to_{end_date}.csv) final_df.to_csv(output_file, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 logger.info(f数据已保存至: {output_file}, 共 {len(final_df)} 条记录。) else: logger.warning(未获取到任何数据。)这个主控流程实现了分页爬取、错误跳过、进度显示和结果合并。你可以通过调整stock_codes列表来控制爬取范围。5. 数据存储、增量更新与工程化考量一次性爬取全量数据后如何存储和后续更新是另一个关键问题。5.1 存储方案选择CSV vs. 数据库CSV/Excel文件适合数据量不大例如只爬取几百家公司最近几年的数据、一次性分析的项目。优点是简单直观无需搭建数据库环境。本项目初步输出结果可以使用CSV。数据库SQLite/MySQL/PostgreSQL如果数据量庞大全市场历史数据或需要频繁查询、关联分析数据库是更好的选择。SQLite是一个轻量级的选择无需服务器单个文件即可管理非常适合个人项目。使用pandas的to_sql方法和sqlalchemy库可以轻松地将DataFrame写入数据库。使用SQLite进行存储的示例import sqlite3 from sqlalchemy import create_engine # 创建数据库连接 engine create_engine(sqlite:///eastmoney_finance.db) # 将DataFrame写入数据库表 final_df.to_sql(namebalance_sheet, conengine, if_existsappend, indexFalse)if_existsappend参数允许我们进行增量更新。但需要注意直接追加可能导致重复数据需要在插入前根据“股票代码”和“报告日期”做去重判断。5.2 实现增量更新机制一个健壮的爬虫应该支持增量更新只爬取新的数据避免重复劳动和请求浪费。记录爬取状态在数据库或一个状态文件中记录每只股票最后一次成功爬取到的报告日期。爬取前检查每次运行爬虫时先读取状态只请求该股票在最后记录日期之后的数据。更新状态成功爬取并存储新数据后更新该股票的“最后爬取日期”状态。这需要修改主控流程增加状态管理的逻辑。一个简单的实现是使用一个JSON文件或一个专门的status数据库表来存储{stock_code: last_crawled_date}。5.3 日志记录与监控完善的日志对于排查问题至关重要。我们之前已经使用了Python的logging模块。应该将日志分级INFO, WARNING, ERROR输出到文件和控制台。# 在项目初始化时配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(crawler.log, encodingutf-8), logging.StreamHandler() ] )定期检查crawler.log文件可以了解爬虫的运行状况、错误频率及时发现IP被封或接口变更等问题。6. 常见问题排查与实战经验分享在实际操作中你几乎一定会遇到下面这些问题。这里是我踩过坑后总结的应对方法。6.1 请求被拒绝或返回空数据症状requests返回403、404或者返回的HTML/JSON里没有目标数据。排查步骤检查Headers特别是User-Agent和Referer。用浏览器开发者工具复制一次成功请求的所有Headers尽量模拟得一模一样。有时Cookie也很关键。检查IP短时间内请求过多IP可能被临时封禁。这就是为什么需要控制频率和使用代理。如果你在本地运行可以尝试重启路由器更换公网IP或者使用代理。验证接口有效性东方财富网的接口地址和参数名可能发生变化。务必用最新的开发者工具抓包确认当前的接口URL和参数格式。尝试动态渲染如果上述方法都无效数据可能是由非常复杂的JS生成的。这时可以祭出备用方案Selenium或Playwright直接模拟浏览器操作获取渲染后的页面源码再用BeautifulSoup解析。但这会慢很多。6.2 数据解析错误或字段缺失症状KeyError找不到字典键或者解析出来的DataFrame列名不对、数据为空。解决方案打印原始响应在解析函数里先把json_data完整地打印或保存下来仔细查看其结构。接口返回的字段名可能和你预期的不一样。动态构建字段映射不要硬编码字段映射字典。可以先写一个函数打印出第一次成功请求返回的JSON的所有键名然后根据这个列表来创建或更新你的column_mapping。处理嵌套结构有时数据可能嵌套在多层字典或列表里。你需要使用.get()方法安全地访问并做好空值判断。6.3 爬取速度慢与效率优化问题A股有几千家上市公司每家又有多年多期报表串行爬取会非常慢。优化方案并发/异步请求使用concurrent.futures的ThreadPoolExecutor可以实现多线程并发请求显著提升速度。但必须谨慎过高的并发会导致IP被迅速封禁。建议控制线程数如3-5个并为每个线程配置独立的代理IP。调整请求参数如果接口支持尽量在一次请求中获取更多数据如增大pageSize减少请求次数。分离IO和计算将网络请求IO密集型和数据分析CPU密集型分开。可以用队列queue模式生产者线程负责请求和下载原始数据消费者线程负责解析和存储。重要经验对于公开网站保持礼貌的爬取节奏是第一原则。在速度和数据完整性之间优先选择后者。设置合理的延时使用代理池并准备好当爬虫因各种原因中断后能从中断点恢复的机制比如记录已成功爬取的股票代码。6.4 法律与伦理风险规避遵守robots.txt检查https://www.eastmoney.com/robots.txt尊重网站规定的不允许爬取的目录。限制爬取频率这是最重要的伦理准则。我们的代码中已经加入了随机延时避免对目标服务器造成压力。数据用途爬取的数据应用于个人学习、研究或内部分析。切勿用于商业用途、公开大量分发或从事任何可能干扰网站正常服务的行为。标识自己在请求头中可以考虑添加一个自定义头如From: your-emailexample.com表明身份和善意方便网站管理员联系。这个项目从分析到实现涵盖了网络爬虫在真实场景下的主要挑战和解决方案。它不仅仅是一个脚本更是一个需要考虑健壮性、可维护性和伦理规范的小型数据工程。当你成功运行起这个爬虫并看到规整的财务数据源源不断地保存下来时那种成就感是实实在在的。剩下的就是基于这些高质量的数据去施展你在金融分析或人工智能算法上的才华了。本文还有配套的精品资源点击获取