diff --git a/docs/free_source_setup.md b/docs/free_source_setup.md index 7473d27..8066bbd 100644 --- a/docs/free_source_setup.md +++ b/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. diff --git a/docs/free_source_setup.zh-CN.md b/docs/free_source_setup.zh-CN.md index 7c5e272..3f078db 100644 --- a/docs/free_source_setup.zh-CN.md +++ b/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。 diff --git a/src/political_event_tracking_research/event_study.py b/src/political_event_tracking_research/event_study.py index 877b604..7c3ab24 100644 --- a/src/political_event_tracking_research/event_study.py +++ b/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, ) diff --git a/src/political_event_tracking_research/official_event_import.py b/src/political_event_tracking_research/official_event_import.py index 947be00..53e69d0 100644 --- a/src/political_event_tracking_research/official_event_import.py +++ b/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 diff --git a/src/political_event_tracking_research/source_mention_extract.py b/src/political_event_tracking_research/source_mention_extract.py index e5e4648..252198d 100644 --- a/src/political_event_tracking_research/source_mention_extract.py +++ b/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 diff --git a/src/political_event_tracking_research/tracker.py b/src/political_event_tracking_research/tracker.py index 1ab312c..71b952e 100644 --- a/src/political_event_tracking_research/tracker.py +++ b/src/political_event_tracking_research/tracker.py @@ -67,6 +67,8 @@ def latest_event(events: list[Event]) -> Event | None: def event_score(events: list[Event]) -> int: score = 0 for event in events: + if event.relationship_type not in {"issuer", "direct_beneficiary"}: + continue score += EVENT_WEIGHTS.get(event.event_type, 0) score += CONFIDENCE_WEIGHTS.get(event.confidence, 0) return score @@ -93,17 +95,20 @@ def build_tracker_rows(items: list[WatchlistItem], events: list[Event]) -> list[ rows: list[dict[str, object]] = [] for item in items: symbol_events = sorted(events_by_symbol.get(item.symbol, []), key=lambda event: (event.event_date, event.event_id)) - latest = latest_event(symbol_events) + company_events = [ + event for event in symbol_events if event.relationship_type in {"issuer", "direct_beneficiary"} + ] + latest = latest_event(company_events) base_score = BUCKET_WEIGHTS.get(item.bucket, 0) - score = base_score + event_score(symbol_events) + score = base_score + event_score(company_events) rows.append( { "symbol": item.symbol, "name": item.name, "bucket": item.bucket, - "trigger_status": trigger_status(item, symbol_events), + "trigger_status": trigger_status(item, company_events), "priority_score": score, - "event_count": len(symbol_events), + "event_count": len(company_events), "latest_event_date": latest.event_date.isoformat() if latest else "", "latest_event_type": latest.event_type if latest else "", "source_url": item.source_url, diff --git a/tests/test_event_study.py b/tests/test_event_study.py index e29771c..3491ba0 100644 --- a/tests/test_event_study.py +++ b/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: @@ -18,6 +18,7 @@ def test_compute_event_returns_uses_next_available_trading_date_and_benchmark() confidence="high", source_url="https://example.com", notes="seed", + relationship_type="issuer", ) ] prices = { @@ -59,3 +60,32 @@ def test_compute_event_returns_skips_events_without_prices() -> None: ] assert compute_event_returns(events, prices={}, windows=(1,)) == [] + + +def test_compute_event_returns_excludes_non_company_relationships() -> None: + events = [ + Event("context", date(2026, 1, 2), "ABC", "public_mention", "bullish", "high", "https://example.com", "", relationship_type="industry_context"), + Event("unknown", date(2026, 1, 2), "ABC", "public_mention", "bullish", "high", "https://example.com", ""), + ] + 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) diff --git a/tests/test_official_event_import.py b/tests/test_official_event_import.py index 5c315ef..2626cc4 100644 --- a/tests/test_official_event_import.py +++ b/tests/test_official_event_import.py @@ -76,3 +76,21 @@ def test_community_lead_records_are_not_in_stable_source_set() -> None: with pytest.raises(ValueError, match="unsupported source_type"): normalize_records([record]) + + +def test_import_rejects_mismatched_entity_and_relationship_types() -> None: + record = OfficialRecord( + record_id="mismatched-relation", + record_date="2026-01-10", + symbol="BAD", + source_type="issuer_release", + event_type="public_mention", + direction="neutral", + source_url="https://example.com/release", + summary="Mismatched evidence fields.", + entity_match_type="issuer", + relationship_type="industry_context", + ) + + with pytest.raises(ValueError, match="must agree"): + normalize_records([record]) diff --git a/tests/test_source_mention_extract.py b/tests/test_source_mention_extract.py index 8f0e529..ec5e29f 100644 --- a/tests/test_source_mention_extract.py +++ b/tests/test_source_mention_extract.py @@ -2,7 +2,11 @@ from pathlib import Path -from political_event_tracking_research.source_mention_extract import extract_source_records, infer_direction, match_symbols +from political_event_tracking_research.source_mention_extract import ( + extract_source_records, + infer_direction, + match_symbols, +) from political_event_tracking_research.source_mention_extract import MentionAlias @@ -51,3 +55,148 @@ def test_extract_source_records_outputs_confidence_by_source_type(tmp_path: Path assert by_symbol["EVT3"]["confidence"] == "low" assert by_symbol["EVT4"]["event_type"] == "procurement" assert output.exists() + + +def test_generic_aliases_are_context_only_and_keep_evidence(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" + "nist-1,2026-04-01T00:00:00Z,government_policy,https://www.nist.gov/example,NIST," + 'NIST guidance discusses nuclear reactor safety and tokenization.\n', + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text( + "symbol,name,aliases\n" + "OKLO,Oklo,Oklo|OKLO|nuclear reactor\n" + "COIN,Coinbase,Coinbase|COIN|tokenization\n", + encoding="utf-8", + ) + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + by_symbol = {row["symbol"]: row for row in rows} + assert by_symbol["OKLO"]["entity_match_type"] == "industry_context" + assert by_symbol["COIN"]["entity_match_type"] == "industry_context" + assert by_symbol["OKLO"]["match_evidence"] == "nuclear reactor" + assert by_symbol["COIN"]["relationship_type"] == "industry_context" + + +def test_explicit_issuer_match_is_company_level(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-1,2026-04-01T00:00:00Z,issuer_release,https://example.com/release,Issuer," + 'Coinbase announced a new product.\n', + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text("symbol,name,aliases\nCOIN,Coinbase,COIN|tokenization\n", encoding="utf-8") + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + assert rows[0]["entity_match_type"] == "issuer" + assert rows[0]["relationship_type"] == "issuer" + 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( + "item_id,published_at,source_type,source_url,author,text\n" + "issuer-2,2026-04-01T00:00:00Z,issuer_release,https://example.com/release,Issuer," + 'Former Brand announced a new product.\n', + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text("symbol,name,aliases\nTEST,Current Company,Current Company|Former Brand\n", encoding="utf-8") + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + assert rows[0]["entity_match_type"] == "issuer" + assert rows[0]["match_evidence"] == "Former Brand" + + +def test_strongest_match_wins_regardless_of_alias_order(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" + "mixed-1,2026-04-01T00:00:00Z,government_procurement,https://www.govinfo.gov/example,Agency," + 'Funding for Current Company supports cybersecurity.\n', + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text( + "symbol,name,aliases\nTEST,Current Company,cybersecurity|Current Company\n", + encoding="utf-8", + ) + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + assert rows[0]["entity_match_type"] == "direct_beneficiary" + assert rows[0]["match_evidence"] == "Current Company" + + +def test_explicit_beneficiary_relation_is_preserved(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" + "award-1,2026-04-01T00:00:00Z,government_procurement,https://www.govinfo.gov/example,Agency," + 'Funding for Coinbase was announced.\n', + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text("symbol,name,aliases\nCOIN,Coinbase,COIN\n", encoding="utf-8") + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + assert rows[0]["entity_match_type"] == "direct_beneficiary" + assert rows[0]["relationship_type"] == "direct_beneficiary" + + +def test_known_ticker_and_industry_collisions_never_become_issuer_events(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" + "mstr,2026-04-01T00:00:00Z,government_policy,https://www.nist.gov/mstr,NIST,Strategy guidance\n" + "oklo,2026-04-01T00:00:00Z,government_policy,https://www.nist.gov/oklo,NIST,nuclear reactor safety\n" + "panw,2026-04-01T00:00:00Z,government_policy,https://www.nist.gov/panw,NIST,generic cybersecurity guidance\n" + "coin,2026-04-01T00:00:00Z,government_policy,https://www.federalreserve.gov/coin,Fed,tokenization policy\n" + "intc,2026-04-01T00:00:00Z,government_policy,https://www.commerce.gov/intc,Commerce,CHIPS Act funding\n", + encoding="utf-8", + ) + aliases = tmp_path / "aliases.csv" + aliases.write_text( + "symbol,name,aliases\n" + "MSTR,Strategy,Strategy|MSTR\n" + "OKLO,Oklo,Oklo|OKLO|nuclear reactor\n" + "PANW,Palo Alto Networks,Palo Alto Networks|PANW|cybersecurity\n" + "COIN,Coinbase,Coinbase|COIN|tokenization\n" + "INTC,Intel,Intel|INTC|CHIPS Act\n", + encoding="utf-8", + ) + + rows = extract_source_records(raw_items, aliases, tmp_path / "events.csv") + + assert {row["symbol"] for row in rows} == {"MSTR", "OKLO", "PANW", "COIN", "INTC"} + assert {row["entity_match_type"] for row in rows} == {"industry_context"} diff --git a/tests/test_tracker.py b/tests/test_tracker.py index c439188..d5457a8 100644 --- a/tests/test_tracker.py +++ b/tests/test_tracker.py @@ -26,8 +26,8 @@ def test_tracker_prioritizes_disclosure_plus_mention() -> None: ), ] events = [ - Event("e1", date(2026, 1, 1), "AAA", "disclosure_buy", "bullish", "low", "https://example.com", ""), - Event("e2", date(2026, 1, 2), "AAA", "public_mention", "bullish", "low", "https://example.com", ""), + Event("e1", date(2026, 1, 1), "AAA", "disclosure_buy", "bullish", "low", "https://example.com", "", relationship_type="issuer"), + Event("e2", date(2026, 1, 2), "AAA", "public_mention", "bullish", "low", "https://example.com", "", relationship_type="issuer"), ] rows = build_tracker_rows(items, events) @@ -38,3 +38,35 @@ def test_tracker_prioritizes_disclosure_plus_mention() -> None: assert rows[0]["latest_event_date"] == "2026-01-02" assert rows[1]["symbol"] == "BBB" assert rows[1]["trigger_status"] == "watchlist" + + +def test_tracker_does_not_score_industry_context_as_company_evidence() -> None: + item = WatchlistItem("AAA", "Alpha", "watchlist", "watchlist", "seed", "https://example.com") + events = [Event("e1", date(2026, 1, 1), "AAA", "public_mention", "bullish", "high", "https://example.com", "", "industry_context", "cybersecurity", "industry_context")] + + rows = build_tracker_rows([item], events) + + assert rows[0]["priority_score"] == 0 + + +def test_tracker_uses_relationship_type_as_scoring_fact() -> None: + item = WatchlistItem("AAA", "Alpha", "watchlist", "watchlist", "seed", "https://example.com") + events = [Event("e1", date(2026, 1, 1), "AAA", "public_mention", "bullish", "high", "https://example.com", "", "industry_context", "cybersecurity", "issuer")] + + rows = build_tracker_rows([item], events) + + assert rows[0]["priority_score"] > 0 + + +def test_tracker_mixed_only_exposes_verified_company_events() -> None: + item = WatchlistItem("AAA", "Alpha", "watchlist", "watchlist", "seed", "https://example.com") + events = [ + Event("context", date(2026, 1, 2), "AAA", "public_mention", "bullish", "high", "https://example.com", "", relationship_type="industry_context"), + Event("issuer", date(2026, 1, 1), "AAA", "public_mention", "bullish", "high", "https://example.com", "", relationship_type="issuer"), + ] + + rows = build_tracker_rows([item], events) + + assert rows[0]["event_count"] == 1 + assert rows[0]["latest_event_date"] == "2026-01-01" + assert rows[0]["latest_event_type"] == "public_mention"