沪深董秘互动平台数据模块（技术学习版）
免责声明：本文档仅供技术学习和学术研究参考，不构成任何投资建议、投资咨询或证券交易指导。股市有风险，投资需谨慎。任何基于本文档内容进行的投资决策，均由投资者自行承担全部风险和责任。
一、模块概述
1.1 模块定位
本模块是一个技术分析工具框架，用于获取近90个交易日沪深交易所董秘互动平台的公开问答数据。模块以 Python 为核心语言，结合 AKShare、Tushare 等开源金融数据接口，实现对上市公司董秘互动数据的结构化采集、清洗、存储与基础 NLP 分析。
本模块不提供任何个股推荐、买卖建议或收益承诺，所有分析功能均以"技术分析工具框架"的形式呈现，仅供学习者和研究者在合规范围内使用。
1.2 董秘互动的重要性
董秘（董事会秘书）是上市公司与投资者之间的重要沟通桥梁。董秘互动平台是投资者与上市公司进行官方公开沟通的法定渠道，其问答内容具有以下特征：
- 公开透明：所有问答均在交易所指定平台公开发布，任何投资者均可查阅。
- 时效性强：上市公司通常需在合理时间内对投资者提问进行回复，反映公司最新动态。
- 信息密度高：问答内容涵盖业务进展、财务数据、战略规划、行业判断等多个维度。
- 研究价值显著：问答数据可用于信息披露质量评估、投资者行为分析、NLP 技术验证等学术研究方向。
1.3 数据来源
| 交易所 | 平台名称 | 网址 |
|---|---|---|
| 上海证券交易所 | e互动 | https://sns.sseinfo.com |
| 深圳证券交易所 | 互动易 | https://irm.cninfo.com.cn |
| 北京证券交易所 | 互动平台 | https://www.bse.cn |
1.4 应用场景（仅供学术研究）
本模块采集的数据可应用于以下纯学术研究方向：
- 信息披露研究：分析上市公司回复率、回复时效性、回复质量等指标。
- NLP 技术应用：文本分类、情感分析、命名实体识别、问答匹配等技术的实践。
- 投资者行为研究：研究投资者关注热点、提问行为模式等。
- 金融科技教学：作为量化金融与自然语言处理交叉领域的学习案例。
明确声明：本模块及所有相关分析功能不用于实际交易、不构成投资建议、不提供任何形式的投资咨询服务。
二、董秘互动平台介绍
2.1 上交所e互动
上交所e互动（sns.sseinfo.com）是上海证券交易所指定的上市公司与投资者互动平台。主要特点：
- 覆盖范围：所有在上交所上市的主板、科创板公司。
- 提问方式：投资者需注册后提交问题，可指定某家上市公司。
- 回复机制：上市公司董秘或证券事务代表在收到问题后进行审核并公开回复。
- 数据格式：公开页面展示提问时间、提问内容、回复时间、回复内容、提问者类型（个人/机构）等字段。
2.2 深交所互动易
深交所互动易（irm.cninfo.com.cn）是深圳证券交易所指定的投资者互动平台。主要特点：
- 覆盖范围：所有在深交所上市的主板、中小板、创业板公司。
- 功能特色：除问答外，还提供投资者关系活动记录、业绩说明会回放等。
- 数据丰富度：问答记录通常包含更多结构化标签，如问答分类、是否涉及重大事项等。
- 历史数据：平台保留较长时间的历史问答记录，便于纵向研究。
2.3 北交所互动平台
北交所互动平台（bse.cn）是北京证券交易所的投资者互动渠道。主要特点：
- 覆盖范围：所有在北交所上市的中小企业。
- 发展阶段：平台功能持续完善中，数据结构与沪深所有差异。
- 研究价值：中小企业信息披露行为研究的重要数据来源。
2.4 问答类型
董秘互动平台的问答内容通常可按主题分为以下类别：
| 类型 | 典型内容 | 示例 |
|---|---|---|
| 业务咨询 | 产品、技术、合作、市场 | “贵公司新产品何时上市？” |
| 财务数据 | 营收、利润、现金流 | “请问公司一季度营收情况如何？” |
| 重大事项 | 并购重组、定增、分红 | “公司是否有并购计划？” |
| 股价波动 | 异动原因、维稳措施 | “近期股价下跌原因是什么？” |
| 股东人数 | 股东户数变化 | “最新股东人数是多少？” |
| 订单产能 | 产能利用率、订单情况 | “目前产能利用率如何？” |
三、数据结构设计
3.1 问答主表（完整Schema）
CREATE TABLE secretary_qa (
id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键ID',
exchange VARCHAR(10) NOT NULL COMMENT '交易所: SH/SZ/BJ',
stock_code VARCHAR(10) NOT NULL COMMENT '证券代码',
stock_name VARCHAR(50) COMMENT '证券简称',
question_id VARCHAR(64) COMMENT '平台问答ID',
question_date DATE COMMENT '提问日期',
question_time DATETIME COMMENT '提问时间',
questioner_type VARCHAR(20) COMMENT '提问者类型: 个人/机构',
question_title VARCHAR(200) COMMENT '提问标题',
question_text TEXT COMMENT '提问内容',
answer_date DATE COMMENT '回复日期',
answer_time DATETIME COMMENT '回复时间',
answer_text TEXT COMMENT '回复内容',
answer_status VARCHAR(20) COMMENT '回复状态: 已回复/未回复/已拒绝',
qa_category VARCHAR(30) COMMENT '问答分类',
reply_delay_hours DECIMAL(8,2) COMMENT '回复延迟(小时)',
question_length INT COMMENT '提问字数',
answer_length INT COMMENT '回复字数',
is_key_qa TINYINT DEFAULT 0 COMMENT '是否关键问答',
sentiment_score DECIMAL(5,4) COMMENT '情感得分(-1到1)',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '入库时间',
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
INDEX idx_stock_code (stock_code),
INDEX idx_question_date (question_date),
INDEX idx_exchange_stock (exchange, stock_code),
INDEX idx_answer_status (answer_status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='董秘互动问答主表';
3.2 关键信息提取表
CREATE TABLE qa_key_info (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
qa_id BIGINT NOT NULL COMMENT '关联问答主表ID',
stock_code VARCHAR(10) NOT NULL,
info_type VARCHAR(30) NOT NULL COMMENT '信息类型: 营收/利润/产能/订单/研发/合作',
info_value VARCHAR(200) COMMENT '提取的具体数值或描述',
info_period VARCHAR(20) COMMENT '信息所属期间',
confidence DECIMAL(3,2) COMMENT '提取置信度(0-1)',
source_text VARCHAR(500) COMMENT '原文片段',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
INDEX idx_qa_id (qa_id),
INDEX idx_stock_info (stock_code, info_type)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='问答关键信息提取表';
3.3 个股问答时间线表
CREATE TABLE qa_timeline (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
stock_code VARCHAR(10) NOT NULL,
event_date DATE NOT NULL,
event_type VARCHAR(30) COMMENT '事件类型: 关键回复/热点提问/公司回应',
event_summary VARCHAR(500) COMMENT '事件摘要',
qa_count INT COMMENT '当日问答数量',
key_qa_count INT COMMENT '当日关键问答数量',
avg_sentiment DECIMAL(5,4) COMMENT '当日平均情感得分',
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
INDEX idx_stock_date (stock_code, event_date)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='个股问答时间线表';
四、数据采集方案
4.1 官方平台（e互动/互动易）
直接访问交易所官方平台是获取数据的最权威方式，但需严格遵守以下规范：
- 遵守 robots.txt：在采集前检查目标网站的 robots.txt 文件。
- 控制请求频率：设置合理的请求间隔（建议不低于 2 秒/次），避免对服务器造成压力。
- 使用官方 User-Agent：标明采集身份，便于平台方识别。
- 仅采集公开数据：不尝试获取需要登录或付费才能查看的内容。
- 数据使用限制：采集的数据仅用于个人学习研究，不得商业化分发。
4.2 API数据源（AKShare/Tushare）
推荐使用开源金融数据接口库获取数据，这些库已经封装了合规的数据访问方式：
| 数据源 | 特点 | 安装方式 |
|---|---|---|
| AKShare | 开源免费、数据覆盖广、更新及时 | pip install akshare |
| Tushare | 接口丰富、社区活跃、部分需积分 | pip install tushare |
4.3 采集规范（遵守网站规则）
采集行为规范清单：
[1] 遵守《中华人民共和国网络安全法》相关规定
[2] 遵守目标网站的 robots.txt 协议
[3] 请求频率不低于 2 秒/次
[4] 不使用多线程/分布式高并发采集
[5] 不绕过反爬机制（如频繁更换IP、伪造Cookie等）
[6] 采集数据仅用于学习研究，不用于商业用途
[7] 不采集、存储用户个人隐私信息
[8] 尊重数据版权，引用时注明数据来源
五、近90日问答数据获取实现
5.1 全市场问答获取
以下代码通过 AKShare 获取全市场董秘互动数据：
import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
def get_full_market_qa(days: int = 90) -> pd.DataFrame:
"""
获取全市场近N日的董秘互动问答数据。
参数:
days: 获取最近多少天的数据，默认90天
返回:
DataFrame 包含问答数据
注意:
- 本函数仅供技术学习与学术研究使用
- 不构成任何投资建议
"""
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
all_data = []
try:
# 通过AKShare获取深交所互动易数据
df_sz = ak.stock_irm_cninfo(
start_date=start_date,
end_date=end_date
)
if df_sz is not None and not df_sz.empty:
df_sz["exchange"] = "SZ"
all_data.append(df_sz)
print(f"[INFO] 深交所互动易获取完成，共 {len(df_sz)} 条记录")
except Exception as e:
print(f"[WARN] 深交所数据获取异常: {e}")
try:
# 通过AKShare获取上交所e互动数据
df_sh = ak.stock_irm_sseinfo(
start_date=start_date,
end_date=end_date
)
if df_sh is not None and not df_sh.empty:
df_sh["exchange"] = "SH"
all_data.append(df_sh)
print(f"[INFO] 上交所e互动获取完成，共 {len(df_sh)} 条记录")
except Exception as e:
print(f"[WARN] 上交所数据获取异常: {e}")
if not all_data:
print("[WARN] 未获取到任何数据，请检查网络连接或API状态")
return pd.DataFrame()
result = pd.concat(all_data, ignore_index=True)
print(f"[INFO] 全市场数据汇总完成，共 {len(result)} 条记录")
return result
5.2 个股问答获取
def get_stock_qa(stock_code: str, days: int = 90) -> pd.DataFrame:
"""
获取指定个股近N日的董秘互动问答数据。
参数:
stock_code: 证券代码，如 "600519"（贵州茅台）
days: 获取最近多少天的数据
返回:
DataFrame 包含该股票的问答数据
注意:
- 本函数仅供技术学习与学术研究使用
- 不构成任何投资建议或个股推荐
"""
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=days)).strftime("%Y%m%d")
all_data = []
try:
# 深交所个股问答
if stock_code.startswith(("0", "3")):
df = ak.stock_irm_cninfo(
stock_code=stock_code,
start_date=start_date,
end_date=end_date
)
if df is not None and not df.empty:
df["exchange"] = "SZ"
all_data.append(df)
except Exception as e:
print(f"[WARN] 深交所个股数据获取异常: {e}")
try:
# 上交所个股问答
if stock_code.startswith("6"):
df = ak.stock_irm_sseinfo(
stock_code=stock_code,
start_date=start_date,
end_date=end_date
)
if df is not None and not df.empty:
df["exchange"] = "SH"
all_data.append(df)
except Exception as e:
print(f"[WARN] 上交所个股数据获取异常: {e}")
if not all_data:
print(f"[WARN] 未获取到 {stock_code} 的问答数据")
return pd.DataFrame()
result = pd.concat(all_data, ignore_index=True)
print(f"[INFO] {stock_code} 问答数据获取完成，共 {len(result)} 条记录")
return result
5.3 关键问答筛选
import re
def filter_key_qa(df: pd.DataFrame) -> pd.DataFrame:
"""
从问答数据中筛选出可能包含关键信息的问答。
筛选规则:
1. 回复内容较长（公司详细回复通常信息量更大）
2. 涉及财务数据、产能、订单等关键词
3. 回复延迟较短（公司重视的问题通常回复较快）
参数:
df: 问答数据 DataFrame
返回:
筛选后的关键问答 DataFrame
注意:
- 筛选结果仅供学术研究参考
- 不构成任何投资判断依据
"""
if df.empty:
return df
# 关键词列表（用于学术研究中的文本分类）
key_keywords = [
"营收", "收入", "利润", "净利润", "毛利率", "净利率",
"产能", "产量", "订单", "出货量", "销量",
"研发", "专利", "技术", "新产品",
"并购", "收购", "重组", "定增", "分红",
"合同", "中标", "签约", "合作",
"股东人数", "机构调研", "减持", "增持"
]
# 计算回复长度（如果存在回复内容列）
answer_col = None
for col in ["answer", "answer_text", "回复内容", "回答内容"]:
if col in df.columns:
answer_col = col
break
question_col = None
for col in ["question", "question_text", "提问内容"]:
if col in df.columns:
question_col = col
break
mask = pd.Series([True] * len(df), index=df.index)
# 规则1：回复内容较长（超过100字视为详细回复）
if answer_col:
df["_answer_len"] = df[answer_col].astype(str).str.len()
mask = mask & (df["_answer_len"] >= 100)
df.drop(columns=["_answer_len"], inplace=True)
# 规则2：包含关键词
text_col = question_col if question_col else answer_col
if text_col:
keyword_mask = df[text_col].astype(str).apply(
lambda x: any(kw in str(x) for kw in key_keywords)
)
mask = mask | keyword_mask
result = df[mask].copy()
print(f"[INFO] 关键问答筛选完成: {len(result)}/{len(df)} 条")
return result
5.4 热门问答追踪
def track_hot_qa(df: pd.DataFrame, top_n: int = 20) -> pd.DataFrame:
"""
追踪近期热门问答（按回复延迟短+回复长度长排序）。
参数:
df: 问答数据 DataFrame
top_n: 返回前N条热门问答
返回:
热门问答 DataFrame
注意:
- 热门标准仅为技术学习中的排序演示
- 不代表任何投资价值判断
"""
if df.empty:
return df
result = df.copy()
# 计算回复长度得分
answer_col = None
for col in ["answer", "answer_text", "回复内容"]:
if col in result.columns:
answer_col = col
break
if answer_col:
result["_length_score"] = result[answer_col].astype(str).str.len()
else:
result["_length_score"] = 0
# 按回复长度降序排列（简化版热门判定）
result = result.sort_values("_length_score", ascending=False)
result = result.drop(columns=["_length_score"]).head(top_n)
print(f"[INFO] 热门问答追踪完成，返回前 {top_n} 条")
return result
六、NLP处理与信息提取
6.1 问答文本清洗
import re
class TextCleaner:
"""问答文本清洗工具类，用于学术研究中的文本预处理。"""
@staticmethod
def clean_text(text: str) -> str:
"""
清洗问答文本。
处理步骤:
1. 去除HTML标签
2. 去除特殊字符和多余空白
3. 统一标点符号
4. 去除无意义的固定回复模板
参数:
text: 原始文本
返回:
清洗后的文本
"""
if not text or not isinstance(text, str):
return ""
# 去除HTML标签
text = re.sub(r'<[^>]+>', '', text)
# 去除URL
text = re.sub(r'http[s]?://\S+', '', text)
# 统一全角/半角标点
text = text.replace('，', ',').replace('。', '.').replace('：', ':')
text = text.replace('；', ';').replace('（', '(').replace('）', ')')
# 去除多余空白
text = re.sub(r'\s+', ' ', text).strip()
# 去除常见无意义模板回复
template_patterns = [
r'感谢您对公司的关注[。，！]*',
r'您好[，,]*感谢您的提问[。，！]*',
r'您的建议已收到[。，！]*',
]
for pattern in template_patterns:
text = re.sub(pattern, '', text)
return text.strip()
@staticmethod
def batch_clean(df: pd.DataFrame, text_columns: list) -> pd.DataFrame:
"""
批量清洗DataFrame中的文本列。
参数:
df: 原始数据
text_columns: 需要清洗的文本列名列表
返回:
清洗后的数据
"""
result = df.copy()
for col in text_columns:
if col in result.columns:
cleaned_col = f"{col}_cleaned"
result[cleaned_col] = result[col].apply(
lambda x: TextCleaner.clean_text(str(x))
)
return result
6.2 关键信息提取（财务/业务/时间/金额）
class InfoExtractor:
"""
关键信息提取工具类。
从问答文本中提取结构化信息，仅供学术研究使用。
"""
# 财务指标正则模式
FINANCIAL_PATTERNS = {
"营收": r'(?:营收|营业收入|销售收入)[约为]?[\s]*(\d+[\.\d]*)\s*(?:亿|万元|元)',
"净利润": r'(?:净利润|归母净利润)[约为]?[\s]*(\d+[\.\d]*)\s*(?:亿|万元|元)',
"毛利率": r'(?:毛利率)[约为]?[\s]*(\d+[\.\d]*)\s*(?:%|个百分点)',
"订单": r'(?:订单|合同金额|中标金额)[约为]?[\s]*(\d+[\.\d]*)\s*(?:亿|万元|元)',
"产能": r'(?:产能|产量|年产能)[约为]?[\s]*(\d+[\.\d]*)\s*(?:万|亿|吨|台|套|GWh|MWh)',
}
# 时间表达正则模式
TIME_PATTERNS = [
r'(\d{4})\s*年\s*(\d{1,2})\s*月',
r'(?:今年|本年度|本报告期)',
r'(?:上半年|下半年|一季度|二季度|三季度|四季度)',
r'(?:近期|目前|当前|截至目前)',
]
@classmethod
def extract_financial_info(cls, text: str) -> list:
"""
从文本中提取财务相关信息。
参数:
text: 清洗后的文本
返回:
提取结果列表，每项包含类型、数值、单位
"""
results = []
for info_type, pattern in cls.FINANCIAL_PATTERNS.items():
matches = re.finditer(pattern, text)
for match in matches:
results.append({
"info_type": info_type,
"raw_match": match.group(0),
"value": match.group(1) if match.lastindex else "",
"position": match.start(),
})
return results
@classmethod
def extract_time_info(cls, text: str) -> list:
"""
从文本中提取时间相关信息。
参数:
text: 清洗后的文本
返回:
时间信息列表
"""
results = []
for pattern in cls.TIME_PATTERNS:
matches = re.finditer(pattern, text)
for match in matches:
results.append({
"raw_match": match.group(0),
"position": match.start(),
})
return results
@classmethod
def extract_all(cls, text: str) -> dict:
"""
综合提取文本中的所有关键信息。
参数:
text: 清洗后的文本
返回:
包含所有提取结果的字典
"""
return {
"financial": cls.extract_financial_info(text),
"time": cls.extract_time_info(text),
"text_length": len(text),
}
6.3 情感分析（仅供参考）
class SentimentAnalyzer:
"""
基于词典的情感分析工具类。
重要声明:
- 本情感分析仅供学术研究参考
- 分析结果不代表任何投资价值判断
- 不建议将情感得分用于任何交易决策
"""
# 正面情感词典（示例，实际研究中应使用专业金融情感词典）
POSITIVE_WORDS = {
"增长", "提升", "改善", "突破", "创新", "领先", "优势",
"超预期", "向好", "稳步", "强劲", "大幅增长", "创历史新高",
"盈利", "回报", "分红", "回购", "增持", "战略合作",
"获批", "中标", "签约", "量产", "交付", "满产",
}
# 负面情感词典（示例）
NEGATIVE_WORDS = {
"下降", "下滑", "亏损", "减少", "萎缩", "放缓", "疲软",
"不及预期", "低于预期", "承压", "挑战", "风险", "不确定",
"减持", "终止", "失败", "违规", "处罚", "诉讼",
"延期", "推迟", "取消", "下跌", "回落",
}
# 否定词
NEGATION_WORDS = {"不", "未", "没有", "无", "并非", "并非没有"}
@classmethod
def analyze(cls, text: str) -> dict:
"""
对文本进行情感分析。
参数:
text: 待分析文本
返回:
包含情感得分和分类的字典
"""
if not text:
return {"score": 0.0, "label": "中性", "positive_count": 0, "negative_count": 0}
pos_count = 0
neg_count = 0
for word in cls.POSITIVE_WORDS:
if word in text:
pos_count += text.count(word)
for word in cls.NEGATIVE_WORDS:
if word in text:
neg_count += text.count(word)
total = pos_count + neg_count
if total == 0:
score = 0.0
label = "中性"
else:
score = round((pos_count - neg_count) / total, 4)
if score > 0.2:
label = "正面"
elif score < -0.2:
label = "负面"
else:
label = "中性"
return {
"score": score,
"label": label,
"positive_count": pos_count,
"negative_count": neg_count,
}
@classmethod
def batch_analyze(cls, df: pd.DataFrame, text_column: str) -> pd.DataFrame:
"""
批量对DataFrame中的文本列进行情感分析。
参数:
df: 原始数据
text_column: 文本列名
返回:
添加了情感分析结果的新DataFrame
"""
result = df.copy()
sentiments = result[text_column].apply(cls.analyze)
result["sentiment_score"] = sentiments.apply(lambda x: x["score"])
result["sentiment_label"] = sentiments.apply(lambda x: x["label"])
result["positive_count"] = sentiments.apply(lambda x: x["positive_count"])
result["negative_count"] = sentiments.apply(lambda x: x["negative_count"])
return result
6.4 问答质量评估
class QAQualityEvaluator:
"""
问答质量评估工具类。
用于学术研究中评估上市公司信息披露质量。
评估维度:
- 回复率: 已回复问题占比
- 回复时效性: 平均回复延迟
- 回复详细度: 回复内容长度
- 回复相关性: 回复是否切题（基于简单文本匹配）
"""
@staticmethod
def evaluate_reply_rate(df: pd.DataFrame) -> dict:
"""评估回复率。"""
status_col = None
for col in ["answer_status", "回复状态"]:
if col in df.columns:
status_col = col
break
if status_col is None:
return {"reply_rate": None, "message": "缺少回复状态列"}
total = len(df)
replied = len(df[df[status_col].str.contains("已回复", na=False)])
rate = round(replied / total * 100, 2) if total > 0 else 0
return {
"total_questions": total,
"replied_count": replied,
"reply_rate": rate,
}
@staticmethod
def evaluate_reply_timeliness(df: pd.DataFrame) -> dict:
"""评估回复时效性。"""
delay_col = None
for col in ["reply_delay_hours", "回复延迟"]:
if col in df.columns:
delay_col = col
break
if delay_col is None:
return {"avg_delay_hours": None, "message": "缺少回复延迟列"}
valid_delays = pd.to_numeric(df[delay_col], errors="coerce").dropna()
if valid_delays.empty:
return {"avg_delay_hours": None}
return {
"avg_delay_hours": round(valid_delays.mean(), 2),
"median_delay_hours": round(valid_delays.median(), 2),
"max_delay_hours": round(valid_delays.max(), 2),
"min_delay_hours": round(valid_delays.min(), 2),
}
@staticmethod
def evaluate_reply_detail(df: pd.DataFrame) -> dict:
"""评估回复详细度。"""
answer_col = None
for col in ["answer", "answer_text", "回复内容"]:
if col in df.columns:
answer_col = col
break
if answer_col is None:
return {"avg_length": None, "message": "缺少回复内容列"}
lengths = df[answer_col].astype(str).str.len()
return {
"avg_length": round(lengths.mean(), 1),
"median_length": round(lengths.median(), 1),
"short_replies_ratio": round(
(lengths < 50).sum() / len(lengths) * 100, 2
),
"detailed_replies_ratio": round(
(lengths >= 200).sum() / len(lengths) * 100, 2
),
}
七、学术研究应用示例
以下所有研究示例仅供学术研究参考，不构成任何投资建议或投资决策依据。
7.1 信息披露效率研究
def disclosure_efficiency_study(df: pd.DataFrame, stock_code: str) -> dict:
"""
信息披露效率研究示例。
研究目标:
分析指定公司在近90日的回复率、回复时效性、回复质量等指标，
评估其信息披露效率。此分析仅供学术研究参考。
参数:
df: 问答数据
stock_code: 研究标的证券代码
返回:
研究指标字典
"""
stock_df = df[df["stock_code"] == stock_code] if "stock_code" in df.columns else df
evaluator = QAQualityEvaluator()
reply_rate = evaluator.evaluate_reply_rate(stock_df)
timeliness = evaluator.evaluate_reply_timeliness(stock_df)
detail = evaluator.evaluate_reply_detail(stock_df)
result = {
"stock_code": stock_code,
"study_period": "近90个交易日",
"disclaimer": "本分析仅供学术研究参考，不构成投资建议",
"reply_rate": reply_rate,
"timeliness": timeliness,
"detail": detail,
}
print(f"[学术研究] {stock_code} 信息披露效率分析:")
print(f" 回复率: {reply_rate.get('reply_rate', 'N/A')}%")
print(f" 平均回复延迟: {timeliness.get('avg_delay_hours', 'N/A')}小时")
print(f" 平均回复长度: {detail.get('avg_length', 'N/A')}字")
print(f" 详细回复占比: {detail.get('detailed_replies_ratio', 'N/A')}%")
return result
7.2 投资者行为研究方法论
def investor_behavior_study(df: pd.DataFrame) -> dict:
"""
投资者行为研究方法论示例。
研究目标:
分析投资者提问的主题分布、时间分布、情感倾向等，
了解投资者关注热点和行为模式。
此分析仅供学术研究参考，不构成投资建议。
参数:
df: 问答数据
返回:
研究结果字典
"""
question_col = None
for col in ["question", "question_text", "提问内容"]:
if col in df.columns:
question_col = col
break
result = {
"disclaimer": "本分析仅供学术研究参考，不构成投资建议",
"total_qa_count": len(df),
}
# 主题分布分析
if question_col:
topic_keywords = {
"业务咨询": ["产品", "业务", "技术", "市场", "客户"],
"财务数据": ["营收", "利润", "收入", "业绩", "财报"],
"重大事项": ["并购", "重组", "定增", "分红", "股权"],
"股价相关": ["股价", "涨跌", "估值", "市盈率"],
"股东人数": ["股东", "户数", "机构"],
"产能订单": ["产能", "订单", "出货", "产量"],
}
topic_counts = {}
for topic, keywords in topic_keywords.items():
count = df[question_col].astype(str).apply(
lambda x: any(kw in x for kw in keywords)
).sum()
topic_counts[topic] = int(count)
result["topic_distribution"] = topic_counts
print("[学术研究] 投资者提问主题分布:")
for topic, count in sorted(topic_counts.items(), key=lambda x: -x[1]):
print(f" {topic}: {count} 条 ({round(count/len(df)*100, 1)}%)")
# 情感分析
if question_col:
analyzer = SentimentAnalyzer()
sentiments = df[question_col].astype(str).apply(analyzer.analyze)
pos_ratio = round((sentiments.apply(lambda x: x["label"]) == "正面").sum() / len(df) * 100, 1)
neg_ratio = round((sentiments.apply(lambda x: x["label"]) == "负面").sum() / len(df) * 100, 1)
neu_ratio = round(100 - pos_ratio - neg_ratio, 1)
result["sentiment_distribution"] = {
"正面": pos_ratio,
"中性": neu_ratio,
"负面": neg_ratio,
}
print(f"[学术研究] 投资者提问情感分布:")
print(f" 正面: {pos_ratio}%")
print(f" 中性: {neu_ratio}%")
print(f" 负面: {neg_ratio}%")
return result
八、完整Python代码实现
8.1 问答数据获取类
class SecretaryQAFetcher:
"""
董秘互动问答数据获取类。
声明:
本类仅供技术学习与学术研究使用。
获取的数据不构成任何投资建议。
"""
def __init__(self, days: int = 90):
"""
初始化数据获取器。
参数:
days: 获取近N天的数据，默认90天
"""
self.days = days
self.end_date = datetime.now().strftime("%Y%m%d")
self.start_date = (
datetime.now() - timedelta(days=days)
).strftime("%Y%m%d")
def fetch_full_market(self) -> pd.DataFrame:
"""获取全市场问答数据。"""
return get_full_market_qa(self.days)
def fetch_by_stock(self, stock_code: str) -> pd.DataFrame:
"""获取个股问答数据。"""
return get_stock_qa(stock_code, self.days)
def fetch_and_process(self, stock_code: str = None) -> pd.DataFrame:
"""
获取数据并进行基础处理。
参数:
stock_code: 如果指定则获取个股数据，否则获取全市场数据
"""
if stock_code:
df = self.fetch_by_stock(stock_code)
else:
df = self.fetch_full_market()
if df.empty:
return df
# 数据清洗
text_cols = []
for col in df.columns:
if any(keyword in col for keyword in
["question", "answer", "提问", "回复", "回答"]):
text_cols.append(col)
if text_cols:
cleaner = TextCleaner()
df = cleaner.batch_clean(df, text_cols)
return df
8.2 NLP处理类
class QANLPProcessor:
"""
问答数据NLP处理类。
声明:
本类输出的所有分析结果仅供学术研究参考，
不构成任何投资建议或投资价值判断。
"""
def __init__(self):
self.cleaner = TextCleaner()
self.extractor = InfoExtractor()
self.sentiment_analyzer = SentimentAnalyzer()
def process_single(self, question: str, answer: str) -> dict:
"""
处理单条问答。
参数:
question: 提问内容
answer: 回复内容
返回:
处理结果字典
"""
q_clean = self.cleaner.clean_text(question)
a_clean = self.cleaner.clean_text(answer)
q_info = self.extractor.extract_all(q_clean)
a_info = self.extractor.extract_all(a_clean)
q_sentiment = self.sentiment_analyzer.analyze(q_clean)
a_sentiment = self.sentiment_analyzer.analyze(a_clean)
return {
"question_cleaned": q_clean,
"answer_cleaned": a_clean,
"question_info": q_info,
"answer_info": a_info,
"question_sentiment": q_sentiment,
"answer_sentiment": a_sentiment,
}
def process_dataframe(self, df: pd.DataFrame,
q_col: str, a_col: str) -> pd.DataFrame:
"""
批量处理DataFrame中的问答数据。
参数:
df: 原始数据
q_col: 提问内容列名
a_col: 回复内容列名
返回:
添加了NLP处理结果的新DataFrame
"""
results = []
for _, row in df.iterrows():
result = self.process_single(str(row[q_col]), str(row[a_col]))
results.append(result)
result_df = pd.DataFrame(results)
return pd.concat([df.reset_index(drop=True), result_df], axis=1)
8.3 情感分析类
class AdvancedSentimentAnalyzer:
"""
进阶情感分析类。
声明:
- 本类所有分析结果仅供学术研究参考
- 不建议将分析结果用于任何投资决策
- 情感分析结果不代表证券未来走势
"""
# 程度副词（用于调节情感强度）
INTENSIFIERS = {
"大幅": 1.5, "显著": 1.5, "明显": 1.3,
"小幅": 0.7, "略有": 0.7, "基本": 0.9,
"持续": 1.2, "稳步": 1.1, "快速": 1.4,
}
@classmethod
def analyze_with_intensity(cls, text: str) -> dict:
"""
带强度调节的情感分析。
参数:
text: 待分析文本
返回:
包含强度调节情感得分的字典
"""
base_result = SentimentAnalyzer.analyze(text)
# 检测程度副词并调节得分
intensity = 1.0
for word, factor in cls.INTENSIFIERS.items():
if word in text:
intensity = max(intensity, factor)
adjusted_score = round(base_result["score"] * intensity, 4)
if adjusted_score > 0.2:
label = "正面"
elif adjusted_score < -0.2:
label = "负面"
else:
label = "中性"
return {
"score": adjusted_score,
"label": label,
"base_score": base_result["score"],
"intensity": intensity,
"disclaimer": "本分析仅供学术研究参考，不构成投资建议",
}
8.4 数据存储类
import sqlite3
import json
from pathlib import Path
class QADataStorage:
"""
问答数据存储类。
使用SQLite进行本地数据存储，
便于离线分析和研究复现。
"""
def __init__(self, db_path: str = "secretary_qa.db"):
"""
初始化数据库连接。
参数:
db_path: 数据库文件路径
"""
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 qa_records (
id INTEGER PRIMARY KEY AUTOINCREMENT,
exchange TEXT,
stock_code TEXT,
stock_name TEXT,
question_date TEXT,
question_text TEXT,
answer_date TEXT,
answer_text TEXT,
answer_status TEXT,
sentiment_score REAL,
sentiment_label TEXT,
raw_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_stock
ON qa_records(stock_code)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_date
ON qa_records(question_date)
""")
conn.commit()
def save(self, df: pd.DataFrame):
"""
将DataFrame数据保存到数据库。
参数:
df: 问答数据
"""
if df.empty:
print("[WARN] 数据为空，跳过保存")
return
with sqlite3.connect(self.db_path) as conn:
for _, row in df.iterrows():
exchange = row.get("exchange", "")
stock_code = row.get("stock_code", row.get("证券代码", ""))
stock_name = row.get("stock_name", row.get("证券简称", ""))
question_text = str(row.get("question", row.get("提问内容", "")))
answer_text = str(row.get("answer", row.get("回复内容", "")))
conn.execute("""
INSERT INTO qa_records
(exchange, stock_code, stock_name,
question_text, answer_text,
raw_data)
VALUES (?, ?, ?, ?, ?, ?)
""", (
exchange, stock_code, stock_name,
question_text, answer_text,
json.dumps(row.to_dict(), ensure_ascii=False)
))
conn.commit()
print(f"[INFO] 数据保存完成，共 {len(df)} 条记录 -> {self.db_path}")
def load(self, stock_code: str = None,
days: int = 90) -> pd.DataFrame:
"""
从数据库加载数据。
参数:
stock_code: 如果指定则加载该股票数据
days: 加载近N天的数据
返回:
DataFrame
"""
with sqlite3.connect(self.db_path) as conn:
query = "SELECT * FROM qa_records WHERE 1=1"
params = []
if stock_code:
query += " AND stock_code = ?"
params.append(stock_code)
if days:
cutoff = (
datetime.now() - timedelta(days=days)
).strftime("%Y-%m-%d")
query += " AND question_date >= ?"
params.append(cutoff)
query += " ORDER BY question_date DESC"
df = pd.read_sql_query(query, conn, params=params)
print(f"[INFO] 数据加载完成，共 {len(df)} 条记录")
return df
8.5 可视化类
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.font_manager as fm
class QAVisualizer:
"""
问答数据可视化类。
声明:
生成的图表仅供学术研究展示使用，
不构成任何投资分析或投资建议。
"""
def __init__(self, output_dir: str = "./charts"):
"""
初始化可视化工具。
参数:
output_dir: 图表输出目录
"""
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)
# 设置中文字体（如系统无中文字体，图表中文可能显示为方块）
plt.rcParams["font.sans-serif"] = [
"SimHei", "WenQuanYi Micro Hei",
"Noto Sans CJK SC", "DejaVu Sans"
]
plt.rcParams["axes.unicode_minus"] = False
def plot_topic_distribution(self, topic_counts: dict,
title: str = "投资者提问主题分布"):
"""
绘制提问主题分布饼图。
参数:
topic_counts: 主题计数字典
title: 图表标题
"""
fig, ax = plt.subplots(figsize=(10, 8))
labels = list(topic_counts.keys())
sizes = list(topic_counts.values())
colors = plt.cm.Set3(range(len(labels)))
wedges, texts, autotexts = ax.pie(
sizes, labels=labels, colors=colors,
autopct="%1.1f%%", startangle=90
)
ax.set_title(f"{title}\n（仅供学术研究参考）", fontsize=14)
output_path = self.output_dir / "topic_distribution.png"
plt.savefig(output_path, dpi=150, bbox_inches="tight")
plt.close()
print(f"[INFO] 主题分布图已保存: {output_path}")
def plot_sentiment_distribution(self, sentiment_counts: dict,
title: str = "投资者提问情感分布"):
"""
绘制情感分布柱状图。
参数:
sentiment_counts: 情感计数字典 {"正面": x, "中性": y, "负面": z}
title: 图表标题
"""
fig, ax = plt.subplots(figsize=(8, 6))
labels = list(sentiment_counts.keys())
values = list(sentiment_counts.values())
colors = ["#4CAF50", "#FFC107", "#F44336"]
bars = ax.bar(labels, values, color=colors, edgecolor="gray")
for bar, val in zip(bars, values):
ax.text(bar.get_x() + bar.get_width() / 2,
bar.get_height() + 0.5,
f"{val}%", ha="center", fontsize=12)
ax.set_title(f"{title}\n（仅供学术研究参考）", fontsize=14)
ax.set_ylabel("占比 (%)")
output_path = self.output_dir / "sentiment_distribution.png"
plt.savefig(output_path, dpi=150, bbox_inches="tight")
plt.close()
print(f"[INFO] 情感分布图已保存: {output_path}")
def plot_qa_timeline(self, df: pd.DataFrame,
stock_code: str = "全市场"):
"""
绘制问答数量时间线图。
参数:
df: 问答数据（需包含日期列）
stock_code: 标题中显示的股票代码
"""
date_col = None
for col in ["question_date", "提问日期", "date"]:
if col in df.columns:
date_col = col
break
if date_col is None:
print("[WARN] 未找到日期列，无法绘制时间线")
return
daily_counts = df.groupby(date_col).size().reset_index(name="count")
daily_counts = daily_counts.sort_values(date_col)
fig, ax = plt.subplots(figsize=(14, 6))
ax.plot(daily_counts[date_col], daily_counts["count"],
color="#2196F3", linewidth=1.5)
ax.fill_between(daily_counts[date_col], daily_counts["count"],
alpha=0.3, color="#2196F3")
ax.set_title(f"问答数量时间线 ({stock_code})\n（仅供学术研究参考）",
fontsize=14)
ax.set_xlabel("日期")
ax.set_ylabel("问答数量")
plt.xticks(rotation=45)
output_path = self.output_dir / "qa_timeline.png"
plt.savefig(output_path, dpi=150, bbox_inches="tight")
plt.close()
print(f"[INFO] 时间线图已保存: {output_path}")
九、合规与法律注意事项
9.1 互动平台数据版权归交易所所有
沪深北交易所董秘互动平台上的所有问答数据，其版权归属于各交易所及相应的上市公司。使用这些数据时必须注意：
- 注明来源：在任何研究论文、报告或分析中引用数据时，必须明确标注数据来源（如"数据来源：深交所互动易"）。
- 禁止商用：未经授权，不得将采集的数据用于商业产品或付费服务中。
- 合理使用：仅在不影响数据平台正常运营的前提下进行合理使用。
9.2 不得利用未公开信息进行交易
董秘互动平台的问答内容属于公开信息，但在使用过程中仍需注意：
- 时效性：公司回复内容在发布前属于未公开信息，不得在发布前利用该信息进行交易。
- 内幕信息：如问答中涉及尚未公告的重大事项，投资者应等待正式公告发布后再做决策。
- 合规交易：所有投资决策应基于公开、合法的信息，遵守证券法律法规。
9.3 个人信息保护
在采集和处理问答数据时，需遵守《中华人民共和国个人信息保护法》：
- 不采集个人信息：不应采集提问者的姓名、联系方式等个人隐私信息。
- 匿名化处理：如数据中包含可识别个人身份的信息，应进行匿名化处理。
- 数据安全：妥善保管采集的数据，防止数据泄露。
9.4 证券投资咨询资质要求
根据《中华人民共和国证券法》和《证券投资咨询业务管理办法》：
- 资质要求：从事证券投资咨询业务需取得相应资质。
- 禁止行为：未取得资质的个人或机构不得向他人提供证券投资建议、投资分析等。
- 合规使用：本模块及文档仅供技术学习，不提供任何投资咨询服务。
十、免责声明与风险提示
10.1 免责声明
- 本文档仅供技术学习和学术研究参考，不构成任何投资建议、投资咨询或证券交易指导。
- 文档中涉及的代码、分析方法、数据结构均为技术学习示例，不保证其完整性、准确性和时效性。
- 文档中提及的任何上市公司名称、证券代码仅作为技术演示的参数示例，不构成对该证券的任何评价或推荐。
- 基于本文档内容进行的任何投资决策，均由投资者自行承担全部风险和责任。
- 本文档作者不对因使用本文档内容而导致的任何直接或间接损失承担责任。
10.2 风险提示
- 股市有风险，投资需谨慎。 证券市场存在系统性风险、政策风险、公司经营风险等多种风险因素。
- 历史数据不代表未来。 过去的问答内容和公司回复不代表未来业绩或股价走势。
- 技术分析有局限性。 NLP情感分析、关键词提取等技术手段存在误判可能，不能作为投资依据。
- 数据可能存在延迟或错误。 采集的数据可能因网络、接口变更等原因存在延迟或不准确的情况。
- 请遵守法律法规。 使用本模块时请严格遵守相关法律法规，不得用于任何违法违规用途。
10.3 合规使用建议
- 仅将本模块用于个人技术学习和学术研究。
- 在公开发表的研究成果中注明数据来源。
- 遵守各交易所平台的使用规则和 robots.txt 协议。
- 控制数据采集频率，不对目标服务器造成过大压力。
- 不将采集的数据用于商业用途或分发给第三方。
- 如有疑问，请咨询专业法律人士。
再次声明：本文档全部内容仅供技术学习和学术研究参考，不构成任何投资建议。股市有风险，投资需谨慎。