| 1 | """Durable adapters and targeted runtime execution for research readiness. |
| 2 | |
| 3 | The read path in this module performs database and canonical-metadata reads |
| 4 | only. Provider work is confined to ``ExistingResearchCapabilityExecutor`` and |
| 5 | is reachable only from the explicit ensure command. |
| 6 | """ |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | import asyncio |
| 10 | import logging |
| 11 | from dataclasses import dataclass, field, replace |
| 12 | from datetime import datetime, time, timedelta, timezone |
| 13 | from decimal import Decimal |
| 14 | from typing import Any, Mapping, Sequence |
| 15 | from urllib.parse import urlparse |
| 16 | from uuid import UUID |
| 17 | |
| 18 | logger = logging.getLogger(__name__) |
| 19 | |
| 20 | from app.research_applicability import classify_requirements |
| 21 | from app.market_sessions import latest_completed_session, next_session_open |
| 22 | |
| 23 | from app.fact_precedence import FactSourceTier, FinancialFact |
| 24 | from app.historical_market_data import has_year_historical_coverage |
| 25 | from app.models import ( |
| 26 | CompanyResearchProfile, |
| 27 | DocumentStatus, |
| 28 | EventImpact, |
| 29 | ResearchDocument, |
| 30 | ResearchEvent, |
| 31 | ResearchEventType, |
| 32 | ResearchLifecycleStatus, |
| 33 | ShareholdingCategory, |
| 34 | SourceClassification, |
| 35 | SourceMode, |
| 36 | StructuredMarketSnapshotRecord, |
| 37 | ) |
| 38 | from app.research_readiness import ( |
| 39 | DurableResearchSnapshot, |
| 40 | ProviderAuthorityRegistry, |
| 41 | ResearchEvidence, |
| 42 | ResearchReadinessResult, |
| 43 | ResearchReadinessService, |
| 44 | ResearchRefreshPlan, |
| 45 | ResearchRefreshPlanner, |
| 46 | ResearchRefreshTarget, |
| 47 | ResearchRequirement, |
| 48 | ResearchRequirementImportance, |
| 49 | ResearchRequirementRegistry, |
| 50 | ResearchRequirementStatus, |
| 51 | ResearchSourceTier, |
| 52 | RuleEngineArea, |
| 53 | ) |
| 54 | |
| 55 | |
| 56 | _FINANCIAL_REQUIREMENTS = frozenset( |
| 57 | { |
| 58 | "BUSINESS_QUALITY_FACTS", |
| 59 | "GROWTH_FACTS", |
| 60 | "BALANCE_SHEET_FACTS", |
| 61 | "QUARTERLY_FINANCIALS", |
| 62 | } |
| 63 | ) |
| 64 | _KNOWN_REQUIREMENTS = frozenset( |
| 65 | item.requirement_id for item in ResearchRequirementRegistry.default().requirements |
| 66 | ) |
| 67 | _ORDER_EVENTS = frozenset( |
| 68 | { |
| 69 | ResearchEventType.NEW_ORDER, |
| 70 | ResearchEventType.ORDER_BACKLOG_CHANGE, |
| 71 | ResearchEventType.MAJOR_CONTRACT, |
| 72 | ResearchEventType.GOVERNMENT_CONTRACT, |
| 73 | ResearchEventType.ORDER_CANCELLED, |
| 74 | ResearchEventType.CAPEX, |
| 75 | ResearchEventType.FACTORY_EXPANSION, |
| 76 | ResearchEventType.CAPACITY_EXPANSION, |
| 77 | ResearchEventType.NEW_FACILITY, |
| 78 | ResearchEventType.PROJECT_DELAY, |
| 79 | ResearchEventType.MANAGEMENT_GUIDANCE, |
| 80 | ResearchEventType.GUIDANCE_RAISED, |
| 81 | ResearchEventType.GUIDANCE_LOWERED, |
| 82 | ResearchEventType.GUIDANCE_CUT, |
| 83 | ResearchEventType.GUIDANCE_MAINTAINED, |
| 84 | ResearchEventType.REVENUE_GUIDANCE, |
| 85 | ResearchEventType.MARGIN_GUIDANCE, |
| 86 | } |
| 87 | ) |
| 88 | _GOVERNANCE_EVENTS = frozenset( |
| 89 | { |
| 90 | ResearchEventType.MANAGEMENT_CHANGE, |
| 91 | ResearchEventType.REGULATORY_EVENT, |
| 92 | ResearchEventType.CREDIT_RATING, |
| 93 | ResearchEventType.BORROWING_CHANGE, |
| 94 | } |
| 95 | ) |
| 96 | _GOVERNANCE_TERMS = ( |
| 97 | "auditor", |
| 98 | "governance", |
| 99 | "regulatory", |
| 100 | "litigation", |
| 101 | "fraud", |
| 102 | "promoter", |
| 103 | "management change", |
| 104 | ) |
| 105 | _MACRO_TERMS = ( |
| 106 | "geopolitical", |
| 107 | "war", |
| 108 | "sanction", |
| 109 | "tariff", |
| 110 | "supply chain", |
| 111 | "commodity", |
| 112 | "currency", |
| 113 | "interest rate", |
| 114 | ) |
| 115 | |
| 116 | |
| 117 | class RepositoryResearchReadinessAdapter: |
| 118 | """Translate existing durable records into provider-neutral evidence.""" |
| 119 | |
| 120 | def __init__(self, repository) -> None: |
| 121 | self.repository = repository |
| 122 | self._canonical_metadata: dict[UUID, dict[str, Any]] = {} |
| 123 | self._evaluation_times: dict[UUID, datetime] = {} |
| 124 | self._refreshing: dict[UUID, frozenset[str]] = {} |
| 125 | self._failures: dict[UUID, dict[str, str]] = {} |
| 126 | self._sessions: dict[UUID, tuple] = {} |
| 127 | |
| 128 | def remember_canonical_metadata( |
| 129 | self, global_instrument_id: UUID, metadata: Mapping[str, Any] |
| 130 | ) -> None: |
| 131 | self._canonical_metadata[global_instrument_id] = dict(metadata) |
| 132 | |
| 133 | def canonical_metadata_for(self, global_instrument_id: UUID) -> dict[str, Any]: |
| 134 | """Return public canonical metadata without exposing mutable adapter state.""" |
| 135 | return dict(self._canonical_metadata.get(global_instrument_id, {})) |
| 136 | |
| 137 | def mark_refreshing(self, global_instrument_id: UUID, requirement_ids: Sequence[str]) -> None: |
| 138 | self._refreshing[global_instrument_id] = frozenset( |
| 139 | str(value).strip().upper() for value in requirement_ids |
| 140 | ) |
| 141 | |
| 142 | def finish_refresh( |
| 143 | self, |
| 144 | global_instrument_id: UUID, |
| 145 | failures: Mapping[str, str] | None = None, |
| 146 | ) -> None: |
| 147 | self._refreshing.pop(global_instrument_id, None) |
| 148 | if failures: |
| 149 | self._failures[global_instrument_id] = { |
| 150 | str(key).strip().upper(): str(value) for key, value in failures.items() |
| 151 | } |
| 152 | else: |
| 153 | self._failures.pop(global_instrument_id, None) |
| 154 | |
| 155 | def load_by_global_instrument_id( |
| 156 | self, |
| 157 | global_instrument_id: UUID, |
| 158 | requirements: Sequence[ResearchRequirement], |
| 159 | ) -> DurableResearchSnapshot: |
| 160 | profile = self.repository.profile(global_instrument_id) |
| 161 | facts = self.repository.financial_facts_for(global_instrument_id) |
| 162 | structured = self.repository.structured_market_snapshots_for( |
| 163 | {global_instrument_id} |
| 164 | ).get(global_instrument_id, []) |
| 165 | observations = self.repository.market_price_observations_for( |
| 166 | {global_instrument_id} |
| 167 | ).get(global_instrument_id, []) |
| 168 | documents = self.repository.documents_for( |
| 169 | global_instrument_id, source_mode=SourceMode.REAL |
| 170 | ) |
| 171 | events = self.repository.events_for( |
| 172 | global_instrument_id, source_mode=SourceMode.REAL |
| 173 | ) |
| 174 | shareholding = self.repository.shareholding_for(global_instrument_id, limit=4) |
| 175 | |
| 176 | evidence: dict[str, list[ResearchEvidence]] = { |
| 177 | requirement.requirement_id: [] for requirement in requirements |
| 178 | } |
| 179 | self._append_financial_evidence(evidence, facts) |
| 180 | self._append_structured_evidence(evidence, structured) |
| 181 | self._append_market_observations(evidence, observations) |
| 182 | from app.valuation_evidence import materialize_valuation |
| 183 | valuation_now = self._evaluation_times.get(global_instrument_id, datetime.now(timezone.utc)) |
| 184 | for name, value in materialize_valuation(structured, observations, now=valuation_now).items(): |
| 185 | evidence['VALUATION_INPUTS'].append(ResearchEvidence(evidence_id='derived-valuation:'+name+':'+str(value.as_of_date), |
| 186 | requirement_id='VALUATION_INPUTS',source='LICENSED_STRUCTURED',source_tier=ResearchSourceTier.LICENSED_STRUCTURED, |
| 187 | retrieved_at=value.retrieved_at,as_of=value.as_of_date,value_fingerprint=str(value.value), |
| 188 | source_url=value.source_url,covered_input_ids=('PE' if name=='trailingPE' else 'PB',))) |
| 189 | self._append_documents(evidence, documents) |
| 190 | self._append_events(evidence, events) |
| 191 | self._append_shareholding(evidence, shareholding) |
| 192 | self._append_canonical_sector(evidence, global_instrument_id, profile) |
| 193 | |
| 194 | metadata = self.canonical_metadata_for(global_instrument_id) |
| 195 | sector = metadata.get("canonicalSector") or metadata.get("sector") |
| 196 | industry = metadata.get("officialIndustry") or metadata.get("industry") |
| 197 | classification_source = "CANONICAL_REFERENCE" if industry else None |
| 198 | for record in sorted(structured, key=lambda record: (int(_structured_source_tier(record.provider) != ResearchSourceTier.OFFICIAL), -record.retrieved_at.timestamp())): |
| 199 | facts_by_name = record.snapshot.facts |
| 200 | sector_fact, industry_fact = facts_by_name.get("sector"), facts_by_name.get("industry") |
| 201 | sector = sector or (sector_fact.value if sector_fact else None) |
| 202 | if not industry and industry_fact and industry_fact.value: |
| 203 | industry, classification_source = str(industry_fact.value), industry_fact.source_url |
| 204 | applicability = classify_requirements(str(sector) if sector else None, str(industry) if industry else None, classification_source) |
| 205 | schedules, exceptions = self._sessions.get(global_instrument_id, ([], [])) |
| 206 | if schedules: |
| 207 | market = profile.mic or profile.exchange |
| 208 | for requirement_id in ("LATEST_PRICE", "VALUATION_INPUTS"): |
| 209 | values = [] |
| 210 | for item in evidence[requirement_id]: |
| 211 | if item.as_of: |
| 212 | session = latest_completed_session(market, schedules, exceptions, item.as_of + timedelta(minutes=15)) |
| 213 | # Only a real close observation is reusable during a closed session. |
| 214 | if session and abs((session - item.as_of).total_seconds()) <= 15 * 60: |
| 215 | valid_until = next_session_open(market, schedules, exceptions, session) |
| 216 | if valid_until and ('LATEST_USABLE_PRICE' in item.covered_input_ids or |
| 217 | item.evidence_id.startswith('derived-valuation:')): |
| 218 | item = replace(item, valid_until=valid_until) |
| 219 | values.append(item) |
| 220 | evidence[requirement_id] = values |
| 221 | |
| 222 | acquisition = {} |
| 223 | observations_loader = getattr(self.repository, "acquisition_observations_for", None) |
| 224 | if callable(observations_loader): |
| 225 | for observation in observations_loader(global_instrument_id): |
| 226 | key = observation["requirement_id"] |
| 227 | history = acquisition.get(key, {}).get("history", []) |
| 228 | acquisition[key] = {**observation, "history": [*history, observation]} |
| 229 | news_loader = getattr(self.repository, 'news_records_for', None) |
| 230 | if callable(news_loader): |
| 231 | from app.news_intelligence import SearchRun, search_state |
| 232 | evaluated_at = self._evaluation_times.get(global_instrument_id, datetime.now(timezone.utc)) |
| 233 | runs = news_loader(global_instrument_id, SearchRun, as_of=evaluated_at) |
| 234 | if runs: |
| 235 | run = runs[-1] |
| 236 | state = search_state(run, evaluated_at) |
| 237 | acquisition['CURRENT_NEWS'] = {'news_readiness':state, 'coverage':run.coverage, |
| 238 | 'run_id':str(run.run_id), 'observed_at':run.completed_at.isoformat(), 'history':[]} |
| 239 | if state in {'READY_WITH_EVENTS','READY_NO_EVENTS'}: |
| 240 | evidence['CURRENT_NEWS'] = [ResearchEvidence(evidence_id='search-run:'+str(run.run_id), |
| 241 | requirement_id='CURRENT_NEWS',source='SEARCH_COVERAGE',source_tier=ResearchSourceTier.APPROVED_SECONDARY, |
| 242 | retrieved_at=run.completed_at,as_of=run.completed_at,event_date=run.completed_at, |
| 243 | valid_until=run.completed_at+timedelta(days=1),confidence=run.coverage, |
| 244 | covered_input_ids=('RELEVANT_CURRENT_EVENT_EVIDENCE',))] |
| 245 | news_checks = [row for row in acquisition.get("CURRENT_NEWS", {}).get("history", []) |
| 246 | if row.get("outcome") in {"SUCCESS", "SUCCESS_EMPTY"}] |
| 247 | if news_checks: |
| 248 | checked_at = _aware_datetime(news_checks[-1].get("observed_at")) |
| 249 | if checked_at: |
| 250 | evidence["CURRENT_NEWS"] = [replace(item, valid_until=checked_at + timedelta(days=1)) |
| 251 | for item in evidence["CURRENT_NEWS"]] |
| 252 | persisted_failures = {key: value["failure_reason"] for key, value in acquisition.items() |
| 253 | if value.get("outcome") == "FAILED" and value.get("failure_reason")} |
| 254 | supported = set(evidence) |
| 255 | if not _is_india(profile) and not shareholding: |
| 256 | supported.discard("SHAREHOLDING") |
| 257 | return DurableResearchSnapshot( |
| 258 | global_instrument_id, |
| 259 | {key: tuple(_deduplicate_evidence(values)) for key, values in evidence.items()}, |
| 260 | supported_requirement_ids=frozenset(supported), |
| 261 | refreshing_requirement_ids=self._refreshing.get(global_instrument_id, frozenset()), |
| 262 | failure_reasons={**persisted_failures, **self._failures.get(global_instrument_id, {})}, |
| 263 | acquisition_observations=acquisition, |
| 264 | applicability_by_requirement=applicability, |
| 265 | ) |
| 266 | |
| 267 | @staticmethod |
| 268 | def _append_financial_evidence( |
| 269 | evidence: dict[str, list[ResearchEvidence]], facts: Sequence[FinancialFact] |
| 270 | ) -> None: |
| 271 | real = [fact for fact in facts if fact.source_mode == SourceMode.REAL] |
| 272 | periods_by_series: dict[tuple[str, str], dict[str, set[str]]] = {} |
| 273 | quarterly_metrics: dict[str, dict[str, set[str]]] = {} |
| 274 | all_quarterly_periods: set[str] = set() |
| 275 | annual_periods: set[str] = set() |
| 276 | for fact in real: |
| 277 | metric = _metric(fact.key.metric) |
| 278 | if not fact.key.period_end: |
| 279 | continue |
| 280 | period = fact.key.period_end[:10] |
| 281 | basis = fact.key.reporting_basis or "UNKNOWN" |
| 282 | periods_by_series.setdefault((fact.key.period_type, basis), {}).setdefault(metric, set()).add(period) |
| 283 | if fact.key.period_type == "QUARTERLY": |
| 284 | all_quarterly_periods.add(period) |
| 285 | family = "revenue" if metric in {"revenue", "total_revenue"} else "pat" if metric in {"pat", "net_income", "net_profit"} else metric |
| 286 | quarterly_metrics.setdefault(basis, {}).setdefault(family, set()).add(period) |
| 287 | elif fact.key.period_type == "ANNUAL": |
| 288 | annual_periods.add(period) |
| 289 | latest_quarter = max(all_quarterly_periods, default=None) |
| 290 | |
| 291 | for fact in real: |
| 292 | metric = _metric(fact.key.metric) |
| 293 | coverage = _financial_fact_coverage( |
| 294 | fact, |
| 295 | metric, |
| 296 | periods_by_series.get((fact.key.period_type, fact.key.reporting_basis or "UNKNOWN"), {}), |
| 297 | quarterly_metrics.get(fact.key.reporting_basis or "UNKNOWN", {}).get("revenue", set()) |
| 298 | & quarterly_metrics.get(fact.key.reporting_basis or "UNKNOWN", {}).get("pat", set()), |
| 299 | annual_periods, |
| 300 | latest_quarter, |
| 301 | ) |
| 302 | for requirement_id, covered_inputs in coverage.items(): |
| 303 | evidence[requirement_id].append( |
| 304 | _evidence_from_financial_fact( |
| 305 | fact, requirement_id, tuple(sorted(covered_inputs)) |
| 306 | ) |
| 307 | ) |
| 308 | |
| 309 | @staticmethod |
| 310 | def _append_structured_evidence( |
| 311 | evidence: dict[str, list[ResearchEvidence]], |
| 312 | records: Sequence[StructuredMarketSnapshotRecord], |
| 313 | ) -> None: |
| 314 | for record in records: |
| 315 | for fact_name, value in record.snapshot.facts.items(): |
| 316 | if fact_name == "latestPrice" and not _positive_number(value.value): |
| 317 | continue |
| 318 | coverage = _structured_fact_coverage(fact_name, record.snapshot.facts) |
| 319 | for requirement_id, covered_inputs in coverage.items(): |
| 320 | evidence[requirement_id].append( |
| 321 | ResearchEvidence( |
| 322 | evidence_id=( |
| 323 | f"structured:{record.provider}:{fact_name}:" |
| 324 | f"{record.retrieved_at.isoformat()}" |
| 325 | ), |
| 326 | requirement_id=requirement_id, |
| 327 | source=_market_source(record.provider), |
| 328 | source_tier=_structured_source_tier(record.provider), |
| 329 | retrieved_at=value.retrieved_at or record.retrieved_at, |
| 330 | as_of=value.as_of_date or record.market_as_of, |
| 331 | published_at=value.published_at, |
| 332 | fact_key=f"{fact_name}:{_date_key(value.as_of_date or record.market_as_of)}", |
| 333 | value_fingerprint=_fingerprint(value.value), |
| 334 | complete=value.value is not None, |
| 335 | confidence=value.confidence, |
| 336 | source_url=value.source_url or record.source_url, |
| 337 | covered_input_ids=tuple(sorted(covered_inputs)), |
| 338 | valid_until=((value.as_of_date or value.published_at or value.retrieved_at)+timedelta(days=120) |
| 339 | if requirement_id=='VALUATION_INPUTS' and fact_name in {'trailingEps','forwardEps','bookValue'} else None), |
| 340 | ) |
| 341 | ) |
| 342 | |
| 343 | @staticmethod |
| 344 | def _append_market_observations( |
| 345 | evidence: dict[str, list[ResearchEvidence]], observations: Sequence[Any] |
| 346 | ) -> None: |
| 347 | usable = sorted( |
| 348 | ( |
| 349 | value |
| 350 | for value in observations |
| 351 | if value.price is not None |
| 352 | and Decimal(str(value.price)).is_finite() |
| 353 | and Decimal(str(value.price)) > 0 |
| 354 | ), |
| 355 | key=lambda value: value.observed_at, |
| 356 | ) |
| 357 | if not usable: |
| 358 | return |
| 359 | latest = usable[-1] |
| 360 | source = _market_source(latest.provider) |
| 361 | source_tier = _structured_source_tier(latest.provider) |
| 362 | price_evidence = ResearchEvidence( |
| 363 | evidence_id=f"price:{latest.provider}:{latest.observed_at.isoformat()}", |
| 364 | requirement_id="LATEST_PRICE", |
| 365 | source=source, |
| 366 | source_tier=source_tier, |
| 367 | retrieved_at=latest.retrieved_at, |
| 368 | as_of=latest.observed_at, |
| 369 | fact_key=f"latest-price:{latest.observed_at.isoformat()}", |
| 370 | value_fingerprint=_fingerprint(latest.price), |
| 371 | source_url=latest.source_url, |
| 372 | covered_input_ids=("LATEST_USABLE_PRICE",), |
| 373 | ) |
| 374 | evidence["LATEST_PRICE"].append(price_evidence) |
| 375 | evidence["VALUATION_INPUTS"].append( |
| 376 | _copy_evidence( |
| 377 | price_evidence, |
| 378 | "VALUATION_INPUTS", |
| 379 | ("LATEST_USABLE_PRICE",), |
| 380 | ) |
| 381 | ) |
| 382 | covered = {"DURABLE_PRICE_OBSERVATIONS"} |
| 383 | if len(usable) >= 50: |
| 384 | covered.add("FIFTY_OBSERVATION_TECHNICAL_BASIS") |
| 385 | if len(usable) >= 150: |
| 386 | covered.add("ONE_HUNDRED_FIFTY_OBSERVATION_TECHNICAL_BASIS") |
| 387 | evidence["HISTORICAL_PRICE_SERIES"].append( |
| 388 | ResearchEvidence( |
| 389 | evidence_id=( |
| 390 | f"price-series:{latest.provider}:{usable[0].observed_at.isoformat()}:" |
| 391 | f"{latest.observed_at.isoformat()}:{len(usable)}" |
| 392 | ), |
| 393 | requirement_id="HISTORICAL_PRICE_SERIES", |
| 394 | source=source, |
| 395 | source_tier=source_tier, |
| 396 | retrieved_at=max(item.retrieved_at for item in usable), |
| 397 | as_of=latest.observed_at, |
| 398 | source_url=latest.source_url, |
| 399 | covered_input_ids=tuple(sorted(covered)), |
| 400 | ) |
| 401 | ) |
| 402 | |
| 403 | @staticmethod |
| 404 | def _append_documents( |
| 405 | evidence: dict[str, list[ResearchEvidence]], documents: Sequence[ResearchDocument] |
| 406 | ) -> None: |
| 407 | for document in documents: |
| 408 | if document.status not in {DocumentStatus.PARSED, DocumentStatus.PROCESSED}: |
| 409 | continue |
| 410 | text = f"{document.title or ''} {document.normalized_text or ''}".casefold() |
| 411 | latest_fact_period = max((item.as_of for item in evidence["QUARTERLY_FINANCIALS"] |
| 412 | if item.as_of and "LATEST_QUARTERLY_RESULT" in item.covered_input_ids), default=None) |
| 413 | document_at = document.published_at or document.retrieved_at |
| 414 | if (any(term in text for term in ("financial result", "quarterly result", "earnings")) |
| 415 | and (latest_fact_period is None or document_at >= latest_fact_period)): |
| 416 | evidence["QUARTERLY_FINANCIALS"].append( |
| 417 | _evidence_from_document( |
| 418 | document, |
| 419 | "QUARTERLY_FINANCIALS", |
| 420 | ("LATEST_QUARTERLY_RESULT",), |
| 421 | ) |
| 422 | ) |
| 423 | if any(term in text for term in _GOVERNANCE_TERMS): |
| 424 | evidence["GOVERNANCE_HISTORY"].append( |
| 425 | _evidence_from_document( |
| 426 | document, |
| 427 | "GOVERNANCE_HISTORY", |
| 428 | ("GOVERNANCE_EVIDENCE",), |
| 429 | unresolved=any( |
| 430 | term in text for term in ("litigation", "fraud", "regulatory") |
| 431 | ), |
| 432 | ) |
| 433 | ) |
| 434 | if any(term in text for term in _MACRO_TERMS): |
| 435 | evidence["SECTOR_MACRO"].append( |
| 436 | _evidence_from_document( |
| 437 | document, |
| 438 | "SECTOR_MACRO", |
| 439 | ("RELEVANT_MACRO_EVENT_EXPOSURE",), |
| 440 | ) |
| 441 | ) |
| 442 | |
| 443 | @staticmethod |
| 444 | def _append_events( |
| 445 | evidence: dict[str, list[ResearchEvidence]], events: Sequence[ResearchEvent] |
| 446 | ) -> None: |
| 447 | seen_news: set[tuple[str, str, str]] = set() |
| 448 | for event in events: |
| 449 | event_key = ( |
| 450 | event.source_url.casefold(), |
| 451 | event.title.strip().casefold(), |
| 452 | _date_key(event.event_date or event.published_at), |
| 453 | ) |
| 454 | if event_key not in seen_news: |
| 455 | seen_news.add(event_key) |
| 456 | evidence["CURRENT_NEWS"].append( |
| 457 | _evidence_from_event( |
| 458 | event, |
| 459 | "CURRENT_NEWS", |
| 460 | ("RELEVANT_CURRENT_EVENT_EVIDENCE",), |
| 461 | ) |
| 462 | ) |
| 463 | if event.event_type in _ORDER_EVENTS: |
| 464 | covered = {"MATERIAL_CATALYST_EVIDENCE"} |
| 465 | if event.event_type in { |
| 466 | ResearchEventType.NEW_ORDER, |
| 467 | ResearchEventType.ORDER_BACKLOG_CHANGE, |
| 468 | ResearchEventType.MAJOR_CONTRACT, |
| 469 | ResearchEventType.GOVERNMENT_CONTRACT, |
| 470 | ResearchEventType.ORDER_CANCELLED, |
| 471 | }: |
| 472 | covered.add("ORDER_BOOK_OR_MAJOR_CONTRACT") |
| 473 | if event.event_type in { |
| 474 | ResearchEventType.CAPEX, |
| 475 | ResearchEventType.FACTORY_EXPANSION, |
| 476 | ResearchEventType.CAPACITY_EXPANSION, |
| 477 | ResearchEventType.NEW_FACILITY, |
| 478 | ResearchEventType.PROJECT_DELAY, |
| 479 | }: |
| 480 | covered.add("CAPACITY_OR_CAPEX_OR_COMMISSIONING") |
| 481 | if "GUIDANCE" in str(event.event_type): |
| 482 | covered.add("MANAGEMENT_GUIDANCE") |
| 483 | evidence["ORDER_BOOK_CAPEX_GUIDANCE"].append( |
| 484 | _evidence_from_event( |
| 485 | event, |
| 486 | "ORDER_BOOK_CAPEX_GUIDANCE", |
| 487 | tuple(sorted(covered)), |
| 488 | ) |
| 489 | ) |
| 490 | if event.event_type in _GOVERNANCE_EVENTS: |
| 491 | unresolved = ( |
| 492 | event.status != ResearchLifecycleStatus.REJECTED |
| 493 | and event.impact |
| 494 | in { |
| 495 | EventImpact.NEGATIVE, |
| 496 | EventImpact.STRONG_NEGATIVE, |
| 497 | EventImpact.UNCERTAIN, |
| 498 | } |
| 499 | ) |
| 500 | evidence["GOVERNANCE_HISTORY"].append( |
| 501 | _evidence_from_event( |
| 502 | event, |
| 503 | "GOVERNANCE_HISTORY", |
| 504 | ("GOVERNANCE_EVIDENCE",), |
| 505 | unresolved=unresolved, |
| 506 | ) |
| 507 | ) |
| 508 | |
| 509 | @staticmethod |
| 510 | def _append_shareholding( |
| 511 | evidence: dict[str, list[ResearchEvidence]], snapshots: Sequence[Any] |
| 512 | ) -> None: |
| 513 | if not snapshots: |
| 514 | return |
| 515 | snapshot = snapshots[0] |
| 516 | categories = {str(value.category) for value in snapshot.values} |
| 517 | covered = {"LATEST_VALID_SHAREHOLDING_PERIOD"} |
| 518 | if categories & { |
| 519 | ShareholdingCategory.PROMOTER.value, |
| 520 | ShareholdingCategory.FII_FPI.value, |
| 521 | ShareholdingCategory.DII.value, |
| 522 | ShareholdingCategory.PUBLIC_RETAIL.value, |
| 523 | }: |
| 524 | covered.add("PROMOTER_INSTITUTIONAL_PUBLIC_CATEGORIES") |
| 525 | if ShareholdingCategory.PROMOTER_PLEDGE.value in categories: |
| 526 | covered.add("PROMOTER_PLEDGE") |
| 527 | evidence["SHAREHOLDING"].append( |
| 528 | ResearchEvidence( |
| 529 | evidence_id=f"shareholding:{snapshot.id}", |
| 530 | requirement_id="SHAREHOLDING", |
| 531 | source=( |
| 532 | "NSE" |
| 533 | if snapshot.source_provider.upper() == "NSE" |
| 534 | else "APPROVED_EXTERNAL_TOOL" |
| 535 | if snapshot.source_provider.upper() == "YAHOO_FINANCE_MCP" |
| 536 | else snapshot.source_provider |
| 537 | ), |
| 538 | source_tier=( |
| 539 | ResearchSourceTier.OFFICIAL |
| 540 | if snapshot.source_provider.upper() == "NSE" |
| 541 | else ResearchSourceTier.APPROVED_EXTERNAL_TOOL |
| 542 | if snapshot.source_provider.upper() == "YAHOO_FINANCE_MCP" |
| 543 | else ResearchSourceTier.LICENSED_STRUCTURED |
| 544 | ), |
| 545 | retrieved_at=snapshot.retrieved_at, |
| 546 | as_of=snapshot.period_end, |
| 547 | published_at=snapshot.published_at, |
| 548 | source_url=snapshot.source_url, |
| 549 | confidence=float(snapshot.confidence), |
| 550 | covered_input_ids=tuple(sorted(covered)), |
| 551 | ) |
| 552 | ) |
| 553 | |
| 554 | def _append_canonical_sector( |
| 555 | self, |
| 556 | evidence: dict[str, list[ResearchEvidence]], |
| 557 | instrument_id: UUID, |
| 558 | profile: CompanyResearchProfile, |
| 559 | ) -> None: |
| 560 | metadata = self._canonical_metadata.get(instrument_id, {}) |
| 561 | sector = ( |
| 562 | metadata.get("canonicalSector") |
| 563 | or metadata.get("sector") |
| 564 | or metadata.get("industrySector") |
| 565 | ) |
| 566 | if not sector: |
| 567 | return |
| 568 | retrieved_at = _aware_datetime( |
| 569 | metadata.get("updatedAt") or metadata.get("retrievedAt") |
| 570 | ) or datetime.now(timezone.utc) |
| 571 | evidence["SECTOR_MACRO"].append( |
| 572 | ResearchEvidence( |
| 573 | evidence_id=f"canonical-sector:{instrument_id}:{str(sector).strip().casefold()}", |
| 574 | requirement_id="SECTOR_MACRO", |
| 575 | source="EXCHANGE_OR_INDEX_PROVIDER", |
| 576 | source_tier=ResearchSourceTier.TRUSTED_MARKET_DATA, |
| 577 | retrieved_at=retrieved_at, |
| 578 | as_of=retrieved_at, |
| 579 | fact_key="canonical-sector", |
| 580 | value_fingerprint=str(sector).strip(), |
| 581 | covered_input_ids=("CANONICAL_SECTOR",), |
| 582 | ) |
| 583 | ) |
| 584 | |
| 585 | |
| 586 | @dataclass(frozen=True) |
| 587 | class CapabilityExecutionResult: |
| 588 | executed_capabilities: tuple[str, ...] = () |
| 589 | failures: Mapping[str, str] = field(default_factory=dict) |
| 590 | satisfied_requirement_ids: tuple[str, ...] = () |
| 591 | |
| 592 | def __post_init__(self) -> None: |
| 593 | object.__setattr__(self, "failures", dict(self.failures or {})) |
| 594 | object.__setattr__( |
| 595 | self, |
| 596 | "satisfied_requirement_ids", |
| 597 | tuple(dict.fromkeys(self.satisfied_requirement_ids or ())), |
| 598 | ) |
| 599 | |
| 600 | |
| 601 | @dataclass |
| 602 | class CapabilityExecutionProgress: |
| 603 | """Request-local acquisition progress retained if the budget cancels execution.""" |
| 604 | |
| 605 | executed_capabilities: list[str] = field(default_factory=list) |
| 606 | failures: dict[str, str] = field(default_factory=dict) |
| 607 | satisfied_requirement_ids: set[str] = field(default_factory=set) |
| 608 | |
| 609 | def executed(self, capability: str) -> None: |
| 610 | if capability not in self.executed_capabilities: |
| 611 | self.executed_capabilities.append(capability) |
| 612 | |
| 613 | def failed(self, requirement_id: str, reason: str) -> None: |
| 614 | self.failures[requirement_id] = _combined_failure_reason( |
| 615 | self.failures.get(requirement_id), reason |
| 616 | ) |
| 617 | |
| 618 | def satisfied(self, requirement_id: str) -> None: |
| 619 | self.satisfied_requirement_ids.add(requirement_id) |
| 620 | |
| 621 | |
| 622 | class ExistingResearchCapabilityExecutor: |
| 623 | """Map planner targets to the narrow provider capabilities already present.""" |
| 624 | |
| 625 | def __init__(self, repository, orchestrator, market_data_population_jobs) -> None: |
| 626 | self.repository = repository |
| 627 | self.orchestrator = orchestrator |
| 628 | self.market_data_population_jobs = market_data_population_jobs |
| 629 | |
| 630 | async def execute_primary( |
| 631 | self, |
| 632 | global_instrument_id: UUID, |
| 633 | targets: Sequence[ResearchRefreshTarget], |
| 634 | *, |
| 635 | jurisdiction: str, |
| 636 | correlation_id: str | None, |
| 637 | identity_headers: Mapping[str, str | None] | None, |
| 638 | progress: CapabilityExecutionProgress | None = None, |
| 639 | ) -> CapabilityExecutionResult: |
| 640 | requirement_ids = {target.requirement_id for target in targets} |
| 641 | unknown = requirement_ids - _KNOWN_REQUIREMENTS |
| 642 | if unknown: |
| 643 | raise KeyError(f"No targeted capability mapping for {sorted(unknown)}") |
| 644 | executed: list[str] = [] |
| 645 | failures: dict[str, str] = {} |
| 646 | |
| 647 | structured_classes: set[str] = set() |
| 648 | if "VALUATION_INPUTS" in requirement_ids: |
| 649 | structured_classes.update({"PRICE", "VALUATION", "FUNDAMENTALS"}) |
| 650 | if "LATEST_PRICE" in requirement_ids: |
| 651 | structured_classes.add("PRICE") |
| 652 | if "SECTOR_MACRO" in requirement_ids: |
| 653 | structured_classes.add("FUNDAMENTALS") |
| 654 | if structured_classes: |
| 655 | executed.append("STRUCTURED_MARKET") |
| 656 | if progress is not None: |
| 657 | progress.executed("STRUCTURED_MARKET") |
| 658 | try: |
| 659 | outcome = await self.orchestrator.ensure_structured_market( |
| 660 | global_instrument_id, structured_classes |
| 661 | ) |
| 662 | if outcome.error: |
| 663 | for requirement_id in requirement_ids & { |
| 664 | "VALUATION_INPUTS", "LATEST_PRICE", "SECTOR_MACRO" |
| 665 | }: |
| 666 | failures[requirement_id] = outcome.error |
| 667 | if progress is not None: |
| 668 | progress.failed(requirement_id, outcome.error) |
| 669 | except Exception as exc: |
| 670 | for requirement_id in requirement_ids & { |
| 671 | "VALUATION_INPUTS", "LATEST_PRICE", "SECTOR_MACRO" |
| 672 | }: |
| 673 | failures[requirement_id] = type(exc).__name__ |
| 674 | if progress is not None: |
| 675 | progress.failed(requirement_id, type(exc).__name__) |
| 676 | |
| 677 | financial = requirement_ids & _FINANCIAL_REQUIREMENTS |
| 678 | repository_categories: set[str] = set() |
| 679 | if financial: |
| 680 | executed.append("FINANCIALS") |
| 681 | if progress is not None: |
| 682 | progress.executed("FINANCIALS") |
| 683 | if jurisdiction == "INDIA": |
| 684 | repository_categories.add("FINANCIAL_RESULTS") |
| 685 | else: |
| 686 | try: |
| 687 | profile = self.repository.profile(global_instrument_id) |
| 688 | result = await self.orchestrator.refresh_international_fundamentals( |
| 689 | profile, |
| 690 | correlation_id=correlation_id, |
| 691 | identity_headers=dict(identity_headers or {}), |
| 692 | ) |
| 693 | if result is None or not result.facts: |
| 694 | for requirement_id in financial: |
| 695 | failures[requirement_id] = "PRIMARY_FINANCIAL_PROVIDER_RETURNED_NO_FACTS" |
| 696 | if progress is not None: |
| 697 | progress.failed( |
| 698 | requirement_id, |
| 699 | "PRIMARY_FINANCIAL_PROVIDER_RETURNED_NO_FACTS", |
| 700 | ) |
| 701 | except Exception as exc: |
| 702 | for requirement_id in financial: |
| 703 | failures[requirement_id] = type(exc).__name__ |
| 704 | if progress is not None: |
| 705 | progress.failed(requirement_id, type(exc).__name__) |
| 706 | |
| 707 | if "SHAREHOLDING" in requirement_ids: |
| 708 | executed.append("SHAREHOLDING") |
| 709 | if progress is not None: |
| 710 | progress.executed("SHAREHOLDING") |
| 711 | repository_categories.add("SHAREHOLDING_PATTERN") |
| 712 | if "ORDER_BOOK_CAPEX_GUIDANCE" in requirement_ids: |
| 713 | executed.append("ORDER_BOOK_CAPACITY_CATALYSTS") |
| 714 | if progress is not None: |
| 715 | progress.executed("ORDER_BOOK_CAPACITY_CATALYSTS") |
| 716 | repository_categories.update( |
| 717 | {"ORDERS_BACKLOG", "CONTRACTS", "CAPEX", "NEW_FACILITIES", "GUIDANCE"} |
| 718 | ) |
| 719 | if "CURRENT_NEWS" in requirement_ids: |
| 720 | executed.append("GLOBAL_NEWS_SEARCH") |
| 721 | if progress is not None: |
| 722 | progress.executed("GLOBAL_NEWS_SEARCH") |
| 723 | news_worker=getattr(self.repository,'refresh_news_intelligence',None) |
| 724 | if callable(news_worker): |
| 725 | try: |
| 726 | await news_worker(global_instrument_id) |
| 727 | except Exception: |
| 728 | failures['CURRENT_NEWS']='NEWS_INTELLIGENCE_UNAVAILABLE' |
| 729 | else: |
| 730 | repository_categories.update({'CATALYSTS','RISKS','REGULATORY','MANAGEMENT','GUIDANCE'}) |
| 731 | if "GOVERNANCE_HISTORY" in requirement_ids: |
| 732 | executed.append("GOVERNANCE_EVIDENCE") |
| 733 | if progress is not None: |
| 734 | progress.executed("GOVERNANCE_EVIDENCE") |
| 735 | repository_categories.update({"RISKS", "REGULATORY", "MANAGEMENT"}) |
| 736 | if repository_categories: |
| 737 | try: |
| 738 | await self.repository.refresh_targeted_categories( |
| 739 | global_instrument_id, |
| 740 | repository_categories, |
| 741 | correlation_id=correlation_id, |
| 742 | allow_demo=True, |
| 743 | ) |
| 744 | except Exception as exc: |
| 745 | for requirement_id in requirement_ids & ( |
| 746 | _FINANCIAL_REQUIREMENTS |
| 747 | | { |
| 748 | "SHAREHOLDING", |
| 749 | "ORDER_BOOK_CAPEX_GUIDANCE", |
| 750 | "CURRENT_NEWS", |
| 751 | "GOVERNANCE_HISTORY", |
| 752 | } |
| 753 | ): |
| 754 | failures[requirement_id] = type(exc).__name__ |
| 755 | if progress is not None: |
| 756 | progress.failed(requirement_id, type(exc).__name__) |
| 757 | |
| 758 | if "HISTORICAL_PRICE_SERIES" in requirement_ids: |
| 759 | executed.append("HISTORICAL_MARKET_DATA") |
| 760 | if progress is not None: |
| 761 | progress.executed("HISTORICAL_MARKET_DATA") |
| 762 | try: |
| 763 | await self._ensure_historical_prices(global_instrument_id) |
| 764 | except Exception as exc: |
| 765 | failures["HISTORICAL_PRICE_SERIES"] = type(exc).__name__ |
| 766 | if progress is not None: |
| 767 | progress.failed("HISTORICAL_PRICE_SERIES", type(exc).__name__) |
| 768 | |
| 769 | return CapabilityExecutionResult(tuple(dict.fromkeys(executed)), failures) |
| 770 | |
| 771 | async def execute_approved_fallbacks( |
| 772 | self, |
| 773 | global_instrument_id: UUID, |
| 774 | targets: Sequence[ResearchRefreshTarget], |
| 775 | ) -> CapabilityExecutionResult: |
| 776 | financial = {target.requirement_id for target in targets} & _FINANCIAL_REQUIREMENTS |
| 777 | if not financial: |
| 778 | return CapabilityExecutionResult() |
| 779 | try: |
| 780 | outcome = await self.orchestrator.ensure_structured_market( |
| 781 | global_instrument_id, {"FUNDAMENTALS"} |
| 782 | ) |
| 783 | failures = ( |
| 784 | {requirement_id: outcome.error for requirement_id in financial} |
| 785 | if outcome.error |
| 786 | else {} |
| 787 | ) |
| 788 | return CapabilityExecutionResult(("APPROVED_STRUCTURED_FINANCIAL_FALLBACK",), failures) |
| 789 | except Exception as exc: |
| 790 | return CapabilityExecutionResult( |
| 791 | ("APPROVED_STRUCTURED_FINANCIAL_FALLBACK",), |
| 792 | {requirement_id: type(exc).__name__ for requirement_id in financial}, |
| 793 | ) |
| 794 | |
| 795 | async def _ensure_historical_prices(self, global_instrument_id: UUID) -> int: |
| 796 | profile = self.repository.profile(global_instrument_id) |
| 797 | provider_ticker = profile.provider_instrument_ids.get("YAHOO_FINANCE") |
| 798 | if not provider_ticker: |
| 799 | raise ValueError("VERIFIED_HISTORICAL_MAPPING_REQUIRED") |
| 800 | observations = ( |
| 801 | await self.repository.market_price_observations_for_instruments( |
| 802 | {global_instrument_id} |
| 803 | ) |
| 804 | ).get(global_instrument_id, []) |
| 805 | now = datetime.now(timezone.utc) |
| 806 | observed_at = [value.observed_at for value in observations] |
| 807 | has_year = bool(observed_at) and has_year_historical_coverage( |
| 808 | min(observed_at), max(observed_at), len(observed_at) |
| 809 | ) |
| 810 | start = ( |
| 811 | max(observed_at) + timedelta(days=1) |
| 812 | if has_year |
| 813 | else now |
| 814 | - timedelta( |
| 815 | days=self.market_data_population_jobs.settings.market_data_population_initial_lookback_days |
| 816 | ) |
| 817 | ) |
| 818 | end = now + timedelta(days=1) |
| 819 | if has_year and start.date() >= end.date(): |
| 820 | return 0 |
| 821 | instrument = { |
| 822 | "globalInstrumentId": str(global_instrument_id), |
| 823 | "structuredProviderTicker": provider_ticker, |
| 824 | "ticker": profile.ticker, |
| 825 | "currency": profile.currency, |
| 826 | } |
| 827 | return await self.market_data_population_jobs.population.populate( |
| 828 | [instrument], start=start, end=end |
| 829 | ) |
| 830 | |
| 831 | |
| 832 | @dataclass(frozen=True) |
| 833 | class TargetedEnsureResult: |
| 834 | readiness: ResearchReadinessResult |
| 835 | planned_requirement_ids: tuple[str, ...] |
| 836 | executed_capabilities: tuple[str, ...] |
| 837 | reused_single_flight: bool = False |
| 838 | failures: Mapping[str, str] = field(default_factory=dict) |
| 839 | |
| 840 | def __post_init__(self) -> None: |
| 841 | object.__setattr__(self, "failures", dict(self.failures or {})) |
| 842 | |
| 843 | |
| 844 | @dataclass(frozen=True) |
| 845 | class _PlanExecutionResult: |
| 846 | executed_capabilities: tuple[str, ...] = () |
| 847 | failures: Mapping[str, str] = field(default_factory=dict) |
| 848 | |
| 849 | def __post_init__(self) -> None: |
| 850 | object.__setattr__(self, "failures", dict(self.failures or {})) |
| 851 | |
| 852 | |
| 853 | @dataclass(frozen=True) |
| 854 | class _EnsureFlight: |
| 855 | task: asyncio.Task[_PlanExecutionResult] |
| 856 | requirement_ids: frozenset[str] |
| 857 | |
| 858 | |
| 859 | class ResearchReadinessRuntime: |
| 860 | """DB-first application service shared by GET and targeted ensure routes.""" |
| 861 | |
| 862 | def __init__( |
| 863 | self, |
| 864 | repository, |
| 865 | data_source: RepositoryResearchReadinessAdapter, |
| 866 | executor: ExistingResearchCapabilityExecutor, |
| 867 | requirement_registry: ResearchRequirementRegistry | None = None, |
| 868 | authority_registry: ProviderAuthorityRegistry | None = None, |
| 869 | ensure_timeout_seconds: float = 25.0, |
| 870 | ) -> None: |
| 871 | if ensure_timeout_seconds <= 0: |
| 872 | raise ValueError("ensure_timeout_seconds must be positive") |
| 873 | self.repository = repository |
| 874 | self.data_source = data_source |
| 875 | self.requirement_registry = requirement_registry or ResearchRequirementRegistry.default() |
| 876 | self.authority_registry = authority_registry or ProviderAuthorityRegistry.default() |
| 877 | self.readiness_service = ResearchReadinessService( |
| 878 | data_source, |
| 879 | requirement_registry=self.requirement_registry, |
| 880 | authority_registry=self.authority_registry, |
| 881 | ) |
| 882 | self.planner = ResearchRefreshPlanner(self.authority_registry) |
| 883 | self.executor = executor |
| 884 | self.ensure_timeout_seconds = ensure_timeout_seconds |
| 885 | self._flights: dict[UUID, _EnsureFlight] = {} |
| 886 | |
| 887 | async def read( |
| 888 | self, |
| 889 | global_instrument_id: UUID, |
| 890 | *, |
| 891 | jurisdiction: str, |
| 892 | now: datetime | None = None, |
| 893 | ) -> ResearchReadinessResult: |
| 894 | if isinstance(self.data_source, RepositoryResearchReadinessAdapter): |
| 895 | self.data_source._evaluation_times[global_instrument_id] = now or datetime.now(timezone.utc) |
| 896 | if callable(getattr(self.repository, "market_session_data", None)): |
| 897 | profile = self.repository.profile(global_instrument_id) |
| 898 | self.data_source._sessions[global_instrument_id] = await self.repository.market_session_data({value for value in (profile.mic, profile.exchange) if value}) |
| 899 | return await self.repository._run_blocking_persistence( |
| 900 | self.readiness_service.assess, |
| 901 | global_instrument_id, |
| 902 | jurisdiction=jurisdiction, |
| 903 | now=now, |
| 904 | ) |
| 905 | |
| 906 | async def ensure( |
| 907 | self, |
| 908 | global_instrument_id: UUID, |
| 909 | *, |
| 910 | jurisdiction: str, |
| 911 | requirement_ids: Sequence[str] | None, |
| 912 | correlation_id: str | None = None, |
| 913 | identity_headers: Mapping[str, str | None] | None = None, |
| 914 | ) -> TargetedEnsureResult: |
| 915 | selected = self._validated_requirement_ids(requirement_ids) |
| 916 | readiness = await self.read(global_instrument_id, jurisdiction=jurisdiction) |
| 917 | plan = self.planner.plan( |
| 918 | readiness, |
| 919 | jurisdiction=jurisdiction, |
| 920 | requirement_ids=selected, |
| 921 | include_non_mandatory=True, |
| 922 | ) |
| 923 | planned_ids = tuple(target.requirement_id for target in plan.targets) |
| 924 | if not plan.targets: |
| 925 | return TargetedEnsureResult(readiness, (), ()) |
| 926 | |
| 927 | existing = self._flights.get(global_instrument_id) |
| 928 | if existing is not None: |
| 929 | shared_execution = await asyncio.shield(existing.task) |
| 930 | attempted = existing.requirement_ids |
| 931 | readiness = await self.read(global_instrument_id, jurisdiction=jurisdiction) |
| 932 | remaining = self.planner.plan( |
| 933 | readiness, |
| 934 | jurisdiction=jurisdiction, |
| 935 | requirement_ids=selected, |
| 936 | include_non_mandatory=True, |
| 937 | ) |
| 938 | remaining_targets = tuple( |
| 939 | target |
| 940 | for target in remaining.targets |
| 941 | if target.requirement_id not in attempted |
| 942 | ) |
| 943 | if not remaining_targets: |
| 944 | return TargetedEnsureResult( |
| 945 | readiness, planned_ids, (), True, shared_execution.failures |
| 946 | ) |
| 947 | plan = ResearchRefreshPlan( |
| 948 | global_instrument_id, remaining_targets, remaining.created_at |
| 949 | ) |
| 950 | |
| 951 | target_ids = frozenset(target.requirement_id for target in plan.targets) |
| 952 | task = asyncio.create_task( |
| 953 | self._execute_plan_bounded( |
| 954 | plan, |
| 955 | jurisdiction=jurisdiction, |
| 956 | correlation_id=correlation_id, |
| 957 | identity_headers=identity_headers, |
| 958 | ) |
| 959 | ) |
| 960 | flight = _EnsureFlight(task, target_ids) |
| 961 | self._flights[global_instrument_id] = flight |
| 962 | |
| 963 | def cleanup(completed: asyncio.Task[_PlanExecutionResult]) -> None: |
| 964 | if self._flights.get(global_instrument_id) is flight: |
| 965 | self._flights.pop(global_instrument_id, None) |
| 966 | |
| 967 | task.add_done_callback(cleanup) |
| 968 | execution = await asyncio.shield(task) |
| 969 | recorder = getattr(self.repository, "record_acquisition_observation", None) |
| 970 | if callable(recorder): |
| 971 | for target in plan.targets: |
| 972 | failure = execution.failures.get(target.requirement_id) |
| 973 | await recorder(global_instrument_id, target.requirement_id, "READINESS_EXECUTOR", |
| 974 | "FAILED" if failure else "COMPLETED", datetime.now(timezone.utc), failure_reason=failure) |
| 975 | readiness = await self.read(global_instrument_id, jurisdiction=jurisdiction) |
| 976 | return TargetedEnsureResult( |
| 977 | readiness, |
| 978 | planned_ids, |
| 979 | execution.executed_capabilities, |
| 980 | failures=execution.failures, |
| 981 | ) |
| 982 | |
| 983 | async def _execute_plan_bounded( |
| 984 | self, |
| 985 | plan: ResearchRefreshPlan, |
| 986 | *, |
| 987 | jurisdiction: str, |
| 988 | correlation_id: str | None, |
| 989 | identity_headers: Mapping[str, str | None] | None, |
| 990 | ) -> _PlanExecutionResult: |
| 991 | progress = CapabilityExecutionProgress() |
| 992 | try: |
| 993 | async with asyncio.timeout(self.ensure_timeout_seconds): |
| 994 | return await self._execute_plan( |
| 995 | plan, |
| 996 | jurisdiction=jurisdiction, |
| 997 | correlation_id=correlation_id, |
| 998 | identity_headers=identity_headers, |
| 999 | progress=progress, |
| 1000 | ) |
| 1001 | except TimeoutError: |
| 1002 | # The interactive route owns a smaller execution budget than the |
| 1003 | # API gateway. Provider work is cancelled here so an unavailable |
| 1004 | # capability becomes a bounded requirement failure instead of a |
| 1005 | # downstream gateway timeout. Any facts already committed remain |
| 1006 | # durable and are reflected by the read below. |
| 1007 | self.data_source.finish_refresh(plan.global_instrument_id) |
| 1008 | readiness = await self.read( |
| 1009 | plan.global_instrument_id, jurisdiction=jurisdiction |
| 1010 | ) |
| 1011 | failures = {} |
| 1012 | for target in plan.targets: |
| 1013 | if self._requires_acquisition( |
| 1014 | readiness.for_requirement(target.requirement_id).status |
| 1015 | ): |
| 1016 | failures[target.requirement_id] = _combined_failure_reason( |
| 1017 | progress.failures.get(target.requirement_id), |
| 1018 | "ACQUISITION_TIMEOUT", |
| 1019 | ) |
| 1020 | self.data_source.finish_refresh(plan.global_instrument_id, failures) |
| 1021 | logger.warning( |
| 1022 | "research_readiness_ensure_timeout globalInstrumentId=%s " |
| 1023 | "budgetSeconds=%s requirements=%s", |
| 1024 | plan.global_instrument_id, |
| 1025 | self.ensure_timeout_seconds, |
| 1026 | sorted(failures), |
| 1027 | ) |
| 1028 | return _PlanExecutionResult( |
| 1029 | tuple(progress.executed_capabilities), failures |
| 1030 | ) |
| 1031 | |
| 1032 | async def _execute_plan( |
| 1033 | self, |
| 1034 | plan: ResearchRefreshPlan, |
| 1035 | *, |
| 1036 | jurisdiction: str, |
| 1037 | correlation_id: str | None, |
| 1038 | identity_headers: Mapping[str, str | None] | None, |
| 1039 | progress: CapabilityExecutionProgress | None = None, |
| 1040 | ) -> _PlanExecutionResult: |
| 1041 | ids = tuple(target.requirement_id for target in plan.targets) |
| 1042 | self.data_source.mark_refreshing(plan.global_instrument_id, ids) |
| 1043 | failures: dict[str, str] = {} |
| 1044 | executed: list[str] = [] |
| 1045 | completed: set[str] = set() |
| 1046 | try: |
| 1047 | primary = await self.executor.execute_primary( |
| 1048 | plan.global_instrument_id, |
| 1049 | plan.targets, |
| 1050 | jurisdiction=jurisdiction, |
| 1051 | correlation_id=correlation_id, |
| 1052 | identity_headers=identity_headers, |
| 1053 | progress=progress, |
| 1054 | ) |
| 1055 | executed.extend(primary.executed_capabilities) |
| 1056 | if progress is not None: |
| 1057 | for capability in primary.executed_capabilities: |
| 1058 | progress.executed(capability) |
| 1059 | for requirement_id, reason in primary.failures.items(): |
| 1060 | progress.failed(requirement_id, reason) |
| 1061 | for requirement_id in primary.satisfied_requirement_ids: |
| 1062 | progress.satisfied(requirement_id) |
| 1063 | for requirement_id, reason in primary.failures.items(): |
| 1064 | failures[requirement_id] = _combined_failure_reason( |
| 1065 | failures.get(requirement_id), reason |
| 1066 | ) |
| 1067 | completed.update(primary.satisfied_requirement_ids) |
| 1068 | finally: |
| 1069 | # Re-read durable state before exposing provider failures so an |
| 1070 | # unavailable primary can be classified as missing/partial and |
| 1071 | # considered for an approved existing fallback. |
| 1072 | self.data_source.finish_refresh(plan.global_instrument_id) |
| 1073 | |
| 1074 | after_primary = await self.read(plan.global_instrument_id, jurisdiction=jurisdiction) |
| 1075 | unresolved: list[ResearchRefreshTarget] = [] |
| 1076 | for target in plan.targets: |
| 1077 | if target.requirement_id in completed: |
| 1078 | continue |
| 1079 | result = after_primary.for_requirement(target.requirement_id) |
| 1080 | if result.status not in { |
| 1081 | ResearchRequirementStatus.MISSING, |
| 1082 | ResearchRequirementStatus.PARTIAL, |
| 1083 | ResearchRequirementStatus.CONFLICTING, |
| 1084 | ResearchRequirementStatus.FAILED, |
| 1085 | }: |
| 1086 | continue |
| 1087 | if target.authority_policy.fallback_policy.permits(result.status): |
| 1088 | unresolved.append(target) |
| 1089 | if unresolved: |
| 1090 | fallback = await self.executor.execute_approved_fallbacks( |
| 1091 | plan.global_instrument_id, unresolved |
| 1092 | ) |
| 1093 | executed.extend(fallback.executed_capabilities) |
| 1094 | if progress is not None: |
| 1095 | for capability in fallback.executed_capabilities: |
| 1096 | progress.executed(capability) |
| 1097 | for requirement_id, reason in fallback.failures.items(): |
| 1098 | progress.failed(requirement_id, reason) |
| 1099 | for requirement_id, reason in fallback.failures.items(): |
| 1100 | failures[requirement_id] = _combined_failure_reason( |
| 1101 | failures.get(requirement_id), reason |
| 1102 | ) |
| 1103 | final_readiness = await self.read( |
| 1104 | plan.global_instrument_id, jurisdiction=jurisdiction |
| 1105 | ) |
| 1106 | unresolved_failures = { |
| 1107 | requirement_id: reason |
| 1108 | for requirement_id, reason in failures.items() |
| 1109 | if self._requires_acquisition( |
| 1110 | final_readiness.for_requirement(requirement_id).status |
| 1111 | ) |
| 1112 | } |
| 1113 | self.data_source.finish_refresh(plan.global_instrument_id, unresolved_failures) |
| 1114 | return _PlanExecutionResult( |
| 1115 | tuple(dict.fromkeys(executed)), unresolved_failures |
| 1116 | ) |
| 1117 | |
| 1118 | @staticmethod |
| 1119 | def _requires_acquisition(status: ResearchRequirementStatus) -> bool: |
| 1120 | return status in { |
| 1121 | ResearchRequirementStatus.MISSING, |
| 1122 | ResearchRequirementStatus.PARTIAL, |
| 1123 | ResearchRequirementStatus.CONFLICTING, |
| 1124 | ResearchRequirementStatus.FAILED, |
| 1125 | } |
| 1126 | |
| 1127 | def _validated_requirement_ids( |
| 1128 | self, requirement_ids: Sequence[str] | None |
| 1129 | ) -> tuple[str, ...] | None: |
| 1130 | if requirement_ids is None: |
| 1131 | return None |
| 1132 | known_ids = { |
| 1133 | item.requirement_id for item in self.requirement_registry.requirements |
| 1134 | } |
| 1135 | area_ids = {area.value for area in RuleEngineArea} |
| 1136 | expanded: list[str] = [] |
| 1137 | unknown: set[str] = set() |
| 1138 | for value in requirement_ids: |
| 1139 | normalized = str(value).strip().upper() |
| 1140 | if normalized in known_ids: |
| 1141 | expanded.append(normalized) |
| 1142 | elif normalized in area_ids: |
| 1143 | expanded.extend( |
| 1144 | item.requirement_id |
| 1145 | for item in self.requirement_registry.for_area( |
| 1146 | RuleEngineArea(normalized) |
| 1147 | ) |
| 1148 | ) |
| 1149 | else: |
| 1150 | unknown.add(normalized) |
| 1151 | if unknown: |
| 1152 | raise ValueError(f"UNKNOWN_RESEARCH_REQUIREMENT:{','.join(sorted(unknown))}") |
| 1153 | return tuple(dict.fromkeys(expanded)) |
| 1154 | |
| 1155 | |
| 1156 | def readiness_response( |
| 1157 | value: ResearchReadinessResult, |
| 1158 | registry: ResearchRequirementRegistry, |
| 1159 | *, |
| 1160 | ensure: TargetedEnsureResult | None = None, |
| 1161 | ) -> dict[str, Any]: |
| 1162 | requirements = [] |
| 1163 | for item in value.requirements: |
| 1164 | contract = registry.get(item.requirement_id) |
| 1165 | requirements.append( |
| 1166 | { |
| 1167 | "requirementId": item.requirement_id, |
| 1168 | "area": item.rule_engine_area.value, |
| 1169 | "areaWeightPct": int(registry.area_weights[item.rule_engine_area] * 100), |
| 1170 | "importance": item.importance.value, |
| 1171 | "mandatory": item.mandatory, |
| 1172 | "status": item.status.value, |
| 1173 | "applicability": item.applicability, |
| 1174 | "applicabilityReason": item.applicability_reason, |
| 1175 | "businessClassification": item.classification, |
| 1176 | "classificationSource": item.classification_source, |
| 1177 | "acquisitionObservation": item.acquisition_observation, |
| 1178 | "sourceProvider": item.source, |
| 1179 | "sourceTier": item.source_tier.value if item.source_tier else None, |
| 1180 | "sourceUrl": item.source_url, |
| 1181 | "asOf": _iso(item.as_of), |
| 1182 | "retrievedAt": _iso(item.retrieved_at), |
| 1183 | "ageSeconds": int(item.age.total_seconds()) if item.age is not None else None, |
| 1184 | "freshnessPolicy": { |
| 1185 | "policyId": item.freshness_policy.policy_id, |
| 1186 | "mode": item.freshness_policy.mode.value, |
| 1187 | "maximumAgeSeconds": ( |
| 1188 | int(item.freshness_policy.maximum_age.total_seconds()) |
| 1189 | if item.freshness_policy.maximum_age is not None |
| 1190 | else None |
| 1191 | ), |
| 1192 | "scoringWindowDays": ( |
| 1193 | item.freshness_policy.scoring_window.days |
| 1194 | if item.freshness_policy.scoring_window is not None |
| 1195 | else None |
| 1196 | ), |
| 1197 | }, |
| 1198 | "evidenceIds": list(item.evidence_ids), |
| 1199 | "coveredInputIds": list(item.covered_input_ids), |
| 1200 | "missingInputIds": list(item.missing_input_ids), |
| 1201 | "concreteRequirements": [ |
| 1202 | { |
| 1203 | "inputId": input_.input_id, |
| 1204 | "importance": input_.importance.value, |
| 1205 | "covered": input_.input_id in item.covered_input_ids, |
| 1206 | "applicability": "NOT_APPLICABLE" if item.applicability == "NOT_APPLICABLE" or input_.input_id in item.not_applicable_input_reasons else "APPLICABLE", |
| 1207 | "applicabilityReason": item.not_applicable_input_reasons.get(input_.input_id) or (item.applicability_reason if item.applicability == "NOT_APPLICABLE" else None), |
| 1208 | } |
| 1209 | for input_ in contract.inputs |
| 1210 | ], |
| 1211 | "coveragePct": item.coverage_pct, |
| 1212 | "criticalCoveragePct": item.critical_coverage_pct, |
| 1213 | "missingReason": item.missing_reason, |
| 1214 | "conflictReason": item.conflict_reason, |
| 1215 | "supportedActions": [action.value for action in item.supported_actions], |
| 1216 | } |
| 1217 | ) |
| 1218 | response: dict[str, Any] = { |
| 1219 | "globalInstrumentId": str(value.global_instrument_id), |
| 1220 | "overallStatus": value.overall_status.value, |
| 1221 | "overallCompletenessPct": value.overall_completeness_pct, |
| 1222 | "criticalCompletenessPct": value.critical_completeness_pct, |
| 1223 | "confidence": value.confidence.value, |
| 1224 | "confidencePct": value.confidence_pct, |
| 1225 | "generatedAt": value.generated_at.isoformat(), |
| 1226 | "requirements": requirements, |
| 1227 | } |
| 1228 | if ensure is not None: |
| 1229 | response["refreshState"] = { |
| 1230 | "plannedRequirements": list(ensure.planned_requirement_ids), |
| 1231 | "executedCapabilities": list(ensure.executed_capabilities), |
| 1232 | "reusedSingleFlight": ensure.reused_single_flight, |
| 1233 | "failureReasons": dict(ensure.failures), |
| 1234 | } |
| 1235 | return response |
| 1236 | |
| 1237 | |
| 1238 | def jurisdiction_for_profile(profile: CompanyResearchProfile) -> str: |
| 1239 | country = profile.country.strip().upper() |
| 1240 | exchange = profile.exchange.strip().upper() |
| 1241 | if country in {"IN", "IND", "INDIA"} or exchange in {"NSE", "XNSE", "BSE", "XBOM"}: |
| 1242 | return "INDIA" |
| 1243 | if country in {"US", "USA", "UNITED STATES"}: |
| 1244 | return "USA" |
| 1245 | if country in { |
| 1246 | "AT", "BE", "CH", "CZ", "DE", "DK", "ES", "EU", "FI", "FR", "GB", "IE", |
| 1247 | "IT", "LU", "NL", "NO", "PL", "PT", "SE", "UK", |
| 1248 | }: |
| 1249 | return "EUROPE" |
| 1250 | return "GLOBAL" |
| 1251 | |
| 1252 | |
| 1253 | def _financial_fact_coverage( |
| 1254 | fact: FinancialFact, |
| 1255 | metric: str, |
| 1256 | periods_by_metric: Mapping[str, set[str]], |
| 1257 | quarterly_periods: set[str], |
| 1258 | annual_periods: set[str], |
| 1259 | latest_quarter: str | None, |
| 1260 | ) -> dict[str, set[str]]: |
| 1261 | coverage: dict[str, set[str]] = {} |
| 1262 | |
| 1263 | def add(requirement_id: str, *input_ids: str) -> None: |
| 1264 | coverage.setdefault(requirement_id, set()).update(input_ids) |
| 1265 | |
| 1266 | if metric in {"eps", "pat", "net_income", "net_profit"}: |
| 1267 | add("VALUATION_INPUTS", "EARNINGS_BASIS") |
| 1268 | if metric in {"pat", "net_income", "net_profit", "roe", "return_on_equity", "roce", "return_on_capital_employed", "operating_margin", "profit_margin", "net_margin", "operating_cash_flow"}: |
| 1269 | if len(periods_by_metric.get(metric, set())) >= 2: |
| 1270 | add("BUSINESS_QUALITY_FACTS", "PROFITABILITY_HISTORY") |
| 1271 | if metric in {"roe", "return_on_equity"}: |
| 1272 | add("BUSINESS_QUALITY_FACTS", "ROE") |
| 1273 | if metric in {"roce", "return_on_capital_employed"}: |
| 1274 | add("BUSINESS_QUALITY_FACTS", "ROCE") |
| 1275 | if metric in {"operating_margin", "profit_margin", "ebitda_margin", "gross_margin"}: |
| 1276 | add("BUSINESS_QUALITY_FACTS", "MARGINS") |
| 1277 | if metric in { |
| 1278 | "free_cash_flow", "operating_cash_flow", "cash_flow_from_operating_activities" |
| 1279 | }: |
| 1280 | add("BUSINESS_QUALITY_FACTS", "CASH_CONVERSION_OR_FCF_QUALITY") |
| 1281 | if metric in {"revenue", "total_revenue"} and len(periods_by_metric.get(metric, set())) >= 2: |
| 1282 | add("GROWTH_FACTS", "REVENUE_HISTORY") |
| 1283 | if metric in {"eps", "pat", "net_income", "net_profit"} and len( |
| 1284 | periods_by_metric.get(metric, set()) |
| 1285 | ) >= 2: |
| 1286 | add("GROWTH_FACTS", "EARNINGS_HISTORY") |
| 1287 | if fact.key.period_type == "QUARTERLY" and len(quarterly_periods) >= 2: |
| 1288 | add("GROWTH_FACTS", "QUARTERLY_YOY_QOQ_TRENDS") |
| 1289 | if fact.key.period_type == "ANNUAL" and len(annual_periods) >= 2: |
| 1290 | add("GROWTH_FACTS", "ANNUAL_CAGR_INPUTS") |
| 1291 | if metric in {"total_debt", "debt", "debt_or_borrowings", "borrowings"}: |
| 1292 | add("BALANCE_SHEET_FACTS", "DEBT") |
| 1293 | if metric in {"total_equity", "equity", "net_worth"}: |
| 1294 | add("BALANCE_SHEET_FACTS", "EQUITY") |
| 1295 | if metric in {"cash", "total_cash", "cash_and_cash_equivalents", "cash_and_equivalents"}: |
| 1296 | add("BALANCE_SHEET_FACTS", "CASH") |
| 1297 | if metric in {"interest_expense", "finance_cost", "finance_costs", "ebit", "operating_profit"}: |
| 1298 | add("BALANCE_SHEET_FACTS", "INTEREST_COVERAGE_INPUTS") |
| 1299 | if metric in {"current_assets", "current_liabilities", "current_ratio"}: |
| 1300 | add("BALANCE_SHEET_FACTS", "LIQUIDITY_CURRENT_RATIO_INPUTS") |
| 1301 | if fact.key.period_type == "QUARTERLY": |
| 1302 | if fact.key.period_end and fact.key.period_end[:10] == latest_quarter: |
| 1303 | add("QUARTERLY_FINANCIALS", "LATEST_QUARTERLY_RESULT") |
| 1304 | if len(quarterly_periods) >= 2: |
| 1305 | add("QUARTERLY_FINANCIALS", "COMPARABLE_QUARTERS") |
| 1306 | if metric in {"revenue", "total_revenue"}: |
| 1307 | add("QUARTERLY_FINANCIALS", "QUARTERLY_REVENUE") |
| 1308 | if metric in {"pat", "net_income", "net_profit"}: |
| 1309 | add("QUARTERLY_FINANCIALS", "QUARTERLY_PAT") |
| 1310 | if metric in {"ebitda", "operating_profit", "operating_income"}: |
| 1311 | add("QUARTERLY_FINANCIALS", "QUARTERLY_EBITDA_OR_OPERATING_PROFIT") |
| 1312 | if metric == "eps": |
| 1313 | add("QUARTERLY_FINANCIALS", "QUARTERLY_EPS") |
| 1314 | if metric in {"operating_margin", "profit_margin", "ebitda_margin"}: |
| 1315 | add("QUARTERLY_FINANCIALS", "QUARTERLY_MARGINS") |
| 1316 | return coverage |
| 1317 | |
| 1318 | |
| 1319 | def _structured_fact_coverage( |
| 1320 | fact_name: str, facts: Mapping[str, Any] |
| 1321 | ) -> dict[str, set[str]]: |
| 1322 | coverage: dict[str, set[str]] = {} |
| 1323 | |
| 1324 | def add(requirement_id: str, *input_ids: str) -> None: |
| 1325 | coverage.setdefault(requirement_id, set()).update(input_ids) |
| 1326 | |
| 1327 | if fact_name == "latestPrice": |
| 1328 | add("LATEST_PRICE", "LATEST_USABLE_PRICE") |
| 1329 | add("VALUATION_INPUTS", "LATEST_USABLE_PRICE") |
| 1330 | if fact_name in {"trailingEps", "forwardEps"}: |
| 1331 | add("VALUATION_INPUTS", "EARNINGS_BASIS") |
| 1332 | if fact_name in {"trailingPE", "forwardPE"}: |
| 1333 | add("VALUATION_INPUTS", "PE") |
| 1334 | if fact_name == "priceToBook": |
| 1335 | add("VALUATION_INPUTS", "PB") |
| 1336 | if fact_name == "evToEbitda": |
| 1337 | add("VALUATION_INPUTS", "EV_EBITDA") |
| 1338 | if fact_name == "freeCashFlow" and facts.get("marketCap") is not None: |
| 1339 | add("VALUATION_INPUTS", "FCF_YIELD") |
| 1340 | if fact_name == "roe": |
| 1341 | add("BUSINESS_QUALITY_FACTS", "ROE") |
| 1342 | if fact_name in {"profitMargin", "operatingMargin"}: |
| 1343 | add("BUSINESS_QUALITY_FACTS", "MARGINS") |
| 1344 | if fact_name in {"freeCashFlow", "operatingCashFlow"}: |
| 1345 | add("BUSINESS_QUALITY_FACTS", "CASH_CONVERSION_OR_FCF_QUALITY") |
| 1346 | if fact_name == "revenueGrowth": |
| 1347 | add("GROWTH_FACTS", "REVENUE_HISTORY") |
| 1348 | if fact_name == "earningsGrowth": |
| 1349 | add("GROWTH_FACTS", "EARNINGS_HISTORY") |
| 1350 | if fact_name in {"totalDebt", "debtToEquity"}: |
| 1351 | add("BALANCE_SHEET_FACTS", "DEBT") |
| 1352 | if fact_name in {"bookValue", "debtToEquity"}: |
| 1353 | add("BALANCE_SHEET_FACTS", "EQUITY") |
| 1354 | if fact_name == "totalCash": |
| 1355 | add("BALANCE_SHEET_FACTS", "CASH") |
| 1356 | if fact_name == "currentRatio": |
| 1357 | add("BALANCE_SHEET_FACTS", "LIQUIDITY_CURRENT_RATIO_INPUTS") |
| 1358 | if fact_name == "sector": |
| 1359 | add("SECTOR_MACRO", "CANONICAL_SECTOR") |
| 1360 | return coverage |
| 1361 | |
| 1362 | |
| 1363 | def _positive_number(value: Any) -> bool: |
| 1364 | try: |
| 1365 | number = Decimal(str(value)) |
| 1366 | return number.is_finite() and number > 0 |
| 1367 | except Exception: |
| 1368 | return False |
| 1369 | |
| 1370 | |
| 1371 | def _evidence_from_financial_fact( |
| 1372 | fact: FinancialFact, requirement_id: str, covered_input_ids: tuple[str, ...] |
| 1373 | ) -> ResearchEvidence: |
| 1374 | source, tier = _financial_source(fact) |
| 1375 | as_of = fact.value.as_of_date or _period_datetime(fact.key.period_end) |
| 1376 | return ResearchEvidence( |
| 1377 | evidence_id=( |
| 1378 | f"financial:{fact.source_identity}:{fact.key.metric}:{fact.key.period_end}:" |
| 1379 | f"{fact.key.period_type}:{requirement_id}" |
| 1380 | ), |
| 1381 | requirement_id=requirement_id, |
| 1382 | source=source, |
| 1383 | source_tier=tier, |
| 1384 | retrieved_at=fact.value.retrieved_at, |
| 1385 | as_of=as_of, |
| 1386 | published_at=fact.value.published_at, |
| 1387 | fact_key=( |
| 1388 | f"{_metric(fact.key.metric)}:{fact.key.period_end}:" |
| 1389 | f"{fact.key.period_type}:{fact.key.reporting_basis}" |
| 1390 | ), |
| 1391 | value_fingerprint=_fingerprint(fact.value.value), |
| 1392 | confidence=fact.value.confidence, |
| 1393 | source_url=fact.value.source_url, |
| 1394 | covered_input_ids=covered_input_ids, |
| 1395 | valid_until=(as_of + timedelta(days=120 if fact.key.period_type == "QUARTERLY" else 400) |
| 1396 | if requirement_id == "VALUATION_INPUTS" and as_of and fact.key.period_type in {"QUARTERLY", "ANNUAL"} else None), |
| 1397 | ) |
| 1398 | |
| 1399 | |
| 1400 | def _evidence_from_document( |
| 1401 | document: ResearchDocument, |
| 1402 | requirement_id: str, |
| 1403 | covered_input_ids: tuple[str, ...], |
| 1404 | *, |
| 1405 | unresolved: bool = False, |
| 1406 | ) -> ResearchEvidence: |
| 1407 | source, tier = _document_source(document, requirement_id) |
| 1408 | return ResearchEvidence( |
| 1409 | evidence_id=f"document:{document.document_id}:{requirement_id}", |
| 1410 | requirement_id=requirement_id, |
| 1411 | source=source, |
| 1412 | source_tier=tier, |
| 1413 | retrieved_at=document.retrieved_at, |
| 1414 | as_of=document.published_at or document.retrieved_at, |
| 1415 | published_at=document.published_at, |
| 1416 | event_date=document.published_at, |
| 1417 | confidence=document.entity_resolution_confidence, |
| 1418 | unresolved=unresolved, |
| 1419 | source_url=document.canonical_url, |
| 1420 | covered_input_ids=covered_input_ids, |
| 1421 | ) |
| 1422 | |
| 1423 | |
| 1424 | def _evidence_from_event( |
| 1425 | event: ResearchEvent, |
| 1426 | requirement_id: str, |
| 1427 | covered_input_ids: tuple[str, ...], |
| 1428 | *, |
| 1429 | unresolved: bool = False, |
| 1430 | ) -> ResearchEvidence: |
| 1431 | source, tier = _event_source(event, requirement_id) |
| 1432 | return ResearchEvidence( |
| 1433 | evidence_id=f"event:{event.event_id}:{requirement_id}", |
| 1434 | requirement_id=requirement_id, |
| 1435 | source=source, |
| 1436 | source_tier=tier, |
| 1437 | retrieved_at=event.retrieved_at or event.detected_at, |
| 1438 | as_of=event.event_date or event.published_at, |
| 1439 | published_at=event.published_at, |
| 1440 | event_date=event.event_date, |
| 1441 | confidence=event.confidence, |
| 1442 | unresolved=unresolved, |
| 1443 | source_url=event.source_url, |
| 1444 | covered_input_ids=covered_input_ids, |
| 1445 | ) |
| 1446 | |
| 1447 | |
| 1448 | def _copy_evidence( |
| 1449 | value: ResearchEvidence, |
| 1450 | requirement_id: str, |
| 1451 | covered_input_ids: tuple[str, ...], |
| 1452 | ) -> ResearchEvidence: |
| 1453 | return ResearchEvidence( |
| 1454 | evidence_id=f"{value.evidence_id}:{requirement_id}", |
| 1455 | requirement_id=requirement_id, |
| 1456 | source=value.source, |
| 1457 | source_tier=value.source_tier, |
| 1458 | retrieved_at=value.retrieved_at, |
| 1459 | as_of=value.as_of, |
| 1460 | published_at=value.published_at, |
| 1461 | event_date=value.event_date, |
| 1462 | valid_until=value.valid_until, |
| 1463 | fact_key=value.fact_key, |
| 1464 | value_fingerprint=value.value_fingerprint, |
| 1465 | complete=value.complete, |
| 1466 | confidence=value.confidence, |
| 1467 | unresolved=value.unresolved, |
| 1468 | source_url=value.source_url, |
| 1469 | covered_input_ids=covered_input_ids, |
| 1470 | ) |
| 1471 | |
| 1472 | |
| 1473 | def _financial_source(fact: FinancialFact) -> tuple[str, ResearchSourceTier]: |
| 1474 | provider = fact.source_provider.strip().upper() |
| 1475 | if fact.source_tier == FactSourceTier.OFFICIAL_NSE: |
| 1476 | return "NSE", ResearchSourceTier.OFFICIAL |
| 1477 | if fact.source_tier == FactSourceTier.OFFICIAL_REGULATORY: |
| 1478 | return ("SEC_EDGAR" if "SEC" in provider else "REGULATORY_FILING"), ResearchSourceTier.REGULATORY |
| 1479 | if fact.source_tier == FactSourceTier.STRUCTURED_FUNDAMENTALS: |
| 1480 | return provider or "LICENSED_STRUCTURED", ResearchSourceTier.LICENSED_STRUCTURED |
| 1481 | if provider == "YAHOO_FINANCE_MCP": |
| 1482 | return "APPROVED_EXTERNAL_TOOL", ResearchSourceTier.APPROVED_EXTERNAL_TOOL |
| 1483 | if fact.source_tier == FactSourceTier.YAHOO: |
| 1484 | return "YAHOO_FINANCE", ResearchSourceTier.APPROVED_SECONDARY |
| 1485 | return provider or "APPROVED_SECONDARY", ResearchSourceTier.APPROVED_SECONDARY |
| 1486 | |
| 1487 | |
| 1488 | def _document_source( |
| 1489 | document: ResearchDocument, requirement_id: str |
| 1490 | ) -> tuple[str, ResearchSourceTier]: |
| 1491 | if document.discovery_provider == "YAHOO_FINANCE_MCP": |
| 1492 | return "APPROVED_EXTERNAL_TOOL", ResearchSourceTier.APPROVED_EXTERNAL_TOOL |
| 1493 | host = (urlparse(document.canonical_url).hostname or "").casefold() |
| 1494 | if document.source_classification == SourceClassification.EXCHANGE: |
| 1495 | return ("NSE" if "nseindia" in host or "nse" in document.source_name.casefold() else "COMPANY_FILING"), ResearchSourceTier.OFFICIAL |
| 1496 | if document.source_classification == SourceClassification.REGULATORY: |
| 1497 | return ("SEC_EDGAR" if "sec.gov" in host else "REGULATORY_FILING"), ResearchSourceTier.REGULATORY |
| 1498 | if document.source_classification == SourceClassification.OFFICIAL_COMPANY: |
| 1499 | return ( |
| 1500 | "OFFICIAL_COMPANY" if requirement_id == "CURRENT_NEWS" else "COMPANY_FILING", |
| 1501 | ResearchSourceTier.OFFICIAL, |
| 1502 | ) |
| 1503 | if document.source_classification == SourceClassification.REPUTABLE_NEWS: |
| 1504 | return "REPUTABLE_NEWS", ResearchSourceTier.APPROVED_SECONDARY |
| 1505 | return "APPROVED_SECONDARY", ResearchSourceTier.APPROVED_SECONDARY |
| 1506 | |
| 1507 | |
| 1508 | def _event_source( |
| 1509 | event: ResearchEvent, requirement_id: str |
| 1510 | ) -> tuple[str, ResearchSourceTier]: |
| 1511 | if str(event.independence_key or "").startswith("YAHOO_FINANCE_MCP:"): |
| 1512 | return "APPROVED_EXTERNAL_TOOL", ResearchSourceTier.APPROVED_EXTERNAL_TOOL |
| 1513 | host = (urlparse(event.source_url).hostname or "").casefold() |
| 1514 | if event.source_classification == SourceClassification.EXCHANGE: |
| 1515 | return ("NSE" if "nseindia" in host else "REGULATORY_FILING"), ResearchSourceTier.OFFICIAL |
| 1516 | if event.source_classification == SourceClassification.REGULATORY: |
| 1517 | return ( |
| 1518 | "REGULATOR_OR_COURT_RECORD" |
| 1519 | if requirement_id == "GOVERNANCE_HISTORY" |
| 1520 | else "REGULATORY_FILING", |
| 1521 | ResearchSourceTier.REGULATORY, |
| 1522 | ) |
| 1523 | if event.source_classification == SourceClassification.OFFICIAL_COMPANY: |
| 1524 | return ( |
| 1525 | "OFFICIAL_COMPANY" if requirement_id == "CURRENT_NEWS" else "COMPANY_FILING", |
| 1526 | ResearchSourceTier.OFFICIAL, |
| 1527 | ) |
| 1528 | return "REPUTABLE_NEWS", ResearchSourceTier.APPROVED_SECONDARY |
| 1529 | |
| 1530 | |
| 1531 | def _structured_source_tier(provider: str) -> ResearchSourceTier: |
| 1532 | normalized = str(provider).strip().upper() |
| 1533 | if normalized == "YAHOO_FINANCE_MCP": |
| 1534 | return ResearchSourceTier.APPROVED_EXTERNAL_TOOL |
| 1535 | if normalized in {"NSE", "NSE_STRUCTURED", "EXCHANGE_MARKET_DATA"}: |
| 1536 | return ResearchSourceTier.TRUSTED_MARKET_DATA |
| 1537 | if normalized == "EODHD": |
| 1538 | return ResearchSourceTier.LICENSED_STRUCTURED |
| 1539 | return ResearchSourceTier.APPROVED_SECONDARY |
| 1540 | |
| 1541 | |
| 1542 | def _market_source(provider: str) -> str: |
| 1543 | normalized = str(provider).strip().upper() |
| 1544 | if normalized in {"NSE", "NSE_STRUCTURED"}: |
| 1545 | return "EXCHANGE_MARKET_DATA" |
| 1546 | return normalized or "CONFIGURED_MARKET_DATA" |
| 1547 | |
| 1548 | |
| 1549 | def _combined_failure_reason(existing: str | None, additional: str | None) -> str: |
| 1550 | values: list[str] = [] |
| 1551 | for candidate in (existing, additional): |
| 1552 | for value in str(candidate or "").split("|"): |
| 1553 | normalized = value.strip() |
| 1554 | if normalized and normalized not in values: |
| 1555 | values.append(normalized) |
| 1556 | return "|".join(values) |
| 1557 | |
| 1558 | |
| 1559 | def _deduplicate_evidence(values: Sequence[ResearchEvidence]) -> list[ResearchEvidence]: |
| 1560 | unique: dict[tuple[str, str | None, str | None], ResearchEvidence] = {} |
| 1561 | for item in values: |
| 1562 | key = (item.evidence_id, item.fact_key, item.value_fingerprint) |
| 1563 | unique.setdefault(key, item) |
| 1564 | return list(unique.values()) |
| 1565 | |
| 1566 | |
| 1567 | def _metric(value: str) -> str: |
| 1568 | return str(value).strip().casefold().replace("-", "_").replace(" ", "_") |
| 1569 | |
| 1570 | |
| 1571 | def _period_datetime(value: str | None) -> datetime | None: |
| 1572 | if not value: |
| 1573 | return None |
| 1574 | try: |
| 1575 | parsed = datetime.fromisoformat(value) |
| 1576 | except ValueError: |
| 1577 | return None |
| 1578 | if parsed.tzinfo is None: |
| 1579 | parsed = datetime.combine(parsed.date(), time.max, tzinfo=timezone.utc) |
| 1580 | return parsed.astimezone(timezone.utc) |
| 1581 | |
| 1582 | |
| 1583 | def _aware_datetime(value: Any) -> datetime | None: |
| 1584 | if isinstance(value, datetime): |
| 1585 | parsed = value |
| 1586 | elif isinstance(value, str) and value.strip(): |
| 1587 | try: |
| 1588 | parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) |
| 1589 | except ValueError: |
| 1590 | return None |
| 1591 | else: |
| 1592 | return None |
| 1593 | if parsed.tzinfo is None: |
| 1594 | parsed = parsed.replace(tzinfo=timezone.utc) |
| 1595 | return parsed.astimezone(timezone.utc) |
| 1596 | |
| 1597 | |
| 1598 | def _fingerprint(value: Any) -> str | None: |
| 1599 | if value is None: |
| 1600 | return None |
| 1601 | if isinstance(value, Decimal): |
| 1602 | return format(value.normalize(), "f") |
| 1603 | return str(value).strip() |
| 1604 | |
| 1605 | |
| 1606 | def _date_key(value: datetime | None) -> str: |
| 1607 | return value.astimezone(timezone.utc).isoformat() if value is not None else "UNKNOWN" |
| 1608 | |
| 1609 | |
| 1610 | def _is_india(profile: CompanyResearchProfile) -> bool: |
| 1611 | return jurisdiction_for_profile(profile) == "INDIA" |
| 1612 | |
| 1613 | |
| 1614 | def _iso(value: datetime | None) -> str | None: |
| 1615 | return value.isoformat() if value is not None else None |