Spaces:
Running on Zero
Running on Zero
| """Immutable taxonomy composition and deterministic ranking contracts.""" | |
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| from abc import ABC, abstractmethod | |
| from collections.abc import Callable, Iterable, Mapping, Sequence | |
| from dataclasses import dataclass, field, fields, is_dataclass, replace | |
| from datetime import datetime, timezone | |
| from enum import Enum | |
| from importlib.resources import files as resource_files | |
| from pathlib import Path | |
| from types import MappingProxyType | |
| from ._core.catalogs import ( | |
| CatalogAlias, | |
| CatalogBundle, | |
| CatalogEntity, | |
| LexicalRule, | |
| active_during, | |
| load_catalogs, | |
| ) | |
| from ._core.config import NewsRelevanceConfig, load_config | |
| from ._core.models import ( | |
| EvidencePath, | |
| ExposureFamily, | |
| ExposureRelevanceStatus, | |
| RankedCandidate, | |
| RelevanceDecision, | |
| ) | |
| _MATCH_MODES = { | |
| "exact_identifier", | |
| "exact_case_sensitive", | |
| "exact_casefold", | |
| "normalized_phrase", | |
| } | |
| _ENTITY_KINDS = {"fund", "index", "issuer_family"} | |
| _DETECTION_METHODS = {"lexical_phrase", "event_rule"} | |
| def _key(value: object) -> str: | |
| if isinstance(value, Enum): | |
| return str(value.value) | |
| return str(value) | |
| def _stable_value(value: object) -> object: | |
| if isinstance(value, Enum): | |
| return value.value | |
| if isinstance(value, Path): | |
| return value.name | |
| if is_dataclass(value): | |
| return { | |
| item.name: _stable_value(getattr(value, item.name)) | |
| for item in fields(value) | |
| } | |
| if isinstance(value, Mapping): | |
| return { | |
| _key(key): _stable_value(item) | |
| for key, item in sorted(value.items(), key=lambda pair: _key(pair[0])) | |
| } | |
| if isinstance(value, (set, frozenset)): | |
| return sorted((_stable_value(item) for item in value), key=repr) | |
| if isinstance(value, (tuple, list)): | |
| return [_stable_value(item) for item in value] | |
| return value | |
| def _fingerprint(config: NewsRelevanceConfig, catalogs: CatalogBundle) -> str: | |
| config_payload = { | |
| item.name: _stable_value(getattr(config, item.name)) | |
| for item in fields(config) | |
| if item.name not in {"entity_catalog_path", "lexical_catalog_path"} | |
| } | |
| payload = { | |
| "config": config_payload, | |
| "catalogs": _stable_value(catalogs), | |
| } | |
| encoded = json.dumps( | |
| payload, | |
| sort_keys=True, | |
| separators=(",", ":"), | |
| ensure_ascii=False, | |
| ).encode("utf-8") | |
| return hashlib.sha256(encoded).hexdigest() | |
| def copy_default_taxonomy( | |
| destination: str | Path, | |
| *, | |
| overwrite: bool = False, | |
| ) -> Path: | |
| """Copy an editable default taxonomy bundle and return its config path. | |
| The resulting directory contains ``taxonomy.yml`` plus the two catalog | |
| files it references. Call :meth:`Taxonomy.from_yaml` or | |
| :meth:`RelevanceEngine.from_yaml` after editing the copies. | |
| """ | |
| root = Path(destination).expanduser().resolve() | |
| package = resource_files("range_curate._core") | |
| copies = ( | |
| (package.joinpath("configs", "default.yml"), root / "taxonomy.yml"), | |
| ( | |
| package.joinpath("catalogs", "entities.yml"), | |
| root / "catalogs" / "entities.yml", | |
| ), | |
| ( | |
| package.joinpath("catalogs", "lexical.yml"), | |
| root / "catalogs" / "lexical.yml", | |
| ), | |
| ) | |
| existing = [target for _source, target in copies if target.exists()] | |
| if existing and not overwrite: | |
| names = ", ".join(str(path) for path in existing) | |
| raise FileExistsError( | |
| f"taxonomy files already exist: {names}; pass overwrite=True to replace them" | |
| ) | |
| for source, target in copies: | |
| target.parent.mkdir(parents=True, exist_ok=True) | |
| target.write_bytes(source.read_bytes()) | |
| return root / "taxonomy.yml" | |
| def _validate(config: NewsRelevanceConfig, catalogs: CatalogBundle) -> None: | |
| if not isinstance(config, NewsRelevanceConfig): | |
| raise TypeError("config must be a NewsRelevanceConfig") | |
| if not isinstance(catalogs, CatalogBundle): | |
| raise TypeError("catalogs must be a CatalogBundle") | |
| for canonical_id, entity in catalogs.entities.items(): | |
| if canonical_id != entity.canonical_id: | |
| raise ValueError( | |
| f"entity mapping key {canonical_id!r} does not match " | |
| f"canonical ID {entity.canonical_id!r}" | |
| ) | |
| for name, value in ( | |
| ("canonical_id", entity.canonical_id), | |
| ("canonical_name", entity.canonical_name), | |
| ("entity_concept", entity.entity_concept), | |
| ): | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError(f"entity {canonical_id!r} has empty {name}") | |
| if entity.entity_kind not in _ENTITY_KINDS: | |
| raise ValueError( | |
| f"entity {canonical_id!r} has unsupported kind " | |
| f"{entity.entity_kind!r}" | |
| ) | |
| active_during( | |
| entity.valid_from, | |
| valid_from=entity.valid_from, | |
| valid_to=entity.valid_to, | |
| ) | |
| alias_ids: set[str] = set() | |
| for alias in catalogs.aliases: | |
| if not isinstance(alias.alias_id, str) or not alias.alias_id.strip(): | |
| raise ValueError("alias ID must be a non-empty string") | |
| if not isinstance(alias.surface, str) or not alias.surface.strip(): | |
| raise ValueError(f"alias {alias.alias_id!r} has an empty surface") | |
| if alias.alias_id in alias_ids: | |
| raise ValueError(f"duplicate alias ID: {alias.alias_id}") | |
| alias_ids.add(alias.alias_id) | |
| if alias.canonical_id not in catalogs.entities: | |
| raise ValueError( | |
| f"alias {alias.alias_id!r} references unknown entity " | |
| f"{alias.canonical_id!r}" | |
| ) | |
| if alias.match_mode not in _MATCH_MODES: | |
| raise ValueError( | |
| f"alias {alias.alias_id!r} has unsupported match mode " | |
| f"{alias.match_mode!r}" | |
| ) | |
| if alias.requires_context and alias.context_rule_id not in config.context_rules: | |
| raise ValueError( | |
| f"alias {alias.alias_id!r} requires unknown context rule " | |
| f"{alias.context_rule_id!r}" | |
| ) | |
| active_during( | |
| alias.valid_from, | |
| valid_from=alias.valid_from, | |
| valid_to=alias.valid_to, | |
| ) | |
| rule_ids: set[str] = set() | |
| for rule in catalogs.lexical_rules: | |
| if not isinstance(rule.rule_id, str) or not rule.rule_id.strip(): | |
| raise ValueError("lexical rule ID must be a non-empty string") | |
| if rule.rule_id in rule_ids: | |
| raise ValueError(f"duplicate lexical rule ID: {rule.rule_id}") | |
| rule_ids.add(rule.rule_id) | |
| if not rule.phrases or any( | |
| not isinstance(phrase, str) or not phrase.strip() for phrase in rule.phrases | |
| ): | |
| raise ValueError( | |
| f"lexical rule {rule.rule_id!r} needs non-empty string phrases" | |
| ) | |
| if rule.detection_method not in _DETECTION_METHODS: | |
| raise ValueError( | |
| f"lexical rule {rule.rule_id!r} has unsupported detection method " | |
| f"{rule.detection_method!r}" | |
| ) | |
| if rule.tag_type not in config.ranking_tiers: | |
| raise ValueError( | |
| f"lexical tag type {rule.tag_type!r} has no configured ranking tier" | |
| ) | |
| if ( | |
| isinstance(rule.context_window_chars, bool) | |
| or not isinstance(rule.context_window_chars, int) | |
| or rule.context_window_chars < 0 | |
| ): | |
| raise ValueError( | |
| f"lexical rule {rule.rule_id!r} has an invalid context window" | |
| ) | |
| memberships: dict[str, ExposureFamily] = {} | |
| for family, members in config.exposure_members.items(): | |
| if not isinstance(family, ExposureFamily): | |
| raise ValueError(f"unsupported exposure family: {family!r}") | |
| for canonical_id in members: | |
| if canonical_id not in catalogs.entities: | |
| raise ValueError( | |
| f"exposure family {family.value} references unknown entity " | |
| f"{canonical_id!r}" | |
| ) | |
| previous = memberships.setdefault(canonical_id, family) | |
| if previous is not family: | |
| raise ValueError( | |
| f"entity {canonical_id!r} belongs to both {previous.value} " | |
| f"and {family.value}" | |
| ) | |
| class Taxonomy: | |
| """A validated, immutable configuration and catalog bundle.""" | |
| config: NewsRelevanceConfig | |
| catalogs: CatalogBundle | |
| fingerprint: str = field(init=False) | |
| def __post_init__(self) -> None: | |
| _validate(self.config, self.catalogs) | |
| object.__setattr__( | |
| self, "fingerprint", _fingerprint(self.config, self.catalogs) | |
| ) | |
| def version(self) -> str: | |
| """Return the version of the decision configuration.""" | |
| return self.config.version | |
| def entity_ids(self) -> tuple[str, ...]: | |
| """Return canonical entity identifiers in stable order.""" | |
| return tuple(sorted(self.catalogs.entities)) | |
| def lexical_rule_ids(self) -> tuple[str, ...]: | |
| """Return lexical rule identifiers in stable order.""" | |
| return tuple(sorted(rule.rule_id for rule in self.catalogs.lexical_rules)) | |
| def describe(self) -> dict[str, object]: | |
| """Return a compact, serializable summary of this taxonomy.""" | |
| return { | |
| "version": self.version, | |
| "fingerprint": self.fingerprint, | |
| "entity_catalog_version": self.catalogs.entity_catalog_version, | |
| "lexical_catalog_version": self.catalogs.lexical_catalog_version, | |
| "entities": len(self.catalogs.entities), | |
| "aliases": len(self.catalogs.aliases), | |
| "relationships": len(self.catalogs.relationships), | |
| "lexical_rules": len(self.catalogs.lexical_rules), | |
| "exposure_families": tuple(family.value for family in ExposureFamily), | |
| } | |
| def from_yaml(cls, path: str | Path) -> "Taxonomy": | |
| """Load a full maintained news-relevance configuration file.""" | |
| config = load_config(path) | |
| return Taxonomy(config=config, catalogs=load_catalogs(config)) | |
| def extend( | |
| self, | |
| configure: Callable[["TaxonomyBuilder"], object], | |
| ) -> "Taxonomy": | |
| """Return a new taxonomy after applying one builder callback.""" | |
| if not callable(configure): | |
| raise TypeError("configure must be callable") | |
| builder = TaxonomyBuilder(self) | |
| configure(builder) | |
| return builder.build() | |
| class DefaultTaxonomy(Taxonomy): | |
| """The bundled financial taxonomy, optionally extended by `configure()`.""" | |
| def __init__(self) -> None: | |
| config = load_config() | |
| catalogs = load_catalogs(config) | |
| builder = TaxonomyBuilder._from_parts(config, catalogs) | |
| self.configure(builder) | |
| built = builder.build() | |
| super().__init__(config=built.config, catalogs=built.catalogs) | |
| def configure(self, builder: "TaxonomyBuilder") -> None: | |
| """Override in a subclass to add rules before the taxonomy is frozen.""" | |
| class TaxonomyBuilder: | |
| """Mutable construction surface whose `build()` result is immutable.""" | |
| __slots__ = ( | |
| "_aliases", | |
| "_catalogs", | |
| "_changed_entities", | |
| "_changed_lexical", | |
| "_config", | |
| "_entities", | |
| "_exposure_members", | |
| "_lexical_rules", | |
| ) | |
| def __init__(self, base: Taxonomy) -> None: | |
| if not isinstance(base, Taxonomy): | |
| raise TypeError("base must be a Taxonomy") | |
| self._initialize(base.config, base.catalogs) | |
| def _from_parts( | |
| cls, | |
| config: NewsRelevanceConfig, | |
| catalogs: CatalogBundle, | |
| ) -> "TaxonomyBuilder": | |
| builder = cls.__new__(cls) | |
| builder._initialize(config, catalogs) | |
| return builder | |
| def _initialize( | |
| self, | |
| config: NewsRelevanceConfig, | |
| catalogs: CatalogBundle, | |
| ) -> None: | |
| self._config = config | |
| self._catalogs = catalogs | |
| self._entities = dict(catalogs.entities) | |
| self._aliases = list(catalogs.aliases) | |
| self._lexical_rules = list(catalogs.lexical_rules) | |
| self._exposure_members = { | |
| family: set(members) for family, members in config.exposure_members.items() | |
| } | |
| self._changed_entities = False | |
| self._changed_lexical = False | |
| def add_entity( | |
| self, | |
| *, | |
| canonical_id: str, | |
| canonical_name: str, | |
| entity_concept: str, | |
| entity_kind: str, | |
| valid_from: str, | |
| valid_to: str | None = None, | |
| exposure_family: ExposureFamily | str | None = None, | |
| ) -> "TaxonomyBuilder": | |
| if not canonical_id.strip() or canonical_id in self._entities: | |
| raise ValueError(f"entity ID must be new and non-empty: {canonical_id!r}") | |
| if entity_kind not in _ENTITY_KINDS: | |
| raise ValueError(f"unsupported entity kind: {entity_kind!r}") | |
| entity = CatalogEntity( | |
| canonical_id=canonical_id, | |
| canonical_name=canonical_name, | |
| entity_concept=entity_concept, | |
| entity_kind=entity_kind, | |
| valid_from=valid_from, | |
| valid_to=valid_to, | |
| ) | |
| self._entities[canonical_id] = entity | |
| if exposure_family is not None: | |
| family = ExposureFamily(exposure_family) | |
| self._exposure_members.setdefault(family, set()).add(canonical_id) | |
| self._changed_entities = True | |
| return self | |
| def add_alias( | |
| self, | |
| *, | |
| alias_id: str, | |
| canonical_id: str, | |
| surface: str, | |
| match_mode: str = "normalized_phrase", | |
| requires_context: bool = False, | |
| context_rule_id: str | None = None, | |
| valid_from: str | None = None, | |
| valid_to: str | None = None, | |
| source_id: str = "user-taxonomy", | |
| ) -> "TaxonomyBuilder": | |
| if any(row.alias_id == alias_id for row in self._aliases): | |
| raise ValueError(f"duplicate alias ID: {alias_id}") | |
| entity = self._entities.get(canonical_id) | |
| if entity is None: | |
| raise ValueError(f"alias references unknown entity: {canonical_id}") | |
| if match_mode not in _MATCH_MODES: | |
| raise ValueError(f"unsupported match mode: {match_mode!r}") | |
| if requires_context and context_rule_id not in self._config.context_rules: | |
| raise ValueError(f"unknown context rule: {context_rule_id!r}") | |
| self._aliases.append( | |
| CatalogAlias( | |
| alias_id=alias_id, | |
| canonical_id=canonical_id, | |
| surface=surface, | |
| match_mode=match_mode, | |
| requires_context=requires_context, | |
| context_rule_id=context_rule_id if requires_context else None, | |
| valid_from=valid_from or entity.valid_from, | |
| valid_to=valid_to if valid_to is not None else entity.valid_to, | |
| source_id=source_id, | |
| ) | |
| ) | |
| self._changed_entities = True | |
| return self | |
| def add_lexical_rule( | |
| self, | |
| *, | |
| rule_id: str, | |
| entity_concept: str, | |
| tag_type: str, | |
| phrases: Sequence[str], | |
| target_id: str | None = None, | |
| detection_method: str = "lexical_phrase", | |
| market_wide_authorized: bool = False, | |
| case_sensitive: bool = False, | |
| context_window_chars: int = 160, | |
| context_any: Sequence[str] = (), | |
| event_family: str | None = None, | |
| source_id: str = "user-taxonomy", | |
| ) -> "TaxonomyBuilder": | |
| if any(row.rule_id == rule_id for row in self._lexical_rules): | |
| raise ValueError(f"duplicate lexical rule ID: {rule_id}") | |
| if detection_method not in _DETECTION_METHODS: | |
| raise ValueError(f"unsupported detection method: {detection_method!r}") | |
| if tag_type not in self._config.ranking_tiers: | |
| raise ValueError( | |
| f"lexical tag type {tag_type!r} has no configured ranking tier" | |
| ) | |
| if isinstance(phrases, (str, bytes)): | |
| raise TypeError("phrases must be a sequence of strings, not one string") | |
| phrase_tuple = tuple(phrases) | |
| if not phrase_tuple or any( | |
| not isinstance(item, str) or not item.strip() for item in phrase_tuple | |
| ): | |
| raise ValueError("phrases must contain non-empty strings") | |
| if ( | |
| isinstance(context_window_chars, bool) | |
| or not isinstance(context_window_chars, int) | |
| or context_window_chars < 0 | |
| ): | |
| raise ValueError("context_window_chars must be non-negative") | |
| if isinstance(context_any, (str, bytes)) or any( | |
| not isinstance(item, str) or not item.strip() for item in context_any | |
| ): | |
| raise ValueError("context_any must contain non-empty strings") | |
| self._lexical_rules.append( | |
| LexicalRule( | |
| rule_id=rule_id, | |
| entity_concept=entity_concept, | |
| tag_type=tag_type, | |
| target_id=target_id, | |
| phrases=phrase_tuple, | |
| detection_method=detection_method, | |
| market_wide_authorized=market_wide_authorized, | |
| case_sensitive=case_sensitive, | |
| context_window_chars=context_window_chars, | |
| context_any=tuple(context_any), | |
| event_family=event_family, | |
| source_id=source_id, | |
| ) | |
| ) | |
| self._changed_lexical = True | |
| return self | |
| def build(self) -> Taxonomy: | |
| exposure_members = MappingProxyType( | |
| { | |
| family: frozenset(members) | |
| for family, members in self._exposure_members.items() | |
| } | |
| ) | |
| config = replace(self._config, exposure_members=exposure_members) | |
| entity_version = self._catalogs.entity_catalog_version | |
| lexical_version = self._catalogs.lexical_catalog_version | |
| if self._changed_entities and not entity_version.endswith("+custom"): | |
| entity_version += "+custom" | |
| if self._changed_lexical and not lexical_version.endswith("+custom"): | |
| lexical_version += "+custom" | |
| catalogs = CatalogBundle( | |
| entity_catalog_version=entity_version, | |
| lexical_catalog_version=lexical_version, | |
| entities=MappingProxyType(dict(self._entities)), | |
| aliases=tuple(self._aliases), | |
| relationships=MappingProxyType(dict(self._catalogs.relationships)), | |
| lexical_rules=tuple(self._lexical_rules), | |
| ) | |
| return Taxonomy(config=config, catalogs=catalogs) | |
| class Ranker(ABC): | |
| """Stable ranking template; subclasses only define the domain sort key.""" | |
| def sort_key( | |
| self, | |
| prediction: RelevanceDecision, | |
| accepted_paths: tuple[EvidencePath, ...], | |
| *, | |
| exposure_family: ExposureFamily, | |
| decision_bucket: str, | |
| ) -> tuple[object, ...]: | |
| """Return an ascending key; article ID is appended as the final tie-break.""" | |
| def rank( | |
| self, | |
| predictions: Iterable[RelevanceDecision], | |
| *, | |
| config: NewsRelevanceConfig, | |
| exposure_family: ExposureFamily | str, | |
| decision_bucket: datetime | str, | |
| limit: int | None = None, | |
| ) -> tuple[RankedCandidate, ...]: | |
| if limit is not None and ( | |
| isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0 | |
| ): | |
| raise ValueError("limit must be a positive integer or None") | |
| family = ExposureFamily(exposure_family) | |
| bucket = str(decision_bucket) | |
| grouped: dict[str, tuple[RelevanceDecision, list[EvidencePath]]] = {} | |
| for prediction in predictions: | |
| exposure = prediction.exposure_decisions.get(family) | |
| if ( | |
| exposure is None | |
| or exposure.status is not ExposureRelevanceStatus.SUPPORTED | |
| ): | |
| continue | |
| paths = [ | |
| path | |
| for path in exposure.evidence_paths | |
| if path.status.value == "accepted" | |
| and path.link_status in config.accepted_link_statuses | |
| ] | |
| if not paths: | |
| continue | |
| existing = grouped.get(prediction.article.article_id) | |
| if existing is not None and existing[0] is not prediction: | |
| raise ValueError("one article ID cannot map to multiple predictions") | |
| if existing is None: | |
| grouped[prediction.article.article_id] = (prediction, paths) | |
| else: | |
| existing[1].extend(paths) | |
| sortable = [] | |
| for article_id, (prediction, paths) in grouped.items(): | |
| accepted = tuple(paths) | |
| custom = self.sort_key( | |
| prediction, | |
| accepted, | |
| exposure_family=family, | |
| decision_bucket=bucket, | |
| ) | |
| if not isinstance(custom, tuple): | |
| raise TypeError("Ranker.sort_key() must return a tuple") | |
| sortable.append(((custom, article_id), prediction, accepted)) | |
| ordered = sorted(sortable, key=lambda row: row[0]) | |
| if limit is not None: | |
| ordered = ordered[:limit] | |
| return tuple( | |
| RankedCandidate( | |
| decision_bucket=bucket, | |
| exposure_family=family, | |
| rank=index, | |
| article_id=prediction.article.article_id, | |
| accepted_evidence_ids=tuple( | |
| sorted(path.evidence_id for path in accepted) | |
| ), | |
| decision=prediction, | |
| ) | |
| for index, (_key_row, prediction, accepted) in enumerate(ordered, 1) | |
| ) | |
| class DefaultRanker(Ranker): | |
| """The maintained tier, evidence-diversity, recency ordering.""" | |
| def sort_key( | |
| self, | |
| prediction: RelevanceDecision, | |
| accepted_paths: tuple[EvidencePath, ...], | |
| *, | |
| exposure_family: ExposureFamily, | |
| decision_bucket: str, | |
| ) -> tuple[object, ...]: | |
| del exposure_family, decision_bucket | |
| timestamp = prediction.article.timestamp.astimezone(timezone.utc).timestamp() | |
| return ( | |
| min(path.tier for path in accepted_paths), | |
| -len({path.tag_detail for path in accepted_paths}), | |
| -len({path.detection_method for path in accepted_paths}), | |
| -timestamp, | |
| ) | |
| __all__ = [ | |
| "DefaultRanker", | |
| "DefaultTaxonomy", | |
| "Ranker", | |
| "Taxonomy", | |
| "TaxonomyBuilder", | |
| "copy_default_taxonomy", | |
| ] | |