A股热点新闻数据模块（学习模拟版本）
免责声明：本文档仅供技术学习和学术研究参考，不构成任何投资建议、投资咨询或证券交易指导。股市有风险，投资需谨慎。任何基于本文档内容进行的投资决策，均由投资者自行承担全部风险和责任。本文档中提及的任何数据源、工具、方法和策略仅为技术演示用途，不保证其准确性、完整性和时效性。
一、模块概述
1.1 模块定位
本模块旨在获取近90个交易日全A股市场公开新闻资讯数据，支持个股关联查询功能。模块采用Python语言开发，整合多种公开数据源接口，构建一套完整的新闻数据采集、处理、存储与分析的技术框架。
核心能力包括：
- 全市场热点新闻的定时采集与增量更新
- 个股维度新闻的精准关联与聚合
- 基于NLP技术的新闻情感分析、实体识别与关键词提取
- 热点话题的自动聚类与趋势追踪
- 面向学术研究的数据因子生成与可视化
1.2 数据来源
本模块所使用的数据来源均为公开合法的财经信息渠道，具体包括：
| 数据来源 | 类型 | 说明 |
|---|---|---|
| 东方财富（eastmoney） | 公开财经媒体 | 提供个股新闻、公告、研报等资讯 |
| 同花顺（10jqka） | 公开财经媒体 | 提供市场热点、概念板块新闻 |
| 财联社（cls） | 公开财经媒体 | 提供实时电报、快讯类资讯 |
| AKShare | 开源Python库 | 封装了多个公开数据接口 |
| Tushare | 开源Python库 | 需注册获取token，有免费额度 |
1.3 数据特点
A股市场新闻数据具有以下显著技术特征：
- 实时性强：重大政策新闻可在数秒内影响市场情绪，数据延迟需控制在分钟级以内
- 数据量大：全市场每日产生的新闻资讯可达数万条，90个交易日累计数据量可观
- 噪音比例高：大量重复新闻、标题党内容需要通过去重与质量过滤机制处理
- 与股价波动高度相关：新闻事件是市场情绪的重要驱动因素，但具体影响机制复杂且非线性
- 多源异构：不同数据源的数据格式、字段定义、更新频率均存在差异
1.4 应用场景
重要声明：本模块及所有相关技术方案仅供学术研究使用，不用于实际交易信号生成。
学术研究方向包括：
- 事件驱动模型研究：探索新闻事件与市场波动之间的统计关系
- 舆情分析方法论：研究基于NLP技术的市场情绪度量方法
- NLP技术应用：金融领域自然语言处理技术的实践与验证
- 数据工程实践：大规模文本数据的采集、存储与处理技术
二、数据结构设计
2.1 新闻数据表结构
新闻数据的主表采用以下Schema设计，支持全文检索与多维分析：
news_schema = {
"news_id": "唯一标识（UUID格式）",
"publish_time": "发布时间（精确到秒，datetime格式）",
"title": "新闻标题（最大长度256字符）",
"content": "新闻正文（TEXT类型，支持全文索引）",
"source": "来源媒体名称",
"source_url": "原文链接（用于溯源与版权标注）",
"news_type": "类型枚举：公告/研报/快讯/深度/评论",
"sentiment": "情感得分（-1到1浮点数，仅供参考，不作为投资依据）",
"related_stocks": "关联股票列表（JSON数组，如['600519.SH','000858.SZ']）",
"industry_tags": "行业标签（JSON数组，如['白酒','消费']）",
"concept_tags": "概念标签（JSON数组，如['新能源','碳中和']）",
"importance_level": "重要等级（1-5整数，基于阅读量与传播速度综合计算）",
"read_count": "阅读数",
"comment_count": "评论数",
"create_time": "数据入库时间",
"update_time": "数据最后更新时间",
"is_deleted": "软删除标记（0/1）"
}
字段设计说明：
news_id采用UUID确保全局唯一性，避免多源数据合并时的主键冲突sentiment字段的得分范围设计为[-1, 1]，其中负值表示负面情绪、正值表示正面情绪、0表示中性。该得分仅为NLP模型输出结果，不代表任何投资建议related_stocks采用JSON数组格式存储，支持一条新闻关联多只股票的场景importance_level采用1-5的整数评级，综合考量阅读量、评论数、转载次数等指标
2.2 个股新闻关联表
个股新闻关联表用于实现从股票维度快速查询相关新闻：
stock_news_relation_schema = {
"relation_id": "关联记录唯一标识",
"stock_code": "股票代码（如600519.SH）",
"stock_name": "股票名称",
"news_id": "关联的新闻ID（外键）",
"relevance_score": "相关度得分（0-1，基于实体共现与语义相似度计算）",
"mention_position": "股票在文中出现的位置信息（JSON）",
"relation_type": "关联类型枚举：直接提及/行业关联/概念关联/供应链关联",
"create_time": "关联创建时间"
}
索引设计：
-- 主索引：按股票代码快速查询
CREATE INDEX idx_stock_code ON stock_news_relation(stock_code);
-- 时间索引：按时间范围筛选
CREATE INDEX idx_news_time ON stock_news_relation(create_time);
-- 复合索引：股票+时间联合查询
CREATE INDEX idx_stock_time ON stock_news_relation(stock_code, create_time);
-- 相关度索引：按相关度排序
CREATE INDEX idx_relevance ON stock_news_relation(relevance_score DESC);
2.3 热点话题聚合表
热点话题聚合表用于存储自动聚类生成的市场热点话题：
hot_topic_schema = {
"topic_id": "话题唯一标识",
"topic_name": "话题名称（由关键词自动生成或人工标注）",
"topic_keywords": "话题关键词列表（JSON数组）",
"news_count": "包含的新闻数量",
"start_time": "话题起始时间",
"end_time": "话题结束时间（持续更新）",
"heat_score": "热度得分（基于新闻数量、传播速度、情感强度综合计算）",
"related_stocks": "话题关联的核心股票列表（JSON数组）",
"related_industries": "话题关联的行业列表（JSON数组）",
"topic_summary": "话题摘要（由NLP自动生成，仅供参考）",
"sentiment_trend": "情感趋势数据（JSON，存储最近N个时间窗口的情感均值）",
"update_time": "最后更新时间"
}
三、数据采集方案
3.1 API数据源（仅列示公开合法接口）
本模块使用以下公开合法的数据获取接口：
（1）AKShare（开源免费，推荐首选）
AKShare是一个开源的金融数据接口库，封装了多个公开数据源，完全免费使用。
import akshare as ak
# 获取实时财经新闻
df_news = ak.stock_news_em(symbol="全部") # 东方财富全市场新闻
# 获取个股相关新闻
df_stock_news = ak.stock_news_em(symbol="600519") # 贵州茅台相关新闻
# 获取概念板块新闻
df_concept = ak.stock_board_concept_name_em() # 概念板块列表
（2）Tushare（需注册，有免费额度）
Tushare是另一个常用的金融数据接口，需要注册获取token。
import tushare as ts
# 初始化（需先注册获取token）
ts.set_token('your_token_here')
pro = ts.pro_api()
# 获取新闻数据
df_news = pro.news(
src='sina', # 数据源
start_date='20250101',
end_date='20250610'
)
（3）东方财富公开API
东方财富提供了部分公开的HTTP接口，可直接调用：
import requests
# 东方财富个股新闻接口（公开接口）
def get_eastmoney_stock_news(stock_code, page=1):
"""获取东方财富个股新闻（公开接口）"""
url = "https://search-api-web.eastmoney.com/search/jsonp"
params = {
"cb": "jQuery",
"param": json.dumps({
"uid": "",
"keyword": stock_code,
"type": ["cmsArticleWebOld"],
"client": "web",
"clientType": "web",
"clientVersion": "curr",
"param": {
"cmsArticleWebOld": {
"searchScope": "default",
"sort": "default",
"pageIndex": page,
"pageSize": 20,
"preTag": "",
"postTag": ""
}
}
})
}
headers = {"User-Agent": "Mozilla/5.0"}
response = requests.get(url, params=params, headers=headers)
return response.text
3.2 采集规范
数据采集必须严格遵守以下规范：
（1）遵守robots.txt
import urllib.robotparser
def check_robots_txt(base_url, path):
"""检查目标路径是否允许爬取"""
rp = urllib.robotparser.RobotFileParser()
rp.set_url(f"{base_url}/robots.txt")
rp.read()
return rp.can_fetch("MyCrawler/1.0", f"{base_url}{path}")
（2）控制请求频率
import time
import random
class RateLimiter:
"""请求频率控制器"""
def __init__(self, min_interval=1.0, max_interval=3.0):
self.min_interval = min_interval
self.max_interval = max_interval
self.last_request_time = 0
def wait(self):
"""在请求前调用，确保满足频率限制"""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_interval:
sleep_time = self.min_interval + random.uniform(0, self.max_interval - self.min_interval)
time.sleep(sleep_time)
self.last_request_time = time.time()
# 使用示例
limiter = RateLimiter(min_interval=1.0, max_interval=2.0)
for url in url_list:
limiter.wait() # 自动等待
response = requests.get(url)
（3）仅采集公开数据
- 不采集需要付费订阅的数据
- 不绕过任何访问控制机制
- 不使用未经授权的API接口
- 所有数据保留原始来源标注，尊重数据版权
3.3 增量更新机制
增量更新采用基于时间戳的拉取策略，避免重复采集：
class IncrementalUpdater:
"""增量更新管理器"""
def __init__(self, db_conn):
self.db = db_conn
self.checkpoint_table = "data_checkpoint"
def get_last_checkpoint(self, source_name):
"""获取指定数据源的最后更新时间点"""
query = """
SELECT last_update_time
FROM data_checkpoint
WHERE source_name = %s
"""
result = self.db.execute(query, (source_name,))
if result:
return result[0]['last_update_time']
return None
def update_checkpoint(self, source_name, last_time):
"""更新数据源的时间检查点"""
query = """
INSERT INTO data_checkpoint (source_name, last_update_time)
VALUES (%s, %s)
ON CONFLICT (source_name)
DO UPDATE SET last_update_time = %s
"""
self.db.execute(query, (source_name, last_time, last_time))
def fetch_incremental(self, source_name, fetch_func):
"""执行增量数据获取"""
last_time = self.get_last_checkpoint(source_name)
new_data = fetch_func(since=last_time)
if new_data and len(new_data) > 0:
max_time = max(item['publish_time'] for item in new_data)
self.update_checkpoint(source_name, max_time)
return new_data
四、近90日数据获取实现
4.1 全市场热点新闻获取
以下代码实现全A股市场近90个交易日的热点新闻批量获取：
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 MarketNewsCollector:
"""全市场热点新闻采集器（仅供技术学习）"""
def __init__(self):
self.rate_limiter = RateLimiter(min_interval=1.5, max_interval=3.0)
self.news_cache = []
def get_trading_days(self, days=90):
"""
获取最近N个交易日的日期列表
使用AKShare获取交易日历
"""
try:
df_trade = ak.tool_trade_date_hist_sina()
today = datetime.now().strftime("%Y-%m-%d")
trading_days = df_trade[df_trade['trade_date'] <= today]['trade_date'].tolist()
return trading_days[-days:]
except Exception as e:
logger.error(f"获取交易日历失败: {e}")
# 回退方案：使用自然日近似
end_date = datetime.now()
start_date = end_date - timedelta(days=int(days * 1.4))
return pd.date_range(start=start_date, end=end_date, freq='B').strftime("%Y-%m-%d").tolist()
def fetch_market_news_by_date(self, date_str):
"""
按日期获取全市场新闻
参数: date_str - 日期字符串，格式 YYYY-MM-DD
"""
self.rate_limiter.wait()
try:
# 使用AKShare获取东方财富全市场新闻
df = ak.stock_news_em(symbol="全部")
if df is not None and not df.empty:
df['publish_date'] = date_str
df['news_type'] = '快讯'
df['source'] = '东方财富'
logger.info(f"日期 {date_str} 获取到 {len(df)} 条新闻")
return df
except Exception as e:
logger.warning(f"获取日期 {date_str} 新闻失败: {e}")
return pd.DataFrame()
def fetch_all_market_news(self, days=90):
"""
批量获取近N个交易日的全市场新闻
注意：实际运行需较长时间，建议分批执行
"""
trading_days = self.get_trading_days(days)
all_news = []
for i, date_str in enumerate(trading_days):
logger.info(f"正在获取第 {i+1}/{len(trading_days)} 天的新闻: {date_str}")
df = self.fetch_market_news_by_date(date_str)
if not df.empty:
all_news.append(df)
# 每获取10天数据保存一次中间结果
if (i + 1) % 10 == 0 and all_news:
interim_df = pd.concat(all_news, ignore_index=True)
interim_df.to_parquet(f'/data/user/work/news_interim_{i+1}.parquet', index=False)
logger.info(f"已保存中间结果，累计 {len(interim_df)} 条新闻")
if all_news:
final_df = pd.concat(all_news, ignore_index=True)
final_df = final_df.drop_duplicates(subset=['title', 'publish_date'], keep='last')
logger.info(f"全市场新闻获取完成，共 {len(final_df)} 条（去重后）")
return final_df
return pd.DataFrame()
4.2 个股关联新闻获取
class StockNewsCollector:
"""个股关联新闻采集器（仅供技术学习）"""
def __init__(self):
self.rate_limiter = RateLimiter(min_interval=2.0, max_interval=4.0)
def fetch_stock_news(self, stock_code, stock_name=None):
"""
获取指定个股的关联新闻
参数:
stock_code: 股票代码，如 "600519"
stock_name: 股票名称（可选，用于日志记录）
"""
self.rate_limiter.wait()
try:
df = ak.stock_news_em(symbol=stock_code)
if df is not None and not df.empty:
df['stock_code'] = stock_code
if stock_name:
df['stock_name'] = stock_name
df['source'] = '东方财富'
logger.info(f"股票 {stock_code} 获取到 {len(df)} 条新闻")
return df
except Exception as e:
logger.warning(f"获取股票 {stock_code} 新闻失败: {e}")
return pd.DataFrame()
def batch_fetch_stock_news(self, stock_list):
"""
批量获取多只股票的关联新闻
参数:
stock_list: 股票列表，格式为 [(code, name), ...]
"""
all_news = []
total = len(stock_list)
for i, (code, name) in enumerate(stock_list):
logger.info(f"正在获取第 {i+1}/{total} 只股票的新闻: {code} {name}")
df = self.fetch_stock_news(code, name)
if not df.empty:
all_news.append(df)
# 控制总采集量，避免过度请求
if len(all_news) > 50000:
logger.warning("已达到单次采集上限50000条，停止采集")
break
if all_news:
result = pd.concat(all_news, ignore_index=True)
result = result.drop_duplicates(subset=['title', 'stock_code'], keep='last')
return result
return pd.DataFrame()
def get_stock_list(self):
"""获取全A股股票列表"""
try:
df = ak.stock_zh_a_spot_em()
stock_list = list(zip(df['代码'], df['名称']))
logger.info(f"获取到 {len(stock_list)} 只A股股票")
return stock_list
except Exception as e:
logger.error(f"获取股票列表失败: {e}")
return []
4.3 热点话题追踪
class HotTopicTracker:
"""热点话题追踪器（仅供技术学习）"""
def __init__(self):
self.rate_limiter = RateLimiter(min_interval=2.0, max_interval=3.0)
def fetch_hot_concepts(self):
"""获取当前热门概念板块"""
self.rate_limiter.wait()
try:
df = ak.stock_board_concept_name_em()
if df is not None and not df.empty:
# 按涨跌幅排序获取热点概念
df_sorted = df.sort_values('涨跌幅', ascending=False)
logger.info(f"获取到 {len(df_sorted)} 个概念板块")
return df_sorted
except Exception as e:
logger.error(f"获取概念板块失败: {e}")
return pd.DataFrame()
def fetch_concept_news(self, concept_name):
"""获取指定概念板块的相关新闻"""
self.rate_limiter.wait()
try:
df = ak.stock_board_concept_cons_em(symbol=concept_name)
if df is not None and not df.empty:
df['concept_name'] = concept_name
return df
except Exception as e:
logger.warning(f"获取概念 {concept_name} 新闻失败: {e}")
return pd.DataFrame()
def fetch_hot_industries(self):
"""获取当前热门行业板块"""
self.rate_limiter.wait()
try:
df = ak.stock_board_industry_name_em()
if df is not None and not df.empty:
df_sorted = df.sort_values('涨跌幅', ascending=False)
return df_sorted
except Exception as e:
logger.error(f"获取行业板块失败: {e}")
return pd.DataFrame()
def generate_daily_hot_report(self):
"""
生成每日热点汇总报告（仅供学术研究参考）
"""
concepts = self.fetch_hot_concepts()
industries = self.fetch_hot_industries()
report = {
"report_date": datetime.now().strftime("%Y-%m-%d"),
"top_concepts": concepts.head(20).to_dict('records') if not concepts.empty else [],
"top_industries": industries.head(20).to_dict('records') if not industries.empty else [],
"disclaimer": "本报告仅供学术研究参考，不构成任何投资建议"
}
return report
五、NLP处理与信息提取
5.1 情感分析（SnowNLP/百度API）
情感分析是金融NLP的核心任务之一。本模块提供两种实现方案：
import snownlp
from snownlp import SnowNLP
import numpy as np
class SentimentAnalyzer:
"""
新闻情感分析器
注意：情感得分仅为NLP模型输出，不代表任何投资建议或未来走势预测
"""
def __init__(self):
# 金融领域正面词汇（示例，可根据研究需要扩展）
self.positive_words = {
'增长', '上涨', '突破', '创新高', '利好', '超预期',
'盈利', '增长', '提升', '优化', '升级', '扩张',
'合作', '签约', '中标', '获批', '回购', '增持'
}
# 金融领域负面词汇（示例）
self.negative_words = {
'下跌', '亏损', '下滑', '暴跌', '利空', '不及预期',
'减持', '抛售', '违约', '退市', '处罚', '调查',
'风险', '警告', '下降', '萎缩', '裁员', '停产'
}
def analyze_snownlp(self, text):
"""
使用SnowNLP进行基础情感分析
返回值范围: 0-1，0表示极度负面，1表示极度正面
"""
try:
s = SnowNLP(text)
score = s.sentiments
# 将0-1映射到-1到1的范围
normalized_score = score * 2 - 1
return round(normalized_score, 4)
except Exception:
return 0.0
def analyze_keyword_based(self, text):
"""
基于金融领域关键词的情感分析（增强方案）
"""
text_lower = text.lower()
pos_count = sum(1 for w in self.positive_words if w in text_lower)
neg_count = sum(1 for w in self.negative_words if w in text_lower)
total = pos_count + neg_count
if total == 0:
return 0.0 # 中性
score = (pos_count - neg_count) / total
return round(score, 4)
def analyze_combined(self, text):
"""
综合情感分析：结合SnowNLP与关键词方法
"""
snownlp_score = self.analyze_snownlp(text)
keyword_score = self.analyze_keyword_based(text)
# 加权融合（权重可根据实验调整）
combined_score = 0.4 * snownlp_score + 0.6 * keyword_score
return round(combined_score, 4)
def batch_analyze(self, texts):
"""批量情感分析"""
results = []
for text in texts:
score = self.analyze_combined(str(text))
results.append(score)
return results
5.2 命名实体识别（HanLP/jieba）
import jieba
import jieba.posseg as pseg
class FinancialEntityRecognizer:
"""金融实体识别器"""
def __init__(self):
# 加载金融领域自定义词典
self._load_financial_dict()
def _load_financial_dict(self):
"""加载金融领域专有名词词典"""
financial_terms = [
'A股', '港股', '美股', '科创板', '创业板', '北交所',
'市盈率', '市净率', 'ROE', 'EPS', 'GDP', 'CPI', 'PPI',
'央行', '美联储', '证监会', '银保监会',
'涨停', '跌停', '熔断', '牛市', '熊市',
'蓝筹股', '白马股', '黑马股', '龙头股',
'融资融券', '沪港通', '深港通', '北向资金',
'市盈率', '换手率', '涨跌幅', '振幅',
'新能源', '人工智能', '半导体', '芯片', '光伏',
'锂电池', '储能', '碳中和', '数字经济'
]
for term in financial_terms:
jieba.add_word(term)
def recognize_entities(self, text):
"""
识别文本中的金融实体
返回: dict，包含各类实体列表
"""
entities = {
'stock_names': [], # 股票名称
'stock_codes': [], # 股票代码
'institutions': [], # 机构名称
'financial_terms': [], # 金融术语
'industries': [], # 行业名称
'numbers': [] # 数值信息
}
# 使用jieba分词与词性标注
words = pseg.cut(text)
# 股票代码正则匹配
import re
code_pattern = re.compile(r'\b(6\d{5}|0\d{5}|3\d{5}|30\d{4}|68\d{5})\b')
codes = code_pattern.findall(text)
entities['stock_codes'] = list(set(codes))
for word, flag in words:
if flag == 'nz' and len(word) >= 2:
entities['stock_names'].append(word)
elif flag == 'nt' or flag == 'ni':
entities['institutions'].append(word)
elif flag in ['n', 'nr', 'ns'] and len(word) >= 2:
entities['financial_terms'].append(word)
# 去重
for key in entities:
entities[key] = list(set(entities[key]))
return entities
def extract_stock_mentions(self, text, stock_dict=None):
"""
从文本中提取提及的股票
参数:
text: 新闻文本
stock_dict: 股票代码-名称映射字典（可选）
"""
mentioned = []
if stock_dict:
for code, name in stock_dict.items():
if name in text or code in text:
mentioned.append({'code': code, 'name': name})
return mentioned
5.3 关键词提取（TF-IDF/TextRank）
import jieba.analyse
class KeywordExtractor:
"""新闻关键词提取器"""
def __init__(self):
# 设置jieba停用词
jieba.analyse.set_stop_words('chinese_stopwords.txt')
def extract_tfidf(self, text, topk=10):
"""
基于TF-IDF算法的关键词提取
"""
try:
keywords = jieba.analyse.extract_tags(
text,
topK=topk,
withWeight=True,
allowPOS=('n', 'vn', 'nz', 'nr', 'ns')
)
return [{'keyword': kw, 'weight': round(wt, 4)} for kw, wt in keywords]
except Exception as e:
return []
def extract_textrank(self, text, topk=10):
"""
基于TextRank算法的关键词提取
"""
try:
keywords = jieba.analyse.textrank(
text,
topK=topk,
withWeight=True,
allowPOS=('n', 'vn', 'nz', 'nr', 'ns')
)
return [{'keyword': kw, 'weight': round(wt, 4)} for kw, wt in keywords]
except Exception as e:
return []
def extract_combined(self, text, topk=10):
"""
综合TF-IDF与TextRank的关键词提取
"""
tfidf_results = self.extract_tfidf(text, topk=topk * 2)
textrank_results = self.extract_textrank(text, topk=topk * 2)
# 合并并重新计算权重
keyword_weights = {}
for item in tfidf_results:
keyword_weights[item['keyword']] = keyword_weights.get(item['keyword'], 0) + item['weight'] * 0.5
for item in textrank_results:
keyword_weights[item['keyword']] = keyword_weights.get(item['keyword'], 0) + item['weight'] * 0.5
# 按权重排序
sorted_keywords = sorted(keyword_weights.items(), key=lambda x: x[1], reverse=True)
return [{'keyword': kw, 'weight': round(wt, 4)} for kw, wt in sorted_keywords[:topk]]
5.4 新闻去重与聚类
from collections import defaultdict
import hashlib
import re
class NewsDeduplicator:
"""新闻去重处理器"""
@staticmethod
def compute_title_hash(title):
"""计算标题的哈希值，用于快速去重"""
# 标准化处理：去除空格、统一标点
normalized = re.sub(r'[\s\u3000]+', '', title)
normalized = re.sub(r'[，。！？、；：""''【】（）]', '', normalized)
return hashlib.md5(normalized.encode('utf-8')).hexdigest()
@staticmethod
def compute_content_fingerprint(content, window=3):
"""
计算内容指纹（SimHash简化版）
用于检测近似重复新闻
"""
if not content or len(content) < 50:
return None
# 取前500字符作为指纹计算基础
text = content[:500]
return hashlib.sha256(text.encode('utf-8')).hexdigest()
@staticmethod
def is_similar_title(title1, title2, threshold=0.7):
"""
判断两个标题是否相似
基于字符级别的Jaccard相似度
"""
set1 = set(title1)
set2 = set(title2)
intersection = set1 & set2
union = set1 | set2
if not union:
return False
similarity = len(intersection) / len(union)
return similarity >= threshold
def deduplicate_news_list(self, news_list):
"""
对新闻列表进行去重
返回去重后的新闻列表
"""
seen_hashes = set()
seen_fingerprints = set()
unique_news = []
for news in news_list:
title_hash = self.compute_title_hash(news.get('title', ''))
content_fp = self.compute_content_fingerprint(news.get('content', ''))
if title_hash in seen_hashes:
continue
if content_fp and content_fp in seen_fingerprints:
continue
seen_hashes.add(title_hash)
if content_fp:
seen_fingerprints.add(content_fp)
unique_news.append(news)
logger.info(f"新闻去重: {len(news_list)} -> {len(unique_news)} 条")
return unique_news
class NewsClusterer:
"""新闻聚类处理器（仅供学术研究）"""
def __init__(self):
self.keyword_extractor = KeywordExtractor()
def cluster_by_keywords(self, news_list, top_keywords=5):
"""
基于关键词重叠度的新闻聚类
"""
# 提取每条新闻的关键词
news_keywords = []
for news in news_list:
text = news.get('title', '') + ' ' + news.get('content', '')[:200]
keywords = self.keyword_extractor.extract_tfidf(text, topk=top_keywords)
keyword_set = set(item['keyword'] for item in keywords)
news_keywords.append(keyword_set)
# 简单聚类：基于关键词重叠
clusters = []
assigned = [False] * len(news_list)
for i in range(len(news_list)):
if assigned[i]:
continue
cluster = [i]
assigned[i] = True
for j in range(i + 1, len(news_list)):
if assigned[j]:
continue
overlap = news_keywords[i] & news_keywords[j]
if len(overlap) >= 2: # 至少共享2个关键词
cluster.append(j)
assigned[j] = True
clusters.append(cluster)
logger.info(f"新闻聚类: {len(news_list)} 条新闻 -> {len(clusters)} 个话题簇")
return clusters
六、数据存储与管理
6.1 数据库设计
本模块推荐使用PostgreSQL作为主数据库，利用其强大的JSON支持和全文检索能力：
# 数据库初始化SQL
CREATE_TABLE_SQL = """
-- 新闻主表
CREATE TABLE IF NOT EXISTS market_news (
news_id VARCHAR(64) PRIMARY KEY,
publish_time TIMESTAMP NOT NULL,
title VARCHAR(512) NOT NULL,
content TEXT,
source VARCHAR(128),
source_url VARCHAR(1024),
news_type VARCHAR(32) DEFAULT '快讯',
sentiment FLOAT DEFAULT 0.0,
related_stocks JSONB DEFAULT '[]',
industry_tags JSONB DEFAULT '[]',
concept_tags JSONB DEFAULT '[]',
importance_level INTEGER DEFAULT 3,
read_count INTEGER DEFAULT 0,
comment_count INTEGER DEFAULT 0,
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
is_deleted INTEGER DEFAULT 0
);
-- 个股新闻关联表
CREATE TABLE IF NOT EXISTS stock_news_relation (
relation_id SERIAL PRIMARY KEY,
stock_code VARCHAR(16) NOT NULL,
stock_name VARCHAR(64),
news_id VARCHAR(64) NOT NULL REFERENCES market_news(news_id),
relevance_score FLOAT DEFAULT 0.0,
relation_type VARCHAR(32) DEFAULT '直接提及',
create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 热点话题聚合表
CREATE TABLE IF NOT EXISTS hot_topics (
topic_id SERIAL PRIMARY KEY,
topic_name VARCHAR(256) NOT NULL,
topic_keywords JSONB DEFAULT '[]',
news_count INTEGER DEFAULT 0,
start_time TIMESTAMP,
end_time TIMESTAMP,
heat_score FLOAT DEFAULT 0.0,
related_stocks JSONB DEFAULT '[]',
related_industries JSONB DEFAULT '[]',
topic_summary TEXT,
sentiment_trend JSONB DEFAULT '[]',
update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 数据检查点表
CREATE TABLE IF NOT EXISTS data_checkpoint (
source_name VARCHAR(64) PRIMARY KEY,
last_update_time TIMESTAMP NOT NULL,
update_count INTEGER DEFAULT 0
);
-- 创建索引
CREATE INDEX idx_news_publish_time ON market_news(publish_time DESC);
CREATE INDEX idx_news_type ON market_news(news_type);
CREATE INDEX idx_news_sentiment ON market_news(sentiment);
CREATE INDEX idx_news_importance ON market_news(importance_level DESC);
CREATE INDEX idx_news_deleted ON market_news(is_deleted);
CREATE INDEX idx_stock_news_code ON stock_news_relation(stock_code);
CREATE INDEX idx_stock_news_time ON stock_news_relation(create_time DESC);
CREATE INDEX idx_topics_heat ON hot_topics(heat_score DESC);
CREATE INDEX idx_topics_time ON hot_topics(update_time DESC);
-- 全文检索索引（PostgreSQL）
ALTER TABLE market_news ADD COLUMN search_vector TSVECTOR;
CREATE INDEX idx_news_search ON market_news USING GIN(search_vector);
-- 自动更新搜索向量的触发器
CREATE OR REPLACE FUNCTION update_search_vector()
RETURNS TRIGGER AS $$
BEGIN
NEW.search_vector := to_tsvector('simple', COALESCE(NEW.title, '') || ' ' || COALESCE(NEW.content, ''));
RETURN NEW;
END
$$ LANGUAGE plpgsql;
CREATE TRIGGER trg_news_search_vector
BEFORE INSERT OR UPDATE ON market_news
FOR EACH ROW EXECUTE FUNCTION update_search_vector();
"""
6.2 数据更新策略
class NewsStorageManager:
"""新闻数据存储管理器"""
def __init__(self, db_config):
self.db_config = db_config
self.conn = None
def connect(self):
"""建立数据库连接"""
import psycopg2
self.conn = psycopg2.connect(**self.db_config)
return self.conn
def insert_news_batch(self, news_list):
"""
批量插入新闻数据
使用UPSERT避免重复插入
"""
if not news_list or not self.conn:
return 0
cursor = self.conn.cursor()
insert_sql = """
INSERT INTO market_news (
news_id, publish_time, title, content, source,
source_url, news_type, sentiment, related_stocks,
industry_tags, concept_tags, importance_level,
read_count, comment_count
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
ON CONFLICT (news_id)
DO UPDATE SET
sentiment = EXCLUDED.sentiment,
importance_level = EXCLUDED.importance_level,
read_count = EXCLUDED.read_count,
comment_count = EXCLUDED.comment_count,
update_time = CURRENT_TIMESTAMP
"""
import json
records = []
for news in news_list:
records.append((
news.get('news_id', ''),
news.get('publish_time'),
news.get('title', ''),
news.get('content', ''),
news.get('source', ''),
news.get('source_url', ''),
news.get('news_type', '快讯'),
news.get('sentiment', 0.0),
json.dumps(news.get('related_stocks', []), ensure_ascii=False),
json.dumps(news.get('industry_tags', []), ensure_ascii=False),
json.dumps(news.get('concept_tags', []), ensure_ascii=False),
news.get('importance_level', 3),
news.get('read_count', 0),
news.get('comment_count', 0)
))
cursor.executemany(insert_sql, records)
self.conn.commit()
inserted = cursor.rowcount
cursor.close()
logger.info(f"批量插入/更新 {inserted} 条新闻")
return inserted
def query_news_by_stock(self, stock_code, days=90):
"""查询指定股票近N天的关联新闻"""
cursor = self.conn.cursor()
query = """
SELECT n.news_id, n.publish_time, n.title, n.content,
n.source, n.sentiment, n.importance_level,
r.relevance_score, r.relation_type
FROM market_news n
INNER JOIN stock_news_relation r ON n.news_id = r.news_id
WHERE r.stock_code = %s
AND n.publish_time >= NOW() - INTERVAL '%s days'
AND n.is_deleted = 0
ORDER BY n.publish_time DESC
LIMIT 1000
"""
cursor.execute(query, (stock_code, days))
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in cursor.fetchall()]
cursor.close()
return results
def query_hot_news(self, limit=50):
"""查询当日热门新闻"""
cursor = self.conn.cursor()
query = """
SELECT news_id, publish_time, title, source,
sentiment, importance_level, read_count
FROM market_news
WHERE publish_time >= CURRENT_DATE
AND is_deleted = 0
ORDER BY importance_level DESC, read_count DESC
LIMIT %s
"""
cursor.execute(query, (limit,))
columns = [desc[0] for desc in cursor.description]
results = [dict(zip(columns, row)) for row in cursor.fetchall()]
cursor.close()
return results
6.3 数据清洗规则
class NewsDataCleaner:
"""新闻数据清洗处理器"""
def clean_text(self, text):
"""
清洗新闻文本
- 去除HTML标签
- 去除特殊字符
- 标准化空白符
"""
if not text:
return ""
import re
# 去除HTML标签
text = re.sub(r'<[^>]+>', '', text)
# 去除特殊控制字符
text = re.sub(r'[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]', '', text)
# 标准化空白符
text = re.sub(r'\s+', ' ', text).strip()
# 去除多余标点
text = re.sub(r'([，。！？])\1+', r'\1', text)
return text
def validate_news(self, news_item):
"""
验证新闻数据完整性
返回: (is_valid, reason)
"""
if not news_item.get('title'):
return False, "标题为空"
if len(news_item.get('title', '')) < 5:
return False, "标题过短"
if not news_item.get('publish_time'):
return False, "发布时间为空"
if len(news_item.get('content', '')) < 10:
return False, "正文过短"
return True, ""
def clean_news_list(self, news_list):
"""
对新闻列表进行完整清洗流程
"""
cleaned = []
skipped = 0
for news in news_list:
# 验证数据完整性
is_valid, reason = self.validate_news(news)
if not is_valid:
skipped += 1
continue
# 清洗文本
news['title'] = self.clean_text(news.get('title', ''))
news['content'] = self.clean_text(news.get('content', ''))
# 标准化来源
news['source'] = self._normalize_source(news.get('source', ''))
cleaned.append(news)
logger.info(f"数据清洗: {len(news_list)} -> {len(cleaned)} 条有效（跳过 {skipped} 条）")
return cleaned
def _normalize_source(self, source):
"""标准化来源名称"""
source_map = {
'东方财富网': '东方财富',
'证券时报网': '证券时报',
'上海证券报': '上证报',
'中国证券报': '中证报',
}
return source_map.get(source, source)
七、学术研究应用示例
重要声明：以下所有示例仅供学术研究参考，不构成任何投资建议或交易策略。所有因子和模型仅为技术分析工具框架，不保证任何投资收益。
7.1 新闻情感因子研究框架
class SentimentFactorGenerator:
"""
新闻情感因子生成器（仅供学术研究参考）
该因子用于研究新闻情感与市场波动之间的统计关系，
不构成任何买卖建议或收益承诺。
"""
def __init__(self, db_manager):
self.db = db_manager
def calculate_stock_sentiment_factor(self, stock_code, window_days=5):
"""
计算个股的情感因子值
研究方法说明：
- 收集窗口期内所有关联新闻
- 对每条新闻进行情感分析
- 按时间加权计算综合情感得分
- 加权因子：越接近当前时间权重越高
注意：该因子仅用于学术研究中的统计检验，
不代表股票未来走势的预测。
"""
news_list = self.db.query_news_by_stock(stock_code, days=window_days)
if not news_list:
return None
# 时间衰减权重
import math
now = datetime.now()
weighted_sentiment = 0.0
total_weight = 0.0
for news in news_list:
pub_time = news['publish_time']
if isinstance(pub_time, str):
pub_time = datetime.strptime(pub_time, "%Y-%m-%d %H:%M:%S")
days_diff = (now - pub_time).total_seconds() / 86400
weight = math.exp(-0.5 * days_diff) # 指数衰减
weighted_sentiment += news['sentiment'] * weight
total_weight += weight
if total_weight == 0:
return 0.0
factor_value = weighted_sentiment / total_weight
return {
'stock_code': stock_code,
'factor_name': 'news_sentiment',
'factor_value': round(factor_value, 4),
'news_count': len(news_list),
'window_days': window_days,
'disclaimer': '本因子仅供学术研究，不构成投资建议'
}
def calculate_market_sentiment_index(self):
"""
计算全市场情绪指数（仅供学术研究参考）
研究方法：
- 统计当日全市场新闻的正面/负面比例
- 计算加权情感均值
- 生成0-100的市场情绪指数
"""
hot_news = self.db.query_hot_news(limit=500)
if not hot_news:
return None
sentiments = [n['sentiment'] for n in hot_news]
avg_sentiment = np.mean(sentiments)
# 映射到0-100范围
index_value = (avg_sentiment + 1) / 2 * 100
return {
'index_name': 'market_sentiment_index',
'index_value': round(index_value, 2),
'news_sample_size': len(hot_news),
'avg_sentiment': round(avg_sentiment, 4),
'disclaimer': '本指数仅供学术研究，不构成投资建议'
}
7.2 事件驱动模型研究方法论
class EventStudyFramework:
"""
事件驱动研究框架（仅供学术研究参考）
研究目的：探索特定新闻事件发布后，
相关股票在事件窗口内的统计表现。
重要声明：本框架仅用于学术研究中的事件研究方法论演示，
不用于实际交易策略，不构成任何投资建议。
"""
def __init__(self, db_manager):
self.db = db_manager
def identify_event_type(self, news_item):
"""
识别新闻事件类型（仅供研究分类使用）
"""
title = news_item.get('title', '').lower()
content = news_item.get('content', '').lower()
text = title + content
event_types = {
'业绩公告': ['业绩', '财报', '营收', '利润', '净利润', '每股收益'],
'政策事件': ['政策', '监管', '法规', '国务院', '央行', '发改委'],
'并购重组': ['并购', '收购', '重组', '合并', '资产注入'],
'管理层变动': ['董事长', '总经理', '高管', '辞职', '任命'],
'产品发布': ['发布', '上市', '新品', '量产', '获批'],
'法律风险': ['处罚', '违规', '诉讼', '调查', '警告'],
}
for event_type, keywords in event_types.items():
if any(kw in text for kw in keywords):
return event_type
return '其他'
def define_event_window(self, event_date, pre_days=5, post_days=10):
"""
定义事件研究窗口
参数:
event_date: 事件发生日期
pre_days: 事件前窗口天数
post_days: 事件后窗口天数
注意：事件窗口的定义仅为学术研究中的标准做法，
不代表任何交易时机建议。
"""
event_dt = datetime.strptime(event_date, "%Y-%m-%d")
window_start = event_dt - timedelta(days=pre_days)
window_end = event_dt + timedelta(days=post_days)
return {
'event_date': event_date,
'window_start': window_start.strftime("%Y-%m-%d"),
'window_end': window_end.strftime("%Y-%m-%d"),
'pre_days': pre_days,
'post_days': post_days,
'disclaimer': '事件窗口定义仅供学术研究方法论演示'
}
def analyze_event_impact(self, stock_code, event_date):
"""
分析事件对股票的统计影响（仅供学术研究）
研究方法：
1. 识别事件类型
2. 定义事件窗口
3. 收集窗口内的新闻数据
4. 统计情感变化趋势
注意：本分析不预测未来股价走势，
不构成任何买卖建议。
"""
news_list = self.db.query_news_by_stock(stock_code, days=30)
event_news = []
for news in news_list:
pub_date = str(news['publish_time'])[:10]
if pub_date == event_date:
event_news.append(news)
if not event_news:
return None
sentiments = [n['sentiment'] for n in event_news]
event_type = self.identify_event_type(event_news[0])
return {
'stock_code': stock_code,
'event_date': event_date,
'event_type': event_type,
'event_news_count': len(event_news),
'avg_sentiment': round(np.mean(sentiments), 4),
'sentiment_std': round(np.std(sentiments), 4),
'disclaimer': '事件影响分析仅供学术研究，不构成投资建议'
}
7.3 舆情监控技术方案
class SentimentMonitor:
"""
舆情监控技术方案（仅供学术研究参考）
本方案演示如何构建一个市场舆情监控系统，
用于研究新闻舆情与市场行为之间的关系。
不用于实际交易决策。
"""
def __init__(self, db_manager, nlp_processor):
self.db = db_manager
self.nlp = nlp_processor
def monitor_stock_sentiment_trend(self, stock_code, days=30):
"""
监控个股舆情趋势（仅供学术研究）
输出每日情感得分的时间序列，
用于研究舆情变化模式。
"""
news_list = self.db.query_news_by_stock(stock_code, days=days)
if not news_list:
return None
# 按日期聚合情感得分
daily_sentiment = defaultdict(list)
for news in news_list:
date_key = str(news['publish_time'])[:10]
daily_sentiment[date_key].append(news['sentiment'])
trend_data = []
for date_key in sorted(daily_sentiment.keys()):
sentiments = daily_sentiment[date_key]
trend_data.append({
'date': date_key,
'avg_sentiment': round(np.mean(sentiments), 4),
'max_sentiment': round(max(sentiments), 4),
'min_sentiment': round(min(sentiments), 4),
'news_count': len(sentiments),
'positive_ratio': round(sum(1 for s in sentiments if s > 0) / len(sentiments), 4)
})
return {
'stock_code': stock_code,
'monitor_period_days': days,
'trend_data': trend_data,
'disclaimer': '舆情趋势数据仅供学术研究，不构成投资建议'
}
def detect_sentiment_anomaly(self, stock_code, days=30, threshold=2.0):
"""
检测舆情异常波动（仅供学术研究）
使用Z-score方法检测情感得分的异常波动，
用于研究突发事件对市场情绪的影响。
注意：异常检测不预测股价走势，
不构成任何交易信号。
"""
monitor_result = self.monitor_stock_sentiment_trend(stock_code, days)
if not monitor_result:
return None
trend_data = monitor_result['trend_data']
if len(trend_data) < 5:
return None
sentiments = [d['avg_sentiment'] for d in trend_data]
mean_sent = np.mean(sentiments)
std_sent = np.std(sentiments)
anomalies = []
for d in trend_data:
z_score = (d['avg_sentiment'] - mean_sent) / std_sent if std_sent > 0 else 0
if abs(z_score) > threshold:
anomalies.append({
'date': d['date'],
'sentiment': d['avg_sentiment'],
'z_score': round(z_score, 4),
'news_count': d['news_count']
})
return {
'stock_code': stock_code,
'mean_sentiment': round(mean_sent, 4),
'std_sentiment': round(std_sent, 4),
'anomaly_threshold': threshold,
'anomalies': anomalies,
'disclaimer': '舆情异常检测仅供学术研究，不构成投资建议'
}
八、完整Python代码实现
8.1 新闻数据获取类（整合版）
"""
A股热点新闻数据获取模块 - 完整实现
仅供技术学习和学术研究使用
"""
import akshare as ak
import pandas as pd
import numpy as np
import requests
import json
import time
import random
import logging
import uuid
from datetime import datetime, timedelta
from collections import defaultdict
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger('NewsModule')
class AShareNewsModule:
"""
A股热点新闻数据模块（技术学习版）
免责声明：本模块仅供技术学习和学术研究参考，
不构成任何投资建议。股市有风险，投资需谨慎。
"""
DISCLAIMER = "本模块仅供技术学习和学术研究参考，不构成任何投资建议。"
def __init__(self):
self.rate_limiter = RateLimiter()
self.sentiment_analyzer = SentimentAnalyzer()
self.entity_recognizer = FinancialEntityRecognizer()
self.keyword_extractor = KeywordExtractor()
self.deduplicator = NewsDeduplicator()
self.clusterer = NewsClusterer()
def get_full_market_news(self, days=90):
"""获取近N个交易日的全市场新闻"""
collector = MarketNewsCollector()
return collector.fetch_all_market_news(days)
def get_stock_news(self, stock_code, stock_name=None):
"""获取指定个股的关联新闻"""
collector = StockNewsCollector()
return collector.fetch_stock_news(stock_code, stock_name)
def process_news_pipeline(self, raw_news_list):
"""
新闻处理流水线：
去重 -> 清洗 -> NLP分析 -> 结构化输出
"""
if not raw_news_list:
return []
# Step 1: 去重
unique_news = self.deduplicator.deduplicate_news_list(raw_news_list)
# Step 2: 清洗
cleaner = NewsDataCleaner()
cleaned_news = cleaner.clean_news_list(unique_news)
# Step 3: NLP分析
processed_news = []
for news in cleaned_news:
text = news.get('title', '') + ' ' + news.get('content', '')[:500]
# 情感分析
news['sentiment'] = self.sentiment_analyzer.analyze_combined(text)
# 实体识别
entities = self.entity_recognizer.recognize_entities(text)
news['industry_tags'] = entities.get('industries', [])
news['financial_terms'] = entities.get('financial_terms', [])
# 关键词提取
keywords = self.keyword_extractor.extract_combined(text, topk=5)
news['keywords'] = [k['keyword'] for k in keywords]
# 生成唯一ID
news['news_id'] = str(uuid.uuid4())
processed_news.append(news)
logger.info(f"处理流水线完成: {len(raw_news_list)} -> {len(processed_news)}")
return processed_news
8.2 数据可视化类
import matplotlib.pyplot as plt
import matplotlib
matplotlib.use('Agg') # 非交互式后端
matplotlib.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
matplotlib.rcParams['axes.unicode_minus'] = False
class NewsVisualizer:
"""
新闻数据可视化工具（仅供学术研究）
"""
def plot_sentiment_trend(self, trend_data, stock_code, save_path=None):
"""
绘制情感趋势图
参数:
trend_data: 舆情监控输出的trend_data列表
stock_code: 股票代码
save_path: 图片保存路径（可选）
"""
if not trend_data:
logger.warning("无趋势数据可绘制")
return
dates = [d['date'] for d in trend_data]
sentiments = [d['avg_sentiment'] for d in trend_data]
counts = [d['news_count'] for d in trend_data]
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8), sharex=True)
# 情感趋势
ax1.plot(dates, sentiments, 'b-o', markersize=4, linewidth=1.5)
ax1.axhline(y=0, color='gray', linestyle='--', alpha=0.5)
ax1.fill_between(dates, sentiments, 0, alpha=0.3)
ax1.set_ylabel('Sentiment Score')
ax1.set_title(f'{stock_code} News Sentiment Trend (Academic Research Only)')
ax1.grid(True, alpha=0.3)
# 新闻数量
ax2.bar(dates, counts, color='steelblue', alpha=0.7)
ax2.set_ylabel('News Count')
ax2.set_xlabel('Date')
ax2.grid(True, alpha=0.3)
plt.tight_layout()
# 添加免责声明
fig.text(0.5, 0.01,
'Disclaimer: For academic research only. Not investment advice.',
ha='center', fontsize=8, color='gray')
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
logger.info(f"图表已保存至: {save_path}")
plt.close()
def plot_topic_distribution(self, topic_data, save_path=None):
"""
绘制热点话题分布图
参数:
topic_data: 话题数据列表，每个元素包含topic_name和news_count
save_path: 图片保存路径（可选）
"""
if not topic_data:
return
# 按新闻数量排序，取前20
sorted_topics = sorted(topic_data, key=lambda x: x.get('news_count', 0), reverse=True)[:20]
names = [t['topic_name'] for t in sorted_topics]
counts = [t['news_count'] for t in sorted_topics]
fig, ax = plt.subplots(figsize=(12, 8))
bars = ax.barh(range(len(names)), counts, color='steelblue', alpha=0.8)
ax.set_yticks(range(len(names)))
ax.set_yticklabels(names)
ax.invert_yaxis()
ax.set_xlabel('News Count')
ax.set_title('Top 20 Hot Topics Distribution (Academic Research Only)')
ax.grid(True, alpha=0.3, axis='x')
# 在柱状图上标注数值
for bar, count in zip(bars, counts):
ax.text(bar.get_width() + 0.5, bar.get_y() + bar.get_height() / 2,
str(count), va='center', fontsize=9)
fig.text(0.5, 0.01,
'Disclaimer: For academic research only. Not investment advice.',
ha='center', fontsize=8, color='gray')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
logger.info(f"话题分布图已保存至: {save_path}")
plt.close()
def plot_sentiment_distribution(self, news_list, save_path=None):
"""
绘制情感得分分布直方图
"""
if not news_list:
return
sentiments = [n.get('sentiment', 0) for n in news_list]
fig, ax = plt.subplots(figsize=(10, 6))
ax.hist(sentiments, bins=50, color='steelblue', alpha=0.7, edgecolor='white')
ax.axvline(x=0, color='red', linestyle='--', linewidth=1.5, label='Neutral')
ax.axvline(x=np.mean(sentiments), color='orange', linestyle='--',
linewidth=1.5, label=f'Mean: {np.mean(sentiments):.3f}')
ax.set_xlabel('Sentiment Score (-1 to 1)')
ax.set_ylabel('Frequency')
ax.set_title('News Sentiment Distribution (Academic Research Only)')
ax.legend()
ax.grid(True, alpha=0.3)
fig.text(0.5, 0.01,
'Disclaimer: For academic research only. Not investment advice.',
ha='center', fontsize=8, color='gray')
plt.tight_layout()
if save_path:
plt.savefig(save_path, dpi=150, bbox_inches='tight')
plt.close()
8.3 研究因子生成类
class ResearchFactorGenerator:
"""
学术研究因子生成器（仅供学术研究参考）
重要声明：
本类生成的所有因子仅用于学术研究中的统计分析和模型验证。
这些因子不构成任何投资建议、买卖信号或收益承诺。
任何使用这些因子进行投资决策的行为，后果由使用者自行承担。
"""
def __init__(self, db_manager):
self.db = db_manager
self.sentiment_analyzer = SentimentAnalyzer()
def generate_news_volume_factor(self, stock_code, window=5):
"""
新闻量因子：统计窗口期内关联新闻的数量
研究意义：新闻量可反映市场关注度，
但新闻量本身不预测股价方向。
"""
news_list = self.db.query_news_by_stock(stock_code, days=window)
return {
'factor_name': 'news_volume',
'value': len(news_list),
'stock_code': stock_code,
'window': window,
'disclaimer': '仅供学术研究'
}
def generate_sentiment_momentum_factor(self, stock_code, short_window=3, long_window=10):
"""
情感动量因子：短期情感均值与长期情感均值之差
研究意义：情感动量反映舆情变化趋势，
但不构成交易信号。
"""
short_news = self.db.query_news_by_stock(stock_code, days=short_window)
long_news = self.db.query_news_by_stock(stock_code, days=long_window)
short_avg = np.mean([n['sentiment'] for n in short_news]) if short_news else 0
long_avg = np.mean([n['sentiment'] for n in long_news]) if long_news else 0
return {
'factor_name': 'sentiment_momentum',
'value': round(short_avg - long_avg, 4),
'stock_code': stock_code,
'short_window': short_window,
'long_window': long_window,
'disclaimer': '仅供学术研究'
}
def generate_news_diversity_factor(self, stock_code, days=10):
"""
新闻多样性因子：衡量关联新闻来源的多样性
研究意义：新闻来源多样性反映信息覆盖面，
但不直接关联投资价值。
"""
news_list = self.db.query_news_by_stock(stock_code, days=days)
if not news_list:
return {'factor_name': 'news_diversity', 'value': 0}
sources = set(n.get('source', '') for n in news_list)
types = set(n.get('news_type', '') for n in news_list)
# Shannon熵计算多样性
source_counts = defaultdict(int)
for n in news_list:
source_counts[n.get('source', 'unknown')] += 1
total = len(news_list)
entropy = 0
for count in source_counts.values():
p = count / total
if p > 0:
entropy -= p * np.log2(p)
return {
'factor_name': 'news_diversity',
'value': round(entropy, 4),
'source_count': len(sources),
'type_count': len(types),
'stock_code': stock_code,
'disclaimer': '仅供学术研究'
}
九、合规与法律注意事项
9.1 数据版权声明
本模块采集的所有新闻数据均来自公开财经媒体，原始数据的版权归属于各数据来源方。使用本模块时须遵守以下规定：
- 仅用于个人学习和学术研究：不得将采集的数据用于商业用途或大规模分发
- 保留来源标注：所有使用的数据必须标注原始来源
- 遵守数据使用协议：各数据源（如Tushare等）可能有特定的使用条款，使用前须仔细阅读并遵守
- 不得转售数据：严禁将采集的数据出售或以其他方式牟利
9.2 个人信息保护
在数据采集和处理过程中，须严格遵守《个人信息保护法》等相关法律法规：
- 不采集个人隐私信息：新闻中的个人姓名、联系方式等隐私信息应脱敏处理
- 数据存储安全：采集的数据应妥善存储，防止数据泄露
- 访问控制：对存储的新闻数据实施适当的访问权限控制
- 数据销毁：研究完成后，涉及个人信息的数据应及时销毁
9.3 证券投资咨询资质要求
根据《证券法》和《证券投资顾问业务暂行规定》：
- 资质要求：从事证券投资咨询业务需取得中国证监会颁发的业务资格证书
- 禁止行为：未取得资质的个人或机构不得提供证券投资建议、投资分析等服务
- 本模块定位：本模块为纯技术工具，不提供任何投资建议，不构成证券投资咨询行为
- 用户责任：使用本模块进行任何投资相关研究的用户，应自行承担全部法律责任
9.4 不得用于非法证券活动
严禁将本模块用于以下非法活动：
- 利用内幕信息进行证券交易（内幕交易）
- 编造并传播虚假信息扰乱市场（操纵市场）
- 未取得资质提供证券投资咨询服务（非法荐股）
- 利用新闻数据进行抢先交易（front-running）
- 其他违反《证券法》《刑法》等相关法律法规的行为
十、免责声明与风险提示
10.1 免责声明
-
本文档及所有代码实现仅供技术学习和学术研究参考，不构成任何形式的投资建议、投资咨询或证券交易指导。
-
股市有风险，投资需谨慎。 任何基于本文档内容进行的投资决策，均由投资者自行承担全部风险和责任。
-
本文档中提及的任何数据源、工具、方法和策略仅为技术演示用途，不保证其准确性、完整性和时效性。
-
文档中的情感分析得分、研究因子、事件研究结果等输出，仅为NLP模型的计算结果，不代表任何投资价值的判断或未来走势的预测。
-
本文档不构成对任何金融产品或服务的推荐或邀约，不构成对任何投资策略有效性的承诺或保证。
10.2 风险提示
- 技术风险：数据采集可能因数据源接口变更、网络中断等原因失败，NLP模型输出可能存在偏差
- 数据风险：公开新闻数据可能存在延迟、错误或遗漏，不反映市场全貌
- 模型风险：情感分析、实体识别等NLP任务的准确率有限，可能产生错误结果
- 法律风险：使用公开数据进行研究须遵守相关法律法规，违规使用可能导致法律后果
- 市场风险：金融市场具有高度不确定性，历史数据和新闻舆情不预示未来表现
10.3 使用者承诺
使用本文档及代码时，使用者承诺：
- 已阅读并理解本免责声明的全部内容
- 仅将本文档用于合法的技术学习和学术研究
- 不会将本文档内容用于任何投资决策或证券交易活动
- 不会将本文档中的代码用于非法数据采集或其他违法活动
- 如因违规使用本文档内容而产生任何法律后果，由使用者自行承担
最后重申：本文档为纯技术学习文档，所有内容均以"技术分析工具框架"的形式呈现。文档中不包含任何具体个股推荐、买卖建议或收益承诺。请严格遵守相关法律法规，理性对待金融市场数据和技术工具。