| 1 | """Provider-neutral historical-close population, deliberately outside request paths.""" |
| 2 | from __future__ import annotations |
| 3 | |
| 4 | import asyncio |
| 5 | from datetime import datetime, timedelta, timezone |
| 6 | from decimal import Decimal |
| 7 | from typing import Any, Protocol |
| 8 | from uuid import UUID |
| 9 | |
| 10 | from app.models import MarketPriceObservation |
| 11 | |
| 12 | |
| 13 | def has_year_historical_coverage( |
| 14 | first_observed_at: datetime, |
| 15 | latest_observed_at: datetime, |
| 16 | observation_count: int, |
| 17 | ) -> bool: |
| 18 | """Return whether durable prices span the YEAR comparison horizon.""" |
| 19 | return ( |
| 20 | observation_count >= 2 |
| 21 | and latest_observed_at - first_observed_at >= timedelta(days=365) |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | class HistoricalPriceProvider(Protocol): |
| 26 | provider_name: str |
| 27 | |
| 28 | async def closes(self, instrument: dict[str, Any], *, start: datetime, end: datetime) -> list[MarketPriceObservation]: ... |
| 29 | |
| 30 | |
| 31 | class HistoricalPriceProviderError(RuntimeError): |
| 32 | pass |
| 33 | |
| 34 | |
| 35 | class HistoricalPricePersistenceError(RuntimeError): |
| 36 | pass |
| 37 | |
| 38 | |
| 39 | class YahooHistoricalPriceProvider: |
| 40 | """Existing structured-market Yahoo integration's historical-close adapter. |
| 41 | |
| 42 | It is invoked only by an explicit/scheduled population worker, never by a |
| 43 | sector-performance request. |
| 44 | """ |
| 45 | provider_name = "YAHOO_FINANCE" |
| 46 | |
| 47 | def __init__(self, ticker_factory: Any) -> None: |
| 48 | self.ticker_factory = ticker_factory |
| 49 | |
| 50 | async def closes(self, instrument: dict[str, Any], *, start: datetime, end: datetime) -> list[MarketPriceObservation]: |
| 51 | return await asyncio.to_thread(self._closes, instrument, start, end) |
| 52 | |
| 53 | def _closes(self, instrument: dict[str, Any], start: datetime, end: datetime) -> list[MarketPriceObservation]: |
| 54 | ticker = str(instrument.get("structuredProviderTicker") or instrument.get("ticker") or "").strip() |
| 55 | if not ticker: |
| 56 | return [] |
| 57 | history = self.ticker_factory(ticker).history(start=start.date(), end=end.date(), auto_adjust=False) |
| 58 | out: list[MarketPriceObservation] = [] |
| 59 | for observed_at, row in history.iterrows(): |
| 60 | try: |
| 61 | close = Decimal(str(row["Close"])) |
| 62 | except Exception: |
| 63 | continue |
| 64 | # Yahoo can include rows whose Close value is NaN. Decimal accepts |
| 65 | # "nan", but ordering it raises decimal.InvalidOperation and would |
| 66 | # discard every otherwise valid row returned for the instrument. |
| 67 | if not close.is_finite() or close <= 0: |
| 68 | continue |
| 69 | stamp = observed_at.to_pydatetime() |
| 70 | if stamp.tzinfo is None: |
| 71 | stamp = stamp.replace(tzinfo=timezone.utc) |
| 72 | out.append(MarketPriceObservation( |
| 73 | instrument_id=UUID(str(instrument["globalInstrumentId"])), observed_at=stamp, |
| 74 | price=close, currency=instrument.get("currency"), provider=self.provider_name, |
| 75 | source_url=f"https://finance.yahoo.com/quote/{ticker}/history", retrieved_at=datetime.now(timezone.utc), |
| 76 | )) |
| 77 | return out |
| 78 | |
| 79 | |
| 80 | class HistoricalPricePopulationService: |
| 81 | """Idempotent population boundary for a scheduled/explicit market-data job.""" |
| 82 | def __init__(self, persistence, provider: HistoricalPriceProvider) -> None: |
| 83 | self.persistence = persistence |
| 84 | self.provider = provider |
| 85 | |
| 86 | async def populate(self, instruments: list[dict[str, Any]], *, start: datetime, end: datetime) -> int: |
| 87 | written = 0 |
| 88 | for instrument in instruments: |
| 89 | try: |
| 90 | observations = await self.provider.closes(instrument, start=start, end=end) |
| 91 | except Exception as exc: |
| 92 | raise HistoricalPriceProviderError("HISTORICAL_PRICE_PROVIDER_UNAVAILABLE") from exc |
| 93 | for observation in observations: |
| 94 | async_writer = getattr(self.persistence, "upsert_market_price_observation_async", None) |
| 95 | try: |
| 96 | if async_writer is not None: |
| 97 | await async_writer(observation) |
| 98 | else: |
| 99 | self.persistence.upsert_market_price_observation(observation) |
| 100 | except Exception as exc: |
| 101 | raise HistoricalPricePersistenceError("HISTORICAL_PRICE_PERSISTENCE_UNAVAILABLE") from exc |
| 102 | written += 1 |
| 103 | return written |