A股个股公告数据模块(配套18)

A股个股公告数据模块(技术学习版)

免责声明:本文档仅供技术学习和学术研究参考,不构成任何投资建议、投资咨询或证券交易指导。股市有风险,投资需谨慎。任何基于本文档内容进行的投资决策,均由投资者自行承担全部风险和责任。


一、模块概述

1.1 模块定位

本模块旨在获取近90个交易日内全A股上市公司的公开公告信息,支持按个股代码查询和按公告类型筛选。模块设计遵循合法合规原则,所有数据均来自中国证监会指定的法定信息披露平台。

核心功能:

  • 全市场公告批量获取(近90个交易日)
  • 个股公告精准查询(按股票代码/简称)
  • 公告类型分类筛选(定期报告、重大事项、股权变动等)
  • 公告文本NLP处理(PDF解析、关键信息抽取、情感分析)
  • 数据持久化存储与可视化

1.2 公告的重要性

信息披露是A股市场的核心制度安排。《中华人民共和国证券法》第七十八条规定,信息披露义务人披露的信息应当真实、准确、完整、简明清晰,通俗易懂,不得有虚假记载、误导性陈述或者重大遗漏。公告作为上市公司与投资者之间最重要的法定信息沟通渠道,是市场透明度的基础保障。

1.3 数据来源

本模块使用的数据源均为公开合法渠道:

数据源 网址 说明
巨潮资讯网 http://www.cninfo.com.cn 深交所旗下法定信息披露平台
上交所 http://www.sse.com.cn 上海证券交易所官网
深交所 http://www.szse.cn 深圳证券交易所官网
北交所 http://www.bse.cn 北京证券交易所官网

1.4 应用场景(仅供学术研究)

本模块仅用于以下学术研究目的:

  • 信息披露制度研究:研究上市公司公告的及时性、完整性与规范性
  • 事件分析方法论:学习事件研究法(Event Study)的理论框架与技术实现
  • 自然语言处理实践:将NLP技术应用于金融文本分析的技术学习
  • 数据工程实践:学习大规模结构化数据的采集、清洗与存储

1.5 明确声明

本模块不用于实际交易信号生成,不构成任何投资建议。 文档中所有策略描述均以"技术分析工具框架"的形式呈现,仅供理解相关技术原理之用。


二、公告类型体系

A股上市公司公告类型繁多,本模块按照证监会信息披露规则和巨潮资讯网的分类体系,将公告划分为以下五大类别:

2.1 定期报告

定期报告是上市公司按照监管要求定期披露的财务与经营信息,是投资者了解公司基本面最核心的信息来源。

子类型 披露周期 核心内容
年度报告 每年4月30日前 经审计的完整财务报表、经营情况讨论、公司治理报告
半年度报告 每年8月31日前 中期财务数据、半年度经营分析
季度报告 季度结束后1个月内 简要财务数据、经营情况概述
业绩预告 定期报告前 预计净利润变动范围及原因
业绩快报 定期报告前 主要财务数据快报(营收、净利润等)

2.2 重大事项

重大事项公告涉及对公司经营、财务或股价可能产生重大影响的事件。

  • 并购重组:资产收购、股权收购、重大资产置换、合并分立
  • 重大合同:中标金额占最近一期经审计净资产10%以上的合同
  • 关联交易:与关联方发生的交易,需独立董事发表意见
  • 诉讼仲裁:涉案金额占净资产10%以上的诉讼或仲裁事项

2.3 股权变动

股权变动类公告直接反映公司股权结构的变化,是公司治理与股东行为研究的重要数据。

子类型 说明
大股东增持/减持 持股5%以上股东的持股变动
股权激励计划 限制性股票、股票期权激励方案
限售股解禁 首发原股东限售股、定向增发限售股解禁
股权质押/解质押 股东将其持有的股份进行质押或解除质押
回购方案 公司回购自身股份的计划与进展

2.4 融资公告

子类型 说明
定向增发 非公开发行股票的预案、实施及结果公告
可转债 可转换公司债券的发行、转股、赎回公告
配股 向原股东配售股份的公告
债券发行 公司债、企业债、中期票据等发行公告

2.5 交易提示

子类型 说明
停牌/复牌 交易暂停与恢复的公告
风险警示 ST、*ST等特别处理警示公告
退市整理 终止上市及退市整理期交易公告
除权除息 权益分派实施公告

三、数据结构设计

3.1 公告主表结构(完整Schema)

