105 lines
4.3 KiB
Python
105 lines
4.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""funnel_lexicon.py — 漏斗第一层确定性事件词表(spec §4.8 F1, P4-3).
|
|
|
|
范式照 sentiment_lexicon(子串零分词+可解释可审计);升级=词表加行非代码改.
|
|
否定前缀守卫(命中前 4 字窗 contains)拦「不减持/终止回购/已终止的质押误入
|
|
pledge_new」类误命中, 被拦命中降级 LLM 候选;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": ("立案", "行政处罚", "警示函", "监管函"),
|
|
}
|
|
|
|
# 否定/动作撤销前缀(命中词前 4 字窗口含此列 → 该命中作废, 降级 LLM 候选)
|
|
NEGATION_PREFIXES: tuple[str, ...] = ("不", "未", "终止", "取消", "解除")
|
|
|
|
# LLM 候选信号词(词表零命中但含任一 → 值得第二层看一眼)
|
|
SIGNAL_WORDS: tuple[str, ...] = (
|
|
"增持", "减持", "回购", "质押", "业绩", "预增", "预亏", "立案",
|
|
"处罚", "中标", "合同", "重组", "收购", "分红", "退市", "违规",
|
|
"担保", "诉讼", "仲裁", "冻结",
|
|
)
|
|
|
|
|
|
def _prefix_negated(text: str, i: int) -> bool:
|
|
"""P2-15(10-05): 命中点前 4 字窗 contains 否定词 —— 旧 2 字 endswith
|
|
只拦紧邻, 漏「已终止的质押」类间隔措辞(方向相反事件 confidence=1.0 落库)."""
|
|
window = text[max(0, i - 4):i]
|
|
return any(n in window for n in NEGATION_PREFIXES)
|
|
|
|
|
|
def _hit(text: str, kw: str) -> bool:
|
|
"""kw 命中且任一出现点前窗无否定词(「不减持」「解除质押」「已终止的质押」作废)."""
|
|
i = text.find(kw)
|
|
while i != -1:
|
|
if not _prefix_negated(text, i):
|
|
return True
|
|
i = text.find(kw, i + 1)
|
|
return False
|
|
|
|
|
|
def _negated_hit(text: str) -> bool:
|
|
"""存在被否定守卫拦下的词表命中 → 降级第二层 LLM 纠偏(P2-15: 不再静默丢)."""
|
|
for kws in LEXICON.values():
|
|
for kw in kws:
|
|
i = text.find(kw)
|
|
while i != -1:
|
|
if _prefix_negated(text, i):
|
|
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 ''}"
|
|
if _negated_hit(blob):
|
|
return True
|
|
return any(kw in blob for kw in SIGNAL_WORDS)
|