1. 项目缘起与核心价值最近在整理一些市场事件分析时我常常需要回溯特定监管事件比如上交所的问询函对相关上市公司股价的即时影响。手动去上交所官网翻公告再去行情软件里对日期查股价效率低不说还容易出错。作为一个习惯用代码解决问题的从业者我决定用Python把这两个环节打通实现一个自动化工具自动爬取上交所发布的问询函并精准定位到发函日然后获取对应股票的当日股价数据。这个项目的核心价值在于它将两个割裂的数据源监管公告和金融市场数据通过“时间”和“证券代码”这两个关键维度关联起来为事件驱动型分析、合规研究或简单的市场观察提供了一个可复现、可扩展的数据基础。无论是想研究监管问询的市场反应还是构建自己的事件分析数据库这个工具都能帮你省去大量重复劳动。整个过程涉及网络爬虫对半结构化网页的解析、关键信息的精准提取、以及金融数据API的调用是一次非常典型的“数据获取-数据清洗-数据关联”的实战。2. 目标网站分析与爬虫策略制定我们的数据源头是上海证券交易所的“监管问询”栏目。首先需要明确爬取的目标和策略。2.1 上交所问询函页面结构剖析上交所的监管问询公告通常发布在官方网站的特定板块。经过分析其列表页和详情页有以下几个特点列表页通常以分页形式展示URL中可能包含页码参数。每一条公告记录包含几个关键元素公告标题内含公司简称和股票代码、公告类型如“问询函”、发布日期即发函日以及一个指向详情页的链接。详情页包含了问询函的完整内容但对我们而言最重要的信息——公司全称、股票代码、发函日期——通常在列表页就已经足够。详情页可能用于后续的内容文本分析但本项目第一阶段聚焦于元数据获取。反爬机制作为重要机构的官网通常会有基本的反爬措施例如请求头校验、访问频率限制等。但一般不会像商业数据平台那样复杂合理的请求间隔和规范的请求头足以应对。关键点股票代码的提取是难点之一。公告标题可能格式不统一例如“关于对XX股份有限公司的问询函”、“关于XX股份600XXX的问询函”。我们需要设计一个健壮的正则表达式或解析逻辑来从中准确提取6位数字代码。2.2 爬虫工具选型与核心库Python生态中爬虫库众多针对本项目特点我的选型如下requestsBeautifulSoup4这是经典组合。requests用于发送HTTP请求简单易用BeautifulSoup4用于解析HTML文档其API对新手友好能快速定位和提取标签内的数据。对于上交所这种静态页面或服务端渲染页面这个组合完全够用。lxml作为BeautifulSoup的解析器之一lxml的解析速度非常快特别是在处理大量页面时优势明显。我们通常用BeautifulSoup(html, lxml)。pandas并非爬虫库但它是数据处理的瑞士军刀。我们将爬取到的结构化数据股票代码、公司名称、发函日期等存入DataFrame便于后续的清洗、筛选和与股价数据合并。rePython内置的正则表达式模块用于从复杂的字符串如公告标题中提取股票代码。为什么不直接用ScrapyScrapy是一个强大的异步爬虫框架适合构建大型、复杂的爬虫项目。本项目目标明确、页面结构相对稳定使用requestsBeautifulSoup更为轻量、直接调试和编写速度更快。注意在实际操作前务必仔细阅读目标网站的robots.txt文件通常位于网站根目录如https://www.sse.com.cn/robots.txt尊重网站的爬虫协议避免对服务器造成不必要的压力。建议设置合理的请求延迟如time.sleep(2)模拟人类浏览行为。3. 爬虫实战抓取问询函列表与关键信息接下来我们进入代码实战环节。我将分步骤拆解并解释每个环节的意图和注意事项。3.1 环境准备与请求模拟首先安装必要的库并配置请求头让我们看起来像一个正常的浏览器访问。import requests from bs4 import BeautifulSoup import pandas as pd import re import time from datetime import datetime # 配置请求头模拟浏览器 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, Accept: text/html,application/xhtmlxml,application/xml;q0.9,image/webp,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } # 上交所问询函列表页示例URL请注意实际URL需根据网站当前结构确定此处为示例 base_url http://www.sse.com.cn/disclosure/credibility/supervision/inquiries/ # 假设分页参数为 pageNo list_url_template base_url ?pageNo{} def fetch_page(page_num): 获取指定页码的HTML内容 url list_url_template.format(page_num) try: resp requests.get(url, headersheaders, timeout10) resp.raise_for_status() # 检查请求是否成功 # 检查编码中文网站常用gbk或utf-8 resp.encoding resp.apparent_encoding or utf-8 return resp.text except requests.exceptions.RequestException as e: print(f请求第{page_num}页失败: {e}) return None关键解释User-Agent是必须的没有它很多网站会拒绝请求或返回错误页面。timeout参数设置了超时时间避免因网络问题导致程序长时间挂起。resp.encoding的设置很重要编码错误会导致中文乱码。apparent_encoding是requests推测的编码有时需要手动指定为gbk。3.2 解析页面与数据提取获取到HTML后我们需要用BeautifulSoup解析并找到存放公告列表的容器通常是table或ul/li需要具体分析。def parse_list_page(html): 解析列表页提取公告链接、标题、日期等信息 soup BeautifulSoup(html, lxml) inquiry_list [] # 这里需要根据实际网页结构定位公告行 # 示例1假设公告在一个class为list的table的tbody的tr中 # table soup.find(table, class_list) # if table: # rows table.find(tbody).find_all(tr) # for row in rows: # cols row.find_all(td) # title_tag cols[0].find(a) # 假设第一列是带链接的标题 # date_text cols[2].text.strip() # 假设第三列是日期 # 示例2更常见的是用ul/li列表 # news_list soup.find(ul, class_news-list) # if news_list: # items news_list.find_all(li) # for item in items: # title_tag item.find(a) # date_span item.find(span, class_date) # **重要**以下为虚拟代码结构你需要用浏览器的开发者工具F12查看实际结构进行替换 # 假设我们找到了所有公告项 announcement_items soup.select(.sse_list_1 li) # 这是一个CSS选择器示例 for item in announcement_items: try: # 提取标题和链接 title_tag item.find(a) if not title_tag: continue title title_tag.text.strip() link title_tag.get(href) # 处理相对链接 if link and not link.startswith(http): link requests.compat.urljoin(base_url, link) # 提取发布日期 date_tag item.find(span, class_time) publish_date date_tag.text.strip() if date_tag else None # 从标题中提取股票代码核心步骤 stock_code extract_stock_code_from_title(title) # 如果成功提取到代码和日期则存入列表 if stock_code and publish_date: # 将日期字符串转为datetime对象便于后续处理 try: publish_date_dt datetime.strptime(publish_date, %Y-%m-%d) except ValueError: publish_date_dt None inquiry_list.append({ stock_code: stock_code, company_name: extract_company_name(title), # 需要另一个函数 inquiry_title: title, publish_date: publish_date_dt, publish_date_str: publish_date, detail_url: link }) except Exception as e: print(f解析公告项时出错: {e}, 原始内容: {item}) continue return inquiry_list def extract_stock_code_from_title(title): 从公告标题中提取6位股票代码 # 使用正则表达式匹配6位连续数字且通常以6沪市或0深市开头但上交所主要是6 # 模式1括号内包含代码如“600519” pattern1 r[\(](\d{6})[\)] # 模式2标题中直接包含6位数字 pattern2 r(?![\d])([0-9]{6})(?![\d]) match re.search(pattern1, title) if not match: match re.search(pattern2, title) if match: return match.group(1) return None def extract_company_name(title): 简易地从标题中提取公司简称实际应用可能需要更复杂的规则或词典 # 移除常见后缀和代码这是一个简化的示例 name re.sub(r[\(].*?[\)], , title) # 去掉括号及其中内容 name re.sub(r关于对|关于|的问询函|的监管问询函|股份有限公司|有限公司, , name) name name.strip() return name if name else title[:20] # 如果提取失败返回标题前20字实操心得与避坑指南结构定位是关键90%的爬虫问题出在定位标签不准。必须使用浏览器的“检查元素”功能仔细分析目标数据所在的HTML标签及其属性class,id等。soup.select()使用CSS选择器通常比find更灵活强大。健壮性处理代码中充满了try...except和if判断这是因为网页结构可能微调或个别条目格式特殊。必须假设任何一步都可能失败并做好异常处理和数据缺失的应对。正则表达式调试提取股票代码的正则表达式需要反复测试。可以在Python交互环境或在线正则工具中用大量不同的公告标题样本来测试你的模式确保覆盖“600519”、“(600519)”、“600519”、“600519”等多种情况。日期格式化将字符串日期转为datetime对象非常重要这为后续的日期比较、排序和查询提供了便利。注意源网站的日期格式strptime的格式字符串要与之匹配。3.3 实现分页爬取与数据存储单页数据有限我们需要爬取多页。def crawl_inquiries(max_pages10, start_page1): 爬取多页问询函列表 all_inquiries [] for page in range(start_page, start_page max_pages): print(f正在爬取第 {page} 页...) html fetch_page(page) if not html: print(f第 {page} 页获取失败可能已无更多内容或网络问题。) break page_data parse_list_page(html) if not page_data: print(f第 {page} 页未解析到数据可能结构已变化或已到末页。) break all_inquiries.extend(page_data) print(f第 {page} 页爬取到 {len(page_data)} 条记录。) # 礼貌性延迟避免请求过快 time.sleep(1.5) # 转换为DataFrame df_inquiries pd.DataFrame(all_inquiries) # 去重基于股票代码和发函日期 df_inquiries.drop_duplicates(subset[stock_code, publish_date_str], inplaceTrue) # 按日期排序 df_inquiries.sort_values(bypublish_date, inplaceTrue, ascendingFalse) df_inquiries.reset_index(dropTrue, inplaceTrue) # 保存到CSV文件 output_file fsse_inquiries_{datetime.now().strftime(%Y%m%d_%H%M%S)}.csv df_inquiries.to_csv(output_file, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel直接打开无乱码 print(f爬取完成共获取 {len(df_inquiries)} 条问询函记录。数据已保存至: {output_file}) return df_inquiries # 执行爬取 if __name__ __main__: df crawl_inquiries(max_pages5) # 先爬5页试试水注意事项max_pages可以设置一个较大的值通过解析失败或数据为空来自然终止循环更稳妥。time.sleep的延迟时间可以根据网站响应情况和自身需求调整通常在1-3秒之间。保存为CSV时使用utf-8-sig编码这是为了兼容Windows系统下的Excel否则用Excel打开可能出现中文乱码。4. 股价数据获取接口选择与日期对齐拿到问询函列表后下一步是获取对应股票在发函日的股价。这里有几个关键问题数据源选哪个如何保证日期对齐4.1 金融数据API选型对比个人开发者或研究者获取股价数据通常有以下几种途径数据源优点缺点适用场景AKShare免费、开源、接口丰富A股、宏观、新闻等、社区活跃。数据源稳定性依赖第三方网站接口可能变动速度可能较慢。个人学习、研究、非实时分析项目。Tushare数据较规范、稳定有基础免费额度社区成熟。高级数据需要积分或付费注册有一定门槛。对数据质量有一定要求的量化入门或研究。Baostock免费、提供较完整的沪深历史行情无需注册。接口相对简单数据更新频率和丰富度可能不及前两者。专注于历史行情数据下载。Yahoo Finance (yfinance)免费、全球市场、接口稳定易用。主要覆盖美股A股数据通过代码后缀如600519.SS获取但可能不全或不及时。同时需要A股和海外市场数据。付费数据商Wind, iFinD数据全面、准确、及时有API支持。费用昂贵个人用户难以承受。专业机构、对数据质量和实时性要求极高的场景。本项目选择AKShare理由很直接它是完全免费的且其“新闻舆情”或“股票数据”接口足以满足我们获取历史日线行情开盘价、收盘价、最高价、最低价、成交量等的需求。对于事件研究日线数据已经足够。4.2 使用AKShare获取历史行情数据首先安装AKSharepip install akshare --upgradeimport akshare as ak def get_stock_daily_price(stock_code, start_date, end_date): 获取指定股票在指定日期区间内的日线行情数据 参数: stock_code: 股票代码如 600519 start_date: 开始日期格式 YYYYMMDD end_date: 结束日期格式 YYYYMMDD 返回: pandas.DataFrame # AKShare的A股日线数据接口 # 注意ak.stock_zh_a_hist 要求代码带市场前缀上交所是sh code_with_prefix fsh{stock_code} if stock_code.startswith(6) else fsz{stock_code} try: # 调整日期格式 start_date_str start_date.replace(-, ) end_date_str end_date.replace(-, ) df_price ak.stock_zh_a_hist(symbolcode_with_prefix, perioddaily, start_datestart_date_str, end_dateend_date_str, adjustqfq) # qfq: 前复权便于历史价格比较 if df_price.empty: print(f警告未获取到股票 {stock_code} 在 {start_date} 至 {end_date} 间的数据。) return None # 规范列名 df_price.rename(columns{ 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 振幅: amplitude, 涨跌幅: pct_chg, 涨跌额: change, 换手率: turnover }, inplaceTrue) # 将日期列转为datetime类型 df_price[trade_date] pd.to_datetime(df_price[trade_date]) return df_price except Exception as e: print(f获取股票 {stock_code} 价格数据时出错: {e}) return None # 示例获取贵州茅台2023年一段时间的股价 # df_600519 get_stock_daily_price(600519, 20230101, 20231231)关键点解析市场前缀AKShare的接口需要区分沪市(sh)和深市(sz)我们通过股票代码首位数字6开头为沪市0或3开头为深市自动添加。复权方式adjustqfq表示前复权。复权处理对于长期股价分析至关重要它消除了分红、送股等事件对股价的跳跃性影响使得历史价格序列连续可比。对于事件研究强烈建议使用复权价格。错误处理网络问题、股票代码错误、日期区间无数据等情况都可能发生必须用try...except包裹并返回None或空DataFrame避免程序崩溃。4.3 核心挑战交易日与发函日的对齐这是本项目最易出错也最关键的一环。问询函的“发函日期”publish_date是日历日期但股票交易只在交易日进行。如果发函日正好是周末或节假日当天是没有股价数据的。解决方案我们需要找到发函日之后或之前的第一个交易日的股价来近似代表“市场得知该消息后的首次反应”。通常我们选择发函日之后的第一个交易日T1。def align_trade_date(publish_date, price_df, directionforward): 将公告日期对齐到最近的交易日。 参数: publish_date: datetime对象公告日期。 price_df: 包含trade_date列的股价DataFrame。 direction: forward (向后找默认T1), backward (向前找T-1), nearest (最近)。 返回: 对齐后的交易日日期(datetime)若找不到则返回None。 if price_df is None or price_df.empty: return None # 确保price_df的trade_date是datetime且已排序 price_dates pd.to_datetime(price_df[trade_date]).sort_values().reset_index(dropTrue) if direction forward: # 找到第一个大于等于公告日的交易日如果公告日是交易日就是它自己否则是之后第一个 mask price_dates publish_date if mask.any(): return price_dates[mask].iloc[0] elif direction backward: # 找到最后一个小于等于公告日的交易日 mask price_dates publish_date if mask.any(): return price_dates[mask].iloc[-1] elif direction nearest: # 找到绝对值差最小的交易日 idx (price_dates - publish_date).abs().argmin() return price_dates.iloc[idx] else: raise ValueError(direction参数必须是 forward, backward 或 nearest) return None # 如果没有找到符合条件的日期应用逻辑对于每一条问询函记录调用get_stock_daily_price获取发函日前后一段时间例如前后各5个交易日的股价数据。使用align_trade_date(publish_date, price_df, forward)获取“发函日后第一个交易日”的日期。从price_df中筛选出该对齐日期的股价记录。注意这里存在一个数据获取边界问题。如果问询函是今天刚发布的那么“T1”的股价数据可能尚未生成收盘后才会更新。在实际脚本中需要加入日期判断如果对齐后的交易日大于当前日期则视为数据不可得。5. 数据整合与结果输出现在我们将爬取到的问询函列表和获取的股价数据整合起来。5.1 主流程函数设计与实现def main_workflow(inquiry_df, days_before2, days_after5): 主工作流为问询函列表中的每条记录获取发函日附近股价。 参数: inquiry_df: 爬取到的问询函DataFrame。 days_before: 需要获取发函日前多少天的股价用于对比。 days_after: 需要获取发函日后多少天的股价用于观察短期影响。 results [] for idx, row in inquiry_df.iterrows(): stock_code row[stock_code] publish_date row[publish_date] # 假设已是datetime对象 publish_str row[publish_date_str] print(f处理 {stock_code} ({row.get(company_name, N/A)})发函日: {publish_str}) # 1. 计算需要获取股价的日期范围 # 获取一个足够宽的日期范围确保包含对齐日及前后观察窗口 start_date (publish_date - pd.Timedelta(daysdays_before10)).strftime(%Y%m%d) end_date (publish_date pd.Timedelta(daysdays_after10)).strftime(%Y%m%d) # 2. 获取股价数据 price_df get_stock_daily_price(stock_code, start_date, end_date) if price_df is None or price_df.empty: print(f - 警告未能获取 {stock_code} 的股价数据跳过。) continue # 3. 日期对齐 aligned_trade_date align_trade_date(publish_date, price_df, directionforward) if aligned_trade_date is None: print(f - 警告无法为发函日 {publish_str} 对齐到交易日跳过。) continue # 4. 提取对齐日的股价 aligned_price_row price_df[price_df[trade_date] aligned_trade_date] if aligned_price_row.empty: # 理论上不会发生因为对齐日期来自price_df自身 print(f - 警告在对齐日 {aligned_trade_date.date()} 未找到股价记录跳过。) continue price_info aligned_price_row.iloc[0] # 5. 可选提取前后几日的股价用于计算涨跌幅等 # 例如计算对齐日相对于前一个交易日的涨跌幅 prev_day_df price_df[price_df[trade_date] aligned_trade_date] if not prev_day_df.empty: prev_close prev_day_df.iloc[-1][close] price_info[pct_chg_vs_prev] (price_info[close] - prev_close) / prev_close * 100 else: price_info[pct_chg_vs_prev] None # 6. 合并结果 result_dict row.to_dict() result_dict[aligned_trade_date] aligned_trade_date # 将股价信息合并进来 for col in [open, close, high, low, volume, pct_chg, pct_chg_vs_prev]: result_dict[fprice_{col}] price_info.get(col) results.append(result_dict) # 礼貌延迟避免对数据源请求过快 time.sleep(0.5) # 转换为最终DataFrame final_df pd.DataFrame(results) # 重新排列列使其更易读 cols_order [stock_code, company_name, publish_date_str, aligned_trade_date, price_close, price_pct_chg, price_pct_chg_vs_prev, price_open, price_high, price_low, price_volume, inquiry_title, detail_url] # 只保留final_df中存在的列 final_cols [c for c in cols_order if c in final_df.columns] final_df final_df[final_cols [c for c in final_df.columns if c not in final_cols]] # 保存最终结果 output_file_final fsse_inquiry_with_price_{datetime.now().strftime(%Y%m%d_%H%M%S)}.csv final_df.to_csv(output_file_final, indexFalse, encodingutf-8-sig) print(f\n数据整合完成共处理 {len(final_df)} 条有效记录。结果已保存至: {output_file_final}) return final_df # 执行整合流程 if __name__ __main__: # 假设我们已经有了爬取好的问询函DataFrame df_inquiries # 如果还没有可以先运行 crawl_inquiries 函数 # df_inquiries crawl_inquiries(max_pages3) # 然后运行主流程 final_result main_workflow(df_inquiries, days_before1, days_after3) print(final_result.head()) # 预览前几条结果5.2 结果解读与初步分析运行成功后你会得到一个CSV文件包含类似以下字段stock_codecompany_namepublish_date_straligned_trade_dateprice_closeprice_pct_chgprice_pct_chg_vs_prev...600519贵州茅台2023-08-152023-08-161850.501.20.8...000858五粮液2023-08-102023-08-11165.30-0.5-1.2...aligned_trade_date就是我们计算出的“市场首次获知消息后的反应日”。price_pct_chg是该股票在aligned_trade_date这一天的日涨跌幅来自数据源。price_pct_chg_vs_prev是我们自己计算的相对于前一个交易日的涨跌幅可以作为另一个参考。有了这个数据集你就可以进行一些简单的分析了例如统计发函后首个交易日股价上涨和下跌的公司比例。计算平均涨跌幅。结合问询函类型如果需要从详情页提取进行分组分析。将结果与同期大盘指数涨跌幅进行对比看是否存在超额收益或亏损。6. 项目优化、扩展与注意事项一个基础版本完成后我们可以从多个角度思考如何让它更健壮、更有用。6.1 错误处理与日志记录强化目前的脚本打印信息到控制台对于长期运行或批量处理更好的方式是使用Python的logging模块。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(inquiry_stock_crawler.log), logging.StreamHandler() ]) logger logging.getLogger(__name__) # 在代码中将 print 替换为 logger.info/warning/error logger.info(f正在爬取第 {page} 页...) logger.warning(f第 {page} 页未解析到数据可能结构已变化。)同时将可能失败的操作如网络请求、数据解析放入更细粒度的try...except中并记录详细的错误信息便于事后排查。6.2 应对网站结构变更网站改版是爬虫的天敌。我们可以采取以下策略将解析规则CSS选择器、正则表达式提取到配置文件中如JSON或YAML而不是硬编码在代码里。这样当网站结构变化时只需修改配置文件无需改动核心代码逻辑。编写监控脚本定期运行爬虫检查返回的数据量是否骤减或为空并设置邮件或消息通知。使用更健壮的定位方式如果网站结构不稳定可以尝试结合多个特征来定位元素例如同时使用class和tag或者通过文本内容进行模糊匹配。6.3 扩展方向这个项目可以作为一个起点向多个方向扩展深入详情页爬取问询函的详细内容利用文本分析如jieba分词、snownlp情感分析对问询函的严厉程度、涉及问题类型进行量化再与股价反应做相关性分析。多数据源关联除了股价还可以关联该股票在发函日前后的融资融券数据、龙虎榜数据、券商研报数量等构建更复杂的分析模型。自动化与定时任务使用schedule库或操作系统的定时任务如cron让脚本每天自动运行爬取最新的问询函并更新数据库实现数据流的自动化。构建简单可视化使用matplotlib或plotly为每只股票绘制发函日前后一段时间的股价走势图并标注发函日直观展示影响。数据库存储当数据量变大时将结果存入SQLite或MySQL数据库便于复杂的查询和分析。6.4 法律与合规提醒最后也是最重要的必须强调合规使用爬虫和数据遵守robots.txt这是网络爬虫的基本礼仪。控制访问频率避免对目标网站服务器造成显著负载这既是道德要求也能减少你被屏蔽的风险。数据用途本项目获取的数据均为公开信息用于个人学习、研究或分析。切勿用于商业用途、频繁交易或任何可能影响市场的行为。金融数据的获取和使用可能受到相关法律法规的约束请确保你的用途合法合规。尊重版权问询函文本的版权归属于上海证券交易所在进行文本分析或传播时需注意相关条款。通过这个项目你不仅学会了如何用Python抓取特定结构的网页信息更重要的是掌握了将不同来源、不同结构的数据通过关键字段进行关联整合的完整数据处理流程。这套方法论可以迁移到无数类似的应用场景中这才是其核心价值所在。在实际操作中最耗费时间的往往不是写代码而是分析网页结构、处理数据异常和调试边界情况耐心和细致是成功的关键。