| 1 | from __future__ import annotations |
| 2 | |
| 3 | from dataclasses import FrozenInstanceError |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | from decimal import Decimal |
| 6 | from uuid import UUID, uuid4 |
| 7 | |
| 8 | import pytest |
| 9 | |
| 10 | from app.historical_market_data import YahooHistoricalPriceProvider |
| 11 | from app.international_fundamentals import EodhdFundamentalProvider, SecEdgarFundamentalProvider |
| 12 | from app.portfolio_context import PortfolioContext, PortfolioHoldingContext |
| 13 | from app.research_readiness import ( |
| 14 | DurableResearchSnapshot, |
| 15 | FreshnessMode, |
| 16 | FreshnessPolicyRegistry, |
| 17 | ProviderAuthorityRegistry, |
| 18 | ResearchCoverageService, |
| 19 | ResearchEvidence, |
| 20 | ResearchReadinessService, |
| 21 | ResearchRefreshPlanner, |
| 22 | ResearchRequirementRegistry, |
| 23 | ResearchRequirementStatus, |
| 24 | ResearchSourceTier, |
| 25 | ResearchSupportedAction, |
| 26 | RuleEngineArea, |
| 27 | ) |
| 28 | from app.structured_market import YahooFinanceProvider |
| 29 | |
| 30 | |
| 31 | NOW = datetime(2026, 9, 10, 12, 0, tzinfo=timezone.utc) |
| 32 | GLOBAL_INSTRUMENT_ID = UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa") |
| 33 | |
| 34 | |
| 35 | class RecordingDataSource: |
| 36 | def __init__(self, snapshot: DurableResearchSnapshot) -> None: |
| 37 | self.snapshot = snapshot |
| 38 | self.loads: list[tuple[UUID, tuple[str, ...]]] = [] |
| 39 | self.provider_calls = 0 |
| 40 | self.holding_mutations = 0 |
| 41 | |
| 42 | def load_by_global_instrument_id(self, global_instrument_id, requirements): |
| 43 | self.loads.append( |
| 44 | (global_instrument_id, tuple(requirement.requirement_id for requirement in requirements)) |
| 45 | ) |
| 46 | return self.snapshot |
| 47 | |
| 48 | |
| 49 | def evidence( |
| 50 | requirement_id: str, |
| 51 | *, |
| 52 | age: timedelta = timedelta(hours=1), |
| 53 | evidence_id: str | None = None, |
| 54 | source: str = "NSE", |
| 55 | source_tier: ResearchSourceTier = ResearchSourceTier.OFFICIAL, |
| 56 | complete: bool = True, |
| 57 | fact_key: str | None = None, |
| 58 | value: str | None = None, |
| 59 | unresolved: bool = False, |
| 60 | ) -> ResearchEvidence: |
| 61 | as_of = NOW - age |
| 62 | return ResearchEvidence( |
| 63 | evidence_id=evidence_id or f"{requirement_id}-{uuid4()}", |
| 64 | requirement_id=requirement_id, |
| 65 | source=source, |
| 66 | source_tier=source_tier, |
| 67 | retrieved_at=as_of, |
| 68 | as_of=as_of, |
| 69 | published_at=as_of, |
| 70 | event_date=as_of, |
| 71 | complete=complete, |
| 72 | fact_key=fact_key, |
| 73 | value_fingerprint=value, |
| 74 | unresolved=unresolved, |
| 75 | ) |
| 76 | |
| 77 | |
| 78 | def complete_snapshot( |
| 79 | *, |
| 80 | overrides: dict[str, tuple[ResearchEvidence, ...]] | None = None, |
| 81 | omit: set[str] | None = None, |
| 82 | supported: frozenset[str] | None = None, |
| 83 | refreshing: frozenset[str] = frozenset(), |
| 84 | failures: dict[str, str] | None = None, |
| 85 | ) -> DurableResearchSnapshot: |
| 86 | registry = ResearchRequirementRegistry.default() |
| 87 | values = { |
| 88 | requirement.requirement_id: ( |
| 89 | evidence( |
| 90 | requirement.requirement_id, |
| 91 | age=timedelta(minutes=5) |
| 92 | if requirement.requirement_id == "LATEST_PRICE" |
| 93 | else timedelta(hours=1), |
| 94 | ), |
| 95 | ) |
| 96 | for requirement in registry.requirements |
| 97 | } |
| 98 | for requirement_id in omit or set(): |
| 99 | values.pop(requirement_id, None) |
| 100 | values.update(overrides or {}) |
| 101 | return DurableResearchSnapshot( |
| 102 | GLOBAL_INSTRUMENT_ID, |
| 103 | values, |
| 104 | supported_requirement_ids=supported, |
| 105 | refreshing_requirement_ids=refreshing, |
| 106 | failure_reasons=failures or {}, |
| 107 | ) |
| 108 | |
| 109 | |
| 110 | def assess(snapshot: DurableResearchSnapshot): |
| 111 | data_source = RecordingDataSource(snapshot) |
| 112 | readiness = ResearchReadinessService(data_source).assess( |
| 113 | GLOBAL_INSTRUMENT_ID, jurisdiction="INDIA", now=NOW |
| 114 | ) |
| 115 | return data_source, readiness, ResearchRefreshPlanner().plan(readiness, jurisdiction="INDIA") |
| 116 | |
| 117 | |
| 118 | def target_ids(plan) -> set[str]: |
| 119 | return {target.requirement_id for target in plan.targets} |
| 120 | |
| 121 | |
| 122 | def test_fresh_required_data_produces_no_provider_candidate() -> None: |
| 123 | data_source, readiness, plan = assess(complete_snapshot()) |
| 124 | |
| 125 | for target in plan.targets: |
| 126 | data_source.provider_calls += 1 |
| 127 | |
| 128 | assert readiness.mandatory_ready is True |
| 129 | assert plan.targets == () |
| 130 | assert data_source.provider_calls == 0 |
| 131 | assert len(data_source.loads) == 1 |
| 132 | |
| 133 | |
| 134 | def test_stale_required_data_is_the_only_targeted_refresh() -> None: |
| 135 | stale = evidence("LATEST_PRICE", age=timedelta(minutes=16)) |
| 136 | _, readiness, plan = assess(complete_snapshot(overrides={"LATEST_PRICE": (stale,)})) |
| 137 | |
| 138 | assert readiness.for_requirement("LATEST_PRICE").status == ResearchRequirementStatus.READY_STALE |
| 139 | assert target_ids(plan) == {"LATEST_PRICE"} |
| 140 | |
| 141 | |
| 142 | def test_missing_required_data_becomes_a_targeted_fetch_candidate() -> None: |
| 143 | _, readiness, plan = assess(complete_snapshot(omit={"QUARTERLY_FINANCIALS"})) |
| 144 | |
| 145 | result = readiness.for_requirement("QUARTERLY_FINANCIALS") |
| 146 | assert result.status == ResearchRequirementStatus.MISSING |
| 147 | assert result.missing_reason == "NO_DURABLE_EVIDENCE" |
| 148 | assert target_ids(plan) == {"QUARTERLY_FINANCIALS"} |
| 149 | |
| 150 | |
| 151 | def test_incomplete_required_data_is_partial_and_targeted() -> None: |
| 152 | incomplete = evidence("QUARTERLY_FINANCIALS", complete=False) |
| 153 | _, readiness, plan = assess( |
| 154 | complete_snapshot(overrides={"QUARTERLY_FINANCIALS": (incomplete,)}) |
| 155 | ) |
| 156 | |
| 157 | result = readiness.for_requirement("QUARTERLY_FINANCIALS") |
| 158 | assert result.status == ResearchRequirementStatus.PARTIAL |
| 159 | assert result.missing_reason == "INSUFFICIENT_COMPLETE_EVIDENCE" |
| 160 | assert target_ids(plan) == {"QUARTERLY_FINANCIALS"} |
| 161 | |
| 162 | |
| 163 | def test_unrelated_fresh_category_is_not_refreshed() -> None: |
| 164 | _, readiness, plan = assess(complete_snapshot(omit={"SECTOR_MACRO"})) |
| 165 | |
| 166 | assert readiness.for_requirement("HISTORICAL_PRICE_SERIES").status == ResearchRequirementStatus.READY_FRESH |
| 167 | assert target_ids(plan) == {"SECTOR_MACRO"} |
| 168 | |
| 169 | |
| 170 | def test_equal_authority_disagreement_is_conflicting() -> None: |
| 171 | conflicting = ( |
| 172 | evidence( |
| 173 | "VALUATION_INPUTS", |
| 174 | evidence_id="valuation-a", |
| 175 | fact_key="trailing-pe", |
| 176 | value="18.2", |
| 177 | ), |
| 178 | evidence( |
| 179 | "VALUATION_INPUTS", |
| 180 | evidence_id="valuation-b", |
| 181 | fact_key="trailing-pe", |
| 182 | value="21.7", |
| 183 | ), |
| 184 | ) |
| 185 | _, readiness, plan = assess(complete_snapshot(overrides={"VALUATION_INPUTS": conflicting})) |
| 186 | |
| 187 | result = readiness.for_requirement("VALUATION_INPUTS") |
| 188 | assert result.status == ResearchRequirementStatus.CONFLICTING |
| 189 | assert result.conflict_reason == "EQUAL_AUTHORITY_CONFLICT:trailing-pe:valuation-a,valuation-b" |
| 190 | assert target_ids(plan) == {"VALUATION_INPUTS"} |
| 191 | |
| 192 | |
| 193 | def test_unsupported_category_is_not_scheduled() -> None: |
| 194 | requirement_ids = { |
| 195 | item.requirement_id for item in ResearchRequirementRegistry.default().requirements |
| 196 | } |
| 197 | supported = frozenset(requirement_ids - {"SECTOR_MACRO"}) |
| 198 | _, readiness, plan = assess(complete_snapshot(supported=supported)) |
| 199 | |
| 200 | result = readiness.for_requirement("SECTOR_MACRO") |
| 201 | assert result.status == ResearchRequirementStatus.UNSUPPORTED |
| 202 | assert ResearchSupportedAction.UPLOAD_EVIDENCE in result.supported_actions |
| 203 | assert ResearchSupportedAction.FIND_DATA not in result.supported_actions |
| 204 | assert "SECTOR_MACRO" not in target_ids(plan) |
| 205 | |
| 206 | |
| 207 | def test_mandatory_and_optional_distinction_is_retained_by_planner() -> None: |
| 208 | _, readiness, plan = assess(complete_snapshot(omit={"LATEST_PRICE", "SHAREHOLDING"})) |
| 209 | |
| 210 | assert readiness.for_requirement("LATEST_PRICE").mandatory is True |
| 211 | assert readiness.for_requirement("SHAREHOLDING").mandatory is False |
| 212 | assert readiness.for_requirement("SHAREHOLDING").status == ResearchRequirementStatus.MISSING |
| 213 | assert target_ids(plan) == {"LATEST_PRICE"} |
| 214 | |
| 215 | |
| 216 | def test_current_news_older_than_thirty_days_is_excluded_but_not_deleted() -> None: |
| 217 | old_news = evidence("CURRENT_NEWS", age=timedelta(days=31), evidence_id="old-news") |
| 218 | boundary_news = evidence( |
| 219 | "CURRENT_NEWS", age=timedelta(days=30), evidence_id="boundary-news" |
| 220 | ) |
| 221 | snapshot = complete_snapshot(overrides={"CURRENT_NEWS": (old_news,)}) |
| 222 | _, readiness, plan = assess(snapshot) |
| 223 | registry = ResearchRequirementRegistry.default() |
| 224 | requirement = registry.get("CURRENT_NEWS") |
| 225 | policy = FreshnessPolicyRegistry.default().get("CURRENT_NEWS") |
| 226 | |
| 227 | result = readiness.for_requirement("CURRENT_NEWS") |
| 228 | assert result.status == ResearchRequirementStatus.MISSING |
| 229 | assert result.missing_reason == "NO_EVIDENCE_IN_CURRENT_NEWS_WINDOW" |
| 230 | assert result.evidence_ids == () |
| 231 | assert ResearchCoverageService().score_input_evidence(requirement, (old_news,), policy, NOW) == () |
| 232 | assert ResearchCoverageService().score_input_evidence( |
| 233 | requirement, (boundary_news,), policy, NOW |
| 234 | ) == (boundary_news,) |
| 235 | assert snapshot.evidence_for("CURRENT_NEWS") == (old_news,) |
| 236 | assert target_ids(plan) == set() # News is optional; explicit news refresh remains supported. |
| 237 | |
| 238 | |
| 239 | def test_old_unresolved_governance_evidence_remains_queryable_and_ready() -> None: |
| 240 | old_issue = evidence( |
| 241 | "GOVERNANCE_HISTORY", |
| 242 | age=timedelta(days=4 * 365), |
| 243 | evidence_id="unresolved-litigation", |
| 244 | unresolved=True, |
| 245 | ) |
| 246 | snapshot = complete_snapshot(overrides={"GOVERNANCE_HISTORY": (old_issue,)}) |
| 247 | data_source, readiness, plan = assess(snapshot) |
| 248 | |
| 249 | assert readiness.for_requirement("GOVERNANCE_HISTORY").status == ResearchRequirementStatus.READY_FRESH |
| 250 | assert ResearchCoverageService().governance_history(data_source.snapshot) == (old_issue,) |
| 251 | assert "GOVERNANCE_HISTORY" not in target_ids(plan) |
| 252 | |
| 253 | |
| 254 | def test_canonical_global_instrument_id_is_required_before_any_db_read() -> None: |
| 255 | data_source = RecordingDataSource(complete_snapshot()) |
| 256 | service = ResearchReadinessService(data_source) |
| 257 | |
| 258 | with pytest.raises(ValueError, match="canonical globalInstrumentId is required"): |
| 259 | service.assess(None, now=NOW) # type: ignore[arg-type] |
| 260 | with pytest.raises(ValueError, match="canonical globalInstrumentId is required"): |
| 261 | service.assess(UUID(int=0), now=NOW) |
| 262 | assert data_source.loads == [] |
| 263 | |
| 264 | |
| 265 | def test_readiness_and_portfolio_projection_do_not_mutate_holdings() -> None: |
| 266 | holding = PortfolioHoldingContext( |
| 267 | global_instrument_id=GLOBAL_INSTRUMENT_ID, |
| 268 | quantity=Decimal("15"), |
| 269 | average_cost=Decimal("101.50"), |
| 270 | market_value=Decimal("1800"), |
| 271 | unrealized_pnl=Decimal("277.50"), |
| 272 | ) |
| 273 | context = PortfolioContext( |
| 274 | context_id="context-1", |
| 275 | owner_session_id="session-1", |
| 276 | portfolio_id=uuid4(), |
| 277 | holdings=(holding,), |
| 278 | expires_at=NOW + timedelta(hours=1), |
| 279 | ) |
| 280 | data_source, _, _ = assess(complete_snapshot()) |
| 281 | |
| 282 | assert context.global_instrument_ids == (GLOBAL_INSTRUMENT_ID,) |
| 283 | assert context.holdings == (holding,) |
| 284 | assert data_source.holding_mutations == 0 |
| 285 | with pytest.raises(FrozenInstanceError): |
| 286 | context.holdings[0].quantity = Decimal("0") # type: ignore[misc] |
| 287 | |
| 288 | |
| 289 | def test_market_data_requirements_keep_distinct_freshness_semantics() -> None: |
| 290 | policies = FreshnessPolicyRegistry.default() |
| 291 | latest = policies.get("LATEST_PRICE") |
| 292 | historical = policies.get("HISTORICAL_PRICE_SERIES") |
| 293 | _, readiness, plan = assess(complete_snapshot()) |
| 294 | |
| 295 | assert latest.mode == FreshnessMode.MARKET_SESSION_AWARE |
| 296 | assert historical.mode == FreshnessMode.DAILY_INCREMENTAL |
| 297 | assert latest.maximum_age != historical.maximum_age |
| 298 | assert readiness.for_requirement("LATEST_PRICE").status == ResearchRequirementStatus.READY_FRESH |
| 299 | assert readiness.for_requirement("HISTORICAL_PRICE_SERIES").status == ResearchRequirementStatus.READY_FRESH |
| 300 | assert not {"LATEST_PRICE", "HISTORICAL_PRICE_SERIES"} & target_ids(plan) |
| 301 | |
| 302 | |
| 303 | def test_existing_provider_adapter_contracts_remain_available() -> None: |
| 304 | assert YahooFinanceProvider.provider_name == "YAHOO_FINANCE" |
| 305 | assert YahooHistoricalPriceProvider.provider_name == "YAHOO_FINANCE" |
| 306 | assert SecEdgarFundamentalProvider.provider_name == "SEC_EDGAR" |
| 307 | assert EodhdFundamentalProvider.provider_name == "EODHD" |
| 308 | |
| 309 | |
| 310 | def test_requirement_contract_exposes_all_readiness_fields_and_area_weights() -> None: |
| 311 | _, readiness, _ = assess(complete_snapshot()) |
| 312 | result = readiness.for_requirement("LATEST_PRICE") |
| 313 | registry = ResearchRequirementRegistry.default() |
| 314 | |
| 315 | assert result.requirement_id == "LATEST_PRICE" |
| 316 | assert result.rule_engine_area == RuleEngineArea.PRICE_TECHNICAL |
| 317 | assert result.source == "NSE" |
| 318 | assert result.source_tier == ResearchSourceTier.OFFICIAL |
| 319 | assert result.as_of == NOW - timedelta(minutes=5) |
| 320 | assert result.retrieved_at == NOW - timedelta(minutes=5) |
| 321 | assert result.age == timedelta(minutes=5) |
| 322 | assert result.freshness_policy.policy_id == "LATEST_PRICE" |
| 323 | assert len(result.evidence_ids) == 1 |
| 324 | assert result.missing_reason is None |
| 325 | assert result.conflict_reason is None |
| 326 | assert result.supported_actions == () |
| 327 | assert set(registry.area_weights) == set(RuleEngineArea) |
| 328 | assert sum(registry.area_weights.values(), Decimal("0")) == Decimal("1.00") |
| 329 | |
| 330 | |
| 331 | def test_uploaded_evidence_does_not_outrank_official_evidence() -> None: |
| 332 | official = evidence( |
| 333 | "QUARTERLY_FINANCIALS", |
| 334 | evidence_id="official", |
| 335 | source="NSE", |
| 336 | source_tier=ResearchSourceTier.OFFICIAL, |
| 337 | fact_key="revenue-2026-q2", |
| 338 | value="100", |
| 339 | ) |
| 340 | upload = evidence( |
| 341 | "QUARTERLY_FINANCIALS", |
| 342 | evidence_id="upload", |
| 343 | source="USER_UPLOAD", |
| 344 | source_tier=ResearchSourceTier.USER_UPLOAD, |
| 345 | fact_key="revenue-2026-q2", |
| 346 | value="999", |
| 347 | ) |
| 348 | _, readiness, _ = assess( |
| 349 | complete_snapshot(overrides={"QUARTERLY_FINANCIALS": (upload, official)}) |
| 350 | ) |
| 351 | |
| 352 | result = readiness.for_requirement("QUARTERLY_FINANCIALS") |
| 353 | assert result.status == ResearchRequirementStatus.READY_FRESH |
| 354 | assert result.source == "NSE" |
| 355 | assert result.conflict_reason is None |
| 356 | |
| 357 | |
| 358 | def test_external_tool_fallback_is_policy_controlled() -> None: |
| 359 | policy = ProviderAuthorityRegistry.default().policy_for("QUARTERLY_FINANCIALS", "USA") |
| 360 | |
| 361 | assert policy.fallback_policy.permits(ResearchRequirementStatus.MISSING) |
| 362 | assert policy.fallback_policy.permits(ResearchRequirementStatus.CONFLICTING) |
| 363 | assert policy.fallback_policy.permits(ResearchRequirementStatus.READY_STALE, confidence=0.4) |
| 364 | assert not policy.fallback_policy.permits(ResearchRequirementStatus.READY_STALE, confidence=0.95) |
| 365 | |
| 366 | |
| 367 | def test_provider_authority_is_fact_specific_and_unregistered_facts_fail_closed() -> None: |
| 368 | registry = ProviderAuthorityRegistry.default() |
| 369 | order_book = registry.policy_for("ORDER_BOOK_CAPEX_GUIDANCE", "GLOBAL") |
| 370 | sector_macro = registry.policy_for("SECTOR_MACRO", "GLOBAL") |
| 371 | |
| 372 | assert { |
| 373 | registry.policy_for(requirement.requirement_id, "GLOBAL").requirement_id |
| 374 | for requirement in ResearchRequirementRegistry.default().requirements |
| 375 | } == { |
| 376 | requirement.requirement_id |
| 377 | for requirement in ResearchRequirementRegistry.default().requirements |
| 378 | } |
| 379 | assert tuple( |
| 380 | item.source |
| 381 | for item in registry.policy_for("QUARTERLY_FINANCIALS", "INDIA").authorities[:2] |
| 382 | ) == ("NSE", "COMPANY_FILING") |
| 383 | assert registry.policy_for("QUARTERLY_FINANCIALS", "USA").authorities[0].source == "SEC_EDGAR" |
| 384 | assert tuple( |
| 385 | item.source |
| 386 | for item in registry.policy_for("QUARTERLY_FINANCIALS", "EUROPE").authorities[:3] |
| 387 | ) == ("REGULATORY_FILING", "COMPANY_FILING", "EODHD") |
| 388 | assert registry.policy_for("LATEST_PRICE", "INDIA").authorities[0].source == ( |
| 389 | "CONFIGURED_MARKET_DATA" |
| 390 | ) |
| 391 | assert registry.policy_for("CANONICAL_IDENTITY", "GLOBAL").authorities[0].source == ( |
| 392 | "ISIN_OR_PERMANENT_ID" |
| 393 | ) |
| 394 | assert order_book.authorities != sector_macro.authorities |
| 395 | with pytest.raises(KeyError, match="No fact-specific provider authority policy"): |
| 396 | registry.policy_for("UNREGISTERED_FACT", "GLOBAL") |
| 397 | |
| 398 | |
| 399 | def test_refreshing_and_failed_states_are_representable() -> None: |
| 400 | refreshing_snapshot = complete_snapshot( |
| 401 | omit={"SECTOR_MACRO"}, refreshing=frozenset({"SECTOR_MACRO"}) |
| 402 | ) |
| 403 | _, refreshing, refreshing_plan = assess(refreshing_snapshot) |
| 404 | assert refreshing.for_requirement("SECTOR_MACRO").status == ResearchRequirementStatus.REFRESHING |
| 405 | assert "SECTOR_MACRO" not in target_ids(refreshing_plan) |
| 406 | |
| 407 | failed_snapshot = complete_snapshot( |
| 408 | omit={"SECTOR_MACRO"}, failures={"SECTOR_MACRO": "PROVIDER_TIMEOUT"} |
| 409 | ) |
| 410 | _, failed, failed_plan = assess(failed_snapshot) |
| 411 | assert failed.for_requirement("SECTOR_MACRO").status == ResearchRequirementStatus.FAILED |
| 412 | assert failed.for_requirement("SECTOR_MACRO").missing_reason == "PROVIDER_TIMEOUT" |
| 413 | assert target_ids(failed_plan) == {"SECTOR_MACRO"} |