| 1 | from __future__ import annotations |
| 2 | |
| 3 | import asyncio |
| 4 | import time |
| 5 | from dataclasses import dataclass |
| 6 | from uuid import NAMESPACE_URL, UUID, uuid5 |
| 7 | from decimal import Decimal |
| 8 | from datetime import datetime, timedelta, timezone |
| 9 | import re |
| 10 | import logging |
| 11 | |
| 12 | import httpx |
| 13 | |
| 14 | from app.models import ( |
| 15 | CompanyResearchProfile, |
| 16 | CategoryEvidence, |
| 17 | EvidenceState, |
| 18 | EtfResearchProfile, |
| 19 | EventImpact, |
| 20 | PortfolioResearchCompany, |
| 21 | PortfolioResearchSummary, |
| 22 | PublicAnalyst, |
| 23 | MarketFundamentals, |
| 24 | ResearchSummary, |
| 25 | StructuredMarketSnapshotRecord, |
| 26 | ) |
| 27 | from app.repository import ResearchRepository |
| 28 | from app.scoring import canonical_read_model_score |
| 29 | from app.settings import Settings |
| 30 | from app.source_registry import registered_sources_for |
| 31 | from app.structured_research import enrich_company_research |
| 32 | from app.structured_market import StructuredProviderError, StructuredResearchProvider, YahooFinanceProvider, _is_financial_identity |
| 33 | from app.market_sessions import class_due, market_session_status, price_sync_eligible |
| 34 | from app.international_fundamentals import InternationalFundamentalsResult, international_provider_for |
| 35 | from app.sector_performance import belongs_to_region |
| 36 | |
| 37 | logger = logging.getLogger(__name__) |
| 38 | |
| 39 | |
| 40 | @dataclass(frozen=True) |
| 41 | class StructuredReconciliationOutcome: |
| 42 | snapshot: object | None |
| 43 | error: str | None |
| 44 | due_classes: frozenset[str] |
| 45 | market_status: str |
| 46 | |
| 47 | |
| 48 | class GlobalInstrumentNotFoundError(Exception): |
| 49 | """The portfolio service confirmed that a global instrument does not exist.""" |
| 50 | |
| 51 | |
| 52 | class PortfolioServiceUnavailableError(Exception): |
| 53 | """A global instrument could not be resolved because portfolio-service is unavailable.""" |
| 54 | |
| 55 | |
| 56 | class WatchlistNotFoundError(Exception): |
| 57 | """The authenticated user does not own the requested watchlist.""" |
| 58 | |
| 59 | |
| 60 | class WatchlistRegionMismatchError(Exception): |
| 61 | """The canonical instrument region does not match the target watchlist.""" |
| 62 | |
| 63 | |
| 64 | class PortfolioResearchOrchestrator: |
| 65 | def __init__( |
| 66 | self, |
| 67 | repository: ResearchRepository, |
| 68 | settings: Settings, |
| 69 | client: httpx.AsyncClient | None = None, |
| 70 | structured_provider: StructuredResearchProvider | None = None, |
| 71 | ) -> None: |
| 72 | self.repository = repository |
| 73 | self.settings = settings |
| 74 | self._client = client or httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) |
| 75 | self.structured_provider = structured_provider or YahooFinanceProvider(settings) |
| 76 | self._structured_by_instrument: dict[UUID, tuple[datetime, object]] = {} |
| 77 | |
| 78 | async def active_global_equities( |
| 79 | self, |
| 80 | *, |
| 81 | correlation_id: str | None = None, |
| 82 | identity_headers: dict[str, str | None] | None = None, |
| 83 | ) -> list[dict]: |
| 84 | """Read the portfolio-service owned active-equity universe; never mutate it.""" |
| 85 | from app.global_scanner import CanonicalEquityUniverse |
| 86 | |
| 87 | try: |
| 88 | return await CanonicalEquityUniverse(self._client, self.settings.portfolio_service_base_url).active_global_equities( |
| 89 | correlation_id=correlation_id, identity_headers=identity_headers, |
| 90 | ) |
| 91 | except (httpx.HTTPError, ValueError) as exc: |
| 92 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for instrument enumeration") from exc |
| 93 | |
| 94 | async def sector_benchmark_contexts(self, instrument_ids: set[UUID], *, correlation_id=None, identity_headers=None): |
| 95 | """Read-only canonical dependencies, prepared before pure Stage-B computation.""" |
| 96 | if not instrument_ids: |
| 97 | return {} |
| 98 | from app.sector_benchmarks import build_sector_contexts |
| 99 | classifications = await self.india_nifty500_universe(correlation_id=correlation_id, identity_headers=identity_headers) |
| 100 | response = await self._client.get(f"{self.settings.portfolio_service_base_url}/api/v1/instruments/benchmarks", |
| 101 | headers={k:v for k,v in (identity_headers or {}).items() if v}) |
| 102 | response.raise_for_status() |
| 103 | registered = response.json() |
| 104 | if not isinstance(registered, list): |
| 105 | raise PortfolioServiceUnavailableError('BENCHMARK_IDENTITY_UNAVAILABLE') |
| 106 | return build_sector_contexts(classifications, registered, instrument_ids) |
| 107 | |
| 108 | async def india_nifty500_universe(self, *, correlation_id: str | None = None, identity_headers: dict[str, str | None] | None = None) -> list[dict]: |
| 109 | """Read portfolio-service owned NSE/Nifty universe; never uses portfolios.""" |
| 110 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 111 | if correlation_id: |
| 112 | headers["X-Correlation-Id"] = correlation_id |
| 113 | values: list[dict] = [] |
| 114 | page = 0 |
| 115 | page_size = 500 |
| 116 | try: |
| 117 | while True: |
| 118 | response = await self._client.get( |
| 119 | f"{self.settings.portfolio_service_base_url}/api/v1/market-universe/india/nifty500", |
| 120 | params={"page": page, "size": page_size}, |
| 121 | headers=headers or None, |
| 122 | ) |
| 123 | response.raise_for_status() |
| 124 | payload = response.json() |
| 125 | if not isinstance(payload, dict) or not isinstance(payload.get("instruments", []), list): |
| 126 | raise ValueError("Invalid India market-universe response") |
| 127 | batch = payload.get("instruments", []) |
| 128 | values.extend(batch) |
| 129 | try: |
| 130 | total = int(payload.get("totalElements", len(values))) |
| 131 | except (TypeError, ValueError) as exc: |
| 132 | raise ValueError("Invalid India market-universe pagination") from exc |
| 133 | if not batch: |
| 134 | if len(values) < total: |
| 135 | raise ValueError("India market-universe pagination ended before totalElements") |
| 136 | break |
| 137 | if len(values) >= total: |
| 138 | break |
| 139 | page += 1 |
| 140 | return values |
| 141 | except (httpx.HTTPError, ValueError, TypeError) as exc: |
| 142 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for India market universe") from exc |
| 143 | |
| 144 | async def refresh_india_nifty500_reference( |
| 145 | self, |
| 146 | *, |
| 147 | correlation_id: str | None = None, |
| 148 | identity_headers: dict[str, str | None] | None = None, |
| 149 | ) -> dict: |
| 150 | """Start the portfolio-owned official reference refresh. |
| 151 | |
| 152 | Callers must supply a server-owned operational identity. Browser role |
| 153 | headers are never promoted inside this client. |
| 154 | """ |
| 155 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 156 | if correlation_id: |
| 157 | headers["X-Correlation-Id"] = correlation_id |
| 158 | try: |
| 159 | response = await self._client.post( |
| 160 | f"{self.settings.portfolio_service_base_url}/api/v1/market-universe/india/nifty500/refresh", |
| 161 | headers=headers or None, |
| 162 | timeout=httpx.Timeout( |
| 163 | self.settings.market_data_nifty_refresh_timeout_seconds, |
| 164 | connect=self.settings.research_connect_timeout_seconds, |
| 165 | ), |
| 166 | ) |
| 167 | response.raise_for_status() |
| 168 | payload = response.json() |
| 169 | if not isinstance(payload, dict): |
| 170 | raise ValueError("Invalid India reference refresh response") |
| 171 | return payload |
| 172 | except (httpx.HTTPError, ValueError) as exc: |
| 173 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for India reference refresh") from exc |
| 174 | |
| 175 | async def global_instrument_metadata( |
| 176 | self, |
| 177 | instrument_id: UUID, |
| 178 | *, |
| 179 | correlation_id: str | None = None, |
| 180 | identity_headers: dict[str, str | None] | None = None, |
| 181 | ) -> dict: |
| 182 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 183 | if correlation_id: |
| 184 | headers["X-Correlation-Id"] = correlation_id |
| 185 | try: |
| 186 | response = await self._client.get( |
| 187 | f"{self.settings.portfolio_service_base_url}/api/v1/instruments/{instrument_id}", |
| 188 | headers=headers or None, |
| 189 | ) |
| 190 | if response.status_code == 404: |
| 191 | raise GlobalInstrumentNotFoundError(str(instrument_id)) |
| 192 | response.raise_for_status() |
| 193 | payload = response.json() |
| 194 | if not isinstance(payload, dict): |
| 195 | raise ValueError("Invalid global instrument response") |
| 196 | return payload |
| 197 | except GlobalInstrumentNotFoundError: |
| 198 | raise |
| 199 | except (httpx.HTTPError, ValueError) as exc: |
| 200 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument lookup") from exc |
| 201 | |
| 202 | async def list_watchlists( |
| 203 | self, |
| 204 | *, |
| 205 | correlation_id: str | None = None, |
| 206 | identity_headers: dict[str, str | None] | None = None, |
| 207 | ) -> list[dict]: |
| 208 | payload = await self._watchlist_request( |
| 209 | "GET", "/api/v1/watchlists", correlation_id=correlation_id, |
| 210 | identity_headers=identity_headers, |
| 211 | ) |
| 212 | if not isinstance(payload, list) or not all(isinstance(value, dict) for value in payload): |
| 213 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid watchlist response") |
| 214 | return payload |
| 215 | |
| 216 | async def ensure_default_watchlist( |
| 217 | self, |
| 218 | region: str, |
| 219 | *, |
| 220 | correlation_id: str | None = None, |
| 221 | identity_headers: dict[str, str | None] | None = None, |
| 222 | ) -> dict: |
| 223 | payload = await self._watchlist_request( |
| 224 | "POST", "/api/v1/watchlists/default/ensure", json={"region": region}, |
| 225 | correlation_id=correlation_id, identity_headers=identity_headers, |
| 226 | ) |
| 227 | if not isinstance(payload, dict): |
| 228 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid watchlist response") |
| 229 | return payload |
| 230 | |
| 231 | async def watchlist( |
| 232 | self, |
| 233 | watchlist_id: UUID, |
| 234 | *, |
| 235 | correlation_id: str | None = None, |
| 236 | identity_headers: dict[str, str | None] | None = None, |
| 237 | ) -> dict: |
| 238 | payload = await self._watchlist_request( |
| 239 | "GET", f"/api/v1/watchlists/{watchlist_id}", correlation_id=correlation_id, |
| 240 | identity_headers=identity_headers, |
| 241 | ) |
| 242 | if not isinstance(payload, dict) or not isinstance(payload.get("instruments"), list): |
| 243 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid watchlist detail response") |
| 244 | return payload |
| 245 | |
| 246 | async def add_watchlist_instrument( |
| 247 | self, |
| 248 | watchlist_id: UUID, |
| 249 | payload: dict, |
| 250 | *, |
| 251 | correlation_id: str | None = None, |
| 252 | identity_headers: dict[str, str | None] | None = None, |
| 253 | ) -> dict: |
| 254 | value = await self._watchlist_request( |
| 255 | "POST", f"/api/v1/watchlists/{watchlist_id}/instruments", json=payload, |
| 256 | correlation_id=correlation_id, identity_headers=identity_headers, |
| 257 | ) |
| 258 | if not isinstance(value, dict): |
| 259 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid watchlist membership response") |
| 260 | return value |
| 261 | |
| 262 | async def remove_watchlist_instrument( |
| 263 | self, |
| 264 | watchlist_id: UUID, |
| 265 | instrument_id: UUID, |
| 266 | *, |
| 267 | correlation_id: str | None = None, |
| 268 | identity_headers: dict[str, str | None] | None = None, |
| 269 | ) -> None: |
| 270 | await self._watchlist_request( |
| 271 | "DELETE", f"/api/v1/watchlists/{watchlist_id}/instruments/{instrument_id}", |
| 272 | correlation_id=correlation_id, identity_headers=identity_headers, |
| 273 | expect_json=False, |
| 274 | ) |
| 275 | |
| 276 | async def _watchlist_request( |
| 277 | self, |
| 278 | method: str, |
| 279 | path: str, |
| 280 | *, |
| 281 | json: dict | None = None, |
| 282 | correlation_id: str | None = None, |
| 283 | identity_headers: dict[str, str | None] | None = None, |
| 284 | expect_json: bool = True, |
| 285 | ): |
| 286 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 287 | if correlation_id: |
| 288 | headers["X-Correlation-Id"] = correlation_id |
| 289 | try: |
| 290 | response = await self._client.request( |
| 291 | method, |
| 292 | f"{self.settings.portfolio_service_base_url}{path}", |
| 293 | headers=headers or None, |
| 294 | json=json, |
| 295 | ) |
| 296 | except httpx.HTTPError as exc: |
| 297 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for watchlists") from exc |
| 298 | if response.status_code == 404: |
| 299 | raise WatchlistNotFoundError(path) |
| 300 | if response.status_code == 409: |
| 301 | try: |
| 302 | code = response.json().get("code") |
| 303 | except (ValueError, AttributeError): |
| 304 | code = None |
| 305 | if code == "WATCHLIST_REGION_MISMATCH": |
| 306 | raise WatchlistRegionMismatchError(code) |
| 307 | try: |
| 308 | response.raise_for_status() |
| 309 | return response.json() if expect_json else None |
| 310 | except (httpx.HTTPError, ValueError) as exc: |
| 311 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for watchlists") from exc |
| 312 | |
| 313 | async def read_global_company_state( |
| 314 | self, |
| 315 | instrument_id: UUID, |
| 316 | *, |
| 317 | metadata: dict | None = None, |
| 318 | correlation_id: str | None = None, |
| 319 | identity_headers: dict[str, str | None] | None = None, |
| 320 | ) -> PortfolioResearchCompany: |
| 321 | """Project one global instrument through the existing durable research read model. |
| 322 | |
| 323 | This is the non-portfolio counterpart of a portfolio company row. It |
| 324 | reads canonical instrument metadata and already-persisted research; it |
| 325 | neither creates a holding nor performs provider work. |
| 326 | """ |
| 327 | payload = metadata or await self.global_instrument_metadata( |
| 328 | instrument_id, |
| 329 | correlation_id=correlation_id, |
| 330 | identity_headers=identity_headers, |
| 331 | ) |
| 332 | instrument = _global_master_instrument(payload, instrument_id) |
| 333 | return await self._read_company_state(instrument) |
| 334 | |
| 335 | async def reconcile_global_instrument_metadata( |
| 336 | self, |
| 337 | instrument_id: UUID, |
| 338 | *, |
| 339 | correlation_id: str | None = None, |
| 340 | identity_headers: dict[str, str | None] | None = None, |
| 341 | ) -> dict: |
| 342 | """Ask portfolio-service's identity owner to validate provider mappings. |
| 343 | |
| 344 | This intentionally returns metadata only. Unlike the interactive |
| 345 | company refresh path, it does not register a process-local research |
| 346 | profile and does not fetch fundamentals. |
| 347 | """ |
| 348 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 349 | if correlation_id: |
| 350 | headers["X-Correlation-Id"] = correlation_id |
| 351 | try: |
| 352 | response = await self._client.post( |
| 353 | f"{self.settings.portfolio_service_base_url}/api/v1/instruments/{instrument_id}/reconcile", |
| 354 | headers=headers or None, |
| 355 | ) |
| 356 | response.raise_for_status() |
| 357 | payload = response.json() |
| 358 | if not isinstance(payload, dict): |
| 359 | raise ValueError("Invalid global instrument reconciliation response") |
| 360 | return payload |
| 361 | except (httpx.HTTPError, ValueError) as exc: |
| 362 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument reconciliation") from exc |
| 363 | |
| 364 | async def restore_global_profile( |
| 365 | self, |
| 366 | global_instrument_id: UUID, |
| 367 | *, |
| 368 | correlation_id: str | None = None, |
| 369 | identity_headers: dict[str, str | None] | None = None, |
| 370 | ) -> bool: |
| 371 | """Restore a process-local profile from the global instrument master API. |
| 372 | |
| 373 | Direct company reads are keyed by the global master ID, never by a local |
| 374 | portfolio instrument ID. The returned shape is normalized to the same |
| 375 | instrument representation used by portfolio-position orchestration so |
| 376 | provider mapping selection remains centralized here. |
| 377 | """ |
| 378 | try: |
| 379 | self.repository.profile(global_instrument_id) |
| 380 | return True |
| 381 | except StopIteration: |
| 382 | pass |
| 383 | try: |
| 384 | self.repository.etf_profile(global_instrument_id) |
| 385 | return True |
| 386 | except StopIteration: |
| 387 | pass |
| 388 | |
| 389 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 390 | if correlation_id: |
| 391 | headers["X-Correlation-Id"] = correlation_id |
| 392 | try: |
| 393 | response = await self._client.get( |
| 394 | f"{self.settings.portfolio_service_base_url}/api/v1/instruments/{global_instrument_id}", |
| 395 | headers=headers or None, |
| 396 | ) |
| 397 | except httpx.HTTPError as exc: |
| 398 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument lookup") from exc |
| 399 | if response.status_code == 404: |
| 400 | raise GlobalInstrumentNotFoundError(str(global_instrument_id)) |
| 401 | try: |
| 402 | response.raise_for_status() |
| 403 | except httpx.HTTPError as exc: |
| 404 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument lookup") from exc |
| 405 | payload = response.json() |
| 406 | if not isinstance(payload, dict): |
| 407 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid global instrument response") |
| 408 | instrument = _global_master_instrument(payload, global_instrument_id) |
| 409 | asset_type = _instrument_asset_type(instrument) |
| 410 | if asset_type == "ETF": |
| 411 | return self._resolve_etf_profile(instrument, register_missing=True) is not None |
| 412 | return self._resolve_profile(instrument, register_missing=True) is not None |
| 413 | |
| 414 | def register_global_profile_metadata( |
| 415 | self, |
| 416 | global_instrument_id: UUID, |
| 417 | payload: dict, |
| 418 | ) -> bool: |
| 419 | """Hydrate the process-local public profile from an already-read identity row.""" |
| 420 | instrument = _global_master_instrument(payload, global_instrument_id) |
| 421 | if _instrument_asset_type(instrument) == "ETF": |
| 422 | return self._resolve_etf_profile(instrument, register_missing=True) is not None |
| 423 | # A canonical global identity must keep ownership of its UUID. The |
| 424 | # general portfolio resolver may legitimately reuse an older profile |
| 425 | # by ISIN/ticker (including a demo seed), which would leave a public |
| 426 | # watchlist/non-held instrument registered under the wrong UUID. |
| 427 | # Prefer an exact profile and otherwise create the canonical profile |
| 428 | # directly from the already-validated instrument-master response. |
| 429 | profile = next( |
| 430 | ( |
| 431 | value |
| 432 | for value in self.repository.list_profiles() |
| 433 | if value.instrument_id == global_instrument_id |
| 434 | ), |
| 435 | None, |
| 436 | ) |
| 437 | if profile is None: |
| 438 | profile = self._register_equity_profile_from_instrument( |
| 439 | instrument, |
| 440 | str(instrument.get("provider") or "").upper(), |
| 441 | str(instrument.get("providerInstrumentId") or "").upper(), |
| 442 | str(instrument.get("isin") or "").upper(), |
| 443 | str(_instrument_ticker(instrument) or "").upper(), |
| 444 | _instrument_exchange(instrument), |
| 445 | _normalize_exchange( |
| 446 | instrument.get("canonicalMic") or instrument.get("mic") |
| 447 | ), |
| 448 | ) |
| 449 | if profile is None: |
| 450 | return False |
| 451 | _refresh_profile_from_global_instrument(profile, instrument) |
| 452 | return True |
| 453 | |
| 454 | async def reconcile_global_profile( |
| 455 | self, |
| 456 | global_instrument_id: UUID, |
| 457 | *, |
| 458 | correlation_id: str | None = None, |
| 459 | identity_headers: dict[str, str | None] | None = None, |
| 460 | ) -> bool: |
| 461 | """Refresh one process-local profile from portfolio-service's mapping owner.""" |
| 462 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 463 | if correlation_id: |
| 464 | headers["X-Correlation-Id"] = correlation_id |
| 465 | url = f"{self.settings.portfolio_service_base_url}/api/v1/instruments/{global_instrument_id}/reconcile" |
| 466 | try: |
| 467 | response = await self._client.post(url, headers=headers or None) |
| 468 | except httpx.HTTPError as exc: |
| 469 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument reconciliation") from exc |
| 470 | if response.status_code == 404: |
| 471 | raise GlobalInstrumentNotFoundError(str(global_instrument_id)) |
| 472 | try: |
| 473 | response.raise_for_status() |
| 474 | except httpx.HTTPError as exc: |
| 475 | raise PortfolioServiceUnavailableError("Portfolio service unavailable for global instrument reconciliation") from exc |
| 476 | payload = response.json() |
| 477 | if not isinstance(payload, dict): |
| 478 | raise PortfolioServiceUnavailableError("Portfolio service returned an invalid global instrument response") |
| 479 | instrument = _global_master_instrument(payload, global_instrument_id) |
| 480 | asset_type = _instrument_asset_type(instrument) |
| 481 | if asset_type == "ETF": |
| 482 | return self._resolve_etf_profile(instrument, register_missing=True) is not None |
| 483 | profile = self._resolve_profile(instrument, register_missing=True) |
| 484 | if profile is None: |
| 485 | return False |
| 486 | _refresh_profile_from_global_instrument(profile, instrument) |
| 487 | return True |
| 488 | |
| 489 | async def refresh_instrument( |
| 490 | self, |
| 491 | instrument_id: UUID, |
| 492 | *, |
| 493 | correlation_id: str | None = None, |
| 494 | allow_demo: bool = True, |
| 495 | pre_resolved_categories: set[str] | None = None, |
| 496 | instrument: dict | None = None, |
| 497 | structured_outcome: StructuredReconciliationOutcome | None = None, |
| 498 | structured_records: list[StructuredMarketSnapshotRecord] | None = None, |
| 499 | market_data=None, |
| 500 | ) -> ResearchSummary | EtfResearchProfile: |
| 501 | """Run the shared explicit refresh and durable structured reconciliation.""" |
| 502 | source_instrument = instrument or self._instrument_for_registered_profile(instrument_id) |
| 503 | asset_type = _instrument_asset_type(source_instrument) |
| 504 | outcome = structured_outcome or await self._reconcile_structured_market( |
| 505 | instrument_id, |
| 506 | source_instrument, |
| 507 | records=structured_records, |
| 508 | market_data=market_data, |
| 509 | ) |
| 510 | if asset_type == "ETF": |
| 511 | return await self.repository.refresh_etf(instrument_id, correlation_id=correlation_id) |
| 512 | categories = set(pre_resolved_categories or set()) |
| 513 | categories.update(_structured_categories(outcome.snapshot)) |
| 514 | return await self.repository.refresh( |
| 515 | instrument_id, |
| 516 | correlation_id=correlation_id, |
| 517 | allow_demo=allow_demo, |
| 518 | pre_resolved_categories=categories, |
| 519 | ) |
| 520 | |
| 521 | def _instrument_for_registered_profile(self, instrument_id: UUID) -> dict: |
| 522 | """Rebuild only persisted identity needed by the provider; never invent a symbol.""" |
| 523 | try: |
| 524 | profile = self.repository.profile(instrument_id) |
| 525 | return { |
| 526 | "instrumentId": str(profile.instrument_id), "assetType": "EQUITY", |
| 527 | "companyName": profile.company_name, "canonicalName": profile.company_name, |
| 528 | "ticker": profile.ticker, "exchange": profile.exchange, "mic": profile.mic, |
| 529 | "isin": profile.isin, "country": profile.country, "currency": profile.currency, |
| 530 | "structuredProviderTicker": profile.provider_instrument_ids.get("YAHOO_FINANCE"), |
| 531 | "structuredProviderStatus": "VERIFIED" if profile.provider_instrument_ids.get("YAHOO_FINANCE") else None, |
| 532 | "nseSymbol": profile.provider_instrument_ids.get("NSE"), |
| 533 | } |
| 534 | except StopIteration: |
| 535 | profile = self.repository.etf_profile(instrument_id) |
| 536 | return { |
| 537 | "instrumentId": str(profile.instrument_id), "assetType": "ETF", |
| 538 | "companyName": profile.fund_name, "canonicalName": profile.fund_name, |
| 539 | "ticker": profile.ticker, "exchange": profile.exchange, "mic": profile.mic, |
| 540 | "isin": profile.isin, "currency": profile.currency, |
| 541 | "structuredProviderTicker": profile.provider_instrument_id if str(profile.provider or "").upper() == "YAHOO_FINANCE" else None, |
| 542 | "structuredProviderStatus": "VERIFIED" if str(profile.provider or "").upper() == "YAHOO_FINANCE" else None, |
| 543 | } |
| 544 | |
| 545 | async def _reconcile_structured_market( |
| 546 | self, |
| 547 | instrument_id: UUID, |
| 548 | instrument: dict, |
| 549 | *, |
| 550 | records: list[StructuredMarketSnapshotRecord] | None = None, |
| 551 | market_data=None, |
| 552 | requested_classes: set[str] | None = None, |
| 553 | force_requested: bool = False, |
| 554 | ) -> StructuredReconciliationOutcome: |
| 555 | """Reconcile durable structured state; summaries never call this path.""" |
| 556 | if not self.settings.structured_provider_enabled or _instrument_asset_type(instrument) not in {"EQUITY", "ETF"}: |
| 557 | return StructuredReconciliationOutcome(None, None, frozenset(), "UNKNOWN") |
| 558 | if records is None: |
| 559 | records = (await self.repository.structured_market_snapshots_for_instruments({instrument_id})).get(instrument_id, []) |
| 560 | record = _preferred_structured_record(records) |
| 561 | market = str(instrument.get("mic") or instrument.get("exchange") or "").upper() |
| 562 | if market_data is None: |
| 563 | market_data = await self.repository.market_session_data({market} if market else set()) |
| 564 | schedules, exceptions = market_data |
| 565 | now = datetime.now(timezone.utc) |
| 566 | status = market_session_status(market, schedules, exceptions, now) |
| 567 | due = _structured_due_classes(record, status, self.settings, now) |
| 568 | if requested_classes is not None: |
| 569 | requested = {str(value).strip().upper() for value in requested_classes} |
| 570 | unknown = requested - {"PRICE", "VALUATION", "FUNDAMENTALS", "ANALYST"} |
| 571 | if unknown: |
| 572 | raise ValueError(f"UNKNOWN_STRUCTURED_DATA_CLASS:{','.join(sorted(unknown))}") |
| 573 | due = requested if force_requested else due & requested |
| 574 | if not due: |
| 575 | return StructuredReconciliationOutcome(record.snapshot if record else None, None, frozenset(), status) |
| 576 | try: |
| 577 | snapshot = await self.structured_provider.collect(instrument) |
| 578 | await self._persist_structured_snapshot(instrument_id, snapshot) |
| 579 | await self.repository.persist_yahoo_statement_facts_async(instrument_id, snapshot) |
| 580 | logger.info("structured_provider_complete canonical_instrument=%s dueClasses=%s marketStatus=%s", instrument_id, sorted(due), status) |
| 581 | return StructuredReconciliationOutcome(snapshot, None, frozenset(due), status) |
| 582 | except Exception as exc: |
| 583 | error = str(exc) if isinstance(exc, StructuredProviderError) else type(exc).__name__ |
| 584 | await self.repository.record_structured_market_failure_async( |
| 585 | instrument_id, getattr(self.structured_provider, "provider_name", "YAHOO_FINANCE"), error, error |
| 586 | ) |
| 587 | logger.warning("structured_provider_failed canonical_instrument=%s dueClasses=%s reason=%s", instrument_id, sorted(due), error) |
| 588 | return StructuredReconciliationOutcome(record.snapshot if record else None, error, frozenset(due), status) |
| 589 | |
| 590 | async def ensure_structured_market( |
| 591 | self, |
| 592 | instrument_id: UUID, |
| 593 | requested_classes: set[str], |
| 594 | ) -> StructuredReconciliationOutcome: |
| 595 | """Run one explicitly selected structured-market capability. |
| 596 | |
| 597 | The provider keeps ownership of collection and persistence. The |
| 598 | readiness planner owns only the requested fact classes. |
| 599 | """ |
| 600 | return await self._reconcile_structured_market( |
| 601 | instrument_id, |
| 602 | self._instrument_for_registered_profile(instrument_id), |
| 603 | requested_classes=requested_classes, |
| 604 | force_requested=True, |
| 605 | ) |
| 606 | |
| 607 | async def refresh_international_fundamentals( |
| 608 | self, |
| 609 | profile: CompanyResearchProfile, |
| 610 | *, |
| 611 | correlation_id: str | None = None, |
| 612 | identity_headers: dict[str, str | None] | None = None, |
| 613 | ) -> InternationalFundamentalsResult | None: |
| 614 | """Ingest a non-India profile through the shared financial-fact store. |
| 615 | |
| 616 | It is deliberately profile-based, rather than portfolio-based, so the |
| 617 | same global instrument can later be refreshed by a market screener. |
| 618 | """ |
| 619 | provider = international_provider_for(profile, self.settings, client=self._client) |
| 620 | if provider is None: |
| 621 | return None |
| 622 | result = await provider.collect(profile) |
| 623 | if result.facts: |
| 624 | await self.repository.persist_international_financial_facts_async(result.facts) |
| 625 | for provider, provider_id in result.verified_provider_ids.items(): |
| 626 | await self._persist_verified_provider_mapping( |
| 627 | profile, |
| 628 | provider, |
| 629 | provider_id, |
| 630 | correlation_id=correlation_id, |
| 631 | identity_headers=identity_headers, |
| 632 | ) |
| 633 | return result |
| 634 | |
| 635 | async def _persist_verified_provider_mapping( |
| 636 | self, |
| 637 | profile: CompanyResearchProfile, |
| 638 | provider: str, |
| 639 | provider_id: str, |
| 640 | *, |
| 641 | correlation_id: str | None = None, |
| 642 | identity_headers: dict[str, str | None] | None = None, |
| 643 | ) -> None: |
| 644 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 645 | if correlation_id: |
| 646 | headers["X-Correlation-Id"] = correlation_id |
| 647 | response = await self._client.put( |
| 648 | f"{self.settings.portfolio_service_base_url}/api/v1/instruments/{profile.instrument_id}/provider-mappings/verified", |
| 649 | headers=headers or None, |
| 650 | json={"provider": provider, "providerSymbol": profile.ticker, "providerInstrumentId": provider_id, |
| 651 | "exchange": profile.exchange, "currency": profile.currency, |
| 652 | "resolutionSource": "RESEARCH_ENGINE_VERIFIED_FUNDAMENTALS", "confidence": 0.90}, |
| 653 | ) |
| 654 | if response.status_code == 404: |
| 655 | raise GlobalInstrumentNotFoundError(str(profile.instrument_id)) |
| 656 | response.raise_for_status() |
| 657 | |
| 658 | async def refresh_portfolio( |
| 659 | self, |
| 660 | portfolio_id: UUID, |
| 661 | correlation_id: str | None = None, |
| 662 | identity_headers: dict[str, str | None] | None = None, |
| 663 | instruments: list[dict] | None = None, |
| 664 | progress_callback=None, |
| 665 | ) -> PortfolioResearchSummary: |
| 666 | instruments = instruments if instruments is not None else await self.prepare_portfolio_refresh(portfolio_id, correlation_id, identity_headers) |
| 667 | logger.info("portfolio_research_refresh_start portfolioId=%s instrumentCount=%s", portfolio_id, len(instruments)) |
| 668 | max_concurrency = self.settings.portfolio_refresh_instrument_concurrency |
| 669 | logger.info( |
| 670 | "portfolio_refresh_concurrency portfolioId=%s instrumentCount=%s maxConcurrency=%s", |
| 671 | portfolio_id, |
| 672 | len(instruments), |
| 673 | max_concurrency, |
| 674 | ) |
| 675 | semaphore = asyncio.Semaphore(max_concurrency) |
| 676 | refreshed_global_instruments: set[UUID] = set() |
| 677 | instrument_ids = {_safe_uuid(instrument.get("instrumentId")) for instrument in instruments} |
| 678 | instrument_ids.discard(None) |
| 679 | structured_records = await self.repository.structured_market_snapshots_for_instruments(instrument_ids) |
| 680 | markets = { |
| 681 | str(instrument.get("mic") or instrument.get("canonicalMic") or instrument.get("exchange") or "").upper() |
| 682 | for instrument in instruments |
| 683 | } |
| 684 | market_data = await self.repository.market_session_data({market for market in markets if market}) |
| 685 | |
| 686 | async def worker(instrument: dict) -> PortfolioResearchSummary: |
| 687 | async with semaphore: |
| 688 | global_instrument_id = instrument.get("instrumentId") |
| 689 | started = time.monotonic() |
| 690 | logger.info("portfolio_refresh_instrument_start globalInstrumentId=%s", global_instrument_id) |
| 691 | try: |
| 692 | local = await self._refresh_portfolio_instrument( |
| 693 | portfolio_id, |
| 694 | instrument, |
| 695 | correlation_id, |
| 696 | refreshed_global_instruments, |
| 697 | structured_records, |
| 698 | market_data, |
| 699 | ) |
| 700 | outcome = local.companies[0].status if local.companies else "NO_COMPANY" |
| 701 | except asyncio.CancelledError: |
| 702 | raise |
| 703 | except Exception as exc: |
| 704 | logger.exception( |
| 705 | "portfolio_refresh_instrument_complete globalInstrumentId=%s correlationId=%s outcome=FAILED reason=%s", |
| 706 | global_instrument_id, |
| 707 | correlation_id or "NONE", |
| 708 | type(exc).__name__, |
| 709 | ) |
| 710 | local = PortfolioResearchSummary(portfolio_id=portfolio_id, companies_requested=1, companies_degraded=1) |
| 711 | local.companies.append(PortfolioResearchCompany( |
| 712 | instrument_id=_safe_uuid(instrument.get("instrumentId")), |
| 713 | company_name=_instrument_name(instrument), |
| 714 | ticker=_instrument_ticker(instrument), |
| 715 | exchange=_instrument_exchange(instrument), |
| 716 | isin=instrument.get("isin"), |
| 717 | provider=instrument.get("provider"), |
| 718 | provider_instrument_id=instrument.get("providerInstrumentId"), |
| 719 | asset_type=_instrument_asset_type(instrument) or None, |
| 720 | status="SEARCH_PROVIDER_UNAVAILABLE", |
| 721 | safe_error_code=type(exc).__name__, |
| 722 | safe_error_message=str(exc)[:500], |
| 723 | )) |
| 724 | outcome = "FAILED" |
| 725 | logger.info( |
| 726 | "portfolio_refresh_instrument_complete globalInstrumentId=%s durationMs=%s outcome=%s", |
| 727 | global_instrument_id, |
| 728 | round((time.monotonic() - started) * 1000), |
| 729 | outcome, |
| 730 | ) |
| 731 | if progress_callback is not None: |
| 732 | await progress_callback(local) |
| 733 | return local |
| 734 | |
| 735 | locals_in_order = await asyncio.gather(*(worker(instrument) for instrument in instruments)) |
| 736 | result = PortfolioResearchSummary(portfolio_id=portfolio_id, companies_requested=len(instruments)) |
| 737 | for local in locals_in_order: |
| 738 | result.companies_resolved += local.companies_resolved |
| 739 | result.companies_failed += local.companies_failed |
| 740 | result.companies_degraded += local.companies_degraded |
| 741 | result.companies_succeeded += local.companies_succeeded |
| 742 | result.documents_created += local.documents_created |
| 743 | result.events_created += local.events_created |
| 744 | result.companies.extend(local.companies) |
| 745 | return await self._finalize_async(result) |
| 746 | |
| 747 | async def prepare_portfolio_refresh(self, portfolio_id: UUID, correlation_id: str | None = None, |
| 748 | identity_headers: dict[str, str | None] | None = None) -> list[dict]: |
| 749 | positions = await self._load_positions(portfolio_id, correlation_id, identity_headers) |
| 750 | return self._dedupe_instruments(positions) |
| 751 | |
| 752 | async def _refresh_portfolio_instrument( |
| 753 | self, |
| 754 | portfolio_id: UUID, |
| 755 | instrument: dict, |
| 756 | correlation_id: str | None, |
| 757 | refreshed_global_instruments: set[UUID], |
| 758 | structured_records: dict[UUID, list[StructuredMarketSnapshotRecord]], |
| 759 | market_data, |
| 760 | ) -> PortfolioResearchSummary: |
| 761 | """Run the existing sequential pipeline for exactly one instrument.""" |
| 762 | result = PortfolioResearchSummary(portfolio_id=portfolio_id, companies_requested=1) |
| 763 | asset_type = _instrument_asset_type(instrument) |
| 764 | logger.info("instrument_research_refresh_start globalInstrumentId=%s company=%s exchange=%s country=%s assetType=%s", instrument.get("instrumentId"), _instrument_name(instrument), _instrument_exchange(instrument), instrument.get("country"), asset_type or "UNKNOWN") |
| 765 | if asset_type == "ETF": |
| 766 | profile = self._resolve_etf_profile(instrument, register_missing=True) |
| 767 | result.companies_resolved += 1 |
| 768 | documents_before = len(self.repository.documents_for(profile.instrument_id)) if profile else 0 |
| 769 | if profile: |
| 770 | await self.refresh_instrument( |
| 771 | profile.instrument_id, |
| 772 | correlation_id=correlation_id, |
| 773 | instrument=instrument, |
| 774 | structured_records=structured_records.get(profile.instrument_id, []), |
| 775 | market_data=market_data, |
| 776 | ) |
| 777 | documents_after = len(self.repository.documents_for(profile.instrument_id)) if profile else documents_before |
| 778 | result.documents_created += max(documents_after - documents_before, 0) |
| 779 | result.companies_degraded += 1 |
| 780 | result.companies.append(_etf_company_from_profile(profile, instrument, self.repository.documents_for(profile.instrument_id) if profile else [], self.repository.last_refresh.get(profile.instrument_id) if profile else None, self.repository.last_live_error.get(profile.instrument_id) if profile else "ETF_RESEARCH_NOT_REFRESHED")) |
| 781 | return result |
| 782 | if asset_type in {"FUND", "BOND", "CASH", "CRYPTO"}: |
| 783 | result.companies_resolved += 1 |
| 784 | result.companies.append(_unsupported_asset_company(instrument, asset_type)) |
| 785 | return result |
| 786 | profile = self._resolve_profile(instrument, register_missing=True) |
| 787 | if profile is None: |
| 788 | result.companies_failed += 1 |
| 789 | result.companies.append(PortfolioResearchCompany(instrument_id=_safe_uuid(instrument.get("instrumentId")), company_name=_instrument_name(instrument), ticker=_instrument_ticker(instrument), exchange=_instrument_exchange(instrument), isin=instrument.get("isin"), provider=instrument.get("provider"), provider_instrument_id=instrument.get("providerInstrumentId"), asset_type=asset_type or None, status="COMPANY_NOT_RESOLVED", safe_error_code="COMPANY_NOT_RESOLVED", safe_error_message="Holding identity did not match a registered research company.")) |
| 790 | return result |
| 791 | result.companies_resolved += 1 |
| 792 | allow_demo = not _is_real_broker_instrument(instrument) |
| 793 | if profile.instrument_id in refreshed_global_instruments: |
| 794 | logger.info("instrument_research_refresh_reused portfolioId=%s globalInstrumentId=%s reason=DUPLICATE_RESOLVED_PROFILE", portfolio_id, profile.instrument_id) |
| 795 | summary = self.repository.summary(profile.instrument_id, allow_demo=allow_demo) |
| 796 | company = _company_from_summary(summary, status=_status_from_summary(summary, self.repository.last_live_error.get(profile.instrument_id)), source_instrument=instrument, safe_error_code=self.repository.last_live_error.get(profile.instrument_id)) |
| 797 | _attach_structured_market(company, self._cached_structured_snapshot(profile.instrument_id), None) |
| 798 | result.companies_succeeded += int(company.status == "RESOLVED_RESEARCH_AVAILABLE") |
| 799 | result.companies_degraded += int(company.status != "RESOLVED_RESEARCH_AVAILABLE") |
| 800 | result.companies.append(company) |
| 801 | return result |
| 802 | refreshed_global_instruments.add(profile.instrument_id) |
| 803 | structured_snapshot = None |
| 804 | structured_error = None |
| 805 | has_registered_sources = bool(registered_sources_for(profile.instrument_id)) |
| 806 | can_live_search = self.settings.research_live_enabled and self.settings.research_search_enabled |
| 807 | structured_outcome = await self._reconcile_structured_market( |
| 808 | profile.instrument_id, |
| 809 | instrument, |
| 810 | records=structured_records.get(profile.instrument_id, []), |
| 811 | market_data=market_data, |
| 812 | ) |
| 813 | structured_snapshot = structured_outcome.snapshot |
| 814 | structured_error = structured_outcome.error |
| 815 | if not has_registered_sources and not can_live_search and not _eligible_for_official_nse_research(profile): |
| 816 | summary = self.repository.summary(profile.instrument_id, allow_demo=allow_demo) |
| 817 | company = _company_from_summary(summary, status=_status_from_summary(summary, self.repository.last_live_error.get(profile.instrument_id)), source_instrument=instrument, safe_error_code=self.repository.last_live_error.get(profile.instrument_id)) |
| 818 | _attach_structured_market(company, structured_snapshot, structured_error) |
| 819 | result.companies_succeeded += int(company.status == "RESOLVED_RESEARCH_AVAILABLE") |
| 820 | result.companies_degraded += int(company.status != "RESOLVED_RESEARCH_AVAILABLE") |
| 821 | result.companies.append(company) |
| 822 | return result |
| 823 | documents_before = len(self.repository.documents_for(profile.instrument_id)) |
| 824 | events_before = len(self.repository.events_for(profile.instrument_id)) |
| 825 | try: |
| 826 | await self.refresh_instrument( |
| 827 | profile.instrument_id, |
| 828 | correlation_id=correlation_id, |
| 829 | allow_demo=True, |
| 830 | pre_resolved_categories=_structured_categories(structured_snapshot), |
| 831 | instrument=instrument, |
| 832 | structured_outcome=structured_outcome, |
| 833 | ) |
| 834 | result.documents_created += max(len(self.repository.documents_for(profile.instrument_id)) - documents_before, 0) |
| 835 | result.events_created += max(len(self.repository.events_for(profile.instrument_id)) - events_before, 0) |
| 836 | summary = self.repository.summary(profile.instrument_id, allow_demo=allow_demo) |
| 837 | company = _company_from_summary(summary, status=_status_from_summary(summary, self.repository.last_live_error.get(profile.instrument_id)), source_instrument=instrument, safe_error_code=self.repository.last_live_error.get(profile.instrument_id)) |
| 838 | except Exception as exc: |
| 839 | result.companies_degraded += 1 |
| 840 | summary = self.repository.summary(profile.instrument_id, allow_demo=allow_demo) |
| 841 | company = _company_from_summary(summary, status="RESOLVED_PARTIAL_DATA" if structured_snapshot else "SEARCH_PROVIDER_UNAVAILABLE", source_instrument=instrument, safe_error_code=type(exc).__name__, safe_error_message=str(exc)[:500]) |
| 842 | else: |
| 843 | result.companies_succeeded += int(company.status == "RESOLVED_RESEARCH_AVAILABLE") |
| 844 | result.companies_degraded += int(company.status != "RESOLVED_RESEARCH_AVAILABLE") |
| 845 | _attach_structured_market(company, structured_snapshot, structured_error) |
| 846 | result.companies.append(company) |
| 847 | return result |
| 848 | |
| 849 | async def read_portfolio_summary( |
| 850 | self, |
| 851 | portfolio_id: UUID, |
| 852 | correlation_id: str | None = None, |
| 853 | identity_headers: dict[str, str | None] | None = None, |
| 854 | ) -> PortfolioResearchSummary: |
| 855 | started = time.perf_counter() |
| 856 | positions = await self._load_positions(portfolio_id, correlation_id, identity_headers) |
| 857 | instruments = self._dedupe_instruments(positions) |
| 858 | result = PortfolioResearchSummary(portfolio_id=portfolio_id, companies_requested=len(instruments)) |
| 859 | for instrument in instruments: |
| 860 | company = await self._read_company_state(instrument) |
| 861 | result.companies.append(company) |
| 862 | if company.status == "COMPANY_NOT_RESOLVED": |
| 863 | result.companies_failed += 1 |
| 864 | else: |
| 865 | result.companies_resolved += 1 |
| 866 | if company.status == "RESOLVED_RESEARCH_AVAILABLE": |
| 867 | result.companies_succeeded += 1 |
| 868 | elif company.status != "RESEARCH_NOT_APPLICABLE": |
| 869 | result.companies_degraded += 1 |
| 870 | await self._attach_durable_structured_snapshots(result) |
| 871 | result = await self._finalize_async(result) |
| 872 | logger.info( |
| 873 | "portfolio_summary_stage stage=TOTAL portfolioId=%s positionCount=%s uniqueInstrumentCount=%s durationMs=%s", |
| 874 | portfolio_id, |
| 875 | len(positions), |
| 876 | len(instruments), |
| 877 | round((time.perf_counter() - started) * 1000), |
| 878 | ) |
| 879 | return result |
| 880 | |
| 881 | def _finalize(self, result: PortfolioResearchSummary) -> PortfolioResearchSummary: |
| 882 | facts_by_instrument = { |
| 883 | company.instrument_id: self.repository.financial_facts_for(company.instrument_id) |
| 884 | for company in result.companies if company.asset_type == "EQUITY" and company.instrument_id |
| 885 | } |
| 886 | return self._finalize_with_facts(result, facts_by_instrument) |
| 887 | |
| 888 | async def _finalize_async(self, result: PortfolioResearchSummary) -> PortfolioResearchSummary: |
| 889 | started = time.perf_counter() |
| 890 | instrument_ids = {company.instrument_id for company in result.companies if company.asset_type == "EQUITY" and company.instrument_id} |
| 891 | facts_by_instrument = await self.repository.financial_facts_for_instruments(instrument_ids) |
| 892 | threshold = Decimal(str(self.settings.research_ownership_change_threshold_percentage_points)) |
| 893 | for index, company in enumerate(result.companies): |
| 894 | if company.asset_type != "EQUITY" or not company.instrument_id: |
| 895 | continue |
| 896 | documents = list(self.repository.documents_for(company.instrument_id)) |
| 897 | events = list(self.repository.events_for(company.instrument_id)) |
| 898 | enrichment_started = time.perf_counter() |
| 899 | enriched = await asyncio.to_thread( |
| 900 | self._enrich_company_snapshot, company.model_copy(deep=True), documents, events, threshold, |
| 901 | list(facts_by_instrument.get(company.instrument_id, [])), |
| 902 | ) |
| 903 | logger.info( |
| 904 | "portfolio_summary_stage stage=ENRICH_COMPANY instrumentId=%s durationMs=%s documentCount=%s eventCount=%s financialFactCount=%s", |
| 905 | company.instrument_id, |
| 906 | round((time.perf_counter() - enrichment_started) * 1000), |
| 907 | len(documents), |
| 908 | len(events), |
| 909 | len(facts_by_instrument.get(company.instrument_id, [])), |
| 910 | ) |
| 911 | if enriched.structured_market: |
| 912 | _attach_structured_market(enriched, enriched.structured_market, None) |
| 913 | result.companies[index] = enriched |
| 914 | result = self._finalize_company_counts(result) |
| 915 | logger.info( |
| 916 | "portfolio_summary_stage stage=FINALIZE companyCount=%s durationMs=%s", |
| 917 | len(result.companies), |
| 918 | round((time.perf_counter() - started) * 1000), |
| 919 | ) |
| 920 | return result |
| 921 | |
| 922 | def _finalize_with_facts(self, result: PortfolioResearchSummary, facts_by_instrument) -> PortfolioResearchSummary: |
| 923 | threshold = Decimal(str(self.settings.research_ownership_change_threshold_percentage_points)) |
| 924 | for company in result.companies: |
| 925 | if company.asset_type == "EQUITY" and company.instrument_id: |
| 926 | enrich_company_research( |
| 927 | company, |
| 928 | self.repository.documents_for(company.instrument_id), |
| 929 | self.repository.events_for(company.instrument_id), |
| 930 | threshold, |
| 931 | facts_by_instrument.get(company.instrument_id, []), |
| 932 | ) |
| 933 | if company.structured_market: |
| 934 | _attach_structured_market(company, company.structured_market, None) |
| 935 | return self._finalize_company_counts(result) |
| 936 | |
| 937 | |
| 938 | def _finalize_company_counts(self, result: PortfolioResearchSummary) -> PortfolioResearchSummary: |
| 939 | result.total_companies = result.companies_requested |
| 940 | result.completed = sum(company.status == "RESOLVED_RESEARCH_AVAILABLE" for company in result.companies) |
| 941 | result.partial = sum(company.status == "RESOLVED_PARTIAL_DATA" for company in result.companies) |
| 942 | result.failed = sum(company.status in { |
| 943 | "COMPANY_NOT_RESOLVED", "SEARCH_PROVIDER_UNAVAILABLE", "SEARCH_RETURNED_ZERO_RESULTS", |
| 944 | "RESULTS_REJECTED", "DOCUMENT_FETCH_FAILED", "EXTRACTION_EMPTY", "RESOLVED_NO_SOURCES", |
| 945 | } for company in result.companies) |
| 946 | result.unsupported = sum(company.status in {"ETF_UNSUPPORTED", "RESEARCH_NOT_APPLICABLE"} for company in result.companies) |
| 947 | result.in_progress = sum(company.status == "IN_PROGRESS" for company in result.companies) |
| 948 | return result |
| 949 | |
| 950 | |
| 951 | @staticmethod |
| 952 | def _enrich_company_snapshot(company, documents, events, threshold, facts): |
| 953 | enrich_company_research(company, documents, events, threshold, facts) |
| 954 | return company |
| 955 | |
| 956 | async def _read_company_state(self, instrument: dict) -> PortfolioResearchCompany: |
| 957 | started = time.perf_counter() |
| 958 | summary_counts = {"documents": 0, "events": 0, "shareholding": 0} |
| 959 | instrument_id = instrument.get("instrumentId") |
| 960 | try: |
| 961 | return await self._read_company_state_impl(instrument, summary_counts) |
| 962 | finally: |
| 963 | logger.info( |
| 964 | "portfolio_summary_stage stage=READ_COMPANY instrumentId=%s durationMs=%s documentCount=%s eventCount=%s shareholdingCount=%s", |
| 965 | instrument_id, |
| 966 | round((time.perf_counter() - started) * 1000), |
| 967 | summary_counts["documents"], |
| 968 | summary_counts["events"], |
| 969 | summary_counts["shareholding"], |
| 970 | ) |
| 971 | |
| 972 | async def _read_company_state_impl( |
| 973 | self, instrument: dict, summary_counts: dict[str, int] |
| 974 | ) -> PortfolioResearchCompany: |
| 975 | asset_type = _instrument_asset_type(instrument) |
| 976 | if asset_type == "ETF": |
| 977 | profile = self._resolve_etf_profile(instrument, register_missing=False) |
| 978 | if profile is None: |
| 979 | profile = self._resolve_etf_profile(instrument, register_missing=True) |
| 980 | company = _etf_company_from_profile( |
| 981 | profile, |
| 982 | instrument, |
| 983 | self.repository.documents_for(profile.instrument_id) if profile else [], |
| 984 | self.repository.last_refresh.get(profile.instrument_id) if profile else None, |
| 985 | self.repository.last_live_error.get(profile.instrument_id) if profile else None, |
| 986 | ) |
| 987 | await self._attach_current_structured_snapshot(company, instrument) |
| 988 | return company |
| 989 | if asset_type in {"FUND", "BOND", "CASH", "CRYPTO"}: |
| 990 | return _unsupported_asset_company(instrument, asset_type) |
| 991 | # A portfolio position already carries a globally resolved instrument |
| 992 | # and verified provider mappings. Absence of prior research must not |
| 993 | # make that global instrument unavailable: register its process-local |
| 994 | # research profile so the UI can offer the first refresh. |
| 995 | profile = self._resolve_profile( |
| 996 | instrument, |
| 997 | register_missing=_safe_uuid(instrument.get("globalInstrumentId")) is not None, |
| 998 | ) |
| 999 | if profile is None: |
| 1000 | enriched = _enriched_equity_company(instrument) |
| 1001 | if enriched is not None: |
| 1002 | return enriched |
| 1003 | return PortfolioResearchCompany( |
| 1004 | instrument_id=_safe_uuid(instrument.get("instrumentId")), |
| 1005 | company_name=_instrument_name(instrument), |
| 1006 | ticker=_instrument_ticker(instrument), |
| 1007 | exchange=_instrument_exchange(instrument), |
| 1008 | isin=instrument.get("isin"), |
| 1009 | provider=instrument.get("provider"), |
| 1010 | provider_instrument_id=instrument.get("providerInstrumentId"), |
| 1011 | asset_type=asset_type or None, |
| 1012 | status="COMPANY_NOT_RESOLVED", |
| 1013 | safe_error_code="COMPANY_NOT_RESOLVED", |
| 1014 | safe_error_message="Holding identity did not match a registered research company.", |
| 1015 | ) |
| 1016 | if not instrument.get("provider"): |
| 1017 | _refresh_profile_from_global_instrument(profile, instrument) |
| 1018 | allow_demo = not _is_real_broker_instrument(instrument) |
| 1019 | summary = self.repository.summary(profile.instrument_id, allow_demo=allow_demo) |
| 1020 | summary_counts.update( |
| 1021 | documents=len(summary.documents), |
| 1022 | events=len(summary.recent_events), |
| 1023 | shareholding=len(summary.shareholding_snapshots), |
| 1024 | ) |
| 1025 | status = _status_from_summary(summary, self.repository.last_live_error.get(profile.instrument_id)) |
| 1026 | company = _company_from_summary( |
| 1027 | summary, |
| 1028 | status=status, |
| 1029 | source_instrument=instrument, |
| 1030 | safe_error_code=self.repository.last_live_error.get(profile.instrument_id), |
| 1031 | ) |
| 1032 | await self._attach_current_structured_snapshot(company, instrument) |
| 1033 | return company |
| 1034 | |
| 1035 | async def _attach_current_structured_snapshot( |
| 1036 | self, company: PortfolioResearchCompany, instrument: dict |
| 1037 | ) -> None: |
| 1038 | """Attach a cached or fresh provider snapshot when serving the drawer. |
| 1039 | |
| 1040 | The research summary must not depend on a previous refresh request having |
| 1041 | hit the same process. Yahoo collection has its own short quote cache and |
| 1042 | a durable verified mapping is supplied by portfolio-service, so this |
| 1043 | fallback reuses the mapping directly and never repeats discovery. |
| 1044 | """ |
| 1045 | if not self.settings.structured_provider_enabled: |
| 1046 | return |
| 1047 | snapshot = self._cached_structured_snapshot(company.instrument_id) if company.instrument_id else None |
| 1048 | _attach_structured_market(company, snapshot, None) |
| 1049 | |
| 1050 | def _cached_structured_snapshot(self, instrument_id: UUID) -> object | None: |
| 1051 | cached = self._structured_by_instrument.get(instrument_id) |
| 1052 | if cached is None: |
| 1053 | return None |
| 1054 | expires_at, snapshot = cached |
| 1055 | if expires_at > datetime.now(timezone.utc): |
| 1056 | return snapshot |
| 1057 | self._structured_by_instrument.pop(instrument_id, None) |
| 1058 | return None |
| 1059 | |
| 1060 | def _store_structured_snapshot(self, instrument_id: UUID, snapshot: object) -> None: |
| 1061 | self._structured_by_instrument[instrument_id] = ( |
| 1062 | datetime.now(timezone.utc) + timedelta(seconds=self.settings.structured_market_price_freshness_seconds), |
| 1063 | snapshot, |
| 1064 | ) |
| 1065 | |
| 1066 | async def _persist_structured_snapshot(self, instrument_id: UUID, snapshot) -> None: |
| 1067 | now = datetime.now(timezone.utc) |
| 1068 | facts = snapshot.facts |
| 1069 | record = StructuredMarketSnapshotRecord( |
| 1070 | instrument_id=instrument_id, provider=snapshot.resolution.provider, provider_instrument_id=snapshot.resolution.provider_ticker, |
| 1071 | exchange=snapshot.resolution.exchange, currency=snapshot.resolution.currency, quote_type=snapshot.resolution.quote_type, |
| 1072 | source_url=snapshot.source_url, source_name=snapshot.source_name, source_type=snapshot.source_type, |
| 1073 | source_identity=snapshot.resolution.provider_ticker, market_as_of=snapshot.market_as_of, retrieved_at=snapshot.retrieved_at, persisted_at=now, |
| 1074 | last_price_at=now if _positive_decimal_fact(facts.get("latestPrice")) is not None else None, |
| 1075 | last_valuation_at=now if any(key in facts for key in ("trailingPE", "forwardPE", "priceToBook")) else None, |
| 1076 | last_fundamentals_at=now if any(key in facts for key in ("trailingEPS", "roe", "roa", "roce")) else None, |
| 1077 | last_analyst_at=now if any(key.startswith("publicAnalyst") for key in facts) else None, |
| 1078 | last_success_at=now, last_provider_attempt_at=now, acquisition_status="SUCCESS", snapshot=snapshot, |
| 1079 | ) |
| 1080 | await self.repository.persist_structured_market_snapshot_async(record) |
| 1081 | self._store_structured_snapshot(instrument_id, snapshot) |
| 1082 | |
| 1083 | async def _attach_durable_structured_snapshots(self, result: PortfolioResearchSummary) -> None: |
| 1084 | ids = {company.instrument_id for company in result.companies if company.instrument_id} |
| 1085 | records = await self.repository.structured_market_snapshots_for_instruments(ids) |
| 1086 | markets = {str(company.primary_exchange or company.exchange or "").upper() for company in result.companies} |
| 1087 | schedules, exceptions = await self.repository.market_session_data(markets) |
| 1088 | now = datetime.now(timezone.utc) |
| 1089 | for company in result.companies: |
| 1090 | if not company.instrument_id: |
| 1091 | continue |
| 1092 | choices = records.get(company.instrument_id, []) |
| 1093 | record = next((item for item in choices if item.provider == "NSE_STRUCTURED"), None) or next((item for item in choices if item.provider == "YAHOO_FINANCE"), None) or (choices[0] if choices else None) |
| 1094 | _attach_durable_structured_market(company, record, schedules, exceptions, now, self.settings) |
| 1095 | |
| 1096 | async def _attach_global_durable_structured_snapshot( |
| 1097 | self, company: PortfolioResearchCompany |
| 1098 | ) -> None: |
| 1099 | """Attach the persisted structured-market snapshot for a single non-held company. |
| 1100 | |
| 1101 | The public/global research path never runs an interactive refresh, so the |
| 1102 | in-process cache attached by ``_attach_current_structured_snapshot`` is |
| 1103 | frequently empty. Reuse the durable snapshot (the same source the |
| 1104 | portfolio path reads in bulk) so the drawer projects provider identity, |
| 1105 | sector, industry and market data instead of falling back to N/A. |
| 1106 | """ |
| 1107 | if not company.instrument_id: |
| 1108 | return |
| 1109 | records = await self.repository.structured_market_snapshots_for_instruments( |
| 1110 | {company.instrument_id} |
| 1111 | ) |
| 1112 | markets = {str(company.primary_exchange or company.exchange or "").upper()} |
| 1113 | if "" in markets: |
| 1114 | markets.discard("") |
| 1115 | schedules, exceptions = await self.repository.market_session_data(markets) if markets else ({}, {}) |
| 1116 | now = datetime.now(timezone.utc) |
| 1117 | choices = records.get(company.instrument_id, []) |
| 1118 | record = _preferred_structured_record(choices) |
| 1119 | _attach_durable_structured_market(company, record, schedules, exceptions, now, self.settings) |
| 1120 | |
| 1121 | async def _finalize_global_company( |
| 1122 | self, company: PortfolioResearchCompany |
| 1123 | ) -> PortfolioResearchCompany: |
| 1124 | """Project durable financial/valuation facts for a single non-held company. |
| 1125 | |
| 1126 | Mirrors the per-company body of ``_finalize_async`` so the public research |
| 1127 | projection reads valuation state, quarterly and statement history and |
| 1128 | shareholding changes from the same durable facts as portfolio positions. |
| 1129 | """ |
| 1130 | if company.asset_type != "EQUITY" or not company.instrument_id: |
| 1131 | return company |
| 1132 | facts_by_instrument = await self.repository.financial_facts_for_instruments( |
| 1133 | {company.instrument_id} |
| 1134 | ) |
| 1135 | threshold = Decimal(str(self.settings.research_ownership_change_threshold_percentage_points)) |
| 1136 | documents = list(self.repository.documents_for(company.instrument_id)) |
| 1137 | events = list(self.repository.events_for(company.instrument_id)) |
| 1138 | enriched = await asyncio.to_thread( |
| 1139 | self._enrich_company_snapshot, company.model_copy(deep=True), documents, events, threshold, |
| 1140 | list(facts_by_instrument.get(company.instrument_id, [])), |
| 1141 | ) |
| 1142 | if enriched.structured_market: |
| 1143 | _attach_structured_market(enriched, enriched.structured_market, None) |
| 1144 | return enriched |
| 1145 | |
| 1146 | async def enrich_global_company_durables( |
| 1147 | self, company: PortfolioResearchCompany |
| 1148 | ) -> PortfolioResearchCompany: |
| 1149 | """Attach all durable facts to a public, non-held company row. |
| 1150 | |
| 1151 | This is the non-portfolio counterpart of ``_finalize_async``: it applies |
| 1152 | durable structured-market facts and durable financial/valuation evidence so |
| 1153 | the Stock Research drawer and research-intelligence sections project real |
| 1154 | data instead of N/A placeholders. No provider work is performed and no |
| 1155 | portfolio/holding identity is mutated. |
| 1156 | """ |
| 1157 | await self._attach_global_durable_structured_snapshot(company) |
| 1158 | return await self._finalize_global_company(company) |
| 1159 | |
| 1160 | async def search_instruments( |
| 1161 | self, |
| 1162 | query: str, |
| 1163 | region: str, |
| 1164 | *, |
| 1165 | limit: int = 20, |
| 1166 | correlation_id: str | None = None, |
| 1167 | identity_headers: dict[str, str | None] | None = None, |
| 1168 | ) -> list[dict]: |
| 1169 | """Search the global canonical instrument universe by name/symbol/ISIN. |
| 1170 | |
| 1171 | India uses the persisted canonical NSE-backed master; the |
| 1172 | USA/EUROPE universes use the verified portfolio-service active-equity |
| 1173 | master. The browser never calls NSE or a price provider directly. |
| 1174 | Results are ranked by canonical identity match and backed by the durable |
| 1175 | read model; financial identity (sector/industry) is projected from the |
| 1176 | durable structured-market snapshot when available. |
| 1177 | """ |
| 1178 | normalized_region = str(region).strip().upper() |
| 1179 | if normalized_region not in {"INDIA", "USA", "EUROPE"}: |
| 1180 | return [] |
| 1181 | if len(query.strip()) < 3: |
| 1182 | return [] |
| 1183 | limit = min(20, max(1, limit)) |
| 1184 | raw_universe = await self.active_global_equities( |
| 1185 | correlation_id=correlation_id, identity_headers=identity_headers |
| 1186 | ) |
| 1187 | candidates = [_normalize_search_item(item) for item in raw_universe] |
| 1188 | query_norm = query.strip().lower() |
| 1189 | matched = [ |
| 1190 | candidate |
| 1191 | for candidate in candidates |
| 1192 | if _search_item_matches(candidate, query_norm) |
| 1193 | and belongs_to_region(candidate, normalized_region) |
| 1194 | and candidate["assetType"] == "EQUITY" |
| 1195 | and _safe_uuid(candidate["globalInstrumentId"]) is not None |
| 1196 | ] |
| 1197 | ids = {UUID(item["globalInstrumentId"]) for item in matched if item["globalInstrumentId"]} |
| 1198 | records = await self.repository.structured_market_snapshots_for_instruments(ids) if ids else {} |
| 1199 | for candidate in matched: |
| 1200 | candidate["region"] = normalized_region |
| 1201 | global_id = candidate["globalInstrumentId"] |
| 1202 | try: |
| 1203 | global_uuid = UUID(global_id) |
| 1204 | except (ValueError, TypeError): |
| 1205 | global_uuid = None |
| 1206 | if global_uuid is not None: |
| 1207 | candidate["sector"] = _sector_from_records(global_uuid, records) or candidate.get("sector") |
| 1208 | candidate["industry"] = _industry_from_records(global_uuid, records) or candidate.get("industry") |
| 1209 | score = self.repository.persisted_canonical_read_model_score(global_uuid) if global_uuid else None |
| 1210 | candidate["score"] = score.overall_score if score is not None else None |
| 1211 | candidate["_rank"] = _search_rank(candidate, query_norm) |
| 1212 | matched.sort(key=lambda item: (item["_rank"], -(item["score"] or 0))) |
| 1213 | for candidate in matched: |
| 1214 | candidate.pop("_rank", None) |
| 1215 | return matched[:limit] |
| 1216 | |
| 1217 | async def structured_quote(self, instrument: dict) -> object: |
| 1218 | if _instrument_asset_type(instrument) not in {"EQUITY", "ETF"}: |
| 1219 | raise StructuredProviderError("RESOLUTION_UNSUPPORTED_ASSET_TYPE") |
| 1220 | snapshot = await self.structured_provider.collect(instrument) |
| 1221 | instrument_id = _safe_uuid(instrument.get("instrumentId")) |
| 1222 | if instrument_id: |
| 1223 | await self._persist_structured_snapshot(instrument_id, snapshot) |
| 1224 | return snapshot |
| 1225 | |
| 1226 | async def _load_positions( |
| 1227 | self, |
| 1228 | portfolio_id: UUID, |
| 1229 | correlation_id: str | None, |
| 1230 | identity_headers: dict[str, str | None] | None, |
| 1231 | ) -> list[dict]: |
| 1232 | started = time.perf_counter() |
| 1233 | headers = {key: value for key, value in (identity_headers or {}).items() if value} |
| 1234 | if correlation_id: |
| 1235 | headers["X-Correlation-Id"] = correlation_id |
| 1236 | response = await self._client.get( |
| 1237 | f"{self.settings.portfolio_service_base_url}/api/v1/portfolios/{portfolio_id}/positions", |
| 1238 | headers=headers or None, |
| 1239 | ) |
| 1240 | response.raise_for_status() |
| 1241 | payload = response.json() |
| 1242 | if not isinstance(payload, list): |
| 1243 | raise ValueError("PORTFOLIO_POSITIONS_INVALID_RESPONSE") |
| 1244 | positions = [item for item in payload if isinstance(item, dict)] |
| 1245 | logger.info( |
| 1246 | "portfolio_summary_stage stage=LOAD_POSITIONS portfolioId=%s durationMs=%s positionCount=%s", |
| 1247 | portfolio_id, |
| 1248 | round((time.perf_counter() - started) * 1000), |
| 1249 | len(positions), |
| 1250 | ) |
| 1251 | return positions |
| 1252 | |
| 1253 | def _dedupe_instruments(self, positions: list[dict]) -> list[dict]: |
| 1254 | instruments: list[dict] = [] |
| 1255 | seen: set[tuple[str, str, str, str, str, str]] = set() |
| 1256 | for position in positions: |
| 1257 | instrument = position.get("instrument") |
| 1258 | if not isinstance(instrument, dict): |
| 1259 | continue |
| 1260 | global_instrument_id = str(instrument.get("globalInstrumentId") or "").strip() |
| 1261 | key = ( |
| 1262 | f"GLOBAL:{global_instrument_id}", "", "", "", "", "" |
| 1263 | ) if global_instrument_id else ( |
| 1264 | str(instrument.get("instrumentId") or ""), |
| 1265 | str(instrument.get("provider") or "").upper(), |
| 1266 | str(instrument.get("providerInstrumentId") or "").upper(), |
| 1267 | str(instrument.get("isin") or "").upper(), |
| 1268 | str(instrument.get("ticker") or "").upper(), |
| 1269 | _normalize_exchange(instrument.get("exchange")), |
| 1270 | ) |
| 1271 | if key in seen: |
| 1272 | continue |
| 1273 | seen.add(key) |
| 1274 | research_instrument = dict(instrument) |
| 1275 | research_instrument["instrumentId"] = instrument.get("globalInstrumentId") or instrument.get("instrumentId") |
| 1276 | for mapping in instrument.get("providerMappings") or []: |
| 1277 | if not _trusted_provider_mapping(mapping): |
| 1278 | continue |
| 1279 | provider = str(mapping.get("provider") or "").upper() |
| 1280 | if provider == "YAHOO_FINANCE": |
| 1281 | research_instrument.update({ |
| 1282 | "structuredProviderTicker": mapping.get("providerSymbol"), |
| 1283 | "structuredProviderExchange": mapping.get("exchange"), |
| 1284 | "structuredProviderCurrency": mapping.get("currency"), |
| 1285 | "structuredProviderStatus": mapping.get("status"), |
| 1286 | }) |
| 1287 | elif provider == "NSE": |
| 1288 | research_instrument["nseSymbol"] = mapping.get("providerSymbol") |
| 1289 | elif provider == "BSE": |
| 1290 | research_instrument["bseSymbol"] = mapping.get("providerSymbol") or mapping.get("providerInstrumentId") |
| 1291 | research_instrument["_positionDataFreshness"] = position.get("dataFreshness") |
| 1292 | research_instrument["_researchCustomDisplayName"] = ( |
| 1293 | position.get("customDisplayName") or position.get("userDisplayName") |
| 1294 | ) |
| 1295 | research_instrument["_researchDisplayName"] = position.get("displayName") |
| 1296 | instruments.append(research_instrument) |
| 1297 | return instruments |
| 1298 | |
| 1299 | def _resolve_profile(self, instrument: dict, *, register_missing: bool) -> CompanyResearchProfile | None: |
| 1300 | instrument_id = _safe_uuid(instrument.get("instrumentId")) |
| 1301 | provider = str(instrument.get("provider") or "").upper() |
| 1302 | provider_instrument_id = str(instrument.get("providerInstrumentId") or "").upper() |
| 1303 | isin = str(instrument.get("isin") or "").upper() |
| 1304 | ticker = str(_instrument_ticker(instrument) or "").upper() |
| 1305 | exchange = _instrument_exchange(instrument) |
| 1306 | mic = _normalize_exchange(instrument.get("canonicalMic") or instrument.get("mic")) |
| 1307 | for profile in self.repository.list_profiles(): |
| 1308 | if provider and provider_instrument_id: |
| 1309 | known = {key.upper(): value.upper() for key, value in profile.provider_instrument_ids.items()} |
| 1310 | if known.get(provider) == provider_instrument_id: |
| 1311 | return _hydrate_verified_exchange_mappings(profile, instrument) |
| 1312 | if instrument_id and profile.instrument_id == instrument_id: |
| 1313 | return _hydrate_verified_exchange_mappings(profile, instrument) |
| 1314 | if isin and profile.isin and profile.isin.upper() == isin: |
| 1315 | return _hydrate_verified_exchange_mappings(profile, instrument) |
| 1316 | known_markets = {profile.exchange.upper(), profile.mic.upper()} |
| 1317 | if mic and mic != "UNKNOWN": |
| 1318 | known_markets.add(mic) |
| 1319 | if ticker == profile.ticker.upper() and exchange and exchange != "UNKNOWN" and exchange in known_markets: |
| 1320 | return _hydrate_verified_exchange_mappings(profile, instrument) |
| 1321 | if not register_missing: |
| 1322 | return None |
| 1323 | return self._register_equity_profile_from_instrument(instrument, provider, provider_instrument_id, isin, ticker, exchange, mic) |
| 1324 | |
| 1325 | def _resolve_etf_profile(self, instrument: dict, *, register_missing: bool) -> EtfResearchProfile | None: |
| 1326 | instrument_id = _safe_uuid(instrument.get("instrumentId")) |
| 1327 | provider = str(instrument.get("provider") or "").upper() |
| 1328 | provider_instrument_id = str(instrument.get("providerInstrumentId") or "").upper() |
| 1329 | isin = str(instrument.get("isin") or "").upper() |
| 1330 | ticker = str(_instrument_ticker(instrument) or "").upper() |
| 1331 | exchange = _instrument_exchange(instrument) |
| 1332 | for profile in self.repository.list_etf_profiles(): |
| 1333 | if instrument_id and profile.instrument_id == instrument_id: |
| 1334 | return profile |
| 1335 | if provider and provider_instrument_id and profile.provider and profile.provider_instrument_id: |
| 1336 | if profile.provider.upper() == provider and profile.provider_instrument_id.upper() == provider_instrument_id: |
| 1337 | return profile |
| 1338 | if isin and profile.isin and profile.isin.upper() == isin: |
| 1339 | return profile |
| 1340 | if not register_missing: |
| 1341 | return None |
| 1342 | if not ticker or ticker == "UNKNOWN" or not exchange or exchange == "UNKNOWN": |
| 1343 | return None |
| 1344 | fund_name = _instrument_name(instrument) |
| 1345 | if not fund_name or fund_name.upper() == "UNKNOWN": |
| 1346 | return None |
| 1347 | profile = EtfResearchProfile( |
| 1348 | instrument_id=instrument_id or uuid5(NAMESPACE_URL, "|".join([provider, provider_instrument_id, isin, ticker, exchange])), |
| 1349 | fund_id=uuid5(NAMESPACE_URL, f"etf|{isin}|{ticker}|{exchange}|{fund_name}"), |
| 1350 | fund_name=fund_name, |
| 1351 | ticker=ticker, |
| 1352 | exchange=exchange, |
| 1353 | mic=_normalize_exchange(instrument.get("canonicalMic") or instrument.get("mic")) or exchange, |
| 1354 | provider=provider or instrument.get("provider"), |
| 1355 | provider_instrument_id=provider_instrument_id or instrument.get("providerInstrumentId"), |
| 1356 | isin=isin or None, |
| 1357 | currency=str(instrument.get("tradingCurrency") or instrument.get("currency") or "").upper() or None, |
| 1358 | fund_provider=_infer_fund_provider_from_name(fund_name), |
| 1359 | underlying_index=_infer_underlying_index_from_name(fund_name), |
| 1360 | ) |
| 1361 | return self.repository.register_etf_profile(profile) |
| 1362 | |
| 1363 | def _register_equity_profile_from_instrument( |
| 1364 | self, |
| 1365 | instrument: dict, |
| 1366 | provider: str, |
| 1367 | provider_instrument_id: str, |
| 1368 | isin: str, |
| 1369 | ticker: str, |
| 1370 | exchange: str, |
| 1371 | mic: str, |
| 1372 | ) -> CompanyResearchProfile | None: |
| 1373 | asset_type = _instrument_asset_type(instrument) |
| 1374 | company_name = _instrument_name(instrument) |
| 1375 | currency = str(instrument.get("tradingCurrency") or instrument.get("currency") or "").upper() |
| 1376 | country = str(instrument.get("country") or "").upper() |
| 1377 | if asset_type not in {"", "EQUITY"}: |
| 1378 | return None |
| 1379 | if not ticker or ticker == "UNKNOWN" or not exchange or exchange == "UNKNOWN": |
| 1380 | return None |
| 1381 | if not company_name or company_name.upper() in {"UNKNOWN", ticker}: |
| 1382 | return None |
| 1383 | instrument_id = _safe_uuid(instrument.get("instrumentId")) or uuid5( |
| 1384 | NAMESPACE_URL, |
| 1385 | "|".join([provider, provider_instrument_id, isin, ticker, exchange]), |
| 1386 | ) |
| 1387 | profile = CompanyResearchProfile( |
| 1388 | instrument_id=instrument_id, |
| 1389 | company_id=uuid5(NAMESPACE_URL, f"company|{isin}|{ticker}|{exchange}|{company_name}"), |
| 1390 | company_name=company_name, |
| 1391 | aliases=_company_aliases(company_name, ticker), |
| 1392 | provider_instrument_ids={ |
| 1393 | **({provider: provider_instrument_id} if provider and provider_instrument_id else {}), |
| 1394 | **({"NSE": str(instrument.get("nseSymbol"))} if instrument.get("nseSymbol") else {}), |
| 1395 | **({"BSE": str(instrument.get("bseSymbol"))} if instrument.get("bseSymbol") else {}), |
| 1396 | **({"YAHOO_FINANCE": str(instrument.get("structuredProviderTicker"))} if instrument.get("structuredProviderTicker") else {}), |
| 1397 | **{str(mapping.get("provider")).upper(): str(mapping.get("providerSymbol") or mapping.get("providerInstrumentId")) |
| 1398 | for mapping in instrument.get("providerMappings") or [] |
| 1399 | if _trusted_provider_mapping(mapping) and mapping.get("provider") |
| 1400 | and (mapping.get("providerSymbol") or mapping.get("providerInstrumentId"))}, |
| 1401 | }, |
| 1402 | isin=isin or None, |
| 1403 | ticker=ticker, |
| 1404 | exchange=exchange, |
| 1405 | mic=mic or exchange, |
| 1406 | country=country or "UNKNOWN", |
| 1407 | currency=currency or "UNKNOWN", |
| 1408 | known_domains=[], |
| 1409 | ) |
| 1410 | self.repository.profiles.append(profile) |
| 1411 | return profile |
| 1412 | |
| 1413 | |
| 1414 | def _company_from_summary( |
| 1415 | summary: ResearchSummary, |
| 1416 | *, |
| 1417 | status: str, |
| 1418 | source_instrument: dict | None = None, |
| 1419 | safe_error_code: str | None = None, |
| 1420 | safe_error_message: str | None = None, |
| 1421 | ) -> PortfolioResearchCompany: |
| 1422 | events = summary.recent_events |
| 1423 | positive = sum(1 for event in events if "POSITIVE" in event.impact) |
| 1424 | negative = sum(1 for event in events if "NEGATIVE" in event.impact) |
| 1425 | neutral = sum(1 for event in events if event.impact in {EventImpact.NEUTRAL, EventImpact.UNCERTAIN}) |
| 1426 | read_model_score = canonical_read_model_score(summary.catalyst_score) |
| 1427 | evidence_coverage = { |
| 1428 | category: evidence.status for category, evidence in read_model_score.category_evidence.items() |
| 1429 | } |
| 1430 | durable_categories = {"ORDERS_BACKLOG", "CAPEX", "CLIENTS"} |
| 1431 | durable_evidence = { |
| 1432 | category: evidence.model_copy(update={"supporting_events": sorted( |
| 1433 | evidence.supporting_events, |
| 1434 | key=lambda event: event.event_date or event.published_at or event.detected_at, |
| 1435 | reverse=True, |
| 1436 | )[:5]}) |
| 1437 | for category, evidence in read_model_score.category_evidence.items() |
| 1438 | if category in durable_categories |
| 1439 | } |
| 1440 | catalyst_events = [ |
| 1441 | event for evidence in durable_evidence.values() for event in evidence.supporting_events |
| 1442 | if str(getattr(event.event_type, "value", event.event_type)) in {"CAPEX", "CAPACITY_EXPANSION", "NEW_FACILITY", "FACTORY_EXPANSION", "NEW_ORDER", "ORDER_WIN", "MAJOR_CONTRACT", "NEW_CONTRACT", "NEW_CUSTOMER", "CLIENT_WIN", "GEOGRAPHIC_EXPANSION", "ACQUISITION", "PRODUCT_LAUNCH", "REGULATORY_APPROVAL"} |
| 1443 | ] |
| 1444 | durable_evidence["CATALYSTS"] = CategoryEvidence( |
| 1445 | category="CATALYSTS", |
| 1446 | status=EvidenceState.NO_EVIDENCE if not catalyst_events else EvidenceState.NEUTRAL_EVIDENCE, |
| 1447 | score=None, |
| 1448 | event_count=len(catalyst_events), |
| 1449 | source_count=len({event.source_document_id for event in catalyst_events}), |
| 1450 | independent_source_count=len({event.independence_key or event.source_document_id for event in catalyst_events}), |
| 1451 | supporting_events=sorted(catalyst_events, key=lambda event: event.event_date or event.published_at or event.detected_at, reverse=True)[:5], |
| 1452 | ) |
| 1453 | missing = [category for category, state in evidence_coverage.items() if state == "NO_EVIDENCE"] |
| 1454 | source_count = len({document.source_independence_key or document.canonical_url for document in summary.documents}) |
| 1455 | listing_provider, listing_symbol, primary_exchange, verified_mappings = _listing_identity(summary.profile, source_instrument) |
| 1456 | return PortfolioResearchCompany( |
| 1457 | instrument_id=summary.profile.instrument_id, |
| 1458 | company_id=summary.profile.company_id, |
| 1459 | company_name=summary.profile.company_name, |
| 1460 | ticker=summary.profile.ticker, |
| 1461 | exchange=summary.profile.exchange, |
| 1462 | isin=summary.profile.isin, |
| 1463 | provider=source_instrument.get("provider") if source_instrument else None, |
| 1464 | provider_instrument_id=source_instrument.get("providerInstrumentId") if source_instrument else None, |
| 1465 | listing_provider=listing_provider, |
| 1466 | listing_symbol=listing_symbol, |
| 1467 | primary_exchange=primary_exchange, |
| 1468 | verified_provider_mappings=verified_mappings, |
| 1469 | asset_type=_instrument_asset_type(source_instrument) if source_instrument else "EQUITY", |
| 1470 | status=status if summary.documents or events or status != "AVAILABLE" else "RESEARCH_NOT_REFRESHED", |
| 1471 | catalyst_score=summary.catalyst_score.overall_score if summary.documents or events else None, |
| 1472 | confidence=summary.catalyst_score.research_confidence if summary.documents or events else None, |
| 1473 | evidence_coverage=evidence_coverage, |
| 1474 | durable_category_evidence=durable_evidence, |
| 1475 | latest_event=events[0] if events else None, |
| 1476 | positive_events_count=positive, |
| 1477 | negative_events_count=negative, |
| 1478 | neutral_events_count=neutral, |
| 1479 | document_count=len(summary.documents), |
| 1480 | event_count=len(events), |
| 1481 | source_count=source_count, |
| 1482 | last_refresh=summary.last_refresh_at, |
| 1483 | freshness=summary.data_freshness, |
| 1484 | mode="DEMO" if summary.demo else summary.data_freshness, |
| 1485 | missing_categories=missing, |
| 1486 | shareholding_snapshots=summary.shareholding_snapshots, |
| 1487 | shareholding_freshness=summary.shareholding_freshness, |
| 1488 | safe_error_code=safe_error_code, |
| 1489 | safe_error_message=safe_error_message, |
| 1490 | ) |
| 1491 | |
| 1492 | |
| 1493 | def _normalize_search_item(item: dict) -> dict: |
| 1494 | """Coerce a universe listing (NSE master or portfolio-service master) into the |
| 1495 | canonical search-candidate shape used by ``search_instruments``.""" |
| 1496 | ticker = str(item.get("canonicalSymbol") or item.get("primarySymbol") or item.get("ticker") or item.get("symbol") or "").strip() |
| 1497 | company_name = ( |
| 1498 | str(item.get("companyName") or item.get("canonicalName") or item.get("name") or ticker).strip() |
| 1499 | ) |
| 1500 | provider_symbols = [ |
| 1501 | str(mapping.get("providerSymbol") or mapping.get("providerInstrumentId") or "").strip() |
| 1502 | for mapping in (item.get("providerMappings") or []) |
| 1503 | if _trusted_provider_mapping(mapping) and str(mapping.get("status")).upper() == "VERIFIED" |
| 1504 | ] |
| 1505 | return { |
| 1506 | "globalInstrumentId": str(item.get("globalInstrumentId") or ""), |
| 1507 | "companyName": company_name, |
| 1508 | "symbol": ticker, |
| 1509 | "canonicalSymbol": ticker, |
| 1510 | "exchange": str(item.get("exchange") or item.get("primaryExchange") or item.get("mic") or "").strip(), |
| 1511 | "mic": str(item.get("canonicalMic") or item.get("mic") or "").strip(), |
| 1512 | "country": str(item.get("country") or "").strip().upper(), |
| 1513 | "currency": str(item.get("currency") or "").strip().upper(), |
| 1514 | "isin": str(item.get("isin") or "").strip().upper(), |
| 1515 | "sector": str(item.get("canonicalSector") or item.get("sector") or "").strip(), |
| 1516 | "industry": str(item.get("officialIndustry") or item.get("industry") or "").strip(), |
| 1517 | "assetType": str(item.get("assetType") or "EQUITY").strip().upper(), |
| 1518 | "providerSymbols": [symbol for symbol in provider_symbols if symbol], |
| 1519 | } |
| 1520 | |
| 1521 | |
| 1522 | def _search_item_matches(candidate: dict, query: str) -> bool: |
| 1523 | """True when the query token appears against a canonical identity field.""" |
| 1524 | if not query: |
| 1525 | return True |
| 1526 | targets: list[str] = [] |
| 1527 | for key in ("symbol", "companyName", "isin"): |
| 1528 | value = candidate.get(key) |
| 1529 | if value: |
| 1530 | targets.append(str(value).strip().lower()) |
| 1531 | targets.extend( |
| 1532 | str(symbol).strip().lower() for symbol in candidate.get("providerSymbols", []) |
| 1533 | ) |
| 1534 | return any(query in value for value in targets if value) |
| 1535 | |
| 1536 | |
| 1537 | def _search_rank(candidate: dict, query: str) -> int: |
| 1538 | """Rank match quality: 0 = exact identity, ascending = weaker match. |
| 1539 | |
| 1540 | Order: exact symbol/ISIN -> provider symbol -> symbol/name prefix -> |
| 1541 | ISIN prefix -> symbol/name substring. |
| 1542 | """ |
| 1543 | symbol = (candidate.get("symbol") or "").strip().lower() |
| 1544 | company_name = (candidate.get("companyName") or "").strip().lower() |
| 1545 | isin = (candidate.get("isin") or "").strip().lower() |
| 1546 | provider_symbols = [ |
| 1547 | str(symbol).strip().lower() for symbol in candidate.get("providerSymbols", []) |
| 1548 | ] |
| 1549 | if query == symbol or query == isin: |
| 1550 | return 0 |
| 1551 | if query and any(query == symbol for symbol in provider_symbols): |
| 1552 | return 1 |
| 1553 | if symbol.startswith(query): |
| 1554 | return 2 |
| 1555 | if company_name.startswith(query): |
| 1556 | return 3 |
| 1557 | if isin.startswith(query): |
| 1558 | return 4 |
| 1559 | if query in symbol or query in company_name: |
| 1560 | return 5 |
| 1561 | return 5 |
| 1562 | |
| 1563 | |
| 1564 | def _sector_from_records(global_instrument_id: UUID, records: dict[UUID, list]) -> str | None: |
| 1565 | record = _preferred_structured_record(records.get(global_instrument_id, [])) |
| 1566 | if not record: |
| 1567 | return None |
| 1568 | facts = getattr(getattr(record, "snapshot", None), "facts", None) |
| 1569 | if not isinstance(facts, dict): |
| 1570 | return None |
| 1571 | fact = facts.get("sector") |
| 1572 | value = getattr(fact, "value", fact) |
| 1573 | return str(value) if value else None |
| 1574 | |
| 1575 | |
| 1576 | def _industry_from_records(global_instrument_id: UUID, records: dict[UUID, list]) -> str | None: |
| 1577 | record = _preferred_structured_record(records.get(global_instrument_id, [])) |
| 1578 | if not record: |
| 1579 | return None |
| 1580 | facts = getattr(getattr(record, "snapshot", None), "facts", None) |
| 1581 | if not isinstance(facts, dict): |
| 1582 | return None |
| 1583 | fact = facts.get("industry") |
| 1584 | value = getattr(fact, "value", fact) |
| 1585 | return str(value) if value else None |
| 1586 | |
| 1587 | |
| 1588 | def _listing_identity(profile: CompanyResearchProfile | EtfResearchProfile, instrument: dict | None) -> tuple[str | None, str | None, str | None, dict[str, str]]: |
| 1589 | """Expose verified listing identities without changing broker provenance fields.""" |
| 1590 | mappings: dict[str, str] = {} |
| 1591 | for mapping in (instrument or {}).get("providerMappings") or []: |
| 1592 | if not _trusted_provider_mapping(mapping): |
| 1593 | continue |
| 1594 | provider = str(mapping.get("provider") or "").upper() |
| 1595 | symbol = str(mapping.get("providerSymbol") or mapping.get("providerInstrumentId") or "").strip() |
| 1596 | if provider and symbol: |
| 1597 | mappings[provider] = symbol |
| 1598 | for provider, symbol in getattr(profile, "provider_instrument_ids", {}).items(): |
| 1599 | if provider and symbol: |
| 1600 | mappings.setdefault(str(provider).upper(), str(symbol)) |
| 1601 | nse_symbol = mappings.get("NSE") |
| 1602 | if nse_symbol: |
| 1603 | return "NSE", nse_symbol, profile.exchange, mappings |
| 1604 | return None, profile.ticker or _instrument_ticker(instrument or {}), profile.exchange, mappings |
| 1605 | |
| 1606 | |
| 1607 | def _etf_company_from_profile( |
| 1608 | profile: EtfResearchProfile | None, |
| 1609 | instrument: dict, |
| 1610 | documents: list, |
| 1611 | last_refresh, |
| 1612 | safe_error_code: str | None, |
| 1613 | ) -> PortfolioResearchCompany: |
| 1614 | status = "ETF_UNSUPPORTED" |
| 1615 | if profile: |
| 1616 | listing_provider, listing_symbol, primary_exchange, verified_mappings = _listing_identity(profile, instrument) |
| 1617 | else: |
| 1618 | listing_provider, listing_symbol, primary_exchange, verified_mappings = None, None, _instrument_exchange(instrument), {} |
| 1619 | return PortfolioResearchCompany( |
| 1620 | instrument_id=profile.instrument_id if profile else _safe_uuid(instrument.get("instrumentId")), |
| 1621 | company_id=profile.fund_id if profile else None, |
| 1622 | company_name=profile.fund_name if profile else _instrument_name(instrument), |
| 1623 | ticker=profile.ticker if profile else _instrument_ticker(instrument), |
| 1624 | exchange=profile.exchange if profile else _instrument_exchange(instrument), |
| 1625 | isin=profile.isin if profile else instrument.get("isin"), |
| 1626 | provider=profile.provider if profile else instrument.get("provider"), |
| 1627 | provider_instrument_id=profile.provider_instrument_id if profile else instrument.get("providerInstrumentId"), |
| 1628 | listing_provider=listing_provider, |
| 1629 | listing_symbol=listing_symbol, |
| 1630 | primary_exchange=primary_exchange, |
| 1631 | verified_provider_mappings=verified_mappings, |
| 1632 | asset_type="ETF", |
| 1633 | status=status, |
| 1634 | document_count=len(documents), |
| 1635 | source_count=len({document.source_independence_key or document.canonical_url for document in documents}), |
| 1636 | last_refresh=last_refresh, |
| 1637 | freshness="REAL" if documents else "UNAVAILABLE", |
| 1638 | mode="REAL" if documents else "UNAVAILABLE", |
| 1639 | missing_categories=[] if documents else ["ETF_PROFILE", "ETF_PERFORMANCE", "INDEX_OUTLOOK", "ETF_RISK"], |
| 1640 | etf_profile=profile, |
| 1641 | safe_error_code=None if documents else safe_error_code, |
| 1642 | safe_error_message="Company-only research is not applicable to ETFs.", |
| 1643 | ) |
| 1644 | |
| 1645 | |
| 1646 | def _unsupported_asset_company(instrument: dict, asset_type: str) -> PortfolioResearchCompany: |
| 1647 | return PortfolioResearchCompany( |
| 1648 | instrument_id=_safe_uuid(instrument.get("instrumentId")), |
| 1649 | company_name=_instrument_name(instrument), |
| 1650 | ticker=_instrument_ticker(instrument), |
| 1651 | exchange=_instrument_exchange(instrument), |
| 1652 | isin=instrument.get("isin"), |
| 1653 | provider=instrument.get("provider"), |
| 1654 | provider_instrument_id=instrument.get("providerInstrumentId"), |
| 1655 | asset_type=asset_type, |
| 1656 | status="RESEARCH_NOT_APPLICABLE", |
| 1657 | freshness="INSTRUMENT_RESOLVED", |
| 1658 | mode="UNSUPPORTED_ASSET_TYPE", |
| 1659 | safe_error_code="RESEARCH_NOT_APPLICABLE", |
| 1660 | safe_error_message="Research is not applicable for this asset type.", |
| 1661 | ) |
| 1662 | |
| 1663 | |
| 1664 | def _safe_uuid(value) -> UUID | None: |
| 1665 | try: |
| 1666 | return UUID(str(value)) |
| 1667 | except (TypeError, ValueError): |
| 1668 | return None |
| 1669 | |
| 1670 | |
| 1671 | def _normalize_exchange(value) -> str: |
| 1672 | upper = str(value or "").upper() |
| 1673 | return { |
| 1674 | "IBIS": "XETR", |
| 1675 | "IBIS2": "XETR", |
| 1676 | "AEB": "XAMS", |
| 1677 | }.get(upper, upper) |
| 1678 | |
| 1679 | |
| 1680 | def _instrument_ticker(instrument: dict) -> str | None: |
| 1681 | return instrument.get("canonicalSymbol") or instrument.get("ticker") or instrument.get("brokerSymbol") |
| 1682 | |
| 1683 | |
| 1684 | def _instrument_exchange(instrument: dict) -> str: |
| 1685 | return _normalize_exchange(instrument.get("canonicalExchange") or instrument.get("exchange")) |
| 1686 | |
| 1687 | |
| 1688 | def _instrument_name(instrument: dict) -> str: |
| 1689 | custom_name = _clean_name(instrument.get("_researchCustomDisplayName")) |
| 1690 | if custom_name: |
| 1691 | return custom_name |
| 1692 | ticker = _clean_name(instrument.get("ticker")) |
| 1693 | structured_name = _clean_name(instrument.get("companyName")) |
| 1694 | if structured_name and structured_name.upper() != ticker.upper() and not _legacy_composite_company_name(structured_name): |
| 1695 | return structured_name |
| 1696 | resolved_name = _clean_name(instrument.get("canonicalName")) |
| 1697 | if resolved_name and resolved_name.upper() != ticker.upper(): |
| 1698 | return resolved_name |
| 1699 | for candidate in (instrument.get("_researchDisplayName"), structured_name, instrument.get("brokerDescription")): |
| 1700 | parsed = _legacy_composite_company_name(_clean_name(candidate)) |
| 1701 | if parsed: |
| 1702 | return parsed |
| 1703 | return ticker or _clean_name(instrument.get("brokerSymbol")) or "Resolved instrument" |
| 1704 | |
| 1705 | |
| 1706 | def _clean_name(value) -> str: |
| 1707 | return str(value or "").strip() |
| 1708 | |
| 1709 | |
| 1710 | def _company_aliases(company_name: str, ticker: str) -> list[str]: |
| 1711 | legal_suffix = re.compile( |
| 1712 | r"\s+(?:n\.?v\.?|s\.?e\.?|a\.?g\.?|plc|ltd\.?|limited|inc\.?|corp\.?|corporation)$", |
| 1713 | re.IGNORECASE, |
| 1714 | ) |
| 1715 | aliases: list[str] = [] |
| 1716 | current = company_name.strip() |
| 1717 | while current: |
| 1718 | stripped = legal_suffix.sub("", current).strip(" ,.-") |
| 1719 | if stripped == current: |
| 1720 | break |
| 1721 | if stripped and stripped.lower() != company_name.lower() and stripped not in aliases: |
| 1722 | aliases.append(stripped) |
| 1723 | current = stripped |
| 1724 | if ticker and ticker not in aliases: |
| 1725 | aliases.append(ticker) |
| 1726 | return aliases |
| 1727 | |
| 1728 | |
| 1729 | def _global_master_instrument(payload: dict, global_instrument_id: UUID) -> dict: |
| 1730 | """Normalize the public global-instrument API to the position instrument shape.""" |
| 1731 | mappings = [mapping for mapping in payload.get("providerMappings") or [] if isinstance(mapping, dict)] |
| 1732 | verified = [mapping for mapping in mappings if _trusted_provider_mapping(mapping)] |
| 1733 | instrument = { |
| 1734 | "instrumentId": str(global_instrument_id), |
| 1735 | "globalInstrumentId": str(global_instrument_id), |
| 1736 | "isin": payload.get("isin"), |
| 1737 | "ticker": payload.get("primarySymbol"), |
| 1738 | "exchange": payload.get("primaryExchange"), |
| 1739 | "companyName": payload.get("canonicalName"), |
| 1740 | "canonicalName": payload.get("canonicalName"), |
| 1741 | "assetType": payload.get("assetType"), |
| 1742 | "country": payload.get("country"), |
| 1743 | "tradingCurrency": payload.get("currency"), |
| 1744 | "currency": payload.get("currency"), |
| 1745 | "providerMappings": mappings, |
| 1746 | } |
| 1747 | for mapping in verified: |
| 1748 | provider = str(mapping.get("provider") or "").upper() |
| 1749 | symbol = mapping.get("providerSymbol") |
| 1750 | if provider == "NSE" and symbol: |
| 1751 | instrument["nseSymbol"] = symbol |
| 1752 | elif provider == "BSE" and (symbol or mapping.get("providerInstrumentId")): |
| 1753 | instrument["bseSymbol"] = symbol or mapping.get("providerInstrumentId") |
| 1754 | elif provider == "YAHOO_FINANCE" and symbol: |
| 1755 | instrument.update({ |
| 1756 | "structuredProviderTicker": symbol, |
| 1757 | "structuredProviderExchange": mapping.get("exchange"), |
| 1758 | "structuredProviderCurrency": mapping.get("currency"), |
| 1759 | "structuredProviderStatus": mapping.get("status"), |
| 1760 | }) |
| 1761 | if not instrument["ticker"]: |
| 1762 | instrument["ticker"] = instrument.get("nseSymbol") or instrument.get("bseSymbol") |
| 1763 | if not instrument["exchange"]: |
| 1764 | instrument["exchange"] = "NSE" if instrument.get("nseSymbol") else ("BSE" if instrument.get("bseSymbol") else None) |
| 1765 | return instrument |
| 1766 | |
| 1767 | |
| 1768 | def _hydrate_verified_exchange_mappings(profile: CompanyResearchProfile, instrument: dict) -> CompanyResearchProfile: |
| 1769 | """Keep a reused global profile current with verified exchange identities.""" |
| 1770 | mappings = dict(profile.provider_instrument_ids) |
| 1771 | if instrument.get("nseSymbol"): |
| 1772 | mappings["NSE"] = str(instrument["nseSymbol"]) |
| 1773 | if instrument.get("bseSymbol"): |
| 1774 | mappings["BSE"] = str(instrument["bseSymbol"]) |
| 1775 | if instrument.get("structuredProviderTicker"): |
| 1776 | mappings["YAHOO_FINANCE"] = str(instrument["structuredProviderTicker"]) |
| 1777 | for mapping in instrument.get("providerMappings") or []: |
| 1778 | if _trusted_provider_mapping(mapping): |
| 1779 | provider = str(mapping.get("provider") or "").upper() |
| 1780 | value = mapping.get("providerSymbol") or mapping.get("providerInstrumentId") |
| 1781 | if provider and value: |
| 1782 | mappings[provider] = str(value) |
| 1783 | if mappings != profile.provider_instrument_ids: |
| 1784 | profile.provider_instrument_ids = mappings |
| 1785 | return profile |
| 1786 | |
| 1787 | |
| 1788 | def _refresh_profile_from_global_instrument(profile: CompanyResearchProfile, instrument: dict) -> None: |
| 1789 | """Replace provider identity metadata while preserving global research identity and evidence.""" |
| 1790 | profile.company_name = _instrument_name(instrument) |
| 1791 | profile.isin = str(instrument.get("isin") or "").upper() or None |
| 1792 | profile.ticker = str(_instrument_ticker(instrument) or profile.ticker).upper() |
| 1793 | profile.exchange = _instrument_exchange(instrument) or profile.exchange |
| 1794 | profile.mic = _normalize_exchange(instrument.get("canonicalMic") or instrument.get("mic")) or profile.exchange |
| 1795 | profile.country = str(instrument.get("country") or profile.country).upper() |
| 1796 | profile.currency = str(instrument.get("tradingCurrency") or instrument.get("currency") or profile.currency).upper() |
| 1797 | profile.aliases = _company_aliases(profile.company_name, profile.ticker) |
| 1798 | authoritative: dict[str, str] = {} |
| 1799 | for mapping in instrument.get("providerMappings") or []: |
| 1800 | if not _trusted_provider_mapping(mapping): |
| 1801 | continue |
| 1802 | provider = str(mapping.get("provider") or "").upper() |
| 1803 | value = mapping.get("providerSymbol") or mapping.get("providerInstrumentId") |
| 1804 | if provider and value: |
| 1805 | authoritative[provider] = str(value) |
| 1806 | profile.provider_instrument_ids = authoritative |
| 1807 | |
| 1808 | |
| 1809 | def _trusted_provider_mapping(mapping: object) -> bool: |
| 1810 | if not isinstance(mapping, dict) or str(mapping.get("status") or "").upper() not in {"VERIFIED", "RESOLVED"}: |
| 1811 | return False |
| 1812 | return not ( |
| 1813 | str(mapping.get("provider") or "").upper() == "NSE" |
| 1814 | and str(mapping.get("resolutionSource") or mapping.get("resolution_source") or "").upper() == "BROKER_IMPORT_IDENTITY" |
| 1815 | ) |
| 1816 | |
| 1817 | |
| 1818 | def _eligible_for_official_nse_research(profile: CompanyResearchProfile) -> bool: |
| 1819 | return ( |
| 1820 | profile.country.upper() in {"IN", "IND", "INDIA"} |
| 1821 | and profile.exchange.upper() in {"NSE", "XNSE"} |
| 1822 | and bool(profile.provider_instrument_ids.get("NSE")) |
| 1823 | ) |
| 1824 | |
| 1825 | |
| 1826 | def _legacy_composite_company_name(value: str) -> str | None: |
| 1827 | segments = [segment.strip() for segment in value.split("/")] |
| 1828 | return segments[2] if len(segments) >= 3 and segments[2] else None |
| 1829 | |
| 1830 | |
| 1831 | def _attach_structured_market(company: PortfolioResearchCompany, snapshot, error: str | None) -> None: |
| 1832 | company.structured_market = snapshot |
| 1833 | company.structured_provider_status = snapshot.status if snapshot else ( |
| 1834 | "STRUCTURED_PROVIDER_UNAVAILABLE" if error else None |
| 1835 | ) |
| 1836 | if snapshot is None: |
| 1837 | return |
| 1838 | facts = snapshot.facts |
| 1839 | company.current_price = _positive_decimal_fact(facts.get("latestPrice")) |
| 1840 | if company.valuation.current_pe is None: |
| 1841 | company.valuation.current_pe = facts.get("trailingPE") |
| 1842 | if company.valuation.roe is None: |
| 1843 | company.valuation.roe = facts.get("roe") |
| 1844 | if company.valuation.roce is None: |
| 1845 | company.valuation.roce = facts.get("roce") |
| 1846 | # Structured evidence is usable even if every document/search provider failed. |
| 1847 | if company.status in { |
| 1848 | "SEARCH_PROVIDER_UNAVAILABLE", "SEARCH_RETURNED_ZERO_RESULTS", "RESULTS_REJECTED", |
| 1849 | "DOCUMENT_FETCH_FAILED", "EXTRACTION_EMPTY", "RESOLVED_NO_SOURCES", "SOURCE_DISCOVERY_UNAVAILABLE", |
| 1850 | }: |
| 1851 | company.status = "RESOLVED_PARTIAL_DATA" |
| 1852 | company.safe_error_message = "Structured market evidence is available; public document discovery remains incomplete." |
| 1853 | |
| 1854 | |
| 1855 | def _attach_durable_structured_market(company: PortfolioResearchCompany, record, schedules, exceptions, now, settings) -> None: |
| 1856 | if record is None: |
| 1857 | company.structured_provider_status = "NEVER_FETCHED" |
| 1858 | company.price_freshness = "NEVER_FETCHED" |
| 1859 | company.market_status = market_session_status(company.primary_exchange or company.exchange, schedules, exceptions, now) |
| 1860 | return |
| 1861 | _attach_structured_market(company, record.snapshot, None) |
| 1862 | status = market_session_status(record.mic or record.exchange or company.primary_exchange or company.exchange, schedules, exceptions, now) |
| 1863 | company.market_status = status |
| 1864 | company.market_as_of = record.market_as_of |
| 1865 | company.structured_provider_status = record.acquisition_status |
| 1866 | company.price_freshness = "FRESH" if not class_due(record.last_price_at, settings.structured_market_price_freshness_seconds, now) else "STALE" |
| 1867 | company.public_analyst = _public_analyst_from_record(record, now, settings) |
| 1868 | company.market_fundamentals = _market_fundamentals_from_record(record, now, settings) |
| 1869 | latest = _positive_decimal_fact(record.snapshot.facts.get("latestPrice")) |
| 1870 | previous = _decimal_fact(record.snapshot.facts.get("previousClose")) |
| 1871 | company.current_price = latest |
| 1872 | if latest is None or previous is None: |
| 1873 | company.price_direction = "UNKNOWN" |
| 1874 | return |
| 1875 | change = latest - previous |
| 1876 | company.price_change = change |
| 1877 | company.price_direction = "UP" if change > 0 else "DOWN" if change < 0 else "UNCHANGED" |
| 1878 | company.price_change_percent = None if previous == 0 else (change / previous) * Decimal("100") |
| 1879 | |
| 1880 | |
| 1881 | def _public_analyst_from_record(record, now, settings) -> PublicAnalyst | None: |
| 1882 | facts = record.snapshot.facts |
| 1883 | keys = { |
| 1884 | "target_low_price": "publicAnalystTargetLowPrice", "target_median_price": "publicAnalystTargetMedianPrice", |
| 1885 | "target_mean_price": "publicAnalystTargetMeanPrice", "target_high_price": "publicAnalystTargetHighPrice", |
| 1886 | "analyst_count": "publicAnalystCount", "recommendation_mean": "publicAnalystRecommendationMean", |
| 1887 | "consensus": "publicAnalystConsensus", |
| 1888 | } |
| 1889 | selected = {field: facts[key] for field, key in keys.items() if facts.get(key) is not None} |
| 1890 | if not selected: |
| 1891 | return None |
| 1892 | provenance = {(value.source_name, value.source_url) for value in selected.values()} |
| 1893 | source_name, source_url = next(iter(provenance)) if len(provenance) == 1 else (None, None) |
| 1894 | as_of_values = {value.as_of_date for value in selected.values()} |
| 1895 | retrieved_values = {value.retrieved_at for value in selected.values()} |
| 1896 | return PublicAnalyst( |
| 1897 | target_low_price=_decimal_fact(selected.get("target_low_price")), target_median_price=_decimal_fact(selected.get("target_median_price")), |
| 1898 | target_mean_price=_decimal_fact(selected.get("target_mean_price")), target_high_price=_decimal_fact(selected.get("target_high_price")), |
| 1899 | analyst_count=int(_decimal_fact(selected["analyst_count"])) if _decimal_fact(selected.get("analyst_count")) is not None else None, |
| 1900 | recommendation_mean=_decimal_fact(selected.get("recommendation_mean")), consensus=str(selected["consensus"].value) if selected.get("consensus") else None, |
| 1901 | currency=next((value.unit for field, value in selected.items() if field.startswith("target_") and value.unit), record.currency), |
| 1902 | provider=record.provider, provider_instrument_id=record.provider_instrument_id, source_name=source_name, source_url=source_url, |
| 1903 | as_of=next(iter(as_of_values)) if len(as_of_values) == 1 else None, |
| 1904 | retrieved_at=next(iter(retrieved_values)) if len(retrieved_values) == 1 else None, |
| 1905 | freshness="FRESH" if not class_due(record.last_analyst_at, settings.structured_analyst_freshness_seconds, now) else "STALE", |
| 1906 | ) |
| 1907 | |
| 1908 | |
| 1909 | def _market_fundamentals_from_record(record, now, settings) -> MarketFundamentals | None: |
| 1910 | keys = {"market_cap":"marketCap","enterprise_value":"enterpriseValue","trailing_pe":"trailingPE","forward_pe":"forwardPE","price_to_book":"priceToBook","price_to_sales":"priceToSales","ev_to_revenue":"evToRevenue","ev_to_ebitda":"evToEbitda","peg_ratio":"pegRatio","trailing_eps":"trailingEps","forward_eps":"forwardEps","book_value_per_share":"bookValue","roe":"roe","roa":"roa","debt_to_equity":"debtToEquity","profit_margin":"profitMargin","operating_margin":"operatingMargin","revenue_growth":"revenueGrowth","earnings_growth":"earningsGrowth","total_cash":"totalCash","total_debt":"totalDebt","free_cash_flow":"freeCashFlow","operating_cash_flow":"operatingCashFlow"} |
| 1911 | selected = {field: record.snapshot.facts[key] for field, key in keys.items() if record.snapshot.facts.get(key) is not None} |
| 1912 | if not selected: return None |
| 1913 | provenance = {(v.source_name, v.source_url) for v in selected.values()} |
| 1914 | source_name, source_url = next(iter(provenance)) if len(provenance) == 1 else (None, None) |
| 1915 | financial = _is_financial_identity({"sector": getattr(record.snapshot.facts.get("sector"), "value", None), "industry": getattr(record.snapshot.facts.get("industry"), "value", None), "longName": record.snapshot.resolution.company_name}) |
| 1916 | semantics = ({"debt_to_equity": "BANK_SPECIFIC_INTERPRETATION_REQUIRED", "total_debt": "BANK_SPECIFIC_INTERPRETATION_REQUIRED", "operating_margin": "BANK_SPECIFIC_INTERPRETATION_REQUIRED", "ev_to_ebitda": "NOT_MEANINGFUL_FOR_FINANCIAL_ENTITY", "roce": "NOT_MEANINGFUL_FOR_FINANCIAL_ENTITY"} if financial else {}) |
| 1917 | return MarketFundamentals(**{field: _decimal_fact(value) for field, value in selected.items()}, provider=record.provider, provider_instrument_id=record.provider_instrument_id, source_name=source_name, source_url=source_url, as_of=record.market_as_of, retrieved_at=record.retrieved_at, freshness="FRESH" if not class_due(record.last_fundamentals_at or record.last_valuation_at, settings.structured_fundamentals_freshness_seconds, now) else "STALE", metric_semantics=semantics) |
| 1918 | |
| 1919 | |
| 1920 | def _decimal_fact(value) -> Decimal | None: |
| 1921 | if value is None: |
| 1922 | return None |
| 1923 | try: |
| 1924 | return Decimal(str(value.value)) |
| 1925 | except Exception: |
| 1926 | return None |
| 1927 | |
| 1928 | |
| 1929 | def _positive_decimal_fact(value) -> Decimal | None: |
| 1930 | number = _decimal_fact(value) |
| 1931 | return number if number is not None and number.is_finite() and number > 0 else None |
| 1932 | |
| 1933 | |
| 1934 | def _structured_categories(snapshot) -> set[str]: |
| 1935 | if snapshot is None: |
| 1936 | return set() |
| 1937 | facts = snapshot.facts |
| 1938 | categories: set[str] = set() |
| 1939 | if any(key in facts for key in ("trailingPE", "forwardPE", "priceToBook", "evToEbitda")): |
| 1940 | categories.add("VALUATION") |
| 1941 | if any(key in facts for key in ("publicAnalystConsensus", "publicAnalystTargetMeanPrice", "publicAnalystCount")): |
| 1942 | categories.update({"ANALYST_OPINION", "ANALYST_TARGETS"}) |
| 1943 | # Yahoo insider/institution percentages intentionally do not satisfy Indian ownership categories. |
| 1944 | return categories |
| 1945 | |
| 1946 | |
| 1947 | def _preferred_structured_record(records: list[StructuredMarketSnapshotRecord]) -> StructuredMarketSnapshotRecord | None: |
| 1948 | return ( |
| 1949 | next((record for record in records if record.provider == "NSE_STRUCTURED"), None) |
| 1950 | or next((record for record in records if record.provider == "YAHOO_FINANCE"), None) |
| 1951 | or (records[0] if records else None) |
| 1952 | ) |
| 1953 | |
| 1954 | |
| 1955 | def _structured_due_classes( |
| 1956 | record: StructuredMarketSnapshotRecord | None, |
| 1957 | market_status: str, |
| 1958 | settings: Settings, |
| 1959 | now: datetime, |
| 1960 | ) -> set[str]: |
| 1961 | """Return durable structured classes due for an explicit refresh. |
| 1962 | |
| 1963 | A missing first-success snapshot deliberately includes slower classes, so |
| 1964 | closed or unknown exchange scheduling cannot starve initial acquisition. |
| 1965 | """ |
| 1966 | if record is None or record.last_success_at is None: |
| 1967 | return {"PRICE", "VALUATION", "FUNDAMENTALS", "ANALYST"} |
| 1968 | due: set[str] = set() |
| 1969 | if price_sync_eligible(market_status, record.last_price_at, settings.structured_market_price_freshness_seconds, now): |
| 1970 | due.add("PRICE") |
| 1971 | if class_due(record.last_valuation_at or record.last_success_at, settings.structured_valuation_freshness_seconds, now): |
| 1972 | due.add("VALUATION") |
| 1973 | if class_due(record.last_fundamentals_at or record.last_success_at, settings.structured_fundamentals_freshness_seconds, now): |
| 1974 | due.add("FUNDAMENTALS") |
| 1975 | if class_due(record.last_analyst_at or record.last_success_at, settings.structured_analyst_freshness_seconds, now): |
| 1976 | due.add("ANALYST") |
| 1977 | return due |
| 1978 | |
| 1979 | |
| 1980 | def _instrument_asset_type(instrument: dict) -> str: |
| 1981 | asset_type = str(instrument.get("assetType") or "").upper() |
| 1982 | security_type = str(instrument.get("securityType") or "").upper() |
| 1983 | identity = " ".join(str(instrument.get(key) or "") for key in ( |
| 1984 | "companyName", "canonicalName", "brokerDescription", "displayName", "ticker", "canonicalSymbol" |
| 1985 | )).upper() |
| 1986 | if ( |
| 1987 | security_type == "ETF" |
| 1988 | or re.search(r"\b(?:ETF|EXCHANGE[ -]TRADED FUND)\b", identity) |
| 1989 | or re.search(r"\b[A-Z]+BEES\b", identity) |
| 1990 | ): |
| 1991 | return "ETF" |
| 1992 | if security_type in {"ETF", "FUND", "MUTUALFUND", "MF"}: |
| 1993 | return "ETF" if security_type == "ETF" else "FUND" |
| 1994 | return asset_type |
| 1995 | |
| 1996 | |
| 1997 | def _enriched_equity_company(instrument: dict) -> PortfolioResearchCompany | None: |
| 1998 | asset_type = _instrument_asset_type(instrument) |
| 1999 | ticker = _instrument_ticker(instrument) |
| 2000 | exchange = _instrument_exchange(instrument) |
| 2001 | company_name = _instrument_name(instrument).strip() |
| 2002 | if not instrument.get("canonicalName") or not instrument.get("canonicalSymbol") or not instrument.get("canonicalExchange"): |
| 2003 | return None |
| 2004 | if asset_type != "EQUITY" or not ticker or not exchange or exchange == "UNKNOWN": |
| 2005 | return None |
| 2006 | if not company_name or company_name.upper() in {"UNKNOWN", str(ticker).upper()}: |
| 2007 | return None |
| 2008 | return PortfolioResearchCompany( |
| 2009 | instrument_id=_safe_uuid(instrument.get("instrumentId")), |
| 2010 | company_id=uuid5(NAMESPACE_URL, f"company|{instrument.get('isin') or ''}|{ticker}|{exchange}|{company_name}"), |
| 2011 | company_name=company_name, |
| 2012 | ticker=ticker, |
| 2013 | exchange=exchange, |
| 2014 | isin=instrument.get("isin"), |
| 2015 | provider=instrument.get("provider"), |
| 2016 | provider_instrument_id=instrument.get("providerInstrumentId"), |
| 2017 | asset_type=asset_type, |
| 2018 | status="RESOLVED_NO_SOURCES", |
| 2019 | freshness="INSTRUMENT_RESOLVED", |
| 2020 | mode="UNAVAILABLE", |
| 2021 | safe_error_code="RESEARCH_NOT_REFRESHED", |
| 2022 | safe_error_message="No shared public research has been collected for this company yet.", |
| 2023 | ) |
| 2024 | |
| 2025 | |
| 2026 | def _is_real_broker_instrument(instrument: dict) -> bool: |
| 2027 | return str(instrument.get("_positionDataFreshness") or "").upper() == "REAL_BROKER" |
| 2028 | |
| 2029 | |
| 2030 | def _infer_fund_provider_from_name(name: str) -> str | None: |
| 2031 | lower = name.lower() |
| 2032 | if "ishares" in lower or "blackrock" in lower: |
| 2033 | return "iShares" |
| 2034 | if "vanguard" in lower: |
| 2035 | return "Vanguard" |
| 2036 | if "xtrackers" in lower: |
| 2037 | return "Xtrackers" |
| 2038 | return None |
| 2039 | |
| 2040 | |
| 2041 | def _infer_underlying_index_from_name(name: str) -> str | None: |
| 2042 | upper = name.upper() |
| 2043 | if "S&P 500" in upper or "SP 500" in upper: |
| 2044 | return "S&P 500" |
| 2045 | if "NASDAQ 100" in upper or "NASDAQ-100" in upper: |
| 2046 | return "NASDAQ 100" |
| 2047 | return None |
| 2048 | |
| 2049 | |
| 2050 | def _status_from_summary(summary: ResearchSummary, live_error: str | None = None) -> str: |
| 2051 | if summary.documents or summary.recent_events: |
| 2052 | categories_with_evidence = sum( |
| 2053 | evidence.status != "NO_EVIDENCE" |
| 2054 | for evidence in summary.catalyst_score.category_evidence.values() |
| 2055 | ) |
| 2056 | return "RESOLVED_RESEARCH_AVAILABLE" if categories_with_evidence >= 3 else "RESOLVED_PARTIAL_DATA" |
| 2057 | if live_error in { |
| 2058 | "SEARCH_PROVIDER_UNAVAILABLE", "SEARCH_RETURNED_ZERO_RESULTS", "RESULTS_REJECTED", |
| 2059 | "DOCUMENT_FETCH_FAILED", "EXTRACTION_EMPTY", |
| 2060 | }: |
| 2061 | return live_error |
| 2062 | if live_error and live_error.startswith("SEARCH_PROVIDER_UNAVAILABLE"): |
| 2063 | return "SEARCH_PROVIDER_UNAVAILABLE" |
| 2064 | if live_error == "GOOGLE_PROVIDER_UNAVAILABLE" or (live_error and live_error.startswith("SEARCH_PROVIDER")): |
| 2065 | return "SEARCH_PROVIDER_UNAVAILABLE" |
| 2066 | if live_error and live_error.startswith("SEARCH_SOURCE_UNAVAILABLE"): |
| 2067 | return "DOCUMENT_FETCH_FAILED" |
| 2068 | if live_error: |
| 2069 | return "SEARCH_PROVIDER_UNAVAILABLE" |
| 2070 | return "RESOLVED_NO_SOURCES" |