| 1 | from __future__ import annotations |
| 2 | |
| 3 | import logging |
| 4 | import re |
| 5 | from dataclasses import dataclass, field |
| 6 | from datetime import datetime, timezone |
| 7 | from decimal import Decimal, InvalidOperation |
| 8 | from typing import Protocol |
| 9 | from urllib.parse import urlparse |
| 10 | |
| 11 | import httpx |
| 12 | |
| 13 | from app.models import ( |
| 14 | CompanyResearchProfile, |
| 15 | DocumentSubtype, |
| 16 | EtfResearchProfile, |
| 17 | ReliabilityLevel, |
| 18 | ShareholdingCategory, |
| 19 | ShareholdingSnapshot, |
| 20 | ShareholdingSnapshotValue, |
| 21 | SourceClassification, |
| 22 | SourceMode, |
| 23 | SourceType, |
| 24 | ) |
| 25 | from app.entity_resolution import _contains_identity |
| 26 | from app.normalization import canonicalize_url |
| 27 | from app.source_registry import RegisteredResearchSource, approved_sources_for_categories |
| 28 | from app.url_security import UnsafeUrlError, validate_public_http_url |
| 29 | |
| 30 | logger = logging.getLogger(__name__) |
| 31 | |
| 32 | |
| 33 | @dataclass(frozen=True) |
| 34 | class DiscoveryResult: |
| 35 | category: str |
| 36 | source: RegisteredResearchSource |
| 37 | |
| 38 | |
| 39 | @dataclass(frozen=True) |
| 40 | class SearchDateWindow: |
| 41 | months: int = 12 |
| 42 | year: int | None = None |
| 43 | query_limit: int | None = None |
| 44 | explicit_queries: tuple[str, ...] | None = None |
| 45 | |
| 46 | |
| 47 | @dataclass(frozen=True) |
| 48 | class CandidateSearchResult: |
| 49 | title: str |
| 50 | url: str |
| 51 | snippet: str |
| 52 | discovered_at: datetime |
| 53 | provider: str |
| 54 | query_id: str |
| 55 | query: str |
| 56 | category: str |
| 57 | |
| 58 | |
| 59 | @dataclass(frozen=True) |
| 60 | class RejectedSearchCandidate: |
| 61 | url: str |
| 62 | reason: str |
| 63 | category: str |
| 64 | provider: str |
| 65 | |
| 66 | |
| 67 | @dataclass |
| 68 | class SearchDiscoveryStats: |
| 69 | candidate_count: int = 0 |
| 70 | accepted_count: int = 0 |
| 71 | rejected_count: int = 0 |
| 72 | categories_attempted: int = 0 |
| 73 | documents_fetched: int = 0 |
| 74 | events_extracted: int = 0 |
| 75 | provider_failure_count: int = 0 |
| 76 | zero_result_query_count: int = 0 |
| 77 | rejected_reasons: dict[str, int] = field(default_factory=dict) |
| 78 | |
| 79 | def reject(self, reason: str) -> None: |
| 80 | self.rejected_count += 1 |
| 81 | self.rejected_reasons[reason] = self.rejected_reasons.get(reason, 0) + 1 |
| 82 | |
| 83 | |
| 84 | class SearchDiscoveryProvider(Protocol): |
| 85 | provider_name: str |
| 86 | |
| 87 | async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]: |
| 88 | """Return candidate publisher URLs only. Search snippets are never investment evidence.""" |
| 89 | |
| 90 | |
| 91 | class DisabledSearchDiscoveryProvider: |
| 92 | provider_name = "disabled" |
| 93 | |
| 94 | async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]: |
| 95 | return [] |
| 96 | |
| 97 | |
| 98 | class SearchProviderConfigurationError(RuntimeError): |
| 99 | pass |
| 100 | |
| 101 | |
| 102 | class SearchProviderError(RuntimeError): |
| 103 | pass |
| 104 | |
| 105 | |
| 106 | class BraveCompatibleSearchDiscoveryProvider: |
| 107 | provider_name = "brave-compatible" |
| 108 | |
| 109 | def __init__( |
| 110 | self, |
| 111 | endpoint: str, |
| 112 | api_key: str, |
| 113 | *, |
| 114 | max_results_per_query: int = 5, |
| 115 | client: httpx.AsyncClient | None = None, |
| 116 | ) -> None: |
| 117 | if not endpoint: |
| 118 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:endpoint") |
| 119 | if not api_key: |
| 120 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:api_key") |
| 121 | self.endpoint = endpoint |
| 122 | self.api_key = api_key |
| 123 | self.max_results_per_query = min(max(max_results_per_query, 1), 20) |
| 124 | self._client = client or httpx.AsyncClient(timeout=httpx.Timeout(8.0, connect=3.0)) |
| 125 | |
| 126 | async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]: |
| 127 | results: list[CandidateSearchResult] = [] |
| 128 | for query_id, query in enumerate(_bounded_search_queries(company, category, date_window), start=1): |
| 129 | response = await _safe_search_get( |
| 130 | self._client, |
| 131 | self.endpoint, |
| 132 | params={"q": query, "count": self.max_results_per_query}, |
| 133 | headers={ |
| 134 | "Accept": "application/json", |
| 135 | "Accept-Encoding": "gzip", |
| 136 | "X-Subscription-Token": self.api_key, |
| 137 | }, |
| 138 | ) |
| 139 | payload = _safe_json(response) |
| 140 | web = payload.get("web", {}) |
| 141 | web_results = web.get("results", []) if isinstance(web, dict) else [] |
| 142 | if not isinstance(web_results, list): |
| 143 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") |
| 144 | for item in web_results[: self.max_results_per_query]: |
| 145 | if not isinstance(item, dict): |
| 146 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") |
| 147 | url = item.get("url") |
| 148 | if not url: |
| 149 | continue |
| 150 | results.append( |
| 151 | CandidateSearchResult( |
| 152 | title=str(item.get("title") or ""), |
| 153 | url=str(url), |
| 154 | snippet=str(item.get("description") or ""), |
| 155 | discovered_at=datetime.now(timezone.utc), |
| 156 | provider=self.provider_name, |
| 157 | query_id=f"{category}:{query_id}", |
| 158 | query=query, |
| 159 | category=category, |
| 160 | ) |
| 161 | ) |
| 162 | return results |
| 163 | |
| 164 | |
| 165 | class GoogleCompatibleSearchDiscoveryProvider: |
| 166 | provider_name = "google-compatible" |
| 167 | |
| 168 | def __init__( |
| 169 | self, |
| 170 | endpoint: str, |
| 171 | api_key: str, |
| 172 | engine_id: str, |
| 173 | *, |
| 174 | max_results_per_query: int = 5, |
| 175 | client: httpx.AsyncClient | None = None, |
| 176 | ) -> None: |
| 177 | if not endpoint: |
| 178 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:endpoint") |
| 179 | if not api_key: |
| 180 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:api_key") |
| 181 | if not engine_id: |
| 182 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:engine_id") |
| 183 | self.endpoint = endpoint |
| 184 | self.api_key = api_key |
| 185 | self.engine_id = engine_id |
| 186 | self.max_results_per_query = max_results_per_query |
| 187 | self._client = client or httpx.AsyncClient(timeout=httpx.Timeout(8.0, connect=3.0)) |
| 188 | |
| 189 | async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]: |
| 190 | results: list[CandidateSearchResult] = [] |
| 191 | for query_id, query in enumerate(_bounded_search_queries(company, category, date_window), start=1): |
| 192 | try: |
| 193 | response = await _safe_search_get( |
| 194 | self._client, |
| 195 | self.endpoint, |
| 196 | params={"q": query, "key": self.api_key, "cx": self.engine_id, "num": self.max_results_per_query}, |
| 197 | ) |
| 198 | except SearchProviderError as exc: |
| 199 | if str(exc) == "SEARCH_PROVIDER_FORBIDDEN": |
| 200 | raise SearchProviderError("GOOGLE_PROVIDER_UNAVAILABLE") from exc |
| 201 | raise |
| 202 | payload = _safe_json(response) |
| 203 | for item in payload.get("items", [])[: self.max_results_per_query]: |
| 204 | link = item.get("link") or item.get("url") |
| 205 | if not link: |
| 206 | continue |
| 207 | results.append( |
| 208 | CandidateSearchResult( |
| 209 | title=str(item.get("title") or ""), |
| 210 | url=str(link), |
| 211 | snippet=str(item.get("snippet") or ""), |
| 212 | discovered_at=datetime.now(timezone.utc), |
| 213 | provider=self.provider_name, |
| 214 | query_id=f"{category}:{query_id}", |
| 215 | query=query, |
| 216 | category=category, |
| 217 | ) |
| 218 | ) |
| 219 | return results |
| 220 | |
| 221 | |
| 222 | class SearxngSearchDiscoveryProvider: |
| 223 | provider_name = "searxng" |
| 224 | |
| 225 | def __init__( |
| 226 | self, |
| 227 | endpoint: str, |
| 228 | *, |
| 229 | max_results_per_query: int = 5, |
| 230 | client: httpx.AsyncClient | None = None, |
| 231 | ) -> None: |
| 232 | if not endpoint: |
| 233 | raise SearchProviderConfigurationError("SEARCH_PROVIDER_NOT_CONFIGURED:endpoint") |
| 234 | self.endpoint = endpoint |
| 235 | self.max_results_per_query = min(max(max_results_per_query, 1), 20) |
| 236 | self._client = client or httpx.AsyncClient(timeout=httpx.Timeout(8.0, connect=3.0)) |
| 237 | |
| 238 | async def discover(self, company: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[CandidateSearchResult]: |
| 239 | results: list[CandidateSearchResult] = [] |
| 240 | failures: list[str] = [] |
| 241 | self.last_query_degraded = False |
| 242 | for query_id, query in enumerate(_bounded_search_queries(company, category, date_window), start=1): |
| 243 | try: |
| 244 | response = await _safe_search_get( |
| 245 | self._client, |
| 246 | self.endpoint, |
| 247 | params={ |
| 248 | "q": query, |
| 249 | "format": "json", |
| 250 | "categories": "general", |
| 251 | "language": "en-US", |
| 252 | "safesearch": 0, |
| 253 | }, |
| 254 | ) |
| 255 | except SearchProviderError as exc: |
| 256 | failures.append(str(exc)) |
| 257 | logger.warning("search_query_failed company=%s category=%s provider=%s query_id=%s reason=%s", |
| 258 | _profile_display_name(company), category, self.provider_name, query_id, str(exc)) |
| 259 | continue |
| 260 | payload = _safe_json(response) |
| 261 | web_results = payload.get("results", []) |
| 262 | if not isinstance(web_results, list): |
| 263 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") |
| 264 | for item in web_results[: self.max_results_per_query]: |
| 265 | if not isinstance(item, dict): |
| 266 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") |
| 267 | url = item.get("url") |
| 268 | if not url: |
| 269 | continue |
| 270 | results.append( |
| 271 | CandidateSearchResult( |
| 272 | title=str(item.get("title") or ""), |
| 273 | url=str(url), |
| 274 | snippet=str(item.get("content") or item.get("snippet") or ""), |
| 275 | discovered_at=datetime.now(timezone.utc), |
| 276 | provider=self.provider_name, |
| 277 | query_id=f"{category}:{query_id}", |
| 278 | query=query, |
| 279 | category=category, |
| 280 | ) |
| 281 | ) |
| 282 | result_engines = sorted({ |
| 283 | str(engine) |
| 284 | for item in web_results if isinstance(item, dict) |
| 285 | for engine in ([item.get("engine")] if item.get("engine") else item.get("engines", [])) |
| 286 | if engine |
| 287 | }) |
| 288 | unresponsive_count = len(payload.get("unresponsive_engines", [])) |
| 289 | self.last_query_degraded = self.last_query_degraded or bool(unresponsive_count) |
| 290 | # An empty result set from engines that did not answer is not |
| 291 | # evidence that the company had no matching public information. |
| 292 | # Let the aggregate service treat this as retryable provider |
| 293 | # degradation rather than a successful zero-result check. |
| 294 | if not result_engines and unresponsive_count: |
| 295 | failures.append(f"SEARCH_PROVIDER_DEGRADED:unresponsive_engines={unresponsive_count}") |
| 296 | logger.warning( |
| 297 | "search_query_degraded company=%s category=%s provider=%s query_id=%s unresponsive_engine_count=%s", |
| 298 | _profile_display_name(company), category, self.provider_name, query_id, unresponsive_count, |
| 299 | ) |
| 300 | continue |
| 301 | logger.info( |
| 302 | "search_query_complete company=%s category=%s provider=%s query_id=%s query=%s result_count=%s engines=%s unresponsive_engine_count=%s", |
| 303 | _profile_display_name(company), category, self.provider_name, query_id, query, |
| 304 | len(web_results), result_engines, unresponsive_count, |
| 305 | ) |
| 306 | if failures and not results: |
| 307 | raise SearchProviderError(f"SEARCH_PROVIDER_UNAVAILABLE:{failures[-1]}") |
| 308 | return results |
| 309 | |
| 310 | |
| 311 | class ApprovedSourceDiscovery: |
| 312 | def discover( |
| 313 | self, |
| 314 | profile: CompanyResearchProfile, |
| 315 | missing_categories: set[str], |
| 316 | already_seen_urls: set[str], |
| 317 | ) -> list[DiscoveryResult]: |
| 318 | results: list[DiscoveryResult] = [] |
| 319 | for source in approved_sources_for_categories(profile.instrument_id, missing_categories): |
| 320 | if canonicalize_url(source.url) in already_seen_urls: |
| 321 | continue |
| 322 | validate_public_http_url(source.url) |
| 323 | host = source.host |
| 324 | if not host or not any(host == domain.lower() or host.endswith(f".{domain.lower()}") for domain in profile.known_domains): |
| 325 | continue |
| 326 | for category in sorted(set(source.categories) & missing_categories): |
| 327 | results.append(DiscoveryResult(category=category, source=source)) |
| 328 | return results |
| 329 | |
| 330 | |
| 331 | class OfficialFilingDiscovery: |
| 332 | """Exchange-first public filing discovery; search remains a fallback.""" |
| 333 | NSE_ANNOUNCEMENTS_URL = "https://www.nseindia.com/api/corporate-announcements" |
| 334 | |
| 335 | def __init__(self, client: httpx.AsyncClient | None = None, announcements_url: str | None = None) -> None: |
| 336 | self.announcements_url = announcements_url or self.NSE_ANNOUNCEMENTS_URL |
| 337 | self.client = client or httpx.AsyncClient(timeout=httpx.Timeout(12.0, connect=4.0), headers={ |
| 338 | "User-Agent": "Mozilla/5.0 (compatible; AIInvestmentResearch/1.0)", |
| 339 | "Accept": "application/json", "Referer": "https://www.nseindia.com/", |
| 340 | }) |
| 341 | |
| 342 | async def discover(self, profile: CompanyResearchProfile, categories: set[str], seen_urls: set[str]) -> list[DiscoveryResult]: |
| 343 | requested = {category for category in categories if category in {"FINANCIAL_RESULTS", "SHAREHOLDING_PATTERN"}} |
| 344 | if not _is_indian_nse_profile(profile) or not categories: |
| 345 | logger.info("official_discovery_result provider=NSE globalInstrumentId=%s status=SKIPPED exchange=%s country=%s reason=INELIGIBLE_PROFILE_OR_CATEGORY", profile.instrument_id, profile.exchange, profile.country) |
| 346 | return [] |
| 347 | # Profile mappings are hydrated from the global master with VERIFIED/ |
| 348 | # RESOLVED status only. Never turn a portfolio or broker display ticker |
| 349 | # into exchange identity here. |
| 350 | symbol = profile.provider_instrument_ids.get("NSE") |
| 351 | if not symbol: |
| 352 | logger.info("shareholding_identity_resolved provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=TRUSTED_NSE_MAPPING_UNAVAILABLE", profile.instrument_id) |
| 353 | return [] |
| 354 | symbol_source = "VERIFIED_NSE_MAPPING" |
| 355 | logger.info( |
| 356 | "official_discovery_start provider=NSE globalInstrumentId=%s exchange=%s country=%s " |
| 357 | "mappingAvailable=%s symbolSource=%s symbol=%r", |
| 358 | profile.instrument_id, |
| 359 | profile.exchange, |
| 360 | profile.country, |
| 361 | bool(profile.provider_instrument_ids.get("NSE")), |
| 362 | symbol_source, |
| 363 | symbol, |
| 364 | ) |
| 365 | try: |
| 366 | response = await self.client.get(self.announcements_url, params={"index": "equities", "symbol": symbol}) |
| 367 | response.raise_for_status() |
| 368 | rows = response.json() |
| 369 | if not isinstance(rows, list): |
| 370 | raise SearchProviderError("NSE_OFFICIAL_INVALID_RESPONSE") |
| 371 | candidates: list[tuple[int, datetime, DiscoveryResult]] = [] |
| 372 | candidate_urls = set(seen_urls) |
| 373 | for row in rows: |
| 374 | if not isinstance(row, dict): |
| 375 | continue |
| 376 | title = " ".join(str(row.get(key) or "") for key in ("desc", "attchmntText")) |
| 377 | url = str(row.get("attchmntFile") or "") |
| 378 | subtype = classify_nse_document_subtype( |
| 379 | desc=row.get("desc"), |
| 380 | attachment_text=row.get("attchmntText"), |
| 381 | attachment_file=row.get("attchmntFile"), |
| 382 | ) |
| 383 | category = "FINANCIAL_RESULTS" if _is_financial_result_announcement(title) and "FINANCIAL_RESULTS" in requested else ( |
| 384 | "SHAREHOLDING_PATTERN" if _is_shareholding_announcement(title) and "SHAREHOLDING_PATTERN" in requested else None |
| 385 | ) |
| 386 | # Preserve required filing selection and add only attachments |
| 387 | # whose *pre-download NSE metadata* carries a high-confidence |
| 388 | # subtype. Generic announcements remain excluded. |
| 389 | if category is None and subtype is not None: |
| 390 | category = subtype.value |
| 391 | if not url or category is None: |
| 392 | continue |
| 393 | try: |
| 394 | canonical = canonicalize_url(url) |
| 395 | validate_public_http_url(canonical) |
| 396 | except ValueError as exc: |
| 397 | logger.warning( |
| 398 | "official_candidate_rejected provider=NSE globalInstrumentId=%s reason=%s host=%s", |
| 399 | profile.instrument_id, |
| 400 | "UNSAFE_URL" if isinstance(exc, UnsafeUrlError) else "MALFORMED_URL", |
| 401 | _safe_attachment_host(url), |
| 402 | ) |
| 403 | continue |
| 404 | if canonical in candidate_urls: |
| 405 | continue |
| 406 | candidate_urls.add(canonical) |
| 407 | published = _nse_datetime(row.get("an_dt")) |
| 408 | source = RegisteredResearchSource( |
| 409 | source_id=f"nse:{category.lower()}:{profile.instrument_id}:{content_hash_key(canonical)}", |
| 410 | instrument_id=profile.instrument_id, url=canonical, source_type=SourceType.EXCHANGE_ANNOUNCEMENT, |
| 411 | source_classification=SourceClassification.EXCHANGE, source_name="NSE corporate announcements", |
| 412 | publisher="NSE", reliability_level=ReliabilityLevel.LEVEL_A, domain=urlparse(canonical).hostname or "", |
| 413 | company_id=profile.company_id, allowed=True, discovery_method="NSE_OFFICIAL_API", |
| 414 | priority=_nse_discovery_priority(subtype, category), categories=(category,), |
| 415 | document_subtype=subtype, official_nse_profile_symbol=symbol, |
| 416 | ) |
| 417 | candidates.append((_nse_discovery_priority(subtype, category), published, DiscoveryResult(category=category, source=source))) |
| 418 | accepted = [item for _, _, item in sorted(candidates, key=lambda value: (value[0], -value[1].timestamp()))] |
| 419 | logger.info("official_filing_discovery provider=NSE globalInstrumentId=%s status=%s candidateCount=%s acceptedCount=%s reason=%s", profile.instrument_id, "SUCCESS" if accepted else "ZERO_CANDIDATES", len(rows), len(accepted), "NONE" if accepted else "NO_QUALIFYING_ATTACHMENT") |
| 420 | return accepted |
| 421 | except Exception as exc: |
| 422 | logger.warning( |
| 423 | "official_discovery_result provider=NSE globalInstrumentId=%s status=FAILED candidateCount=0 acceptedCount=0 reason=%s", |
| 424 | profile.instrument_id, |
| 425 | type(exc).__name__, |
| 426 | ) |
| 427 | raise |
| 428 | |
| 429 | |
| 430 | class OfficialNseShareholdingDiscovery: |
| 431 | """Read NSE's dedicated quarterly shareholding-pattern feed. |
| 432 | |
| 433 | This is intentionally separate from corporate announcements: NSE publishes |
| 434 | Regulation 31 shareholding data as a dedicated corporate-filings product, |
| 435 | with an official XBRL artifact per reported period. |
| 436 | """ |
| 437 | |
| 438 | NSE_SHAREHOLDINGS_URL = "https://www.nseindia.com/api/corporate-share-holdings-master" |
| 439 | |
| 440 | def __init__(self, client: httpx.AsyncClient | None = None, shareholdings_url: str | None = None) -> None: |
| 441 | self.shareholdings_url = shareholdings_url or self.NSE_SHAREHOLDINGS_URL |
| 442 | self.client = client or httpx.AsyncClient(timeout=httpx.Timeout(12.0, connect=4.0), headers={ |
| 443 | "User-Agent": "Mozilla/5.0 (compatible; AIInvestmentResearch/1.0)", |
| 444 | "Accept": "application/json", "Referer": "https://www.nseindia.com/", |
| 445 | }) |
| 446 | |
| 447 | async def discover(self, profile: CompanyResearchProfile) -> list[ShareholdingSnapshot]: |
| 448 | if not _is_indian_nse_profile(profile): |
| 449 | logger.info("shareholding_identity_resolved provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=INELIGIBLE_PROFILE", profile.instrument_id) |
| 450 | return [] |
| 451 | # This mapping is hydrated only after the global-provider mapping |
| 452 | # layer has excluded broker-import identities. There is deliberately |
| 453 | # no profile-ticker or portfolio-alias fallback here. |
| 454 | symbol = profile.provider_instrument_ids.get("NSE") |
| 455 | if not symbol: |
| 456 | logger.info("shareholding_identity_resolved provider=NSE globalInstrumentId=%s outcome=SKIPPED reason=TRUSTED_NSE_MAPPING_UNAVAILABLE", profile.instrument_id) |
| 457 | return [] |
| 458 | |
| 459 | logger.info("shareholding_official_discovery provider=NSE globalInstrumentId=%s symbol=%r outcome=START", profile.instrument_id, symbol) |
| 460 | try: |
| 461 | response = await self.client.get(self.shareholdings_url, params={"index": "equities", "symbol": symbol}) |
| 462 | response.raise_for_status() |
| 463 | rows = response.json() |
| 464 | if not isinstance(rows, list): |
| 465 | raise SearchProviderError("NSE_SHAREHOLDING_OFFICIAL_INVALID_RESPONSE") |
| 466 | except httpx.HTTPError as exc: |
| 467 | logger.warning("shareholding_official_discovery provider=NSE globalInstrumentId=%s outcome=UNAVAILABLE reason=%s", profile.instrument_id, type(exc).__name__) |
| 468 | raise SearchProviderError("NSE_SHAREHOLDING_OFFICIAL_UNAVAILABLE") from exc |
| 469 | |
| 470 | candidates = [snapshot for row in rows if isinstance(row, dict) |
| 471 | if (snapshot := _nse_shareholding_snapshot(profile, symbol, row)) is not None] |
| 472 | candidates.sort(key=lambda snapshot: (snapshot.period_end, snapshot.published_at or datetime.min.replace(tzinfo=timezone.utc)), reverse=True) |
| 473 | # Keep the latest four distinct reporting periods, but retain every |
| 474 | # official record selected for those periods. A revised filing has its |
| 475 | # own recordId and must remain durably distinguishable from the filing |
| 476 | # it supersedes; the read model selects the latest publication. |
| 477 | periods: set[datetime] = set() |
| 478 | for candidate in candidates: |
| 479 | periods.add(candidate.period_end) |
| 480 | if len(periods) == 4: |
| 481 | break |
| 482 | snapshots = [candidate for candidate in candidates if candidate.period_end in periods] |
| 483 | logger.info("shareholding_official_discovery provider=NSE globalInstrumentId=%s outcome=%s rowCount=%s snapshotCount=%s", profile.instrument_id, "SUCCESS" if snapshots else "ZERO_RESULTS", len(rows), len(snapshots)) |
| 484 | return snapshots |
| 485 | |
| 486 | |
| 487 | def _is_indian_nse_profile(profile: CompanyResearchProfile) -> bool: |
| 488 | return profile.country.upper() in {"IN", "IND", "INDIA"} and profile.exchange.upper() in {"NSE", "XNSE"} |
| 489 | |
| 490 | |
| 491 | def _is_financial_result_announcement(value: str) -> bool: |
| 492 | lowered = value.lower() |
| 493 | return any(term in lowered for term in ("financial results", "financial result", "unaudited financial", "audited financial", "results for the period ended")) |
| 494 | |
| 495 | |
| 496 | def _is_shareholding_announcement(value: str) -> bool: |
| 497 | # NSE announcements carry many ordinary ownership references. Only the |
| 498 | # explicit quarterly filing name is a SHAREHOLDING_PATTERN artifact. |
| 499 | normalized = re.sub(r"\s+", " ", value.casefold()).strip() |
| 500 | return bool(re.search(r"\bshare(?:holding|\s+holder)\s+pattern\b", normalized)) |
| 501 | |
| 502 | |
| 503 | def classify_nse_document_subtype( |
| 504 | *, |
| 505 | desc: object = None, |
| 506 | attachment_text: object = None, |
| 507 | attachment_file: object = None, |
| 508 | ) -> DocumentSubtype | None: |
| 509 | """Classify only explicit, pre-download NSE announcement metadata. |
| 510 | |
| 511 | A subtype is an optional selection hint, never a substitute for document |
| 512 | identity or post-download evidence extraction. The deliberately narrow |
| 513 | rules fail closed for notices and isolated keywords. |
| 514 | """ |
| 515 | normalized = _nse_metadata_text(desc, attachment_text, attachment_file) |
| 516 | if not normalized: |
| 517 | return None |
| 518 | # Single deterministic precedence for metadata containing more than one |
| 519 | # signal. A substantive transcript/presentation outranks a notice. |
| 520 | if _has_any_phrase(normalized, ( |
| 521 | "conference call transcript", "earnings call transcript", "analyst call transcript", |
| 522 | "concall transcript", "conference call presentation", "earnings call presentation", |
| 523 | )): |
| 524 | return DocumentSubtype.CONFERENCE_CALL_MATERIAL |
| 525 | if _has_any_phrase(normalized, ("agm presentation", "egm presentation", "shareholder meeting presentation")): |
| 526 | return None |
| 527 | if _has_any_phrase(normalized, ( |
| 528 | "investor presentation", "investors presentation", "corporate presentation", |
| 529 | "earnings presentation", "results presentation", "analyst presentation", |
| 530 | )): |
| 531 | return DocumentSubtype.INVESTOR_PRESENTATION |
| 532 | if _has_any_phrase(normalized, ("statutory notice", "compliance notice", "compliance certificate", "trading window")): |
| 533 | return None |
| 534 | if _has_any_phrase(normalized, ("investor release", "press release", "media release", "earnings release", "business update")): |
| 535 | return DocumentSubtype.INVESTOR_RELEASE |
| 536 | if _has_any_phrase(normalized, ( |
| 537 | "receipt of order", "receipt of orders", "received an order", "order received", |
| 538 | "order win", "letter of award", "work order", "contract awarded", "award of contract", |
| 539 | )): |
| 540 | return DocumentSubtype.ORDER_CONTRACT_DISCLOSURE |
| 541 | if _has_any_phrase(normalized, ( |
| 542 | "capacity expansion", "commissioning of", "new manufacturing facility", "new plant", |
| 543 | "greenfield expansion", "brownfield expansion", "expansion project", |
| 544 | )): |
| 545 | return DocumentSubtype.CAPEX_CAPACITY_DISCLOSURE |
| 546 | return None |
| 547 | |
| 548 | |
| 549 | def _nse_metadata_text(*values: object) -> str: |
| 550 | joined = " ".join(str(value or "") for value in values) |
| 551 | joined = re.sub(r"[_./\\-]+", " ", joined.casefold()) |
| 552 | return re.sub(r"\s+", " ", joined).strip() |
| 553 | |
| 554 | |
| 555 | def _has_any_phrase(value: str, phrases: tuple[str, ...]) -> bool: |
| 556 | return any(phrase in value for phrase in phrases) |
| 557 | |
| 558 | |
| 559 | def _nse_discovery_priority(subtype: DocumentSubtype | None, category: str) -> int: |
| 560 | return { |
| 561 | DocumentSubtype.CONFERENCE_CALL_MATERIAL: 0, |
| 562 | DocumentSubtype.INVESTOR_PRESENTATION: 1, |
| 563 | DocumentSubtype.INVESTOR_RELEASE: 2, |
| 564 | DocumentSubtype.ORDER_CONTRACT_DISCLOSURE: 3, |
| 565 | DocumentSubtype.CAPEX_CAPACITY_DISCLOSURE: 4, |
| 566 | }.get(subtype, 5 if category == "FINANCIAL_RESULTS" else 6) |
| 567 | |
| 568 | |
| 569 | def _nse_datetime(value: object) -> datetime: |
| 570 | try: |
| 571 | return datetime.strptime(str(value), "%d-%b-%Y %H:%M:%S").replace(tzinfo=timezone.utc) |
| 572 | except ValueError: |
| 573 | return datetime.min.replace(tzinfo=timezone.utc) |
| 574 | |
| 575 | |
| 576 | def _nse_shareholding_snapshot( |
| 577 | profile: CompanyResearchProfile, |
| 578 | trusted_symbol: str, |
| 579 | row: dict[object, object], |
| 580 | ) -> ShareholdingSnapshot | None: |
| 581 | # A server response must still identify the requested trusted NSE symbol. |
| 582 | # The verified mapping, not a broker alias or a fuzzy company-name match, |
| 583 | # is the identity proof for this dedicated NSE feed. |
| 584 | if str(row.get("symbol") or "").strip().upper() != trusted_symbol.strip().upper(): |
| 585 | return None |
| 586 | # NSE calls this field "As on Date" in the dedicated Shareholding Pattern |
| 587 | # table. This quarterly feature must not treat an arbitrary special-event |
| 588 | # as-on date as a Regulation 31 quarter. The feed exposes no separate |
| 589 | # reporting-period field for such rows, so do not round or infer one. |
| 590 | period_end = _nse_shareholding_date(row.get("date")) |
| 591 | if period_end is not None and not _is_nse_quarter_end(period_end): |
| 592 | logger.info( |
| 593 | "shareholding_official_discovery provider=NSE globalInstrumentId=%s outcome=REJECTED reason=NON_QUARTER_REPORTING_DATE recordId=%s asOnDate=%s", |
| 594 | profile.instrument_id, |
| 595 | str(row.get("recordId") or "NONE"), |
| 596 | period_end.date().isoformat(), |
| 597 | ) |
| 598 | return None |
| 599 | xbrl_url = str(row.get("xbrl") or "").strip() |
| 600 | if period_end is None or not xbrl_url: |
| 601 | return None |
| 602 | try: |
| 603 | canonical_url = canonicalize_url(xbrl_url) |
| 604 | validate_public_http_url(canonical_url) |
| 605 | except ValueError: |
| 606 | return None |
| 607 | values = _nse_shareholding_values(row) |
| 608 | if not values: |
| 609 | return None |
| 610 | identity = str(row.get("recordId") or "").strip() or canonical_url |
| 611 | published_at = _nse_shareholding_date(row.get("submissionDate")) or _nse_datetime(row.get("broadcastDate")) |
| 612 | return ShareholdingSnapshot( |
| 613 | instrument_id=profile.instrument_id, |
| 614 | period_end=period_end, |
| 615 | filing_basis=str(row.get("typeOfSubmission") or "").strip() or None, |
| 616 | source_provider="NSE", |
| 617 | source_type="NSE_SHAREHOLDING_XBRL", |
| 618 | source_identity_key=f"NSE_SHAREHOLDING:{identity}", |
| 619 | source_url=canonical_url, |
| 620 | published_at=published_at, |
| 621 | confidence=Decimal("0.95"), |
| 622 | reliability_level=ReliabilityLevel.LEVEL_A, |
| 623 | source_mode=SourceMode.REAL, |
| 624 | values=values, |
| 625 | ) |
| 626 | |
| 627 | |
| 628 | def _nse_shareholding_values(row: dict[object, object]) -> list[ShareholdingSnapshotValue]: |
| 629 | # Only the promoter-and-promoter-group field has a matching durable |
| 630 | # ownership category. NSE's aggregate public shareholding includes several |
| 631 | # possible institutional and non-retail classes, so it must not be coerced |
| 632 | # into PUBLIC_RETAIL (or any other detailed category). |
| 633 | fields = ( |
| 634 | ("pr_and_prgrp", "Promoter and Promoter Group", ShareholdingCategory.PROMOTER), |
| 635 | ) |
| 636 | values: list[ShareholdingSnapshotValue] = [] |
| 637 | for field, label, category in fields: |
| 638 | percentage = _nse_percentage(row.get(field)) |
| 639 | if percentage is None: |
| 640 | continue |
| 641 | values.append(ShareholdingSnapshotValue( |
| 642 | category=category, |
| 643 | percentage=percentage, |
| 644 | raw_source_label=label, |
| 645 | source_locator=f"nse-shareholdings-master:{field}", |
| 646 | evidence_text=f"{label}: {percentage}%", |
| 647 | )) |
| 648 | return values |
| 649 | |
| 650 | |
| 651 | def _nse_percentage(value: object) -> Decimal | None: |
| 652 | try: |
| 653 | percentage = Decimal(str(value).strip()) |
| 654 | except (InvalidOperation, ValueError): |
| 655 | return None |
| 656 | return percentage if Decimal("0") <= percentage <= Decimal("100") else None |
| 657 | |
| 658 | |
| 659 | def _nse_shareholding_date(value: object) -> datetime | None: |
| 660 | raw = str(value or "").strip() |
| 661 | for fmt in ("%d-%b-%Y", "%d-%b-%Y %H:%M:%S", "%d-%m-%Y", "%d/%m/%Y"): |
| 662 | try: |
| 663 | return datetime.strptime(raw, fmt).replace(tzinfo=timezone.utc) |
| 664 | except ValueError: |
| 665 | continue |
| 666 | return None |
| 667 | |
| 668 | |
| 669 | def _is_nse_quarter_end(value: datetime) -> bool: |
| 670 | return (value.month, value.day) in {(3, 31), (6, 30), (9, 30), (12, 31)} |
| 671 | |
| 672 | |
| 673 | def content_hash_key(value: str) -> str: |
| 674 | return re.sub(r"[^a-z0-9]", "", value.lower())[-32:] |
| 675 | |
| 676 | |
| 677 | def _safe_attachment_host(url: str) -> str: |
| 678 | try: |
| 679 | return (urlparse(url).hostname or "").lower() |
| 680 | except ValueError: |
| 681 | return "" |
| 682 | |
| 683 | |
| 684 | class SearchDiscoveryService: |
| 685 | def __init__( |
| 686 | self, |
| 687 | provider: SearchDiscoveryProvider, |
| 688 | *, |
| 689 | max_queries_per_category: int = 6, |
| 690 | max_results_per_query: int = 5, |
| 691 | max_documents_per_refresh: int = 6, |
| 692 | allowed_domains: list[str] | None = None, |
| 693 | ) -> None: |
| 694 | self.provider = provider |
| 695 | self.max_queries_per_category = max_queries_per_category |
| 696 | self.max_results_per_query = max_results_per_query |
| 697 | self.max_documents_per_refresh = max_documents_per_refresh |
| 698 | self.allowed_domains = {domain.lower() for domain in (allowed_domains or []) if domain} |
| 699 | self.last_stats = SearchDiscoveryStats() |
| 700 | self.rejected_candidates: list[RejectedSearchCandidate] = [] |
| 701 | |
| 702 | async def discover(self, profile: CompanyResearchProfile | EtfResearchProfile, missing_categories: set[str], already_seen_urls: set[str]) -> list[DiscoveryResult]: |
| 703 | stats = SearchDiscoveryStats() |
| 704 | rejected: list[RejectedSearchCandidate] = [] |
| 705 | accepted: list[DiscoveryResult] = [] |
| 706 | seen = set(already_seen_urls) |
| 707 | window = SearchDateWindow(year=datetime.now(timezone.utc).year, query_limit=self.max_queries_per_category) |
| 708 | per_category_limit = max(1, self.max_documents_per_refresh // max(len(missing_categories), 1)) |
| 709 | provider_errors: list[str] = [] |
| 710 | for category in sorted(missing_categories): |
| 711 | accepted_for_category = 0 |
| 712 | stats.categories_attempted += 1 |
| 713 | try: |
| 714 | candidates = await self.provider.discover(profile, category, window) |
| 715 | except SearchProviderError as exc: |
| 716 | stats.provider_failure_count += 1 |
| 717 | if not provider_errors: |
| 718 | stats.reject(str(exc)) |
| 719 | provider_errors.append(str(exc)) |
| 720 | logger.warning("search_category_failed company=%s category=%s provider=%s reason=%s", |
| 721 | _profile_display_name(profile), category, self.provider.provider_name, str(exc)) |
| 722 | continue |
| 723 | if not candidates: |
| 724 | stats.zero_result_query_count += 1 |
| 725 | # Search ranking is not filing selection. Prefer an actual result |
| 726 | # document from an exchange/company over a generic IR landing page, |
| 727 | # then let the report parser choose the latest fiscal period. |
| 728 | ranked_candidates = sorted( |
| 729 | candidates[: self.max_queries_per_category * self.max_results_per_query], |
| 730 | key=lambda candidate: _candidate_rank(profile, candidate), reverse=True, |
| 731 | ) |
| 732 | for candidate in ranked_candidates: |
| 733 | stats.candidate_count += 1 |
| 734 | try: |
| 735 | source = _candidate_to_source(self, profile, candidate) |
| 736 | except ValueError as exc: |
| 737 | reason = _discovery_rejection_reason(str(exc)) |
| 738 | stats.reject(reason) |
| 739 | rejected.append(RejectedSearchCandidate(candidate.url, reason, category, candidate.provider)) |
| 740 | continue |
| 741 | canonical = canonicalize_url(source.url) |
| 742 | if canonical in seen: |
| 743 | stats.reject("DUPLICATE") |
| 744 | rejected.append(RejectedSearchCandidate(candidate.url, "DUPLICATE", category, candidate.provider)) |
| 745 | continue |
| 746 | seen.add(canonical) |
| 747 | accepted.append(DiscoveryResult(category=category, source=source)) |
| 748 | stats.accepted_count += 1 |
| 749 | accepted_for_category += 1 |
| 750 | if len(accepted) >= self.max_documents_per_refresh: |
| 751 | break |
| 752 | if accepted_for_category >= per_category_limit: |
| 753 | break |
| 754 | if len(accepted) >= self.max_documents_per_refresh: |
| 755 | break |
| 756 | accepted.sort(key=lambda result: _source_rank(result.source.source_classification)) |
| 757 | self.last_stats = stats |
| 758 | self.rejected_candidates = rejected |
| 759 | logger.info( |
| 760 | "search_discovery_complete company=%s provider=%s categories=%s candidate_count=%s accepted_count=%s rejected_count=%s provider_failures=%s zero_result_categories=%s rejected_reasons=%s", |
| 761 | _profile_display_name(profile), |
| 762 | self.provider.provider_name, |
| 763 | stats.categories_attempted, |
| 764 | stats.candidate_count, |
| 765 | stats.accepted_count, |
| 766 | stats.rejected_count, |
| 767 | stats.provider_failure_count, |
| 768 | stats.zero_result_query_count, |
| 769 | stats.rejected_reasons, |
| 770 | ) |
| 771 | if not accepted and provider_errors and stats.provider_failure_count == stats.categories_attempted: |
| 772 | raise SearchProviderError(f"SEARCH_PROVIDER_UNAVAILABLE:{provider_errors[-1]}") |
| 773 | return accepted |
| 774 | |
| 775 | |
| 776 | def _candidate_rank(profile: CompanyResearchProfile | EtfResearchProfile, candidate: CandidateSearchResult) -> tuple[int, int, int, datetime]: |
| 777 | haystack = f"{candidate.title} {candidate.snippet} {candidate.url}".lower() |
| 778 | host = (urlparse(candidate.url).hostname or "").lower() |
| 779 | official = any(host == domain.lower() or host.endswith(f".{domain.lower()}") for domain in profile.known_domains) |
| 780 | exchange = any(token in host for token in ("nseindia", "bseindia", "euronext", "deutsche-boerse", "londonstockexchange", "sec.gov")) |
| 781 | filing_terms = ("quarterly result", "financial result", "unaudited result", "audited result", "earnings release", "results for quarter", "quarter ended", "financial statement") |
| 782 | ownership_terms = ("shareholding pattern", "shareholder pattern", "promoter", "fii", "fpi", "dii") |
| 783 | is_pdf = ".pdf" in candidate.url.lower() or " pdf" in haystack |
| 784 | relevant = any(term in haystack for term in filing_terms if candidate.category == "FINANCIAL_RESULTS") or any( |
| 785 | term in haystack for term in ownership_terms if candidate.category in {"Ownership", "INSTITUTIONAL_ACTIVITY"} |
| 786 | ) |
| 787 | return (3 if exchange else 2 if official else 0, int(relevant), int(is_pdf), candidate.discovered_at) |
| 788 | |
| 789 | |
| 790 | def _candidate_to_source(service: SearchDiscoveryService, profile: CompanyResearchProfile | EtfResearchProfile, candidate: CandidateSearchResult) -> RegisteredResearchSource: |
| 791 | try: |
| 792 | validate_public_http_url(candidate.url) |
| 793 | except ValueError as exc: |
| 794 | raise ValueError("DOMAIN_VALIDATION_FAILED") from exc |
| 795 | canonical = canonicalize_url(candidate.url) |
| 796 | host = (urlparse(canonical).hostname or "").lower() |
| 797 | if service.allowed_domains and not any(host == domain or host.endswith(f".{domain}") for domain in service.allowed_domains): |
| 798 | raise ValueError("DOMAIN_VALIDATION_FAILED") |
| 799 | classification = classify_source(host, profile, candidate) |
| 800 | if isinstance(profile, CompanyResearchProfile) and _candidate_issuer_relevance(profile, candidate, host) == "NEGATIVE": |
| 801 | raise ValueError("COMPANY_RELEVANCE_FAILED") |
| 802 | reliability = reliability_for_classification(classification) |
| 803 | source_type = source_type_for_classification(classification) |
| 804 | priority = 2 if classification in {SourceClassification.OFFICIAL_COMPANY, SourceClassification.REGULATORY, SourceClassification.EXCHANGE} else 4 |
| 805 | if classification == SourceClassification.OFFICIAL_COMPANY and host not in {domain.lower() for domain in profile.known_domains}: |
| 806 | profile.known_domains.append(host) |
| 807 | return RegisteredResearchSource( |
| 808 | source_id=f"search:{candidate.provider}:{candidate.query_id}:{host}", instrument_id=profile.instrument_id, |
| 809 | url=canonical, source_type=source_type, source_classification=classification, |
| 810 | source_name=f"{candidate.provider} discovered publisher page", publisher=publisher_from_host(host), |
| 811 | reliability_level=reliability, domain=host, company_id=getattr(profile, "company_id", getattr(profile, "fund_id", None)), |
| 812 | allowed=True, discovery_method="SEARCH_DISCOVERY", priority=priority, categories=(candidate.category,), |
| 813 | ) |
| 814 | |
| 815 | |
| 816 | def _candidate_issuer_relevance(profile: CompanyResearchProfile, candidate: CandidateSearchResult, host: str) -> str: |
| 817 | """Return positive, neutral, or contradictory issuer evidence from search metadata. |
| 818 | |
| 819 | Search metadata is not document evidence. It is only used to stop a |
| 820 | clearly named different issuer from being registered under this profile. |
| 821 | Exchange and regulatory hosts are deliberately neutral: they publish for |
| 822 | many issuers and therefore cannot establish company relevance by domain. |
| 823 | """ |
| 824 | if any(host == domain.lower() or host.endswith(f".{domain.lower()}") for domain in profile.known_domains): |
| 825 | return "POSITIVE" |
| 826 | |
| 827 | evidence = f"{candidate.title} {candidate.snippet} {candidate.url}" |
| 828 | if _candidate_has_positive_issuer_evidence(profile, evidence): |
| 829 | return "POSITIVE" |
| 830 | if _candidate_names_other_issuer(evidence): |
| 831 | return "NEGATIVE" |
| 832 | return "NEUTRAL" |
| 833 | |
| 834 | |
| 835 | def _candidate_has_positive_issuer_evidence(profile: CompanyResearchProfile, evidence: str) -> bool: |
| 836 | normalized_isin = _normalized_issuer_token(profile.isin) |
| 837 | normalized_evidence = _normalized_issuer_token(evidence) |
| 838 | if normalized_isin and normalized_isin in normalized_evidence: |
| 839 | return True |
| 840 | |
| 841 | identities = [profile.company_name, *profile.aliases, *profile.provider_instrument_ids.values()] |
| 842 | return any( |
| 843 | identity and len(identity.strip()) >= 4 and _contains_identity(evidence.casefold(), identity.casefold()) |
| 844 | for identity in identities |
| 845 | ) |
| 846 | |
| 847 | |
| 848 | def _candidate_names_other_issuer(evidence: str) -> bool: |
| 849 | """Recognize an explicitly named corporate issuer without guessing one.""" |
| 850 | return re.search( |
| 851 | r"\b[A-Z][A-Z&.' -]{3,}?\s+(?:LIMITED|LTD\.?|INC\.?|CORPORATION|CORP\.?|PLC|P\.L\.C\.)\b", |
| 852 | evidence, |
| 853 | flags=re.IGNORECASE, |
| 854 | ) is not None |
| 855 | |
| 856 | |
| 857 | def _normalized_issuer_token(value: str | None) -> str: |
| 858 | return re.sub(r"[^a-z0-9]", "", str(value or "").casefold()) |
| 859 | |
| 860 | |
| 861 | def generate_search_queries(profile: CompanyResearchProfile | EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[str]: |
| 862 | if isinstance(profile, EtfResearchProfile): |
| 863 | return generate_etf_search_queries(profile, category, date_window) |
| 864 | year = date_window.year or datetime.now(timezone.utc).year |
| 865 | names = [profile.company_name, *profile.aliases, profile.ticker] |
| 866 | identities = _dedupe_identity_terms(names)[:3] |
| 867 | broad_patterns = [ |
| 868 | "investor relations", |
| 869 | "annual report", |
| 870 | "quarterly results", |
| 871 | "earnings", |
| 872 | "guidance", |
| 873 | "orders backlog", |
| 874 | "contracts", |
| 875 | "capex", |
| 876 | "acquisition", |
| 877 | "clients", |
| 878 | "analyst rating", |
| 879 | "target price", |
| 880 | "institutional ownership", |
| 881 | "valuation", |
| 882 | "news", |
| 883 | ] |
| 884 | category_patterns = { |
| 885 | "FINANCIAL_RESULTS": ["financial results", "quarterly results", "earnings", "annual report"], |
| 886 | "GUIDANCE": ["guidance", "outlook", "guidance raised", "guidance cut"], |
| 887 | "ORDERS_BACKLOG": ["order intake", "order backlog", "orders", "backlog"], |
| 888 | "CONTRACTS": ["contracts", "contract award", "framework agreement"], |
| 889 | "CAPEX": ["capex", "capital expenditure", "investment"], |
| 890 | "NEW_FACILITIES": ["new facility", "new plant", "capacity expansion", "factory investment"], |
| 891 | "ACQUISITIONS": ["acquisition", "merger", "divestment"], |
| 892 | "CLIENTS": ["clients", "customers", "customer contract", "customer win"], |
| 893 | "PRODUCTS": ["products", "product launch"], |
| 894 | "MANAGEMENT": ["management", "executive board", "CEO", "CFO"], |
| 895 | "ANALYST_OPINION": ["analyst rating", "analyst opinion", "broker rating"], |
| 896 | "ANALYST_TARGETS": ["analyst target price", "target price", "price target"], |
| 897 | "INSTITUTIONAL_ACTIVITY": ["institutional ownership", "shareholder structure", "major shareholders"], |
| 898 | "VALUATION": ["valuation", "multiples", "market capitalization"], |
| 899 | "RISKS": ["risks", "risk factors"], |
| 900 | "CATALYSTS": ["catalysts", "news", "contracts", "guidance"], |
| 901 | "Customers": [ |
| 902 | "new customer", |
| 903 | "customer order", |
| 904 | "customer win", |
| 905 | "customer contract", |
| 906 | "customer qualification", |
| 907 | "strategic customer", |
| 908 | "customer loss", |
| 909 | ], |
| 910 | "Growth": ["earnings results", "revenue growth", "profit margins", "order growth", "market expansion", "product ramp"], |
| 911 | "Orders & Backlog": ["order wins", "order intake", "contract", "backlog", "order cancellation"], |
| 912 | "CAPEX & Capacity": ["CAPEX", "new plant investment", "capacity expansion", "factory investment"], |
| 913 | "Guidance": ["management guidance", "guidance", "outlook", "guidance raised", "guidance cut"], |
| 914 | "Ownership": ["institutional ownership", "FII DII ownership", "shareholding changes"], |
| 915 | "Analyst": ["analyst target", "analyst ratings"], |
| 916 | "M&A": ["acquisition", "merger", "divestment"], |
| 917 | "Regulatory": ["regulatory announcement", "stock exchange announcement"], |
| 918 | } |
| 919 | patterns = category_patterns.get(category, [category]) |
| 920 | mapped_category = category in category_patterns |
| 921 | queries: list[str] = [] |
| 922 | indian_market = profile.country.upper() in {"IN", "IND", "INDIA"} or profile.exchange.upper() in { |
| 923 | "NSE", "BSE", "XNSE", "XBOM" |
| 924 | } |
| 925 | if indian_market: |
| 926 | nse_symbol = profile.provider_instrument_ids.get("NSE") or profile.ticker |
| 927 | bse_symbol = profile.provider_instrument_ids.get("BSE") or profile.provider_instrument_ids.get("NSE") or profile.ticker |
| 928 | india_patterns = { |
| 929 | "FINANCIAL_RESULTS": ["quarterly financial results", "corporate financial results"], |
| 930 | "Ownership": ["shareholding pattern", "promoter FII DII holding", "promoter pledge"], |
| 931 | "INSTITUTIONAL_ACTIVITY": ["shareholding pattern", "promoter FII DII holding"], |
| 932 | "ORDERS_BACKLOG": ["order win corporate announcement", "contract award"], |
| 933 | "CAPEX": ["capex new plant capacity expansion"], |
| 934 | "NEW_FACILITIES": ["new plant capacity expansion"], |
| 935 | "Regulatory": ["corporate announcement"], |
| 936 | }.get(category, patterns) |
| 937 | primary_pattern = india_patterns[0] |
| 938 | queries.extend([ |
| 939 | f'"{profile.company_name}" {primary_pattern} site:nseindia.com', |
| 940 | f'"{profile.company_name}" {primary_pattern} site:bseindia.com', |
| 941 | f'"{profile.company_name}" {primary_pattern}', |
| 942 | ]) |
| 943 | if profile.isin: |
| 944 | queries.append(f'"{profile.isin}"') |
| 945 | queries.extend([f'"{nse_symbol}" NSE', f'"{bse_symbol}" BSE']) |
| 946 | for pattern in india_patterns[1:]: |
| 947 | queries.extend([ |
| 948 | f'"{profile.company_name}" {pattern} site:nseindia.com', |
| 949 | f'"{profile.company_name}" {pattern} site:bseindia.com', |
| 950 | f'"{profile.company_name}" {pattern}', |
| 951 | ]) |
| 952 | if not mapped_category: |
| 953 | return [f"{identity} {category} {year}" for identity in identities] |
| 954 | queries.extend(f"{profile.company_name} {pattern}" for pattern in broad_patterns) |
| 955 | queries.extend(f"{profile.company_name} {pattern} {profile.ticker}" for pattern in broad_patterns if profile.ticker) |
| 956 | for identity in identities: |
| 957 | for pattern in patterns: |
| 958 | queries.append(f"{identity} {pattern} {year}") |
| 959 | if mapped_category: |
| 960 | queries.append(f"{identity} {pattern}") |
| 961 | if profile.company_name and profile.ticker and category in {"Analyst", "ANALYST_OPINION", "ANALYST_TARGETS"}: |
| 962 | ticker_patterns = ["analyst", "target price", "earnings", "investor relations"] |
| 963 | for pattern in ticker_patterns: |
| 964 | queries.append(f"{profile.company_name} {profile.ticker} {pattern}") |
| 965 | return queries |
| 966 | |
| 967 | |
| 968 | def generate_etf_search_queries(profile: EtfResearchProfile, category: str, date_window: SearchDateWindow) -> list[str]: |
| 969 | name = profile.fund_name |
| 970 | index = profile.underlying_index or _infer_underlying_index(name) |
| 971 | category_patterns = { |
| 972 | "ETF_PROFILE": ["factsheet", "holdings", "expense ratio", "AUM", "distribution policy", "NAV"], |
| 973 | "ETF_PERFORMANCE": ["performance", "tracking difference", "premium discount", "dividend yield"], |
| 974 | "INDEX_OUTLOOK": ["outlook", "valuation", "earnings outlook", "analyst outlook", "macro environment"], |
| 975 | "ETF_RISK": ["concentration risk", "sector concentration", "drawdown", "rate sensitivity", "liquidity"], |
| 976 | } |
| 977 | patterns = category_patterns.get(category, [category]) |
| 978 | queries = [f"{name} {pattern}" for pattern in patterns] |
| 979 | if profile.ticker: |
| 980 | queries.extend(f"{name} {profile.ticker} {pattern}" for pattern in patterns) |
| 981 | if index: |
| 982 | queries.extend( |
| 983 | [ |
| 984 | f"{index} outlook", |
| 985 | f"{index} valuation", |
| 986 | f"{index} earnings outlook", |
| 987 | f"{index} analyst outlook", |
| 988 | ] |
| 989 | ) |
| 990 | return _dedupe_identity_terms(queries) |
| 991 | |
| 992 | |
| 993 | def _bounded_search_queries(profile: CompanyResearchProfile, category: str, date_window: SearchDateWindow) -> list[str]: |
| 994 | queries = list(date_window.explicit_queries[:10]) if date_window.explicit_queries is not None else generate_search_queries(profile, category, date_window) |
| 995 | if date_window.query_limit is None: |
| 996 | return queries |
| 997 | return queries[: max(date_window.query_limit, 0)] |
| 998 | |
| 999 | |
| 1000 | def classify_source(host: str, profile: CompanyResearchProfile | EtfResearchProfile, candidate: CandidateSearchResult | None = None) -> SourceClassification: |
| 1001 | if any(host == domain.lower() or host.endswith(f".{domain.lower()}") for domain in profile.known_domains): |
| 1002 | return SourceClassification.OFFICIAL_COMPANY |
| 1003 | if candidate and _candidate_matches_company_domain(host, profile, candidate): |
| 1004 | return SourceClassification.OFFICIAL_COMPANY |
| 1005 | if host in { |
| 1006 | "sec.gov", |
| 1007 | "www.sec.gov", |
| 1008 | "sebi.gov.in", |
| 1009 | "www.sebi.gov.in", |
| 1010 | "afm.nl", |
| 1011 | "www.afm.nl", |
| 1012 | "bundesanzeiger.de", |
| 1013 | "www.bundesanzeiger.de", |
| 1014 | }: |
| 1015 | return SourceClassification.REGULATORY |
| 1016 | if ( |
| 1017 | host.endswith("deutsche-boerse.com") |
| 1018 | or host.endswith("deutsche-boerse-cash-market.com") |
| 1019 | or host.endswith("euronext.com") |
| 1020 | or host.endswith("nseindia.com") |
| 1021 | or host.endswith("bseindia.com") |
| 1022 | or host.endswith("lse.co.uk") |
| 1023 | or host.endswith("nasdaq.com") |
| 1024 | ): |
| 1025 | return SourceClassification.EXCHANGE |
| 1026 | if host.endswith("infineon.com") or host.endswith("onsemi.com") or host.endswith("stmicroelectronics.com"): |
| 1027 | return SourceClassification.CUSTOMER |
| 1028 | if ( |
| 1029 | host.endswith("eqs-news.com") |
| 1030 | or host.endswith("reuters.com") |
| 1031 | or host.endswith("bloomberg.com") |
| 1032 | or host.endswith("finance.yahoo.com") |
| 1033 | or host.endswith("ft.com") |
| 1034 | or host.endswith("wsj.com") |
| 1035 | or host.endswith("cnbc.com") |
| 1036 | or host.endswith("marketwatch.com") |
| 1037 | ): |
| 1038 | return SourceClassification.REPUTABLE_NEWS |
| 1039 | if ( |
| 1040 | host.endswith("marketscreener.com") |
| 1041 | or host.endswith("moneycontrol.com") |
| 1042 | or host.endswith("trendlyne.com") |
| 1043 | or host.endswith("screener.in") |
| 1044 | or host.endswith("morningstar.com") |
| 1045 | or host.endswith("investing.com") |
| 1046 | ): |
| 1047 | return SourceClassification.INVESTMENT_RESEARCH |
| 1048 | return SourceClassification.OTHER |
| 1049 | |
| 1050 | |
| 1051 | def _candidate_matches_company_domain(host: str, profile: CompanyResearchProfile | EtfResearchProfile, candidate: CandidateSearchResult) -> bool: |
| 1052 | domain_label = _registrable_label(host) |
| 1053 | display_name = _profile_display_name(profile) |
| 1054 | company_tokens = _company_domain_tokens(display_name) |
| 1055 | if isinstance(profile, EtfResearchProfile) and profile.fund_provider: |
| 1056 | company_tokens |= _company_domain_tokens(profile.fund_provider) |
| 1057 | if not domain_label or not company_tokens: |
| 1058 | return False |
| 1059 | if domain_label not in company_tokens and not any(token in domain_label or domain_label in token for token in company_tokens): |
| 1060 | return False |
| 1061 | evidence = f"{candidate.title} {candidate.snippet} {candidate.query}".lower() |
| 1062 | company_name = display_name.lower() |
| 1063 | if company_name in evidence: |
| 1064 | return True |
| 1065 | matched_tokens = [token for token in company_tokens if re.search(rf"(?<![a-z0-9]){re.escape(token)}(?![a-z0-9])", evidence)] |
| 1066 | official_terms = ["investor", "annual report", "quarterly", "results", "earnings", "guidance", "press", "news", "factsheet", "holdings", "fund"] |
| 1067 | return bool(matched_tokens) and any(term in evidence for term in official_terms) |
| 1068 | |
| 1069 | |
| 1070 | def _company_domain_tokens(company_name: str) -> set[str]: |
| 1071 | legal_suffixes = { |
| 1072 | "ag", |
| 1073 | "asa", |
| 1074 | "corp", |
| 1075 | "corporation", |
| 1076 | "gmbh", |
| 1077 | "group", |
| 1078 | "holding", |
| 1079 | "holdings", |
| 1080 | "inc", |
| 1081 | "limited", |
| 1082 | "ltd", |
| 1083 | "nv", |
| 1084 | "n.v", |
| 1085 | "plc", |
| 1086 | "sa", |
| 1087 | "se", |
| 1088 | "the", |
| 1089 | } |
| 1090 | tokens = [ |
| 1091 | token |
| 1092 | for token in re.findall(r"[a-z0-9]+", company_name.lower()) |
| 1093 | if len(token) >= 4 and token not in legal_suffixes |
| 1094 | ] |
| 1095 | compact = "".join(tokens) |
| 1096 | result = set(tokens) |
| 1097 | if len(compact) >= 4: |
| 1098 | result.add(compact) |
| 1099 | return result |
| 1100 | |
| 1101 | |
| 1102 | def _registrable_label(host: str) -> str: |
| 1103 | parts = [part for part in host.lower().split(".") if part and part != "www"] |
| 1104 | if len(parts) < 2: |
| 1105 | return parts[0] if parts else "" |
| 1106 | return parts[-2] |
| 1107 | |
| 1108 | |
| 1109 | def _source_rank(classification: SourceClassification) -> int: |
| 1110 | return { |
| 1111 | SourceClassification.OFFICIAL_COMPANY: 1, |
| 1112 | SourceClassification.REGULATORY: 2, |
| 1113 | SourceClassification.EXCHANGE: 3, |
| 1114 | SourceClassification.INVESTMENT_RESEARCH: 5, |
| 1115 | SourceClassification.REPUTABLE_NEWS: 6, |
| 1116 | SourceClassification.CUSTOMER: 8, |
| 1117 | SourceClassification.PARTNER: 8, |
| 1118 | SourceClassification.SUPPLIER: 8, |
| 1119 | SourceClassification.OTHER: 9, |
| 1120 | }[classification] |
| 1121 | |
| 1122 | |
| 1123 | def reliability_for_classification(classification: SourceClassification) -> ReliabilityLevel: |
| 1124 | return { |
| 1125 | SourceClassification.OFFICIAL_COMPANY: ReliabilityLevel.LEVEL_B, |
| 1126 | SourceClassification.REGULATORY: ReliabilityLevel.LEVEL_A, |
| 1127 | SourceClassification.EXCHANGE: ReliabilityLevel.LEVEL_A, |
| 1128 | SourceClassification.CUSTOMER: ReliabilityLevel.LEVEL_B, |
| 1129 | SourceClassification.PARTNER: ReliabilityLevel.LEVEL_B, |
| 1130 | SourceClassification.SUPPLIER: ReliabilityLevel.LEVEL_C, |
| 1131 | SourceClassification.REPUTABLE_NEWS: ReliabilityLevel.LEVEL_C, |
| 1132 | SourceClassification.INVESTMENT_RESEARCH: ReliabilityLevel.LEVEL_D, |
| 1133 | SourceClassification.OTHER: ReliabilityLevel.LEVEL_E, |
| 1134 | }[classification] |
| 1135 | |
| 1136 | |
| 1137 | def source_type_for_classification(classification: SourceClassification) -> SourceType: |
| 1138 | if classification == SourceClassification.OFFICIAL_COMPANY: |
| 1139 | return SourceType.INVESTOR_RELATIONS |
| 1140 | if classification == SourceClassification.REGULATORY: |
| 1141 | return SourceType.REGULATORY_FILING |
| 1142 | if classification == SourceClassification.EXCHANGE: |
| 1143 | return SourceType.EXCHANGE_ANNOUNCEMENT |
| 1144 | if classification == SourceClassification.REPUTABLE_NEWS: |
| 1145 | return SourceType.NEWS |
| 1146 | return SourceType.SEARCH_DISCOVERY |
| 1147 | |
| 1148 | |
| 1149 | def publisher_from_host(host: str) -> str: |
| 1150 | parts = host.split(".") |
| 1151 | if len(parts) >= 2: |
| 1152 | return parts[-2].replace("-", " ").title() |
| 1153 | return host |
| 1154 | |
| 1155 | |
| 1156 | def _discovery_rejection_reason(reason: str) -> str: |
| 1157 | if reason in {"DOMAIN_VALIDATION_FAILED", "SOURCE_QUALITY_REJECTED", "DUPLICATE"}: |
| 1158 | return reason |
| 1159 | if reason in {"UNSUPPORTED_URL", "DOMAIN_BLOCKED"}: |
| 1160 | return "DOMAIN_VALIDATION_FAILED" |
| 1161 | if "not permitted" in reason or "private" in reason.lower() or "local" in reason.lower(): |
| 1162 | return "DOMAIN_VALIDATION_FAILED" |
| 1163 | return reason |
| 1164 | |
| 1165 | |
| 1166 | def _profile_display_name(profile: CompanyResearchProfile | EtfResearchProfile) -> str: |
| 1167 | return profile.company_name if isinstance(profile, CompanyResearchProfile) else profile.fund_name |
| 1168 | |
| 1169 | |
| 1170 | def _infer_underlying_index(name: str) -> str | None: |
| 1171 | upper = name.upper() |
| 1172 | if "S&P 500" in upper or "SP 500" in upper: |
| 1173 | return "S&P 500" |
| 1174 | if "NASDAQ 100" in upper or "NASDAQ-100" in upper: |
| 1175 | return "NASDAQ 100" |
| 1176 | return None |
| 1177 | |
| 1178 | |
| 1179 | def _dedupe_identity_terms(values: list[str]) -> list[str]: |
| 1180 | seen: set[str] = set() |
| 1181 | result: list[str] = [] |
| 1182 | for value in values: |
| 1183 | normalized = value.strip() |
| 1184 | key = normalized.lower() |
| 1185 | if normalized and key not in seen: |
| 1186 | seen.add(key) |
| 1187 | result.append(normalized) |
| 1188 | return result |
| 1189 | |
| 1190 | |
| 1191 | async def _safe_search_get( |
| 1192 | client: httpx.AsyncClient, |
| 1193 | endpoint: str, |
| 1194 | *, |
| 1195 | params: dict[str, str | int], |
| 1196 | headers: dict[str, str] | None = None, |
| 1197 | ) -> httpx.Response: |
| 1198 | try: |
| 1199 | response = await client.get(endpoint, params=params, headers=headers) |
| 1200 | except httpx.TimeoutException as exc: |
| 1201 | raise SearchProviderError("SEARCH_PROVIDER_TIMEOUT") from exc |
| 1202 | except httpx.HTTPError as exc: |
| 1203 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:http_error") from exc |
| 1204 | if response.status_code in {401, 403}: |
| 1205 | raise SearchProviderError("SEARCH_PROVIDER_FORBIDDEN") |
| 1206 | if response.status_code == 429: |
| 1207 | raise SearchProviderError("SEARCH_PROVIDER_RATE_LIMITED") |
| 1208 | if response.status_code >= 500: |
| 1209 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE") |
| 1210 | if response.status_code >= 400: |
| 1211 | raise SearchProviderError(f"SEARCH_PROVIDER_UNAVAILABLE:http_status_{response.status_code}") |
| 1212 | return response |
| 1213 | |
| 1214 | |
| 1215 | def _safe_json(response: httpx.Response) -> dict: |
| 1216 | try: |
| 1217 | payload = response.json() |
| 1218 | except ValueError as exc: |
| 1219 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") from exc |
| 1220 | if not isinstance(payload, dict): |
| 1221 | raise SearchProviderError("SEARCH_PROVIDER_UNAVAILABLE:invalid_response") |
| 1222 | return payload |