fix: fail closed on generic political entity matches

docs/free_source_setup.md

@@ -9,6 +9,20 @@
 - `data/live/political_watchlist.csv`: initial watchlist for the live publish path.
 - `data/live/source_items.csv`: latest raw text pulled by the scheduled RSS pipeline.
 - `data/live/source_events.csv`: events extracted deterministically from `source_items.csv`.
+
+`source_events.csv` keeps the original columns and adds the following compatible
+evidence columns: `entity_match_type`, `match_evidence`, and
+`relationship_type`. Values are `issuer`, `direct_beneficiary`,
+`industry_context`, or `unverified`. Only the first two are company-level
+evidence; industry vocabulary is retained for auditability but is not scored as
+an issuer event. Rows without a verifiable entity match are not emitted by the
+live extractor. Legacy imported rows default to `unverified` and therefore fail
+closed in downstream company-event scoring.
+
+Historical event-study compatibility is opt-in only. Use
+`--historical-compatibility --compatibility-reason "..."` for a pre-schema CSV.
+Rows remain `unverified`; output records include `compatibility_used`,
+`compatibility_reason`, and `legacy_provenance`.
 - `data/live/political_events.csv`: stable Advisor input, refreshed by RSS/source pipeline or maintained after manual review.
 - `data/live/source_tracker.csv`: merged watchlist and event tracker.
 

docs/free_source_setup.zh-CN.md

@@ -9,6 +9,17 @@
 - `data/live/political_watchlist.csv`：真实发布链路的初始观察池。
 - `data/live/source_items.csv`：定时 RSS pipeline 最近一次拉取的公开原始文本。
 - `data/live/source_events.csv`：从 `source_items.csv` 确定性抽取出的事件。
+
+`source_events.csv` 保留原有字段，并增加兼容字段：
+`entity_match_type`、`match_evidence`、`relationship_type`。关系值为
+`issuer`、`direct_beneficiary`、`industry_context` 或 `unverified`。
+只有前两类可作为公司级证据；行业通用词会保留用于审计，但不会进入公司事件评分。
+live 抽取器不会输出无法验证实体关联的行；旧格式导入行默认为
+`unverified`，下游公司事件评分会 fail-closed。
+历史 event study 兼容模式必须显式启用并提供原因：
+`--historical-compatibility --compatibility-reason "..."`。旧行仍保持
+`unverified`，输出记录 `compatibility_used`、`compatibility_reason` 和
+`legacy_provenance`，不会自动升级为 verified。
 - `data/live/political_events.csv`：Advisor 稳定读取的真实事件输入，由 RSS/source pipeline 刷新，也可以人工核验后维护。
 - `data/live/source_tracker.csv`：观察池与事件合并后的 tracker。
 

src/political_event_tracking_research/event_study.py

@@ -18,6 +18,12 @@ class Event:
     confidence: str
     source_url: str
     notes: str
+    entity_match_type: str = "unverified"
+    match_evidence: str = ""
+    relationship_type: str = "unverified"
+    legacy_compatibility: bool = False
+    compatibility_reason: str = ""
+    legacy_provenance: str = ""
 
 
 @dataclass(frozen=True)
@@ -35,6 +41,9 @@ class EventReturn:
     benchmark_symbol: str
     benchmark_return_pct: float | None
     abnormal_return_pct: float | None
+    compatibility_used: bool = False
+    compatibility_reason: str = ""
+    legacy_provenance: str = ""
 
     def to_row(self) -> dict[str, object]:
         return {
@@ -51,19 +60,28 @@ def to_row(self) -> dict[str, object]:
             "benchmark_symbol": self.benchmark_symbol,
             "benchmark_return_pct": "" if self.benchmark_return_pct is None else f"{self.benchmark_return_pct:.6f}",
             "abnormal_return_pct": "" if self.abnormal_return_pct is None else f"{self.abnormal_return_pct:.6f}",
+            "compatibility_used": str(self.compatibility_used).lower(),
+            "compatibility_reason": self.compatibility_reason,
+            "legacy_provenance": self.legacy_provenance,
         }
 
 
 PriceTable = dict[str, dict[date, float]]
+VERIFIED_RELATIONSHIPS = frozenset({"issuer", "direct_beneficiary"})
 
 
 def parse_date(value: str) -> date:
     return date.fromisoformat(value.strip())
 
 
