1. 项目概述从监管问询中挖掘市场信号在A股市场交易所的问询函一直是投资者尤其是像我这样喜欢从公开信息里“淘金”的人重点关注的对象。一份问询函特别是针对年报、重大资产重组或者异常波动的问询往往像一面“照妖镜”能照出上市公司财报里那些语焉不详的细节或是经营中潜在的“雷点”。但信息本身是孤立的它的价值需要放在市场反应的背景下才能被充分解读。这个项目的核心想法很简单自动化地抓取上海证券交易所上交所发布的监管问询函并精准定位到相关上市公司然后获取这些公司在收到问询函当日的股价数据。听起来像是把两个独立的数据源监管公告和行情数据串联起来但实操中你会发现从数据获取、清洗、匹配到分析每一步都藏着不少门道。这不仅仅是写个爬虫那么简单它涉及到对监管信息披露规则的理解、对网络反爬策略的应对、对金融数据API的熟练运用以及最终如何将这两类数据“对齐”的逻辑设计。我之所以花时间折腾这个是因为手动操作效率太低。每天盯着公告再切到行情软件查历史K线费时费力还容易出错。通过Python将这个过程自动化我就能快速构建一个历史数据库用于回溯研究比如市场对不同类型的问询函年报问询、重组问询等反应如何是“利空出尽”还是“跌跌不休”问询函发布的时机盘前、盘中、盘后对股价冲击有影响吗这些问题的答案都藏在数据里。无论你是对量化研究感兴趣的开发者还是希望用数据辅助决策的投资者亦或是学习Python网络爬虫和数据分析的学生这个项目都能提供一个非常贴近实战的练手场景。接下来我会详细拆解从思路到实现的每一个环节并分享那些在官方文档里找不到的“踩坑”经验和操作技巧。2. 核心思路与架构设计做这个项目不能一上来就埋头写代码。首先得想清楚数据从哪来、怎么拿、拿了以后怎么拼在一起。这就像做菜得先备齐食材和厨具。2.1 数据源分析与选型问询函数据源上交所官方网站这是最权威、最直接的数据来源。上交所的“监管信息公开”栏目会发布各类监管函件。我们的目标是爬取其中的“问询函”。这里的关键不是简单的requests.get而是要理解它的页面结构。通常这类列表页是动态加载的或者有翻页机制。我们需要分析其网络请求XHR找到返回真实数据的API接口。一个常见的发现是数据往往通过一个POST或GET请求以JSON格式返回里面包含了公告标题、代码、日期、PDF链接等。注意直接爬取HTML页面然后解析对于这类现代网站往往不是最优解。优先寻找隐藏在背后的JSON API接口效率更高且数据结构更清晰。股价数据源第三方金融数据API获取精确的历史日级股价数据开盘、收盘、最高、最低、成交量我们不可能自己去解析行情软件。这里有几个可靠的备选AKShare一个基于Python的免费金融数据接口库数据源丰富对国内股票数据支持很好。它封装了多个数据源稳定性和易用性都不错非常适合个人研究和学习。Tushare或Baostock同样是国内常用的免费金融数据平台需要注册获取Token。它们的数据质量较高但有调用频率或积分限制。商用API如聚宽、万得数据最全最准但需要付费。对于本项目我推荐使用AKShare。理由很简单免费、无需注册、接口直接、社区活跃。它足以满足我们获取某一天股票行情数据的需求。2.2 系统流程设计整个自动化流程可以分解为以下几个核心步骤我画了一个简单的逻辑图在脑子里爬取问询函列表模拟请求上交所的公告列表接口获取一段时间内例如最近30天所有问询函的列表数据。解析并过滤数据从列表中提取每份问询函的关键信息公告标题、股票代码需从标题或内容中提取、公告日期、问询函PDF的下载链接。数据清洗与存储清洗股票代码统一为6位数字格式将结构化数据标题、代码、日期、链接存储起来比如用CSV文件或SQLite数据库。PDF文件可以选择性下载。股价数据获取遍历存储的问询函记录根据股票代码和公告日期调用AKShare接口获取该股票在公告日当天的日线行情数据。数据关联与输出将问询函信息与对应的股价数据合并生成一份包含“何时、何股、因何事被问询、当日股价表现如何”的综合数据表。进阶定时任务与可视化可以使用schedule库设置定时任务每日自动运行更新。用pandas和matplotlib对结果进行初步分析可视化比如统计股价涨跌分布。这个流程的难点和精髓在于第二步提取股票代码和第四步日期对齐。问询函标题可能包含多个公司代码如涉及并购公告发布日期与实际市场交易日也可能存在错位如函件在周末发布这些都需要在代码逻辑中进行精细处理。3. 关键技术实现与代码拆解接下来我们进入实操环节。我会分模块展示核心代码并解释每一处为什么要这么写。3.1 环境准备与依赖安装首先创建一个新的Python虚拟环境是个好习惯。然后安装必要的库pip install requests pandas aksharerequests用于发送HTTP请求爬取网页数据。pandas数据处理和分析的核心库用于清洗、整合和保存数据。akshare免费获取股价数据。3.2 爬取上交所问询函列表这一步的目标是拿到结构化的问询函列表。如上所述我们需要找到上交所披露问询函的真实数据接口。通过浏览器开发者工具F12的“网络(Network)”标签筛选XHR请求在公告列表页进行翻页操作通常能找到一个返回JSON数据的请求。假设我们找到了一个这样的接口请注意以下URL和参数为示例实际需根据当时网站结构分析import requests import pandas as pd from datetime import datetime, timedelta def fetch_inquiry_letters(start_date, end_date): 抓取指定时间段内的上交所问询函列表 # 示例API端点需根据实际情况替换 url http://www.sse.com.cn/disclosure/credibility/supervision/inquiries/ # 常见的请求参数通常包括页码、时间范围、类型等 params { startDate: start_date, # 格式2023-01-01 endDate: end_date, pageNum: 1, pageSize: 50, type: 问询函 # 这个参数名需要根据实际接口确定 } 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://www.sse.com.cn/disclosure/credibility/supervision/inquiries/, Accept: application/json, text/javascript, */*; q0.01 } all_letters [] page 1 while True: params[pageNum] page try: resp requests.get(url, paramsparams, headersheaders, timeout10) resp.raise_for_status() # 检查请求是否成功 data resp.json() # 解析返回的JSON数据这个结构需要你根据实际接口响应来调整 letter_list data.get(data, []) if not letter_list: print(f第{page}页无数据爬取结束。) break for item in letter_list: # 提取关键信息 letter_info { title: item.get(title), code: item.get(code), # 有时接口直接提供代码 publish_date: item.get(publishDate), pdf_url: item.get(pdfUrl) } all_letters.append(letter_info) print(f已爬取第{page}页共{len(letter_list)}条记录。) page 1 # 简单延迟避免请求过快 time.sleep(1) except requests.exceptions.RequestException as e: print(f请求第{page}页时发生错误: {e}) break except ValueError as e: print(f解析JSON响应失败: {e}) break return pd.DataFrame(all_letters) # 使用示例爬取最近30天的问询函 end_date datetime.now().strftime(%Y-%m-%d) start_date (datetime.now() - timedelta(days30)).strftime(%Y-%m-%d) df_letters fetch_inquiry_letters(start_date, end_date) print(df_letters.head())实操心得网站的接口和参数可能会变动。如果上述方法找不到可能需要尝试分析页面初始加载时的一个包含大量数据的JSON块或者使用Selenium模拟浏览器获取渲染后的页面数据。Selenium是最后的手段因为它速度慢、资源消耗大。3.3 从标题中提取股票代码如果接口返回的数据中没有直接的code字段或者为了双重校验我们需要从公告标题中提取股票代码。A股股票代码是6位数字。import re def extract_stock_code(title): 从问询函标题中提取股票代码。 常见模式如“关于对XXXXXX(股票代码)的定期报告问询函” if not isinstance(title, str): return None # 匹配6位连续数字 patterns [ r[\(](\d{6})[\)], # 括号内的6位数字 r关于对(\d{6})的, # “关于对”后面的6位数字 r股票代码[:]\s*(\d{6}), # 明确写明股票代码 r\b(\d{6})\b # 独立的6位数字需谨慎可能误匹配日期 ] for pattern in patterns: match re.search(pattern, title) if match: code match.group(1) # 简单校验A股代码通常以6沪市、3创业板、0深市主板开头 if code.startswith((0, 3, 6)): return code return None # 应用函数 if code not in df_letters.columns or df_letters[code].isnull().any(): df_letters[extracted_code] df_letters[title].apply(extract_stock_code) # 优先使用接口返回的代码缺失则用提取的 df_letters[stock_code] df_letters[code].combine_first(df_letters[extracted_code]) else: df_letters[stock_code] df_letters[code] # 清洗删除无法识别代码的记录 df_letters_clean df_letters.dropna(subset[stock_code]).copy() df_letters_clean[stock_code] df_letters_clean[stock_code].astype(str).str.zfill(6) # 补全为6位 print(f清洗后剩余{len(df_letters_clean)}条有效问询函记录。)3.4 使用AKShare获取发函日股价这是项目的另一半核心。我们有了股票代码和日期现在需要获取那一天的行情。import akshare as ak import time def get_stock_price_on_date(stock_code, target_date): 获取指定股票在指定日期的日线行情数据。 参数 stock_code: 6位股票代码如 600000 target_date: 字符串格式 2023-08-01 返回 包含当日行情数据的Series如果当天无交易或数据缺失则返回None # AKShare的接口需要市场前缀上交所股票是sh深交所是sz # 但ak.stock_zh_a_hist接口可以直接用6位代码 try: # 获取该股票一段时间的历史数据确保包含目标日期 # 这里获取目标日期前后共10天的数据以防日期偏差 start_date (pd.to_datetime(target_date) - timedelta(days5)).strftime(%Y%m%d) end_date (pd.to_datetime(target_date) timedelta(days5)).strftime(%Y%m%d) # 调用AKShare接口 df_stock ak.stock_zh_a_hist(symbolstock_code, perioddaily, start_datestart_date, end_dateend_date, adjustqfq) if df_stock.empty: print(f警告未获取到股票 {stock_code} 在{start_date}到{end_date}间的数据。) return None # 将日期列转换为datetime格式以便匹配 df_stock[日期] pd.to_datetime(df_stock[日期]) target_datetime pd.to_datetime(target_date) # 查找目标日期的数据 day_data df_stock[df_stock[日期] target_datetime] if not day_data.empty: # 通常只有一行 return day_data.iloc[0] else: # 目标日期可能为非交易日周末、节假日尝试寻找前一个交易日 # 获取所有交易日日期并排序 trading_dates df_stock[日期].sort_values() # 找到目标日期之前最近的交易日 previous_trading_days trading_dates[trading_dates target_datetime] if not previous_trading_days.empty: last_trading_date previous_trading_days.iloc[-1] day_data df_stock[df_stock[日期] last_trading_date] print(f提示{target_date}为非交易日或数据缺失股票{stock_code}使用最近交易日{last_trading_date.date()}的数据。) return day_data.iloc[0] else: print(f警告股票 {stock_code} 在{target_date}及之前无交易数据。) return None except Exception as e: print(f获取股票 {stock_code} 数据时发生异常: {e}) return None # 遍历问询函DataFrame获取股价 price_data_list [] for idx, row in df_letters_clean.iterrows(): stock_code row[stock_code] publish_date row[publish_date] # 假设已经是YYYY-MM-DD格式 price_info get_stock_price_on_date(stock_code, publish_date) if price_info is not None: price_dict { stock_code: stock_code, inquiry_date: publish_date, open: price_info[开盘], close: price_info[收盘], high: price_info[最高], low: price_info[最低], volume: price_info[成交量], pct_change: (price_info[收盘] - price_info[开盘]) / price_info[开盘] * 100 if price_info[开盘] ! 0 else 0 } price_data_list.append(price_dict) else: # 记录获取失败的情况 price_data_list.append({ stock_code: stock_code, inquiry_date: publish_date, open: None, close: None, high: None, low: None, volume: None, pct_change: None }) # 礼貌性延迟避免对数据源造成压力 time.sleep(0.5) df_prices pd.DataFrame(price_data_list)3.5 数据合并与最终输出最后将问询函信息与股价信息通过股票代码和日期进行合并。# 合并两个DataFrame # 使用股票代码和日期作为合并键但要注意股价数据里我们用的是‘inquiry_date’而问询函里是‘publish_date’ df_letters_clean[publish_date] pd.to_datetime(df_letters_clean[publish_date]) df_prices[inquiry_date] pd.to_datetime(df_prices[inquiry_date]) # 假设我们根据股票代码和日期允许股价使用最近交易日进行关联 df_final pd.merge( df_letters_clean, df_prices, howleft, left_on[stock_code, publish_date], right_on[stock_code, inquiry_date] ) # 选择并重命名我们关心的列 output_columns [ publish_date, stock_code, title, open, close, high, low, volume, pct_change, pdf_url ] df_output df_final[output_columns] # 保存到CSV文件 output_filename fsse_inquiry_with_price_{start_date}_to_{end_date}.csv df_output.to_csv(output_filename, indexFalse, encodingutf-8-sig) # utf-8-sig支持Excel中文 print(f数据已保存至: {output_filename}) print(df_output.head())至此一个完整的、从数据抓取到关联分析的原型就完成了。你得到的是一个CSV文件里面清晰地列出了每一份问询函及其对应的股票在发函日的市场表现。4. 常见问题、优化策略与避坑指南在实际操作中你肯定会遇到比我这里列举的更多问题。下面是我在多次运行类似脚本后总结的一些核心要点和进阶优化方向。4.1 数据获取中的典型问题1. 网站反爬与IP封锁这是爬虫项目的老大难问题。上交所官网对频繁访问会有防御。策略降低频率在请求间添加随机延时例如time.sleep(random.uniform(1, 3))。轮换User-Agent准备一个列表每次请求随机选择。使用代理IP池如果数据量巨大这是必须的。可以考虑使用一些免费的代理IP服务但稳定性和速度需要测试。对于个人小规模使用控制好频率通常足够。尊重robots.txt检查目标网站的robots.txt文件避免爬取禁止的目录。2. 接口变更与页面结构变动今天能用的接口明天可能就变了。策略模块化设计将数据抓取函数fetch_inquiry_letters单独封装。一旦接口失效只需修改这个函数不影响后续数据处理逻辑。添加异常监控在脚本中增加日志记录当解析失败或返回数据为空时能及时发出警报如发送邮件到你自己。定期运行测试将脚本设置为每日低峰期运行一次而不是一次性爬取大量历史数据可以减少因接口变动导致的大规模数据缺失。3. 股票代码提取错误正则表达式不可能覆盖所有标题格式。策略人工抽样校验首次运行后一定要人工检查提取出的代码和标题是否匹配。可以随机抽样几十条记录进行核对。多模式组合与优先级如上面代码所示使用多个正则模式并按可靠性设置优先级。括号内的代码通常最准确。建立纠错词典对于经常出错的公司如标题中用简称可以建立一个小型的“公司简称-股票代码”映射表进行后期校正。4. AKShare数据缺失或延迟免费数据源的数据可能存在延迟或者某些股票如刚上市、退市的历史数据不全。策略日期容错处理正如我们在get_stock_price_on_date函数里做的当目标日期无数据时自动寻找前一个交易日。这符合金融市场分析的一般逻辑。设置重试机制网络请求可能失败对AKShare的调用可以包装在重试逻辑中如使用tenacity库。备用数据源可以考虑集成Tushare作为备用当AKShare失败时尝试切换。但要注意不同API的返回格式需要适配。4.2 项目进阶优化方向当基础功能跑通后你可以考虑以下方向让这个项目变得更强大1. 数据持久化与增量更新不要每次都从头爬取所有历史数据。使用数据库将df_letters_clean和df_prices存入SQLite或MySQL数据库。每次运行只爬取最新日期的问询函然后与数据库存量合并实现增量更新。状态记录在数据库中记录每条问询函的“股价是否已获取”状态避免重复查询。2. 丰富数据维度下载并解析PDF使用pdfplumber或PyPDF2库下载问询函PDF并尝试提取关键文本如问询问题数量、涉及会计科目等进行文本分析。获取更多市场数据不仅获取发函日股价还可以获取前后N天的股价用于计算事件窗口的累计异常收益率、当日的市场指数如上证指数用于计算相对涨跌、甚至当日的换手率、资金流向等。3. 构建分析模型这是从“数据收集”到“数据应用”的飞跃。事件研究法计算每只股票在问询函发布前后一段时间内的“异常收益率”量化问询函事件对股价的纯粹影响。分类统计根据问询函类型年报问询、重组问询、监管工作函对股价反应进行分类统计看看市场对不同类型监管信号的敏感度。构建预警指标结合问询函内容通过文本分析得到的关键词和股价反应尝试建立简单的预警模型用于关注潜在风险较高的公司。4. 自动化与可视化定时任务使用schedule库或操作系统的cron/Task Scheduler让脚本在每天收盘后自动运行更新数据。生成报告使用Jinja2模板引擎将每日新增的问询函及股价表现自动生成HTML或PDF简报。简单看板用Flask或Streamlit快速搭建一个内部网页展示问询函统计图表、近期高关注度公司列表等。4.3 我的几点核心心得合法性是底线本项目所有数据均来自公开信息用于个人研究学习。务必控制爬取频率避免对目标网站服务器造成压力。绝对不要试图爬取非公开、需要登录或有明确技术防护如高强度验证码的数据。错误处理要细致网络爬虫和数据获取脚本里try...except块是你的好朋友。对每一个网络请求、数据解析步骤都要做好异常捕获和日志记录否则脚本会在半夜悄无声息地崩溃。数据清洗比获取更重要原始数据往往是脏的、乱的。花在数据清洗处理缺失值、格式统一、去重、逻辑校验上的时间通常会超过数据获取本身。这部分工作直接决定了后续分析的可靠性。从最小可行产品开始不要一开始就想做一个全自动、带数据库、有前端看板的复杂系统。先像我上面那样写一个能跑通的脚本输出一个CSV文件。验证整个逻辑闭环是可行的然后再一步步迭代增加新功能。这个项目就像一把瑞士军刀它为你打开了一扇门让你能够用一种可量化、可回溯的方式去观察和理解“监管信号”这个抽象概念在真实市场中的涟漪。当你手里有了这样一份数据你会发现很多之前模糊的感觉现在可以变成清晰的图表和数字。这才是数据工具最大的魅力所在。