| 1 | """Yahoo MCP-first acquisition behind the 5A ExternalResearchToolGateway seam. |
| 2 | |
| 3 | The read-side readiness and rule-engine paths never import or call this module. |
| 4 | Only the targeted readiness executor uses it, after DB-first planning has found |
| 5 | a stale or missing requirement. Yahoo acquisition priority is kept separate |
| 6 | from the existing durable fact authority/merge rules. |
| 7 | """ |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import hashlib |
| 11 | from dataclasses import dataclass |
| 12 | from datetime import datetime, timezone |
| 13 | from decimal import Decimal |
| 14 | from typing import Any, Mapping, Protocol, Sequence |
| 15 | from uuid import UUID, uuid5, NAMESPACE_URL |
| 16 | |
| 17 | import httpx |
| 18 | from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator |
| 19 | |
| 20 | from app.fact_precedence import FactSourceTier, FinancialFact, FinancialFactKey |
| 21 | from app.models import ( |
| 22 | CompanyResearchProfile, |
| 23 | DocumentStatus, |
| 24 | DocumentType, |
| 25 | EventImpact, |
| 26 | MarketPriceObservation, |
| 27 | ProvenancedValue, |
| 28 | ReliabilityLevel, |
| 29 | ResearchDocument, |
| 30 | ResearchEvent, |
| 31 | ResearchEventType, |
| 32 | ResearchLifecycleStatus, |
| 33 | ShareholdingCategory, |
| 34 | ShareholdingSnapshot, |
| 35 | ShareholdingSnapshotValue, |
| 36 | SourceClassification, |
| 37 | SourceMode, |
| 38 | SourceType, |
| 39 | StructuredInstrumentResolution, |
| 40 | StructuredMarketSnapshot, |
| 41 | StructuredMarketSnapshotRecord, |
| 42 | TimeHorizon, |
| 43 | ) |
| 44 | from app.research_readiness import ExternalResearchToolAuthorization, ResearchRefreshTarget |
| 45 | from app.research_readiness_runtime import CapabilityExecutionProgress, CapabilityExecutionResult |
| 46 | |
| 47 | |
| 48 | YAHOO_FINANCE_MCP = "YAHOO_FINANCE_MCP" |
| 49 | _SUPPORTED_REGIONS = frozenset({"INDIA", "USA", "EUROPE"}) |
| 50 | _MCP_FIRST_REQUIREMENTS = frozenset( |
| 51 | { |
| 52 | "LATEST_PRICE", |
| 53 | "HISTORICAL_PRICE_SERIES", |
| 54 | "VALUATION_INPUTS", |
| 55 | "BUSINESS_QUALITY_FACTS", |
| 56 | "GROWTH_FACTS", |
| 57 | "BALANCE_SHEET_FACTS", |
| 58 | "QUARTERLY_FINANCIALS", |
| 59 | "CURRENT_NEWS", |
| 60 | "ORDER_BOOK_CAPEX_GUIDANCE", |
| 61 | "SHAREHOLDING", |
| 62 | "SECTOR_MACRO", |
| 63 | } |
| 64 | ) |
| 65 | |
| 66 | |
| 67 | class ExternalMcpAcquisitionError(Exception): |
| 68 | def __init__(self, safe_code: str) -> None: |
| 69 | super().__init__(safe_code) |
| 70 | self.safe_code = safe_code |
| 71 | |
| 72 | |
| 73 | class _WireModel(BaseModel): |
| 74 | model_config = ConfigDict(extra="forbid", populate_by_name=True) |
| 75 | |
| 76 | |
| 77 | class _Fact(_WireModel): |
| 78 | metric: str |
| 79 | value: Any |
| 80 | unit: str | None = None |
| 81 | as_of: datetime | None = Field(default=None, alias="asOf") |
| 82 | published_at: datetime | None = Field(default=None, alias="publishedAt") |
| 83 | source_url: str = Field(alias="sourceUrl") |
| 84 | confidence: float = Field(ge=0, le=1) |
| 85 | raw_field_origin: str | None = Field(default=None, alias="rawFieldOrigin") |
| 86 | |
| 87 | |
| 88 | class _FinancialFact(_Fact): |
| 89 | period_end: str = Field(alias="periodEnd") |
| 90 | period_type: str = Field(alias="periodType") |
| 91 | reporting_basis: str = Field(alias="reportingBasis") |
| 92 | |
| 93 | |
| 94 | class _Observation(_WireModel): |
| 95 | observed_at: datetime = Field(alias="observedAt") |
| 96 | price: Decimal |
| 97 | currency: str | None = None |
| 98 | |
| 99 | @field_validator("price") |
| 100 | @classmethod |
| 101 | def positive_finite(cls, value: Decimal) -> Decimal: |
| 102 | if not value.is_finite() or value <= 0: |
| 103 | raise ValueError("market price must be positive and finite") |
| 104 | return value |
| 105 | |
| 106 | |
| 107 | class _Article(_WireModel): |
| 108 | headline: str |
| 109 | url: str |
| 110 | published_at: datetime = Field(alias="publishedAt") |
| 111 | publisher: str |
| 112 | issuer_symbol: str = Field(alias="issuerSymbol") |
| 113 | summary: str | None = None |
| 114 | event_type: str | None = Field(default=None, alias="eventType") |
| 115 | |
| 116 | |
| 117 | class _CompanyProfile(_WireModel): |
| 118 | company_name: str | None = Field(default=None, alias="companyName") |
| 119 | sector: str | None = None |
| 120 | industry: str | None = None |
| 121 | |
| 122 | |
| 123 | class _Shareholding(_WireModel): |
| 124 | period_end: datetime = Field(alias="periodEnd") |
| 125 | promoter_holding_percent: Decimal | None = Field(default=None, alias="promoterHoldingPercent") |
| 126 | promoter_pledge_percent: Decimal | None = Field(default=None, alias="promoterPledgePercent") |
| 127 | promoter_pledge_basis: str | None = Field(default=None, alias="promoterPledgeBasis") |
| 128 | fii_fpi_percent: Decimal | None = Field(default=None, alias="fiiFpiPercent") |
| 129 | dii_percent: Decimal | None = Field(default=None, alias="diiPercent") |
| 130 | public_retail_percent: Decimal | None = Field(default=None, alias="publicRetailPercent") |
| 131 | institutional_ownership_percent: Decimal | None = Field( |
| 132 | default=None, alias="institutionalOwnershipPercent" |
| 133 | ) |
| 134 | |
| 135 | |
| 136 | class YahooMcpNormalizedResult(_WireModel): |
| 137 | adapter_version: str = Field(alias="adapterVersion") |
| 138 | provider_id: str = Field(alias="providerId") |
| 139 | source_tier: str = Field(alias="sourceTier") |
| 140 | source_tool: str = Field(alias="sourceTool") |
| 141 | region: str |
| 142 | requirement_id: str = Field(alias="requirementId") |
| 143 | global_instrument_id: UUID = Field(alias="globalInstrumentId") |
| 144 | symbol: str |
| 145 | exchange: str | None = None |
| 146 | currency: str | None = None |
| 147 | retrieved_at: datetime = Field(alias="retrievedAt") |
| 148 | observed_at: datetime | None = Field(default=None, alias="observedAt") |
| 149 | source_url: str = Field(alias="sourceUrl") |
| 150 | confidence: float = Field(ge=0, le=1) |
| 151 | freshness: str |
| 152 | structured_facts: tuple[_Fact, ...] = Field(alias="structuredFacts") |
| 153 | financial_facts: tuple[_FinancialFact, ...] = Field(alias="financialFacts") |
| 154 | acquisition_outcome: str = Field(default="SUCCESS", alias="acquisitionOutcome") |
| 155 | market_observations: tuple[_Observation, ...] = Field(alias="marketObservations") |
| 156 | company_profile: _CompanyProfile | None = Field(default=None, alias="companyProfile") |
| 157 | news: tuple[_Article, ...] = () |
| 158 | events: tuple[_Article, ...] = () |
| 159 | shareholding: _Shareholding | None = None |
| 160 | |
| 161 | @field_validator("provider_id") |
| 162 | @classmethod |
| 163 | def yahoo_only(cls, value: str) -> str: |
| 164 | if value != YAHOO_FINANCE_MCP: |
| 165 | raise ValueError("unexpected external provider") |
| 166 | return value |
| 167 | |
| 168 | |
| 169 | @dataclass(frozen=True) |
| 170 | class ProviderAcquisitionRoute: |
| 171 | region: str |
| 172 | requirement_id: str |
| 173 | providers: tuple[str, ...] |
| 174 | |
| 175 | |
| 176 | class McpFirstProviderPriority: |
| 177 | """Central acquisition order. Durable evidence authority lives elsewhere.""" |
| 178 | |
| 179 | _REGIONAL_FALLBACKS: Mapping[str, Mapping[str, tuple[str, ...]]] = { |
| 180 | "INDIA": { |
| 181 | "LATEST_PRICE": ("YAHOO_FINANCE_REST",), |
| 182 | "HISTORICAL_PRICE_SERIES": ("YAHOO_FINANCE_REST",), |
| 183 | "VALUATION_INPUTS": ("NSE_OR_APPROVED_STRUCTURED",), |
| 184 | "BUSINESS_QUALITY_FACTS": ("NSE",), |
| 185 | "GROWTH_FACTS": ("NSE",), |
| 186 | "BALANCE_SHEET_FACTS": ("NSE",), |
| 187 | "QUARTERLY_FINANCIALS": ("NSE",), |
| 188 | "CURRENT_NEWS": ("GLOBAL_NEWS_SEARCH", "NSE"), |
| 189 | "ORDER_BOOK_CAPEX_GUIDANCE": ("NSE",), |
| 190 | "SHAREHOLDING": ("NSE_XBRL",), |
| 191 | "SECTOR_MACRO": ("EXISTING_APPROVED_RESEARCH",), |
| 192 | }, |
| 193 | "USA": { |
| 194 | "LATEST_PRICE": ("YAHOO_FINANCE_REST",), |
| 195 | "HISTORICAL_PRICE_SERIES": ("YAHOO_FINANCE_REST",), |
| 196 | "VALUATION_INPUTS": ("SEC_EDGAR_OR_APPROVED_STRUCTURED",), |
| 197 | "BUSINESS_QUALITY_FACTS": ("SEC_EDGAR",), |
| 198 | "GROWTH_FACTS": ("SEC_EDGAR",), |
| 199 | "BALANCE_SHEET_FACTS": ("SEC_EDGAR",), |
| 200 | "QUARTERLY_FINANCIALS": ("SEC_EDGAR",), |
| 201 | "CURRENT_NEWS": ("GLOBAL_NEWS_SEARCH",), |
| 202 | "ORDER_BOOK_CAPEX_GUIDANCE": ("SEC_EDGAR_OR_APPROVED_RESEARCH",), |
| 203 | "SHAREHOLDING": ("UNAVAILABLE",), |
| 204 | "SECTOR_MACRO": ("EXISTING_APPROVED_RESEARCH",), |
| 205 | }, |
| 206 | "EUROPE": { |
| 207 | "LATEST_PRICE": ("YAHOO_FINANCE_REST",), |
| 208 | "HISTORICAL_PRICE_SERIES": ("YAHOO_FINANCE_REST",), |
| 209 | "VALUATION_INPUTS": ("EODHD_OR_APPROVED_STRUCTURED",), |
| 210 | "BUSINESS_QUALITY_FACTS": ("EODHD",), |
| 211 | "GROWTH_FACTS": ("EODHD",), |
| 212 | "BALANCE_SHEET_FACTS": ("EODHD",), |
| 213 | "QUARTERLY_FINANCIALS": ("EODHD",), |
| 214 | "CURRENT_NEWS": ("GLOBAL_NEWS_SEARCH",), |
| 215 | "ORDER_BOOK_CAPEX_GUIDANCE": ("EODHD_OR_APPROVED_RESEARCH",), |
| 216 | "SHAREHOLDING": ("UNAVAILABLE",), |
| 217 | "SECTOR_MACRO": ("EXISTING_APPROVED_RESEARCH",), |
| 218 | }, |
| 219 | } |
| 220 | |
| 221 | def route(self, region: str, requirement_id: str) -> ProviderAcquisitionRoute: |
| 222 | normalized_region = region.strip().upper() |
| 223 | normalized_requirement = requirement_id.strip().upper() |
| 224 | fallbacks = self._REGIONAL_FALLBACKS.get(normalized_region, {}).get( |
| 225 | normalized_requirement, () |
| 226 | ) |
| 227 | providers = ( |
| 228 | (YAHOO_FINANCE_MCP, *fallbacks) |
| 229 | if normalized_region in _SUPPORTED_REGIONS |
| 230 | and normalized_requirement in _MCP_FIRST_REQUIREMENTS |
| 231 | else fallbacks |
| 232 | ) |
| 233 | return ProviderAcquisitionRoute(normalized_region, normalized_requirement, providers) |
| 234 | |
| 235 | |
| 236 | class ExternalResearchToolGatewayClient(Protocol): |
| 237 | async def acquire_requirement( |
| 238 | self, |
| 239 | profile: CompanyResearchProfile, |
| 240 | *, |
| 241 | region: str, |
| 242 | requirement_id: str, |
| 243 | authorization: ExternalResearchToolAuthorization, |
| 244 | request_id: str, |
| 245 | ) -> YahooMcpNormalizedResult: ... |
| 246 | |
| 247 | |
| 248 | class HttpExternalResearchToolGateway: |
| 249 | """Narrow internal HTTP command; callers cannot submit a provider tool name.""" |
| 250 | |
| 251 | def __init__(self, base_url: str, timeout_seconds: float, service_identity: str) -> None: |
| 252 | self.base_url = base_url.rstrip("/") |
| 253 | self.timeout_seconds = timeout_seconds |
| 254 | self.service_identity = service_identity |
| 255 | |
| 256 | async def acquire_requirement( |
| 257 | self, |
| 258 | profile: CompanyResearchProfile, |
| 259 | *, |
| 260 | region: str, |
| 261 | requirement_id: str, |
| 262 | authorization: ExternalResearchToolAuthorization, |
| 263 | request_id: str, |
| 264 | ) -> YahooMcpNormalizedResult: |
| 265 | symbol = profile.provider_instrument_ids.get("YAHOO_FINANCE") |
| 266 | if not symbol: |
| 267 | raise ExternalMcpAcquisitionError("VERIFIED_YAHOO_MAPPING_REQUIRED") |
| 268 | payload = { |
| 269 | "providerId": YAHOO_FINANCE_MCP, |
| 270 | "region": region, |
| 271 | "requirementId": requirement_id, |
| 272 | "globalInstrumentId": str(profile.instrument_id), |
| 273 | "providerSymbol": symbol, |
| 274 | "expectedExchange": profile.exchange, |
| 275 | "expectedCurrency": profile.currency, |
| 276 | "authorization": authorization.as_gateway_payload(), |
| 277 | } |
| 278 | headers = { |
| 279 | "X-Request-ID": request_id, |
| 280 | "X-Correlation-ID": request_id, |
| 281 | "X-AIP-Service-Identity": self.service_identity, |
| 282 | } |
| 283 | try: |
| 284 | async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: |
| 285 | response = await client.post( |
| 286 | f"{self.base_url}/internal/v1/external-research/acquire", |
| 287 | json=payload, |
| 288 | headers=headers, |
| 289 | ) |
| 290 | except httpx.TimeoutException as exc: |
| 291 | raise ExternalMcpAcquisitionError("DOWNSTREAM_TIMEOUT") from exc |
| 292 | except httpx.HTTPError as exc: |
| 293 | raise ExternalMcpAcquisitionError("EXTERNAL_PROVIDER_UNAVAILABLE") from exc |
| 294 | try: |
| 295 | envelope = response.json() |
| 296 | except ValueError as exc: |
| 297 | raise ExternalMcpAcquisitionError("EXTERNAL_SCHEMA_INVALID") from exc |
| 298 | if not isinstance(envelope, dict): |
| 299 | raise ExternalMcpAcquisitionError("EXTERNAL_SCHEMA_INVALID") |
| 300 | if response.status_code >= 400 or envelope.get("ok") is not True: |
| 301 | code = str((envelope.get("error") or {}).get("code") or "EXTERNAL_PROVIDER_UNAVAILABLE") |
| 302 | raise ExternalMcpAcquisitionError(code) |
| 303 | try: |
| 304 | result = YahooMcpNormalizedResult.model_validate(envelope.get("data")) |
| 305 | except ValidationError as exc: |
| 306 | raise ExternalMcpAcquisitionError("EXTERNAL_SCHEMA_INVALID") from exc |
| 307 | if result.global_instrument_id != profile.instrument_id: |
| 308 | raise ExternalMcpAcquisitionError("EXTERNAL_IDENTITY_CONFLICT") |
| 309 | return result |
| 310 | |
| 311 | |
| 312 | class YahooMcpResultPersister: |
| 313 | """Translate the wire contract into existing provider-neutral durable models.""" |
| 314 | |
| 315 | def __init__(self, repository) -> None: |
| 316 | self.repository = repository |
| 317 | |
| 318 | async def persist( |
| 319 | self, result: YahooMcpNormalizedResult, profile: CompanyResearchProfile |
| 320 | ) -> int: |
| 321 | if result.global_instrument_id != profile.instrument_id: |
| 322 | raise ExternalMcpAcquisitionError("EXTERNAL_IDENTITY_CONFLICT") |
| 323 | if result.symbol.upper() != profile.provider_instrument_ids.get( |
| 324 | "YAHOO_FINANCE", "" |
| 325 | ).upper(): |
| 326 | raise ExternalMcpAcquisitionError("EXTERNAL_IDENTITY_CONFLICT") |
| 327 | written = 0 |
| 328 | if result.structured_facts: |
| 329 | await self._persist_structured(result, profile) |
| 330 | written += len(result.structured_facts) |
| 331 | financial = [ |
| 332 | self._financial_fact(item, result, profile) |
| 333 | for item in result.financial_facts |
| 334 | ] |
| 335 | if financial: |
| 336 | written += await self.repository.persist_international_financial_facts_async(financial) |
| 337 | # A lower-tier fact may be rejected because an official value already exists. |
| 338 | # The valid provider result still satisfies acquisition without downgrading it. |
| 339 | if written == 0: |
| 340 | written += len(financial) |
| 341 | for item in result.market_observations: |
| 342 | await self.repository.upsert_market_price_observation_async( |
| 343 | MarketPriceObservation( |
| 344 | instrument_id=profile.instrument_id, |
| 345 | observed_at=_aware(item.observed_at), |
| 346 | price=item.price, |
| 347 | currency=item.currency or result.currency or profile.currency, |
| 348 | provider=YAHOO_FINANCE_MCP, |
| 349 | source_url=result.source_url, |
| 350 | retrieved_at=_aware(result.retrieved_at), |
| 351 | ) |
| 352 | ) |
| 353 | written += 1 |
| 354 | for article in (*result.news, *result.events): |
| 355 | await self._persist_article(article, result, profile) |
| 356 | written += 1 |
| 357 | if result.shareholding is not None: |
| 358 | await self._persist_shareholding(result.shareholding, result, profile) |
| 359 | written += 1 |
| 360 | empty_news = result.requirement_id == "CURRENT_NEWS" and result.acquisition_outcome == "SUCCESS_EMPTY" and not result.news |
| 361 | if written < 1 and not empty_news: |
| 362 | raise ExternalMcpAcquisitionError("EXTERNAL_RESULT_INCOMPLETE") |
| 363 | recorder = getattr(self.repository, "record_acquisition_observation", None) |
| 364 | if callable(recorder): |
| 365 | await recorder(profile.instrument_id, result.requirement_id, YAHOO_FINANCE_MCP, |
| 366 | "SUCCESS_EMPTY" if empty_news else "SUCCESS", result.retrieved_at, result.source_url, evidence_count=written) |
| 367 | return written |
| 368 | |
| 369 | async def _persist_structured( |
| 370 | self, result: YahooMcpNormalizedResult, profile: CompanyResearchProfile |
| 371 | ) -> None: |
| 372 | facts = { |
| 373 | item.metric: ProvenancedValue( |
| 374 | value=item.value, |
| 375 | unit=item.unit, |
| 376 | as_of_date=_aware(item.as_of) if item.as_of else result.observed_at, |
| 377 | source_url=item.source_url, |
| 378 | source_name="Yahoo Finance MCP", |
| 379 | source_type="EXTERNAL_MCP_PROVIDER", |
| 380 | published_at=_aware(item.published_at) if item.published_at else None, |
| 381 | retrieved_at=_aware(result.retrieved_at), |
| 382 | confidence=item.confidence, |
| 383 | calculation_basis=( |
| 384 | f"{result.source_tool}:{item.raw_field_origin}" |
| 385 | if item.raw_field_origin |
| 386 | else result.source_tool |
| 387 | ), |
| 388 | ) |
| 389 | for item in result.structured_facts |
| 390 | } |
| 391 | resolution = StructuredInstrumentResolution( |
| 392 | instrument_id=profile.instrument_id, |
| 393 | provider=YAHOO_FINANCE_MCP, |
| 394 | provider_ticker=result.symbol, |
| 395 | company_name=profile.company_name, |
| 396 | exchange=result.exchange, |
| 397 | currency=result.currency, |
| 398 | confidence=result.confidence, |
| 399 | resolved_at=_aware(result.retrieved_at), |
| 400 | status="VERIFIED_MAPPING_VALIDATED", |
| 401 | ) |
| 402 | snapshot = StructuredMarketSnapshot( |
| 403 | resolution=resolution, |
| 404 | status="STRUCTURED_PROVIDER_AVAILABLE", |
| 405 | retrieved_at=_aware(result.retrieved_at), |
| 406 | market_as_of=_aware(result.observed_at) if result.observed_at else None, |
| 407 | source_name="Yahoo Finance MCP", |
| 408 | source_type="EXTERNAL_MCP_PROVIDER", |
| 409 | source_url=result.source_url, |
| 410 | facts=facts, |
| 411 | statement_facts=[], |
| 412 | news=[], |
| 413 | accepted_fields_count=len(facts), |
| 414 | ) |
| 415 | now = datetime.now(timezone.utc) |
| 416 | record = StructuredMarketSnapshotRecord( |
| 417 | instrument_id=profile.instrument_id, |
| 418 | provider=YAHOO_FINANCE_MCP, |
| 419 | provider_instrument_id=result.symbol, |
| 420 | exchange=result.exchange, |
| 421 | mic=profile.mic, |
| 422 | currency=result.currency, |
| 423 | source_url=result.source_url, |
| 424 | source_name="Yahoo Finance MCP", |
| 425 | source_type="EXTERNAL_MCP_PROVIDER", |
| 426 | source_identity=f"{YAHOO_FINANCE_MCP}:{result.symbol}:{result.source_tool}", |
| 427 | market_as_of=snapshot.market_as_of, |
| 428 | retrieved_at=snapshot.retrieved_at, |
| 429 | persisted_at=now, |
| 430 | last_price_at=now if "latestPrice" in facts else None, |
| 431 | last_valuation_at=( |
| 432 | now |
| 433 | if any(key in facts for key in ("trailingPE", "forwardPE", "priceToBook")) |
| 434 | else None |
| 435 | ), |
| 436 | last_fundamentals_at=( |
| 437 | now |
| 438 | if any(key in facts for key in ("trailingEps", "roe", "roa", "roce", "sector")) |
| 439 | else None |
| 440 | ), |
| 441 | last_success_at=now, |
| 442 | last_provider_attempt_at=now, |
| 443 | acquisition_status="SUCCESS", |
| 444 | snapshot=snapshot, |
| 445 | ) |
| 446 | await self.repository.persist_structured_market_snapshot_async(record) |
| 447 | |
| 448 | @staticmethod |
| 449 | def _financial_fact( |
| 450 | item: _FinancialFact, |
| 451 | result: YahooMcpNormalizedResult, |
| 452 | profile: CompanyResearchProfile, |
| 453 | ) -> FinancialFact: |
| 454 | return FinancialFact( |
| 455 | FinancialFactKey( |
| 456 | profile.instrument_id, |
| 457 | item.metric, |
| 458 | item.period_end, |
| 459 | item.period_type, |
| 460 | item.reporting_basis, |
| 461 | ), |
| 462 | ProvenancedValue( |
| 463 | value=Decimal(str(item.value)), |
| 464 | unit=item.unit, |
| 465 | as_of_date=_aware(item.as_of) if item.as_of else _period_datetime(item.period_end), |
| 466 | source_url=item.source_url, |
| 467 | source_name="Yahoo Finance MCP", |
| 468 | source_type="EXTERNAL_MCP_PROVIDER", |
| 469 | published_at=_aware(item.published_at) if item.published_at else None, |
| 470 | retrieved_at=_aware(result.retrieved_at), |
| 471 | confidence=item.confidence, |
| 472 | calculation_basis=( |
| 473 | f"{result.source_tool}:{item.raw_field_origin}" |
| 474 | if item.raw_field_origin |
| 475 | else result.source_tool |
| 476 | ), |
| 477 | ), |
| 478 | FactSourceTier.YAHOO, |
| 479 | YAHOO_FINANCE_MCP, |
| 480 | ( |
| 481 | f"{YAHOO_FINANCE_MCP}:{result.symbol}:{result.source_tool}:" |
| 482 | f"{item.metric}:{item.period_end}:{item.period_type}" |
| 483 | ), |
| 484 | SourceMode.REAL, |
| 485 | ) |
| 486 | |
| 487 | async def _persist_article( |
| 488 | self, |
| 489 | article: _Article, |
| 490 | result: YahooMcpNormalizedResult, |
| 491 | profile: CompanyResearchProfile, |
| 492 | ) -> None: |
| 493 | identity = f"{article.url}|{article.headline}|{article.published_at.date().isoformat()}" |
| 494 | document_id = uuid5(NAMESPACE_URL, identity) |
| 495 | document = ResearchDocument( |
| 496 | document_id=document_id, |
| 497 | canonical_url=article.url, |
| 498 | original_url=article.url, |
| 499 | title=article.headline, |
| 500 | source_type=SourceType.NEWS, |
| 501 | source_classification=SourceClassification.REPUTABLE_NEWS, |
| 502 | source_name="Yahoo Finance MCP", |
| 503 | publisher=article.publisher, |
| 504 | published_at=_aware(article.published_at), |
| 505 | retrieved_at=_aware(result.retrieved_at), |
| 506 | content_type="application/json", |
| 507 | document_type=DocumentType.TEXT, |
| 508 | raw_text=None, |
| 509 | normalized_text=None, |
| 510 | content_hash=hashlib.sha256(identity.encode("utf-8")).hexdigest(), |
| 511 | instrument_id=profile.instrument_id, |
| 512 | company_id=profile.company_id, |
| 513 | country=profile.country, |
| 514 | exchange=profile.exchange, |
| 515 | status=DocumentStatus.PROCESSED, |
| 516 | reliability_level=ReliabilityLevel.LEVEL_B, |
| 517 | entity_resolution_confidence=0.99, |
| 518 | source_mode=SourceMode.REAL, |
| 519 | freshness="REAL", |
| 520 | discovery_provider=YAHOO_FINANCE_MCP, |
| 521 | source_independence_key=article.url.casefold(), |
| 522 | ) |
| 523 | event_type = _event_type(article.event_type) |
| 524 | event = ResearchEvent( |
| 525 | event_id=uuid5(NAMESPACE_URL, f"event|{identity}"), |
| 526 | instrument_id=profile.instrument_id, |
| 527 | company_id=profile.company_id, |
| 528 | event_type=event_type, |
| 529 | event_date=_aware(article.published_at), |
| 530 | detected_at=_aware(result.retrieved_at), |
| 531 | title=article.headline, |
| 532 | summary=article.summary or article.headline, |
| 533 | source_document_id=document_id, |
| 534 | source_url=article.url, |
| 535 | source_type=SourceType.NEWS, |
| 536 | source_classification=SourceClassification.REPUTABLE_NEWS, |
| 537 | reliability=ReliabilityLevel.LEVEL_B, |
| 538 | source_mode=SourceMode.REAL, |
| 539 | confidence=result.confidence, |
| 540 | impact=EventImpact.UNCERTAIN, |
| 541 | time_horizon=TimeHorizon.UNKNOWN, |
| 542 | status=ResearchLifecycleStatus.VALIDATED, |
| 543 | raw_evidence_reference=article.headline, |
| 544 | published_at=_aware(article.published_at), |
| 545 | retrieved_at=_aware(result.retrieved_at), |
| 546 | independence_key=f"{YAHOO_FINANCE_MCP}:{article.url.casefold()}", |
| 547 | ) |
| 548 | await self.repository.persist_external_mcp_evidence_async(document, event) |
| 549 | |
| 550 | async def _persist_shareholding( |
| 551 | self, |
| 552 | value: _Shareholding, |
| 553 | result: YahooMcpNormalizedResult, |
| 554 | profile: CompanyResearchProfile, |
| 555 | ) -> None: |
| 556 | exact = { |
| 557 | ShareholdingCategory.PROMOTER: value.promoter_holding_percent, |
| 558 | ShareholdingCategory.PROMOTER_PLEDGE: value.promoter_pledge_percent, |
| 559 | ShareholdingCategory.FII_FPI: value.fii_fpi_percent, |
| 560 | ShareholdingCategory.DII: value.dii_percent, |
| 561 | ShareholdingCategory.PUBLIC_RETAIL: value.public_retail_percent, |
| 562 | } |
| 563 | values = [ |
| 564 | ShareholdingSnapshotValue( |
| 565 | category=category, |
| 566 | percentage=percentage, |
| 567 | metric_basis=( |
| 568 | value.promoter_pledge_basis |
| 569 | if category == ShareholdingCategory.PROMOTER_PLEDGE |
| 570 | else None |
| 571 | ), |
| 572 | raw_source_label=category.value, |
| 573 | source_locator=result.source_tool, |
| 574 | ) |
| 575 | for category, percentage in exact.items() |
| 576 | if percentage is not None |
| 577 | ] |
| 578 | snapshot = ShareholdingSnapshot( |
| 579 | instrument_id=profile.instrument_id, |
| 580 | period_end=_aware(value.period_end), |
| 581 | filing_basis="YAHOO_MCP_EXACT_NORMALIZED_FIELDS", |
| 582 | source_provider=YAHOO_FINANCE_MCP, |
| 583 | source_type="EXTERNAL_MCP_PROVIDER", |
| 584 | source_identity_key=( |
| 585 | f"{YAHOO_FINANCE_MCP}:{result.symbol}:" |
| 586 | f"{value.period_end.date().isoformat()}" |
| 587 | ), |
| 588 | source_url=result.source_url, |
| 589 | retrieved_at=_aware(result.retrieved_at), |
| 590 | confidence=Decimal(str(result.confidence)), |
| 591 | reliability_level=ReliabilityLevel.LEVEL_B, |
| 592 | source_mode=SourceMode.REAL, |
| 593 | values=values, |
| 594 | ) |
| 595 | await self.repository.persist_external_mcp_shareholding_async(snapshot) |
| 596 | |
| 597 | |
| 598 | class McpFirstResearchCapabilityExecutor: |
| 599 | """Try configured Yahoo MCP capabilities, then call the unchanged executor.""" |
| 600 | |
| 601 | def __init__( |
| 602 | self, |
| 603 | legacy_executor, |
| 604 | repository, |
| 605 | gateway: ExternalResearchToolGatewayClient, |
| 606 | *, |
| 607 | enabled: bool, |
| 608 | priority: McpFirstProviderPriority | None = None, |
| 609 | ) -> None: |
| 610 | self.legacy_executor = legacy_executor |
| 611 | self.repository = repository |
| 612 | self.gateway = gateway |
| 613 | self.enabled = enabled |
| 614 | self.priority = priority or McpFirstProviderPriority() |
| 615 | self.persister = YahooMcpResultPersister(repository) |
| 616 | |
| 617 | async def execute_primary( |
| 618 | self, |
| 619 | global_instrument_id: UUID, |
| 620 | targets: Sequence[ResearchRefreshTarget], |
| 621 | *, |
| 622 | jurisdiction: str, |
| 623 | correlation_id: str | None, |
| 624 | identity_headers: Mapping[str, str | None] | None, |
| 625 | progress: CapabilityExecutionProgress | None = None, |
| 626 | ) -> CapabilityExecutionResult: |
| 627 | completed: set[str] = set() |
| 628 | executed: list[str] = [] |
| 629 | mcp_failures: dict[str, str] = {} |
| 630 | profile = self.repository.profile(global_instrument_id) |
| 631 | request_id = correlation_id or str(global_instrument_id) |
| 632 | if self.enabled: |
| 633 | for target in targets: |
| 634 | route = self.priority.route(jurisdiction, target.requirement_id) |
| 635 | if not route.providers or route.providers[0] != YAHOO_FINANCE_MCP: |
| 636 | continue |
| 637 | authorization = target.authority_policy.fallback_policy.authorize_external_tool( |
| 638 | global_instrument_id=global_instrument_id, |
| 639 | requirement_id=target.requirement_id, |
| 640 | status=target.reason, |
| 641 | confidence=0.0, |
| 642 | permitted_provider_ids=(YAHOO_FINANCE_MCP,), |
| 643 | ) |
| 644 | if authorization is None: |
| 645 | mcp_failures[target.requirement_id] = "EXTERNAL_FALLBACK_NOT_AUTHORIZED" |
| 646 | if progress is not None: |
| 647 | progress.failed( |
| 648 | target.requirement_id, "EXTERNAL_FALLBACK_NOT_AUTHORIZED" |
| 649 | ) |
| 650 | continue |
| 651 | capability = f"{YAHOO_FINANCE_MCP}:{target.requirement_id}" |
| 652 | executed.append(capability) |
| 653 | if progress is not None: |
| 654 | progress.executed(capability) |
| 655 | try: |
| 656 | result = await self.gateway.acquire_requirement( |
| 657 | profile, |
| 658 | region=jurisdiction, |
| 659 | requirement_id=target.requirement_id, |
| 660 | authorization=authorization, |
| 661 | request_id=request_id, |
| 662 | ) |
| 663 | await self.persister.persist(result, profile) |
| 664 | except ExternalMcpAcquisitionError as exc: |
| 665 | mcp_failures[target.requirement_id] = exc.safe_code |
| 666 | if progress is not None: |
| 667 | progress.failed(target.requirement_id, exc.safe_code) |
| 668 | continue |
| 669 | except Exception: |
| 670 | # Provider and persistence details never escape targeted ensure. |
| 671 | # The unchanged regional provider receives the requirement. |
| 672 | mcp_failures[target.requirement_id] = "EXTERNAL_PROVIDER_UNAVAILABLE" |
| 673 | if progress is not None: |
| 674 | progress.failed( |
| 675 | target.requirement_id, "EXTERNAL_PROVIDER_UNAVAILABLE" |
| 676 | ) |
| 677 | continue |
| 678 | if result.acquisition_outcome == "SUCCESS_EMPTY": |
| 679 | # Empty acquisition metadata is not event evidence. Try approved fallbacks. |
| 680 | continue |
| 681 | completed.add(target.requirement_id) |
| 682 | if progress is not None: |
| 683 | progress.satisfied(target.requirement_id) |
| 684 | |
| 685 | remaining = tuple(target for target in targets if target.requirement_id not in completed) |
| 686 | legacy = ( |
| 687 | await self.legacy_executor.execute_primary( |
| 688 | global_instrument_id, |
| 689 | remaining, |
| 690 | jurisdiction=jurisdiction, |
| 691 | correlation_id=correlation_id, |
| 692 | identity_headers=identity_headers, |
| 693 | progress=progress, |
| 694 | ) |
| 695 | if remaining |
| 696 | else CapabilityExecutionResult() |
| 697 | ) |
| 698 | failures = dict(legacy.failures) |
| 699 | for requirement_id, safe_code in mcp_failures.items(): |
| 700 | if requirement_id in failures: |
| 701 | failures[requirement_id] = f"{safe_code}|{failures[requirement_id]}" |
| 702 | else: |
| 703 | # A legacy executor may complete normally while finding no |
| 704 | # evidence. Retain the concrete MCP result until the runtime |
| 705 | # re-reads durable evidence and can prove the requirement was |
| 706 | # satisfied by that fallback. |
| 707 | failures[requirement_id] = safe_code |
| 708 | return CapabilityExecutionResult( |
| 709 | (*executed, *legacy.executed_capabilities), |
| 710 | failures, |
| 711 | tuple(sorted(completed | set(legacy.satisfied_requirement_ids))), |
| 712 | ) |
| 713 | |
| 714 | async def execute_approved_fallbacks( |
| 715 | self, global_instrument_id: UUID, targets: Sequence[ResearchRefreshTarget] |
| 716 | ) -> CapabilityExecutionResult: |
| 717 | return await self.legacy_executor.execute_approved_fallbacks(global_instrument_id, targets) |
| 718 | |
| 719 | |
| 720 | def _event_type(value: str | None) -> ResearchEventType: |
| 721 | normalized = str(value or "").strip().upper() |
| 722 | try: |
| 723 | return ResearchEventType(normalized) |
| 724 | except ValueError: |
| 725 | return ResearchEventType.OTHER |
| 726 | |
| 727 | |
| 728 | def _period_datetime(value: str) -> datetime | None: |
| 729 | try: |
| 730 | parsed = datetime.fromisoformat(value) |
| 731 | except ValueError: |
| 732 | return None |
| 733 | return _aware(parsed) |
| 734 | |
| 735 | |
| 736 | def _aware(value: datetime) -> datetime: |
| 737 | return ( |
| 738 | value.replace(tzinfo=timezone.utc) |
| 739 | if value.tzinfo is None or value.utcoffset() is None |
| 740 | else value.astimezone(timezone.utc) |
| 741 | ) |