-def load_events(path: str | Path) -> list[Event]:
+def load_events(path: str | Path, historical_compatibility: bool = False, compatibility_reason: str = "") -> list[Event]:
     events: list[Event] = []
-    for row in read_csv_rows(path):
+    rows = read_csv_rows(path)
+    legacy_schema = bool(rows) and "relationship_type" not in rows[0]
+    if historical_compatibility and not compatibility_reason.strip():
+        raise ValueError("compatibility_reason is required for historical compatibility")
+    for row in rows:
+        legacy_compatibility = historical_compatibility and legacy_schema
         events.append(
             Event(
                 event_id=row["event_id"],
@@ -74,6 +92,12 @@ def load_events(path: str | Path) -> list[Event]:
                 confidence=row.get("confidence", ""),
                 source_url=row.get("source_url", ""),
                 notes=row.get("notes", ""),
+                entity_match_type=row.get("entity_match_type", "unverified") or "unverified",
+                match_evidence=row.get("match_evidence", ""),
+                relationship_type=row.get("relationship_type", "unverified") or "unverified",
+                legacy_compatibility=legacy_compatibility,
+                compatibility_reason=compatibility_reason if legacy_compatibility else "",
+                legacy_provenance=str(path) if legacy_compatibility else "",
             )
         )
     return events
@@ -123,11 +147,15 @@ def compute_event_returns(
     prices: PriceTable,
     windows: tuple[int, ...] = (1, 5, 20),
     benchmark_symbol: str = "SPY",
+    historical_compatibility: bool = False,
 ) -> list[EventReturn]:
     results: list[EventReturn] = []
     benchmark_symbol = benchmark_symbol.upper()
 
     for event in events:
+        use_legacy = historical_compatibility and event.legacy_compatibility
+        if event.relationship_type not in VERIFIED_RELATIONSHIPS and not use_legacy:
+            continue
         symbol_dates = sorted_dates(prices, event.symbol)
         base_date = first_trading_date_on_or_after(symbol_dates, event.event_date)
         if base_date is None:
@@ -166,6 +194,9 @@ def compute_event_returns(
                     benchmark_symbol=benchmark_symbol,
                     benchmark_return_pct=benchmark_return,
                     abnormal_return_pct=abnormal_return,
+                    compatibility_used=use_legacy,
+                    compatibility_reason=event.compatibility_reason if use_legacy else "",
+                    legacy_provenance=event.legacy_provenance if use_legacy else "",
                 )
             )
 
@@ -187,10 +218,18 @@ def run_event_study(
     output_path: str | Path,
     windows: tuple[int, ...] = (1, 5, 20),
     benchmark_symbol: str = "SPY",
+    historical_compatibility: bool = False,
+    compatibility_reason: str = "",
 ) -> list[EventReturn]:
-    events = load_events(events_path)
+    events = load_events(events_path, historical_compatibility=historical_compatibility, compatibility_reason=compatibility_reason)
     prices = load_prices(prices_path)
-    results = compute_event_returns(events, prices, windows=windows, benchmark_symbol=benchmark_symbol)
+    results = compute_event_returns(
+        events,
+        prices,
+        windows=windows,
+        benchmark_symbol=benchmark_symbol,
+        historical_compatibility=historical_compatibility,
+    )
     write_csv_rows(
         output_path,
         [
@@ -207,6 +246,9 @@ def run_event_study(
             "benchmark_symbol",
             "benchmark_return_pct",
             "abnormal_return_pct",
+            "compatibility_used",
+            "compatibility_reason",
+            "legacy_provenance",
         ],
         [result.to_row() for result in results],
     )
@@ -220,6 +262,8 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--output", required=True, help="Output CSV path.")
     parser.add_argument("--windows", default="1,5,20", help="Comma-separated trading-day offsets.")
     parser.add_argument("--benchmark", default="SPY", help="Benchmark symbol for abnormal returns.")
+    parser.add_argument("--historical-compatibility", action="store_true", help="Explicitly analyze legacy event CSVs.")
+    parser.add_argument("--compatibility-reason", default="", help="Required audit reason for legacy compatibility.")
     return parser
 
 
@@ -231,6 +275,8 @@ def main(argv: list[str] | None = None) -> None:
         output_path=args.output,
         windows=parse_windows(args.windows),
         benchmark_symbol=args.benchmark,
+        historical_compatibility=args.historical_compatibility,
+        compatibility_reason=args.compatibility_reason,
     )
 
 