CREATE TABLE IF NOT EXISTS stock_announcement (
    id              BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键ID',
    announcement_id VARCHAR(64)  NOT NULL COMMENT '公告唯一标识(巨潮公告ID)',
    stock_code      VARCHAR(16)  NOT NULL COMMENT '股票代码(如000001)',
    stock_name      VARCHAR(64)  NOT NULL COMMENT '股票简称',
    announcement_title VARCHAR(512) NOT NULL COMMENT '公告标题',
    announcement_type VARCHAR(32) NOT NULL COMMENT '公告类型编码',
    announcement_type_name VARCHAR(64) COMMENT '公告类型名称',
    announcement_date DATE        NOT NULL COMMENT '公告披露日期',
    announcement_time DATETIME    COMMENT '公告披露时间',
    announcement_content TEXT     COMMENT '公告正文(纯文本)',
    pdf_url         VARCHAR(512) COMMENT '公告PDF原文链接',
    annex_count     INT          DEFAULT 0 COMMENT '附件数量',
    announcement_year VARCHAR(4) COMMENT '公告年份',
    announcement_month VARCHAR(2) COMMENT '公告月份',
    announcement_week VARCHAR(2) COMMENT '公告所在周数',
    word_count      INT          DEFAULT 0 COMMENT '公告正文字数',
    sentiment_score DECIMAL(5,4) COMMENT '情感分析得分(-1到1,仅供参考)',
    event_type      VARCHAR(64)  COMMENT '识别的事件类型(NLP识别结果)',
    key_entities    JSON         COMMENT '抽取的关键实体(JSON格式)',
    created_at      DATETIME     DEFAULT CURRENT_TIMESTAMP COMMENT '数据入库时间',
    updated_at      DATETIME     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '数据更新时间',
    UNIQUE KEY uk_announcement (announcement_id),
    KEY idx_stock_code (stock_code),
    KEY idx_stock_date (stock_code, announcement_date),
    KEY idx_type_date (announcement_type, announcement_date),
    KEY idx_date (announcement_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='A股个股公告主表';

3.2 公告分类编码表

CREATE TABLE IF NOT EXISTS announcement_category (
    category_code   VARCHAR(16)  PRIMARY KEY COMMENT '分类编码',
    category_name   VARCHAR(64)  NOT NULL COMMENT '分类名称',
    parent_code     VARCHAR(16)  COMMENT '上级分类编码',
    level           INT          DEFAULT 1 COMMENT '分类层级(1-一级,2-二级)',
    description     VARCHAR(256) COMMENT '分类说明',
    sort_order      INT          DEFAULT 0 COMMENT '排序权重'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='公告分类编码表';

-- 初始化分类数据
INSERT INTO announcement_category VALUES
('REGULAR',    '定期报告', NULL, 1, '年报、半年报、季报等定期披露报告', 1),
('REG-ANNUAL', '年度报告', 'REGULAR', 2, '经审计的年度财务报告', 1),
('REG-HALF',   '半年度报告', 'REGULAR', 2, '半年度财务报告', 2),
('REG-QUARTER','季度报告', 'REGULAR', 2, '季度财务报告', 3),
('REG-FORECAST','业绩预告', 'REGULAR', 2, '业绩预告公告', 4),
('REG-EXPRESS','业绩快报', 'REGULAR', 2, '主要财务数据快报', 5),
('MAJOR',      '重大事项', NULL, 1, '对公司有重大影响的事项公告', 2),
('MA-MERGER',  '并购重组', 'MAJOR', 2, '资产收购、合并分立等', 1),
('MA-CONTRACT','重大合同', 'MAJOR', 2, '重大合同中标公告', 2),
('MA-RELATED', '关联交易', 'MAJOR', 2, '关联方交易公告', 3),
('MA-LAWSUIT', '诉讼仲裁', 'MAJOR', 2, '重大诉讼仲裁公告', 4),
('EQUITY',     '股权变动', NULL, 1, '股权结构变动相关公告', 3),
('EQ-HOLDING', '增持减持', 'EQUITY', 2, '大股东持股变动', 1),
('EQ-INCENTIVE','股权激励', 'EQUITY', 2, '股权激励计划', 2),
('EQ-LIFTBAN', '限售解禁', 'EQUITY', 2, '限售股解禁公告', 3),
('EQ-PLEDGE',  '股权质押', 'EQUITY', 2, '股权质押与解质押', 4),
('EQ-BUYBACK', '回购方案', 'EQUITY', 2, '股份回购计划', 5),
('FINANCING',  '融资公告', NULL, 1, '融资相关公告', 4),
('FI-PRIVATE', '定向增发', 'FINANCING', 2, '非公开发行股票', 1),
('FI-BOND',    '可转债', 'FINANCING', 2, '可转换公司债券', 2),
('FI-RIGHTS',  '配股', 'FINANCING', 2, '向原股东配售股份', 3),
('TRADING',    '交易提示', NULL, 1, '交易状态变更提示', 5),
('TR-SUSPEND', '停牌复牌', 'TRADING', 2, '停牌与复牌公告', 1),
('TR-RISK',    '风险警示', 'TRADING', 2, 'ST/*ST等风险警示', 2),
('TR-DELIST',  '退市整理', 'TRADING', 2, '退市整理期公告', 3),
('TR-EXDIV',   '除权除息', 'TRADING', 2, '权益分派实施公告', 4);

3.3 个股公告时间线表

CREATE TABLE IF NOT EXISTS stock_announcement_timeline (
    id              BIGINT PRIMARY KEY AUTO_INCREMENT,
    stock_code      VARCHAR(16)  NOT NULL COMMENT '股票代码',
    announcement_date DATE      NOT NULL COMMENT '公告日期',
    day_offset      INT          NOT NULL COMMENT '相对基准日的偏移天数',
    announcement_count INT      DEFAULT 0 COMMENT '当日公告总数',
    regular_count   INT          DEFAULT 0 COMMENT '定期报告数量',
    major_count     INT          DEFAULT 0 COMMENT '重大事项数量',
    equity_count    INT          DEFAULT 0 COMMENT '股权变动数量',
    financing_count INT          DEFAULT 0 COMMENT '融资公告数量',
    trading_count   INT          DEFAULT 0 COMMENT '交易提示数量',
    avg_sentiment   DECIMAL(5,4) COMMENT '当日公告平均情感得分',
    has_key_event   TINYINT(1)   DEFAULT 0 COMMENT '是否包含关键事件',
    KEY idx_stock_date (stock_code, announcement_date),
    KEY idx_offset (stock_code, day_offset)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='个股公告时间线表';

四、数据采集方案

4.1 官方数据源

平台 数据格式 覆盖范围 访问方式
巨潮资讯网 HTML/PDF 全A股(沪深北) HTTP API + 网页爬取
上交所 HTML/PDF 沪市主板、科创板 官网公告栏目
深交所 HTML/PDF 深市主板、创业板 官网公告栏目
北交所 HTML/PDF 北交所上市公司 官网公告栏目

4.2 API数据源(公开接口)

接口 说明 优势
Tushare Pro (tushare.pro) 提供结构化公告数据接口 数据规范、更新及时、字段完整
AKShare (akshare.xyz) 开源金融数据接口库 免费、社区维护、覆盖面广
巨潮资讯网API cninfo官方搜索接口 数据权威、实时更新

4.3 采集规范

严格遵守以下规范:

  1. 频率控制:单IP请求间隔不低于1秒,避免对数据源服务器造成压力
  2. 遵守robots.txt:尊重网站爬虫协议,不爬取禁止访问的内容
  3. 数据用途限制:仅用于个人学习研究,不得商业化使用
  4. 版权声明:巨潮资讯网数据版权归深圳证券交易所所有
  5. 异常处理:网络异常时自动重试,最多重试3次,采用指数退避策略

五、近90日公告获取实现

5.1 全市场公告获取

以下代码使用AKShare获取全市场公告数据(需先安装:pip install akshare):

import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)


class AnnouncementFetcher:
    """A股公告数据获取器(仅供技术学习)"""

    def __init__(self):
        self.data = pd.DataFrame()

    def get_recent_announcements(self, days: int = 90) -> pd.DataFrame:
        """
        获取近N个交易日的全市场公告信息。
        使用AKShare的巨潮资讯网公告接口。

        参数:
            days: 获取最近多少天的公告(自然日)

        返回:
            包含公告信息的DataFrame

        注意:
            本方法仅用于学术研究,不构成投资建议。
        """
        end_date = datetime.now().strftime('%Y%m%d')
        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')

        logger.info(f"开始获取 {start_date} 至 {end_date} 的全市场公告...")

        try:
            # AKShare 巨潮资讯网公告接口
            df = ak.stock_info_a_code_name()
            stock_list = df[['code', 'name']].to_dict('records')

            all_announcements = []
            for idx, stock in enumerate(stock_list):
                try:
                    stock_df = self._get_stock_announcements(
                        stock['code'], start_date, end_date
                    )
                    if stock_df is not None and not stock_df.empty:
                        all_announcements.append(stock_df)
                    # 频率控制:每只股票间隔0.5秒
                    time.sleep(0.5)
                except Exception as e:
                    logger.warning(f"获取 {stock['code']} 公告失败: {e}")
                    continue

                if (idx + 1) % 100 == 0:
                    logger.info(f"已处理 {idx + 1}/{len(stock_list)} 只股票")

            if all_announcements:
                self.data = pd.concat(all_announcements, ignore_index=True)
                logger.info(f"共获取 {len(self.data)} 条公告记录")
            else:
                logger.warning("未获取到任何公告数据")

        except Exception as e:
            logger.error(f"获取全市场公告异常: {e}")

        return self.data

    def _get_stock_announcements(
        self, stock_code: str, start_date: str, end_date: str
    ) -> pd.DataFrame:
        """
        获取单只股票在指定日期范围内的公告。
        """
        try:
            df = ak.stock_notice_report(symbol=stock_code)
            if df is not None and not df.empty:
                df['stock_code'] = stock_code
                # 日期过滤
                if '公告日期' in df.columns:
                    df['公告日期'] = pd.to_datetime(df['公告日期'], errors='coerce')
                    start_dt = pd.to_datetime(start_date)
                    end_dt = pd.to_datetime(end_date)
                    df = df[
                        (df['公告日期'] >= start_dt) & (df['公告日期'] <= end_dt)
                    ]
                return df
        except Exception as e:
            logger.debug(f"获取 {stock_code} 公告详情失败: {e}")
        return pd.DataFrame()

5.2 个股公告获取

    def get_stock_announcements(
        self, stock_code: str, days: int = 90
    ) -> pd.DataFrame:
        """
        获取指定个股近N天的公告信息。

        参数:
            stock_code: 股票代码(如 '000001')
            days: 回溯天数

        返回:
            该股票公告DataFrame
        """
        end_date = datetime.now().strftime('%Y%m%d')
        start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')

        logger.info(f"获取 {stock_code} 近{days}天公告 ({start_date}~{end_date})...")

        try:
            df = ak.stock_notice_report(symbol=stock_code)
            if df is not None and not df.empty:
                if '公告日期' in df.columns:
                    df['公告日期'] = pd.to_datetime(df['公告日期'], errors='coerce')
                    start_dt = pd.to_datetime(start_date)
                    end_dt = pd.to_datetime(end_date)
                    df = df[
                        (df['公告日期'] >= start_dt) & (df['公告日期'] <= end_dt)
                    ]
                df = df.sort_values('公告日期', ascending=False).reset_index(drop=True)
                logger.info(f"获取到 {len(df)} 条公告")
            else:
                logger.info(f"未获取到 {stock_code} 的公告数据")
            return df
        except Exception as e:
            logger.error(f"获取个股公告失败: {e}")
            return pd.DataFrame()

5.3 特定类型公告获取

    def get_announcements_by_type(
        self, stock_code: str, keyword: str, days: int = 90
    ) -> pd.DataFrame:
        """
        按关键词筛选特定类型公告。

        参数:
            stock_code: 股票代码
            keyword: 公告标题关键词(如'年报'、'减持'、'重组')
            days: 回溯天数

        返回:
            筛选后的公告DataFrame
        """
        df = self.get_stock_announcements(stock_code, days)
        if df.empty:
            return df

        # 在标题列中搜索关键词
        title_col = None
        for col in df.columns:
            if '标题' in col or 'title' in col.lower():
                title_col = col
                break

        if title_col:
            filtered = df[df[title_col].str.contains(keyword, na=False)]
            logger.info(f"筛选到 {len(filtered)} 条包含'{keyword}'的公告")
            return filtered

        logger.warning("未找到标题列,无法按关键词筛选")
        return pd.DataFrame()

    def get_announcement_statistics(self, df: pd.DataFrame) -> dict:
        """
        对公告数据进行统计分析(仅供学术研究)。

        参数:
            df: 公告DataFrame

        返回:
            统计结果字典
        """
        if df.empty:
            return {}

        stats = {
            'total_count': len(df),
            'date_range': None,
            'type_distribution': {},
        }

        # 日期范围
        date_col = None
        for col in df.columns:
            if '日期' in col or 'date' in col.lower():
                date_col = col
                break
        if date_col:
            dates = pd.to_datetime(df[date_col], errors='coerce').dropna()
            if not dates.empty:
                stats['date_range'] = {
                    'start': dates.min().strftime('%Y-%m-%d'),
                    'end': dates.max().strftime('%Y-%m-%d'),
                }

        # 按公告类型统计
        type_col = None
        for col in df.columns:
            if '类型' in col or 'type' in col.lower():
                type_col = col
                break
        if type_col:
            stats['type_distribution'] = df[type_col].value_counts().to_dict()

        return stats

六、公告NLP处理

6.1 PDF转文本解析

import re
import os
import requests
import pdfplumber
from typing import Optional


class AnnouncementNLP:
    """公告文本NLP处理器(仅供技术学习)"""

    def __init__(self, download_dir: str = './announcements_pdf'):
        self.download_dir = download_dir
        os.makedirs(download_dir, exist_ok=True)

    def download_pdf(self, pdf_url: str, filename: str) -> Optional[str]:
        """
        下载公告PDF文件。

        参数:
            pdf_url: PDF文件URL
            filename: 保存的文件名

        返回:
            本地文件路径,失败返回None
        """
        filepath = os.path.join(self.download_dir, filename)
        if os.path.exists(filepath):
            return filepath

        try:
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
                              'AcademicResearchBot/1.0'
            }
            response = requests.get(pdf_url, headers=headers, timeout=30)
            response.raise_for_status()

            with open(filepath, 'wb') as f:
                f.write(response.content)
            logger.info(f"PDF已下载: {filepath}")
            return filepath
        except Exception as e:
            logger.error(f"下载PDF失败: {e}")
            return None

    def pdf_to_text(self, pdf_path: str) -> str:
        """
        将PDF公告文件转换为纯文本。

        参数:
            pdf_path: PDF文件路径

        返回:
            提取的文本内容
        """
        text_parts = []
        try:
            with pdfplumber.open(pdf_path) as pdf:
                for page in pdf.pages:
                    page_text = page.extract_text()
                    if page_text:
                        text_parts.append(page_text)
        except Exception as e:
            logger.error(f"PDF解析失败: {e}")

        full_text = '\n'.join(text_parts)
        return full_text

6.2 关键信息抽取

    def extract_financial_data(self, text: str) -> dict:
        """
        从公告文本中抽取关键财务数据。
        使用正则表达式匹配常见财务数据模式。

        参数:
            text: 公告纯文本

        返回:
            抽取的财务数据字典
        """
        result = {}

        # 营业收入(亿元)
        revenue_patterns = [
            r'营业收入[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*亿元',
            r'营业收入[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*万元',
        ]
        for pattern in revenue_patterns:
            match = re.search(pattern, text)
            if match:
                result['revenue'] = match.group(1).replace(',', '')
                break

        # 净利润
        profit_patterns = [
            r'净利润[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*亿元',
            r'净利润[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*万元',
            r'归属于.*?净利润[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*亿元',
        ]
        for pattern in profit_patterns:
            match = re.search(pattern, text)
            if match:
                result['net_profit'] = match.group(1).replace(',', '')
                break

        # 总资产
        asset_match = re.search(
            r'总资产[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*亿元', text
        )
        if asset_match:
            result['total_assets'] = asset_match.group(1).replace(',', '')

        # 金额(合同金额、交易金额等)
        amount_patterns = [
            r'(?:合同|交易|中标)金额[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*(亿元|万元|元)',
            r'(?:涉及)?金额[为约]?\s*[::]?\s*([\d,]+\.?\d*)\s*(亿元|万元|元)',
        ]
        for pattern in amount_patterns:
            match = re.search(pattern, text)
            if match:
                result['amount'] = match.group(1).replace(',', '')
                result['amount_unit'] = match.group(2)
                break

        # 日期信息
        date_patterns = [
            r'(\d{4})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日',
            r'(\d{4})[/-](\d{1,2})[/-](\d{1,2})',
        ]
        dates_found = []
        for pattern in date_patterns:
            dates_found.extend(re.findall(pattern, text))
        if dates_found:
            result['dates'] = [
                f"{d[0]}-{d[1].zfill(2)}-{d[2].zfill(2)}" for d in dates_found
            ]

        return result

    def extract_key_entities(self, text: str) -> dict:
        """
        从公告文本中抽取关键实体信息。

        参数:
            text: 公告纯文本

        返回:
            实体字典
        """
        entities = {}

        # 公司名称
        company_patterns = [
            r'(?:本公司|公司|甲方|乙方)\s*[((]([^\))]+)[))]',
            r'([^\s,。]+(?:股份有限公司|有限责任公司|集团有限公司))',
        ]
        companies = set()
        for pattern in company_patterns:
            matches = re.findall(pattern, text)
            companies.update(matches)
        if companies:
            entities['companies'] = list(companies)[:10]

        # 百分比数据
        percentages = re.findall(r'([\d]+\.?\d*)\s*%', text)
        if percentages:
            entities['percentages'] = percentages[:10]

        # 股份数量
        shares = re.findall(
            r'([\d,]+\.?\d*)\s*(万股|亿股|股)', text
        )
        if shares:
            entities['share_amounts'] = shares[:10]

        return entities

6.3 情感分析(仅供参考)

    def analyze_sentiment(self, text: str) -> dict:
        """
        对公告文本进行简单的情感分析。
        基于关键词词典方法,结果仅供参考,不构成投资建议。

        参数:
            text: 公告纯文本

        返回:
            情感分析结果字典
        """
        # 利好关键词(权重为正)
        positive_keywords = {
            '增长': 1, '增加': 1, '提升': 1, '盈利': 2, '扭亏': 2,
            '预增': 2, '超预期': 3, '回购': 1, '增持': 1, '中标': 1,
            '签约': 1, '突破': 2, '创新高': 3, '分红': 1, '派息': 1,
            '获批': 1, '获得': 1, '完成': 1, '达标': 1, '改善': 1,
        }

        # 利空关键词(权重为负)
        negative_keywords = {
            '下降': -1, '减少': -1, '亏损': -2, '预减': -2, '下滑': -1,
            '减持': -1, '质押': -1, '诉讼': -1, '处罚': -2, '警示': -1,
            '退市': -3, '终止': -2, '低于预期': -3, '违规': -2, '风险': -1,
            '下跌': -1, '跌': -1, '商誉减值': -2, '计提': -1, '停产': -2,
        }

        score = 0
        matched_keywords = []

        for keyword, weight in positive_keywords.items():
            count = len(re.findall(keyword, text))
            if count > 0:
                score += weight * count
                matched_keywords.append({'keyword': keyword, 'weight': weight, 'count': count})

        for keyword, weight in negative_keywords.items():
            count = len(re.findall(keyword, text))
            if count > 0:
                score += weight * count
                matched_keywords.append({'keyword': keyword, 'weight': weight, 'count': count})

        # 归一化到 [-1, 1]
        max_possible = sum(abs(w) * 5 for w in list(positive_keywords.values()) + list(negative_keywords.values()))
        normalized_score = round(max(-1, min(1, score / max(max_possible, 1))), 4)

        sentiment_label = 'positive' if normalized_score > 0.1 else (
            'negative' if normalized_score < -0.1 else 'neutral'
        )

        return {
            'raw_score': score,
            'normalized_score': normalized_score,
            'sentiment_label': sentiment_label,
            'matched_keywords': sorted(matched_keywords, key=lambda x: abs(x['weight']), reverse=True),
            'disclaimer': '本情感分析结果仅基于简单关键词匹配,仅供参考学习,不构成投资建议。',
        }

6.4 事件类型识别

    def classify_event_type(self, title: str, text: str = '') -> dict:
        """
        基于规则的事件类型识别(仅供学术研究)。

        参数:
            title: 公告标题
            text: 公告正文(可选)

        返回:
            事件分类结果
        """
        combined_text = f"{title} {text}"

        # 事件类型规则库
        event_rules = {
            '业绩预告': {
                'category': 'REGULAR',
                'keywords': ['业绩预告', '预计净利润', '业绩预计'],
            },
            '年度报告': {
                'category': 'REGULAR',
                'keywords': ['年度报告', '年报', '年度财务报告'],
            },
            '并购重组': {
                'category': 'MAJOR',
                'keywords': ['收购', '重组', '合并', '资产置换', '重大资产'],
            },
            '重大合同': {
                'category': 'MAJOR',
                'keywords': ['中标', '合同', '框架协议', '合作协议', '签约'],
            },
            '股东增持': {
                'category': 'EQUITY',
                'keywords': ['增持', '举牌', '一致行动人'],
            },
            '股东减持': {
                'category': 'EQUITY',
                'keywords': ['减持', '减持计划', '集中竞价减持'],
            },
            '股权激励': {
                'category': 'EQUITY',
                'keywords': ['股权激励', '限制性股票', '股票期权', '员工持股'],
            },
            '限售解禁': {
                'category': 'EQUITY',
                'keywords': ['限售股解禁', '解除限售', '上市流通'],
            },
            '定向增发': {
                'category': 'FINANCING',
                'keywords': ['非公开发行', '定向增发', '定增'],
            },
            '停牌': {
                'category': 'TRADING',
                'keywords': ['停牌', '暂停上市'],
            },
            '风险警示': {
                'category': 'TRADING',
                'keywords': ['退市风险', '*ST', 'ST', '风险警示'],
            },
        }

        matched_events = []
        for event_name, rule in event_rules.items():
            for keyword in rule['keywords']:
                if keyword in combined_text:
                    matched_events.append({
                        'event_type': event_name,
                        'category': rule['category'],
                        'matched_keyword': keyword,
                    })
                    break

        return {
            'matched_events': matched_events,
            'primary_event': matched_events[0] if matched_events else None,
            'disclaimer': '事件识别基于简单规则匹配,仅供学术研究参考。',
        }

七、学术研究应用示例

声明:以下所有示例仅供学术研究参考,不构成任何投资建议或交易策略。

7.1 信息披露效率研究框架

信息披露效率是金融学中的重要研究课题。研究者可以通过公告数据构建以下分析框架:

研究问题示例:

  • 上市公司公告披露的及时性是否存在系统性差异?
  • 不同类型公告的市场反应速度有何不同?
  • 信息披露质量与公司治理结构之间的关系

技术分析工具框架:

def disclosure_timeliness_analysis(announcements_df: pd.DataFrame) -> dict:
    """
    信息披露及时性分析框架(仅供学术研究)。

    分析维度:
    1. 公告披露时间分布(工作日 vs 周末、盘中 vs 盘后)
    2. 业绩预告与实际报告的时间间隔
    3. 不同公司规模的信息披露速度差异

    参数:
        announcements_df: 公告数据DataFrame

    返回:
        分析结果字典
    """
    results = {}

    # 分析1:公告披露时间分布
    # 检查是否有时间列
    time_col = None
    for col in announcements_df.columns:
        if '时间' in col or 'time' in col.lower():
            time_col = col
            break

    if time_col:
        announcements_df['_hour'] = pd.to_datetime(
            announcements_df[time_col], errors='coerce'
        ).dt.hour

        # 盘后披露比例(17:00之后)
        after_hours = announcements_df['_hour'].apply(
            lambda x: x >= 17 if pd.notna(x) else False
        )
        results['after_hours_ratio'] = round(after_hours.mean(), 4)

    # 分析2:公告数量按星期分布
    date_col = None
    for col in announcements_df.columns:
        if '日期' in col or 'date' in col.lower():
            date_col = col
            break

    if date_col:
        announcements_df['_weekday'] = pd.to_datetime(
            announcements_df[date_col], errors='coerce'
        ).dt.dayofweek
        weekday_dist = announcements_df['_weekday'].value_counts().sort_index().to_dict()
        results['weekday_distribution'] = {
            '周一': weekday_dist.get(0, 0),
            '周二': weekday_dist.get(1, 0),
            '周三': weekday_dist.get(2, 0),
            '周四': weekday_dist.get(3, 0),
            '周五': weekday_dist.get(4, 0),
            '周六': weekday_dist.get(5, 0),
            '周日': weekday_dist.get(6, 0),
        }

    results['disclaimer'] = '本分析仅供学术研究参考,不构成投资建议。'
    return results

7.2 事件研究方法论

事件研究法(Event Study)是金融学中衡量事件对证券价格影响的经典实证方法。以下为技术实现框架:

仅供学术研究参考的事件研究步骤:

  1. 事件定义:确定研究的事件类型(如业绩预告公告)
  2. 事件窗口:定义事件日前后各N天为事件窗口
  3. 估计窗口:选取事件窗口之前的期间用于估计正常收益率
  4. 正常收益率模型:市场模型法(CAPM)或市场调整模型
  5. 异常收益率计算:AR = 实际收益率 - 正常收益率
  6. 累计异常收益率:CAR = 事件窗口内AR的累加
  7. 统计检验:检验CAR是否显著异于零
def event_study_framework(
    stock_returns: pd.DataFrame,
    market_returns: pd.Series,
    event_dates: list,
    event_window: int = 10,
    estimation_window: int = 120,
) -> dict:
    """
    事件研究法技术框架(仅供学术研究参考)。

    参数:
        stock_returns: 个股日收益率Series(index为日期)
        market_returns: 市场日收益率Series(index为日期)
        event_dates: 事件日期列表
        event_window: 事件窗口半径(事件日前后各N天)
        estimation_window: 估计窗口长度

    返回:
        事件研究结果字典

    注意:
        本函数仅为方法论演示,不构成投资建议。
        实际学术研究需考虑更多因素(如CAPM模型、统计显著性等)。
    """
    results = {
        'event_dates': event_dates,
        'event_window': event_window,
        'estimation_window': estimation_window,
        'car_results': [],
        'disclaimer': '本事件研究框架仅供学术研究方法论学习,不构成投资建议。',
    }

    for event_date in event_dates:
        event_dt = pd.to_datetime(event_date)

        # 估计窗口
        est_start = event_dt - pd.Timedelta(days=estimation_window + event_window)
        est_end = event_dt - pd.Timedelta(days=event_window + 1)

        # 事件窗口
        evt_start = event_dt - pd.Timedelta(days=event_window)
        evt_end = event_dt + pd.Timedelta(days=event_window)

        try:
            # 提取估计窗口数据
            est_mask = (stock_returns.index >= est_start) & (stock_returns.index <= est_end)
            est_stock = stock_returns.loc[est_mask]
            est_market = market_returns.loc[est_mask]

            if len(est_stock) < 30:
                results['car_results'].append({
                    'event_date': str(event_dt.date()),
                    'status': '估计窗口数据不足',
                })
                continue

            # 市场调整模型:正常收益率 = 市场收益率
            # 提取事件窗口数据
            evt_mask = (stock_returns.index >= evt_start) & (stock_returns.index <= evt_end)
            evt_stock = stock_returns.loc[evt_mask]
            evt_market = market_returns.loc[evt_mask]

            # 计算异常收益率
            common_idx = evt_stock.index.intersection(evt_market.index)
            ar = evt_stock.loc[common_idx] - evt_market.loc[common_idx]

            # 计算累计异常收益率
            car = ar.sum()

            results['car_results'].append({
                'event_date': str(event_dt.date()),
                'CAR': round(float(car), 6),
                'AR_series': {str(d): round(float(v), 6) for d, v in ar.items()},
                'event_window_days': len(ar),
            })
        except Exception as e:
            results['car_results'].append({
                'event_date': str(event_dt.date()),
                'status': f'计算失败: {str(e)}',
            })

    return results

7.3 公告文本分析技术方案

def announcement_text_analysis(
    announcements_df: pd.DataFrame, nlp_processor: 'AnnouncementNLP'
) -> dict:
    """
    公告文本综合分析方案(仅供学术研究参考)。

    分析内容:
    1. 公告文本长度分布
    2. 情感分析分布
    3. 事件类型分布
    4. 关键词频率统计

    参数:
        announcements_df: 公告数据
        nlp_processor: NLP处理器实例

    返回:
        分析结果
    """
    analysis = {
        'text_length_stats': {},
        'sentiment_distribution': {'positive': 0, 'neutral': 0, 'negative': 0},
        'event_type_distribution': {},
        'top_keywords': [],
        'disclaimer': '本分析仅供学术研究参考,不构成投资建议。',
    }

    title_col = None
    for col in announcements_df.columns:
        if '标题' in col or 'title' in col.lower():
            title_col = col
            break

    if not title_col:
        return analysis

    # 文本长度统计
    announcements_df['_title_len'] = announcements_df[title_col].str.len()
    analysis['text_length_stats'] = {
        'mean': round(announcements_df['_title_len'].mean(), 1),
        'median': round(announcements_df['_title_len'].median(), 1),
        'max': int(announcements_df['_title_len'].max()),
        'min': int(announcements_df['_title_len'].min()),
    }

    # 情感分析
    for _, row in announcements_df.iterrows():
        sentiment = nlp_processor.analyze_sentiment(str(row[title_col]))
        label = sentiment['sentiment_label']
        analysis['sentiment_distribution'][label] += 1

    # 事件类型识别
    for _, row in announcements_df.iterrows():
        event_result = nlp_processor.classify_event_type(str(row[title_col]))
        if event_result['primary_event']:
            evt = event_result['primary_event']['event_type']
            analysis['event_type_distribution'][evt] = (
                analysis['event_type_distribution'].get(evt, 0) + 1
            )

    return analysis

八、完整Python代码实现

8.1 公告数据获取类

"""
公告数据获取类 - 完整实现
"""
import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
import time
import logging

logger = logging.getLogger(__name__)


class AnnouncementDataFetcher:
    """A股公告数据获取器"""

    def __init__(self, cache_dir: str = './cache'):
        self.cache_dir = cache_dir
        self.session = None

    def fetch_stock_announcements(
        self, stock_code: str, days: int = 90
    ) -> pd.DataFrame:
        """获取个股近N天公告"""
        try:
            df = ak.stock_notice_report(symbol=stock_code)
            if df is not None and not df.empty:
                date_col = self._find_date_column(df)
                if date_col:
                    df[date_col] = pd.to_datetime(df[date_col], errors='coerce')
                    cutoff = datetime.now() - timedelta(days=days)
                    df = df[df[date_col] >= cutoff]
                return df.sort_values(date_col or df.columns[0], ascending=False)
        except Exception as e:
            logger.error(f"获取公告失败 {stock_code}: {e}")
        return pd.DataFrame()

    def fetch_market_announcements_batch(
        self, stock_codes: list, days: int = 90, delay: float = 0.5
    ) -> pd.DataFrame:
        """批量获取多只股票公告"""
        all_data = []
        for code in stock_codes:
            df = self.fetch_stock_announcements(code, days)
            if not df.empty:
                df['stock_code'] = code
                all_data.append(df)
            time.sleep(delay)
        if all_data:
            return pd.concat(all_data, ignore_index=True)
        return pd.DataFrame()

    @staticmethod
    def _find_date_column(df: pd.DataFrame) -> str:
        for col in df.columns:
            if '日期' in col or 'date' in col.lower():
                return col
        return ''

8.2 PDF解析与NLP处理类

"""
PDF解析与NLP处理类 - 完整实现
"""
import re
import os
import pdfplumber
import requests
from typing import Optional
import logging

logger = logging.getLogger(__name__)


class PDFParserAndNLP:
    """PDF解析与NLP处理"""

    def __init__(self, pdf_dir: str = './pdf_cache'):
        self.pdf_dir = pdf_dir
        os.makedirs(pdf_dir, exist_ok=True)

    def download_and_parse(self, url: str, filename: str) -> str:
        """下载PDF并解析为文本"""
        filepath = os.path.join(self.pdf_dir, filename)
        if not os.path.exists(filepath):
            try:
                resp = requests.get(url, timeout=30, headers={
                    'User-Agent': 'AcademicResearchBot/1.0'
                })
                resp.raise_for_status()
                with open(filepath, 'wb') as f:
                    f.write(resp.content)
            except Exception as e:
                logger.error(f"下载失败: {e}")
                return ''

        return self._extract_text(filepath)

    def _extract_text(self, path: str) -> str:
        """提取PDF文本"""
        texts = []
        try:
            with pdfplumber.open(path) as pdf:
                for page in pdf.pages:
                    t = page.extract_text()
                    if t:
                        texts.append(t)
        except Exception as e:
            logger.error(f"PDF解析失败: {e}")
        return '\n'.join(texts)

    def extract_numbers(self, text: str) -> list:
        """提取文本中的数字信息"""
        patterns = [
            r'([\d,]+\.?\d*)\s*亿元',
            r'([\d,]+\.?\d*)\s*万元',
            r'([\d,]+\.?\d*)\s*(万股|亿股)',
            r'([\d]+\.?\d*)\s*%',
        ]
        results = []
        for p in patterns:
            matches = re.findall(p, text)
            results.extend(matches)
        return results

    def keyword_sentiment(self, text: str) -> dict:
        """基于关键词的情感分析(仅供参考)"""
        positive = ['增长', '盈利', '预增', '中标', '增持', '回购', '分红']
        negative = ['下降', '亏损', '预减', '减持', '诉讼', '处罚', '退市']

        pos_score = sum(text.count(k) for k in positive)
        neg_score = sum(text.count(k) for k in negative)
        total = pos_score + neg_score

        if total == 0:
            return {'label': 'neutral', 'score': 0.0}
        score = (pos_score - neg_score) / total
        label = 'positive' if score > 0.2 else ('negative' if score < -0.2 else 'neutral')
        return {'label': label, 'score': round(score, 4)}

8.3 事件识别类

"""
事件识别类 - 完整实现
"""
import re
import logging

logger = logging.getLogger(__name__)


class EventClassifier:
    """公告事件类型识别器"""

    RULES = {
        '业绩预告': ['业绩预告', '预计净利润', '业绩预计'],
        '年度报告': ['年度报告', '年报', '年度财务'],
        '并购重组': ['收购', '重组', '合并', '资产置换'],
        '重大合同': ['中标', '重大合同', '框架协议'],
        '股东增持': ['增持', '举牌', '一致行动人'],
        '股东减持': ['减持', '减持计划', '集中竞价减持'],
        '股权激励': ['股权激励', '限制性股票', '股票期权'],
        '限售解禁': ['限售股解禁', '解除限售', '上市流通'],
        '定向增发': ['非公开发行', '定向增发', '定增'],
        '停牌复牌': ['停牌', '复牌', '暂停上市'],
        '风险警示': ['退市风险', '*ST', 'ST', '风险警示'],
        '除权除息': ['除权', '除息', '权益分派', '分红派息'],
    }

    CATEGORY_MAP = {
        '业绩预告': 'REGULAR', '年度报告': 'REGULAR',
        '并购重组': 'MAJOR', '重大合同': 'MAJOR',
        '股东增持': 'EQUITY', '股东减持': 'EQUITY',
        '股权激励': 'EQUITY', '限售解禁': 'EQUITY',
        '定向增发': 'FINANCING',
        '停牌复牌': 'TRADING', '风险警示': 'TRADING',
        '除权除息': 'TRADING',
    }

    def classify(self, title: str, content: str = '') -> dict:
        """对公告进行事件分类"""
        text = f"{title} {content}"
        matched = []

        for event_type, keywords in self.RULES.items():
            for kw in keywords:
                if kw in text:
                    matched.append({
                        'event_type': event_type,
                        'category': self.CATEGORY_MAP.get(event_type, 'OTHER'),
                        'keyword': kw,
                    })
                    break

        return {
            'matched_events': matched,
            'primary': matched[0] if matched else None,
        }

8.4 数据存储类

"""
数据存储类 - 完整实现
"""
import sqlite3
import json
import pandas as pd
import logging
from datetime import datetime

logger = logging.getLogger(__name__)


class AnnouncementStorage:
    """公告数据SQLite存储"""

    def __init__(self, db_path: str = './announcements.db'):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        """初始化数据库表"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute('''
                CREATE TABLE IF NOT EXISTS announcements (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    announcement_id TEXT UNIQUE,
                    stock_code TEXT NOT NULL,
                    stock_name TEXT,
                    title TEXT NOT NULL,
                    announcement_type TEXT,
                    announcement_date TEXT,
                    content TEXT,
                    pdf_url TEXT,
                    sentiment_score REAL,
                    event_type TEXT,
                    key_entities TEXT,
                    created_at TEXT DEFAULT CURRENT_TIMESTAMP
                )
            ''')
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_stock
                ON announcements(stock_code)
            ''')
            conn.execute('''
                CREATE INDEX IF NOT EXISTS idx_date
                ON announcements(announcement_date)
            ''')
            conn.commit()

    def save_announcement(self, record: dict) -> bool:
        """保存单条公告"""
        try:
            with sqlite3.connect(self.db_path) as conn:
                conn.execute('''
                    INSERT OR REPLACE INTO announcements
                    (announcement_id, stock_code, stock_name, title,
                     announcement_type, announcement_date, content,
                     pdf_url, sentiment_score, event_type, key_entities)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ''', (
                    record.get('announcement_id', ''),
                    record.get('stock_code', ''),
                    record.get('stock_name', ''),
                    record.get('title', ''),
                    record.get('announcement_type', ''),
                    record.get('announcement_date', ''),
                    record.get('content', ''),
                    record.get('pdf_url', ''),
                    record.get('sentiment_score'),
                    record.get('event_type'),
                    json.dumps(record.get('key_entities', {}), ensure_ascii=False),
                ))
                conn.commit()
            return True
        except Exception as e:
            logger.error(f"保存失败: {e}")
            return False

    def save_batch(self, records: list) -> int:
        """批量保存公告"""
        count = 0
        for r in records:
            if self.save_announcement(r):
                count += 1
        logger.info(f"批量保存完成: {count}/{len(records)}")
        return count

    def query_by_stock(self, stock_code: str, days: int = 90) -> pd.DataFrame:
        """查询个股公告"""
        with sqlite3.connect(self.db_path) as conn:
            df = pd.read_sql_query('''
                SELECT * FROM announcements
                WHERE stock_code = ?
                AND announcement_date >= date('now', '-' || ? || ' days')
                ORDER BY announcement_date DESC
            ''', conn, params=(stock_code, days))
        return df

    def query_by_type(
        self, announcement_type: str, days: int = 90
    ) -> pd.DataFrame:
        """按类型查询公告"""
        with sqlite3.connect(self.db_path) as conn:
            df = pd.read_sql_query('''
                SELECT * FROM announcements
                WHERE announcement_type = ?
                AND announcement_date >= date('now', '-' || ? || ' days')
                ORDER BY announcement_date DESC
            ''', conn, params=(announcement_type, days))
        return df

8.5 可视化类

"""
可视化类 - 完整实现
"""
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg')  # 非交互式后端
matplotlib.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
matplotlib.rcParams['axes.unicode_minus'] = False
import logging

logger = logging.getLogger(__name__)


class AnnouncementVisualizer:
    """公告数据可视化工具(仅供学术研究)"""

    def __init__(self, output_dir: str = './charts'):
        self.output_dir = output_dir
        import os
        os.makedirs(output_dir, exist_ok=True)

    def plot_announcement_timeline(
        self, df: pd.DataFrame, stock_code: str, save: bool = True
    ):
        """
        绘制个股公告时间线图(仅供学术研究)。

        参数:
            df: 公告数据(需包含日期列和类型列)
            stock_code: 股票代码
            save: 是否保存图片
        """
        date_col = self._find_col(df, ['日期', 'date'])
        type_col = self._find_col(df, ['类型', 'type'])

        if not date_col:
            logger.warning("未找到日期列")
            return

        fig, ax = plt.subplots(figsize=(14, 6))

        if type_col:
            # 按类型分组统计每日公告数量
            daily = df.groupby([date_col, type_col]).size().unstack(fill_value=0)
            daily.plot(kind='bar', stacked=True, ax=ax, width=0.8)
            ax.legend(title='公告类型', bbox_to_anchor=(1.02, 1), loc='upper left')
        else:
            daily = df.groupby(date_col).size()
            ax.bar(range(len(daily)), daily.values)
            ax.set_xticks(range(len(daily)))
            ax.set_xticklabels(
                [str(d)[:10] for d in daily.index], rotation=45, ha='right'
            )

        ax.set_title(f'{stock_code} 公告时间线(仅供学术研究)', fontsize=14)
        ax.set_xlabel('日期')
        ax.set_ylabel('公告数量')

        plt.tight_layout()
        if save:
            filepath = f'{self.output_dir}/{stock_code}_timeline.png'
            plt.savefig(filepath, dpi=150, bbox_inches='tight')
            logger.info(f"图表已保存: {filepath}")
        plt.close()

    def plot_type_distribution(
        self, df: pd.DataFrame, save: bool = True
    ):
        """绘制公告类型分布饼图(仅供学术研究)"""
        type_col = self._find_col(df, ['类型', 'type'])
        if not type_col:
            logger.warning("未找到类型列")
            return

        counts = df[type_col].value_counts()

        fig, ax = plt.subplots(figsize=(10, 8))
        ax.pie(
            counts.values, labels=counts.index, autopct='%1.1f%%',
            startangle=90, textprops={'fontsize': 10}
        )
        ax.set_title('公告类型分布(仅供学术研究)', fontsize=14)

        plt.tight_layout()
        if save:
            filepath = f'{self.output_dir}/type_distribution.png'
            plt.savefig(filepath, dpi=150, bbox_inches='tight')
            logger.info(f"图表已保存: {filepath}")
        plt.close()

    def plot_sentiment_distribution(
        self, df: pd.DataFrame, save: bool = True
    ):
        """绘制情感分析分布图(仅供学术研究,不构成投资建议)"""
        if 'sentiment_score' not in df.columns:
            logger.warning("数据中不包含情感得分列")
            return

        fig, ax = plt.subplots(figsize=(10, 6))
        ax.hist(df['sentiment_score'].dropna(), bins=50, edgecolor='black', alpha=0.7)
        ax.axvline(x=0, color='red', linestyle='--', label='中性线')
        ax.set_title('公告情感得分分布(仅供参考,不构成投资建议)', fontsize=14)
        ax.set_xlabel('情感得分')
        ax.set_ylabel('频次')
        ax.legend()

        plt.tight_layout()
        if save:
            filepath = f'{self.output_dir}/sentiment_distribution.png'
            plt.savefig(filepath, dpi=150, bbox_inches='tight')
            logger.info(f"图表已保存: {filepath}")
        plt.close()

    @staticmethod
    def _find_col(df: pd.DataFrame, keywords: list) -> str:
        for col in df.columns:
            for kw in keywords:
                if kw in col.lower():
                    return col
        return ''

九、合规与法律注意事项

9.1 数据版权

巨潮资讯网(cninfo.com.cn)及其披露的所有公告数据,版权归深圳证券交易所所有。使用公告数据时应当遵守以下规定:

  • 不得将数据用于商业用途
  • 不得对数据进行转售或再分发
  • 引用数据时应当注明数据来源
  • 遵守巨潮资讯网的使用条款和robots.txt协议

9.2 信息使用限制

根据《中华人民共和国证券法》及相关法规:

  1. 不得利用内幕信息交易:在公告正式披露前获取的信息属于内幕信息,利用内幕信息进行交易构成内幕交易罪
  2. 不得编造虚假信息:不得基于公告数据进行断章取义、歪曲事实的传播
  3. 不得操纵市场:不得利用公告信息进行市场操纵行为

9.3 证券投资咨询资质要求

根据《证券期货投资咨询管理暂行办法》,从事证券投资咨询业务需要取得中国证监会颁发的投资咨询执业资格。本模块的开发者和使用者均不具备证券投资咨询资质,因此不得将本模块的分析结果作为投资建议提供给他人。

9.4 技术使用伦理

  • 数据采集应遵守目标网站的robots.txt协议和服务条款
  • 请求频率应合理控制,避免对数据源服务器造成过大压力
  • 所有数据处理应在本地完成,不得未经授权传输至境外服务器
  • 代码和文档中应明确标注免责声明和合规要求

十、免责声明与风险提示

10.1 免责声明

本文档及配套代码仅供技术学习和学术研究参考之用,不构成任何形式的投资建议、投资咨询或证券交易指导。

本文档中包含的所有分析框架、代码实现、技术方案均为教学演示目的,其输出结果不应对任何实际投资决策产生影响。文档作者不对任何人因使用本文档内容而导致的任何直接或间接损失承担责任。

10.2 风险提示

  1. 股市有风险,投资需谨慎。A股市场波动较大,投资者应充分认识投资风险。
  2. 历史数据不代表未来表现。基于历史公告数据的任何分析均不能预测未来股价走势。
  3. 技术分析的局限性。本文档中的NLP分析、情感分析、事件识别等技术均为简化模型,存在大量误判和遗漏的可能。
  4. 数据准确性。公开数据可能存在延迟、错误或不完整的情况,使用者应自行验证数据准确性。
  5. 法规变化。证券市场监管法规可能随时变化,使用者应关注最新法规动态。

10.3 使用承诺

使用本文档及配套代码时,使用者承诺:

  • 仅将本文档用于合法合规的技术学习和学术研究
  • 不将分析结果用于任何实际投资决策或提供给他人作为投资建议
  • 遵守所有适用的法律法规,包括但不限于《证券法》《刑法》中关于证券犯罪的规定
  • 尊重数据版权,遵守数据源的使用条款
  • 在传播或分享本文档时,保留完整的免责声明

再次声明:本文档全部内容仅供技术学习和学术研究参考,不构成任何投资建议。股市有风险,投资需谨慎。

这是一套完整的合集,上方的可以单独使用可以组合使用,使用方法,复制此行上方的完整信息,让trae帮你在本地生成一套完整的Skill,为了防止模型迷路我将完整的链接贴在下方。