Spaces:
Running on Zero
Running on Zero
| """Immutable public data contracts for news relevance decisions.""" | |
| from __future__ import annotations | |
| import html | |
| from collections.abc import Mapping | |
| from dataclasses import dataclass, field | |
| from datetime import datetime | |
| from enum import Enum | |
| from types import MappingProxyType | |
| from zoneinfo import ZoneInfo | |
| _ARTICLE_TIMEZONE = ZoneInfo("America/New_York") | |
| def parse_article_time(value: datetime | str) -> datetime: | |
| """Parse an article time, interpreting naive values as America/New_York.""" | |
| if isinstance(value, datetime): | |
| timestamp = value | |
| elif isinstance(value, str) and value.strip(): | |
| timestamp = datetime.fromisoformat(value.strip()) | |
| else: | |
| raise TypeError("timestamp must be a datetime or non-empty ISO-8601 string") | |
| if timestamp.tzinfo is None: | |
| timestamp = timestamp.replace(tzinfo=_ARTICLE_TIMEZONE) | |
| return timestamp | |
| class ArticleAdmissibilityStatus(str, Enum): | |
| ACCEPTED = "accepted" | |
| PLAUSIBLE_UNRESOLVED = "plausible_unresolved" | |
| REJECTED = "rejected" | |
| class ExposureRelevanceStatus(str, Enum): | |
| SUPPORTED = "supported" | |
| PLAUSIBLE_UNRESOLVED = "plausible_unresolved" | |
| NOT_SUPPORTED = "not_supported" | |
| class ExposureFamily(str, Enum): | |
| SP500 = "EXPOSURE:SP500" | |
| NASDAQ100 = "EXPOSURE:NASDAQ100" | |
| RUSSELL2000 = "EXPOSURE:RUSSELL2000" | |
| MARKET_WIDE = "MARKET_WIDE" | |
| def label(self) -> str: | |
| """Return the concise name used in user-facing summaries.""" | |
| return { | |
| ExposureFamily.SP500: "S&P 500", | |
| ExposureFamily.NASDAQ100: "Nasdaq-100", | |
| ExposureFamily.RUSSELL2000: "Russell 2000", | |
| ExposureFamily.MARKET_WIDE: "Broad market", | |
| }[self] | |
| class IdentityStatus(str, Enum): | |
| ACCEPTED = "accepted" | |
| AMBIGUOUS = "ambiguous" | |
| REJECTED = "rejected" | |
| UNRESOLVED = "unresolved" | |
| class EvidenceStatus(str, Enum): | |
| ACCEPTED = "accepted" | |
| UNRESOLVED = "unresolved" | |
| REJECTED = "rejected" | |
| class Article: | |
| article_id: str | |
| timestamp: datetime | str | |
| text: str | |
| source: str = "unknown" | |
| content_id: str | None = None | |
| decision_bucket: str | None = None | |
| def __post_init__(self) -> None: | |
| if not self.article_id.strip(): | |
| raise ValueError("article_id must be non-empty") | |
| if not isinstance(self.text, str) or not self.text.strip(): | |
| raise ValueError("text must be a non-empty string") | |
| object.__setattr__(self, "timestamp", parse_article_time(self.timestamp)) | |
| object.__setattr__(self, "content_id", self.content_id or self.article_id) | |
| class Observation: | |
| observation_id: str | |
| article_id: str | |
| matched_surface: str | |
| span_start: int | |
| span_end: int | |
| entity_concept: str | |
| detection_method: str | |
| confidence: float | None = None | |
| candidate_canonical_id: str | None = None | |
| catalog_entry_id: str | None = None | |
| catalog_version: str | None = None | |
| rule_id: str | None = None | |
| tag_type: str | None = None | |
| event_family: str | None = None | |
| market_wide_authorized: bool = False | |
| context_status: str = "not_required" | |
| source_id: str | None = None | |
| def __post_init__(self) -> None: | |
| if self.span_start < 0 or self.span_end <= self.span_start: | |
| raise ValueError("observation offsets must describe a non-empty span") | |
| if not self.observation_id or not self.article_id or not self.matched_surface: | |
| raise ValueError("observation identity and surface must be non-empty") | |
| if self.confidence is not None and not 0 <= self.confidence <= 1: | |
| raise ValueError("confidence must be in [0, 1]") | |
| class ObservationBundle: | |
| article: Article | |
| observations: tuple[Observation, ...] = () | |
| class IdentityResolution: | |
| resolution_id: str | |
| observation_id: str | |
| article_id: str | |
| status: IdentityStatus | |
| canonical_id: str | None | |
| reason_code: str | |
| resolution_method: str | |
| catalog_version: str | None = None | |
| def __post_init__(self) -> None: | |
| if self.status is IdentityStatus.ACCEPTED and not self.canonical_id: | |
| raise ValueError("accepted identities require a canonical_id") | |
| if self.status is not IdentityStatus.ACCEPTED and self.canonical_id is not None: | |
| raise ValueError("only accepted identities may assign canonical_id") | |
| class EvidencePath: | |
| evidence_id: str | |
| article_id: str | |
| tag_type: str | |
| tag_detail: str | |
| matched_surface: str | |
| detection_method: str | |
| status: EvidenceStatus | |
| link_status: str | |
| target_id: str | None | |
| exposure_family: ExposureFamily | None | |
| reason_code: str | |
| tier: int | |
| observation_ids: tuple[str, ...] = () | |
| identity_resolution_ids: tuple[str, ...] = () | |
| relationship_ids: tuple[str, ...] = () | |
| catalog_version: str | None = None | |
| event_family: str | None = None | |
| explicit_market_wide_rule: bool = False | |
| def __post_init__(self) -> None: | |
| if self.tier < 0: | |
| raise ValueError("evidence tier must be non-negative") | |
| if self.status is EvidenceStatus.ACCEPTED and self.exposure_family is None: | |
| raise ValueError("accepted exposure evidence requires an exposure family") | |
| if self.link_status == "resolved_exposure" and not self.relationship_ids: | |
| raise ValueError("resolved exposure evidence requires relationship lineage") | |
| if self.exposure_family is ExposureFamily.MARKET_WIDE and ( | |
| self.link_status != "market_wide" or not self.explicit_market_wide_rule | |
| ): | |
| raise ValueError("MARKET_WIDE requires an explicitly authorized rule") | |
| class ArticleAdmissibilityDecision: | |
| article_id: str | |
| status: ArticleAdmissibilityStatus | |
| accepted_evidence: tuple[EvidencePath, ...] = () | |
| unresolved_evidence: tuple[EvidencePath, ...] = () | |
| rejected_evidence: tuple[EvidencePath, ...] = () | |
| reason_codes: tuple[str, ...] = () | |
| class ExposureRelevanceDecision: | |
| article_id: str | |
| exposure_family: ExposureFamily | |
| status: ExposureRelevanceStatus | |
| evidence_paths: tuple[EvidencePath, ...] = () | |
| unresolved_paths: tuple[EvidencePath, ...] = () | |
| reason_codes: tuple[str, ...] = () | |
| def __post_init__(self) -> None: | |
| if self.status is ExposureRelevanceStatus.SUPPORTED and not self.evidence_paths: | |
| raise ValueError("supported exposure decisions require evidence lineage") | |
| if ( | |
| self.status is ExposureRelevanceStatus.PLAUSIBLE_UNRESOLVED | |
| and not self.unresolved_paths | |
| ): | |
| raise ValueError( | |
| "plausible unresolved exposure decisions require unresolved evidence" | |
| ) | |
| if any( | |
| path.exposure_family is not self.exposure_family | |
| for path in self.evidence_paths | |
| ): | |
| raise ValueError("evidence paths must support the decision exposure family") | |
| class RelevanceDecision: | |
| article: Article | |
| article_decision: ArticleAdmissibilityDecision | |
| exposure_decisions: Mapping[ExposureFamily, ExposureRelevanceDecision] | |
| observations: tuple[Observation, ...] = () | |
| resolutions: tuple[IdentityResolution, ...] = () | |
| def __post_init__(self) -> None: | |
| if self.article.article_id != self.article_decision.article_id: | |
| raise ValueError("article decision provenance mismatch") | |
| decisions = dict(self.exposure_decisions) | |
| if any(key is not value.exposure_family for key, value in decisions.items()): | |
| raise ValueError("exposure decision mapping keys must match their values") | |
| observation_ids = {row.observation_id for row in self.observations} | |
| if len(observation_ids) != len(self.observations): | |
| raise ValueError("observation IDs must be unique") | |
| if any(row.article_id != self.article.article_id for row in self.observations): | |
| raise ValueError("observation provenance mismatch") | |
| resolution_ids = {row.resolution_id for row in self.resolutions} | |
| if len(resolution_ids) != len(self.resolutions): | |
| raise ValueError("identity resolution IDs must be unique") | |
| for resolution in self.resolutions: | |
| if resolution.article_id != self.article.article_id: | |
| raise ValueError("identity resolution provenance mismatch") | |
| if resolution.observation_id not in observation_ids: | |
| raise ValueError( | |
| "identity resolution references an unknown observation" | |
| ) | |
| article_paths = ( | |
| *self.article_decision.accepted_evidence, | |
| *self.article_decision.unresolved_evidence, | |
| *self.article_decision.rejected_evidence, | |
| ) | |
| exposure_paths = tuple( | |
| path | |
| for decision in decisions.values() | |
| for path in (*decision.evidence_paths, *decision.unresolved_paths) | |
| ) | |
| for path in (*article_paths, *exposure_paths): | |
| if path.article_id != self.article.article_id: | |
| raise ValueError("evidence path provenance mismatch") | |
| if not set(path.observation_ids).issubset(observation_ids): | |
| raise ValueError("evidence path references an unknown observation") | |
| if not set(path.identity_resolution_ids).issubset(resolution_ids): | |
| raise ValueError( | |
| "evidence path references an unknown identity resolution" | |
| ) | |
| object.__setattr__(self, "exposure_decisions", MappingProxyType(decisions)) | |
| def status(self) -> str: | |
| """Return the plain-language overall result.""" | |
| return { | |
| ArticleAdmissibilityStatus.ACCEPTED: "relevant", | |
| ArticleAdmissibilityStatus.PLAUSIBLE_UNRESOLVED: "needs_review", | |
| ArticleAdmissibilityStatus.REJECTED: "not_relevant", | |
| }[self.article_decision.status] | |
| def accepted(self) -> bool: | |
| """Whether the article has accepted financial evidence.""" | |
| return self.article_decision.status is ArticleAdmissibilityStatus.ACCEPTED | |
| def relevant_to(self) -> tuple[str, ...]: | |
| """Plain-language markets with a completed relevance path.""" | |
| return tuple( | |
| family.label | |
| for family in ExposureFamily | |
| if ( | |
| (result := self.exposure_decisions.get(family)) is not None | |
| and result.status is ExposureRelevanceStatus.SUPPORTED | |
| ) | |
| ) | |
| def _surface(path: EvidencePath) -> str: | |
| surface = " ".join(path.matched_surface.split()) | |
| return surface if len(surface) <= 60 else surface[:57].rstrip() + "..." | |
| def _why_for_path(cls, path: EvidencePath) -> str: | |
| surface = cls._surface(path) | |
| if path.link_status == "resolved_exposure" and path.exposure_family: | |
| return f"{surface} connects to {path.exposure_family.label}" | |
| if path.link_status == "direct_asset": | |
| return f"fund or ticker found: {surface}" | |
| if path.link_status == "direct_benchmark": | |
| return f"index or benchmark found: {surface}" | |
| if path.link_status == "market_wide": | |
| return f"broad-market news found: {surface}" | |
| return { | |
| "macroeconomic_indicator": f"economic indicator found: {surface}", | |
| "financial_regulator": f"financial regulator found: {surface}", | |
| "sector_or_industry": f"industry or sector found: {surface}", | |
| "company_event": f"company event found: {surface}", | |
| }.get(path.tag_type, f"financial information found: {surface}") | |
| def why(self) -> str: | |
| """Return a short explanation; detailed lineage remains available separately.""" | |
| if self.article_decision.status is ArticleAdmissibilityStatus.REJECTED: | |
| return "No financial match found." | |
| paths = ( | |
| self.article_decision.accepted_evidence | |
| if self.article_decision.status is ArticleAdmissibilityStatus.ACCEPTED | |
| else self.article_decision.unresolved_evidence | |
| ) | |
| phrases: list[str] = [] | |
| seen: set[str] = set() | |
| for path in sorted( | |
| paths, | |
| key=lambda item: ( | |
| item.tier, | |
| item.matched_surface.casefold(), | |
| item.evidence_id, | |
| ), | |
| ): | |
| key = path.matched_surface.casefold().strip() | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| phrases.append(self._why_for_path(path)) | |
| if not phrases: | |
| return ( | |
| "Possible financial information was found, but it needs review." | |
| if self.status == "needs_review" | |
| else "Financial information found." | |
| ) | |
| visible = phrases[:3] | |
| reason = "; ".join(visible) | |
| if len(phrases) > len(visible): | |
| reason += f"; and {len(phrases) - len(visible)} more match(es)" | |
| if self.status == "needs_review": | |
| reason += ", but it could not be connected to a configured market" | |
| return reason[0].upper() + reason[1:] + "." | |
| def summary(self) -> dict[str, object]: | |
| """Return the small, JSON-friendly result most callers need.""" | |
| return { | |
| "article_id": self.article.article_id, | |
| "status": self.status, | |
| "accepted": self.accepted, | |
| "why": self.why, | |
| "relevant_to": list(self.relevant_to), | |
| } | |
| def __repr__(self) -> str: | |
| return ( | |
| "RelevanceDecision(" | |
| f"article_id={self.article.article_id!r}, " | |
| f"status={self.status!r}, " | |
| f"relevant_to={self.relevant_to!r}, " | |
| f"why={self.why!r})" | |
| ) | |
| def _repr_markdown_(self) -> str: | |
| """Render a concise decision card in Jupyter without an IPython dependency.""" | |
| def markdown(value: object) -> str: | |
| escaped = html.escape(str(value)) | |
| for character in ("\\", "`", "*", "_", "[", "]"): | |
| escaped = escaped.replace(character, f"\\{character}") | |
| return escaped | |
| heading = { | |
| "relevant": "Keep", | |
| "needs_review": "Check", | |
| "not_relevant": "Ignore", | |
| }[self.status] | |
| lines = [f"### {heading}", "", f"**Why:** {markdown(self.why)}"] | |
| if self.relevant_to: | |
| lines.extend(("", f"**Markets:** {', '.join(self.relevant_to)}")) | |
| lines.extend( | |
| ( | |
| "", | |
| ( | |
| f"*{markdown(self.article.source)} | " | |
| f"{markdown(self.article.timestamp.isoformat())} | " | |
| f"{markdown(self.article.article_id)}*" | |
| ), | |
| ) | |
| ) | |
| return "\n".join(lines) | |
| class RankedCandidate: | |
| decision_bucket: str | |
| exposure_family: ExposureFamily | |
| rank: int | |
| article_id: str | |
| accepted_evidence_ids: tuple[str, ...] | |
| decision: RelevanceDecision = field(compare=False, repr=False) | |