src/political_event_tracking_research/official_event_import.py

@@ -45,6 +45,9 @@ class OfficialRecord:
     direction: str
     source_url: str
     summary: str
+    entity_match_type: str = "unverified"
+    match_evidence: str = ""
+    relationship_type: str = "unverified"
 
 
 def slug(value: str) -> str:
@@ -76,6 +79,9 @@ def load_official_records(path: str | Path) -> list[OfficialRecord]:
                 direction=row.get("direction", ""),
                 source_url=row["source_url"],
                 summary=row.get("summary", ""),
+                entity_match_type=row.get("entity_match_type", "unverified") or "unverified",
+                match_evidence=row.get("match_evidence", ""),
+                relationship_type=row.get("relationship_type", "unverified") or "unverified",
             )
         )
     return records
@@ -87,6 +93,12 @@ def validate_record(record: OfficialRecord) -> None:
         raise ValueError("record_id is required")
     if not record.symbol.strip():
         raise ValueError(f"{record.record_id}: symbol is required")
+    if record.entity_match_type not in {"issuer", "direct_beneficiary", "industry_context", "unverified"}:
+        raise ValueError(f"{record.record_id}: unsupported entity_match_type {record.entity_match_type!r}")
+    if record.relationship_type not in {"issuer", "direct_beneficiary", "industry_context", "unverified"}:
+        raise ValueError(f"{record.record_id}: unsupported relationship_type {record.relationship_type!r}")
+    if record.entity_match_type != record.relationship_type:
+        raise ValueError(f"{record.record_id}: entity_match_type and relationship_type must agree")
     if record.event_type not in ALLOWED_EVENT_TYPES:
         raise ValueError(f"{record.record_id}: unsupported event_type {record.event_type!r}")
     if record.source_type in GOVERNMENT_SOURCE_TYPES:
@@ -125,6 +137,9 @@ def normalize_records(records: list[OfficialRecord]) -> list[dict[str, str]]:
                 "confidence": confidence_for_source(record),
                 "source_url": record.source_url,
                 "notes": record.summary,
+                "entity_match_type": record.entity_match_type,
+                "match_evidence": record.match_evidence,
+                "relationship_type": record.relationship_type,
             }
         )
     rows.sort(key=lambda row: (row["event_date"], row["symbol"], row["event_id"]))
@@ -136,7 +151,19 @@ def import_official_events(input_path: str | Path, output_path: str | Path) -> l
     rows = normalize_records(records)
     write_csv_rows(
         output_path,
-        ["event_id", "event_date", "symbol", "event_type", "direction", "confidence", "source_url", "notes"],
+        [
+            "event_id",
+            "event_date",
+            "symbol",
+            "event_type",
+            "direction",
+            "confidence",
+            "source_url",
+            "notes",
+            "entity_match_type",
+            "match_evidence",
+            "relationship_type",
+        ],
         rows,
     )
     return rows

src/political_event_tracking_research/source_mention_extract.py

@@ -10,6 +10,31 @@
 from .official_event_import import OfficialRecord, normalize_records
 
 
+GENERIC_ALIAS_DENYLIST = frozenset(
+    {
+        "advanced nuclear",
+        "bitcoin treasury",
+        "chips act",
+        "crypto assets",
+        "cybersecurity",
+        "digital assets",
+        "energy-related infrastructure",
+        "foundry",
+        "nuclear energy",
+        "nuclear reactor",
+        "semiconductor manufacturing",
+        "strategy",
+        "tokenization",
+    }
+)
+RELATIONSHIP_PRIORITY = {
+    "unverified": 0,
+    "industry_context": 1,
+    "issuer": 2,
+    "direct_beneficiary": 3,
+}
+
+
 @dataclass(frozen=True)
 class RawSourceItem:
     item_id: str
@@ -24,6 +49,7 @@ class RawSourceItem:
 class MentionAlias:
     symbol: str
     aliases: tuple[str, ...]
