A股量化投研 - 新闻事件驱动模块技术规范
版本：v1.0 | 更新日期：2026-05-13
本文档定义了事件驱动策略中新闻获取、事件抽取、事件回测、预警订阅及信号化的完整技术规范。
1. 新闻获取规范
1.1 数据源优先级
| 优先级 | 数据源 | 说明 |
|---|---|---|
| P0 | 用户提供的新闻文件 | 用户自定义数据，优先级最高 |
| P1 | 公开API（财联社/东方财富/同花顺） | 标准化接口，数据质量可控 |
| P2 | 网页抓取 | 仅在合规前提下作为补充，需遵守 robots.txt |
1.2 API 接入示例
import akshare as ak
import pandas as pd
from pathlib import Path
def fetch_eastmoney_news() -> pd.DataFrame:
"""获取东方财富7x24快讯"""
df = ak.stock_info_global_em()
return df.rename(columns={"发布时间": "timestamp", "标题": "headline", "内容": "content", "来源": "source"})
def fetch_stock_news_em(stock_code: str) -> pd.DataFrame:
"""获取个股相关新闻（东方财富）"""
return ak.stock_news_em(symbol=stock_code)
def load_user_news(file_path: str) -> pd.DataFrame:
"""加载用户提供的新闻文件（CSV/Excel/JSON）"""
path = Path(file_path)
if path.suffix == ".csv": return pd.read_csv(path, parse_dates=["timestamp"])
elif path.suffix in (".xlsx", ".xls"): return pd.read_excel(path, parse_dates=["timestamp"])
elif path.suffix == ".json": return pd.read_json(path)
else: raise ValueError(f"不支持的文件格式: {path.suffix}")
1.3 去重规则
from difflib import SequenceMatcher
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
def deduplicate_news(df: pd.DataFrame, sim_threshold: float = 0.8, time_window_hours: int = 1) -> pd.DataFrame:
"""
新闻去重：标题相似度（编辑距离/余弦相似度 > 0.8 视为重复）+ 时间窗口（1小时内同主题合并）
"""
df = df.sort_values("timestamp").reset_index(drop=True)
if len(df) <= 1: return df
titles = df["headline"].tolist()
# TF-IDF余弦相似度
tfidf = TfidfVectorizer(char_wb_ngram_range=(2, 3)).fit_transform(titles)
sim_matrix = cosine_similarity(tfidf)
keep_mask = np.ones(len(df), dtype=bool)
time_window = pd.Timedelta(hours=time_window_hours)
for i in range(len(df)):
if not keep_mask[i]: continue
for j in range(i + 1, len(df)):
if not keep_mask[j]: continue
if (df.loc[j, "timestamp"] - df.loc[i, "timestamp"]) <= time_window and sim_matrix[i, j] > sim_threshold:
keep_mask[j] = False
return df[keep_mask].reset_index(drop=True)
1.4 实体对齐
# 股票代码映射表（维护证券简称变更与历史曾用名）
STOCK_NAME_MAP = {"中国平安": "601318.SH", "贵州茅台": "600519.SH"} # 持续维护
STOCK_ALIAS_MAP = {"匹凸匹": "600696.SH", "多伦股份": "600696.SH"} # 历史曾用名
INDUSTRY_MAP = {"银行": "801780.SI", "食品饮料": "801230.SI"} # 申万行业
CONCEPT_MAP = {"人工智能": "BK0800", "新能源车": "BK0477"} # 概念板块
def align_entities(text: str) -> dict:
"""从文本中提取并对齐实体（股票/行业/概念）"""
entities = {"stocks": [], "industries": [], "concepts": []}
for name, code in {**STOCK_NAME_MAP, **STOCK_ALIAS_MAP}.items():
if name in text: entities["stocks"].append({"name": name, "code": code})
for name, code in INDUSTRY_MAP.items():
if name in text: entities["industries"].append({"name": name, "code": code})
for name, code in CONCEPT_MAP.items():
if name in text: entities["concepts"].append({"name": name, "code": code})
return entities
2. 事件抽取与结构化
2.1 事件结构化输出 Schema（YAML）
event:
event_id: "EVT_20260513_001" # 唯一事件ID
timestamp: "2026-05-13T09:30:00+08:00" # 事件发生时间
source: "eastmoney" # 数据源
entities:
stocks: [{name: "贵州茅台", code: "600519.SH"}]
industries: [{name: "食品饮料", code: "801230.SI"}]
concepts: [{name: "白酒", code: "BK0478"}]
event_type: "业绩类" # 一级事件类型
event_subtype: "业绩预增" # 二级事件子类型
sentiment:
label: "positive" # positive / negative / neutral
confidence: 0.92 # 置信度 [0, 1]
intensity: 0.85 # 情绪强度 [0, 1]
content:
headline: "贵州茅台2026Q1净利润同比增长20%"
summary: "茅台发布一季报，营收同比增长18%，净利润同比增长20%，超市场预期。" # 100字以内
keywords: ["茅台", "一季报", "净利润", "超预期"]
impact_pathway:
channel: "基本面" # 基本面 / 资金面 / 情绪面 / 政策面
effect: "盈利上调"
magnitude: "high" # high / medium / low
action_hint:
type: "关注" # 关注 / 规避 / 核实 / 无
urgency: "medium" # high / medium / low
rationale: "业绩超预期可能引发分析师上调盈利预测"
2.2 事件类型清单
| 事件类型 | 子类型 | 典型情绪倾向 | 影响时效 | 说明 |
|---|---|---|---|---|
| 业绩类 | 业绩预增/预减、扭亏/首亏、财报超/低于预期 | 预增正面，预减负面 | 5-20日 | 对股价影响最直接 |
| 并购类 | 资产收购、股权出售、合并重组、要约收购 | 视溢价率而定 | 10-30日 | 需关注交易对价与协同效应 |
| 减持类 | 大股东减持、高管减持、减持计划公告 | 负面 | 3-10日 | 关注减持比例与动机 |
| 监管类 | 立案调查、行政处罚、问询函、监管函 | 负面 | 5-60日 | 严重程度差异大 |
| 政策类 | 行业扶持、产业限制、税收优惠、准入政策 | 视政策方向 | 10-60日 | 影响整个行业板块 |
| 事故类 | 安全事故、环保处罚、停产整顿 | 负面 | 3-20日 | 关注停产范围与持续时间 |
| 诉讼类 | 重大诉讼、仲裁、知识产权纠纷 | 负面 | 5-30日 | 关注涉案金额占净利润比例 |
| 分红类 | 高送转、特别分红、分红率提升、取消分红 | 正面/负面 | 3-10日 | 高送转短期效应明显 |
| 股权变动类 | 控股股东变更、股权激励、回购计划、举牌 | 视具体情况 | 5-30日 | 控股股东变更影响最大 |
| 人事变动类 | 董事长变更、CEO变更、核心高管离职 | 负面居多 | 3-10日 | 关注继任者背景 |
2.3 NLP 处理方案
基础方案 - 关键词匹配：
EVENT_KEYWORDS = {
"业绩类": {"业绩预增": ["净利润增长", "营收增长", "超预期"], "业绩预减": ["净利润下降", "亏损", "不及预期"]},
"减持类": {"大股东减持": ["减持", "减持计划"], "高管减持": ["高管减持", "董监高减持"]},
"监管类": {"立案调查": ["立案调查", "证监会立案"], "行政处罚": ["行政处罚", "罚款"]},
"并购类": {"资产收购": ["收购", "并购", "资产重组"], "股权出售": ["出售资产", "剥离资产"]},
}
SENTIMENT_LEXICON = {
"positive": ["增长", "提升", "超预期", "利好", "中标"],
"negative": ["下降", "亏损", "处罚", "调查", "减持", "退市"],
}
def extract_event_keyword(text: str) -> dict:
"""基于关键词匹配的事件抽取（基础方案）"""
result = {"event_type": None, "event_subtype": None, "sentiment": "neutral"}
for evt_type, subtypes in EVENT_KEYWORDS.items():
for subtype, keywords in subtypes.items():
if any(kw in text for kw in keywords):
result.update(event_type=evt_type, event_subtype=subtype); break
if result["event_type"]: break
pos = sum(1 for w in SENTIMENT_LEXICON["positive"] if w in text)
neg = sum(1 for w in SENTIMENT_LEXICON["negative"] if w in text)
result["sentiment"] = "positive" if pos > neg else "negative" if neg > pos else "neutral"
return result
进阶方案 - LLM 抽取：
import json
EVENT_EXTRACTION_PROMPT = """你是A股新闻事件抽取系统。从新闻中提取结构化事件信息。
新闻标题：{headline} 新闻内容：{content}
输出JSON：{{"event_type","event_subtype","entities":{{"stocks","industries","concepts"}},
"sentiment":{{"label","confidence","intensity"}},"summary"(100字内),"keywords",
"impact_pathway":{{"channel","effect","magnitude"}},"action_hint":{{"type","urgency","rationale"}}}}
只输出JSON。"""
def extract_event_llm(headline: str, content: str, llm_client) -> dict:
"""基于LLM的事件抽取（进阶方案）"""
response = llm_client.chat(EVENT_EXTRACTION_PROMPT.format(headline=headline, content=content))
try: return json.loads(response)
except json.JSONDecodeError: return {"error": "LLM输出解析失败", "raw": response}
情感分析模型（FinBERT微调）：
from transformers import pipeline
class FinancialSentimentAnalyzer:
def __init__(self, model_name: str = "bert-base-chinese"):
self.classifier = pipeline("sentiment-analysis", model=model_name, return_all_scores=True)
def analyze(self, text: str) -> dict:
results = self.classifier(text)[0]
scores = {r["label"]: r["score"] for r in results}
best = max(scores, key=scores.get)
return {"label": best, "confidence": scores[best], "scores": scores}
3. 事件回测框架（事件研究法）
3.1 完整事件研究法 Python 实现
import numpy as np
import pandas as pd
from scipy import stats
class EventStudy:
"""
事件研究法：估计窗口T-120至T-11，事件窗口T-10至T+10，正常收益率模型为市场模型（CAPM）
AR = 实际收益 - 正常收益；CAR = 事件窗口内AR求和；t检验判断CAR是否显著异于零
"""
def __init__(self, estimation_window: int = 120, event_window_pre: int = 10, event_window_post: int = 10):
self.est_window, self.pre_window, self.post_window = estimation_window, event_window_pre, event_window_post
def estimate_market_model(self, stock_returns: pd.Series, market_returns: pd.Series, event_date: pd.Timestamp) -> dict:
"""估计窗口：事件日前120日至前11日，OLS回归得到alpha/beta/sigma"""
est_start = event_date - pd.Timedelta(days=self.est_window + self.pre_window + 30)
est_end = event_date - pd.Timedelta(days=self.pre_window + 5)
common_idx = stock_returns.index.intersection(market_returns.index)
est_idx = common_idx[(common_idx >= est_start) & (common_idx <= est_end)]
if len(est_idx) < 60: raise ValueError(f"估计窗口数据不足: {len(est_idx)} < 60")
y, X = stock_returns.loc[est_idx].values, market_returns.loc[est_idx].values
beta_vec = np.linalg.lstsq(np.column_stack([np.ones(len(X)), X]), y, rcond=None)[0]
residuals = y - (beta_vec[0] + beta_vec[1] * X)
return {"alpha": beta_vec[0], "beta": beta_vec[1], "sigma": np.std(residuals, ddof=2)}
def calculate_abnormal_returns(self, stock_returns: pd.Series, market_returns: pd.Series,
event_date: pd.Timestamp, params: dict) -> pd.DataFrame:
"""计算事件窗口AR和CAR"""
window_start = event_date - pd.Timedelta(days=self.pre_window + 15)
window_end = event_date + pd.Timedelta(days=self.post_window + 15)
common_idx = stock_returns.index.intersection(market_returns.index)
evt_idx = common_idx[(common_idx >= window_start) & (common_idx <= window_end)]
if len(evt_idx) == 0: raise ValueError("事件窗口内无数据")
actual = stock_returns.loc[evt_idx].values
predicted = params["alpha"] + params["beta"] * market_returns.loc[evt_idx].values
ar = actual - predicted
# 计算相对事件日位置
trading_days = sorted(common_idx)
event_pos = trading_days.index(event_date) if event_date in trading_days else np.searchsorted(trading_days, event_date)
positions = np.arange(len(evt_idx)) - (np.searchsorted(trading_days, evt_idx) - event_pos)
result = pd.DataFrame({"date": evt_idx, "relative_day": positions, "AR": ar})
event_mask = (positions >= -self.pre_window) & (positions <= self.post_window)
result["CAR"] = np.nan
result.loc[event_mask, "CAR"] = np.cumsum(result.loc[event_mask, "AR"].values)
return result
def statistical_test(self, ar_series: pd.Series, sigma: float, n_est: int) -> dict:
"""t检验：H0: CAR=0, H1: CAR!=0"""
car = ar_series.sum()
t_stat = car / (sigma * np.sqrt(len(ar_series)))
p_value = 2 * (1 - stats.t.cdf(abs(t_stat), df=n_est - 2))
return {"CAR": car, "AR_mean": ar_series.mean(), "t_statistic": t_stat, "p_value": p_value,
"significant_at_5pct": p_value < 0.05, "significant_at_1pct": p_value < 0.01}
def run_single_event(self, stock_returns: pd.Series, market_returns: pd.Series, event_date: pd.Timestamp) -> dict:
"""对单个事件运行完整事件研究，返回AR/CAR/统计检验/T+N日CAR"""
params = self.estimate_market_model(stock_returns, market_returns, event_date)
ar_df = self.calculate_abnormal_returns(stock_returns, market_returns, event_date, params)
event_mask = (ar_df["relative_day"] >= -self.pre_window) & (ar_df["relative_day"] <= self.post_window)
test_result = self.statistical_test(ar_df.loc[event_mask, "AR"], params["sigma"], self.est_window)
car_points = {}
for n in [1, 3, 5, 10]:
mask_n = (ar_df["relative_day"] >= 0) & (ar_df["relative_day"] <= n)
if mask_n.sum() > 0: car_points[f"CAR_T+{n}"] = ar_df.loc[mask_n, "AR"].sum()
return {"event_date": event_date, "model_params": params, "daily_ar": ar_df,
"test_result": test_result, "car_points": car_points}
3.2 批量回测与分层分析
class BatchEventStudy:
"""批量事件研究：对多只股票的多个事件进行汇总分析"""
def __init__(self, event_study: EventStudy): self.es = event_study
def run_batch(self, events: list[dict], return_data: dict, market_returns: pd.Series) -> pd.DataFrame:
"""events: [{"stock_code","event_date","event_type"}], return_data: {code: Series}"""
results = []
for evt in events:
code, date = evt["stock_code"], evt["event_date"]
if code not in return_data: continue
try:
single = self.es.run_single_event(return_data[code], market_returns, date)
results.append({"stock_code": code, "event_type": evt.get("event_type", ""),
"event_date": date, **single["test_result"], **single["car_points"]})
except Exception as e: print(f"失败: {code} {date} - {e}")
return pd.DataFrame(results)
def aggregate_aar_car(self, results_df: pd.DataFrame) -> pd.DataFrame:
"""计算平均异常收益率（AAR）和平均累计异常收益率（CAAR）"""
return results_df.groupby("event_type").agg(
sample_size=("stock_code", "count"), mean_CAR=("CAR", "mean"),
median_CAR=("CAR", "median"), pct_positive=("CAR", lambda x: (x > 0).mean()),
mean_t_stat=("t_statistic", "mean"), pct_significant_5pct=("significant_at_5pct", "mean")).round(4)
def stratified_analysis(results_df: pd.DataFrame, market_cap_data: pd.Series) -> dict:
"""分层分析：按事件类型、市值（大/中/小盘）、市场状态（牛/熊/震荡）分层"""
q = market_cap_data.quantile([0.33, 0.66])
results_df["cap_group"] = results_df["stock_code"].apply(
lambda c: "大盘" if market_cap_data.get(c, 0) >= q[0.66] else "中盘" if market_cap_data.get(c, 0) >= q[0.33] else "小盘")
return {
"by_market_cap": results_df.groupby("cap_group").agg(
sample_size=("CAR", "count"), mean_CAR=("CAR", "mean"), pct_significant=("significant_at_5pct", "mean")).round(4),
"by_type_and_cap": results_df.groupby(["event_type", "cap_group"]).agg(
sample_size=("CAR", "count"), mean_CAR=("CAR", "mean"), pct_significant=("significant_at_5pct", "mean")).round(4)}
def decay_analysis(results_df: pd.DataFrame) -> pd.DataFrame:
"""事件效应衰减分析：T+1/T+3/T+5/T+10的CAR变化"""
decay_cols = [c for c in results_df.columns if c.startswith("CAR_T+")]
if not decay_cols: return pd.DataFrame()
decay_summary = results_df.groupby("event_type")[decay_cols].agg(["mean", "median", "std", "count"]).round(4)
if "CAR_T+1" in decay_cols and "CAR_T+10" in decay_cols:
decay_summary["decay_ratio"] = (decay_summary[("CAR_T+10", "mean")] / decay_summary[("CAR_T+1", "mean")]).round(4)
return decay_summary
4. 事件效应报告模板
4.1 报告生成函数
def generate_event_report(results_df: pd.DataFrame, stratified: dict, decay_df: pd.DataFrame) -> str:
"""生成事件效应研究报告，包含事件统计、异质性分析、可交易性评估、结论"""
r = []
r.append("=" * 60 + "\n 事件驱动策略 - 事件效应研究报告\n" + "=" * 60)
# 一、事件统计
r.append(f"\n一、事件统计总览\n{'-'*40}\n总样本量: {len(results_df)} 事件类型数: {results_df['event_type'].nunique()}"
f"\n平均事件日CAR: {results_df['CAR'].mean():.4f} CAR为正比例: {(results_df['CAR']>0).mean():.2%}"
f"\n5%显著性比例: {results_df['significant_at_5pct'].mean():.2%}")
# 二、分事件类型统计
type_summary = results_df.groupby("event_type").agg(N=("CAR","count"), CAR_mean=("CAR","mean"),
CAR_median=("CAR","median"), t_stat=("t_statistic","mean"), p_val=("p_value","mean"),
sig_pct=("significant_at_5pct","mean")).round(4)
r.append(f"\n二、分事件类型统计\n{'-'*40}\n{type_summary.to_string()}")
# 三、异质性分析
if "by_market_cap" in stratified:
r.append(f"\n三、异质性分析（按市值分层）\n{'-'*40}\n{stratified['by_market_cap'].to_string()}")
# 四、衰减分析
if not decay_df.empty:
r.append(f"\n四、事件效应衰减分析\n{'-'*40}\n{decay_df.to_string()}")
# 五、可交易性评估
r.append(f"\n五、可交易性评估\n{'-'*40}")
for evt_type in results_df["event_type"].unique():
sub = results_df[results_df["event_type"] == evt_type]
sig_pct, car_mean = sub["significant_at_5pct"].mean(), sub["CAR"].mean()
if sig_pct > 0.3 and abs(car_mean) > 0.01: trad, hold = "高", "T+3至T+5"
elif sig_pct > 0.15 and abs(car_mean) > 0.005: trad, hold = "中", "T+1至T+3"
else: trad, hold = "低（接近噪声）", "不建议基于此事件交易"
r.append(f" {evt_type}: 可交易性={trad} 建议持有期={hold}")
# 六、结论
sig_evts = results_df.groupby("event_type")["significant_at_5pct"].mean()
r.append(f"\n六、结论\n{'-'*40}")
r.append(f" 统计显著: {', '.join(sig_evts[sig_evts>0.2].index.tolist()) or '无'}")
r.append(f" 基本噪声: {', '.join(sig_evts[sig_evts<=0.2].index.tolist()) or '无'}")
r.append("\n" + "=" * 60 + "\n免责声明：本报告仅供研究参考，不构成投资建议。\n" + "=" * 60)
return "\n".join(r)
4.2 报告输出示例
============================================================
事件驱动策略 - 事件效应研究报告
============================================================
一、事件统计总览
----------------------------------------
总样本量: 1,256 事件类型数: 10
平均事件日CAR: 0.0085 CAR为正比例: 58.20%
5%显著性比例: 28.50%
二、分事件类型统计
----------------------------------------
N CAR_mean CAR_median t_stat p_val sig_pct
event_type
业绩类 320 0.0152 0.0123 2.150 0.038 0.4200
并购类 185 0.0098 0.0065 1.560 0.120 0.2500
减持类 210 -0.0125 -0.0098 -2.030 0.044 0.3800
监管类 145 -0.0180 -0.0150 -2.580 0.010 0.5200
三、异质性分析（按市值分层）
----------------------------------------
sample_size mean_CAR pct_significant
cap_group
大盘 420 0.0042 0.2200
中盘 415 0.0098 0.3100
小盘 421 0.0115 0.3300
五、可交易性评估
----------------------------------------
业绩类: 可交易性=高 建议持有期=T+3至T+5
监管类: 可交易性=高 建议持有期=T+3至T+5
减持类: 可交易性=高 建议持有期=T+1至T+3
六、结论
----------------------------------------
统计显著: 业绩类, 监管类, 减持类, 并购类
基本噪声: 人事变动类, 分红类
============================================================
免责声明：本报告仅供研究参考，不构成投资建议。
============================================================
5. 预警与订阅系统
5.1 预警规则配置（YAML）
alert_rules:
watchlist_major_events:
name: "自选股重大事件预警"
enabled: true
conditions:
- event_type: "业绩类"
event_subtype: ["业绩预减", "首亏", "大幅下滑"]
min_sentiment_intensity: 0.6
- event_type: "监管类"
event_subtype: ["立案调查"]
- event_type: "诉讼类"
event_subtype: ["重大诉讼"]
min_impact_magnitude: "medium"
scope: {type: "watchlist"}
output: {urgency: "high"}
portfolio_risk_events:
name: "持仓风险事件预警"
enabled: true
conditions:
- event_type: "减持类"
event_subtype: ["大股东减持", "高管减持"]
- event_type: "监管类"
event_subtype: ["ST风险", "退市整理"]
- event_type: "事故类"
event_subtype: ["停产整顿"]
scope: {type: "portfolio"}
output: {urgency: "high"}
industry_policy_changes:
name: "行业政策变化预警"
enabled: true
conditions:
- event_type: "政策类"
channel: "政策面"
min_impact_magnitude: "medium"
scope: {type: "industry", industries: ["新能源", "半导体", "医药生物", "房地产"]}
output: {urgency: "medium"}
macro_inflection:
name: "宏观拐点预警"
enabled: true
conditions:
- indicator: "LPR", change_threshold: 0.10 # 变动超过10BP
- indicator: "PMI", change_threshold: 1.0, direction: "cross_50" # 穿越荣枯线
- indicator: "USD_CNY", change_threshold: 0.02 # 变动超过200BP
scope: {type: "macro"}
output: {urgency: "medium"}
5.2 预警输出与引擎
import uuid
from datetime import datetime
from collections import defaultdict
def generate_alert(rule_name: str, trigger_event: dict, urgency: str = "medium") -> dict:
"""生成预警：alert_id, timestamp, urgency, trigger, details, impact_assessment, action_hint"""
return {
"alert_id": f"ALT_{datetime.now().strftime('%Y%m%d%H%M%S')}_{uuid.uuid4().hex[:6]}",
"timestamp": datetime.now().isoformat(), "urgency": urgency, "trigger": rule_name,
"details": {"event_type": trigger_event.get("event_type"), "event_subtype": trigger_event.get("event_subtype"),
"headline": trigger_event.get("headline"),
"affected_stocks": trigger_event.get("entities", {}).get("stocks", [])},
"impact_assessment": {"sentiment": trigger_event.get("sentiment", {}).get("label"),
"magnitude": trigger_event.get("impact_pathway", {}).get("magnitude"),
"channel": trigger_event.get("impact_pathway", {}).get("channel")},
"action_hint": {"type": trigger_event.get("action_hint", {}).get("type", "关注"),
"rationale": trigger_event.get("action_hint", {}).get("rationale", "")},
"disclaimer": "本预警仅为信息提示，不构成投资建议。投资决策请基于独立判断。"}
class AlertEngine:
"""预警引擎：根据规则配置检测事件并生成预警"""
def __init__(self, rules_config: dict):
self.rules = rules_config.get("alert_rules", {})
self._last_alert = defaultdict(lambda: None) # 频率控制
def check_event(self, event: dict, context: dict = None) -> list[dict]:
alerts = []
for rule_name, rule in self.rules.items():
if not rule.get("enabled", True): continue
if not self._check_scope(rule, event, context): continue
if self._check_conditions(rule["conditions"], event):
# 频率控制：同一标的同一类型24小时内最多1次
stocks = [s["code"] for s in event.get("entities", {}).get("stocks", [])]
key = (stocks[0] if stocks else "", event.get("event_type", ""))
if self._last_alert[key] and (datetime.now() - self._last_alert[key]).total_seconds() < 86400:
continue
self._last_alert[key] = datetime.now()
alerts.append(generate_alert(rule_name, event, rule.get("output", {}).get("urgency", "medium")))
return alerts
def _check_scope(self, rule: dict, event: dict, context: dict) -> bool:
scope = rule.get("scope", {})
st = scope.get("type")
event_stocks = [s["code"] for s in event.get("entities", {}).get("stocks", [])]
if st == "watchlist" and context: return bool(set(event_stocks) & set(context.get("watchlist", [])))
if st == "portfolio" and context: return bool(set(event_stocks) & set(context.get("portfolio", [])))
if st == "industry":
return bool(set(i["name"] for i in event.get("entities", {}).get("industries", [])) & set(scope.get("industries", [])))
return True
def _check_conditions(self, conditions: list, event: dict) -> bool:
for cond in conditions:
if cond.get("event_type") and event.get("event_type") != cond["event_type"]: continue
if cond.get("event_subtype"):
subs = cond["event_subtype"] if isinstance(cond["event_subtype"], list) else [cond["event_subtype"]]
if event.get("event_subtype") not in subs: continue
if cond.get("min_sentiment_intensity") and event.get("sentiment", {}).get("intensity", 0) < cond["min_sentiment_intensity"]: continue
return True
return False
5.3 预警不越界原则
核心原则：预警系统只做信息提示，不构成投资建议。
- 所有预警输出必须包含免责声明
action_hint.type仅限：关注、规避、核实、无，不得使用买入、卖出等投资操作指令action_hint.rationale仅描述事件本身的逻辑影响，不给出目标价或仓位建议- 预警频率控制：同一标的同一类型事件24小时内最多触发1次预警
6. 事件信号化
6.1 事件得分计算
# 事件类型权重（基于历史回测结果校准）
EVENT_TYPE_WEIGHTS = {
"业绩类": {"positive": 1.2, "negative": 1.3}, "并购类": {"positive": 0.8, "negative": 0.7},
"减持类": {"positive": 0.3, "negative": 1.1}, "监管类": {"positive": 0.2, "negative": 1.4},
"政策类": {"positive": 1.0, "negative": 0.9}, "事故类": {"positive": 0.1, "negative": 1.2},
"诉讼类": {"positive": 0.2, "negative": 1.0}, "分红类": {"positive": 0.6, "negative": 0.5},
"股权变动类": {"positive": 0.9, "negative": 0.8}, "人事变动类": {"positive": 0.4, "negative": 0.6},
}
def calculate_event_score(event: dict) -> float:
"""事件得分 = 情绪得分 x 置信度 x 事件类型权重，返回 [-1.5, 1.5]"""
sentiment = event.get("sentiment", {})
label, confidence, intensity = sentiment.get("label", "neutral"), sentiment.get("confidence", 0.5), sentiment.get("intensity", 0.5)
sentiment_score = {"positive": 1.0, "negative": -1.0, "neutral": 0.0}.get(label, 0.0)
weight = EVENT_TYPE_WEIGHTS.get(event.get("event_type", ""), {"positive": 0.5, "negative": 0.5}).get(label, 0.5)
return round(sentiment_score * confidence * weight * (0.5 + 0.5 * intensity), 4)
6.2 事件信号与量价信号结合
def combine_event_quant_signals(event_score: float, momentum_score: float, volume_score: float, event_weight: float = 0.4) -> float:
"""综合得分 = event_weight * 事件得分(归一化) + (1-event_weight) * 量价综合得分，返回 [-1, 1]"""
norm_event = np.clip(event_score / 1.5, -1, 1)
quant_score = 0.6 * momentum_score + 0.4 * volume_score
return round(event_weight * norm_event + (1 - event_weight) * quant_score, 4)
6.3 事件冷却期
from collections import defaultdict
class EventCooldown:
"""冷却期管理：同一标的同一类型事件在N个自然日内不重复触发信号"""
COOLDOWN_CONFIG = {"业绩类": 10, "减持类": 5, "监管类": 7, "政策类": 3, "并购类": 15,
"事故类": 7, "诉讼类": 10, "分红类": 10, "股权变动类": 7, "人事变动类": 5}
def __init__(self, default_days: int = 5):
self.default = default_days
self._last_trigger = defaultdict(lambda: None)
def check_and_trigger(self, stock_code: str, event_type: str, current_date: pd.Timestamp) -> bool:
key = (stock_code, event_type)
last = self._last_trigger[key]
if last is None or (current_date - last).days >= self.COOLDOWN_CONFIG.get(event_type, self.default):
self._last_trigger[key] = current_date
return True
return False
6.4 事件信号生成完整流程
class EventSignalGenerator:
"""从原始事件到可交易信号的完整流程：计算得分 -> 冷却期检查 -> 生成信号"""
HOLDING_MAP = {"业绩类": 5, "减持类": 3, "监管类": 5, "政策类": 10, "并购类": 10,
"事故类": 3, "诉讼类": 5, "分红类": 3, "股权变动类": 5, "人事变动类": 3}
def __init__(self): self.cooldown = EventCooldown()
def generate_signal(self, event: dict, current_date: pd.Timestamp) -> list[dict] | None:
score = calculate_event_score(event)
if abs(score) < 0.1: return None # 得分过低，无信号
stocks = event.get("entities", {}).get("stocks", [])
if not stocks: return None
signals = []
for stock in stocks:
code, evt_type = stock["code"], event.get("event_type", "")
if not self.cooldown.check_and_trigger(code, evt_type, current_date): continue
signals.append({
"signal_id": f"SIG_{current_date.strftime('%Y%m%d')}_{code}",
"date": current_date.isoformat(), "stock_code": code, "stock_name": stock.get("name", ""),
"direction": "long" if score > 0 else "short", "event_score": score,
"event_type": evt_type, "event_subtype": event.get("event_subtype", ""),
"sentiment": event.get("sentiment", {}).get("label"),
"confidence": event.get("sentiment", {}).get("confidence"),
"headline": event.get("content", {}).get("headline", ""),
"holding_days": self.HOLDING_MAP.get(evt_type, 5)})
return signals or None
文档维护说明：本规范应随策略迭代持续更新。事件类型权重、冷却期参数等需基于最新回测结果定期校准。所有代码示例均经过类型标注，可直接集成至量化投研框架中。
上述配置只作为个人学习参考示例，本文是作为本人技术学习记录，不代表本人投资建议，股市有风险投资需谨慎