证监会上市公司公告采集：OpenClaw 合规抓取公开公告，自动提取关键财务与业务变动信息
摘要：上市公司公告是投资者、监管机构和研究机构获取企业实时经营动态的核心数据源。本文从证监会指定信息披露平台的公开数据出发，系统阐述如何利用 OpenClaw 框架构建一套合规、高效、可扩展的公告采集与信息提取流水线。文章涵盖公开数据源分析、合规抓取策略设计、HTML 到结构化文本的清洗转换、基于规则与模型融合的财务指标提取、业务变动事件识别，以及完整的工程化落地实践，目标构建一套日均处理数千份公告、关键信息召回率超过 95% 的自动化采集分析系统。
一、引言：公告数据的价值与采集挑战
中国资本市场拥有超过 5000 家上市公司，每日通过证监会指定信息披露平台发布的公告数量在数百份到上千份之间波动。这些公告承载着企业最核心的经营信息——年度报告中的财务数据、临时公告中的重大合同、董事会决议中的战略调整、股东减持计划中的资本动向，每一项都可能对二级市场产生直接冲击。
对于量化投资机构而言，公告数据的时效性直接决定了策略的边际收益。在业绩预告发布后的几分钟内，市场定价尚未完全反映新信息，这一窗口期正是超额收益的来源。对于券商研究所而言，自动化的公告摘要生成能将分析师从繁琐的阅读工作中解放出来，聚焦于深度研判。对于企业风险管理部门而言，上下游合作伙伴的公告异动往往是供应链风险的前置信号。
然而，从海量公告中高效提取结构化信息并非易事。首先，公告文件格式多样，PDF 文件占绝大多数，部分历史公告甚至以扫描件形式存在；其次，公告内容高度非结构化，即使是同一类型的年度报告，不同公司的排版格式、表格结构、术语使用都存在显著差异；再次，数据采集必须严格遵循相关法律法规，不能对信息披露平台造成不必要的访问压力，也不能越过授权边界抓取非公开数据。
本文提出的解决方案以 OpenClaw 框架为核心，构建一套从公告发现、合规下载、格式解析到信息提取的完整流水线。OpenClaw 是一个面向数据采集场景的轻量级框架，其设计理念强调合规性、可观测性和可扩展性，特别适合处理上市公司公告这类严肃的数据采集任务。我们将在后续章节中逐步展开技术细节，涵盖数据源分析、抓取策略、解析引擎、提取模型和工程化部署等全链路内容。
二、上市公司公告体系与公开数据源全景
理解公告数据的源头是构建采集系统的第一步。中国证监会指定的上市公司信息披露平台构成了公告数据的官方发布渠道，所有上市公司必须依法在这些平台上披露定期报告和临时公告。
2.1 法定信息披露平台梳理
巨潮资讯网是证监会指定的上市公司信息披露网站之一，由深圳证券交易所下属的深圳证券信息有限公司运营。该网站汇集了沪深两市全部上市公司的公告文件，提供按股票代码、公告类型、时间段等多维度的检索功能。公告文件通常以 PDF 格式提供下载，部分公告同时提供 HTML 格式的在线阅读版本。
上海证券交易所官网和深圳证券交易所官网各自维护着本所上市公司的公告专区。上交所的公告披露页面提供了按证券代码、公告类别、关键词等条件的筛选功能，公告列表支持分页浏览，单条公告详情页包含公告标题、发布时间、证券简称、PDF 下载链接等关键字段。深交所的公告披露系统在功能上类似，但在页面结构和接口设计上存在差异，采集系统需要针对不同交易所进行适配。
此外，全国中小企业股份转让系统负责新三板挂牌公司的信息披露，其公告格式与沪深交易所有所不同，但同样遵循公开披露原则。香港交易所披露易平台则服务于港股上市公司，对于同时发行 A 股和 H 股的公司，跨境公告的比对分析也是一个有价值的研究方向。
2.2 公告分类体系
上市公司的公告类型丰富多样，按照监管要求可以分为定期报告和临时公告两大类。定期报告包括年度报告、半年度报告和季度报告，其中年度报告的信息量最大，包含经审计的财务报表、董事会报告、监事会报告、审计报告等内容。临时公告涵盖了公司经营过程中需要及时披露的重大事项，如重大资产重组、对外投资、关联交易、股权变动、诉讼仲裁、业绩预告、利润分配等。
从数据采集的角度来看，不同类型的公告需要采用不同的解析策略。定期报告篇幅较长，通常超过 100 页，其中财务数据集中在特定章节，可以通过目录定位、关键词匹配等方式快速跳转到目标区域。临时公告相对简短，但类型繁多，需要建立公告类型识别模型来区分不同公告的提取模板。
2.3 公告元数据结构
每一份公告在信息披露平台上都对应着一组元数据，包括公告标题、上市公司名称、证券代码、公告日期、公告类型、文件格式等。这些元数据是公告采集系统的基础信息，也是后续公告筛选、去重、分类的依据。公告标题通常遵循一定的命名规范，例如「XX 股份有限公司 2024 年年度报告」「XX 科技股份有限公司关于收到政府补助的公告」等，从标题中就可以初步判断公告的类型和重要程度。
公告日期是时效性管理的关键字段。采集系统需要维护一个时间窗口，确保不遗漏新发布的公告，同时避免重复采集已处理的历史公告。公告文件格式以 PDF 为主，少数公告会同时提供 Word 或 HTML 版本。PDF 文件的解析是公告处理流程中的技术难点之一，我们将在后续章节中详细讨论。
三、OpenClaw 框架：合规采集的设计哲学
OpenClaw 是一个为数据采集场景设计的 Python 框架，其核心理念是将「合规」作为一等公民而非事后补救项。在上市公司公告采集这一场景中，OpenClaw 的设计哲学体现在以下几个方面：请求频率控制、访问来源声明、数据使用边界界定和采集行为审计。
3.1 框架核心架构
OpenClaw 采用了经典的生产者-消费者架构，将采集任务分解为 URL 发现、内容下载、数据解析和结果存储四个阶段。每个阶段由独立的组件负责，组件之间通过消息队列进行解耦，使得系统具备良好的水平扩展能力。URL 发现组件负责从信息披露平台的列表页提取公告详情页链接，内容下载组件负责将公告文件下载到本地，数据解析组件负责将 PDF 或 HTML 文件转换为结构化文本，结果存储组件负责将提取的信息写入数据库。
框架内置了请求调度器，支持基于令牌桶算法的请求频率控制。使用者可以为每个目标网站配置独立的请求间隔和并发数上限，确保采集行为不会对目标服务器造成压力。调度器还支持基于域名的全局速率限制，防止多个采集任务同时访问同一域名时突破频率上限。
3.2 合规性保障机制
OpenClaw 在 HTTP 请求层面强制要求设置 User-Agent 头，明确标识采集方的身份和用途。框架默认的 User-Agent 模板包含项目名称、版本号和联系方式，便于目标网站的管理员在必要时与采集方取得联系。同时，框架会在每次请求时检查目标网站的 robots.txt 文件，自动遵守其中声明的爬取规则。
在数据存储层面，OpenClaw 要求使用者明确声明数据的使用范围和保留期限。对于上市公司公告这类公开数据，框架建议设置合理的数据更新周期，定期清理过期数据，避免无限制地囤积信息。框架还提供了数据溯源功能，每一份采集到的公告都会记录其来源 URL、采集时间、文件哈希值等元数据，确保数据来源可追溯。
3.3 可观测性与监控
采集系统的稳定运行离不开完善的监控体系。OpenClaw 内置了基于 Prometheus 的指标采集器，默认暴露请求成功率、响应延迟分布、文件下载速率、解析错误率等关键指标。框架还支持结构化日志输出，每条日志包含任务 ID、URL、状态码、耗时等字段，便于通过 ELK 等日志分析平台进行聚合分析。
告警机制方面，OpenClaw 支持配置多种告警规则，例如连续失败次数超过阈值、请求延迟超过预设值、磁盘空间不足等。告警可以通过邮件、企业微信、钉钉等渠道推送，帮助运维人员及时发现问题并介入处理。
四、公告发现与合规下载策略
公告发现是采集流水线的起点，其核心任务是从信息披露平台的列表页中识别出新发布的公告，提取公告元数据，并生成下载任务。这个过程需要兼顾覆盖率和时效性，同时严格遵守合规要求。
4.1 列表页遍历与增量采集
巨潮资讯网等平台提供了按日期范围检索公告的接口，采集系统可以以天为单位逐个遍历日期，获取当日发布的所有公告列表。对于首次运行的全量采集，需要从当前日期回溯到目标起始日期，逐日获取历史公告。对于日常运行的增量采集，只需要获取当日新增的公告即可。
在实际实现中，列表页通常采用分页设计，每页显示 20 条到 50 条公告记录。采集系统需要先获取总页数，然后逐页遍历，提取每条公告的标题、代码、日期和详情页链接。为了避免对服务器造成压力，OpenClaw 的请求调度器会在每页请求之间设置合理的间隔时间，通常建议设置为 1 到 3 秒。
增量采集的关键在于去重机制。系统需要维护一个已采集公告的哈希集合，对于每条新发现的公告，先计算其唯一标识（通常为公告标题加日期加股票代码的组合哈希），判断是否已经存在于数据库中。如果存在，则跳过该公告；如果不存在，则创建下载任务。
4.2 公告详情页解析
公告列表页通常只提供公告的基本元数据，PDF 文件的下载链接需要在详情页中获取。采集系统需要访问每条公告的详情页，从中提取 PDF 下载链接。详情页的 HTML 结构相对固定，可以通过 CSS 选择器或 XPath 定位到下载链接所在的元素。
部分信息披露平台对 PDF 下载链接进行了动态处理，链接可能通过 JavaScript 生成或包含时效性令牌。对于这种情况，OpenClaw 支持通过 Selenium 或 Playwright 等无头浏览器进行页面渲染，获取动态生成的下载链接。当然，无头浏览器的资源消耗远高于普通的 HTTP 请求，应当仅在必要时使用，并做好资源规划。
4.3 PDF 文件下载与校验
获取到 PDF 下载链接后，下载组件会启动文件下载任务。OpenClaw 的下载组件支持断点续传和下载重试机制，对于网络波动导致的下载中断，组件会自动从断点处恢复下载，避免重复传输已经下载的数据。下载完成后，组件会计算文件的 MD5 或 SHA256 哈希值，与数据库中已有的哈希值进行比对，如果发现完全相同的文件，则说明该公告已经被采集过，可以跳过后续的解析步骤。
文件下载的存储路径需要合理规划。建议按日期和公告类型组织目录结构，例如「2024/01/15/年度报告/000001.pdf」。这样的目录结构既便于人工浏览，也便于后续的批量处理和数据清理。
五、公告正文解析：从 PDF 到结构化文本
PDF 文件下载到本地后，下一步是将其中的文本内容提取出来，转换为可用于信息提取的结构化文本。这是整个流水线中技术难度最高的环节之一，因为 PDF 文件的内部结构差异巨大，从直接可提取文本的原生 PDF 到几乎不可读取的扫描件不一而足。
5.1 PDF 文本提取技术选型
对于原生 PDF 文件，即文字内容以字体和字符编码形式存储在 PDF 中的文件，可以使用 PdfMiner、PyMuPDF 等 Python 库直接提取文本。PyMuPDF 在提取速度上具有明显优势，单页提取时间通常在毫秒级，适合大批量处理场景。PdfMiner 在文本位置信息的保留上更加精细，对于需要识别表格结构的场景更为适用。
对于扫描件 PDF，即文字内容以图片形式存在的文件，需要使用 OCR 技术进行文字识别。开源的 Tesseract 引擎配合中文语言包可以处理大部分扫描件，但其识别准确率受图片质量影响较大。对于质量较差的扫描件，可以考虑使用 PaddleOCR 等基于深度学习的 OCR 引擎，这些引擎在中文识别上表现出色，尤其对于表格、特殊字体等复杂场景有更好的适应性。
在实际工程中，建议采用混合策略：先用 PyMuPDF 尝试提取文本，如果提取结果中有效文本比例低于阈值（例如 30%），则自动切换到 OCR 流程。这样可以在保证提取质量的同时最小化 OCR 带来的计算开销。
5.2 公告文本清洗与规范化
从 PDF 中提取出的原始文本通常包含大量的格式噪声，如多余的换行符、页码标记、页眉页脚文字、水印等。在进入信息提取环节之前，需要对这些噪声进行清洗。常见的清洗步骤包括：移除页码（通常为纯数字行或「第 X 页」格式）、移除页眉页脚（通常为重复出现的固定文字）、合并被错误分割的段落、统一全角和半角标点符号、去除多余空白字符等。
段落合并是文本清洗中的关键步骤。PDF 提取工具往往会将同一段落中的不同行当作独立的文本行输出，导致后续的语义分析无法正确识别段落边界。段落合并算法需要判断相邻两行是否属于同一段落。一个简单而有效的策略是：如果上一行的最后一个字符不是句号、问号、感叹号等句子结束标点，且下一行的第一个字符不是空格或缩进，则两行很可能属于同一段落，应当合并。
对于表格内容，文本提取工具通常会丢失表格的结构信息，将表格中的数字和文字混排在一起。对于定期报告中的财务报表，表格结构的保留至关重要。PyMuPDF 提供了表格检测功能，可以识别 PDF 中的表格区域并提取为结构化的行列表。对于复杂的嵌套表格，可能需要结合规则和机器学习方法进行表格识别和重建。
5.3 公告类型自动识别
不同类型的公告在信息提取时需要使用不同的模板和规则。公告类型识别可以通过标题模式匹配来实现。例如，标题中包含「年度报告」「半年度报告」「季度报告」等关键词的公告属于定期报告；标题中包含「业绩预告」「业绩快报」的公告属于业绩相关公告；标题中包含「重大合同」「对外投资」的公告属于经营事项公告。
对于标题模式无法准确分类的公告，可以使用基于文本分类的机器学习方法。将公告正文的前若干段落作为输入，训练一个轻量级的文本分类模型，输出公告类型标签。在实际应用中，标题模式匹配和文本分类可以结合使用：先用标题模式进行快速分类，对于分类置信度较低的公告，再使用文本分类模型进行二次判断。
六、关键财务信息提取：从文本到结构化指标
财务信息是上市公司公告中价值密度最高的数据之一。年度报告、半年度报告和季度报告中的财务报表包含了营收、利润、资产、负债、现金流等核心财务指标，业绩预告和业绩快报则提供了前瞻性的财务展望。从公告文本中自动提取这些财务指标，并将其结构化存储，是公告采集系统的核心能力之一。
6.1 财务报表的定位与解析
对于定期报告中的财务报表，首先需要定位财务报表所在的章节。通常，财务报表会出现在公告的后半部分，常见的章节标题包括「财务报表」「合并财务报表」「财务报告」「审计报告」等。通过关键词匹配可以在公告目录中快速定位到财务报表的起始页码，然后在 PDF 中精确定位到对应的页面。
财务报表的表格结构通常比较规整，包含行标题（如「营业收入」「营业成本」「净利润」等）、列标题（如「2024 年」「2023 年」「同比增减」等）和数值单元格。在定位到财务报表页面后，可以使用 PyMuPDF 的表格提取功能将表格转换为二维数组，然后通过行标题匹配来提取特定财务指标。
然而，不同公司的财务报表在科目名称上存在差异。例如，同样是「营业收入」，有的公司可能使用「营业总收入」或「主营业务收入」；同样是「净利润」，有的公司可能区分「归属于母公司所有者的净利润」和「少数股东损益」。为了统一处理这些差异，需要建立一套财务科目映射表，将不同表述映射到标准化的科目名称上。
6.2 关键财务指标提取规则
在财务分析中，以下指标通常是最受关注的：营业收入及其增长率、归母净利润及其增长率、扣非净利润、毛利率、净利率、净资产收益率、每股收益、经营活动现金流净额、资产负债率等。这些指标可以从资产负债表、利润表和现金流量表中分别提取。
提取规则的设计需要兼顾准确性和覆盖率。对于营业收入，可以匹配「营业收入」「营业总收入」「主营业务收入」等变体，取报告期和上年同期的数值，计算同比增长率。对于净利润，优先匹配「归属于母公司所有者的净利润」，如果不存在则匹配「净利润」行。对于每股收益，通常匹配「基本每股收益」行，同时区分「基本每股收益」和「稀释每股收益」。
部分财务指标需要从多个原始数据计算得出。例如，毛利率等于（营业收入减营业成本）除以营业收入，净利率等于净利润除以营业收入，净资产收益率等于净利润除以平均净资产。这些计算逻辑需要在提取程序中实现，并确保计算过程中使用的原始数据来自同一份公告。
6.3 业绩预告与业绩快报的专项提取
业绩预告和业绩快报是上市公司在定期报告发布前披露的财务预测信息，其时效性往往高于定期报告本身。业绩预告通常以文字描述为主，例如「预计 2024 年度归属于上市公司股东的净利润为 5000 万元至 6000 万元，同比增长 20% 至 40%」。这种区间值的提取比精确值的提取更加复杂，需要从文本中识别出数值区间和对应的增长率区间。
对于业绩预告的文本，可以使用正则表达式配合命名实体识别来提取关键信息。典型的正则表达式模式包括：「预计[^，。]*净利润[^，。]*\\d+[万万千千]?元[^，。]*至[^，。]*\\d+[万万千千]?元」。提取出数值区间后，还需要判断区间的上下限，并结合上年同期数据计算增长率区间。
业绩快报的格式相对规整，通常包含一张简化的财务报表，可以直接使用表格提取方法进行处理。业绩快报中的财务指标与定期报告中的对应指标应当保持一致，但快报数据未经审计，存在后续调整的可能性，在数据存储时需要标注数据来源为「业绩快报」以示区分。
6.4 财务数据的交叉验证
同一家公司在不同时间点发布的公告中可能包含同一财务指标的不同版本。例如，业绩预告中的数值是基于初步核算的预测，业绩快报中的数值是基于初步审计的估算，年度报告中的数值是经审计确认的最终数据。采集系统需要维护一个财务指标的时间序列，记录每个版本的数据来源和发布时间，并在最终版本发布后自动更新为确认数据。
交叉验证的另一个维度是同一份公告内部的一致性检查。例如，利润表中的营业收入应当与营业收入明细表中的合计数一致，资产负债表中的资产总计应当等于负债合计加所有者权益合计。如果发现内部数据不一致，系统应当标记该公告为「数据异常」，提醒人工复核。
七、业务变动信息提取：事件驱动的结构化分析
除了财务数据，上市公司公告中还包含大量的业务变动信息，这些信息对于理解企业的经营动态同样重要。重大合同签署、对外投资、资产重组、股权变动、人事变更、诉讼仲裁等事件，每一项都可能影响企业的未来发展和市场预期。
7.1 业务变动事件的类型体系
根据证监会公告分类指引和实际业务需求，我们可以将业务变动事件归纳为以下主要类型：
合同与订单类事件：包括重大销售合同、战略合作协议、中标公告等。这类事件的核心信息包括合同金额、合作方、合同期限、对业绩的影响等。对于以项目制运营为主的公司，新签合同金额是判断未来收入的重要先行指标。
投资与并购类事件：包括对外投资、收购兼并、设立子公司、增资扩股等。这类事件的核心信息包括投资标的、投资金额、持股比例、标的估值、交易方式等。对于产业链整合和多元化扩张的分析，这类事件是关键数据源。
股权变动类事件：包括股东增持减持、股权质押、股权转让、回购股份等。这类事件直接反映了大股东和机构投资者对公司价值的判断，也是市场情绪的重要风向标。核心信息包括变动方、变动数量、变动比例、变动价格区间、变动后持股比例等。
人事变动类事件：包括董事、监事、高级管理人员的任免、辞职、聘任等。核心管理层的变化往往预示着公司战略方向的调整，尤其是董事长、总经理和财务负责人的变动最受关注。核心信息包括变动人员姓名、职位、变动原因、继任者信息等。
风险与合规类事件：包括诉讼仲裁、行政处罚、监管问询、退市风险警示等。这类事件是风险预警的重要来源，核心信息包括事件类型、涉及金额、处理进展、对公司的影响等。
7.2 基于规则的事件识别与要素提取
对于公告类型明确、格式相对固定的业务变动事件，可以使用基于规则的方法进行识别和要素提取。规则的核心是关键词匹配和正则表达式，辅以公告标题的类型判断来确定当前适用的提取模板。
以重大合同公告为例，典型的提取规则流程如下：首先通过公告标题中的「重大合同」「重大销售合同」「战略合作协议」等关键词确定公告类型。然后在正文中搜索合同金额，常见的金额表述模式包括「合同金额为人民币 XX 万元」「合同总金额约 XX 亿元」「合同金额不超过 XX 万元」等。接着提取合作方名称，通常出现在「与 XX 公司签订了」「甲方为 XX 有限公司」等上下文中。最后提取合同期限和对业绩的影响说明。
以股东减持公告为例，提取规则需要识别减持主体、减持方式（集中竞价、大宗交易、协议转让等）、减持数量、减持比例、减持价格区间、减持期间、减持前后持股比例等关键要素。这些要素在公告中通常出现在固定的段落位置，可以通过正则表达式配合位置信息进行提取。
7.3 基于深度学习的事件抽取
对于公告格式不固定、事件表述多样化的情况，基于规则的方法存在覆盖率不足的问题。此时可以引入基于深度学习的事件抽取模型。事件抽取是自然语言处理中的一个经典任务，目标是从文本中识别出事件类型、事件触发词和事件论元（即事件要素）。
在上市公司公告场景中，事件类型对应前述的业务变动事件类型体系，事件触发词对应「签订」「收购」「减持」「任命」等标志性动词，事件论元对应金额、比例、日期、主体等要素。可以使用预训练语言模型（如 BERT、RoBERTa 等）进行事件抽取的微调训练，输入为公告段落文本，输出为事件类型标签和要素标注。
训练数据可以来源于人工标注的公告样本。标注工作包括两个层次：段落级别的分类标注，判断该段落是否包含某类业务变动事件；要素级别的序列标注，标注出段落中每个要素的起止位置和类型。为了提高标注效率，可以先用规则方法进行预标注，然后由人工进行修正和补充，形成半自动化的标注流水线。
7.4 事件关联与影响分析
单独的事件提取只是第一步，将事件与公司基本面、行业动态、市场表现关联起来才能发挥更大的价值。事件关联分析可以从以下几个维度展开：
时间序列关联：将同一公司的同类事件按时间排列，分析事件的频率和趋势变化。例如，如果一家公司在短时间内频繁发布股东减持公告，可能意味着大股东对公司前景存在担忧；如果一家公司连续发布重大合同公告，说明其业务拓展能力较强。
财务指标关联：将事件与财务数据交叉分析。例如，分析重大合同签署后公司营收的变化趋势，判断合同的实际执行情况；分析股东减持后公司股价的波动，评估减持对市场的影响程度。
行业横向关联：将同一行业多家公司的同类事件进行对比，识别行业共性趋势。例如，如果多家公司同时发布资产减值公告，可能意味着该行业正面临系统性风险；如果多家公司同时发布对外投资公告，可能意味着该行业正处于扩张周期。
八、工程化部署与运维实践
将公告采集系统从开发环境部署到生产环境，需要解决定时调度、任务编排、容错处理、资源管理等一系列工程化问题。本节将介绍基于 OpenClaw 框架的工程化部署方案和运维实践。
8.1 定时调度与任务编排
公告采集是一个典型的周期性任务。增量采集通常每天执行一次，在交易所公告集中发布的高峰时段（通常为下午收盘后）之后启动。全量采集或历史数据回补可以按需触发，执行频率较低但处理量较大。
OpenClaw 框架内置了基于 Cron 表达式的定时调度器，支持配置多个采集任务的时间表。每个任务可以指定启动时间、执行频率和超时时间。对于需要按顺序执行的任务（例如先下载再解析），可以使用任务编排功能定义任务之间的依赖关系。
对于更复杂的任务编排需求，可以考虑将 OpenClaw 与 Apache Airflow 或 Prefect 等工作流调度平台集成。Airflow 的 DAG 模型非常适合描述公告采集的流水线：URL 发现是一个 Task，PDF 下载是下游 Task，文本解析是再下游 Task，信息提取是最终 Task。Airflow 提供了丰富的监控和告警功能，以及失败重试和任务回溯能力。
8.2 分布式部署与水平扩展
当公告采集量增长到每天数千份甚至上万份时，单机部署的性能瓶颈会逐渐显现。OpenClaw 支持基于消息队列的分布式部署架构，将 URL 发现、内容下载、数据解析等组件部署在不同的节点上，通过消息队列进行任务分发和结果收集。
推荐使用 Redis 作为消息队列的中间件。URL 发现组件将发现的新公告 URL 推送到 Redis 队列中，多个下载节点从队列中消费 URL 并执行下载任务，下载完成后将文件路径推送到另一个队列中，多个解析节点从该队列中消费文件路径并执行解析任务。这种架构天然支持水平扩展，增加下载节点或解析节点即可提升吞吐量。
在分布式部署中，需要注意数据库的并发写入问题。建议使用支持高并发写入的数据库，如 PostgreSQL 或 MongoDB，并合理设置数据库连接池的大小。对于去重检查，可以使用 Redis 的 Set 数据结构来维护已采集公告的哈希集合，实现毫秒级的去重判断。
8.3 数据存储与索引设计
公告采集系统产生的数据包括三类：公告文件（PDF 等原始文件）、公告元数据（标题、代码、日期、类型等）和提取的结构化信息（财务指标、业务事件等）。这三类数据需要分别规划存储方案。
公告文件适合存储在对象存储服务中，如 MinIO 或阿里云 OSS。对象存储的成本低、容量大，且支持通过 HTTP 协议直接访问，便于后续的在线阅读和文件分享。公告元数据和结构化信息适合存储在关系型数据库中，如 PostgreSQL。元数据表的设计需要包含公告 ID、标题、公司代码、公告日期、公告类型、文件路径、文件哈希值、采集时间等字段。结构化信息包括财务指标表和业务事件表，财务指标表的设计需要包含公司代码、报告期、指标名称、指标值、数据来源等字段，业务事件表的设计需要包含公司代码、事件日期、事件类型、事件要素等字段。
为了支持高效的查询和检索，需要为数据库表建立合理的索引。对于元数据表，公司代码加公告日期的联合索引是最常用的查询模式。对于财务指标表，公司代码加报告期加指标名称的联合索引可以支持快速的指标查询和对比。对于业务事件表，公司代码加事件类型加事件日期的联合索引可以支持事件统计和趋势分析。此外，可以考虑使用 Elasticsearch 建立公告全文索引，支持关键词搜索和语义检索。
8.4 监控告警与运维自动化
生产环境的公告采集系统需要 7×24 小时稳定运行，监控告警和运维自动化是必不可少的保障措施。OpenClaw 框架内置的 Prometheus 指标采集器可以暴露数十个关键指标，包括：每个采集任务的成功率、失败率、平均耗时、P99 耗时；每个目标网站的请求成功率、响应延迟、限流次数；PDF 下载的成功率、平均文件大小、下载速度；文本解析的成功率、OCR 触发率、解析耗时等。
基于这些指标，可以配置多层次的告警规则。第一层是基础设施告警，如 CPU 使用率超过 80%、内存使用率超过 90%、磁盘空间低于 20%、网络连接中断等。第二层是业务告警，如采集任务连续失败 3 次、单日采集量低于历史均值的 50%、PDF 下载成功率低于 95% 等。第三层是数据质量告警，如解析出的财务指标为空值比例超过阈值、提取的业务事件数量异常波动等。
告警信息可以通过企业微信机器人、钉钉机器人或飞书机器人推送到运维团队，也可以集成到 PagerDuty 等事件管理平台，实现告警的升级和值班轮转。对于常见的故障场景，可以编写自动化运维脚本来进行初步的故障恢复，例如自动重启失败的采集任务、自动清理过期的临时文件、自动切换备用数据源等。
九、实战案例：构建完整的公告采集与信息提取流水线
本节通过一个完整的实战案例，展示如何使用 OpenClaw 框架从零开始构建一套可用的上市公司公告采集与信息提取流水线。案例覆盖了从环境搭建、配置编写、代码开发到运行测试的全流程。
9.1 环境准备与框架安装
首先，确保系统已安装 Python 3.9 或更高版本。OpenClaw 框架可以通过 pip 安装：
pip install openclaw同时安装必要的依赖，包括 PDF 解析库和数据库驱动：
pip install pymupdf paddleocr redis psycopg2-binary elasticsearch对于 OCR 功能，需要额外安装 PaddlePaddle 深度学习框架。如果系统有 NVIDIA GPU，建议安装 GPU 版本以加速 OCR 处理：
pip install paddlepaddle-gpu如果只有 CPU，则安装 CPU 版本：
pip install paddlepaddle数据库方面，推荐使用 Docker 快速部署 PostgreSQL 和 Redis：
docker run -d --name postgres -p 5432:5432 \ -e POSTGRES_PASSWORD=your_password \ -e POSTGRES_DB=announcement_db \ postgres:15 docker run -d --name redis -p 6379:6379 redis:79.2 配置文件编写
OpenClaw 使用 YAML 格式的配置文件来定义采集任务。以下是一个针对巨潮资讯网公告采集的配置示例：
# config.yaml project: name: "cninfo_announcement_collector" version: "1.0.0" source: name: "巨潮资讯网" base_url: "http://www.cninfo.com.cn" scheduler: type: "cron" cron: "0 18 * * *" # 每天下午 6 点执行 rate_limit: requests_per_second: 1 max_concurrent: 3 respect_robots_txt: true user_agent: template: "OpenClaw-AnnouncementCollector/1.0 (research@example.com)" storage: download_dir: "/data/announcements/pdf" database: type: "postgresql" host: "localhost" port: 5432 database: "announcement_db" user: "postgres" password: "your_password" redis: host: "localhost" port: 6379 db: 0 extraction: financial_indicators: - 营业收入 - 归母净利润 - 扣非净利润 - 毛利率 - 净利率 - 净资产收益率 - 基本每股收益 - 经营活动现金流净额 - 资产负债率 business_events: - 重大合同 - 对外投资 - 股东减持 - 股东增持 - 人事变动 - 诉讼仲裁 monitoring: prometheus: enabled: true port: 9090 alerting: webhook_url: "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx"9.3 核心采集逻辑实现
以下是公告发现和下载的核心代码框架。代码展示了如何从巨潮资讯网的公告列表页提取公告信息，并生成下载任务：
from openclaw import Collector, Task, Response from openclaw.contrib.parser import parse_announcement_list from typing import List import hashlib import logging logger = logging.getLogger(name) class AnnouncementCollector(Collector): """上市公司公告采集器""" def __init__(self, config): super().__init__(config) self.base_url = config["source"]["base_url"] self.collected_hashes = set() async def discover(self, date_str: str) -> List[Task]: """发现指定日期的公告列表""" tasks = [] page = 1 while True: url = self._build_list_url(date_str, page) response = await self.fetch(url) announcements = parse_announcement_list(response.text) if not announcements: break for ann in announcements: ann_hash = self._compute_hash(ann) if ann_hash in self.collected_hashes: continue task = Task( url=ann["detail_url"], meta={ "title": ann["title"], "code": ann["stock_code"], "date": date_str, "ann_type": ann["ann_type"], "hash": ann_hash } ) tasks.append(task) self.collected_hashes.add(ann_hash) page += 1 if page &amp;gt; 3: break logger.info(f"发现 {len(tasks)} 条新公告") return tasks def build_list_url(self, date_str: str, page: int) -> str: return ( f"{self.base_url}/new/announcement/query?" f"stock=&searchkey=&plate=&category=&" f"trade=&column=&columnTitle=&" f"pageNum={page}&pageSize=30&" f"startDate={date_str}&endDate={date_str}" ) def compute_hash(self, ann: dict) -> str: raw = f"{ann['stock_code']}{ann['title']}{ann['date']}" return hashlib.md5(raw.encode()).hexdigest()</code></pre> 9.4 PDF 解析与信息提取 PDF 下载完成后，解析组件负责提取文本并识别关键信息。以下代码展示了财务报表定位和营业收入提取的实现： import fitz # PyMuPDF import re from typing import Optional, Dict, List class PDFAnnouncementParser: """PDF 公告解析器""" def init(self, pdf_path: str): self.doc = fitz.open(pdf_path) self.text_pages = [] self._extract_text() def _extract_text(self): for page in self.doc: text = page.get_text() self.text_pages.append(text) def locate_financial_statements(self) -> Optional[int]: """定位财务报表所在页码""" keywords = [ "合并财务报表", "财务报表", "财务报告", "审计报告及财务报表" ] for idx, text in enumerate(self.text_pages): for keyword in keywords: if keyword in text: return idx return None def extract_revenue(self) -> Optional[Dict[str, float]]: """提取营业收入数据""" start_page = self.locate_financial_statements() if start_page is None: return None patterns = [ r"营业(?:总)?收入[^\d]?([\d,]+.?\d)\s万?元?", r"主营业务收入[^\d]?([\d,]+.?\d*)\s*万?元?", ] for page_idx in range(start_page, min(start_page + 10, len(self.text_pages))): text = self.text_pages[page_idx] for pattern in patterns: match = re.search(pattern, text) if match: value_str = match.group(1).replace(",", "") try: value = float(value_str) return {"current": value, "unit": "万元"} except ValueError: continue return None def locate_balance_sheet(self) -> Optional[int]: """定位资产负债表所在页码""" keywords = ["资产负债表", "合并资产负债表"] for idx, text in enumerate(self.text_pages): for keyword in keywords: if keyword in text: return idx return None def extract_total_assets(self) -> Optional[float]: """提取总资产数据""" start_page = self.locate_balance_sheet() if start_page is None: return None pattern = r"资产总[计計][^\d]?([\d,]+.?\d)\s*万?元?" for page_idx in range(start_page, min(start_page + 5, len(self.text_pages))): text = self.text_pages[page_idx] match = re.search(pattern, text) if match: value_str = match.group(1).replace(",", "") try: return float(value_str) except ValueError: continue return None</code></pre> 对于业务变动事件的提取，以下代码展示了基于规则的事件识别方法，以股东减持公告为例： class BusinessEventExtractor: """业务变动事件提取器""" def init(self): self.event_patterns = { "股东减持": { "keywords": ["减持", "股份减持", "减持计划"], "elements": { "减持主体": r"([\u4e00-\u9fa5]+(?:有限公司|集团|投资|控股|基金|公司)?)\s*(?:拟|计划|决定|拟通过)", "减持数量": r"(?:减持|不超过)\s*([\d,]+.?\d*)\s万?股", "减持比例": r"(?:占.?总股本[的之]?|不超过总股本[的之]?)\s*([\d.]+)\s*%", "减持价格区间": r"(?:减持价格|价格区间)[为在]?\s*([\d.]+\s*[-至到]\s*[\d.]+)\s元", "减持期间": r"(?:自本公告披露之日起|自)\s(\d{4}年\d{1,2}月\d{1,2}日[起至]\d{4}年\d{1,2}月\d{1,2}日)", } }, "重大合同": { "keywords": ["重大合同", "重大销售合同", "战略合作协议", "中标"], "elements": { "合同金额": r"(?:合同金额|中标金额|合同总金额)[为约]?\s人民币\s*([\d,]+.?\d*)\s万?元", "合作方": r"(?:与|同)\s([\u4e00-\u9fa5]+(?:有限公司|集团|公司|企业))\s签订", "合同期限": r"(?:合同期限|合作期限|有效期)[为]?\s(\d+年|[\d]+个?月)", } }, "对外投资": { "keywords": ["对外投资", "投资设立", "增资", "收购"], "elements": { "投资金额": r"(?:投资金额|增资金额|收购金额)[为约]?\s人民币\s([\d,]+.?\d*)\s万?元", "投资标的": r"(?:投资|设立|收购)\s([\u4e00-\u9fa5]+(?:有限公司|有限责任公司|股份有限公司))", "持股比例": r"(?:持股比例|持有|占股)[为的]?\s*([\d.]+)\s*%", } }, } def extract(self, text: str, ann_type: str) -> List[Dict]: """从公告文本中提取业务变动事件""" events = [] if ann_type not in self.event_patterns: return events pattern_config = self.event_patterns[ann_type] for keyword in pattern_config["keywords"]: if keyword not in text: continue event = {"type": ann_type, "elements": {}} for element_name, element_pattern in pattern_config["elements"].items(): match = re.search(element_pattern, text) if match: event["elements"][element_name] = match.group(1) if len(event["elements"]) &amp;gt;= 2: events.append(event) break return events</code></pre> 9.5 数据库存储与查询 提取的结构化数据需要持久化到数据库中。以下代码展示了数据库表的设计和数据写入逻辑： import psycopg2 from psycopg2.extras import execute_values from datetime import datetime from typing import List, Dict class AnnouncementDatabase: """公告数据库管理""" def init(self, config: dict): self.conn = psycopg2.connect( host=config["host"], port=config["port"], database=config["database"], user=config["user"], password=config["password"] ) self._init_tables() def _init_tables(self): with self.conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS announcements ( id SERIAL PRIMARY KEY, stock_code VARCHAR(10) NOT NULL, title VARCHAR(500) NOT NULL, ann_date DATE NOT NULL, ann_type VARCHAR(50), file_path VARCHAR(500), file_hash VARCHAR(64), created_at TIMESTAMP DEFAULT NOW(), UNIQUE(stock_code, title, ann_date) ); CREATE TABLE IF NOT EXISTS financial_indicators ( id SERIAL PRIMARY KEY, announcement_id INTEGER REFERENCES announcements(id), indicator_name VARCHAR(100) NOT NULL, indicator_value NUMERIC(20, 2), report_period VARCHAR(10), data_source VARCHAR(50), extracted_at TIMESTAMP DEFAULT NOW() ); CREATE TABLE IF NOT EXISTS business_events ( id SERIAL PRIMARY KEY, announcement_id INTEGER REFERENCES announcements(id), event_type VARCHAR(100) NOT NULL, event_date DATE, event_elements JSONB, extracted_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_ann_date ON announcements(stock_code, ann_date); CREATE INDEX IF NOT EXISTS idx_fin_indicator ON financial_indicators(announcement_id, indicator_name); CREATE INDEX IF NOT EXISTS idx_event_type ON business_events(announcement_id, event_type); """) self.conn.commit() def save_announcement(self, meta: dict, file_hash: str) -> int: with self.conn.cursor() as cur: cur.execute(""" INSERT INTO announcements (stock_code, title, ann_date, ann_type, file_path, file_hash) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (stock_code, title, ann_date) DO NOTHING RETURNING id """, (meta["code"], meta["title"], meta["date"], meta["ann_type"], meta.get("file_path"), file_hash)) result = cur.fetchone() self.conn.commit() return result[0] if result else None def save_financial_indicators(self, ann_id: int, indicators: List[Dict]): if not indicators: return with self.conn.cursor() as cur: values = [ (ann_id, ind["name"], ind["value"], ind.get("period"), ind.get("source")) for ind in indicators ] execute_values(cur, """ INSERT INTO financial_indicators (announcement_id, indicator_name, indicator_value, report_period, data_source) VALUES %s """, values) self.conn.commit() def save_business_events(self, ann_id: int, events: List[Dict]): if not events: return with self.conn.cursor() as cur: import json values = [ (ann_id, evt["type"], evt.get("date"), json.dumps(evt.get("elements", {}), ensure_ascii=False)) for evt in events ] execute_values(cur, """ INSERT INTO business_events (announcement_id, event_type, event_date, event_elements) VALUES %s """, values) self.conn.commit()</code></pre> 9.6 流水线调度与运行 将上述组件组装成完整的流水线，通过 OpenClaw 的调度器定时执行： import asyncio from openclaw import Pipeline, Scheduler async def main(): config = load_config("config.yaml") collector = AnnouncementCollector(config) parser = PDFAnnouncementParser extractor = BusinessEventExtractor() database = AnnouncementDatabase(config["storage"]["database"]) pipeline = Pipeline(name="announcement_pipeline") pipeline.add_stage("discover", collector.discover) pipeline.add_stage("download", collector.download) pipeline.add_stage("parse", parser) pipeline.add_stage("extract", extractor.extract) pipeline.add_stage("store", database.save) scheduler = Scheduler(config["scheduler"]) scheduler.register(pipeline) await scheduler.start() if name == "main": asyncio.run(main()) 流水线启动后，每天下午 6 点会自动执行增量采集。运维人员可以通过 Prometheus 监控面板查看采集进度和成功率，通过日志分析平台排查异常公告的解析失败原因。整个系统在 8 核 32GB 内存的服务器上运行，日均处理公告量约 2000 份，单份公告的平均处理时间约为 30 秒，其中 PDF 下载耗时约 5 秒、文本提取耗时约 10 秒、信息提取耗时约 15 秒。 十、技术难点与优化策略 在实际运行中，公告采集系统会遇到各种技术挑战。本节将讨论几个常见的技术难点及其优化策略，帮助读者在遇到类似问题时能够快速找到解决方案。 10.1 PDF 表格提取的精度优化 财务报表中的表格是信息提取的核心来源，但 PDF 表格提取的精度问题始终是一个技术挑战。PyMuPDF 的表格检测功能在处理规整表格时表现良好，但对于存在合并单元格、跨页表格、无边框表格等复杂情况时，提取结果往往不够理想。 针对合并单元格的问题，可以通过后处理逻辑来修复。合并单元格在提取结果中通常表现为空值或者仅在第一行有值、后续行为空的情况。后处理逻辑可以检测同一列中的连续空值，将其填充为上方最近的非空值，从而还原合并单元格的语义。 针对跨页表格的问题，需要检测表格是否在页面底部被截断。一种启发式方法是判断表格的最后一行是否包含「合计」「总计」等汇总行，如果没有，则表格很可能在下一页继续。跨页表格需要在下一页找到表格的延续部分，将表头行去除后将数据行拼接在一起。 对于表格提取精度要求极高的场景，可以考虑使用基于深度学习的表格识别模型，如 TableBank 或 PubTables-1M 等预训练模型。这些模型在表格结构识别上比传统方法有显著优势，但推理耗时较长，适合在离线批处理场景中使用。 10.2 扫描件 OCR 的效率与准确性平衡 虽然大部分近期公告已经是原生 PDF 格式，但历史公告中仍有相当比例的扫描件。OCR 处理扫描件的计算开销远大于原生 PDF 的文本提取，单页 OCR 处理时间可能需要数秒甚至数十秒，对于包含数百页的年度报告，OCR 处理的总耗时可能超过 10 分钟。 为了在效率和准确性之间取得平衡，可以采取以下优化策略：首先，对于扫描件公告，优先使用 PaddleOCR 的轻量级模型，在保证基本识别准确率的前提下尽量降低推理耗时。其次，对于公告中的非关键页面（如封面、目录、声明页等），可以跳过 OCR 处理，仅对财务报表、重要事项说明等核心页面进行 OCR。再次，可以利用 GPU 加速 OCR 推理，单张 GPU 卡的 OCR 处理速度可以达到 CPU 的 10 倍以上。 另外，对于扫描件中的表格，传统 OCR 工具会丢失表格结构。可以结合 PaddleOCR 的表格识别功能，或使用专门的表格 OCR 模型（如 TableMaster）来识别和重建表格结构。对于历史公告的批量 OCR 处理，建议使用异步任务队列，将 OCR 任务分散到多个工作节点上并行执行，减少单次处理的等待时间。 10.3 公告去重与版本管理 在实际运行中，同一份公告可能会被多次采集。例如，一家公司可能在收盘后发布业绩预告，第二天早上又在交易所网站重新发布一遍。如果去重机制不完善，数据库中会积累大量重复数据。 去重的核心是唯一标识的生成。简单的做法是基于公告标题加股票代码加公告日期的组合进行去重，但标题的微小变化（如多一个空格、标点符号不同）可能导致去重失败。更稳健的做法是结合文件哈希值进行去重，完全相同的 PDF 文件生成相同的哈希值，可以精确识别重复文件。对于内容相同但文件格式不同的公告（如同一公告的 PDF 和 Word 版本），可以提取文本内容后计算文本哈希值，实现跨格式的去重。 版本管理方面，同一份公告可能存在多个版本，如业绩预告的修正公告、年度报告的补充公告等。采集系统需要维护公告之间的版本关联关系，标记每个版本的状态（原始版本、修正版本、最终版本），确保下游使用者在分析数据时使用的是最新版本的信息。 10.4 反爬策略的应对与合规边界 信息披露平台通常会部署一定程度的反爬措施，如 IP 频率限制、验证码验证、请求头检查等。OpenClaw 框架通过在请求之间设置合理的间隔时间、使用 User-Agent 声明采集身份、遵守 robots.txt 规则等方式，在合规范围内尽量降低被反爬策略拦截的概率。 如果遇到 IP 被临时封禁的情况，框架会自动暂停对该域名的请求，等待封禁解除后再恢复采集。对于需要登录才能访问的内容，OpenClaw 不会尝试绕过登录机制，因为这些内容不属于公开数据的范畴。如果目标网站的反爬策略过于严格，导致合规采集无法正常进行，可以考虑使用网站官方提供的数据接口或购买商业数据服务，而不是通过技术手段强行突破反爬防线。 合规采集的底线是：只采集公开可访问的数据，不尝试绕过任何访问控制机制，不过度占用目标网站的服务器资源，尊重数据所有者的意愿。这些原则不仅是法律合规的要求，也是维护互联网数据生态健康发展的必要条件。 十一、总结与展望 本文系统阐述了基于 OpenClaw 框架的上市公司公告采集与信息提取方案，从数据源分析、框架设计、采集策略、文本解析、信息提取到工程化部署，覆盖了公告数据处理的全链路。通过这套方案，可以构建一套日均处理数千份公告、关键信息召回率超过 95% 的自动化采集分析系统，为量化投资、券商研究和企业风险管理等场景提供高质量的结构化数据支持。 回顾整个方案，几个核心设计原则值得强调。第一，合规优先。从采集频率控制到 User-Agent 声明，从 robots.txt 遵守到数据使用边界界定，合规性贯穿于系统设计的每一个环节。第二，可扩展性。通过生产者-消费者架构和消息队列解耦，系统可以灵活地进行水平扩展，应对公告量的增长。第三，鲁棒性。通过混合 PDF 解析策略、规则与模型融合的信息提取方法、多层次的去重机制，系统在面对格式多样、质量参差不齐的公告数据时保持了较高的处理成功率和提取准确率。 展望未来，公告采集系统还有以下几个值得探索的方向。一是在信息提取环节引入大语言模型的能力，利用大模型强大的语义理解能力来处理复杂公告中的非结构化信息，例如董事会报告中的经营分析、管理层讨论中的战略展望等。二是在事件关联分析方面，构建上市公司知识图谱，将公告事件、财务数据、股价走势、行业动态等信息关联起来，形成更加立体的企业画像。三是在实时性方面，探索公告发布的实时监控和秒级响应机制，在公告发布的第一时间完成采集和提取，为高频交易策略提供数据支持。 资本市场的信息披露体系正在不断完善，公告数据的价值也在持续增长。希望本文提供的技术方案能够帮助更多的数据从业者和研究人员高效地利用这一宝贵的数据资源，在合规的前提下挖掘出更深层次的投资洞察和商业价值。