+    name: str = ""
 
 
 def load_raw_items(path: str | Path) -> list[RawSourceItem]:
@@ -51,10 +77,13 @@ def load_aliases(path: str | Path) -> list[MentionAlias]:
     records: list[MentionAlias] = []
     for row in read_csv_rows(path):
         symbol = row["symbol"].upper()
+        name = row.get("name", "").strip()
         aliases = split_aliases(row.get("aliases", ""))
+        if name and name.casefold() not in {alias.casefold() for alias in aliases}:
+            aliases = (name, *aliases)
         if symbol not in aliases:
             aliases = (symbol, *aliases)
-        records.append(MentionAlias(symbol=symbol, aliases=aliases))
+        records.append(MentionAlias(symbol=symbol, aliases=aliases, name=name))
     return records
 
 
@@ -88,6 +117,36 @@ def match_symbols(text: str, aliases: list[MentionAlias]) -> list[str]:
     return sorted(dict.fromkeys(matches))
 
 
+def _is_generic_alias(alias_record: MentionAlias, alias: str) -> bool:
+    normalized = normalize_match_text(alias).strip()
+    if normalized.upper() == alias_record.symbol.upper():
+        return False
+    return normalized.casefold() in GENERIC_ALIAS_DENYLIST
+
+
+def match_evidence(text: str, alias_record: MentionAlias, source_type: str = "") -> tuple[str, str] | None:
+    normalized_text = normalize_match_text(text)
+    matches: list[tuple[str, str]] = []
+    for alias in alias_record.aliases:
+        if not alias_pattern(alias).search(normalized_text):
+            continue
+        is_canonical_name = bool(alias_record.name) and alias.casefold() == alias_record.name.casefold()
+        if is_canonical_name and source_type == "issuer_release":
+            matches.append(("issuer", normalize_match_text(alias)))
+        elif _is_generic_alias(alias_record, alias):
+            matches.append(("industry_context", normalize_match_text(alias)))
+            continue
+        relationship = "direct_beneficiary" if re.search(
+            rf"(?:awarded to|contract with|funding for|benefit(?:s|ed)? from)\s+(?:the\s+)?{re.escape(normalize_match_text(alias))}",
+            normalized_text,
+            re.IGNORECASE,
+        ) else "issuer"
+        matches.append((relationship, normalize_match_text(alias)))
+    if not matches:
+        return None
+    return max(matches, key=lambda match: (RELATIONSHIP_PRIORITY[match[0]], match[1].casefold()))
+
+
 def infer_event_type(item: RawSourceItem) -> str:
     source_type = item.source_type
     text = item.text.lower()
@@ -133,24 +192,42 @@ def extract_source_records(raw_items_path: str | Path, aliases_path: str | Path,
     aliases = load_aliases(aliases_path)
     records: list[OfficialRecord] = []
     for item in raw_items:
-        symbols = match_symbols(item.text, aliases)
-        for symbol in symbols:
+        for alias_record in aliases:
+            evidence = match_evidence(item.text, alias_record, item.source_type)
+            if evidence is None:
+                continue
+            entity_match_type, matched_text = evidence
             records.append(
                 OfficialRecord(
-                    record_id=f"{item.item_id}-{symbol.lower()}",
+                    record_id=f"{item.item_id}-{alias_record.symbol.lower()}",
                     record_date=item_date(item),
-                    symbol=symbol,
+                    symbol=alias_record.symbol,
                     source_type=item.source_type,
                     event_type=infer_event_type(item),
                     direction=infer_direction(item.text),
                     source_url=item.source_url,
                     summary=f"{item.author}: {item.text}".strip(": "),
+                    entity_match_type=entity_match_type,
+                    match_evidence=matched_text,
+                    relationship_type=entity_match_type,
                 )
             )
     rows = normalize_records(records)
     write_csv_rows(
         output_path,
-        ["event_id", "event_date", "symbol", "event_type", "direction", "confidence", "source_url", "notes"],
+        [
+            "event_id",
+            "event_date",
+            "symbol",
+            "event_type",
+            "direction",
+            "confidence",
+            "source_url",
+            "notes",
+            "entity_match_type",
+            "match_evidence",
+            "relationship_type",
+        ],
         rows,
     )
     return rows