当前位置：首页 > news >正文

搭建自己的金融数据源和量化分析平台（七）：定时更新上市公司所属行业门类及大类

news 2025/11/22 18:19:09

0x00 前言

由于此前从深交所下载的股票信息中只有行业门类信息，没有行业大类信息，导致后续解析三大报表和量化选股的时候无法进行：
在这里插入图片描述
可以看到深交所的股票是没有大类信息的。
再看看上交所的保险股：

因此需要将深交所股票的所属大类信息也添加上。
这里可以直接使用中国上市公司协会每隔一段时间发布的《上市公司行业分类结果》。
目前最新版本是《2023年下半年上市公司行业分类结果》
在这里插入图片描述
具体的解析逻辑不再赘述，分析一下HTML的格式就能把最新的pdf拿到手来解析。
直接上爬虫代码：

import osimport pdfplumber
import requests
from lxml import etree'''中国上市公司协会的爬虫，读取和解析最新上市公司行业分类结果 返回格式为：股票代码:[一级行业代码,二级行业代码]。举例如下
{"stock_code1":[industry,industry_2],"stock_code2":[industry,industry_2]
}
'''
def get_A_industry_list():basic_url = 'https://www.capco.org.cn/pub/zgssgsxh/xhgg/hyfl/hyfljg/index.html'mid_url = 'https://www.capco.org.cn/pub/zgssgsxh/xhgg/hyfl/hyfljg/'cache_file_path = "./corporation_category.pdf"response = requests.get(basic_url)response.encoding = 'UTF-8'href_cut = etree.HTML(response.text).xpath(".//div[@class='fr listCon']/h3/a")response.close()href_mid = etree.tostring(element_or_tree=href_cut[0], encoding='utf-8').decode('utf-8')latest_result = href_mid.split("<a href=\"")[1].split("\">")[0].split("./")[1]response = requests.get(mid_url+latest_result)response.encoding = 'UTF-8'href_cut = etree.HTML(response.text).xpath(".//a[@style='font-size:12px; color:#0066cc;']")response.close()pdf_url_mid = etree.tostring(element_or_tree=href_cut[0], encoding='utf-8').decode('utf-8')pdf_url = pdf_url_mid.split("href=\"")[1].split("\" title=\"")[0]response = requests.get(pdf_url)open(cache_file_path, "wb").write(response.content)response.close()result = {}with pdfplumber.open(cache_file_path) as pdf:for page in pdf.pages:tables = page.extract_tables()for table in tables:for line in table:if line[0].find("上市公司") < 0:result[line[0]] = [line[2], line[2]+line[6]]os.remove(cache_file_path)return result

然后控制器那边这样写：

# 更新上市公司所属行业门类及大类
def update_A_corporation_category():database = "stock_a"select_sql = "SELECT stock_code,industry,industry_2 FROM stock_list"update_sql = "update stock_list set industry=%s,industry_2=%s where stock_code=%s"update_rows = []category = get_A_industry_list()select_result = ExecSelect(database, select_sql)  # 读取查询结果for stock in select_result:if stock[2] is None:try:update_rows.append((category[stock[0]][0], category[stock[0]][1], stock[0]))except KeyError:print(stock[0], "暂无大类分类结果")continue# 更新数据库中存在的股票信息if len(update_rows) > 0:result = ExecInsert(database, update_sql, update_rows)if result == 'success':print("更新上市公司行业分类成功.")else:raise CustomException("更新上市公司行业分类时发生数据库异常:" + result)print("上市公司行业分类更新结束.")