| 1 | from __future__ import annotations |
| 2 | |
| 3 | import asyncio |
| 4 | import re |
| 5 | import logging |
| 6 | import threading |
| 7 | import time |
| 8 | from dataclasses import dataclass |
| 9 | from datetime import date, datetime, timedelta, timezone |
| 10 | from urllib.parse import urlparse |
| 11 | from uuid import UUID |
| 12 | |
| 13 | from app.deduplication import DocumentDeduplicator |
| 14 | from app.entity_resolution import EntityResolver |
| 15 | from app.events import company_updated, document_event, research_event_extracted |
| 16 | from app.extraction import RuleBasedEventExtractor |
| 17 | from app.models import ( |
| 18 | CompanyResearchProfile, |
| 19 | DocumentSubtype, |
| 20 | DocumentStatus, |
| 21 | DocumentType, |
| 22 | EtfResearchProfile, |
| 23 | PlatformEvent, |
| 24 | ReliabilityLevel, |
| 25 | ResearchDocument, |
| 26 | ResearchEvidenceSource, |
| 27 | ResearchEvent, |
| 28 | ResearchEventType, |
| 29 | ResearchSummary, |
| 30 | ProvenancedValue, |
| 31 | SourceClassification, |
| 32 | SourceMode, |
| 33 | ShareholdingSnapshot, |
| 34 | SourceType, |
| 35 | StructuredMarketSnapshotRecord, |
| 36 | DailyMarketBar, |
| 37 | ) |
| 38 | from app.normalization import canonicalize_url, content_hash, detect_document_type, extract_published_at, extract_text, normalize_text |
| 39 | from app.research_fetching import FetchError, HttpResearchFetcher, PdfExtractionTimeoutError, RestrictedFetchError, TransportFetchError |
| 40 | from app.scoring import CatalystScorer, canonical_read_model_score |
| 41 | from app.settings import Settings |
| 42 | from app.shareholding import parse_nse_shareholding_xbrl, parse_official_shareholding |
| 43 | from app.persistence import ResearchPersistence, SqliteResearchPersistence, persistence_from_settings |
| 44 | from app.source_discovery import ( |
| 45 | ApprovedSourceDiscovery, |
| 46 | BraveCompatibleSearchDiscoveryProvider, |
| 47 | DisabledSearchDiscoveryProvider, |
| 48 | DiscoveryResult, |
| 49 | GoogleCompatibleSearchDiscoveryProvider, |
| 50 | OfficialFilingDiscovery, |
| 51 | OfficialNseShareholdingDiscovery, |
| 52 | SearchDiscoveryProvider, |
| 53 | SearchProviderConfigurationError, |
| 54 | SearchProviderError, |
| 55 | SearchDiscoveryService, |
| 56 | SearxngSearchDiscoveryProvider, |
| 57 | ) |
| 58 | from app.source_registry import RegisteredResearchSource, registered_sources_for |
| 59 | from app.structured_research import financial_result_history_from_facts, financial_statement_history_from_facts, latest_quarterly_result_from_facts, latest_quarterly_result, parsed_nse_balance_sheet_periods, parsed_nse_cash_flow_periods, parsed_nse_income_statement_periods |
| 60 | from app.fact_precedence import FinancialFact, FinancialFactKey, FactSourceTier |
| 61 | |
| 62 | logger = logging.getLogger(__name__) |
| 63 | _TRUSTED_NSE_PROFILE_IDENTITY = object() |
| 64 | |
| 65 | |
| 66 | @dataclass(frozen=True) |
| 67 | class _InstrumentRefreshGate: |
| 68 | missing_categories: set[str] |
| 69 | shareholding_backfill_needed: bool |
| 70 | shareholding_category_enrichment_needed: bool |
| 71 | state: str |
| 72 | |
| 73 | @property |
| 74 | def requires_provider_work(self) -> bool: |
| 75 | return bool( |
| 76 | self.missing_categories |
| 77 | or self.shareholding_backfill_needed |
| 78 | or self.shareholding_category_enrichment_needed |
| 79 | ) |
| 80 | |
| 81 | |
| 82 | @dataclass(frozen=True) |
| 83 | class _CategoryRefreshStrategy: |
| 84 | name: str |
| 85 | lightweight_check_interval: timedelta |
| 86 | |
| 87 | |
| 88 | _SHORT_TTL = _CategoryRefreshStrategy("SHORT_TTL", timedelta(minutes=5)) |
| 89 | _DAILY_LIGHTWEIGHT = _CategoryRefreshStrategy("DAILY_LIGHTWEIGHT", timedelta(days=1)) |
| 90 | _PERIODIC_SLOW = _CategoryRefreshStrategy("PERIODIC_SLOW", timedelta(days=7)) |
| 91 | _QUARTERLY_WINDOW = _CategoryRefreshStrategy("QUARTERLY_WINDOW", timedelta(days=3)) |
| 92 | _ANNUAL_WINDOW = _CategoryRefreshStrategy("ANNUAL_WINDOW", timedelta(days=14)) |
| 93 | |
| 94 | _CATEGORY_STRATEGIES: dict[str, _CategoryRefreshStrategy] = { |
| 95 | "SHAREHOLDING_PATTERN": _QUARTERLY_WINDOW, |
| 96 | "FINANCIAL_RESULTS": _QUARTERLY_WINDOW, |
| 97 | "ANNUAL_REPORT": _ANNUAL_WINDOW, |
| 98 | "VALUATION": _SHORT_TTL, |
| 99 | "ANALYST_OPINION": _DAILY_LIGHTWEIGHT, |
| 100 | "ANALYST_TARGETS": _DAILY_LIGHTWEIGHT, |
| 101 | "ORDERS_BACKLOG": _DAILY_LIGHTWEIGHT, |
| 102 | "CONTRACTS": _DAILY_LIGHTWEIGHT, |
| 103 | "CAPEX": _DAILY_LIGHTWEIGHT, |
| 104 | "NEW_FACILITIES": _DAILY_LIGHTWEIGHT, |
| 105 | "ACQUISITIONS": _DAILY_LIGHTWEIGHT, |
| 106 | "CLIENTS": _DAILY_LIGHTWEIGHT, |
| 107 | "GUIDANCE": _DAILY_LIGHTWEIGHT, |
| 108 | "MANAGEMENT": _DAILY_LIGHTWEIGHT, |
| 109 | "REGULATORY": _DAILY_LIGHTWEIGHT, |
| 110 | "CATALYSTS": _DAILY_LIGHTWEIGHT, |
| 111 | "ORDERS_BACKLOG": _DAILY_LIGHTWEIGHT, |
| 112 | "CAPEX": _PERIODIC_SLOW, |
| 113 | "NEW_FACILITIES": _PERIODIC_SLOW, |
| 114 | "CLIENTS": _PERIODIC_SLOW, |
| 115 | "PRODUCTS": _PERIODIC_SLOW, |
| 116 | "GROWTH": _PERIODIC_SLOW, |
| 117 | } |
| 118 | |
| 119 | _REFRESH_CATEGORY_ALIASES = { |
| 120 | "FINANCIAL RESULTS": "FINANCIAL_RESULTS", |
| 121 | "SHAREHOLDING PATTERN": "SHAREHOLDING_PATTERN", |
| 122 | "ANNUAL REPORT": "ANNUAL_REPORT", |
| 123 | "ANALYST OPINION": "ANALYST_OPINION", |
| 124 | "ANALYST TARGETS": "ANALYST_TARGETS", |
| 125 | "INSTITUTIONAL ACTIVITY": "INSTITUTIONAL_ACTIVITY", |
| 126 | "GUIDANCE": "GUIDANCE", |
| 127 | "GROWTH": "GROWTH", |
| 128 | "CUSTOMERS": "CLIENTS", |
| 129 | "NEW CUSTOMERS": "CLIENTS", |
| 130 | "CLIENTS": "CLIENTS", |
| 131 | "ORDERS BACKLOG": "ORDERS_BACKLOG", |
| 132 | "NEW ORDERS": "ORDERS_BACKLOG", |
| 133 | "ORDERS_BACKLOG": "ORDERS_BACKLOG", |
| 134 | "CAPEX CAPACITY": "CAPEX", |
| 135 | "CAPEX": "CAPEX", |
| 136 | "OWNERSHIP": "INSTITUTIONAL_ACTIVITY", |
| 137 | "REGULATORY": "REGULATORY", |
| 138 | "MANAGEMENT": "MANAGEMENT", |
| 139 | } |
| 140 | |
| 141 | |
| 142 | class ResearchRepository: |
| 143 | def __init__( |
| 144 | self, |
| 145 | settings: Settings | None = None, |
| 146 | fetcher: HttpResearchFetcher | None = None, |
| 147 | discovery: ApprovedSourceDiscovery | None = None, |
| 148 | search_discovery: SearchDiscoveryService | None = None, |
| 149 | official_filing_discovery: OfficialFilingDiscovery | None = None, |
| 150 | official_shareholding_discovery: OfficialNseShareholdingDiscovery | None = None, |
| 151 | persistence: ResearchPersistence | None = None, |
| 152 | ) -> None: |
| 153 | self.settings = settings or Settings() |
| 154 | self.profiles = _demo_profiles() |
| 155 | self.etf_profiles: list[EtfResearchProfile] = [] |
| 156 | self.documents: dict[UUID, ResearchDocument] = {} |
| 157 | self.events: dict[UUID, ResearchEvent] = {} |
| 158 | self.shareholding_snapshots: dict[UUID, ShareholdingSnapshot] = {} |
| 159 | self.platform_events: list[PlatformEvent] = [] |
| 160 | self.last_refresh: dict[UUID, datetime] = {} |
| 161 | self.last_live_error: dict[UUID, str] = {} |
| 162 | self._category_refresh: dict[tuple[UUID, str], datetime] = {} |
| 163 | # A successful provider response with no qualifying evidence is not |
| 164 | # freshness. Keep its check cadence separate from durable evidence so |
| 165 | # normal refreshes do not rediscover the same missing category. |
| 166 | self._category_successful_no_change_checks: dict[tuple[UUID, str], datetime] = {} |
| 167 | self._deduplicator = DocumentDeduplicator() |
| 168 | self._event_keys: set[tuple[UUID, ResearchEventType, str, str | None, str | None]] = set() |
| 169 | self._resolver = EntityResolver(self.profiles) |
| 170 | self._extractor = RuleBasedEventExtractor() |
| 171 | self._scorer = CatalystScorer() |
| 172 | self._fetcher = fetcher or HttpResearchFetcher(self.settings) |
| 173 | self._persistence = persistence or persistence_from_settings(self.settings) |
| 174 | self._discovery = discovery or ApprovedSourceDiscovery() |
| 175 | self._search_discovery = search_discovery or SearchDiscoveryService( |
| 176 | _search_provider_from_settings(self.settings), |
| 177 | max_queries_per_category=self.settings.research_search_max_queries_per_category, |
| 178 | max_results_per_query=self.settings.research_search_max_results_per_query, |
| 179 | max_documents_per_refresh=self.settings.research_search_max_documents_per_refresh, |
| 180 | allowed_domains=self.settings.research_search_allowed_domains, |
| 181 | ) |
| 182 | self._official_filing_discovery = official_filing_discovery or OfficialFilingDiscovery( |
| 183 | announcements_url=self.settings.nse_announcements_url |
| 184 | ) |
| 185 | self._official_shareholding_discovery = official_shareholding_discovery or OfficialNseShareholdingDiscovery( |
| 186 | shareholdings_url=self.settings.nse_shareholdings_url |
| 187 | ) |
| 188 | # This is deliberately process-local. Persistence remains the durable |
| 189 | # cross-process deduplication boundary. |
| 190 | self._official_filing_flights: dict[tuple[UUID, str], asyncio.Task[ResearchDocument]] = {} |
| 191 | self._instrument_refresh_flights: dict[UUID, asyncio.Task[ResearchSummary]] = {} |
| 192 | # The production persistence adapter owns one synchronous database |
| 193 | # connection. Worker operations are serialized per repository so two |
| 194 | # refreshes do not interleave transactions on that connection. |
| 195 | self._persistence_worker_lock = threading.RLock() |
| 196 | self._news_worker_lock = asyncio.Lock() |
| 197 | self._seed_demo_data() |
| 198 | self._load_persisted_research() |
| 199 | |
| 200 | def news_records_for(self, instrument_id, model, *, as_of): |
| 201 | loader = getattr(self._persistence, 'load_news_records', None) |
| 202 | return loader(model, instrument_id, as_of=as_of) if callable(loader) else [] |
| 203 | |
| 204 | async def append_news_record(self, record): |
| 205 | return await self._run_blocking_persistence(self._persistence.append_news_record, record) |
| 206 | |
| 207 | async def refresh_news_intelligence(self, instrument_id, *, industry=None): |
| 208 | from app.news_acquisition import acquire_news |
| 209 | async with self._news_worker_lock: |
| 210 | return await acquire_news(self,self.profile(instrument_id),providers=[self._search_discovery.provider],industry=industry, |
| 211 | max_queries=min(20,max(1,self.settings.research_search_max_queries_per_category)), |
| 212 | max_documents=min(20,max(1,self.settings.research_search_max_documents_per_refresh))) |
| 213 | |
| 214 | def list_profiles(self) -> list[CompanyResearchProfile]: |
| 215 | return self.profiles |
| 216 | |
| 217 | @property |
| 218 | def persistence(self): |
| 219 | return self._persistence |
| 220 | |
| 221 | def list_etf_profiles(self) -> list[EtfResearchProfile]: |
| 222 | return self.etf_profiles |
| 223 | |
| 224 | def financial_facts_for(self, instrument_id: UUID): |
| 225 | return self._persistence.load_financial_facts({instrument_id}) |
| 226 | |
| 227 | async def financial_facts_for_instruments(self, instrument_ids: set[UUID]) -> dict[UUID, list[FinancialFact]]: |
| 228 | started = time.perf_counter() |
| 229 | facts = await self._run_blocking_persistence(self._persistence.load_financial_facts, instrument_ids) |
| 230 | grouped: dict[UUID, list[FinancialFact]] = {instrument_id: [] for instrument_id in instrument_ids} |
| 231 | for fact in facts: |
| 232 | if fact.key.instrument_id in grouped: |
| 233 | grouped[fact.key.instrument_id].append(fact) |
| 234 | logger.info( |
| 235 | "portfolio_summary_stage stage=FINANCIAL_FACTS durationMs=%s requestedInstrumentCount=%s returnedFactCount=%s", |
| 236 | round((time.perf_counter() - started) * 1000), |
| 237 | len(instrument_ids), |
| 238 | sum(len(values) for values in grouped.values()), |
| 239 | ) |
| 240 | return grouped |
| 241 | |
| 242 | def persist_yahoo_statement_facts(self, instrument_id: UUID, snapshot) -> None: |
| 243 | for raw in getattr(snapshot, "statement_facts", []): |
| 244 | if not isinstance(raw, dict) or raw.get("value") is None or not raw.get("periodEnd") or raw.get("periodType") not in {"ANNUAL", "QUARTERLY"}: |
| 245 | continue |
| 246 | self._persistence.upsert_financial_fact(FinancialFact( |
| 247 | FinancialFactKey(instrument_id, str(raw["metric"]), str(raw["periodEnd"]), str(raw["periodType"]), "UNKNOWN"), |
| 248 | ProvenancedValue(value=raw["value"], unit=raw.get("unit"), source_url=str(raw["sourceUrl"]), |
| 249 | source_name=str(raw["sourceName"]), source_type=str(raw["sourceType"]), retrieved_at=raw["retrievedAt"], confidence=raw.get("confidence")), |
| 250 | FactSourceTier.YAHOO, "YAHOO_FINANCE", f"{snapshot.resolution.provider_ticker}:{raw['metric']}:{raw['periodEnd']}:{raw['periodType']}", SourceMode.REAL, |
| 251 | )) |
| 252 | |
| 253 | async def persist_yahoo_statement_facts_async(self, instrument_id: UUID, snapshot) -> None: |
| 254 | await self._run_blocking_persistence(self.persist_yahoo_statement_facts, instrument_id, snapshot) |
| 255 | |
| 256 | def persist_international_financial_facts(self, facts: list[FinancialFact]) -> int: |
| 257 | """Persist provider-neutral international facts through the existing durable model.""" |
| 258 | written = 0 |
| 259 | for fact in facts: |
| 260 | if self._persistence.upsert_financial_fact(fact): |
| 261 | written += 1 |
| 262 | return written |
| 263 | |
| 264 | async def persist_international_financial_facts_async(self, facts: list[FinancialFact]) -> int: |
| 265 | return await self._run_blocking_persistence(self.persist_international_financial_facts, facts) |
| 266 | |
| 267 | def structured_market_snapshots_for(self, instrument_ids: set[UUID]) -> dict[UUID, list[StructuredMarketSnapshotRecord]]: |
| 268 | grouped = {instrument_id: [] for instrument_id in instrument_ids} |
| 269 | for record in self._persistence.load_structured_market_snapshots(instrument_ids): |
| 270 | grouped.setdefault(record.instrument_id, []).append(record) |
| 271 | return grouped |
| 272 | |
| 273 | async def structured_market_snapshots_for_instruments(self, instrument_ids: set[UUID]): |
| 274 | return await self._run_blocking_persistence(self.structured_market_snapshots_for, instrument_ids) |
| 275 | |
| 276 | def market_price_observations_for(self, instrument_ids: set[UUID]): |
| 277 | grouped = {instrument_id: [] for instrument_id in instrument_ids} |
| 278 | for observation in self._persistence.load_market_price_observations(instrument_ids): |
| 279 | grouped.setdefault(observation.instrument_id, []).append(observation) |
| 280 | return grouped |
| 281 | |
| 282 | async def market_price_observations_for_instruments(self, instrument_ids: set[UUID]): |
| 283 | return await self._run_blocking_persistence(self.market_price_observations_for, instrument_ids) |
| 284 | |
| 285 | async def market_price_coverage_for_instruments(self, instrument_ids: set[UUID]): |
| 286 | return await self._run_blocking_persistence( |
| 287 | self._persistence.load_market_price_coverage, instrument_ids |
| 288 | ) |
| 289 | |
| 290 | async def upsert_market_price_observation_async(self, observation) -> None: |
| 291 | await self._run_blocking_persistence(self._persistence.upsert_market_price_observation, observation) |
| 292 | |
| 293 | def daily_market_bars_for(self, instrument_ids: set[UUID], *, start_date: date | None = None, |
| 294 | end_date: date | None = None, provider: str | None = None) -> dict[UUID, list[DailyMarketBar]]: |
| 295 | grouped = {instrument_id: [] for instrument_id in instrument_ids} |
| 296 | for bar in self._persistence.load_daily_market_bars( |
| 297 | instrument_ids, start_date=start_date, end_date=end_date, provider=provider): |
| 298 | grouped[bar.global_instrument_id].append(bar) |
| 299 | return grouped |
| 300 | |
| 301 | async def daily_market_bars_for_instruments(self, instrument_ids: set[UUID], *, start_date: date | None = None, |
| 302 | end_date: date | None = None, provider: str | None = None) -> dict[UUID, list[DailyMarketBar]]: |
| 303 | return await self._run_blocking_persistence(self.daily_market_bars_for, instrument_ids, |
| 304 | start_date=start_date, end_date=end_date, provider=provider) |
| 305 | |
| 306 | async def upsert_daily_market_bar_async(self, bar: DailyMarketBar) -> None: |
| 307 | await self._run_blocking_persistence(self._persistence.upsert_daily_market_bar, bar) |
| 308 | |
| 309 | async def upsert_daily_market_bars_async(self, bars: list[DailyMarketBar]) -> int: |
| 310 | return await self._run_blocking_persistence(self._persistence.upsert_daily_market_bars, bars) |
| 311 | |
| 312 | async def stock_rule_engine_result( |
| 313 | self, |
| 314 | global_instrument_id: UUID, |
| 315 | rule_engine_version: str, |
| 316 | input_fingerprint: str, |
| 317 | ) -> dict | None: |
| 318 | return await self._run_blocking_persistence( |
| 319 | self._persistence.load_stock_rule_engine_result, |
| 320 | global_instrument_id, |
| 321 | rule_engine_version, |
| 322 | input_fingerprint, |
| 323 | ) |
| 324 | |
| 325 | async def persist_stock_rule_engine_result(self, result: dict) -> None: |
| 326 | await self._run_blocking_persistence( |
| 327 | self._persistence.upsert_stock_rule_engine_result, result |
| 328 | ) |
| 329 | |
| 330 | async def persist_structured_market_snapshot_async(self, record: StructuredMarketSnapshotRecord) -> None: |
| 331 | await self._run_blocking_persistence(self._persistence.upsert_structured_market_snapshot, record) |
| 332 | |
| 333 | async def record_structured_market_failure_async(self, instrument_id: UUID, provider: str, code: str, message: str) -> None: |
| 334 | await self._run_blocking_persistence(self._persistence.record_structured_market_failure, instrument_id, provider, datetime.now(timezone.utc), code, message) |
| 335 | |
| 336 | async def market_session_data(self, markets: set[str]): |
| 337 | schedules, exceptions = await asyncio.gather( |
| 338 | self._run_blocking_persistence(self._persistence.load_market_schedules, markets), |
| 339 | self._run_blocking_persistence(self._persistence.load_market_calendar_exceptions, markets), |
| 340 | ) |
| 341 | return schedules, exceptions |
| 342 | |
| 343 | async def _run_blocking_persistence(self, operation, *args, **kwargs): |
| 344 | """Keep production database work out of the request event loop. |
| 345 | |
| 346 | The in-memory SQLite adapter is deliberately single-thread-affine and |
| 347 | is used only by the local/test persistence configuration. Production |
| 348 | uses the psycopg adapter, whose synchronous operations are safe to run |
| 349 | in a worker thread. Keeping this compatibility branch here avoids |
| 350 | moving repository-owned state or changing the SQLite adapter contract. |
| 351 | """ |
| 352 | if isinstance(self._persistence, SqliteResearchPersistence) and self._persistence.__class__.__module__ != "app.postgres_persistence": |
| 353 | return operation(*args, **kwargs) |
| 354 | return await asyncio.to_thread(self._run_serialized_persistence_operation, operation, args, kwargs) |
| 355 | |
| 356 | def _run_serialized_persistence_operation(self, operation, args, kwargs): |
| 357 | with self._persistence_worker_lock: |
| 358 | return operation(*args, **kwargs) |
| 359 | |
| 360 | def etf_profile(self, instrument_id: UUID) -> EtfResearchProfile: |
| 361 | return next(profile for profile in self.etf_profiles if profile.instrument_id == instrument_id) |
| 362 | |
| 363 | def profile(self, instrument_id: UUID) -> CompanyResearchProfile: |
| 364 | return next(profile for profile in self.profiles if profile.instrument_id == instrument_id) |
| 365 | |
| 366 | def instrument_refresh_state(self, instrument_id: UUID) -> str: |
| 367 | """Return the global public-research gate without doing provider work.""" |
| 368 | return self._instrument_refresh_gate( |
| 369 | self.profile(instrument_id), set(), datetime.now(timezone.utc) |
| 370 | ).state |
| 371 | |
| 372 | async def backfill(self, instrument_id: UUID, correlation_id: str | None = None) -> ResearchSummary: |
| 373 | """Intentional, global deep repair for missing supported evidence. |
| 374 | |
| 375 | This bypasses normal due scheduling only; durable document/snapshot |
| 376 | reuse and process-local per-instrument single-flight remain in force. |
| 377 | """ |
| 378 | existing = self._instrument_refresh_flights.get(instrument_id) |
| 379 | if existing is not None: |
| 380 | return await asyncio.shield(existing) |
| 381 | task = asyncio.create_task(self._backfill_once(instrument_id, correlation_id)) |
| 382 | self._instrument_refresh_flights[instrument_id] = task |
| 383 | |
| 384 | def cleanup(completed: asyncio.Task[ResearchSummary]) -> None: |
| 385 | if self._instrument_refresh_flights.get(instrument_id) is completed: |
| 386 | self._instrument_refresh_flights.pop(instrument_id, None) |
| 387 | |
| 388 | task.add_done_callback(cleanup) |
| 389 | return await asyncio.shield(task) |
| 390 | |
| 391 | async def _backfill_once(self, instrument_id: UUID, correlation_id: str | None) -> ResearchSummary: |
| 392 | profile = self.profile(instrument_id) |
| 393 | if self.settings.research_live_enabled: |
| 394 | await self._refresh_live(instrument_id, set(), force=True) |
| 395 | self.last_refresh[instrument_id] = datetime.now(timezone.utc) |
| 396 | logger.info("research_backfill_complete globalInstrumentId=%s correlationId=%s", instrument_id, correlation_id or "NONE") |
| 397 | return self.summary(instrument_id, allow_demo=True) |
| 398 | |
| 399 | def documents_for(self, instrument_id: UUID, source_mode: SourceMode | None = None) -> list[ResearchDocument]: |
| 400 | return sorted( |
| 401 | [ |
| 402 | doc |
| 403 | for doc in self.documents.values() |
| 404 | if doc.instrument_id == instrument_id and (source_mode is None or doc.source_mode == source_mode) |
| 405 | ], |
| 406 | key=lambda doc: doc.published_at or doc.retrieved_at, |
| 407 | reverse=True, |
| 408 | ) |
| 409 | |
| 410 | def register_etf_profile(self, profile: EtfResearchProfile) -> EtfResearchProfile: |
| 411 | for existing in self.etf_profiles: |
| 412 | if existing.instrument_id == profile.instrument_id: |
| 413 | return existing |
| 414 | if profile.provider and profile.provider_instrument_id: |
| 415 | if ( |
| 416 | existing.provider |
| 417 | and existing.provider.upper() == profile.provider.upper() |
| 418 | and existing.provider_instrument_id |
| 419 | and existing.provider_instrument_id.upper() == profile.provider_instrument_id.upper() |
| 420 | ): |
| 421 | return existing |
| 422 | if profile.isin and existing.isin and profile.isin.upper() == existing.isin.upper(): |
| 423 | return existing |
| 424 | self.etf_profiles.append(profile) |
| 425 | return profile |
| 426 | |
| 427 | def events_for( |
| 428 | self, |
| 429 | instrument_id: UUID, |
| 430 | event_type: ResearchEventType | None = None, |
| 431 | impact: str | None = None, |
| 432 | reliability: ReliabilityLevel | None = None, |
| 433 | source_mode: SourceMode | None = None, |
| 434 | ) -> list[ResearchEvent]: |
| 435 | values = [event for event in self.events.values() if event.instrument_id == instrument_id] |
| 436 | if event_type: |
| 437 | values = [event for event in values if event.event_type == event_type] |
| 438 | if impact: |
| 439 | values = [event for event in values if event.impact == impact] |
| 440 | if reliability: |
| 441 | values = [event for event in values if event.reliability == reliability] |
| 442 | if source_mode: |
| 443 | values = [event for event in values if event.source_mode == source_mode] |
| 444 | return sorted(values, key=lambda event: event.event_date or event.detected_at, reverse=True) |
| 445 | |
| 446 | def shareholding_for(self, instrument_id: UUID, *, limit: int = 4) -> list[ShareholdingSnapshot]: |
| 447 | ordered = sorted( |
| 448 | [snapshot for snapshot in self.shareholding_snapshots.values() |
| 449 | if snapshot.instrument_id == instrument_id |
| 450 | and snapshot.source_mode == SourceMode.REAL |
| 451 | and _is_quarter_end(snapshot.period_end)], |
| 452 | key=lambda snapshot: (snapshot.period_end, snapshot.published_at or datetime.min.replace(tzinfo=timezone.utc), snapshot.retrieved_at), reverse=True, |
| 453 | ) |
| 454 | latest_by_period: list[ShareholdingSnapshot] = [] |
| 455 | periods: set[datetime] = set() |
| 456 | for snapshot in ordered: |
| 457 | if snapshot.period_end in periods: |
| 458 | continue |
| 459 | periods.add(snapshot.period_end) |
| 460 | latest_by_period.append(snapshot) |
| 461 | if len(latest_by_period) == limit: |
| 462 | break |
| 463 | return latest_by_period |
| 464 | |
| 465 | def persist_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool: |
| 466 | if not snapshot.values or snapshot.source_mode != SourceMode.REAL: |
| 467 | return False |
| 468 | try: |
| 469 | created = self._persistence.upsert_shareholding_snapshot(snapshot) |
| 470 | except Exception as exc: |
| 471 | raise FetchError("SHAREHOLDING_PERSIST_FAILED") from exc |
| 472 | self._record_persisted_shareholding_snapshot(snapshot, created) |
| 473 | return created |
| 474 | |
| 475 | async def _persist_shareholding_snapshot_async(self, snapshot: ShareholdingSnapshot) -> bool: |
| 476 | if not snapshot.values or snapshot.source_mode != SourceMode.REAL: |
| 477 | return False |
| 478 | try: |
| 479 | created = await self._run_blocking_persistence(self._persistence.upsert_shareholding_snapshot, snapshot) |
| 480 | except Exception as exc: |
| 481 | raise FetchError("SHAREHOLDING_PERSIST_FAILED") from exc |
| 482 | self._record_persisted_shareholding_snapshot(snapshot, created) |
| 483 | return created |
| 484 | |
| 485 | async def persist_external_mcp_shareholding_async( |
| 486 | self, snapshot: ShareholdingSnapshot |
| 487 | ) -> bool: |
| 488 | """Persist exact normalized MCP ownership categories after adapter validation.""" |
| 489 | if snapshot.source_provider != "YAHOO_FINANCE_MCP": |
| 490 | raise ValueError("UNAPPROVED_EXTERNAL_MCP_PROVIDER") |
| 491 | return await self._persist_shareholding_snapshot_async(snapshot) |
| 492 | |
| 493 | async def record_acquisition_observation(self, instrument_id, requirement_id, provider, outcome, observed_at, source_url=None, failure_reason=None, evidence_count=0): |
| 494 | await self._run_blocking_persistence(self._persistence.upsert_acquisition_observation, |
| 495 | instrument_id, requirement_id, provider, outcome, observed_at, source_url, failure_reason, evidence_count) |
| 496 | |
| 497 | def acquisition_observations_for(self, instrument_id): |
| 498 | loader = getattr(self._persistence, "load_acquisition_observations", None) |
| 499 | return loader(instrument_id) if loader else [] |
| 500 | |
| 501 | async def persist_external_mcp_evidence_async( |
| 502 | self, document: ResearchDocument, event: ResearchEvent |
| 503 | ) -> None: |
| 504 | """Persist normalized MCP metadata without provider-text reclassification.""" |
| 505 | if ( |
| 506 | document.discovery_provider != "YAHOO_FINANCE_MCP" |
| 507 | or event.instrument_id != document.instrument_id |
| 508 | or event.company_id != document.company_id |
| 509 | or event.source_url != document.canonical_url |
| 510 | ): |
| 511 | raise ValueError("INVALID_EXTERNAL_MCP_EVIDENCE") |
| 512 | accepted = await self._apply_prepared_ingested_document_async( |
| 513 | document, document_status=DocumentStatus.PROCESSED |
| 514 | ) |
| 515 | if accepted.status == DocumentStatus.DUPLICATE and accepted.duplicate_of_document_id: |
| 516 | event.source_document_id = accepted.duplicate_of_document_id |
| 517 | await self._apply_ingested_event_async(event, source_mode=SourceMode.REAL) |
| 518 | |
| 519 | def _record_persisted_shareholding_snapshot(self, snapshot: ShareholdingSnapshot, created: bool) -> None: |
| 520 | if created: |
| 521 | self.shareholding_snapshots[snapshot.id] = snapshot |
| 522 | logger.info("shareholding_snapshot_persisted globalInstrumentId=%s periodEnd=%s values=%s outcome=SUCCESS", |
| 523 | snapshot.instrument_id, snapshot.period_end.date().isoformat(), len(snapshot.values)) |
| 524 | |
| 525 | def summary(self, instrument_id: UUID, *, allow_demo: bool = True) -> ResearchSummary: |
| 526 | profile = self.profile(instrument_id) |
| 527 | real_events = self.events_for(instrument_id, source_mode=SourceMode.REAL) |
| 528 | real_documents = self.documents_for(instrument_id, source_mode=SourceMode.REAL) |
| 529 | if real_events or real_documents: |
| 530 | events = real_events |
| 531 | documents = real_documents |
| 532 | data_freshness = "REAL" |
| 533 | demo = False |
| 534 | elif allow_demo and self.settings.research_demo_enabled: |
| 535 | events = self.events_for(instrument_id, source_mode=SourceMode.DEMO) |
| 536 | documents = self.documents_for(instrument_id, source_mode=SourceMode.DEMO) |
| 537 | data_freshness = "DEMO_FALLBACK" if self.settings.research_live_enabled else "DEMO" |
| 538 | demo = True |
| 539 | else: |
| 540 | events = [] |
| 541 | documents = [] |
| 542 | data_freshness = "SOURCE_UNAVAILABLE" if self.last_live_error.get(instrument_id) else "UNAVAILABLE" |
| 543 | demo = False |
| 544 | score = self._scorer.score(instrument_id, events) |
| 545 | source_mix: dict[str, int] = {} |
| 546 | for document in documents: |
| 547 | key = str(document.source_type) |
| 548 | source_mix[key] = source_mix.get(key, 0) + 1 |
| 549 | shareholding = self.shareholding_for(instrument_id) |
| 550 | financial_facts = self.financial_facts_for(instrument_id) |
| 551 | return ResearchSummary( |
| 552 | profile=profile, |
| 553 | catalyst_score=score, |
| 554 | recent_events=events[:10], |
| 555 | documents=documents[:10], |
| 556 | last_refresh_at=self.last_refresh.get(instrument_id), |
| 557 | data_freshness=data_freshness, |
| 558 | demo=demo, |
| 559 | source_mix=source_mix, |
| 560 | shareholding_snapshots=shareholding, |
| 561 | shareholding_freshness="REAL" if shareholding else "UNAVAILABLE", |
| 562 | latest_quarterly_result=latest_quarterly_result_from_facts(financial_facts), |
| 563 | financial_result_history=financial_result_history_from_facts(financial_facts), |
| 564 | balance_sheet_history=financial_statement_history_from_facts( |
| 565 | financial_facts, period_type={"AS_AT", "QUARTERLY", "ANNUAL"}, metrics={ |
| 566 | "total_assets", "total_liabilities", "total_equity", "equity", |
| 567 | "cash_and_cash_equivalents", "cash_and_equivalents", "total_debt", |
| 568 | "debt_or_borrowings", "current_assets", "current_liabilities", |
| 569 | }, |
| 570 | ), |
| 571 | cash_flow_history=financial_statement_history_from_facts( |
| 572 | financial_facts, period_type={"QUARTERLY", "ANNUAL"}, metrics={ |
| 573 | "operating_cash_flow", "investing_cash_flow", "financing_cash_flow", |
| 574 | "cash_flow_from_operating_activities", "cash_flow_from_investing_activities", |
| 575 | "cash_flow_from_financing_activities", |
| 576 | }, |
| 577 | ), |
| 578 | ) |
| 579 | |
| 580 | def persisted_canonical_read_model_score(self, instrument_id: UUID): |
| 581 | """Project the existing canonical score from durable real research events. |
| 582 | |
| 583 | This intentionally does not require a process-local profile: callers that |
| 584 | already have an authoritative global instrument ID can read the same |
| 585 | scorer input used by ``summary`` without restoring or creating a profile. |
| 586 | No events means there is no usable persisted research read model. |
| 587 | """ |
| 588 | events = self.events_for(instrument_id, source_mode=SourceMode.REAL) |
| 589 | if not events: |
| 590 | return None |
| 591 | return canonical_read_model_score(self._scorer.score(instrument_id, events)) |
| 592 | |
| 593 | def ingest_fixture( |
| 594 | self, |
| 595 | *, |
| 596 | original_url: str, |
| 597 | source_type: SourceType, |
| 598 | source_name: str, |
| 599 | publisher: str, |
| 600 | content_type: str, |
| 601 | body: str, |
| 602 | reliability: ReliabilityLevel, |
| 603 | published_at: datetime | None = None, |
| 604 | source_mode: SourceMode = SourceMode.DEMO, |
| 605 | source_classification: SourceClassification = SourceClassification.OTHER, |
| 606 | discovered_at: datetime | None = None, |
| 607 | discovery_provider: str | None = None, |
| 608 | expected_profile: CompanyResearchProfile | None = None, |
| 609 | document_status: DocumentStatus = DocumentStatus.PARSED, |
| 610 | allow_empty_content: bool = False, |
| 611 | _trusted_profile_identity: object | None = None, |
| 612 | document_subtype: DocumentSubtype | None = None, |
| 613 | _metadata_only_nse_financial_result: bool = False, |
| 614 | ) -> ResearchDocument: |
| 615 | document = self._prepare_ingested_document( |
| 616 | original_url=original_url, source_type=source_type, source_name=source_name, publisher=publisher, |
| 617 | content_type=content_type, body=body, reliability=reliability, published_at=published_at, |
| 618 | source_mode=source_mode, source_classification=source_classification, discovered_at=discovered_at, |
| 619 | discovery_provider=discovery_provider, expected_profile=expected_profile, |
| 620 | document_status=document_status, allow_empty_content=allow_empty_content, |
| 621 | trusted_profile_identity=_trusted_profile_identity, |
| 622 | document_subtype=document_subtype, |
| 623 | ) |
| 624 | return self._apply_prepared_ingested_document( |
| 625 | document, |
| 626 | document_status=document_status, |
| 627 | metadata_only_nse_financial_result=_metadata_only_nse_financial_result, |
| 628 | ) |
| 629 | |
| 630 | async def _apply_prepared_ingested_document_async( |
| 631 | self, |
| 632 | document: ResearchDocument, |
| 633 | *, |
| 634 | document_status: DocumentStatus, |
| 635 | metadata_only_nse_financial_result: bool = False, |
| 636 | ) -> ResearchDocument: |
| 637 | duplicate = self._deduplicator.add(document) |
| 638 | if duplicate: |
| 639 | document.status = DocumentStatus.DUPLICATE |
| 640 | document.duplicate_of_document_id = duplicate.document_id |
| 641 | return document |
| 642 | if document_status != DocumentStatus.FAILED: |
| 643 | document.status = DocumentStatus.PROCESSED |
| 644 | self.documents[document.document_id] = document |
| 645 | if document.source_mode == SourceMode.REAL: |
| 646 | started = time.monotonic() |
| 647 | try: |
| 648 | await self._run_blocking_persistence( |
| 649 | self._persist_ingested_document, |
| 650 | document, |
| 651 | nse_financial_result=metadata_only_nse_financial_result, |
| 652 | ) |
| 653 | except Exception as exc: |
| 654 | self.documents.pop(document.document_id, None) |
| 655 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=DOCUMENT_PERSIST elapsedMs=%s outcome=FAILED", document.instrument_id, document.document_id, _elapsed_ms(started)) |
| 656 | raise FetchError("DOCUMENT_PERSIST_FAILED") from exc |
| 657 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=DOCUMENT_PERSIST elapsedMs=%s outcome=SUCCESS", document.instrument_id, document.document_id, _elapsed_ms(started)) |
| 658 | if (document.instrument_id and document.source_mode == SourceMode.REAL and document_status != DocumentStatus.FAILED |
| 659 | and document.source_classification in {SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY}): |
| 660 | started = time.monotonic() |
| 661 | try: |
| 662 | parsed, _ = await self._run_blocking_persistence(self._persist_official_financial_facts, document) |
| 663 | except Exception: |
| 664 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=FINANCIAL_FACTS elapsedMs=%s outcome=FAILED", document.instrument_id, document.document_id, _elapsed_ms(started)) |
| 665 | raise |
| 666 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=FINANCIAL_FACTS elapsedMs=%s outcome=%s", document.instrument_id, document.document_id, _elapsed_ms(started), "SUCCESS" if parsed else "SKIPPED") |
| 667 | extraction_started = time.monotonic() |
| 668 | try: |
| 669 | candidates = await asyncio.to_thread(self._extract_ingested_event_candidates, document, document_status=document_status) |
| 670 | except Exception: |
| 671 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=EVENT_EXTRACT elapsedMs=%s outcome=FAILED", document.instrument_id, document.document_id, _elapsed_ms(extraction_started)) |
| 672 | raise |
| 673 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=EVENT_EXTRACT elapsedMs=%s outcome=%s", document.instrument_id, document.document_id, _elapsed_ms(extraction_started), "SUCCESS" if candidates else "SKIPPED") |
| 674 | self.platform_events.append(document_event("research.document.processed", document)) |
| 675 | for event in candidates: |
| 676 | await self._apply_ingested_event_async(event, source_mode=document.source_mode) |
| 677 | return await self._continue_ingested_document_after_events_async(document) |
| 678 | |
| 679 | def _apply_prepared_ingested_document( |
| 680 | self, |
| 681 | document: ResearchDocument, |
| 682 | *, |
| 683 | document_status: DocumentStatus, |
| 684 | metadata_only_nse_financial_result: bool = False, |
| 685 | ) -> ResearchDocument: |
| 686 | """Apply a prepared document using the legacy synchronous semantics.""" |
| 687 | duplicate = self._deduplicator.add(document) |
| 688 | if duplicate: |
| 689 | document.status = DocumentStatus.DUPLICATE |
| 690 | document.duplicate_of_document_id = duplicate.document_id |
| 691 | return document |
| 692 | # Reuse eligibility is durable. Persist the successful terminal state, |
| 693 | # rather than leaving a pre-processing status in storage and only |
| 694 | # changing the in-memory object afterwards. |
| 695 | if document_status != DocumentStatus.FAILED: |
| 696 | document.status = DocumentStatus.PROCESSED |
| 697 | self.documents[document.document_id] = document |
| 698 | if document.source_mode == SourceMode.REAL: |
| 699 | try: |
| 700 | self._persist_ingested_document( |
| 701 | document, |
| 702 | nse_financial_result=metadata_only_nse_financial_result, |
| 703 | ) |
| 704 | except Exception as exc: |
| 705 | self.documents.pop(document.document_id, None) |
| 706 | raise FetchError("DOCUMENT_PERSIST_FAILED") from exc |
| 707 | return self._continue_ingested_document_after_persistence(document, document_status=document_status) |
| 708 | |
| 709 | def _continue_ingested_document_after_persistence(self, document: ResearchDocument, *, document_status: DocumentStatus, financial_processed: bool = False, event_candidates: list[ResearchEvent] | None = None) -> ResearchDocument: |
| 710 | if (not financial_processed and document.instrument_id and document.source_mode == SourceMode.REAL and document_status != DocumentStatus.FAILED |
| 711 | and document.source_classification in {SourceClassification.EXCHANGE, SourceClassification.REGULATORY, SourceClassification.OFFICIAL_COMPANY}): |
| 712 | self._persist_official_financial_facts(document) |
| 713 | self.platform_events.append(document_event("research.document.processed", document)) |
| 714 | for event in (event_candidates if event_candidates is not None else self._extract_ingested_event_candidates(document, document_status=document_status)): |
| 715 | self._apply_ingested_event(event, source_mode=document.source_mode) |
| 716 | return self._continue_ingested_document_after_events(document) |
| 717 | |
| 718 | def _continue_ingested_document_after_events(self, document: ResearchDocument) -> ResearchDocument: |
| 719 | if document.instrument_id: |
| 720 | snapshot = parse_official_shareholding(document) |
| 721 | if snapshot is not None: |
| 722 | self.persist_shareholding_snapshot(snapshot) |
| 723 | self.last_refresh[document.instrument_id] = datetime.now(timezone.utc) |
| 724 | self.platform_events.append(company_updated(document.instrument_id)) |
| 725 | return document |
| 726 | |
| 727 | async def _continue_ingested_document_after_events_async(self, document: ResearchDocument) -> ResearchDocument: |
| 728 | snapshot = await asyncio.to_thread(parse_official_shareholding, document) if document.instrument_id else None |
| 729 | if snapshot is not None: |
| 730 | await self._persist_shareholding_snapshot_async(snapshot) |
| 731 | if document.instrument_id: |
| 732 | self.last_refresh[document.instrument_id] = datetime.now(timezone.utc) |
| 733 | self.platform_events.append(company_updated(document.instrument_id)) |
| 734 | return document |
| 735 | |
| 736 | def _apply_ingested_event(self, event: ResearchEvent, *, source_mode: SourceMode) -> None: |
| 737 | event_key = _event_key(event) |
| 738 | if event_key in self._event_keys: |
| 739 | return |
| 740 | self._event_keys.add(event_key) |
| 741 | self.events[event.event_id] = event |
| 742 | if source_mode == SourceMode.REAL: |
| 743 | self._persist_ingested_event(event) |
| 744 | self.platform_events.append(research_event_extracted(event)) |
| 745 | |
| 746 | async def _apply_ingested_event_async(self, event: ResearchEvent, *, source_mode: SourceMode) -> None: |
| 747 | event_key = _event_key(event) |
| 748 | if event_key in self._event_keys: |
| 749 | return |
| 750 | self._event_keys.add(event_key) |
| 751 | self.events[event.event_id] = event |
| 752 | if source_mode == SourceMode.REAL: |
| 753 | started = time.monotonic() |
| 754 | try: |
| 755 | await self._run_blocking_persistence(self._persist_ingested_event, event) |
| 756 | except Exception: |
| 757 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s eventId=%s stage=EVENT_PERSIST elapsedMs=%s outcome=FAILED", event.instrument_id, event.source_document_id, event.event_id, _elapsed_ms(started)) |
| 758 | raise |
| 759 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s eventId=%s stage=EVENT_PERSIST elapsedMs=%s outcome=SUCCESS", event.instrument_id, event.source_document_id, event.event_id, _elapsed_ms(started)) |
| 760 | self.platform_events.append(research_event_extracted(event)) |
| 761 | |
| 762 | def _persist_ingested_event(self, event: ResearchEvent) -> None: |
| 763 | """Durably store an accepted event without touching repository state.""" |
| 764 | self._persistence.upsert_event(event) |
| 765 | |
| 766 | def _extract_ingested_event_candidates(self, document: ResearchDocument, *, document_status: DocumentStatus) -> list[ResearchEvent]: |
| 767 | if document_status == DocumentStatus.FAILED: |
| 768 | return [] |
| 769 | candidates = self._extractor.extract(document) |
| 770 | for event in candidates: |
| 771 | event.source_mode = document.source_mode |
| 772 | event.source_classification = document.source_classification |
| 773 | event.published_at = document.published_at |
| 774 | event.retrieved_at = document.retrieved_at |
| 775 | event.independence_key = document.source_independence_key |
| 776 | event.supporting_sources = [_evidence_source(document)] |
| 777 | return candidates |
| 778 | |
| 779 | def _persist_ingested_document( |
| 780 | self, |
| 781 | document: ResearchDocument, |
| 782 | *, |
| 783 | nse_financial_result: bool = False, |
| 784 | ) -> None: |
| 785 | """Durably store a non-duplicate prepared document.""" |
| 786 | persisted = self._metadata_only_nse_quarterly_document( |
| 787 | document, |
| 788 | nse_financial_result=nse_financial_result, |
| 789 | ) |
| 790 | self._persistence.upsert_document(persisted) |
| 791 | |
| 792 | def _metadata_only_nse_quarterly_document( |
| 793 | self, |
| 794 | document: ResearchDocument, |
| 795 | *, |
| 796 | nse_financial_result: bool = False, |
| 797 | ) -> ResearchDocument: |
| 798 | """Keep new official NSE quarterly PDF writes free of document content. |
| 799 | |
| 800 | PDF bytes are already transient in ``HttpResearchFetcher``. This |
| 801 | final persistence guard also omits extracted text when official |
| 802 | discovery classified the attachment as a financial result, or when |
| 803 | the parser proved that it contains normalized quarterly facts. The |
| 804 | in-memory document remains available to the existing extraction path |
| 805 | for the duration of this request. |
| 806 | """ |
| 807 | if ( |
| 808 | document.content_type != "application/pdf" |
| 809 | or document.discovery_provider != "NSE_OFFICIAL_API" |
| 810 | or document.source_classification != SourceClassification.EXCHANGE |
| 811 | or not _is_nse_official_document_url(document.canonical_url) |
| 812 | ): |
| 813 | return document |
| 814 | has_quarterly_facts = any( |
| 815 | fact.key.period_type == "QUARTERLY" |
| 816 | for fact in self._official_financial_fact_candidates(document) |
| 817 | ) |
| 818 | if not nse_financial_result and not has_quarterly_facts: |
| 819 | return document |
| 820 | logger.info( |
| 821 | "official_document_persist provider=NSE globalInstrumentId=%s documentId=%s " |
| 822 | "storage=URL_METADATA_AND_NORMALIZED_FACTS", |
| 823 | document.instrument_id, |
| 824 | document.document_id, |
| 825 | ) |
| 826 | return document.model_copy(update={"raw_text": None, "normalized_text": None}) |
| 827 | |
| 828 | def _prepare_ingested_document(self, *, original_url: str, source_type: SourceType, source_name: str, |
| 829 | publisher: str, content_type: str, body: str, reliability: ReliabilityLevel, |
| 830 | published_at: datetime | None, source_mode: SourceMode, |
| 831 | source_classification: SourceClassification, discovered_at: datetime | None, |
| 832 | discovery_provider: str | None, expected_profile: CompanyResearchProfile | None, |
| 833 | document_status: DocumentStatus, allow_empty_content: bool, |
| 834 | trusted_profile_identity: object | None, |
| 835 | document_subtype: DocumentSubtype | None = None) -> ResearchDocument: |
| 836 | canonical = canonicalize_url(original_url) |
| 837 | title, extracted = extract_text(body, content_type) |
| 838 | normalized = normalize_text(extracted or "") |
| 839 | if source_mode == SourceMode.REAL and not allow_empty_content: |
| 840 | if not normalized: |
| 841 | raise FetchError("CONTENT_EMPTY") |
| 842 | if len(normalized) < 40: |
| 843 | raise FetchError("CONTENT_TOO_SHORT") |
| 844 | parsed_published_at = published_at or extract_published_at(normalized) |
| 845 | resolution = self._resolver.resolve(title, normalized, canonical) |
| 846 | if expected_profile is not None and not allow_empty_content: |
| 847 | if trusted_profile_identity is _TRUSTED_NSE_PROFILE_IDENTITY: |
| 848 | resolution = resolution.model_copy(update={ |
| 849 | "instrument_id": expected_profile.instrument_id, |
| 850 | "company_id": expected_profile.company_id, |
| 851 | "confidence": 0.99, |
| 852 | }) |
| 853 | else: |
| 854 | self._validate_document_relevance(expected_profile, resolution.instrument_id, resolution.confidence) |
| 855 | if expected_profile is not None and allow_empty_content: |
| 856 | resolution = resolution.model_copy(update={ |
| 857 | "instrument_id": expected_profile.instrument_id, |
| 858 | "company_id": expected_profile.company_id, |
| 859 | "confidence": 0.95, |
| 860 | }) |
| 861 | document = ResearchDocument( |
| 862 | canonical_url=canonical, |
| 863 | original_url=original_url, |
| 864 | title=title, |
| 865 | source_type=source_type, |
| 866 | source_classification=source_classification, |
| 867 | source_name=source_name, |
| 868 | publisher=publisher, |
| 869 | published_at=parsed_published_at, |
| 870 | content_type=content_type, |
| 871 | document_type=detect_document_type(content_type, canonical), |
| 872 | document_subtype=document_subtype, |
| 873 | raw_text=body if len(body) < 20_000 else None, |
| 874 | normalized_text=normalized, |
| 875 | content_hash=content_hash(normalized or canonical), |
| 876 | instrument_id=resolution.instrument_id, |
| 877 | company_id=resolution.company_id, |
| 878 | status=document_status, |
| 879 | reliability_level=reliability, |
| 880 | entity_resolution_confidence=resolution.confidence, |
| 881 | source_mode=source_mode, |
| 882 | freshness=source_mode.value, |
| 883 | discovered_at=discovered_at, |
| 884 | discovery_provider=discovery_provider, |
| 885 | source_independence_key=content_hash(normalized or canonical), |
| 886 | ) |
| 887 | return document |
| 888 | |
| 889 | def _persist_official_financial_facts(self, document: ResearchDocument) -> tuple[bool, int]: |
| 890 | """Persist explicit NSE quarterly/annual income-statement facts.""" |
| 891 | if not document.instrument_id: |
| 892 | return False, 0 |
| 893 | facts = self._official_financial_fact_candidates(document) |
| 894 | if not facts: |
| 895 | return False, 0 |
| 896 | existing = {fact.key: fact for fact in self._persistence.load_financial_facts()} |
| 897 | written = 0 |
| 898 | for fact in facts: |
| 899 | prior = existing.get(fact.key) |
| 900 | same_document_correction = ( |
| 901 | prior is not None |
| 902 | and prior.source_tier == FactSourceTier.OFFICIAL_NSE |
| 903 | and prior.source_identity == str(document.document_id) |
| 904 | ) |
| 905 | if self._persistence.upsert_financial_fact(fact, allow_same_tier_correction=same_document_correction): |
| 906 | written += 1 |
| 907 | return True, written |
| 908 | |
| 909 | def _reconcile_persisted_official_financial_document(self, document: ResearchDocument) -> tuple[bool, int]: |
| 910 | """Atomically reconcile a known trusted document without retrieval. |
| 911 | |
| 912 | This is deliberately explicit: normal reads and ordinary completeness |
| 913 | repair retain their existing missing-fact selection behavior. |
| 914 | """ |
| 915 | facts = self._official_financial_fact_candidates(document) |
| 916 | if not facts: |
| 917 | return False, 0 |
| 918 | return True, self._persistence.reconcile_financial_facts_for_source( |
| 919 | document.instrument_id, |
| 920 | str(document.document_id), |
| 921 | facts, |
| 922 | ) |
| 923 | |
| 924 | @staticmethod |
| 925 | def _official_financial_fact_candidates(document: ResearchDocument) -> list[FinancialFact]: |
| 926 | if not document.instrument_id: |
| 927 | return [] |
| 928 | periods = [ |
| 929 | *parsed_nse_income_statement_periods([document]), |
| 930 | *parsed_nse_balance_sheet_periods([document]), |
| 931 | *parsed_nse_cash_flow_periods([document]), |
| 932 | ] |
| 933 | return [ |
| 934 | FinancialFact( |
| 935 | FinancialFactKey(document.instrument_id, metric, period.period_end, period.period_type, period.reporting_basis), |
| 936 | value, |
| 937 | FactSourceTier.OFFICIAL_NSE, |
| 938 | "NSE", |
| 939 | str(document.document_id), |
| 940 | SourceMode.REAL, |
| 941 | ) |
| 942 | for period in periods |
| 943 | for metric, value in period.metrics |
| 944 | ] |
| 945 | |
| 946 | def _validate_document_relevance( |
| 947 | self, |
| 948 | profile: CompanyResearchProfile, |
| 949 | instrument_id: UUID | None, |
| 950 | confidence: float, |
| 951 | ) -> None: |
| 952 | if instrument_id != profile.instrument_id or confidence < 0.30: |
| 953 | raise FetchError("COMPANY_RELEVANCE_FAILED") |
| 954 | |
| 955 | async def refresh( |
| 956 | self, |
| 957 | instrument_id: UUID, |
| 958 | correlation_id: str | None = None, |
| 959 | *, |
| 960 | allow_demo: bool = True, |
| 961 | pre_resolved_categories: set[str] | None = None, |
| 962 | ) -> ResearchSummary: |
| 963 | existing = self._instrument_refresh_flights.get(instrument_id) |
| 964 | if existing is not None: |
| 965 | logger.info( |
| 966 | "research_refresh_gate globalInstrumentId=%s category=ALL outcome=REUSED_IN_FLIGHT", |
| 967 | instrument_id, |
| 968 | ) |
| 969 | return await asyncio.shield(existing) |
| 970 | |
| 971 | task = asyncio.create_task( |
| 972 | self._refresh_once( |
| 973 | instrument_id, |
| 974 | correlation_id=correlation_id, |
| 975 | allow_demo=allow_demo, |
| 976 | pre_resolved_categories=pre_resolved_categories or set(), |
| 977 | ) |
| 978 | ) |
| 979 | self._instrument_refresh_flights[instrument_id] = task |
| 980 | |
| 981 | def cleanup(completed: asyncio.Task[ResearchSummary]) -> None: |
| 982 | if self._instrument_refresh_flights.get(instrument_id) is completed: |
| 983 | self._instrument_refresh_flights.pop(instrument_id, None) |
| 984 | |
| 985 | task.add_done_callback(cleanup) |
| 986 | return await asyncio.shield(task) |
| 987 | |
| 988 | async def refresh_targeted_categories( |
| 989 | self, |
| 990 | instrument_id: UUID, |
| 991 | categories: set[str], |
| 992 | *, |
| 993 | correlation_id: str | None = None, |
| 994 | allow_demo: bool = True, |
| 995 | ) -> ResearchSummary: |
| 996 | """Run only planner-selected legacy capabilities under the existing flight. |
| 997 | |
| 998 | The public refresh method above retains its compatibility behavior. |
| 999 | This boundary deliberately canonicalizes and filters categories before |
| 1000 | reaching discovery so a readiness ensure cannot widen into an ALL |
| 1001 | refresh. |
| 1002 | """ |
| 1003 | selected = {_canonical_refresh_category(value) for value in categories if str(value).strip()} |
| 1004 | if not selected: |
| 1005 | return self.summary(instrument_id, allow_demo=allow_demo) |
| 1006 | existing = self._instrument_refresh_flights.get(instrument_id) |
| 1007 | if existing is not None: |
| 1008 | logger.info( |
| 1009 | "research_refresh_gate globalInstrumentId=%s categories=%s outcome=REUSED_IN_FLIGHT", |
| 1010 | instrument_id, |
| 1011 | sorted(selected), |
| 1012 | ) |
| 1013 | return await asyncio.shield(existing) |
| 1014 | task = asyncio.create_task( |
| 1015 | self._refresh_targeted_categories_once( |
| 1016 | instrument_id, |
| 1017 | selected, |
| 1018 | correlation_id=correlation_id, |
| 1019 | allow_demo=allow_demo, |
| 1020 | ) |
| 1021 | ) |
| 1022 | self._instrument_refresh_flights[instrument_id] = task |
| 1023 | |
| 1024 | def cleanup(completed: asyncio.Task[ResearchSummary]) -> None: |
| 1025 | if self._instrument_refresh_flights.get(instrument_id) is completed: |
| 1026 | self._instrument_refresh_flights.pop(instrument_id, None) |
| 1027 | |
| 1028 | task.add_done_callback(cleanup) |
| 1029 | # The synchronous readiness request owns a newly-created targeted |
| 1030 | # refresh. Propagate its budget cancellation into provider/search work. |
| 1031 | # A refresh started by another caller remains shielded in the branch |
| 1032 | # above and is still managed by its original owner. |
| 1033 | return await task |
| 1034 | |
| 1035 | async def _refresh_targeted_categories_once( |
| 1036 | self, |
| 1037 | instrument_id: UUID, |
| 1038 | categories: set[str], |
| 1039 | *, |
| 1040 | correlation_id: str | None, |
| 1041 | allow_demo: bool, |
| 1042 | ) -> ResearchSummary: |
| 1043 | profile = self.profile(instrument_id) |
| 1044 | run = await self._run_blocking_persistence( |
| 1045 | self._persistence.start_refresh_run, |
| 1046 | instrument_id=profile.instrument_id, |
| 1047 | company_id=profile.company_id, |
| 1048 | correlation_id=correlation_id, |
| 1049 | mode="TARGETED_LIVE" if self.settings.research_live_enabled else "TARGETED_DEMO", |
| 1050 | ) |
| 1051 | documents_before = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1052 | events_before = len(self.events_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1053 | try: |
| 1054 | if self.settings.research_live_enabled: |
| 1055 | await self._refresh_live( |
| 1056 | instrument_id, |
| 1057 | set(), |
| 1058 | force=True, |
| 1059 | requested_categories=categories, |
| 1060 | ) |
| 1061 | except asyncio.CancelledError: |
| 1062 | await self._run_blocking_persistence( |
| 1063 | self._persistence.complete_refresh_run, |
| 1064 | run, |
| 1065 | status="FAILED", |
| 1066 | documents_discovered=0, |
| 1067 | documents_accepted=0, |
| 1068 | events_extracted=0, |
| 1069 | events_created=0, |
| 1070 | events_updated=0, |
| 1071 | deduplicated_count=0, |
| 1072 | safe_error_code="ACQUISITION_CANCELLED", |
| 1073 | safe_error_message="Synchronous readiness execution budget exhausted", |
| 1074 | ) |
| 1075 | logger.info( |
| 1076 | "research_targeted_ensure_cancelled globalInstrumentId=%s categories=%s", |
| 1077 | instrument_id, |
| 1078 | sorted(categories), |
| 1079 | ) |
| 1080 | raise |
| 1081 | except Exception as exc: |
| 1082 | await self._run_blocking_persistence( |
| 1083 | self._persistence.complete_refresh_run, |
| 1084 | run, |
| 1085 | status="FAILED", |
| 1086 | documents_discovered=0, |
| 1087 | documents_accepted=0, |
| 1088 | events_extracted=0, |
| 1089 | events_created=0, |
| 1090 | events_updated=0, |
| 1091 | deduplicated_count=0, |
| 1092 | safe_error_code=type(exc).__name__, |
| 1093 | safe_error_message=str(exc)[:500], |
| 1094 | ) |
| 1095 | raise |
| 1096 | self.last_refresh[instrument_id] = datetime.now(timezone.utc) |
| 1097 | documents_after = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1098 | events_after = len(self.events_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1099 | await self._run_blocking_persistence( |
| 1100 | self._persistence.complete_refresh_run, |
| 1101 | run, |
| 1102 | status="COMPLETED", |
| 1103 | documents_discovered=max(documents_after - documents_before, 0), |
| 1104 | documents_accepted=max(documents_after - documents_before, 0), |
| 1105 | events_extracted=max(events_after - events_before, 0), |
| 1106 | events_created=max(events_after - events_before, 0), |
| 1107 | events_updated=0, |
| 1108 | deduplicated_count=0, |
| 1109 | ) |
| 1110 | logger.info( |
| 1111 | "research_targeted_ensure_complete globalInstrumentId=%s categories=%s", |
| 1112 | instrument_id, |
| 1113 | sorted(categories), |
| 1114 | ) |
| 1115 | return self.summary(instrument_id, allow_demo=allow_demo) |
| 1116 | |
| 1117 | async def _refresh_once( |
| 1118 | self, |
| 1119 | instrument_id: UUID, |
| 1120 | *, |
| 1121 | correlation_id: str | None, |
| 1122 | allow_demo: bool, |
| 1123 | pre_resolved_categories: set[str], |
| 1124 | ) -> ResearchSummary: |
| 1125 | profile = self.profile(instrument_id) |
| 1126 | gate = self._instrument_refresh_gate(profile, pre_resolved_categories, datetime.now(timezone.utc)) |
| 1127 | if not gate.requires_provider_work: |
| 1128 | # A fresh category gate is intentionally about provider work, not |
| 1129 | # about whether canonical facts have already been materialised. |
| 1130 | # Let the live boundary perform its persisted-official-document |
| 1131 | # repair check before returning the existing fresh summary. |
| 1132 | if self.settings.research_live_enabled: |
| 1133 | await self._refresh_live(instrument_id, pre_resolved_categories) |
| 1134 | logger.info( |
| 1135 | "research_refresh_gate globalInstrumentId=%s category=ALL outcome=REUSE_FRESH state=%s", |
| 1136 | instrument_id, |
| 1137 | gate.state, |
| 1138 | ) |
| 1139 | return self.summary(instrument_id, allow_demo=allow_demo) |
| 1140 | logger.info( |
| 1141 | "research_refresh_gate globalInstrumentId=%s category=ALL outcome=TARGETED_REFRESH reason=%s missingCategories=%s", |
| 1142 | instrument_id, |
| 1143 | gate.state, |
| 1144 | sorted(gate.missing_categories), |
| 1145 | ) |
| 1146 | run = await self._run_blocking_persistence(self._persistence.start_refresh_run, |
| 1147 | instrument_id=profile.instrument_id, |
| 1148 | company_id=profile.company_id, |
| 1149 | correlation_id=correlation_id, |
| 1150 | mode="LIVE" if self.settings.research_live_enabled else "DEMO", |
| 1151 | ) |
| 1152 | documents_before = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1153 | events_before = len(self.events_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1154 | if self.settings.research_live_enabled: |
| 1155 | try: |
| 1156 | await self._refresh_live(instrument_id, pre_resolved_categories) |
| 1157 | except Exception as exc: |
| 1158 | await self._run_blocking_persistence(self._persistence.complete_refresh_run, |
| 1159 | run, |
| 1160 | status="FAILED", |
| 1161 | documents_discovered=0, |
| 1162 | documents_accepted=0, |
| 1163 | events_extracted=0, |
| 1164 | events_created=0, |
| 1165 | events_updated=0, |
| 1166 | deduplicated_count=0, |
| 1167 | safe_error_code=type(exc).__name__, |
| 1168 | safe_error_message=str(exc)[:500], |
| 1169 | ) |
| 1170 | raise |
| 1171 | self.last_refresh[instrument_id] = datetime.now(timezone.utc) |
| 1172 | documents_after = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1173 | events_after = len(self.events_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1174 | await self._run_blocking_persistence(self._persistence.complete_refresh_run, |
| 1175 | run, |
| 1176 | status="COMPLETED", |
| 1177 | documents_discovered=max(documents_after - documents_before, 0), |
| 1178 | documents_accepted=max(documents_after - documents_before, 0), |
| 1179 | events_extracted=max(events_after - events_before, 0), |
| 1180 | events_created=max(events_after - events_before, 0), |
| 1181 | events_updated=0, |
| 1182 | deduplicated_count=0, |
| 1183 | ) |
| 1184 | return self.summary(instrument_id, allow_demo=allow_demo) |
| 1185 | |
| 1186 | async def refresh_etf(self, instrument_id: UUID, correlation_id: str | None = None) -> EtfResearchProfile: |
| 1187 | profile = self.etf_profile(instrument_id) |
| 1188 | run = await self._run_blocking_persistence(self._persistence.start_refresh_run, |
| 1189 | instrument_id=profile.instrument_id, |
| 1190 | company_id=profile.fund_id, |
| 1191 | correlation_id=correlation_id, |
| 1192 | mode="LIVE" if self.settings.research_live_enabled else "DEMO", |
| 1193 | ) |
| 1194 | documents_before = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1195 | if self.settings.research_live_enabled and self.settings.research_search_enabled: |
| 1196 | await self._refresh_etf_search_discovery(profile) |
| 1197 | self.last_refresh[instrument_id] = datetime.now(timezone.utc) |
| 1198 | documents_after = len(self.documents_for(instrument_id, source_mode=SourceMode.REAL)) |
| 1199 | status = "COMPLETED" if documents_after > documents_before else "FAILED" |
| 1200 | await self._run_blocking_persistence(self._persistence.complete_refresh_run, |
| 1201 | run, |
| 1202 | status=status, |
| 1203 | documents_discovered=max(documents_after - documents_before, 0), |
| 1204 | documents_accepted=max(documents_after - documents_before, 0), |
| 1205 | events_extracted=0, |
| 1206 | events_created=0, |
| 1207 | events_updated=0, |
| 1208 | deduplicated_count=0, |
| 1209 | safe_error_code=self.last_live_error.get(instrument_id) if status == "FAILED" else None, |
| 1210 | ) |
| 1211 | return profile |
| 1212 | |
| 1213 | async def _refresh_live( |
| 1214 | self, |
| 1215 | instrument_id: UUID, |
| 1216 | pre_resolved_categories: set[str], |
| 1217 | *, |
| 1218 | force: bool = False, |
| 1219 | requested_categories: set[str] | None = None, |
| 1220 | ) -> None: |
| 1221 | sources = registered_sources_for(instrument_id) |
| 1222 | profile = self.profile(instrument_id) |
| 1223 | now = datetime.now(timezone.utc) |
| 1224 | gate = self._instrument_refresh_gate(profile, pre_resolved_categories, now, force=force) |
| 1225 | due_categories = set(gate.missing_categories) |
| 1226 | if requested_categories is not None: |
| 1227 | requested_categories = { |
| 1228 | _canonical_refresh_category(value) for value in requested_categories |
| 1229 | } |
| 1230 | due_categories.intersection_update(requested_categories) |
| 1231 | logger.info( |
| 1232 | "research_refresh_gate globalInstrumentId=%s outcome=DUE_CATEGORIES dueCategories=%s", |
| 1233 | instrument_id, |
| 1234 | sorted(due_categories), |
| 1235 | ) |
| 1236 | if requested_categories is None or "FINANCIAL_RESULTS" in requested_categories: |
| 1237 | await self._reconcile_incomplete_persisted_official_financial_facts(profile) |
| 1238 | shareholding_selected = ( |
| 1239 | requested_categories is None or "SHAREHOLDING_PATTERN" in requested_categories |
| 1240 | ) |
| 1241 | has_shareholding_reconciliation = shareholding_selected and ( |
| 1242 | gate.shareholding_backfill_needed or gate.shareholding_category_enrichment_needed |
| 1243 | ) |
| 1244 | if not due_categories and not has_shareholding_reconciliation: |
| 1245 | logger.info("research_refresh_gate globalInstrumentId=%s category=ALL outcome=REUSE_FRESH", instrument_id) |
| 1246 | return |
| 1247 | if not sources: |
| 1248 | self.last_live_error[instrument_id] = "SOURCE_UNAVAILABLE" |
| 1249 | for source in sources: |
| 1250 | if source.priority != 1: |
| 1251 | continue |
| 1252 | if not any( |
| 1253 | _canonical_refresh_category(category) in due_categories |
| 1254 | for category in source.categories |
| 1255 | ): |
| 1256 | continue |
| 1257 | try: |
| 1258 | await self._fetch_registered_source(profile, source) |
| 1259 | self.last_live_error.pop(instrument_id, None) |
| 1260 | except RestrictedFetchError: |
| 1261 | self.last_live_error[instrument_id] = "ACCESS_RESTRICTED" |
| 1262 | except (FetchError, ValueError) as exc: |
| 1263 | self.last_live_error[instrument_id] = f"SOURCE_UNAVAILABLE:{exc}" |
| 1264 | await self._refresh_targeted( |
| 1265 | profile, |
| 1266 | pre_resolved_categories, |
| 1267 | now=now, |
| 1268 | force=force, |
| 1269 | requested_categories=requested_categories, |
| 1270 | ) |
| 1271 | |
| 1272 | def _instrument_refresh_gate( |
| 1273 | self, |
| 1274 | profile: CompanyResearchProfile, |
| 1275 | pre_resolved_categories: set[str], |
| 1276 | now: datetime, |
| 1277 | *, |
| 1278 | force: bool = False, |
| 1279 | ) -> _InstrumentRefreshGate: |
| 1280 | structured_categories = { |
| 1281 | "FINANCIAL_RESULTS", "Ownership", "INSTITUTIONAL_ACTIVITY", "SHAREHOLDING_PATTERN", "VALUATION", |
| 1282 | "ORDERS_BACKLOG", "CONTRACTS", "CAPEX", "NEW_FACILITIES", "ACQUISITIONS", |
| 1283 | "CLIENTS", "GUIDANCE", "ANALYST_OPINION", "ANALYST_TARGETS", "Regulatory", |
| 1284 | } |
| 1285 | missing = self._missing_categories(profile, pre_resolved_categories, now, structured_categories, force=force) |
| 1286 | eligible_missing: set[str] = set() |
| 1287 | for category in missing: |
| 1288 | eligible, next_eligible = (True, None) if force else self._category_is_eligible_to_check(profile.instrument_id, category, now) |
| 1289 | if eligible: |
| 1290 | eligible_missing.add(category) |
| 1291 | else: |
| 1292 | logger.info( |
| 1293 | "research_refresh_gate globalInstrumentId=%s category=%s outcome=SKIP_NOT_DUE nextEligibleCheckAt=%s", |
| 1294 | profile.instrument_id, |
| 1295 | category, |
| 1296 | next_eligible.isoformat() if next_eligible else "NONE", |
| 1297 | ) |
| 1298 | missing = eligible_missing |
| 1299 | valid_shareholding_periods = len(self.shareholding_for(profile.instrument_id, limit=4)) |
| 1300 | shareholding_backfill_needed = ( |
| 1301 | self._category_is_fresh(profile.instrument_id, "SHAREHOLDING_PATTERN", now) |
| 1302 | and valid_shareholding_periods < 4 |
| 1303 | and _eligible_for_nse_shareholding_reconciliation(profile) |
| 1304 | ) |
| 1305 | shareholding_category_enrichment_needed = ( |
| 1306 | self._category_is_fresh(profile.instrument_id, "SHAREHOLDING_PATTERN", now) |
| 1307 | and _shareholding_xbrl_enrichment_needed(self.shareholding_for(profile.instrument_id, limit=4)) |
| 1308 | and _eligible_for_nse_shareholding_reconciliation(profile) |
| 1309 | ) |
| 1310 | if not missing and not shareholding_backfill_needed and not shareholding_category_enrichment_needed: |
| 1311 | state = "FRESH_AND_COMPLETE" |
| 1312 | elif not missing and (shareholding_backfill_needed or shareholding_category_enrichment_needed): |
| 1313 | state = "INCOMPLETE" |
| 1314 | elif ( |
| 1315 | missing == {"SHAREHOLDING_PATTERN"} |
| 1316 | and valid_shareholding_periods >= 4 |
| 1317 | and _eligible_for_nse_shareholding_reconciliation(profile) |
| 1318 | ): |
| 1319 | state = "NEEDS_LIGHTWEIGHT_CHECK" |
| 1320 | elif any(self._category_has_qualifying_evidence(profile.instrument_id, category) for category in missing): |
| 1321 | state = "STALE" |
| 1322 | else: |
| 1323 | state = "INCOMPLETE" |
| 1324 | return _InstrumentRefreshGate( |
| 1325 | missing_categories=missing, |
| 1326 | shareholding_backfill_needed=shareholding_backfill_needed, |
| 1327 | shareholding_category_enrichment_needed=shareholding_category_enrichment_needed, |
| 1328 | state=state, |
| 1329 | ) |
| 1330 | |
| 1331 | async def _refresh_targeted( |
| 1332 | self, |
| 1333 | profile: CompanyResearchProfile, |
| 1334 | pre_resolved_categories: set[str], |
| 1335 | *, |
| 1336 | now: datetime | None = None, |
| 1337 | force: bool = False, |
| 1338 | requested_categories: set[str] | None = None, |
| 1339 | ) -> None: |
| 1340 | now = now or datetime.now(timezone.utc) |
| 1341 | gate = self._instrument_refresh_gate(profile, pre_resolved_categories, now, force=force) |
| 1342 | missing = set(gate.missing_categories) |
| 1343 | if requested_categories is not None: |
| 1344 | requested_categories = { |
| 1345 | _canonical_refresh_category(value) for value in requested_categories |
| 1346 | } |
| 1347 | missing.intersection_update(requested_categories) |
| 1348 | due_categories = set(missing) |
| 1349 | shareholding_selected = ( |
| 1350 | requested_categories is None or "SHAREHOLDING_PATTERN" in requested_categories |
| 1351 | ) |
| 1352 | shareholding_backfill_needed = gate.shareholding_backfill_needed and shareholding_selected |
| 1353 | shareholding_category_enrichment_needed = ( |
| 1354 | gate.shareholding_category_enrichment_needed and shareholding_selected |
| 1355 | ) |
| 1356 | valid_shareholding_periods = len(self.shareholding_for(profile.instrument_id, limit=4)) |
| 1357 | shareholding_check_succeeded = False |
| 1358 | successful_check_categories: set[str] = set() |
| 1359 | logger.info( |
| 1360 | "research_refresh_gate globalInstrumentId=%s outcome=DUE_CATEGORIES dueCategories=%s", |
| 1361 | profile.instrument_id, sorted(due_categories), |
| 1362 | ) |
| 1363 | logger.info("research_missing_categories globalInstrumentId=%s categories=%s", profile.instrument_id, sorted(due_categories)) |
| 1364 | if shareholding_backfill_needed: |
| 1365 | logger.info( |
| 1366 | "shareholding_reconciliation provider=NSE globalInstrumentId=%s outcome=START reason=FRESH_BUT_INCOMPLETE validQuarterCount=%s targetQuarterCount=4", |
| 1367 | profile.instrument_id, |
| 1368 | valid_shareholding_periods, |
| 1369 | ) |
| 1370 | if shareholding_category_enrichment_needed: |
| 1371 | logger.info( |
| 1372 | "shareholding_category_enrichment provider=NSE globalInstrumentId=%s outcome=START reason=LEGACY_XBRL_PROVENANCE_MISSING validQuarterCount=%s", |
| 1373 | profile.instrument_id, |
| 1374 | valid_shareholding_periods, |
| 1375 | ) |
| 1376 | if not missing and not shareholding_backfill_needed and not shareholding_category_enrichment_needed: |
| 1377 | logger.info("official_discovery_gate globalInstrumentId=%s eligible=false reason=NO_MISSING_CATEGORIES", profile.instrument_id) |
| 1378 | return |
| 1379 | seen_urls = {doc.canonical_url for doc in self.documents_for(profile.instrument_id, source_mode=SourceMode.REAL)} |
| 1380 | # Resolve authoritative filings before broad research searches can |
| 1381 | # exhaust public-search engines. This is global-instrument research. |
| 1382 | official_due_categories = {"FINANCIAL_RESULTS", "SHAREHOLDING_PATTERN"} & due_categories |
| 1383 | eligible_official = ( |
| 1384 | profile.country.upper() in {"IN", "IND", "INDIA"} |
| 1385 | and profile.exchange.upper() in {"NSE", "XNSE"} |
| 1386 | and bool(official_due_categories) |
| 1387 | ) |
| 1388 | logger.info("official_discovery_gate globalInstrumentId=%s eligible=%s reason=%s", profile.instrument_id, eligible_official, "NSE_OFFICIAL_CATEGORY" if eligible_official else "PROFILE_OR_CATEGORY_INELIGIBLE") |
| 1389 | if eligible_official: |
| 1390 | try: |
| 1391 | # Let official discovery return known URLs too: the official fetch |
| 1392 | # loop can then explicitly reuse a durable global document, while |
| 1393 | # failed/scanned historical attempts remain eligible for retry. |
| 1394 | official_filings = await self._official_filing_discovery.discover(profile, official_due_categories, set()) |
| 1395 | except Exception as exc: |
| 1396 | official_filings = [] |
| 1397 | self.last_live_error[profile.instrument_id] = f"OFFICIAL_FILING_DISCOVERY_UNAVAILABLE:{type(exc).__name__}" |
| 1398 | # OfficialFilingDiscovery emits the terminal provider diagnostic. |
| 1399 | # Keep this boundary diagnostic for injected/legacy implementations. |
| 1400 | logger.warning("official_discovery_handled provider=NSE globalInstrumentId=%s status=FAILED reason=%s", profile.instrument_id, type(exc).__name__) |
| 1401 | if await self._fetch_official_filings(profile, official_filings, seen_urls): |
| 1402 | successful_check_categories.update({"FINANCIAL_RESULTS"} & official_due_categories) |
| 1403 | if ( |
| 1404 | "SHAREHOLDING_PATTERN" in due_categories |
| 1405 | or shareholding_backfill_needed |
| 1406 | or shareholding_category_enrichment_needed |
| 1407 | ) and _eligible_for_nse_shareholding_reconciliation(profile): |
| 1408 | try: |
| 1409 | # NSE publishes quarterly Regulation 31 data through its |
| 1410 | # dedicated shareholding feed, not necessarily as a corporate |
| 1411 | # announcement attachment. The provider returns only real, |
| 1412 | # source-labelled values and durable persistence is keyed by |
| 1413 | # this global instrument and the official filing identity. |
| 1414 | persisted = 0 |
| 1415 | discovered_snapshots = await self._official_shareholding_discovery.discover(profile) |
| 1416 | shareholding_check_succeeded = True |
| 1417 | successful_check_categories.add("SHAREHOLDING_PATTERN") |
| 1418 | snapshots_to_process = [ |
| 1419 | snapshot for snapshot in discovered_snapshots |
| 1420 | if self._shareholding_snapshot_needs_processing(snapshot) |
| 1421 | ] |
| 1422 | if not snapshots_to_process: |
| 1423 | logger.info( |
| 1424 | "research_refresh_gate globalInstrumentId=%s category=SHAREHOLDING_PATTERN outcome=LIGHTWEIGHT_CHECK_UNCHANGED latestSourceIdentity=%s", |
| 1425 | profile.instrument_id, |
| 1426 | discovered_snapshots[0].source_identity_key if discovered_snapshots else "NONE", |
| 1427 | ) |
| 1428 | for snapshot in snapshots_to_process: |
| 1429 | enriched_snapshot = await self._enrich_nse_shareholding_snapshot(snapshot) |
| 1430 | persisted += int(await self._persist_shareholding_snapshot_async(enriched_snapshot)) |
| 1431 | if shareholding_backfill_needed or shareholding_category_enrichment_needed: |
| 1432 | logger.info( |
| 1433 | "shareholding_reconciliation provider=NSE globalInstrumentId=%s outcome=COMPLETE persistedCount=%s validQuarterCountBefore=%s validQuarterCountAfter=%s reason=%s", |
| 1434 | profile.instrument_id, |
| 1435 | persisted, |
| 1436 | valid_shareholding_periods, |
| 1437 | len(self.shareholding_for(profile.instrument_id, limit=4)), |
| 1438 | "FRESH_BUT_INCOMPLETE" if shareholding_backfill_needed else "LEGACY_XBRL_PROVENANCE_MISSING", |
| 1439 | ) |
| 1440 | except SearchProviderError as exc: |
| 1441 | self.last_live_error[profile.instrument_id] = str(exc) |
| 1442 | except (FetchError, ValueError) as exc: |
| 1443 | self.last_live_error[profile.instrument_id] = f"NSE_SHAREHOLDING_UNAVAILABLE:{type(exc).__name__}" |
| 1444 | if due_categories: |
| 1445 | discovery_categories = _search_discovery_categories(due_categories) |
| 1446 | discovered = ( |
| 1447 | _profile_source_discovery(profile, discovery_categories, seen_urls) |
| 1448 | if self.settings.research_search_enabled and not registered_sources_for(profile.instrument_id) |
| 1449 | else [] |
| 1450 | ) |
| 1451 | discovered.extend(self._discovery.discover(profile, discovery_categories, seen_urls | {result.source.url for result in discovered})) |
| 1452 | fetched_source_ids: set[str] = set() |
| 1453 | for result in discovered: |
| 1454 | source = result.source |
| 1455 | if source.source_id in fetched_source_ids: |
| 1456 | continue |
| 1457 | fetched_source_ids.add(source.source_id) |
| 1458 | try: |
| 1459 | self._validate_registered_source(profile, source) |
| 1460 | await self._fetch_registered_source(profile, source) |
| 1461 | except (FetchError, RestrictedFetchError, ValueError) as exc: |
| 1462 | self.last_live_error[profile.instrument_id] = f"TARGETED_SOURCE_UNAVAILABLE:{source.source_id}:{exc}" |
| 1463 | except Exception: |
| 1464 | self.last_live_error[profile.instrument_id] = f"TARGETED_SOURCE_UNAVAILABLE:{source.source_id}:HTTP_FETCH_FAILED" |
| 1465 | if self.settings.research_search_enabled: |
| 1466 | self._mark_qualifying_categories_fresh( |
| 1467 | profile.instrument_id, |
| 1468 | _checked_categories(missing, shareholding_check_succeeded), |
| 1469 | now, |
| 1470 | ) |
| 1471 | # Keep the pre-discovery due set authoritative. Some refresh |
| 1472 | # categories (for example REGULATORY) are deliberately not scorer |
| 1473 | # buckets, so recomputing missing only from score coverage would |
| 1474 | # drop them here before their successful no-change check can be |
| 1475 | # recorded. Evidence that arrived during this refresh still |
| 1476 | # removes its category from fallback. |
| 1477 | search_categories = { |
| 1478 | category |
| 1479 | for category in due_categories |
| 1480 | if not self._category_has_qualifying_evidence(profile.instrument_id, category) |
| 1481 | } |
| 1482 | if search_categories: |
| 1483 | logger.info("search_fallback_start globalInstrumentId=%s missingCategories=%s", profile.instrument_id, sorted(search_categories)) |
| 1484 | refreshed_seen_urls = { |
| 1485 | doc.canonical_url for doc in self.documents_for(profile.instrument_id, source_mode=SourceMode.REAL) |
| 1486 | } |
| 1487 | if await self._refresh_search_discovery( |
| 1488 | profile, _search_discovery_categories(search_categories), refreshed_seen_urls |
| 1489 | ): |
| 1490 | successful_check_categories.update(search_categories) |
| 1491 | self._mark_qualifying_categories_fresh( |
| 1492 | profile.instrument_id, |
| 1493 | _checked_categories(missing, shareholding_check_succeeded), |
| 1494 | now, |
| 1495 | ) |
| 1496 | self._mark_successful_categories_checked(profile.instrument_id, successful_check_categories, now) |
| 1497 | |
| 1498 | async def _incomplete_persisted_official_financial_documents(self, profile: CompanyResearchProfile) -> list[ResearchDocument]: |
| 1499 | if (profile.country.upper() not in {"IN", "IND", "INDIA"} or profile.exchange.upper() not in {"NSE", "XNSE"} |
| 1500 | or not profile.provider_instrument_ids.get("NSE")): |
| 1501 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=INELIGIBLE_PROFILE", profile.instrument_id) |
| 1502 | return [] |
| 1503 | documents = [document for document in self.documents_for(profile.instrument_id, source_mode=SourceMode.REAL) |
| 1504 | if document.company_id == profile.company_id and document.source_classification == SourceClassification.EXCHANGE |
| 1505 | and document.source_type == SourceType.EXCHANGE_ANNOUNCEMENT and bool(document.normalized_text) |
| 1506 | and _is_nse_official_document_url(document.canonical_url)] |
| 1507 | if not documents: |
| 1508 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=NO_QUALIFYING_PERSISTED_DOCUMENT", profile.instrument_id) |
| 1509 | return [] |
| 1510 | def incomplete_window_documents(): |
| 1511 | # Source freshness and derived-fact completeness are deliberately |
| 1512 | # independent. Parse only the durable, trusted documents already |
| 1513 | # held for this instrument, then use their explicit periods to |
| 1514 | # define the rolling window. No calendar-derived target periods |
| 1515 | # are invented here. |
| 1516 | parsed_by_document = [ |
| 1517 | (document, parsed_nse_income_statement_periods([document])) |
| 1518 | for document in documents |
| 1519 | ] |
| 1520 | parsed_by_document = [item for item in parsed_by_document if item[1]] |
| 1521 | if not parsed_by_document: |
| 1522 | return None |
| 1523 | |
| 1524 | available_periods: dict[tuple[str, str | None], set[str]] = {} |
| 1525 | periods_by_document: dict[UUID, set[tuple[str, str | None, str]]] = {} |
| 1526 | for document, periods in parsed_by_document: |
| 1527 | document_periods: set[tuple[str, str | None, str]] = set() |
| 1528 | for period in periods: |
| 1529 | if period.period_type not in {"QUARTERLY", "ANNUAL"}: |
| 1530 | continue |
| 1531 | # A parsed result column is evidence of an explicitly |
| 1532 | # reported period even when one core field is absent. It |
| 1533 | # must not silently make the rolling window complete. |
| 1534 | if not dict(period.metrics): |
| 1535 | continue |
| 1536 | group = (period.period_type, period.reporting_basis) |
| 1537 | available_periods.setdefault(group, set()).add(period.period_end) |
| 1538 | document_periods.add((period.period_type, period.reporting_basis, period.period_end)) |
| 1539 | if document_periods: |
| 1540 | periods_by_document[document.document_id] = document_periods |
| 1541 | |
| 1542 | target_periods = { |
| 1543 | (period_type, reporting_basis, period_end) |
| 1544 | for (period_type, reporting_basis), periods in available_periods.items() |
| 1545 | for period_end in sorted(periods, reverse=True)[:4] |
| 1546 | } |
| 1547 | if not target_periods: |
| 1548 | return None |
| 1549 | |
| 1550 | facts = self._persistence.load_financial_facts() |
| 1551 | present_by_period: dict[tuple[str, str | None, str], set[str]] = {} |
| 1552 | for fact in facts: |
| 1553 | if ( |
| 1554 | fact.key.instrument_id != profile.instrument_id |
| 1555 | or fact.key.period_type not in {"QUARTERLY", "ANNUAL"} |
| 1556 | or fact.source_tier != FactSourceTier.OFFICIAL_NSE |
| 1557 | or fact.source_mode != SourceMode.REAL |
| 1558 | or not fact.key.period_end |
| 1559 | ): |
| 1560 | continue |
| 1561 | key = (fact.key.period_type, fact.key.reporting_basis, fact.key.period_end) |
| 1562 | present_by_period.setdefault(key, set()).add(fact.key.metric) |
| 1563 | |
| 1564 | incomplete = { |
| 1565 | key for key in target_periods |
| 1566 | if not {"revenue", "pat"} <= present_by_period.get(key, set()) |
| 1567 | } |
| 1568 | if not incomplete: |
| 1569 | return [], target_periods, incomplete |
| 1570 | selected = [ |
| 1571 | document for document, _periods in parsed_by_document |
| 1572 | if periods_by_document.get(document.document_id, set()) & incomplete |
| 1573 | ] |
| 1574 | return selected, target_periods, incomplete |
| 1575 | |
| 1576 | window = await self._run_blocking_persistence(incomplete_window_documents) |
| 1577 | if window is None: |
| 1578 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=NO_PARSEABLE_QUARTERLY_RESULT", profile.instrument_id) |
| 1579 | return [] |
| 1580 | selected, target_periods, incomplete = window |
| 1581 | if not selected: |
| 1582 | logger.info( |
| 1583 | "official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=ROLLING_WINDOW_COMPLETE targetPeriodCount=%s", |
| 1584 | profile.instrument_id, |
| 1585 | len(target_periods), |
| 1586 | ) |
| 1587 | return [] |
| 1588 | logger.info( |
| 1589 | "official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=REQUIRED targetPeriodCount=%s incompletePeriodCount=%s documentCount=%s", |
| 1590 | profile.instrument_id, |
| 1591 | len(target_periods), |
| 1592 | len(incomplete), |
| 1593 | len(selected), |
| 1594 | ) |
| 1595 | return selected |
| 1596 | |
| 1597 | async def _reconcile_incomplete_persisted_official_financial_facts( |
| 1598 | self, |
| 1599 | profile: CompanyResearchProfile, |
| 1600 | ) -> bool: |
| 1601 | documents = await self._incomplete_persisted_official_financial_documents(profile) |
| 1602 | for document in documents: |
| 1603 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=START documentId=%s", profile.instrument_id, document.document_id) |
| 1604 | parsed, written = await self._run_blocking_persistence(self._persist_official_financial_facts, document) |
| 1605 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s outcome=COMPLETE documentId=%s parserResult=%s factsWritten=%s", profile.instrument_id, document.document_id, "PARSED" if parsed else "NO_RESULT", written) |
| 1606 | return bool(documents) |
| 1607 | |
| 1608 | def _shareholding_snapshot_needs_processing(self, snapshot: ShareholdingSnapshot) -> bool: |
| 1609 | existing = next( |
| 1610 | ( |
| 1611 | item for item in self.shareholding_snapshots.values() |
| 1612 | if item.instrument_id == snapshot.instrument_id |
| 1613 | and item.source_provider == snapshot.source_provider |
| 1614 | and item.source_identity_key == snapshot.source_identity_key |
| 1615 | ), |
| 1616 | None, |
| 1617 | ) |
| 1618 | if existing is None: |
| 1619 | return True |
| 1620 | return not any((value.source_locator or "").startswith("nse-xbrl:") for value in existing.values) |
| 1621 | |
| 1622 | async def _enrich_nse_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> ShareholdingSnapshot: |
| 1623 | """Attach source-specific XBRL category facts without changing discovery identity.""" |
| 1624 | if snapshot.source_type != "NSE_SHAREHOLDING_XBRL": |
| 1625 | return snapshot |
| 1626 | existing = next((item for item in self.shareholding_snapshots.values() |
| 1627 | if item.instrument_id == snapshot.instrument_id |
| 1628 | and item.source_provider == snapshot.source_provider |
| 1629 | and item.source_identity_key == snapshot.source_identity_key), None) |
| 1630 | if existing and any((value.source_locator or "").startswith("nse-xbrl:") for value in existing.values): |
| 1631 | return snapshot |
| 1632 | try: |
| 1633 | fetch_xbrl = getattr(self._fetcher, "fetch_nse_shareholding_xbrl", None) |
| 1634 | result = await (fetch_xbrl(snapshot.source_url) if callable(fetch_xbrl) |
| 1635 | else self._fetcher.fetch(snapshot.source_url)) |
| 1636 | except (FetchError, RestrictedFetchError, ValueError) as exc: |
| 1637 | logger.warning( |
| 1638 | "shareholding_xbrl_fetch globalInstrumentId=%s host=%s path=%s outcome=UNAVAILABLE reason=%s", |
| 1639 | snapshot.instrument_id, |
| 1640 | (urlparse(snapshot.source_url).hostname or "").lower(), |
| 1641 | _safe_url_path(snapshot.source_url), |
| 1642 | str(exc)[:160], |
| 1643 | ) |
| 1644 | return snapshot |
| 1645 | values = await asyncio.to_thread(parse_nse_shareholding_xbrl, result.text) |
| 1646 | if not values: |
| 1647 | logger.info("shareholding_xbrl_parse globalInstrumentId=%s outcome=REJECTED reason=NO_SUPPORTED_EXPLICIT_VALUES", snapshot.instrument_id) |
| 1648 | return snapshot |
| 1649 | logger.info("shareholding_xbrl_parse globalInstrumentId=%s outcome=SUCCESS valueCount=%s", snapshot.instrument_id, len(values)) |
| 1650 | return snapshot.model_copy(update={"values": values}) |
| 1651 | |
| 1652 | async def _fetch_official_filings( |
| 1653 | self, |
| 1654 | profile: CompanyResearchProfile, |
| 1655 | filings: list[DiscoveryResult], |
| 1656 | seen_urls: set[str], |
| 1657 | ) -> bool: |
| 1658 | """Fetch a small, newest-first official filing set within an interactive budget.""" |
| 1659 | attempted = 0 |
| 1660 | completed_without_failure = True |
| 1661 | host_transport_failures: dict[str, int] = {} |
| 1662 | scheduled_filings = _fair_official_filing_order(filings) |
| 1663 | for filing_index, result in enumerate(scheduled_filings): |
| 1664 | source = result.source |
| 1665 | reusable = self._reusable_official_document(profile.instrument_id, source.url) |
| 1666 | if reusable is not None: |
| 1667 | if source.document_subtype and reusable.document_subtype is None: |
| 1668 | reusable.document_subtype = source.document_subtype |
| 1669 | await self._run_blocking_persistence( |
| 1670 | self._persist_ingested_document, |
| 1671 | reusable, |
| 1672 | nse_financial_result="FINANCIAL_RESULTS" in source.categories, |
| 1673 | ) |
| 1674 | seen_urls.add(reusable.canonical_url) |
| 1675 | await self._run_blocking_persistence(self._reconcile_reused_official_financial_facts, profile, source, reusable) |
| 1676 | logger.info( |
| 1677 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=REUSED reason=ALREADY_PERSISTED documentId=%s", |
| 1678 | profile.instrument_id, |
| 1679 | (urlparse(source.url).hostname or "").lower(), |
| 1680 | _safe_url_path(source.url), |
| 1681 | reusable.document_id, |
| 1682 | ) |
| 1683 | continue |
| 1684 | if attempted >= self.settings.research_official_document_max_attempts_per_refresh: |
| 1685 | logger.info( |
| 1686 | "official_document_fetch provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=ATTEMPT_BUDGET", |
| 1687 | profile.instrument_id, |
| 1688 | ) |
| 1689 | # Continue so a later durable reusable filing can still be |
| 1690 | # recorded without consuming network budget. |
| 1691 | continue |
| 1692 | host = (urlparse(source.url).hostname or "").lower() |
| 1693 | if host_transport_failures.get(host, 0) >= self.settings.research_official_document_max_transport_failures_per_host: |
| 1694 | logger.info( |
| 1695 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s outcome=SKIPPED reason=HOST_TRANSPORT_FAILURE_BUDGET", |
| 1696 | profile.instrument_id, |
| 1697 | host, |
| 1698 | ) |
| 1699 | remaining_hosts = { |
| 1700 | (urlparse(item.source.url).hostname or "").lower() |
| 1701 | for item in scheduled_filings[filing_index:] |
| 1702 | } |
| 1703 | if remaining_hosts == {host}: |
| 1704 | logger.info( |
| 1705 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s outcome=STOPPED reason=HOST_TRANSPORT_FAILURE_BUDGET", |
| 1706 | profile.instrument_id, |
| 1707 | host, |
| 1708 | ) |
| 1709 | break |
| 1710 | continue |
| 1711 | attempted += 1 |
| 1712 | started = time.monotonic() |
| 1713 | try: |
| 1714 | document, joined_in_flight = await self._single_flight_official_filing(profile, source) |
| 1715 | if document.status != DocumentStatus.DUPLICATE: |
| 1716 | seen_urls.add(document.canonical_url) |
| 1717 | if document.status != DocumentStatus.DUPLICATE and not _is_usable_durable_document(document): |
| 1718 | # A completed transport attempt is not a successful |
| 1719 | # no-change check when extraction/persistence left only a |
| 1720 | # terminal non-usable document. Keep it retryable. |
| 1721 | completed_without_failure = False |
| 1722 | if joined_in_flight: |
| 1723 | logger.info( |
| 1724 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=REUSED reason=IN_FLIGHT_SINGLE_FLIGHT documentId=%s", |
| 1725 | profile.instrument_id, host, _safe_url_path(source.url), document.document_id, |
| 1726 | ) |
| 1727 | logger.info( |
| 1728 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=SUCCESS reason=NONE elapsedMs=%s httpStatus=%s", |
| 1729 | profile.instrument_id, |
| 1730 | host, |
| 1731 | _safe_url_path(source.url), |
| 1732 | _elapsed_ms(started), |
| 1733 | 200, |
| 1734 | ) |
| 1735 | logger.info( |
| 1736 | "official_document_persist provider=NSE globalInstrumentId=%s documentType=%s outcome=%s", |
| 1737 | profile.instrument_id, |
| 1738 | document.document_type, |
| 1739 | document.status, |
| 1740 | ) |
| 1741 | except TimeoutError: |
| 1742 | completed_without_failure = False |
| 1743 | exc = TransportFetchError("NETWORK_TIMEOUT") |
| 1744 | host_transport_failures[host] = host_transport_failures.get(host, 0) + 1 |
| 1745 | self.last_live_error[profile.instrument_id] = "OFFICIAL_FILING_FETCH_FAILED:NETWORK_TIMEOUT" |
| 1746 | logger.warning( |
| 1747 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=FAILED reason=%s elapsedMs=%s httpStatus=%s", |
| 1748 | profile.instrument_id, host, _safe_url_path(source.url), exc, _elapsed_ms(started), "NONE", |
| 1749 | ) |
| 1750 | except TransportFetchError as exc: |
| 1751 | completed_without_failure = False |
| 1752 | host_transport_failures[host] = host_transport_failures.get(host, 0) + 1 |
| 1753 | self.last_live_error[profile.instrument_id] = f"OFFICIAL_FILING_FETCH_FAILED:{type(exc).__name__}" |
| 1754 | logger.warning( |
| 1755 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=FAILED reason=%s elapsedMs=%s httpStatus=%s", |
| 1756 | profile.instrument_id, host, _safe_url_path(source.url), str(exc), _elapsed_ms(started), "NONE", |
| 1757 | ) |
| 1758 | except (FetchError, RestrictedFetchError, ValueError) as exc: |
| 1759 | completed_without_failure = False |
| 1760 | if _is_transport_fetch_failure(exc): |
| 1761 | host_transport_failures[host] = host_transport_failures.get(host, 0) + 1 |
| 1762 | self.last_live_error[profile.instrument_id] = f"OFFICIAL_FILING_FETCH_FAILED:{type(exc).__name__}" |
| 1763 | logger.warning( |
| 1764 | "official_document_fetch provider=NSE globalInstrumentId=%s host=%s path=%s outcome=FAILED reason=%s elapsedMs=%s httpStatus=%s", |
| 1765 | profile.instrument_id, |
| 1766 | host, |
| 1767 | _safe_url_path(source.url), |
| 1768 | str(exc), |
| 1769 | _elapsed_ms(started), |
| 1770 | getattr(exc, "status_code", "NONE"), |
| 1771 | ) |
| 1772 | return completed_without_failure |
| 1773 | |
| 1774 | def _reconcile_reused_official_financial_facts( |
| 1775 | self, |
| 1776 | profile: CompanyResearchProfile, |
| 1777 | source: RegisteredResearchSource, |
| 1778 | document: ResearchDocument, |
| 1779 | ) -> None: |
| 1780 | reason = self._reusable_financial_fact_reconcile_skip_reason(profile, source, document) |
| 1781 | if reason is not None: |
| 1782 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s documentId=%s outcome=SKIPPED reason=%s", |
| 1783 | profile.instrument_id, document.document_id, reason) |
| 1784 | return |
| 1785 | parsed, written = self._persist_official_financial_facts(document) |
| 1786 | outcome, reason = ("PROCESSED", "FACTS_UPSERTED" if written else "NO_CHANGES") if parsed else ("SKIPPED", "PARSE_NO_RESULT") |
| 1787 | logger.info("official_financial_fact_reconcile provider=NSE globalInstrumentId=%s documentId=%s outcome=%s reason=%s", |
| 1788 | profile.instrument_id, document.document_id, outcome, reason) |
| 1789 | |
| 1790 | def _reusable_financial_fact_reconcile_skip_reason( |
| 1791 | self, |
| 1792 | profile: CompanyResearchProfile, |
| 1793 | source: RegisteredResearchSource, |
| 1794 | document: ResearchDocument, |
| 1795 | ) -> str | None: |
| 1796 | if not self._has_trusted_nse_profile_identity(profile, source, profile): |
| 1797 | return "UNTRUSTED_SOURCE" |
| 1798 | if "FINANCIAL_RESULTS" not in source.categories: |
| 1799 | return "NOT_FINANCIAL_RESULT" |
| 1800 | if (document.source_mode != SourceMode.REAL or document.instrument_id != profile.instrument_id |
| 1801 | or document.company_id != profile.company_id |
| 1802 | or document.source_classification != SourceClassification.EXCHANGE |
| 1803 | or document.source_type != SourceType.EXCHANGE_ANNOUNCEMENT): |
| 1804 | return "UNTRUSTED_SOURCE" |
| 1805 | if not document.normalized_text: |
| 1806 | return "NO_NORMALIZED_TEXT" |
| 1807 | return None |
| 1808 | |
| 1809 | async def _single_flight_official_filing( |
| 1810 | self, |
| 1811 | profile: CompanyResearchProfile, |
| 1812 | source: RegisteredResearchSource, |
| 1813 | ) -> tuple[ResearchDocument, bool]: |
| 1814 | """Share one official-file operation without retaining completed keys.""" |
| 1815 | key = (profile.instrument_id, canonicalize_url(source.url)) |
| 1816 | flight = self._official_filing_flights.get(key) |
| 1817 | joined_in_flight = flight is not None |
| 1818 | if flight is None: |
| 1819 | flight = asyncio.create_task( |
| 1820 | self._run_official_filing_flight(key, profile, source) |
| 1821 | ) |
| 1822 | self._official_filing_flights[key] = flight |
| 1823 | try: |
| 1824 | # A cancelled follower must not cancel the leader's work. |
| 1825 | return await asyncio.shield(flight), joined_in_flight |
| 1826 | except asyncio.CancelledError: |
| 1827 | # The caller that created the flight owns cancellation. This also |
| 1828 | # makes its failure visible to followers and guarantees cleanup in |
| 1829 | # the worker's finally block. |
| 1830 | if not joined_in_flight and not flight.done(): |
| 1831 | flight.cancel() |
| 1832 | raise |
| 1833 | |
| 1834 | async def _run_official_filing_flight( |
| 1835 | self, |
| 1836 | key: tuple[UUID, str], |
| 1837 | profile: CompanyResearchProfile, |
| 1838 | source: RegisteredResearchSource, |
| 1839 | ) -> ResearchDocument: |
| 1840 | try: |
| 1841 | self._validate_registered_source(profile, source) |
| 1842 | parsed_url = urlparse(source.url) |
| 1843 | host = (parsed_url.hostname or "").lower() |
| 1844 | logger.info( |
| 1845 | "official_document_fetch_start provider=NSE globalInstrumentId=%s scheme=%s host=%s path=%s timeoutSeconds=%s connectTimeoutSeconds=%s retries=%s", |
| 1846 | profile.instrument_id, parsed_url.scheme, host, _safe_url_path(source.url), |
| 1847 | self.settings.research_official_document_timeout_seconds, |
| 1848 | self.settings.research_connect_timeout_seconds, self.settings.research_max_retries, |
| 1849 | ) |
| 1850 | if isinstance(self._fetcher, HttpResearchFetcher): |
| 1851 | async with asyncio.timeout(self.settings.research_official_document_timeout_seconds): |
| 1852 | network_result = await self._fetcher.fetch_network( |
| 1853 | source.url, |
| 1854 | headers={"User-Agent": self.settings.research_official_document_user_agent}, |
| 1855 | max_bytes=self.settings.research_official_document_max_bytes, |
| 1856 | ) |
| 1857 | else: |
| 1858 | network_result = None |
| 1859 | if network_result is not None: |
| 1860 | fetch_result = await self._fetcher.process_network_response_async( |
| 1861 | network_result, |
| 1862 | max_bytes=self.settings.research_official_document_max_bytes, |
| 1863 | extraction_timeout_seconds=self.settings.research_official_document_extraction_timeout_seconds, |
| 1864 | ) |
| 1865 | logger.info("fetch_persist_start provider=NSE globalInstrumentId=%s host=%s path=%s", profile.instrument_id, host, _safe_url_path(source.url)) |
| 1866 | document = await self._ingest_registered_fetch_result_async(profile, source, fetch_result, expected_profile=profile) |
| 1867 | logger.info("fetch_persist_complete provider=NSE globalInstrumentId=%s host=%s path=%s", profile.instrument_id, host, _safe_url_path(source.url)) |
| 1868 | return document |
| 1869 | return await self._fetch_registered_source(profile, source, expected_profile=profile) |
| 1870 | finally: |
| 1871 | # Identity check prevents an old, cancelled flight from removing a |
| 1872 | # retry flight that was installed for the same key. |
| 1873 | if self._official_filing_flights.get(key) is asyncio.current_task(): |
| 1874 | self._official_filing_flights.pop(key, None) |
| 1875 | |
| 1876 | def _reusable_official_document(self, instrument_id: UUID, url: str) -> ResearchDocument | None: |
| 1877 | canonical = canonicalize_url(url) |
| 1878 | return next( |
| 1879 | ( |
| 1880 | document |
| 1881 | for document in self.documents_for(instrument_id, source_mode=SourceMode.REAL) |
| 1882 | if document.canonical_url == canonical and _is_usable_durable_document(document) |
| 1883 | ), |
| 1884 | None, |
| 1885 | ) |
| 1886 | |
| 1887 | def _missing_categories( |
| 1888 | self, |
| 1889 | profile: CompanyResearchProfile, |
| 1890 | pre_resolved_categories: set[str], |
| 1891 | now: datetime, |
| 1892 | structured_categories: set[str], |
| 1893 | *, |
| 1894 | force: bool = False, |
| 1895 | ) -> set[str]: |
| 1896 | coverage = self._scorer.score( |
| 1897 | profile.instrument_id, |
| 1898 | self.events_for(profile.instrument_id, source_mode=SourceMode.REAL), |
| 1899 | ).category_evidence |
| 1900 | missing = { |
| 1901 | _canonical_refresh_category(category) |
| 1902 | for category, evidence in coverage.items() |
| 1903 | if evidence.status == "NO_EVIDENCE" |
| 1904 | } |
| 1905 | missing.update(_canonical_refresh_category(category) for category in structured_categories) |
| 1906 | missing.difference_update(_canonical_refresh_category(category) for category in pre_resolved_categories) |
| 1907 | if force: |
| 1908 | return missing |
| 1909 | return { |
| 1910 | category for category in missing |
| 1911 | if not self._category_is_fresh(profile.instrument_id, category, now) |
| 1912 | or self._quarterly_window_open(profile.instrument_id, category, now) |
| 1913 | } |
| 1914 | |
| 1915 | def _quarterly_window_open(self, instrument_id: UUID, category: str, now: datetime) -> bool: |
| 1916 | if _CATEGORY_STRATEGIES.get(category) is not _QUARTERLY_WINDOW: |
| 1917 | return False |
| 1918 | _evidence_at, period_end = self._category_evidence_timing(instrument_id, category) |
| 1919 | return period_end is not None and now >= _next_quarter_window(period_end) |
| 1920 | |
| 1921 | def _category_is_eligible_to_check( |
| 1922 | self, |
| 1923 | instrument_id: UUID, |
| 1924 | category: str, |
| 1925 | now: datetime, |
| 1926 | ) -> tuple[bool, datetime | None]: |
| 1927 | """Keep provider polling separate from evidence freshness. |
| 1928 | |
| 1929 | `_category_refresh` records the last successful provider check in this |
| 1930 | process. Evidence time remains on the durable document/snapshot and is |
| 1931 | never changed when an official listing is unchanged. |
| 1932 | """ |
| 1933 | category = _canonical_refresh_category(category) |
| 1934 | strategy = _CATEGORY_STRATEGIES.get(category, _PERIODIC_SLOW) |
| 1935 | evidence_at, period_end = self._category_evidence_timing(instrument_id, category) |
| 1936 | if evidence_at is None: |
| 1937 | checked_at = self._category_successful_no_change_checks.get((instrument_id, category)) |
| 1938 | if checked_at is None: |
| 1939 | # Failed provider attempts never write this state and remain |
| 1940 | # immediately retryable. |
| 1941 | return True, None |
| 1942 | next_eligible = checked_at + strategy.lightweight_check_interval |
| 1943 | return now >= next_eligible, next_eligible |
| 1944 | checked_at = self._category_refresh.get((instrument_id, category)) |
| 1945 | if strategy is _QUARTERLY_WINDOW and period_end is not None: |
| 1946 | next_window = _next_quarter_window(period_end) |
| 1947 | next_eligible = next_window |
| 1948 | if checked_at is not None and checked_at >= next_window: |
| 1949 | next_eligible = checked_at + strategy.lightweight_check_interval |
| 1950 | return now >= next_eligible, next_eligible |
| 1951 | if strategy is _ANNUAL_WINDOW and period_end is not None: |
| 1952 | next_window = _next_annual_window(period_end) |
| 1953 | next_eligible = next_window |
| 1954 | if checked_at is not None and checked_at >= next_window: |
| 1955 | next_eligible = checked_at + strategy.lightweight_check_interval |
| 1956 | return now >= next_eligible, next_eligible |
| 1957 | next_eligible = (checked_at or evidence_at) + strategy.lightweight_check_interval |
| 1958 | return now >= next_eligible, next_eligible |
| 1959 | |
| 1960 | def _category_evidence_timing( |
| 1961 | self, |
| 1962 | instrument_id: UUID, |
| 1963 | category: str, |
| 1964 | ) -> tuple[datetime | None, datetime | None]: |
| 1965 | category = _canonical_refresh_category(category) |
| 1966 | if category == "SHAREHOLDING_PATTERN": |
| 1967 | snapshots = self.shareholding_for(instrument_id, limit=1) |
| 1968 | if snapshots: |
| 1969 | return snapshots[0].retrieved_at, snapshots[0].period_end |
| 1970 | return None, None |
| 1971 | if category == "FINANCIAL_RESULTS": |
| 1972 | documents = [ |
| 1973 | document for document in self.documents_for(instrument_id, source_mode=SourceMode.REAL) |
| 1974 | if _is_usable_durable_document(document) |
| 1975 | and any(term in f"{document.title or ''} {document.normalized_text or ''}".lower() |
| 1976 | for term in ("financial result", "quarterly result", "earnings", "annual report")) |
| 1977 | ] |
| 1978 | if documents: |
| 1979 | latest = documents[0] |
| 1980 | return latest.retrieved_at, _explicit_quarter_end(latest.normalized_text or latest.raw_text or "") |
| 1981 | evidence = self._qualifying_category_evidence(instrument_id, category) |
| 1982 | if evidence and isinstance(evidence.get("evidence_at"), str): |
| 1983 | return _parse_iso_datetime(evidence["evidence_at"]), None |
| 1984 | return None, None |
| 1985 | |
| 1986 | def _mark_qualifying_categories_fresh(self, instrument_id: UUID, categories: set[str], now: datetime) -> None: |
| 1987 | for category in categories: |
| 1988 | category = _canonical_refresh_category(category) |
| 1989 | evidence = self._qualifying_category_evidence(instrument_id, category) |
| 1990 | if evidence is not None: |
| 1991 | self._category_refresh[(instrument_id, category)] = now |
| 1992 | logger.info( |
| 1993 | "research_refresh_gate globalInstrumentId=%s category=%s outcome=CHECK_COMPLETE reason=%s documentId=%s lastCheckedAt=%s evidenceAt=%s", |
| 1994 | instrument_id, |
| 1995 | category, |
| 1996 | evidence["reason"], |
| 1997 | evidence.get("document_id", "NONE"), |
| 1998 | now.isoformat(), |
| 1999 | evidence.get("evidence_at", "NONE"), |
| 2000 | ) |
| 2001 | |
| 2002 | def _mark_successful_categories_checked(self, instrument_id: UUID, categories: set[str], now: datetime) -> None: |
| 2003 | """Record successful no-change checks without manufacturing evidence.""" |
| 2004 | for raw_category in categories: |
| 2005 | category = _canonical_refresh_category(raw_category) |
| 2006 | if self._category_has_qualifying_evidence(instrument_id, category): |
| 2007 | self._category_refresh[(instrument_id, category)] = now |
| 2008 | continue |
| 2009 | self._category_successful_no_change_checks[(instrument_id, category)] = now |
| 2010 | logger.info( |
| 2011 | "research_refresh_gate globalInstrumentId=%s category=%s outcome=LIGHTWEIGHT_CHECK_UNCHANGED lastCheckedAt=%s evidenceAt=NONE", |
| 2012 | instrument_id, category, now.isoformat(), |
| 2013 | ) |
| 2014 | |
| 2015 | def _category_has_qualifying_evidence(self, instrument_id: UUID, category: str) -> bool: |
| 2016 | return self._qualifying_category_evidence(instrument_id, category) is not None |
| 2017 | |
| 2018 | def _qualifying_category_evidence(self, instrument_id: UUID, category: str) -> dict[str, object] | None: |
| 2019 | category = _canonical_refresh_category(category) |
| 2020 | if category == "SHAREHOLDING_PATTERN": |
| 2021 | snapshots = self.shareholding_for(instrument_id, limit=1) |
| 2022 | if snapshots: |
| 2023 | snapshot = snapshots[0] |
| 2024 | return { |
| 2025 | "reason": "DURABLE_REAL_SHAREHOLDING_SNAPSHOT", |
| 2026 | "document_id": snapshot.research_document_id or "NONE", |
| 2027 | "evidence_at": snapshot.retrieved_at.isoformat(), |
| 2028 | } |
| 2029 | category_evidence = self._scorer.score( |
| 2030 | instrument_id, |
| 2031 | self.events_for(instrument_id, source_mode=SourceMode.REAL), |
| 2032 | ).category_evidence |
| 2033 | for evidence_key in _refresh_evidence_keys(category): |
| 2034 | evidence = category_evidence.get(evidence_key) |
| 2035 | if evidence is not None and evidence.status != "NO_EVIDENCE": |
| 2036 | return {"reason": "REAL_EVENT_EVIDENCE"} |
| 2037 | if category != "FINANCIAL_RESULTS": |
| 2038 | return None |
| 2039 | result_terms = ("financial result", "quarterly result", "earnings", "annual report") |
| 2040 | for document in self.documents_for(instrument_id, source_mode=SourceMode.REAL): |
| 2041 | if _is_usable_durable_document(document) and any( |
| 2042 | term in f"{document.title or ''} {document.normalized_text or ''}".lower() for term in result_terms |
| 2043 | ): |
| 2044 | return { |
| 2045 | "reason": "DURABLE_FINANCIAL_RESULT_DOCUMENT", |
| 2046 | "document_id": document.document_id, |
| 2047 | "evidence_at": document.retrieved_at.isoformat(), |
| 2048 | } |
| 2049 | return None |
| 2050 | |
| 2051 | def _category_is_fresh(self, instrument_id: UUID, category: str, now: datetime) -> bool: |
| 2052 | category = _canonical_refresh_category(category) |
| 2053 | refreshed_at = self._category_refresh.get((instrument_id, category)) |
| 2054 | if category == "SHAREHOLDING_PATTERN" and refreshed_at is None: |
| 2055 | snapshots = self.shareholding_for(instrument_id, limit=1) |
| 2056 | if snapshots: |
| 2057 | refreshed_at = snapshots[0].retrieved_at |
| 2058 | if refreshed_at is None or not self._category_has_qualifying_evidence(instrument_id, category): |
| 2059 | return False |
| 2060 | if category == "FINANCIAL_RESULTS": |
| 2061 | ttl = self.settings.research_quarterly_freshness_seconds |
| 2062 | elif category in {"Ownership", "INSTITUTIONAL_ACTIVITY", "SHAREHOLDING_PATTERN"}: |
| 2063 | ttl = self.settings.research_shareholding_freshness_seconds |
| 2064 | elif category in {"ORDERS_BACKLOG", "CONTRACTS", "CAPEX", "NEW_FACILITIES", "ACQUISITIONS", "CLIENTS", "GUIDANCE", "REGULATORY"}: |
| 2065 | ttl = self.settings.research_catalyst_freshness_seconds |
| 2066 | elif category in {"ANALYST_OPINION", "ANALYST_TARGETS", "VALUATION"}: |
| 2067 | ttl = self.settings.research_analyst_freshness_seconds |
| 2068 | else: |
| 2069 | ttl = self.settings.research_search_refresh_cooldown_seconds |
| 2070 | return (now - refreshed_at).total_seconds() < ttl |
| 2071 | |
| 2072 | async def _refresh_search_discovery(self, profile: CompanyResearchProfile, missing: set[str], seen_urls: set[str]) -> bool: |
| 2073 | try: |
| 2074 | discovered = await self._search_discovery.discover(profile, missing, seen_urls) |
| 2075 | except SearchProviderError as exc: |
| 2076 | reason = str(exc) |
| 2077 | self._search_discovery.last_stats.reject(reason) |
| 2078 | self.last_live_error[profile.instrument_id] = f"SEARCH_PROVIDER_UNAVAILABLE:{reason}" |
| 2079 | logger.warning("research_discovery_terminal company=%s provider=%s status=SEARCH_PROVIDER_UNAVAILABLE reason=%s", |
| 2080 | profile.company_name, self._search_discovery.provider.provider_name, reason) |
| 2081 | return False |
| 2082 | if not discovered: |
| 2083 | stats = self._search_discovery.last_stats |
| 2084 | if stats.candidate_count == 0: |
| 2085 | terminal = "SEARCH_RETURNED_ZERO_RESULTS" |
| 2086 | else: |
| 2087 | terminal = "RESULTS_REJECTED" |
| 2088 | self.last_live_error[profile.instrument_id] = terminal |
| 2089 | logger.warning("research_discovery_terminal company=%s provider=%s status=%s candidates=%s accepted=%s rejected_reasons=%s", |
| 2090 | profile.company_name, self._search_discovery.provider.provider_name, terminal, |
| 2091 | stats.candidate_count, stats.accepted_count, stats.rejected_reasons) |
| 2092 | # A zero-result response is a successful no-change check only |
| 2093 | # when the provider completed the requested search work. Partial |
| 2094 | # engine/provider failures remain retryable and must not advance |
| 2095 | # the no-change schedule. |
| 2096 | return stats.provider_failure_count == 0 |
| 2097 | fetched_documents = 0 |
| 2098 | extracted_before = len(self.events) |
| 2099 | fetch_failures: dict[str, int] = {} |
| 2100 | for result in discovered: |
| 2101 | if fetched_documents >= self.settings.research_search_max_documents_per_refresh: |
| 2102 | break |
| 2103 | source = result.source |
| 2104 | try: |
| 2105 | self._validate_registered_source(profile, source) |
| 2106 | document = await self._fetch_registered_source(profile, source, expected_profile=profile) |
| 2107 | if document.status == DocumentStatus.DUPLICATE: |
| 2108 | self._search_discovery.last_stats.reject("DUPLICATE") |
| 2109 | continue |
| 2110 | fetched_documents += 1 |
| 2111 | self._search_discovery.last_stats.reject("SEARCH_RESULT_ACCEPTED") |
| 2112 | except RestrictedFetchError as exc: |
| 2113 | self._search_discovery.last_stats.reject("ROBOTS_OR_ACCESS_BLOCKED") |
| 2114 | fetch_failures["ROBOTS_OR_ACCESS_BLOCKED"] = fetch_failures.get("ROBOTS_OR_ACCESS_BLOCKED", 0) + 1 |
| 2115 | self.last_live_error[profile.instrument_id] = f"SEARCH_SOURCE_UNAVAILABLE:{source.source_id}:{exc}" |
| 2116 | except FetchError as exc: |
| 2117 | failure = _fetch_rejection_reason(exc) |
| 2118 | self._search_discovery.last_stats.reject(failure) |
| 2119 | fetch_failures[failure] = fetch_failures.get(failure, 0) + 1 |
| 2120 | self.last_live_error[profile.instrument_id] = f"SEARCH_SOURCE_UNAVAILABLE:{source.source_id}:{exc}" |
| 2121 | except ValueError as exc: |
| 2122 | self._search_discovery.last_stats.reject("PARSER_FAILED") |
| 2123 | fetch_failures["PARSER_FAILED"] = fetch_failures.get("PARSER_FAILED", 0) + 1 |
| 2124 | self.last_live_error[profile.instrument_id] = f"SEARCH_SOURCE_UNAVAILABLE:{source.source_id}:{exc}" |
| 2125 | except Exception as exc: |
| 2126 | self._search_discovery.last_stats.reject("HTTP_FETCH_FAILED") |
| 2127 | fetch_failures["HTTP_FETCH_FAILED"] = fetch_failures.get("HTTP_FETCH_FAILED", 0) + 1 |
| 2128 | self.last_live_error[profile.instrument_id] = f"SEARCH_SOURCE_UNAVAILABLE:{source.source_id}:HTTP_FETCH_FAILED" |
| 2129 | if fetched_documents > 0: |
| 2130 | self.last_live_error.pop(profile.instrument_id, None) |
| 2131 | elif discovered: |
| 2132 | self.last_live_error[profile.instrument_id] = "DOCUMENT_FETCH_FAILED" |
| 2133 | self._search_discovery.last_stats.documents_fetched = fetched_documents |
| 2134 | self._search_discovery.last_stats.events_extracted = len(self.events) - extracted_before |
| 2135 | logger.info("research_fetch_complete company=%s provider=%s accepted_results=%s document_fetch_count=%s extraction_count=%s terminal_status=%s failure_reasons=%s", |
| 2136 | profile.company_name, self._search_discovery.provider.provider_name, len(discovered), fetched_documents, |
| 2137 | self._search_discovery.last_stats.events_extracted, |
| 2138 | "RESOLVED_PARTIAL_DATA" if fetched_documents else "DOCUMENT_FETCH_FAILED", fetch_failures) |
| 2139 | return not fetch_failures |
| 2140 | |
| 2141 | async def _refresh_etf_search_discovery(self, profile: EtfResearchProfile) -> None: |
| 2142 | categories = {"ETF_PROFILE", "ETF_PERFORMANCE", "INDEX_OUTLOOK", "ETF_RISK"} |
| 2143 | seen_urls = {doc.canonical_url for doc in self.documents_for(profile.instrument_id, source_mode=SourceMode.REAL)} |
| 2144 | try: |
| 2145 | discovered = await self._search_discovery.discover(profile, categories, seen_urls) |
| 2146 | except SearchProviderError as exc: |
| 2147 | reason = str(exc) |
| 2148 | self._search_discovery.last_stats.reject(reason) |
| 2149 | self.last_live_error[profile.instrument_id] = reason |
| 2150 | return |
| 2151 | if not discovered: |
| 2152 | self.last_live_error[profile.instrument_id] = "ETF_RESEARCH_SOURCE_UNAVAILABLE:NO_ACCEPTABLE_DOCUMENT" |
| 2153 | return |
| 2154 | fetched_documents = 0 |
| 2155 | for result in discovered: |
| 2156 | if fetched_documents >= self.settings.research_search_max_documents_per_refresh: |
| 2157 | break |
| 2158 | try: |
| 2159 | self._validate_registered_source(profile, result.source) |
| 2160 | document = await self._fetch_etf_source(profile, result.source) |
| 2161 | if document.status == DocumentStatus.DUPLICATE: |
| 2162 | self._search_discovery.last_stats.reject("DUPLICATE") |
| 2163 | continue |
| 2164 | fetched_documents += 1 |
| 2165 | self._search_discovery.last_stats.reject("SEARCH_RESULT_ACCEPTED") |
| 2166 | except RestrictedFetchError: |
| 2167 | self._search_discovery.last_stats.reject("ROBOTS_OR_ACCESS_BLOCKED") |
| 2168 | except FetchError as exc: |
| 2169 | self._search_discovery.last_stats.reject(_fetch_rejection_reason(exc)) |
| 2170 | except ValueError: |
| 2171 | self._search_discovery.last_stats.reject("PARSER_FAILED") |
| 2172 | except Exception: |
| 2173 | self._search_discovery.last_stats.reject("HTTP_FETCH_FAILED") |
| 2174 | if fetched_documents > 0: |
| 2175 | self.last_live_error.pop(profile.instrument_id, None) |
| 2176 | else: |
| 2177 | self.last_live_error[profile.instrument_id] = "ETF_RESEARCH_SOURCE_UNAVAILABLE:NO_ACCEPTABLE_DOCUMENT" |
| 2178 | self._search_discovery.last_stats.documents_fetched = fetched_documents |
| 2179 | |
| 2180 | async def _fetch_registered_source( |
| 2181 | self, |
| 2182 | profile: CompanyResearchProfile, |
| 2183 | source: RegisteredResearchSource, |
| 2184 | *, |
| 2185 | expected_profile: CompanyResearchProfile | None = None, |
| 2186 | ) -> ResearchDocument: |
| 2187 | self._validate_registered_source(profile, source) |
| 2188 | result = await self._fetcher.fetch(source.url) |
| 2189 | return await self._ingest_registered_fetch_result_async(profile, source, result, expected_profile=expected_profile) |
| 2190 | |
| 2191 | async def _ingest_registered_fetch_result_async( |
| 2192 | self, profile: CompanyResearchProfile, source: RegisteredResearchSource, result, *, |
| 2193 | expected_profile: CompanyResearchProfile | None = None, |
| 2194 | ) -> ResearchDocument: |
| 2195 | if result.status_code >= 400: |
| 2196 | raise FetchError(f"Registered source returned {result.status_code}") |
| 2197 | if source.discovery_method == "NSE_OFFICIAL_API": |
| 2198 | logger.info("official_document_fetch provider=NSE globalInstrumentId=%s documentType=%s httpStatus=%s extractionStatus=%s", profile.instrument_id, result.content_type, result.status_code, result.extraction_status) |
| 2199 | document_status = DocumentStatus.FAILED if result.extraction_status != "EXTRACTED" else DocumentStatus.PARSED |
| 2200 | trusted_identity = _TRUSTED_NSE_PROFILE_IDENTITY if self._has_trusted_nse_profile_identity(profile, source, expected_profile) else None |
| 2201 | started = time.monotonic() |
| 2202 | try: |
| 2203 | document = await asyncio.to_thread( |
| 2204 | self._prepare_ingested_document, |
| 2205 | original_url=result.final_url, source_type=source.source_type, source_classification=source.source_classification, |
| 2206 | source_name=source.source_name, publisher=source.publisher, content_type=result.content_type, body=result.text, |
| 2207 | reliability=source.reliability_level, published_at=None, source_mode=SourceMode.REAL, |
| 2208 | discovered_at=None, |
| 2209 | discovery_provider=(source.discovery_method if source.discovery_method == "SEARCH_DISCOVERY" or trusted_identity else None), |
| 2210 | expected_profile=expected_profile, document_status=document_status, |
| 2211 | allow_empty_content=result.extraction_status != "EXTRACTED", trusted_profile_identity=trusted_identity, |
| 2212 | document_subtype=source.document_subtype, |
| 2213 | ) |
| 2214 | except Exception: |
| 2215 | logger.info("document_ingest_stage globalInstrumentId=%s stage=PREPARE elapsedMs=%s outcome=FAILED", profile.instrument_id, _elapsed_ms(started)) |
| 2216 | raise |
| 2217 | logger.info("document_ingest_stage globalInstrumentId=%s documentId=%s stage=PREPARE elapsedMs=%s outcome=SUCCESS", profile.instrument_id, document.document_id, _elapsed_ms(started)) |
| 2218 | return await self._apply_prepared_ingested_document_async( |
| 2219 | document, |
| 2220 | document_status=document_status, |
| 2221 | metadata_only_nse_financial_result="FINANCIAL_RESULTS" in source.categories, |
| 2222 | ) |
| 2223 | |
| 2224 | def _ingest_registered_fetch_result( |
| 2225 | self, |
| 2226 | profile: CompanyResearchProfile, |
| 2227 | source: RegisteredResearchSource, |
| 2228 | result, |
| 2229 | *, |
| 2230 | expected_profile: CompanyResearchProfile | None = None, |
| 2231 | ) -> ResearchDocument: |
| 2232 | if result.status_code >= 400: |
| 2233 | raise FetchError(f"Registered source returned {result.status_code}") |
| 2234 | if source.discovery_method == "NSE_OFFICIAL_API": |
| 2235 | logger.info("official_document_fetch provider=NSE globalInstrumentId=%s documentType=%s httpStatus=%s extractionStatus=%s", profile.instrument_id, result.content_type, result.status_code, result.extraction_status) |
| 2236 | return self.ingest_fixture( |
| 2237 | original_url=result.final_url, |
| 2238 | source_type=source.source_type, |
| 2239 | source_classification=source.source_classification, |
| 2240 | source_name=source.source_name, |
| 2241 | publisher=source.publisher, |
| 2242 | content_type=result.content_type, |
| 2243 | body=result.text, |
| 2244 | reliability=source.reliability_level, |
| 2245 | source_mode=SourceMode.REAL, |
| 2246 | discovery_provider=(source.discovery_method if source.discovery_method == "SEARCH_DISCOVERY" |
| 2247 | or self._has_trusted_nse_profile_identity(profile, source, expected_profile) else None), |
| 2248 | expected_profile=expected_profile, |
| 2249 | document_status=DocumentStatus.FAILED if result.extraction_status != "EXTRACTED" else DocumentStatus.PARSED, |
| 2250 | allow_empty_content=result.extraction_status != "EXTRACTED", |
| 2251 | _trusted_profile_identity=( |
| 2252 | _TRUSTED_NSE_PROFILE_IDENTITY |
| 2253 | if self._has_trusted_nse_profile_identity(profile, source, expected_profile) else None |
| 2254 | ), |
| 2255 | document_subtype=source.document_subtype, |
| 2256 | _metadata_only_nse_financial_result="FINANCIAL_RESULTS" in source.categories, |
| 2257 | ) |
| 2258 | |
| 2259 | @staticmethod |
| 2260 | def _has_trusted_nse_profile_identity( |
| 2261 | profile: CompanyResearchProfile, |
| 2262 | source: RegisteredResearchSource, |
| 2263 | expected_profile: CompanyResearchProfile | None, |
| 2264 | ) -> bool: |
| 2265 | """Allow NSE API discovery, not generic retrieval, to bind a filing's identity.""" |
| 2266 | return ( |
| 2267 | expected_profile is profile |
| 2268 | and source.instrument_id == profile.instrument_id |
| 2269 | and source.company_id == profile.company_id |
| 2270 | and source.source_classification == SourceClassification.EXCHANGE |
| 2271 | and source.discovery_method == "NSE_OFFICIAL_API" |
| 2272 | and source.official_nse_profile_symbol is not None |
| 2273 | and source.official_nse_profile_symbol == profile.provider_instrument_ids.get("NSE") |
| 2274 | ) |
| 2275 | |
| 2276 | async def _fetch_etf_source(self, profile: EtfResearchProfile, source: RegisteredResearchSource) -> ResearchDocument: |
| 2277 | self._validate_registered_source(profile, source) |
| 2278 | result = await self._fetcher.fetch(source.url) |
| 2279 | if result.status_code >= 400: |
| 2280 | raise FetchError(f"Registered source returned {result.status_code}") |
| 2281 | return self.ingest_etf_fixture( |
| 2282 | profile, |
| 2283 | original_url=result.final_url, |
| 2284 | source_type=source.source_type, |
| 2285 | source_classification=source.source_classification, |
| 2286 | source_name=source.source_name, |
| 2287 | publisher=source.publisher, |
| 2288 | content_type=result.content_type, |
| 2289 | body=result.text, |
| 2290 | reliability=source.reliability_level, |
| 2291 | discovery_provider=source.discovery_method if source.discovery_method == "SEARCH_DISCOVERY" else None, |
| 2292 | ) |
| 2293 | |
| 2294 | def ingest_etf_fixture( |
| 2295 | self, |
| 2296 | profile: EtfResearchProfile, |
| 2297 | *, |
| 2298 | original_url: str, |
| 2299 | source_type: SourceType, |
| 2300 | source_name: str, |
| 2301 | publisher: str, |
| 2302 | content_type: str, |
| 2303 | body: str, |
| 2304 | reliability: ReliabilityLevel, |
| 2305 | source_classification: SourceClassification = SourceClassification.OTHER, |
| 2306 | discovery_provider: str | None = None, |
| 2307 | ) -> ResearchDocument: |
| 2308 | canonical = canonicalize_url(original_url) |
| 2309 | title, extracted = extract_text(body, content_type) |
| 2310 | normalized = normalize_text(extracted or "") |
| 2311 | if not normalized: |
| 2312 | raise FetchError("CONTENT_EMPTY") |
| 2313 | if len(normalized) < 40: |
| 2314 | raise FetchError("CONTENT_TOO_SHORT") |
| 2315 | if not _etf_document_relevant(profile, title, normalized): |
| 2316 | raise FetchError("COMPANY_RELEVANCE_FAILED") |
| 2317 | document = ResearchDocument( |
| 2318 | canonical_url=canonical, |
| 2319 | original_url=original_url, |
| 2320 | title=title, |
| 2321 | source_type=source_type, |
| 2322 | source_classification=source_classification, |
| 2323 | source_name=source_name, |
| 2324 | publisher=publisher, |
| 2325 | published_at=extract_published_at(normalized), |
| 2326 | content_type=content_type, |
| 2327 | document_type=DocumentType.PDF_REFERENCE if canonical.lower().endswith(".pdf") else DocumentType.HTML, |
| 2328 | raw_text=body if len(body) < 20_000 else None, |
| 2329 | normalized_text=normalized, |
| 2330 | content_hash=content_hash(normalized or canonical), |
| 2331 | instrument_id=profile.instrument_id, |
| 2332 | company_id=profile.fund_id, |
| 2333 | status=DocumentStatus.PARSED, |
| 2334 | reliability_level=reliability, |
| 2335 | entity_resolution_confidence=1.0, |
| 2336 | source_mode=SourceMode.REAL, |
| 2337 | freshness=SourceMode.REAL.value, |
| 2338 | discovery_provider=discovery_provider, |
| 2339 | source_independence_key=content_hash(normalized or canonical), |
| 2340 | ) |
| 2341 | duplicate = self._deduplicator.add(document) |
| 2342 | if duplicate: |
| 2343 | document.status = DocumentStatus.DUPLICATE |
| 2344 | document.duplicate_of_document_id = duplicate.document_id |
| 2345 | return document |
| 2346 | self.documents[document.document_id] = document |
| 2347 | self._persistence.upsert_document(document) |
| 2348 | self._update_etf_facts(profile, document) |
| 2349 | self.last_refresh[profile.instrument_id] = datetime.now(timezone.utc) |
| 2350 | return document |
| 2351 | |
| 2352 | def _validate_registered_source(self, profile: CompanyResearchProfile | EtfResearchProfile, source: RegisteredResearchSource) -> None: |
| 2353 | if not source.allowed: |
| 2354 | raise FetchError("SOURCE_QUALITY_REJECTED") |
| 2355 | if source.instrument_id != profile.instrument_id: |
| 2356 | raise FetchError("COMPANY_RELEVANCE_FAILED") |
| 2357 | host = (urlparse(source.url).hostname or "").lower() |
| 2358 | if source.host and host != source.host: |
| 2359 | raise FetchError("DOMAIN_VALIDATION_FAILED") |
| 2360 | first_party = source.source_classification == SourceClassification.OFFICIAL_COMPANY |
| 2361 | authoritative_public = source.source_classification in {SourceClassification.EXCHANGE, SourceClassification.REGULATORY} |
| 2362 | known_company_domain = any(host == domain.lower() or host.endswith(f".{domain.lower()}") for domain in profile.known_domains) |
| 2363 | if (first_party or (source.priority <= 2 and not authoritative_public)) and not known_company_domain: |
| 2364 | raise FetchError("DOMAIN_VALIDATION_FAILED") |
| 2365 | |
| 2366 | def _update_etf_facts(self, profile: EtfResearchProfile, document: ResearchDocument) -> None: |
| 2367 | text = document.normalized_text or "" |
| 2368 | source = ProvenancedValue(value="", source_url=document.canonical_url, source_name=document.source_name, retrieved_at=document.retrieved_at) |
| 2369 | inferred_provider = profile.fund_provider or _infer_fund_provider(profile.fund_name, text) |
| 2370 | inferred_index = profile.underlying_index or _infer_underlying_index(profile.fund_name, text) |
| 2371 | if inferred_provider: |
| 2372 | profile.fund_provider = inferred_provider |
| 2373 | profile.facts["fundProvider"] = source.model_copy(update={"value": inferred_provider}) |
| 2374 | if inferred_index: |
| 2375 | profile.underlying_index = inferred_index |
| 2376 | profile.facts["underlyingIndex"] = source.model_copy(update={"value": inferred_index}) |
| 2377 | for key, value in _extract_etf_facts(text).items(): |
| 2378 | profile.facts[key] = source.model_copy(update={"value": value}) |
| 2379 | |
| 2380 | def _seed_demo_data(self) -> None: |
| 2381 | fixtures = [ |
| 2382 | ( |
| 2383 | "https://ir.aixtron.example/releases/order-capacity?utm_source=test", |
| 2384 | "AIXTRON SE announced a new order worth €350 million from a leading power electronics customer. The XETR AIXA order supports silicon-carbide equipment demand and increases backlog by +42%.", |
| 2385 | SourceType.INVESTOR_RELATIONS, |
| 2386 | SourceClassification.OFFICIAL_COMPANY, |
| 2387 | ReliabilityLevel.LEVEL_B, |
| 2388 | ), |
| 2389 | ( |
| 2390 | "https://exchange.example/xams/besi-capacity", |
| 2391 | "BE Semiconductor Industries BESI XAMS announced capacity expansion of 73.15 MW equivalent production capability and a new facility in the Netherlands. CAPEX is €120 million and the project is under construction.", |
| 2392 | SourceType.EXCHANGE_ANNOUNCEMENT, |
| 2393 | SourceClassification.EXCHANGE, |
| 2394 | ReliabilityLevel.LEVEL_A, |
| 2395 | ), |
| 2396 | ( |
| 2397 | "https://nse.example/reliance-filing", |
| 2398 | "RELIANCE XNSE INE002A01018 disclosed an investment of ₹2,000 crore in new energy manufacturing capacity in India. Management maintained revenue guidance.", |
| 2399 | SourceType.REGULATORY_FILING, |
| 2400 | SourceClassification.REGULATORY, |
| 2401 | ReliabilityLevel.LEVEL_A, |
| 2402 | ), |
| 2403 | ( |
| 2404 | "https://news.example/nvda-delay", |
| 2405 | "NVIDIA Corporation NVDA XNAS reported that a factory ramp was delayed by one quarter while demand remains strong.", |
| 2406 | SourceType.NEWS, |
| 2407 | SourceClassification.REPUTABLE_NEWS, |
| 2408 | ReliabilityLevel.LEVEL_C, |
| 2409 | ), |
| 2410 | ] |
| 2411 | for url, body, source_type, source_classification, reliability in fixtures: |
| 2412 | self.ingest_fixture( |
| 2413 | original_url=url, |
| 2414 | source_type=source_type, |
| 2415 | source_classification=source_classification, |
| 2416 | source_name="DEMO fixture source", |
| 2417 | publisher="DEMO", |
| 2418 | content_type="text/html", |
| 2419 | body=f"<html><head><title>DEMO research fixture</title></head><body>{body}</body></html>", |
| 2420 | reliability=reliability, |
| 2421 | published_at=datetime(2026, 1, 15, tzinfo=timezone.utc), |
| 2422 | source_mode=SourceMode.DEMO, |
| 2423 | ) |
| 2424 | |
| 2425 | def _load_persisted_research(self) -> None: |
| 2426 | for document in self._persistence.load_documents(): |
| 2427 | if document.document_id in self.documents: |
| 2428 | continue |
| 2429 | self.documents[document.document_id] = document |
| 2430 | self._deduplicator.add(document) |
| 2431 | for event in self._persistence.load_events(): |
| 2432 | if event.event_id in self.events: |
| 2433 | continue |
| 2434 | self.events[event.event_id] = event |
| 2435 | self._event_keys.add(_event_key(event)) |
| 2436 | self.last_refresh[event.instrument_id] = max( |
| 2437 | self.last_refresh.get(event.instrument_id, event.detected_at), |
| 2438 | event.detected_at, |
| 2439 | ) |
| 2440 | for snapshot in self._persistence.load_shareholding_snapshots(): |
| 2441 | self.shareholding_snapshots.setdefault(snapshot.id, snapshot) |
| 2442 | |
| 2443 | |
| 2444 | def _demo_profiles() -> list[CompanyResearchProfile]: |
| 2445 | return [ |
| 2446 | CompanyResearchProfile( |
| 2447 | instrument_id=UUID("11111111-1111-1111-1111-111111111111"), |
| 2448 | company_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa1"), |
| 2449 | company_name="AIXTRON SE", |
| 2450 | aliases=["AIXTRON", "AIXA"], |
| 2451 | isin="DE000A0WMPJ6", |
| 2452 | ticker="AIXA", |
| 2453 | exchange="XETR", |
| 2454 | mic="XETR", |
| 2455 | country="DE", |
| 2456 | currency="EUR", |
| 2457 | known_domains=["aixtron.example", "aixtron.com"], |
| 2458 | official_website="https://www.aixtron.com/en", |
| 2459 | investor_relations_url="https://www.aixtron.com/en/investors", |
| 2460 | press_release_url="https://www.aixtron.com/en/press/press-releases", |
| 2461 | ), |
| 2462 | CompanyResearchProfile( |
| 2463 | instrument_id=UUID("22222222-2222-2222-2222-222222222222"), |
| 2464 | company_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa2"), |
| 2465 | company_name="BE Semiconductor Industries", |
| 2466 | aliases=["BESI", "Besi", "BE Semiconductor", "BE Semiconductor Industries N.V."], |
| 2467 | isin="NL0012866412", |
| 2468 | ticker="BESI", |
| 2469 | exchange="XAMS", |
| 2470 | mic="XAMS", |
| 2471 | country="NL", |
| 2472 | currency="EUR", |
| 2473 | known_domains=["besi.example", "besi.com"], |
| 2474 | official_website="https://www.besi.com/", |
| 2475 | investor_relations_url="https://www.besi.com/investor-relations/", |
| 2476 | press_release_url="https://www.besi.com/investor-relations/press-releases/", |
| 2477 | ), |
| 2478 | CompanyResearchProfile( |
| 2479 | instrument_id=UUID("33333333-3333-3333-3333-333333333333"), |
| 2480 | company_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa3"), |
| 2481 | company_name="NVIDIA Corporation", |
| 2482 | aliases=["NVIDIA", "NVDA"], |
| 2483 | isin="US67066G1040", |
| 2484 | ticker="NVDA", |
| 2485 | exchange="XNAS", |
| 2486 | mic="XNAS", |
| 2487 | country="US", |
| 2488 | currency="USD", |
| 2489 | known_domains=["nvidia.example"], |
| 2490 | ), |
| 2491 | CompanyResearchProfile( |
| 2492 | instrument_id=UUID("44444444-4444-4444-4444-444444444444"), |
| 2493 | company_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaa4"), |
| 2494 | company_name="Reliance Industries Limited", |
| 2495 | aliases=["RELIANCE", "Reliance Industries", "RIL"], |
| 2496 | isin="INE002A01018", |
| 2497 | ticker="RELIANCE", |
| 2498 | exchange="XNSE", |
| 2499 | mic="XNSE", |
| 2500 | country="IN", |
| 2501 | currency="INR", |
| 2502 | known_domains=["ril.example", "ril.com"], |
| 2503 | official_website="https://www.ril.com/", |
| 2504 | investor_relations_url="https://www.ril.com/investors/investor-relations", |
| 2505 | ), |
| 2506 | ] |
| 2507 | |
| 2508 | |
| 2509 | def _event_key(event: ResearchEvent) -> tuple[UUID, ResearchEventType, str, str | None, str | None]: |
| 2510 | return ( |
| 2511 | event.instrument_id, |
| 2512 | event.event_type, |
| 2513 | content_hash(event.raw_evidence_reference), |
| 2514 | event.monetary_original, |
| 2515 | event.customer, |
| 2516 | ) |
| 2517 | |
| 2518 | |
| 2519 | def _search_provider_from_settings(settings: Settings) -> SearchDiscoveryProvider: |
| 2520 | if not settings.research_search_enabled: |
| 2521 | return DisabledSearchDiscoveryProvider() |
| 2522 | if settings.research_search_provider == "google-compatible": |
| 2523 | missing = [] |
| 2524 | if not settings.research_search_endpoint: |
| 2525 | missing.append("endpoint") |
| 2526 | if not settings.research_search_api_key: |
| 2527 | missing.append("api_key") |
| 2528 | if not settings.research_search_engine_id: |
| 2529 | missing.append("engine_id") |
| 2530 | if missing: |
| 2531 | raise SearchProviderConfigurationError(f"SEARCH_PROVIDER_NOT_CONFIGURED:{','.join(missing)}") |
| 2532 | return GoogleCompatibleSearchDiscoveryProvider( |
| 2533 | settings.research_search_endpoint, |
| 2534 | settings.research_search_api_key, |
| 2535 | settings.research_search_engine_id, |
| 2536 | max_results_per_query=settings.research_search_max_results_per_query, |
| 2537 | ) |
| 2538 | if settings.research_search_provider == "brave-compatible": |
| 2539 | missing = [] |
| 2540 | if not settings.research_search_endpoint: |
| 2541 | missing.append("endpoint") |
| 2542 | if not settings.research_search_api_key: |
| 2543 | missing.append("api_key") |
| 2544 | if missing: |
| 2545 | raise SearchProviderConfigurationError(f"SEARCH_PROVIDER_NOT_CONFIGURED:{','.join(missing)}") |
| 2546 | return BraveCompatibleSearchDiscoveryProvider( |
| 2547 | settings.research_search_endpoint, |
| 2548 | settings.research_search_api_key, |
| 2549 | max_results_per_query=settings.research_search_max_results_per_query, |
| 2550 | ) |
| 2551 | if settings.research_search_provider == "searxng": |
| 2552 | if not settings.research_search_endpoint: |
| 2553 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:endpoint") |
| 2554 | return SearxngSearchDiscoveryProvider( |
| 2555 | settings.research_search_endpoint, |
| 2556 | max_results_per_query=settings.research_search_max_results_per_query, |
| 2557 | ) |
| 2558 | raise SearchProviderConfigurationError(f"SEARCH_PROVIDER_NOT_CONFIGURED:unsupported_provider:{settings.research_search_provider}") |
| 2559 | |
| 2560 | |
| 2561 | def _profile_source_discovery( |
| 2562 | profile: CompanyResearchProfile, |
| 2563 | missing_categories: set[str], |
| 2564 | seen_urls: set[str], |
| 2565 | ) -> list[DiscoveryResult]: |
| 2566 | if not missing_categories: |
| 2567 | return [] |
| 2568 | source_urls = [ |
| 2569 | ("official-website", profile.official_website, SourceType.COMPANY_WEBSITE), |
| 2570 | ("investor-relations", profile.investor_relations_url, SourceType.INVESTOR_RELATIONS), |
| 2571 | ("press-releases", profile.press_release_url, SourceType.INVESTOR_RELATIONS), |
| 2572 | ("annual-reports", profile.annual_reports_url, SourceType.INVESTOR_RELATIONS), |
| 2573 | ("exchange-announcements", profile.exchange_announcements_url, SourceType.EXCHANGE_ANNOUNCEMENT), |
| 2574 | ("regulatory-filings", profile.regulatory_filings_url, SourceType.REGULATORY_FILING), |
| 2575 | ] |
| 2576 | results: list[DiscoveryResult] = [] |
| 2577 | for source_key, url, source_type in source_urls: |
| 2578 | if url is None: |
| 2579 | continue |
| 2580 | canonical = canonicalize_url(str(url)) |
| 2581 | if canonical in seen_urls: |
| 2582 | continue |
| 2583 | host = (urlparse(canonical).hostname or "").lower() |
| 2584 | classification = SourceClassification.OFFICIAL_COMPANY |
| 2585 | reliability = ReliabilityLevel.LEVEL_B |
| 2586 | if source_type == SourceType.EXCHANGE_ANNOUNCEMENT: |
| 2587 | classification = SourceClassification.EXCHANGE |
| 2588 | reliability = ReliabilityLevel.LEVEL_A |
| 2589 | elif source_type == SourceType.REGULATORY_FILING: |
| 2590 | classification = SourceClassification.REGULATORY |
| 2591 | reliability = ReliabilityLevel.LEVEL_A |
| 2592 | source = RegisteredResearchSource( |
| 2593 | source_id=f"profile:{profile.instrument_id}:{source_key}", |
| 2594 | instrument_id=profile.instrument_id, |
| 2595 | url=canonical, |
| 2596 | source_type=source_type, |
| 2597 | source_classification=classification, |
| 2598 | source_name=f"{profile.company_name} {source_key.replace('-', ' ')}", |
| 2599 | publisher=profile.company_name, |
| 2600 | reliability_level=reliability, |
| 2601 | domain=host, |
| 2602 | company_id=profile.company_id, |
| 2603 | allowed=True, |
| 2604 | discovery_method="PROFILE_OFFICIAL", |
| 2605 | priority=2, |
| 2606 | categories=tuple(sorted(missing_categories)), |
| 2607 | ) |
| 2608 | for category in sorted(missing_categories): |
| 2609 | results.append(DiscoveryResult(category=category, source=source)) |
| 2610 | return results |
| 2611 | |
| 2612 | |
| 2613 | def _fetch_rejection_reason(exc: FetchError) -> str: |
| 2614 | message = str(exc) |
| 2615 | if message in { |
| 2616 | "CONTENT_EMPTY", |
| 2617 | "CONTENT_TOO_SHORT", |
| 2618 | "COMPANY_RELEVANCE_FAILED", |
| 2619 | "DOCUMENT_PERSIST_FAILED", |
| 2620 | "DOMAIN_VALIDATION_FAILED", |
| 2621 | "SOURCE_QUALITY_REJECTED", |
| 2622 | }: |
| 2623 | return message |
| 2624 | if "Unsupported content type" in message: |
| 2625 | return "PARSER_FAILED" |
| 2626 | if "not approved" in message or "not permitted" in message: |
| 2627 | return "DOMAIN_VALIDATION_FAILED" |
| 2628 | if "HTTP status" in message or "returned" in message or "timed out" in message: |
| 2629 | return "HTTP_FETCH_FAILED" |
| 2630 | return "PARSER_FAILED" |
| 2631 | |
| 2632 | |
| 2633 | def _is_transport_fetch_failure(exc: FetchError) -> bool: |
| 2634 | """Only transport failures trip the short-lived per-host official budget.""" |
| 2635 | return isinstance(exc, TransportFetchError) |
| 2636 | |
| 2637 | |
| 2638 | def _is_quarter_end(value: datetime) -> bool: |
| 2639 | return (value.month, value.day) in {(3, 31), (6, 30), (9, 30), (12, 31)} |
| 2640 | |
| 2641 | |
| 2642 | def _next_quarter_window(period_end: datetime) -> datetime: |
| 2643 | """First day of the next reporting quarter, in UTC, without instrument rules.""" |
| 2644 | month = period_end.month + 3 |
| 2645 | year = period_end.year + (1 if month > 12 else 0) |
| 2646 | month = month - 12 if month > 12 else month |
| 2647 | return datetime(year, month, 1, tzinfo=timezone.utc) |
| 2648 | |
| 2649 | |
| 2650 | def _next_annual_window(period_end: datetime) -> datetime: |
| 2651 | return datetime(period_end.year + 1, period_end.month, 1, tzinfo=timezone.utc) |
| 2652 | |
| 2653 | |
| 2654 | def _canonical_financial_period_end(period: str | None) -> str | None: |
| 2655 | if not period: |
| 2656 | return None |
| 2657 | if re.fullmatch(r"20\d{2}-\d{2}-\d{2}", period): |
| 2658 | try: |
| 2659 | return datetime.fromisoformat(period).date().isoformat() |
| 2660 | except ValueError: |
| 2661 | return None |
| 2662 | match = re.fullmatch(r"Q([1-4])\s*FY\s*(\d{2,4})", period.strip(), re.I) |
| 2663 | if not match: |
| 2664 | return None |
| 2665 | financial_year = int(match.group(2)) |
| 2666 | if financial_year < 100: |
| 2667 | financial_year += 2000 |
| 2668 | month, day = {"1": (6, 30), "2": (9, 30), "3": (12, 31), "4": (3, 31)}[match.group(1)] |
| 2669 | year = financial_year - 1 if match.group(1) != "4" else financial_year |
| 2670 | return datetime(year, month, day).date().isoformat() |
| 2671 | |
| 2672 | |
| 2673 | def _parse_iso_datetime(value: object) -> datetime | None: |
| 2674 | try: |
| 2675 | parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) |
| 2676 | except ValueError: |
| 2677 | return None |
| 2678 | return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) |
| 2679 | |
| 2680 | |
| 2681 | def _explicit_quarter_end(text: str) -> datetime | None: |
| 2682 | """Use an explicitly stated reporting end date; never round an arbitrary date.""" |
| 2683 | match = re.search( |
| 2684 | r"(?:quarter|three\s+months)\s+ended\s+([A-Za-z]+\s+\d{1,2},?\s+\d{4})", |
| 2685 | text, |
| 2686 | re.IGNORECASE, |
| 2687 | ) |
| 2688 | if not match: |
| 2689 | return None |
| 2690 | for pattern in ("%B %d, %Y", "%B %d %Y", "%b %d, %Y", "%b %d %Y"): |
| 2691 | try: |
| 2692 | parsed = datetime.strptime(match.group(1), pattern).replace(tzinfo=timezone.utc) |
| 2693 | except ValueError: |
| 2694 | continue |
| 2695 | return parsed if _is_quarter_end(parsed) else None |
| 2696 | return None |
| 2697 | |
| 2698 | |
| 2699 | def _eligible_for_nse_shareholding_reconciliation(profile: CompanyResearchProfile) -> bool: |
| 2700 | return ( |
| 2701 | profile.country.upper() in {"IN", "IND", "INDIA"} |
| 2702 | and profile.exchange.upper() in {"NSE", "XNSE"} |
| 2703 | and bool(profile.provider_instrument_ids.get("NSE")) |
| 2704 | ) |
| 2705 | |
| 2706 | |
| 2707 | def _shareholding_xbrl_enrichment_needed(snapshots: list[ShareholdingSnapshot]) -> bool: |
| 2708 | """Detect legacy structured filings without treating optional categories as required. |
| 2709 | |
| 2710 | Successful XBRL parsing persists at least one value with a stable |
| 2711 | ``nse-xbrl:`` locator. That durable provenance marker, rather than a |
| 2712 | complete category set, prevents repeat fetches when an official filing |
| 2713 | legitimately omits an optional ownership class. |
| 2714 | """ |
| 2715 | return any( |
| 2716 | snapshot.source_provider.upper() == "NSE" |
| 2717 | and snapshot.source_type == "NSE_SHAREHOLDING_XBRL" |
| 2718 | and not any((value.source_locator or "").startswith("nse-xbrl:") for value in snapshot.values) |
| 2719 | for snapshot in snapshots |
| 2720 | ) |
| 2721 | |
| 2722 | |
| 2723 | def _checked_categories(categories: set[str], shareholding_check_succeeded: bool) -> set[str]: |
| 2724 | """A provider outage must not advance the shareholding check schedule.""" |
| 2725 | if shareholding_check_succeeded or "SHAREHOLDING_PATTERN" not in categories: |
| 2726 | return categories |
| 2727 | return categories - {"SHAREHOLDING_PATTERN"} |
| 2728 | |
| 2729 | |
| 2730 | def _fair_official_filing_order(filings: list[DiscoveryResult]) -> list[DiscoveryResult]: |
| 2731 | """Interleave categories while preserving discovery order within each one.""" |
| 2732 | buckets: dict[str, list[DiscoveryResult]] = {} |
| 2733 | category_order: list[str] = [] |
| 2734 | for filing in filings: |
| 2735 | if filing.category not in buckets: |
| 2736 | buckets[filing.category] = [] |
| 2737 | category_order.append(filing.category) |
| 2738 | buckets[filing.category].append(filing) |
| 2739 | ordered: list[DiscoveryResult] = [] |
| 2740 | offsets = {category: 0 for category in category_order} |
| 2741 | while True: |
| 2742 | emitted = False |
| 2743 | for category in category_order: |
| 2744 | index = offsets[category] |
| 2745 | bucket = buckets[category] |
| 2746 | if index >= len(bucket): |
| 2747 | continue |
| 2748 | ordered.append(bucket[index]) |
| 2749 | offsets[category] = index + 1 |
| 2750 | emitted = True |
| 2751 | if not emitted: |
| 2752 | return ordered |
| 2753 | |
| 2754 | |
| 2755 | def _is_usable_durable_document(document: ResearchDocument) -> bool: |
| 2756 | """A durable official result suitable for reuse and category evidence.""" |
| 2757 | if document.source_mode != SourceMode.REAL: |
| 2758 | return False |
| 2759 | if document.status == DocumentStatus.PROCESSED: |
| 2760 | return True |
| 2761 | # Historical rows can legitimately be PARSED. Require persisted extracted |
| 2762 | # text so an empty/scanned/incomplete row remains retryable. |
| 2763 | return document.status == DocumentStatus.PARSED and bool((document.normalized_text or "").strip()) |
| 2764 | |
| 2765 | |
| 2766 | def _canonical_refresh_category(value: str) -> str: |
| 2767 | normalized = re.sub(r"[_&]+", " ", str(value or "").strip()).upper() |
| 2768 | normalized = re.sub(r"\s+", " ", normalized) |
| 2769 | return _REFRESH_CATEGORY_ALIASES.get(normalized, normalized) |
| 2770 | |
| 2771 | |
| 2772 | def _refresh_evidence_keys(category: str) -> tuple[str, ...]: |
| 2773 | return { |
| 2774 | "GROWTH": ("Growth", "GROWTH"), |
| 2775 | "CLIENTS": ("CLIENTS", "Customers", "New Customers"), |
| 2776 | "ORDERS_BACKLOG": ("ORDERS_BACKLOG", "Orders & Backlog", "New Orders"), |
| 2777 | "CAPEX": ("CAPEX", "CAPEX & Capacity"), |
| 2778 | "GUIDANCE": ("GUIDANCE", "Guidance"), |
| 2779 | "INSTITUTIONAL_ACTIVITY": ("INSTITUTIONAL_ACTIVITY", "Ownership"), |
| 2780 | "REGULATORY": ("REGULATORY", "Regulatory"), |
| 2781 | "MANAGEMENT": ("MANAGEMENT", "Management"), |
| 2782 | }.get(category, (category,)) |
| 2783 | |
| 2784 | |
| 2785 | def _search_discovery_categories(categories: set[str]) -> set[str]: |
| 2786 | """Translate canonical scheduling keys to the existing search vocabulary.""" |
| 2787 | display = { |
| 2788 | "GROWTH": "Growth", |
| 2789 | "CLIENTS": "Customers", |
| 2790 | "ORDERS_BACKLOG": "Orders & Backlog", |
| 2791 | "CAPEX": "CAPEX & Capacity", |
| 2792 | "GUIDANCE": "Guidance", |
| 2793 | "INSTITUTIONAL_ACTIVITY": "Ownership", |
| 2794 | "REGULATORY": "Regulatory", |
| 2795 | "MANAGEMENT": "Management", |
| 2796 | } |
| 2797 | return {display.get(_canonical_refresh_category(category), _canonical_refresh_category(category)) for category in categories} |
| 2798 | |
| 2799 | |
| 2800 | def _safe_url_path(url: str) -> str: |
| 2801 | try: |
| 2802 | return urlparse(url).path or "/" |
| 2803 | except ValueError: |
| 2804 | return "/" |
| 2805 | |
| 2806 | |
| 2807 | def _is_nse_official_document_url(url: str) -> bool: |
| 2808 | """Persisted-document repair may only reuse a known NSE archive host.""" |
| 2809 | try: |
| 2810 | host = (urlparse(url).hostname or "").lower() |
| 2811 | except ValueError: |
| 2812 | return False |
| 2813 | return host == "nseindia.com" or host.endswith(".nseindia.com") |
| 2814 | |
| 2815 | |
| 2816 | def _elapsed_ms(started: float) -> int: |
| 2817 | return round((time.monotonic() - started) * 1000) |
| 2818 | |
| 2819 | |
| 2820 | def _etf_document_relevant(profile: EtfResearchProfile, title: str | None, text: str) -> bool: |
| 2821 | haystack = f"{title or ''} {text}".lower() |
| 2822 | name_tokens = [token for token in re.findall(r"[a-z0-9]+", profile.fund_name.lower()) if len(token) >= 4] |
| 2823 | ticker_match = profile.ticker and re.search(rf"(?<![a-z0-9]){re.escape(profile.ticker.lower())}(?![a-z0-9])", haystack) |
| 2824 | index = profile.underlying_index or _infer_underlying_index(profile.fund_name, text) |
| 2825 | index_match = bool(index and index.lower() in haystack) |
| 2826 | token_matches = sum(1 for token in set(name_tokens) if token in haystack) |
| 2827 | fund_context = any(term in haystack for term in ["etf", "fund", "ucits", "factsheet", "holdings", "expense ratio", "aum"]) |
| 2828 | return fund_context and (token_matches >= 2 or bool(ticker_match) or index_match) |
| 2829 | |
| 2830 | |
| 2831 | def _infer_fund_provider(name: str, text: str) -> str | None: |
| 2832 | haystack = f"{name} {text}".lower() |
| 2833 | if "ishares" in haystack or "blackrock" in haystack: |
| 2834 | return "iShares" |
| 2835 | if "vanguard" in haystack: |
| 2836 | return "Vanguard" |
| 2837 | if "xtrackers" in haystack: |
| 2838 | return "Xtrackers" |
| 2839 | return None |
| 2840 | |
| 2841 | |
| 2842 | def _infer_underlying_index(name: str, text: str = "") -> str | None: |
| 2843 | haystack = f"{name} {text}".upper() |
| 2844 | if "S&P 500" in haystack or "SP 500" in haystack: |
| 2845 | return "S&P 500" |
| 2846 | if "NASDAQ 100" in haystack or "NASDAQ-100" in haystack: |
| 2847 | return "NASDAQ 100" |
| 2848 | return None |
| 2849 | |
| 2850 | |
| 2851 | def _extract_etf_facts(text: str) -> dict[str, object]: |
| 2852 | facts: dict[str, object] = {} |
| 2853 | lower = text.lower() |
| 2854 | expense = re.search(r"(?:expense ratio|ter|total expense ratio)\D{0,30}(\d+(?:\.\d+)?)\s?%", lower) |
| 2855 | if expense: |
| 2856 | facts["expenseRatio"] = f"{expense.group(1)}%" |
| 2857 | holdings = re.search(r"(?:holdings count|number of holdings|holdings)\D{0,30}(\d{2,5})", lower) |
| 2858 | if holdings: |
| 2859 | facts["holdingsCount"] = int(holdings.group(1)) |
| 2860 | dividend = re.search(r"(?:dividend yield|distribution yield)\D{0,30}(\d+(?:\.\d+)?)\s?%", lower) |
| 2861 | if dividend: |
| 2862 | facts["dividendYield"] = f"{dividend.group(1)}%" |
| 2863 | if "accumulating" in lower or "acc" in lower: |
| 2864 | facts["distributionPolicy"] = "Accumulating" |
| 2865 | elif "distributing" in lower or "dist" in lower: |
| 2866 | facts["distributionPolicy"] = "Distributing" |
| 2867 | aum = re.search(r"(?:aum|assets under management|fund size)\D{0,30}((?:EUR|USD|GBP)?\s?\d+(?:\.\d+)?\s?(?:bn|billion|mn|million))", text, re.IGNORECASE) |
| 2868 | if aum: |
| 2869 | facts["AUM"] = aum.group(1).strip() |
| 2870 | nav = re.search(r"\bNAV\D{0,20}((?:EUR|USD|GBP)?\s?\d+(?:\.\d+)?)", text, re.IGNORECASE) |
| 2871 | if nav: |
| 2872 | facts["NAV"] = nav.group(1).strip() |
| 2873 | if any(term in lower for term in ["apple", "microsoft", "nvidia", "amazon", "meta"]): |
| 2874 | top = [name for name in ["Apple", "Microsoft", "NVIDIA", "Amazon", "Meta"] if name.lower() in lower] |
| 2875 | facts["topHoldings"] = top[:10] |
| 2876 | return facts |
| 2877 | |
| 2878 | |
| 2879 | def _evidence_source(document: ResearchDocument) -> ResearchEvidenceSource: |
| 2880 | return ResearchEvidenceSource( |
| 2881 | publisher=document.publisher, |
| 2882 | url=document.canonical_url, |
| 2883 | source_type=document.source_classification, |
| 2884 | published_at=document.published_at, |
| 2885 | retrieved_at=document.retrieved_at, |
| 2886 | reliability=document.reliability_level, |
| 2887 | source_mode=document.source_mode, |
| 2888 | document_id=document.document_id, |
| 2889 | source_name=document.source_name, |
| 2890 | canonical_url=document.canonical_url, |
| 2891 | independent=document.duplicate_of_document_id is None, |
| 2892 | ) |