Spaces:
Running on Zero
Running on Zero
| """Strict configuration loader for the public news-relevance API.""" | |
| from __future__ import annotations | |
| from collections.abc import Mapping | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| from types import MappingProxyType | |
| from typing import Any | |
| import yaml | |
| from .models import ExposureFamily | |
| _PACKAGE_ROOT = Path(__file__).resolve().parent | |
| DEFAULT_CONFIG_PATH = _PACKAGE_ROOT / "configs" / "default.yml" | |
| _TOP_LEVEL_KEYS = { | |
| "version", | |
| "mode", | |
| "catalogs", | |
| "normalization", | |
| "alias_policy_overrides", | |
| "context_rules", | |
| "market_wide_conjunctions", | |
| "exposure_families", | |
| "relationship_paths", | |
| "constituent_evidence_enabled", | |
| "licensed_adapters", | |
| "precedence", | |
| "relevant_k", | |
| "ranked_k", | |
| } | |
| _TIE_BREAKERS = ( | |
| "accepted_tag_details_desc", | |
| "independent_detection_methods_desc", | |
| "article_timestamp_desc", | |
| "article_id_asc", | |
| ) | |
| def _mapping(value: object, name: str) -> Mapping[str, Any]: | |
| if not isinstance(value, Mapping): | |
| raise TypeError(f"{name} must be a mapping") | |
| return value | |
| def _exact_keys(value: Mapping[str, Any], allowed: set[str], name: str) -> None: | |
| unknown = set(value) - allowed | |
| if unknown: | |
| raise ValueError(f"unknown {name} keys: {sorted(unknown)}") | |
| def _catalog_path(value: object, name: str, *, base_dir: Path) -> Path: | |
| if not isinstance(value, str) or not value.strip(): | |
| raise ValueError(f"{name} must be a non-empty path") | |
| resolved = (base_dir / value).resolve() | |
| if not resolved.is_file(): | |
| raise ValueError(f"{name} does not exist: {resolved}") | |
| return resolved | |
| class ContextRule: | |
| window_chars: int | |
| accept_any_clauses: tuple[tuple[str, ...], ...] | |
| reject_any_terms: tuple[str, ...] | |
| class MarketWideConjunctionRule: | |
| actor_phrases: tuple[str, ...] | |
| actor_identifiers: tuple[str, ...] | |
| action_phrases: tuple[str, ...] | |
| object_phrases: tuple[str, ...] | |
| scope_phrases: tuple[str, ...] | |
| entity_concept: str | |
| tag_type: str | |
| target_id: str | |
| event_family: str | |
| class NewsRelevanceConfig: | |
| version: str | |
| mode: str | |
| entity_catalog_path: Path | |
| lexical_catalog_path: Path | |
| alias_policy_overrides: Mapping[str, Mapping[str, Any]] | |
| context_rules: Mapping[str, ContextRule] | |
| market_wide_conjunctions: Mapping[str, MarketWideConjunctionRule] | |
| exposure_members: Mapping[ExposureFamily, frozenset[str]] | |
| relationship_paths: Mapping[str, tuple[str, ...]] | |
| precedence: tuple[str, ...] | |
| accepted_link_statuses: frozenset[str] | |
| ranking_tiers: Mapping[str, int] | |
| ranking_budget: int | |
| constituent_evidence_enabled: bool | |
| def load_config(path: str | Path | None = None) -> NewsRelevanceConfig: | |
| """Load a fail-closed configuration; unknown fields are rejected.""" | |
| config_path = Path(path).resolve() if path is not None else DEFAULT_CONFIG_PATH | |
| catalog_root = config_path.parent if path is not None else _PACKAGE_ROOT | |
| with config_path.open(encoding="utf-8") as handle: | |
| root = yaml.safe_load(handle) | |
| root = _mapping(root, "config") | |
| _exact_keys(root, _TOP_LEVEL_KEYS, "top-level") | |
| if root.get("mode") != "catalog_lite": | |
| raise ValueError("only mode=catalog_lite is currently implemented") | |
| if root.get("constituent_evidence_enabled") is not False: | |
| raise ValueError("catalog_lite must fail closed on constituent evidence") | |
| adapters = _mapping(root.get("licensed_adapters"), "licensed_adapters") | |
| _exact_keys(adapters, {"crsp", "compustat"}, "licensed_adapters") | |
| if any(value is not False for value in adapters.values()): | |
| raise ValueError("licensed adapters are unavailable in the public core") | |
| catalogs = _mapping(root.get("catalogs"), "catalogs") | |
| _exact_keys(catalogs, {"entities", "lexical"}, "catalogs") | |
| normalization = _mapping(root.get("normalization"), "normalization") | |
| _exact_keys(normalization, {"version", "fuzzy_edits"}, "normalization") | |
| if normalization != { | |
| "version": "financial-phrase-nfkc-v2", | |
| "fuzzy_edits": False, | |
| }: | |
| raise ValueError("the conservative v2 normalization contract is required") | |
| overrides = _mapping(root.get("alias_policy_overrides"), "alias_policy_overrides") | |
| override_values: dict[str, Mapping[str, Any]] = {} | |
| for alias_id, raw in overrides.items(): | |
| value = _mapping(raw, f"alias_policy_overrides.{alias_id}") | |
| _exact_keys( | |
| value, {"context_required", "context_rule_id"}, f"alias override {alias_id}" | |
| ) | |
| if not isinstance(value.get("context_required"), bool): | |
| raise TypeError(f"alias override {alias_id} needs boolean context_required") | |
| override_values[str(alias_id)] = MappingProxyType(dict(value)) | |
| rules = _mapping(root.get("context_rules"), "context_rules") | |
| context_rules: dict[str, ContextRule] = {} | |
| for rule_id, raw in rules.items(): | |
| value = _mapping(raw, f"context_rules.{rule_id}") | |
| _exact_keys( | |
| value, | |
| {"window_chars", "accept_any_clauses", "reject_any_terms"}, | |
| f"context rule {rule_id}", | |
| ) | |
| width = value.get("window_chars") | |
| if isinstance(width, bool) or not isinstance(width, int) or width <= 0: | |
| raise ValueError(f"context rule {rule_id} needs positive window_chars") | |
| context_rules[str(rule_id)] = ContextRule( | |
| window_chars=width, | |
| accept_any_clauses=tuple( | |
| tuple(str(term) for term in clause) | |
| for clause in value.get("accept_any_clauses", ()) | |
| ), | |
| reject_any_terms=tuple( | |
| str(term) for term in value.get("reject_any_terms", ()) | |
| ), | |
| ) | |
| raw_conjunctions = _mapping( | |
| root.get("market_wide_conjunctions"), "market_wide_conjunctions" | |
| ) | |
| market_wide_conjunctions: dict[str, MarketWideConjunctionRule] = {} | |
| conjunction_keys = { | |
| "actor_phrases", | |
| "actor_identifiers", | |
| "action_phrases", | |
| "object_phrases", | |
| "scope_phrases", | |
| "entity_concept", | |
| "tag_type", | |
| "target_id", | |
| "event_family", | |
| } | |
| for rule_id, raw in raw_conjunctions.items(): | |
| value = _mapping(raw, f"market_wide_conjunctions.{rule_id}") | |
| _exact_keys(value, conjunction_keys, f"market-wide conjunction {rule_id}") | |
| sequence_fields = ( | |
| "actor_phrases", | |
| "actor_identifiers", | |
| "action_phrases", | |
| "object_phrases", | |
| "scope_phrases", | |
| ) | |
| sequences: dict[str, tuple[str, ...]] = {} | |
| for field in sequence_fields: | |
| raw_values = value.get(field) | |
| if not isinstance(raw_values, list) or not raw_values: | |
| raise TypeError( | |
| f"market-wide conjunction {rule_id}.{field} must be a non-empty list" | |
| ) | |
| sequences[field] = tuple(str(item) for item in raw_values) | |
| scalar_values = {} | |
| for field in ("entity_concept", "tag_type", "target_id", "event_family"): | |
| raw_value = value.get(field) | |
| if not isinstance(raw_value, str) or not raw_value.strip(): | |
| raise TypeError( | |
| f"market-wide conjunction {rule_id}.{field} must be non-empty" | |
| ) | |
| scalar_values[field] = raw_value | |
| if scalar_values["target_id"] != ExposureFamily.MARKET_WIDE.value: | |
| raise ValueError("market-wide conjunction target must be MARKET_WIDE") | |
| market_wide_conjunctions[str(rule_id)] = MarketWideConjunctionRule( | |
| **sequences, | |
| **scalar_values, | |
| ) | |
| raw_families = _mapping(root.get("exposure_families"), "exposure_families") | |
| exposure_members: dict[ExposureFamily, frozenset[str]] = {} | |
| seen_members: set[str] = set() | |
| for family_name, raw in raw_families.items(): | |
| family = ExposureFamily(str(family_name)) | |
| value = _mapping(raw, f"exposure_families.{family_name}") | |
| _exact_keys(value, {"canonical_members"}, f"exposure family {family_name}") | |
| members = frozenset( | |
| str(member) for member in value.get("canonical_members", ()) | |
| ) | |
| if not members or seen_members & members: | |
| raise ValueError("exposure family members must be non-empty and disjoint") | |
| seen_members.update(members) | |
| exposure_members[family] = members | |
| raw_paths = _mapping(root.get("relationship_paths"), "relationship_paths") | |
| relationship_paths = { | |
| str(canonical_id): tuple(str(item) for item in path_value) | |
| for canonical_id, path_value in raw_paths.items() | |
| } | |
| if any(not path_value for path_value in relationship_paths.values()): | |
| raise ValueError("relationship paths cannot be empty") | |
| precedence = _mapping(root.get("precedence"), "precedence") | |
| _exact_keys(precedence, {"detection_methods"}, "precedence") | |
| methods = tuple(str(item) for item in precedence.get("detection_methods", ())) | |
| if not methods or len(methods) != len(set(methods)): | |
| raise ValueError("precedence detection methods must be unique and non-empty") | |
| relevant = _mapping(root.get("relevant_k"), "relevant_k") | |
| _exact_keys(relevant, {"accepted_link_statuses"}, "relevant_k") | |
| link_statuses = frozenset( | |
| str(item) for item in relevant.get("accepted_link_statuses", ()) | |
| ) | |
| if not link_statuses: | |
| raise ValueError("Relevant-K requires accepted link statuses") | |
| ranked = _mapping(root.get("ranked_k"), "ranked_k") | |
| _exact_keys(ranked, {"strategy", "budget", "tiers", "tie_breakers"}, "ranked_k") | |
| if ranked.get("strategy") != "per_bucket_exposure_lexicographic_v2": | |
| raise ValueError("unsupported Ranked-K strategy") | |
| if tuple(ranked.get("tie_breakers", ())) != _TIE_BREAKERS: | |
| raise ValueError("Ranked-K tie-breakers differ from the frozen contract") | |
| budget = ranked.get("budget") | |
| if isinstance(budget, bool) or not isinstance(budget, int) or budget <= 0: | |
| raise ValueError("ranked_k.budget must be a positive integer") | |
| tiers_raw = _mapping(ranked.get("tiers"), "ranked_k.tiers") | |
| tiers = {str(name): int(value) for name, value in tiers_raw.items()} | |
| if any(value < 0 for value in tiers.values()): | |
| raise ValueError("ranking tiers must be non-negative") | |
| missing_conjunction_tiers = { | |
| rule.tag_type | |
| for rule in market_wide_conjunctions.values() | |
| if rule.tag_type not in tiers | |
| } | |
| if missing_conjunction_tiers: | |
| raise ValueError( | |
| "market-wide conjunction tag types need ranking tiers: " | |
| f"{sorted(missing_conjunction_tiers)}" | |
| ) | |
| version = root.get("version") | |
| if not isinstance(version, str) or not version.strip(): | |
| raise ValueError("version must be non-empty") | |
| return NewsRelevanceConfig( | |
| version=version, | |
| mode="catalog_lite", | |
| entity_catalog_path=_catalog_path( | |
| catalogs.get("entities"), "catalogs.entities", base_dir=catalog_root | |
| ), | |
| lexical_catalog_path=_catalog_path( | |
| catalogs.get("lexical"), "catalogs.lexical", base_dir=catalog_root | |
| ), | |
| alias_policy_overrides=MappingProxyType(override_values), | |
| context_rules=MappingProxyType(context_rules), | |
| market_wide_conjunctions=MappingProxyType(market_wide_conjunctions), | |
| exposure_members=MappingProxyType(exposure_members), | |
| relationship_paths=MappingProxyType(relationship_paths), | |
| precedence=methods, | |
| accepted_link_statuses=link_statuses, | |
| ranking_tiers=MappingProxyType(tiers), | |
| ranking_budget=budget, | |
| constituent_evidence_enabled=False, | |
| ) | |