Python 爬取财务报表

在本文中，我们将介绍如何使用Python编写一个简单的数据抓取器，用于爬取东方财富网上的各类财务报表数据。我们将利用requests和lxml库进行数据请求和解析，并将抓取到的数据保存到CSV文件中。

1. 准备工作

首先，确保你已经安装了以下Python库：

pip install requests
pip install lxml

2. 创建数据抓取器

我们将创建一个名为DataScraper的类，用于封装所有数据抓取相关的方法。数据抓取器的主要功能包括：

获取报表数据
解析并提取表头信息
将数据写入CSV文件

2.1 初始化

在DataScraper类的__init__方法中，我们将初始化一些必要的属性，如报表类型、报表名称等。此外，我们还需要设置请求URL和请求头，以便稍后进行数据请求。

class DataScraper:def __init__(self):self.pagename_type = {# ...}self.pagename_en = {# ...}self.en_list = []self.url = 'https://datacenter-web.eastmoney.com/api/data/v1/get'self.headers = {# ...}

2.2 获取报表数据

我们定义一个名为get_table的方法，用于向东方财富网发送请求并获取报表数据。传入页数作为参数，返回当前页的报表数据。

    def get_table(self, page):# ...

2.3 解析表头

在抓取数据之前，我们需要解析表头信息。我们创建一个名为get_header的方法，传入一个包含所有英文表头的列表。该方法将请求报表页面，使用lxml库解析HTML，并提取中文表头信息。

    def get_header(self, all_en_list):# ...

2.4 写入表头

接下来，我们创建一个名为write_header的方法，用于将解析到的表头信息写入CSV文件。在该方法中，我们首先调用get_header方法获取表头信息，然后使用csv.writer将其写入CSV文件。

    def write_header(self, table_data):# ...

2.5 写入报表数据

定义一个名为write_table的方法，用于将抓取到的报表数据逐行写入CSV文件。在该方法中，我们遍历抓取到的数据，并将每一行数据写入CSV文件。

    def write_table(self, table_data):# ...

2.6 获取时间列表

为了让用户选择爬取的报表时间，我们定义一个名为get_timeList的方法。该方法将发送请求到东方财富网，解析并提取可选的时间列表。

    def get_timeList(self):# ...

3 使用数据抓取器

在创建好DataScraper类之后，我们可以使用以下代码来实例化它并爬取所需的报表数据：

if __name__ == '__main__':scraper = DataScraper()timeList = scraper.get_timeList()for index, value in enumerate(timeList):if (index + 1) % 5 == 0:print(value)else:print(value, end=' ; ')timePoint = str(input('\n请选择时间（可选项如上）:'))pagename = str(input('请输入报表类型（业绩报表;业绩快报;业绩预告;预约披露时间;资产负债表;利润表；现金流量表）:'))# 校验输入assert timePoint in timeList, '时间输入错误'assert pagename in list(scraper.pagename_type.keys()), '报表类型输入错误'table_type = scraper.pagename_type[pagename]filename = f'{pagename}_{timePoint}.csv'# 写入表头scraper.write_header(scraper.get_table(1))# 循环遍历所有页数page = 1while True:table = scraper.get_table(page)if table:scraper.write_table(table)else:breakpage += 1

4 完整代码及结果截图

import csv
import json
import requests
from lxml import etreeclass DataScraper:def __init__(self):self.pagename_type = {"业绩报表": "RPT_LICO_FN_CPD","业绩快报": "RPT_FCI_PERFORMANCEE","业绩预告": "RPT_PUBLIC_OP_NEWPREDICT","预约披露时间": "RPT_PUBLIC_BS_APPOIN","资产负债表": "RPT_DMSK_FN_BALANCE","利润表": "RPT_DMSK_FN_INCOME","现金流量表": "RPT_DMSK_FN_CASHFLOW"}self.pagename_en = {"业绩报表": "yjbb","业绩快报": "yjkb","业绩预告": "yjyg","预约披露时间": "yysj","资产负债表": "zcfz","利润表": "lrb","现金流量表": "xjll"}self.en_list = []self.url = 'https://datacenter-web.eastmoney.com/api/data/v1/get'self.headers = {'Accept': '*/*','Accept-Language': 'zh-CN,zh;q=0.9','Connection': 'closed','Referer': 'https://data.eastmoney.com/','User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/111.0.0.0 Safari/537.36','sec-ch-ua': '"Google Chrome";v="111", "Not(A:Brand";v="8", "Chromium";v="111"','sec-ch-ua-mobile': '?0','sec-ch-ua-platform': '"Windows"'}def get_table(self, page):params = {'sortTypes': '-1,-1','reportName': self.table_type,'columns': 'ALL','filter': f'(REPORT_DATE=\'{self.timePoint}\')'}if self.table_type in ['RPT_LICO_FN_CPD']:params['filter'] = f'(REPORTDATE=\'{self.timePoint}\')'params['pageNumber'] = str(page)response = requests.get(url=self.url, params=params, headers=self.headers)data = json.loads(response.text)if data['result']:return data['result']['data']else:returndef get_header(self, all_en_list):ch_list = []url = f'https://data.eastmoney.com/bbsj/{self.pagename_en[self.pagename]}.html'response = requests.get(url)res = etree.HTML(response.text)for en in all_en_list:ch = ''.join([i.strip() for i in res.xpath(f'//div[@class="dataview"]//table[1]//th[@data-field="{en}"]//text()')])if ch:ch_list.append(ch)self.en_list.append(en)return ch_listdef write_header(self, table_data):with open(self.filename, 'w', encoding='utf-8', newline='') as f:writer = csv.writer(f)headers = self.get_header(list(table_data[0].keys()))writer.writerow(headers)def write_table(self, table_data):with open(self.filename, 'a', encoding='utf-8', newline='') as csvfile:writer = csv.writer(csvfile)for item in table_data:row = []for key in item.keys():if key in self.en_list:row.append(str(item[key]))print(row)writer.writerow(row)def get_timeList(self):headers = {'Referer': 'https://data.eastmoney.com/bbsj/202206.html',}response = requests.get('https://data.eastmoney.com/bbsj/202206.html', headers=headers)res = etree.HTML(response.text)return res.xpath('//*[@id="filter_date"]//option/text()')def run(self):self.timeList = self.get_timeList()for index, value in enumerate(self.timeList):if (index + 1) % 5 == 0:print(value)else:print(value, end=' ; ')self.timePoint = str(input('\n请选择时间（可选项如上）:'))self.pagename = str(input('请输入报表类型（业绩报表;业绩快报;业绩预告;预约披露时间;资产负债表;利润表；现金流量表）:'))assert self.timePoint in self.timeList, '时间输入错误'assert self.pagename in list(self.pagename_type.keys()), '报表类型输入错误'self.table_type = self.pagename_type[self.pagename]self.filename = f'{self.pagename}_{self.timePoint}.csv'self.write_header(self.get_table(1))page = 1while True:table = self.get_table(page)if table:self.write_table(table)else:breakpage += 1if __name__ == '__main__':scraper = DataScraper()scraper.run()

关于更多东方财富网、巨潮网、中国知网的爬虫，欢迎来参观我的github仓库