简介本资源是一套面向人工智能与数据分析初学者的实战型网络爬虫项目聚焦于从东方财富网批量获取上市公司结构化财务数据解决金融数据采集难、动态页面解析复杂等实际问题。压缩包共12个文件含10个CSV格式财报数据表涵盖资产负债表、利润表、现金流量表、业绩预告及预约披露时间表等和2个核心爬虫脚本Selenium与Requests双实现总大小19.88MB其中eastmoney_crawler2.py采用Requests逆向分析方案效率较Selenium版本提升百倍具备生产级可用性。已有351人学习下载配套完整运行说明支持按年份、季度、报表类型及页码范围灵活配置结果自动落盘为标准CSV便于后续导入Pandas或接入AI建模流程。1. 项目缘起从数据焦虑到自动化解决方案作为一名长期在金融科技和数据科学领域摸爬滚打的从业者我深知数据是驱动一切分析、决策和模型训练的基石。无论是做基本面分析、量化策略回测还是构建企业信用风险模型上市公司的财务报表数据都是不可或缺的“燃料”。然而获取这些数据的过程常常让人感到一种深深的“数据焦虑”——手动从东方财富网这类财经门户网站复制粘贴不仅效率低下、容易出错而且当需要追踪成百上千家公司、跨越多个报告期时这项工作几乎不可能靠人力完成。几年前为了一个企业财务健康度评估项目我需要获取A股市场近五年所有上市公司的三大报表资产负债表、利润表、现金流量表数据。最初的尝试是笨拙的打开浏览器搜索公司进入财务页面找到报表手动选取数据区域复制到Excel……做了不到十家公司我就意识到这条路行不通。时间成本高得离谱而且数据格式不统一后续清洗工作量巨大。这正是我转向网络爬虫技术的契机——用程序自动化地完成这项繁琐、重复但至关重要的数据采集工作。“人工智能-项目实践-网络爬虫-爬取东方财富网上市公司的财务报表数据”这个项目正是这种需求的典型体现。它不是一个炫技的复杂算法演示而是一个解决实际业务痛点的、具有高度实用价值的工程实践。通过Python爬虫技术我们可以系统化、批量化地从东方财富网抓取结构化的财务数据为后续的人工智能分析如财务造假识别、业绩预测、行业对比提供高质量的数据源。本篇文章我将以一个过来人的身份手把手拆解这个项目的完整实现逻辑、技术细节以及我踩过的那些“坑”目标是让你不仅能复现这个爬虫更能理解其背后的设计思想并能举一反三应用到其他类似的数据采集场景中。2. 目标网站分析与爬虫策略设计在动手写一行代码之前花时间透彻分析目标网站的结构和行为是决定爬虫项目成败的关键。盲目开干很容易撞上反爬机制或者写出脆弱、难以维护的代码。我们对东方财富网以quote.eastmoney.com等子域为例的财务报表页面进行了深入剖析。2.1 页面结构与数据定位东方财富网的财务数据通常以两种形式呈现HTML表格和异步加载的JSON数据。通过浏览器开发者工具F12的“网络”Network选项卡进行观察是发现真相的利器。静态页面分析访问一个具体的公司财务页面例如http://quote.eastmoney.com/concept/sz000002.html?f财务分析查看页面源代码。你会发现核心的财务报表数据并非直接写在初始HTML中而是通过后续的JavaScript请求动态加载的。这意味着简单的requests.get()无法直接拿到数据。动态请求捕获在开发者工具的Network选项卡中筛选XHR/Fetch请求然后操作页面如下拉选择不同报表类型或报告期。你会看到浏览器向特定API地址发送了请求并返回了结构清晰的JSON数据。例如利润表数据可能来自一个类似http://datacenter.eastmoney.com/api/data/get?typeRPT_LICO_FN_CPD的接口。这个接口地址和其参数就是我们爬虫需要模拟的核心。参数逆向工程仔细查看这个API请求的“标头”Headers和“负载”Payload。关键信息通常在“查询字符串参数”Query String Parameters或“请求负载”Request Payload中。常见的参数包括stockCode: 股票代码如000002。reportType: 报告类型如年报、季报。reportDate: 报告期如2023-12-31。可能还有pageNum,pageSize用于分页以及一些固定的type、sty等标识参数。此外请求头Headers中的User-Agent,Referer, 有时甚至是Cookie或特定的Token都可能被服务器用于校验请求的合法性。注意东方财富网的API接口地址和参数格式并非一成不变可能会随着网站改版而更新。本文基于当前2024年可观察到的模式进行讲解你在实践时务必使用开发者工具进行实时验证。2.2 反爬机制与应对策略东方财富网作为主流财经网站具备一定的反爬虫能力。我们的策略必须稳健且尊重对方服务器。频率限制Rate Limiting这是最常见的反爬措施。我们的策略是在请求间添加随机延时。不要使用固定的sleep(1)而是使用time.sleep(random.uniform(1, 3))这样的随机间隔让请求行为更接近人类。请求头校验必须设置合理的请求头。一个最基本的配置应包括headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Referer: http://quote.eastmoney.com/, # 通常需要设置为数据页面的来源域名 Accept: application/json, text/javascript, */*; q0.01, Accept-Language: zh-CN,zh;q0.9, }IP封锁如果请求过于频繁可能会触发IP封锁。对于个人学习和少量数据抓取遵守规则、降低频率是首要原则。对于大规模抓取需要考虑使用代理IP池但这涉及更高成本和更复杂的管理且必须确保代理的合法性本文不展开讨论。数据校验获取到的JSON数据首先要检查其状态码或success字段。例如很多API返回的JSON中有一个success键值为true才表示成功。同时要检查数据部分如result或data是否为空。核心策略总结我们的爬虫将采用“模拟浏览器API请求”的方式而非解析HTML。这种方式更直接、更稳定因为数据源头就是JSON。我们需要做的就是1) 找到正确的API地址2) 构造合法的请求参数和头信息3) 以合理的频率发送请求4) 解析返回的JSON数据。3. 技术栈选型与核心工具详解工欲善其事必先利其器。选择一个合适的技术栈能让开发事半功倍。以下是本项目经过实践检验的核心工具包。3.1 核心库Requests 与 PandasRequestsPython中公认的HTTP库之王简单易用功能强大。我们将用它来发送所有HTTP请求获取API返回的JSON数据。import requests response requests.get(url, headersheaders, paramsparams, timeout10) data response.json() # 直接解析为Python字典/列表为什么选它相比Python内置的urllibrequests的API设计极其人性化代码简洁明了。社区支持强大遇到问题容易找到解决方案。Pandas数据分析和处理的瑞士军刀。我们爬取到的财务数据通常是列表形式的字典Pandas可以轻松地将其转换为结构化的DataFrame方便进行数据清洗、转换和最终导出为Excel或CSV。import pandas as pd df pd.DataFrame(data_list) df.to_excel(财务报表.xlsx, indexFalse)为什么选它对于表格型数据Pandas提供了无与伦比的便捷性。后续若需进行简单的数据透视或计算如计算同比增长率Pandas也能无缝衔接。3.2 辅助库Time, Random, JSON, osTime Random用于控制请求频率模拟人类操作避免触发反爬。import time import random time.sleep(random.uniform(0.5, 2.0)) # 随机等待0.5到2秒JSON用于解析API返回的字符串虽然requests的.json()方法更常用以及在调试时美观地打印字典数据。import json print(json.dumps(data, indent2, ensure_asciiFalse)) # 漂亮打印支持中文os用于创建目录、检查文件路径实现数据的按公司、按类型有序存储。import os if not os.path.exists(‘data/财务报表’): os.makedirs(‘data/财务报表’)3.3 开发环境与工程化建议IDE推荐使用VSCode或PyCharm。它们对代码提示、调试特别是设置断点查看网络请求返回的数据结构支持非常好。虚拟环境务必使用venv或conda创建独立的Python环境使用requirements.txt文件管理依赖。# requirements.txt requests2.28.0 pandas1.5.0 openpyxl3.0.0 # 用于Pandas导出Excel项目结构即使是小项目良好的结构也有助于维护。financial_spider/ ├── main.py # 主程序入口 ├── config.py # 配置文件API地址、请求头、股票列表等 ├── spider.py # 核心爬虫类 ├── data_processor.py # 数据处理与保存类 ├── utils.py # 工具函数延时、日志等 ├── requirements.txt └── data/ # 数据输出目录 ├── balance_sheet/ ├── income_statement/ └── cash_flow/实操心得不要把所有代码都堆在一个文件里。将网络请求、数据处理、文件存储逻辑分离会让代码更清晰也更容易定位问题。例如当解析逻辑出错时你只需要检查data_processor.py当请求失败时则查看spider.py。4. 爬虫核心实现分步拆解与代码实战接下来我们进入最核心的编码环节。我将把爬虫拆解成几个关键函数并附上详细的代码和注释。4.1 步骤一获取单家公司单张报表数据这是最基础的单元操作。假设我们已经通过分析找到了利润表的API接口和参数规律。import requests import pandas as pd import time import random def fetch_financial_data(stock_code, report_type利润表, report_date2023-12-31): 获取指定公司、指定报告期的财务数据 :param stock_code: 股票代码如 000002 (不带市场前缀) :param report_type: 报表类型如 利润表、资产负债表、现金流量表 :param report_date: 报告期如 2023-12-31 :return: 包含数据的Pandas DataFrame如果失败返回None # 1. 定义API基础URL和请求头 (需要根据实际情况调整) # 注意以下URL和参数仅为示例务必用开发者工具获取最新的 base_url http://datacenter.eastmoney.com/api/data/get # 映射报表类型到具体的API参数 report_type_map { 利润表: {type: RPT_LICO_FN_CPD, sty: SECURITY_CODE,REPORT_DATE}, 资产负债表: {type: RPT_FINA_BALANCE_SHEET, sty: SECURITY_CODE,REPORT_DATE}, 现金流量表: {type: RPT_FINA_CASH_FLOW, sty: SECURITY_CODE,REPORT_DATE}, } if report_type not in report_type_map: print(f不支持的报表类型: {report_type}) return None api_params report_type_map[report_type] # 2. 构造查询参数 params { type: api_params[type], sty: api_params[sty], filter: f(SECURITY_CODE{stock_code})(REPORT_DATE\{report_date}\), p: 1, # 页码 ps: 50, # 每页大小一般足够 sr: -1, st: REPORT_DATE, source: WEB, client: WEB, } 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, Referer: fhttp://quote.eastmoney.com/concept/sz{stock_code}.html, Accept: application/json, text/javascript, */*; q0.01, } # 3. 发送HTTP GET请求 try: response requests.get(base_url, paramsparams, headersheaders, timeout15) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 json_data response.json() # 4. 解析JSON响应 if json_data.get(success) is True: # 数据通常在 result - data 路径下 data_list json_data.get(result, {}).get(data, []) if data_list: df pd.DataFrame(data_list) print(f成功获取 {stock_code} {report_date} {report_type}共{len(df)}行数据。) return df else: print(f警告{stock_code} {report_date} {report_type} 返回数据为空。) return pd.DataFrame() # 返回空DataFrame else: print(f请求失败{json_data.get(message, 未知错误)}) return None except requests.exceptions.RequestException as e: print(f网络请求异常: {e}) return None except ValueError as e: print(fJSON解析异常: {e}) return None # 5. 请求后随机延时礼貌爬取 time.sleep(random.uniform(1, 2))代码解读与避坑点参数动态化filter参数中的股票代码和报告期是动态的我们通过函数参数传入。注意报告期日期的格式可能需要与API要求严格一致如2023-12-31。异常处理使用try...except包裹网络请求和JSON解析避免程序因单次请求失败而崩溃。response.raise_for_status()能有效捕获HTTP错误如404 403。数据校验不仅检查HTTP状态码还要检查JSON中的业务状态码如success和实际数据是否为空。有时API返回success: true但data为空这可能是该公司该期确实没有数据。延时策略函数末尾的sleep是必须的这是爬虫的道德底线和技术保障。4.2 步骤二批量获取多家公司多期数据单次获取是基础批量获取才是生产力。我们需要一个股票代码列表并循环遍历。def batch_fetch_data(stock_list, report_types, start_date2020-12-31, end_date2023-12-31, freqyear): 批量获取财务数据 :param stock_list: 股票代码列表如 [000001, 000002, 600000] :param report_types: 需要获取的报表类型列表如 [利润表, 资产负债表] :param start_date: 开始报告期 :param end_date: 结束报告期 :param freq: 报告频率year代表年报quarter代表季报需要调整API参数 :return: 一个字典键为(股票代码, 报表类型)值为DataFrame列表按报告期 all_data {} # 生成报告期列表简化版仅生成年末日期 # 注意实际项目中需要根据freq生成季度末或年度末日期列表 report_dates [2020-12-31, 2021-12-31, 2022-12-31, 2023-12-31] total_requests len(stock_list) * len(report_types) * len(report_dates) request_count 0 for stock_code in stock_list: for rpt_type in report_types: key (stock_code, rpt_type) all_data[key] [] for rpt_date in report_dates: request_count 1 print(f[进度 {request_count}/{total_requests}] 正在获取 {stock_code} - {rpt_type} - {rpt_date}) df fetch_financial_data(stock_code, rpt_type, rpt_date) if df is not None and not df.empty: # 可以为DataFrame添加标识列 df[股票代码] stock_code df[报表类型] rpt_type df[报告期] rpt_date all_data[key].append(df) else: print(f 跳过 {stock_code} {rpt_date} {rpt_type}) # 更长的批次间延时防止IP被封锁 time.sleep(random.uniform(2, 4)) return all_data关键设计进度提示在循环中加入进度打印对于长时间运行的爬虫至关重要让你知道程序还在正常工作以及大概需要多久。数据结构使用字典all_data来组织数据键是(股票代码 报表类型)值是该组合下所有报告期DataFrame的列表。这种结构便于后续按公司或按报表类型进行合并处理。批次延时在每完成一次完整的公司-报表-日期请求后设置一个更长的延时2-4秒这比在每次fetch_financial_data内部延时更有利于控制整体节奏避免短时间内对同一主机发起过多连接。4.3 步骤三数据清洗、整合与存储爬取到的原始数据往往包含无关字段、格式不一需要进行清洗和整合。def clean_and_save_data(all_data, output_dir./data): 清洗数据并保存到本地文件 :param all_data: batch_fetch_data返回的字典 :param output_dir: 输出根目录 import os for (stock_code, rpt_type), df_list in all_data.items(): if not df_list: continue # 1. 纵向合并同一公司同一报表的所有期数据 combined_df pd.concat(df_list, ignore_indexTrue) # 2. 数据清洗示例 # a. 重命名列将API返回的英文/缩写列名改为中文 # 需要根据实际API返回的字段名建立映射关系 column_name_map { SECURITY_NAME_ABBR: 公司简称, REPORT_DATE: 报告期, TOTAL_OPERATE_INCOME: 营业总收入, OPERATE_INCOME: 营业收入, NETPROFIT: 净利润, # ... 更多字段映射 } combined_df.rename(columnscolumn_name_map, inplaceTrue) # b. 筛选需要的列 useful_columns [股票代码, 公司简称, 报告期, 营业总收入, 营业收入, 净利润] # 按需添加 combined_df combined_df[[col for col in useful_columns if col in combined_df.columns]] # c. 处理缺失值和异常值示例 # 将某些字段转换为数值类型错误强制转换为NaN numeric_cols [营业总收入, 净利润] for col in numeric_cols: if col in combined_df.columns: combined_df[col] pd.to_numeric(combined_df[col], errorscoerce) # 3. 按报表类型分目录保存 type_dir os.path.join(output_dir, rpt_type) os.makedirs(type_dir, exist_okTrue) # 保存为CSV和Excel双备份 csv_path os.path.join(type_dir, f{stock_code}_{rpt_type}.csv) excel_path os.path.join(type_dir, f{stock_code}_{rpt_type}.xlsx) combined_df.to_csv(csv_path, indexFalse, encodingutf-8-sig) # utf-8-sig解决Excel中文乱码 combined_df.to_excel(excel_path, indexFalse) print(f已保存: {csv_path})清洗要点列名映射API返回的字段名可能是英文或缩写将其映射为有业务意义的中文名极大提升数据可读性。字段筛选财务报表字段众多只保留你分析所需的列能简化后续处理。类型转换财务数据是数值型的但爬取下来可能是字符串。使用pd.to_numeric(..., errors‘coerce’)安全转换无效值会变成NaN便于后续处理。多格式保存同时保存为CSV轻量通用和Excel便于手动查看是个好习惯。utf-8-sig编码可以确保用Excel打开CSV时中文不乱码。4.4 步骤四主程序流程与错误恢复机制将以上模块组装起来并增加健壮性设计。def main(): # 配置 stock_codes [000001, 000002, 600036, 600519] # 示例股票池 report_types [利润表, 资产负债表] output_directory ./financial_data # 创建输出目录 os.makedirs(output_directory, exist_okTrue) # 尝试加载已有的进度简易断点续传 progress_file os.path.join(output_directory, progress.log) completed_pairs set() if os.path.exists(progress_file): with open(progress_file, r, encodingutf-8) as f: completed_pairs set(line.strip() for line in f) print(开始批量爬取财务数据...) all_data {} for stock_code in stock_codes: for rpt_type in report_types: pair_key f{stock_code}|{rpt_type} if pair_key in completed_pairs: print(f跳过已完成的: {pair_key}) continue print(f\n 处理 {stock_code} - {rpt_type} ) # 这里简化处理实际应调用batch_fetch_data的逻辑但只处理当前pair # 为了示例我们假设调用一个单线程版本 df_list fetch_data_for_stock_type(stock_code, rpt_type) # 假设的函数 if df_list: all_data[(stock_code, rpt_type)] df_list # 记录进度 with open(progress_file, a, encodingutf-8) as f: f.write(pair_key \n) else: print(f 获取失败已记录。) # 清洗并保存所有成功获取的数据 if all_data: clean_and_save_data(all_data, output_directory) print(\n所有数据爬取与保存完成) else: print(\n未获取到任何有效数据。) if __name__ __main__: main()错误恢复与健壮性进度日志实现一个简单的“断点续传”机制。将成功处理的(股票代码 报表类型)对记录到日志文件中。如果程序中途因网络或错误中断重新运行时可以跳过已完成的组合从断点处开始。这对于爬取大量数据时非常有用。异常隔离在batch_fetch_data或循环中单次请求的失败不应导致整个程序崩溃。我们的设计已经通过fetch_financial_data返回None或空DataFrame并在上层判断来处理了这一点。5. 高级话题效率优化与反爬进阶当数据量变大时基础的循环爬取会非常慢。此外网站的反爬策略也可能升级。5.1 使用并发提升爬取效率顺序请求几百家公司多年的数据耗时可能长达数小时。使用concurrent.futures库进行线程池或进程池并发可以大幅缩短时间。from concurrent.futures import ThreadPoolExecutor, as_completed def fetch_single_task(args): 包装单个抓取任务供线程池调用 stock_code, rpt_type, rpt_date args return stock_code, rpt_type, rpt_date, fetch_financial_data(stock_code, rpt_type, rpt_date) def concurrent_batch_fetch(stock_list, report_types, report_dates, max_workers5): 使用线程池并发爬取 :param max_workers: 并发线程数不宜过大如3-5避免对目标服务器造成压力 all_tasks [] for s in stock_list: for t in report_types: for d in report_dates: all_tasks.append((s, t, d)) results {} with ThreadPoolExecutor(max_workersmax_workers) as executor: future_to_task {executor.submit(fetch_single_task, task): task for task in all_tasks} for future in as_completed(future_to_task): stock_code, rpt_type, rpt_date, df future.result() if df is not None and not df.empty: key (stock_code, rpt_type) df[股票代码] stock_code df[报表类型] rpt_type df[报告期] rpt_date results.setdefault(key, []).append(df) # 注意并发环境下延时控制需要更精细可以考虑在fetch_single_task内部或使用速率限制器。 return results重要警告并发爬取是一把双刃剑。它虽然快但会显著增加对方服务器的负载更容易触发反爬机制如IP封锁。务必谨慎使用将max_workers设置得很小例如3或5。在fetch_single_task函数内部保留随机延时。最好用于抓取那些明确允许或反爬不严的公开数据。对于东方财富网这类主流网站建议优先使用低速、礼貌的单线程爬取除非你拥有充足的代理IP池和更复杂的反反爬策略。5.2 应对动态Token与签名验证一些网站为了加强反爬会使用动态Token或对请求参数进行签名。这增加了爬虫的难度。现象在开发者工具中你会发现每次请求的URL中都有一个不断变化的参数如tokenabcd1234...或signxyz789...。解决方案寻找Token生成逻辑在网站的JavaScript源代码中搜索这个token或sign。通常它是由页面加载时返回的某个值或者由固定的算法如对某些参数进行MD5加密计算得出。你需要使用requests先获取这个种子值或者用Python复现其加密算法。使用Selenium或Playwright当JavaScript逻辑过于复杂难以逆向时可以动用浏览器自动化工具。让真实的浏览器加载页面执行JS然后从浏览器环境中提取出生成的Token或直接获取渲染后的数据。但这会极大降低爬取速度资源消耗也大应作为最后的手段。Session维持有时Token与登录态或Session相关。你可能需要先用requests.Session()对象模拟登录然后使用同一个Session去请求数据API服务器会自动关联Session和Token。实操心得遇到动态Token不要慌。90%的情况下通过仔细分析首个HTML页面或某个初始化接口的响应都能找到生成Token的线索。只有极少数情况才需要动用浏览器自动化。6. 数据质量校验与后续应用方向数据爬取下来不是终点确保数据质量并知道如何用才是价值所在。6.1 数据质量校验清单在将数据投入分析前请进行以下检查完整性检查检查每个公司、每种报表、各报告期的数据是否齐全。是否有大量NaN值可以使用df.isnull().sum()快速查看各列缺失情况。一致性检查同一家公司资产负债表是否满足“资产负债所有者权益”的会计恒等式允许微小计算误差可以利用爬取的数据进行验证。异常值检查查看关键财务指标如净利润、营业收入是否存在离谱的极大或极小值可能是单位错误或数据错误。重复数据检查检查是否有完全重复的行。格式统一确保“报告期”等字段格式一致方便后续按时间排序和分析。6.2 数据应用方向示例干净、结构化的财务数据是金矿可以支持多种人工智能和分析场景财务比率分析与可视化计算流动比率、资产负债率、净资产收益率ROE等经典指标使用Matplotlib或Plotly绘制趋势图进行公司间横向对比。财务风险预警模型利用多期数据构建逻辑回归、随机森林等机器学习模型尝试预测企业下一期是否会发生ST特别处理或财务困境。特征可以选取各种财务比率和增长指标。行业聚类分析使用无监督学习算法如K-Means对同一行业内公司的财务指标进行聚类发现不同的财务表现模式如稳健型、成长型、风险型。财报文本挖掘虽然本文聚焦数字报表但东方财富网也有财报全文和公告。可以结合NLP技术对管理层讨论与分析MDA部分进行情感分析或主题提取与财务数字相互印证。这个爬虫项目本身是一个典型的数据工程Data Engineering实践它为下游的数据科学Data Science和人工智能AI应用提供了坚实的数据基础。从我个人的经验来看在数据科学项目中用于数据采集、清洗和准备的时间往往占到整个项目周期的60%以上。因此掌握像这样的自动化数据获取技能不仅能解放你的双手更能让你将宝贵的精力集中在更有创造性的模型构建和业务分析上。本文还有配套的精品资源点击获取