feat(data): 漏斗第一层词表 funnel_lexicon——八类事件+否定前缀守卫, sentiment_lexicon 范式 [nas] [no-doc]

This commit is contained in:
2026-10-02 09:51:59 +08:00
parent 7b489d1606
commit 8bf935b1a0
2 changed files with 147 additions and 0 deletions
+83
View File
@@ -0,0 +1,83 @@
# -*- coding: utf-8 -*-
"""funnel_lexicon.py — 漏斗第一层确定性事件词表(spec §4.8 F1, P4-3).
范式照 sentiment_lexicon(子串零分词+可解释可审计);升级=词表加行非代码改.
否定前缀守卫拦「不减持/终止回购/解除质押误入 pledge_new」类误命中;v1 接受
残余噪声换确定性,方向=分类学固有先验不做上下文翻转(ambiguous 边界判给
第二层 LLM, 见 corpus_funnel.is_llm_candidate).
"""
from __future__ import annotations
# 事件分类学 pilot 八类(code 稳定供下游枚举校验;扩充=加行)
EVENT_TYPES: dict[str, str] = {
"holder_increase": "股东增持",
"holder_decrease": "股东减持",
"buyback": "回购",
"pledge_new": "新增质押",
"pledge_release": "解除质押",
"earnings_pre_up": "业绩预增",
"earnings_pre_down": "业绩预减或预亏",
"investigation": "立案调查或处罚",
}
# 方向先验(pos/neg), 分类学固有
EVENT_DIRECTION: dict[str, str] = {
"holder_increase": "pos", "holder_decrease": "neg", "buyback": "pos",
"pledge_new": "neg", "pledge_release": "pos", "earnings_pre_up": "pos",
"earnings_pre_down": "neg", "investigation": "neg",
}
# 词表(命中即该类型;否定前缀守卫在 _hit)
LEXICON: dict[str, tuple[str, ...]] = {
"pledge_release": ("解除质押", "解押"),
"holder_increase": ("拟增持", "增持计划", "股东增持", "继续增持",
"增持公司股份", "累计增持"),
"holder_decrease": ("拟减持", "减持计划", "股东减持", "继续减持",
"减持公司股份", "累计减持"),
"buyback": ("回购股份", "回购公司股份", "回购报告书", "回购实施"),
"pledge_new": ("质押",),
"earnings_pre_up": ("业绩预增", "预增", "扭亏"),
"earnings_pre_down": ("业绩预减", "预亏", "预减", "首亏", "续亏"),
"investigation": ("立案", "行政处罚", "警示函", "监管函"),
}
# 否定/动作撤销前缀(命中词前 2 字窗口含此列 → 该命中作废)
NEGATION_PREFIXES: tuple[str, ...] = ("不", "未", "终止", "取消", "解除")
# LLM 候选信号词(词表零命中但含任一 → 值得第二层看一眼)
SIGNAL_WORDS: tuple[str, ...] = (
"增持", "减持", "回购", "质押", "业绩", "预增", "预亏", "立案",
"处罚", "中标", "合同", "重组", "收购", "分红", "退市", "违规",
"担保", "诉讼", "仲裁", "冻结",
)
def _hit(text: str, kw: str) -> bool:
"""kw 命中且前 2 字窗口不含否定前缀(「不减持」「解除质押」里的裸命中作废)."""
i = text.find(kw)
while i != -1:
prefix = text[max(0, i - 2):i]
if not any(prefix.endswith(n) for n in NEGATION_PREFIXES):
return True
i = text.find(kw, i + 1)
return False
def match_events(title: str, text: str | None = None) -> list[dict]:
"""词表命中(类型去重) → [{event_type, direction, confidence: 1.0}]."""
blob = f"{title or ''}\n{text or ''}"
out = []
for etype, kws in LEXICON.items():
if any(_hit(blob, kw) for kw in kws):
out.append({"event_type": etype,
"direction": EVENT_DIRECTION[etype],
"confidence": 1.0})
return out
def is_llm_candidate(title: str, text: str | None = None) -> bool:
"""词表零命中但含信号词 → 第二层候选(简单子串无守卫, 边界判给 LLM)."""
if match_events(title, text):
return False
blob = f"{title or ''}\n{text or ''}"
return any(kw in blob for kw in SIGNAL_WORDS)