fix: add bounded legacy event compatibility

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

docs/free_source_setup.md

@@ -18,6 +18,11 @@ 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

@@ -16,6 +16,10 @@
 只有前两类可作为公司级证据；行业通用词会保留用于审计，但不会进入公司事件评分。
 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

@@ -21,6 +21,9 @@ class Event:
     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)
@@ -38,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 {
@@ -54,6 +60,9 @@ 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,
         }
 
 
@@ -65,9 +74,14 @@ 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"],
@@ -81,6 +95,9 @@ def load_events(path: str | Path) -> list[Event]:
                 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
@@ -130,12 +147,14 @@ 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:
-        if event.relationship_type not in VERIFIED_RELATIONSHIPS:
+        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)
@@ -175,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 "",
                 )
             )
 
@@ -196,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,
         [
@@ -216,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],
     )
@@ -229,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
 
 
@@ -240,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/source_mention_extract.py

@@ -124,13 +124,16 @@ def _is_generic_alias(alias_record: MentionAlias, alias: str) -> bool:
     return normalized.casefold() in GENERIC_ALIAS_DENYLIST
 
 
-def match_evidence(text: str, alias_record: MentionAlias) -> tuple[str, str] | None:
+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
-        if _is_generic_alias(alias_record, alias):
+        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(
@@ -190,7 +193,7 @@ def extract_source_records(raw_items_path: str | Path, aliases_path: str | Path,
     records: list[OfficialRecord] = []
     for item in raw_items:
         for alias_record in aliases:
-            evidence = match_evidence(item.text, alias_record)
+            evidence = match_evidence(item.text, alias_record, item.source_type)
             if evidence is None:
                 continue
             entity_match_type, matched_text = evidence

tests/test_event_study.py

@@ -4,7 +4,7 @@
 
 import pytest
 
-from political_event_tracking_research.event_study import Event, compute_event_returns
+from political_event_tracking_research.event_study import Event, compute_event_returns, load_events
 
 
 def test_compute_event_returns_uses_next_available_trading_date_and_benchmark() -> None:
@@ -70,3 +70,22 @@ def test_compute_event_returns_excludes_non_company_relationships() -> None:
     prices = {"ABC": {date(2026, 1, 5): 100.0, date(2026, 1, 6): 110.0}}
 
     assert compute_event_returns(events, prices, windows=(1,)) == []
+
+
+def test_legacy_event_csv_requires_explicit_compatibility_and_provenance(tmp_path) -> None:
+    path = tmp_path / "legacy_events.csv"
+    path.write_text(
+        "event_id,event_date,symbol,event_type,direction,confidence,source_url,notes\n"
+        "legacy-1,2026-01-02,ABC,public_mention,bullish,high,https://example.com,legacy\n",
+        encoding="utf-8",
+    )
+    prices = {"ABC": {date(2026, 1, 5): 100.0, date(2026, 1, 6): 110.0}}
+
+    legacy = load_events(path, historical_compatibility=True, compatibility_reason="reproduce 2026 baseline")
+    assert legacy[0].relationship_type == "unverified"
+    assert compute_event_returns(legacy, prices, windows=(1,)) == []
+    results = compute_event_returns(legacy, prices, windows=(1,), historical_compatibility=True)
+
+    assert results[0].compatibility_used is True
+    assert results[0].compatibility_reason == "reproduce 2026 baseline"
+    assert results[0].legacy_provenance == str(path)

tests/test_source_mention_extract.py

@@ -100,6 +100,26 @@ def test_explicit_issuer_match_is_company_level(tmp_path: Path) -> None:
     assert rows[0]["match_evidence"] == "Coinbase"
 
 
+def test_canonical_name_on_trusted_issuer_release_beats_generic_denylist(tmp_path: Path) -> None:
+    raw_items = tmp_path / "source_items.csv"
+    raw_items.write_text(
+        "item_id,published_at,source_type,source_url,author,text\n"
+        "issuer-strategy,2026-04-01T00:00:00Z,issuer_release,https://example.com/strategy,Strategy,"
+        'Strategy announced a new product.\n'
+        "policy-strategy,2026-04-01T00:00:00Z,government_policy,https://www.nist.gov/strategy,NIST,"
+        'Strategy guidance was published.\n',
+        encoding="utf-8",
+    )
+    aliases = tmp_path / "aliases.csv"
+    aliases.write_text("symbol,name,aliases\nMSTR,Strategy,Strategy|MSTR\n", encoding="utf-8")
+
+    rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv")
+    by_id = {row["event_id"]: row for row in rows}
+
+    assert by_id["official-issuer-release-issuer-strategy-mstr"]["entity_match_type"] == "issuer"
+    assert by_id["official-government-policy-policy-strategy-mstr"]["entity_match_type"] == "industry_context"
+
+
 def test_proper_noun_alternate_alias_remains_company_level(tmp_path: Path) -> None:
     raw_items = tmp_path / "source_items.csv"
     raw_items.write_text(