目录
1. 背景介绍
2. 爬取 Wind 上市公司年报
2.1 获取年报地址
2.2 下载年报 PDF 文件
3. 爬取 CSMAR 上市公司年报
4. 总结
5. 附：文中涉及的主要 Python 代码
关于我们
目前，上市公司公告主要从巨潮网、上交所以及深交所等网站获取。爬取的步骤分为以下两步，一是获取公告地址，二是通过公告地址下载 PDF 文件。通常来说，获取公告地址比较麻烦。本文将根据 Wind 和 CSMAR 数据库的公告信息简化获取公告地址过程，并下载 PDF 文件，这里以年报为例。
首先，打开 Wind 终端，按照公司公告->年度报告->高级检索的步骤检索上市公司年报信息。
年报检索
其次，点击导出列表，将检索到的年报信息导出到 Excel 文件。需要注意的是，Wind 最多导出 9999 条记录。实际操作中，可以通过在检索中设置时间区间分批导出。
下载年报信息
最后，公告标题在 Excel 中是以超链接形式存在，通过 Excel 中的函数 FORMULATEXT() 可以将超链接分解成链接和文本两部分。进一步，可以通过调用 Python 进行拆分，提取公告地址链接。
获取年报地址
def address(str): #定义提取公告地址函数
return str.split(''')[1]
data['公告地址'] = data['公告地址'].apply(address)
打开年报地址，可以看到年报页面是由年报 PDF 超链接和年报内容两部分组成。其中，PDF 超链接可以通过 Xpath 或 正则表达式 获取。
年报地址内容
#利用Xpath提取年报PDF链接
def pdf_url(url):
html = requests.get(url).text #获取网页源代码
tree = etree.HTML(html) #解析网页
url = tree.xpath('//div[2]/a/@href') #获取PDF链接
return 'http://news.windin.com/ns/' + url[0]
data['PDF地址'] = data['公告地址'].apply(pdf_url)
#利用正则表达式提取年报PDF链接
def pdf_url(url):
html = requests.get(url).text
url = re.findall('(?<=href=).*(?=class)', html) #?<= 正则肯定式向后看 ?= 正则肯定式向前看
return 'http://news.windin.com/ns/' + url[0]
data['PDF地址'] = data['公告地址'].apply(pdf_url)
根据提取到的年报 PDF 链接下载文件。
for index, row in data.iterrows():
name = row['证券代码'][:6] + '_' + row['公告日期'] + '.pdf' #文件名称
url = row['PDF地址'] #pdf地址
times = 1 #失败后，重新获取次数
while times <= 3: #3次都失败后跳出循环
try:
urlretrieve(url, filename = r'./PDF/' + name) #下载pdf
print(f'成功下载{name}！')
break
except:
times += 1
print(f'休息5秒！再试第{times}次!')
time.sleep(5)
print('成功下载所有PDF文件！')
首先，打开 CSMAR 数据库，按照市场咨询系列->公告->公告分类关联表和公告证券关联表步骤，下载上市公司公告信息。其中，年报正文分类编码为 01030101。
CSMAR 公告信息
其次，根据公告分类关联表和公告证券关联表共同字段，合并两个表。
data1 = pd.read_excel('ANN_Classify.xlsx')[2:] #剔除前两行中文标题和单位
data2 = pd.read_excel('ANN_Security.xlsx')[2:]
data = pd.merge(data1, data2, on = ['AnnouncementID', 'DeclareDate', 'FullDeclareDate', 'Title'])
分类和证券关联表合并
最后，生成年报 PDF 链接，并下载。
#巨潮咨询网年报链接形式
#http://static.cninfo.com.cn/finalpage/2020-04-14/1207489969.PDF
#生成年报PDF链接
data['pdf_url'] = 'http://static.cninfo.com.cn/finalpage/' + data['DeclareDate'] + '/' + data['AnnouncementID'] + '.PDF'
#根据PDF链接下载
for index, row in data.iterrows():
name = row['Symbol'] + '_' + row['DeclareDate'] + '.pdf' #文件名称
print(name)
url = row['pdf_url'] #pdf地址
times = 1 #失败后，重新获取次数
while times <= 3: #3次都失败后跳出循环
try:
urlretrieve(url, filename = r'./PDF/' + name)
print(f'成功下载{name}！')
break
except:
times += 1
print(f'休息5秒！再试第{times}次!')
time.sleep(5)
print('成功下载所有PDF文件！')
通过数据库提供的公告信息，可以大大简化爬取公告难度。同时，相较于 CSMAR 而言，Wind 提供的公告信息更加详尽。
当然，爬取公告只是文本分析的开始，后续还有很多工作要做，如提取文本、分词、词频统计、关键词提取、情感分析、以及文本相似度计算等。
其中，从 PDF 中提取信息就是一项繁杂的工作。我们应该尽力去找那些以 网页形式 展示公告内容的网站，如 Wind、网易财经等，其次可以考虑用专门的 PDF转换软件 或 API 接口将 PDF 转换为TXT，
最后，可考虑 Python 的 pdfminer3k 和 pdfplumber 等包，并且 pdfplumber 可以很好提取表格数据。
关于图片文字识别，可以参考 百度AI 提供的文字识别 API 接口。
温馨提示： 文中链接在微信中无法生效。请点击底部「阅读原文」。
连享会 - 效率分析专题，2020年5月29-31日
📙 主讲嘉宾：连玉君 | 鲁晓东 | 张宁
🍏 课程主页：https://gitee.com/arlionn/TE
完整 .py 文件，可以到百度云盘下载：
https://pan.baidu.com/s/1Vr_-YCSJAmyoKvfayTQqpw如下是文中使用的主要代码：
#导入相应的包
import os
import pandas as pd
import requests
import re
from lxml import etree #解析网页
from urllib.request import urlretrieve #下载网络文件到本地
import time
#设置地址
os.chdir(r'D:\爬取上市公告') #修改为自己文件路径
os.getcwd()
#爬取上市公司年报_Wind
data = pd.read_excel('公司公告.xlsx')[:-1] #读入数据，并删除最后一行（最后一行为空值）
def address(str): #定义提取公告地址函数
return str.split(''')[1]
data['公告地址'] = data['公告地址'].apply(address)
def pdf_url(url): #利用Xpath提取年报PDF链接
html = requests.get(url).text
tree = etree.HTML(html) #解析网页
url = tree.xpath('//div[2]/a/@href') #获取PDF链接
return 'http://news.windin.com/ns/' + url[0]
data['PDF地址'] = data['公告地址'].apply(pdf_url)
for index, row in data.iterrows(): #下载年报
name = row['证券代码'][:6] + '_' + row['公告日期'] + '.pdf' #文件名称
url = row['PDF地址'] #pdf地址
times = 1 #失败后，重新获取次数
while times <= 3: #3次都失败后跳出循环
try:
urlretrieve(url, filename = r'./PDF/' + name)
print(f'成功下载{name}！')
break
except:
times += 1
print(f'休息5秒！再试第{times}次!')
time.sleep(5)
print('成功下载所有PDF文件！')
#爬取上市公司年报_CSMAR
data1 = pd.read_excel('ANN_Classify.xlsx')[2:] #剔除前两行中文标题和单位
data2 = pd.read_excel('ANN_Security.xlsx')[2:]
data = pd.merge(data1, data2, on = ['AnnouncementID', 'DeclareDate', 'FullDeclareDate', 'Title'])
data['pdf_url'] = 'http://static.cninfo.com.cn/finalpage/' + data['DeclareDate'] + '/' + data['AnnouncementID'] + '.PDF' #根据巨潮资讯网年报链接形式，生产PDF年报链接
for index, row in data.iterrows(): #下载前年报
name = row['Symbol'] + '_' + row['DeclareDate'] + '.pdf' #文件名称
print(name)
url = row['pdf_url'] #pdf地址
times = 1 #失败后，重新获取次数
while times <= 3: #3次都失败后跳出循环
try:
urlretrieve(url, filename = r'./PDF/' + name)
print(f'成功下载{name}！')
break
except:
times += 1
print(f'休息5秒！再试第{times}次!')
time.sleep(5)
print('成功下载所有PDF文件！')
连享会-在线课堂
http://lianxh.duanshu.com
免费公开课：
联系客服
