| 1 | from dataclasses import replace |
| 2 | from datetime import datetime, timedelta, timezone |
| 3 | from decimal import Decimal |
| 4 | import pytest |
| 5 | |
| 6 | from app.research_readiness import ResearchReadinessService, ResearchRequirementStatus, FreshnessPolicyRegistry |
| 7 | from app.research_readiness_runtime import RepositoryResearchReadinessAdapter |
| 8 | from app.valuation_evidence import materialize_valuation |
| 9 | from test_research_readiness_runtime import DurableRepositoryFixture, _profile, _fact |
| 10 | from test_stock_rule_engine import _structured, _prices |
| 11 | from test_news_intelligence_v2 import NOW, run, outcome |
| 12 | |
| 13 | |
| 14 | @pytest.mark.parametrize('period,status', [('2026-03-31','READY_STALE'),('2026-06-30','READY_FRESH')]) |
| 15 | def test_tmcv_mandatory_balance_dates_not_newer_supporting_income(period,status): |
| 16 | profile=_profile(); repo=DurableRepositoryFixture(profile,complete=False) |
| 17 | repo.facts=[_fact(profile,'total_debt','100',period,'QUARTERLY'), |
| 18 | _fact(profile,'total_equity','200',period,'QUARTERLY'), |
| 19 | _fact(profile,'finance_cost','10','2026-06-30','QUARTERLY')] |
| 20 | adapter=RepositoryResearchReadinessAdapter(repo) |
| 21 | result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=NOW) |
| 22 | balance=result.for_requirement('BALANCE_SHEET_FACTS') |
| 23 | assert balance.status==status |
| 24 | assert balance.as_of.date().isoformat()==period |
| 25 | policy=FreshnessPolicyRegistry.default().get('QUARTERLY_FINANCIALS') |
| 26 | assert policy.maximum_age==timedelta(days=120) |
| 27 | if status=='READY_STALE': |
| 28 | assert balance.missing_reason=='FRESHNESS_POLICY_EXPIRED:DEBT,EQUITY' |
| 29 | else: |
| 30 | assert balance.age==NOW-datetime(2026,6,30,tzinfo=timezone.utc) |
| 31 | |
| 32 | |
| 33 | def test_release_aware_anchor_explicit_validity_not_retrieval(): |
| 34 | from test_research_readiness import evidence |
| 35 | policy=FreshnessPolicyRegistry.default().get('QUARTERLY_FINANCIALS') |
| 36 | e=replace(evidence('BALANCE_SHEET_FACTS'),as_of=datetime(2026,6,30,tzinfo=timezone.utc), |
| 37 | retrieved_at=NOW,published_at=datetime(2026,8,10,tzinfo=timezone.utc)) |
| 38 | assert policy.evidence_time(e)==e.as_of |
| 39 | assert policy.is_fresh(e,NOW) |
| 40 | assert not policy.is_fresh(e,NOW+timedelta(days=60)) |
| 41 | assert policy.is_fresh(replace(e,valid_until=NOW+timedelta(days=61)),NOW+timedelta(days=60)) |
| 42 | |
| 43 | |
| 44 | def valuation_fixture(): |
| 45 | record=_structured(trailingEps=Decimal('10'),bookValue=Decimal('25')) |
| 46 | for key in ('trailingEps','bookValue'): |
| 47 | record.snapshot.facts[key]=record.snapshot.facts[key].model_copy(update={ |
| 48 | 'as_of_date':NOW-timedelta(days=75),'retrieved_at':NOW-timedelta(days=7)}) |
| 49 | price=_prices(1)[0].model_copy(update={'price':Decimal('200'),'observed_at':NOW,'retrieved_at':NOW}) |
| 50 | return record,price |
| 51 | |
| 52 | |
| 53 | def test_price_recomputes_ratios_using_non_daily_basis(): |
| 54 | record,price=valuation_fixture() |
| 55 | values=materialize_valuation([record],[price],now=NOW) |
| 56 | assert values['trailingPE'].value==Decimal('20') |
| 57 | assert values['priceToBook'].value==Decimal('8') |
| 58 | next_price=price.model_copy(update={'price':Decimal('210'),'observed_at':NOW+timedelta(days=1),'retrieved_at':NOW+timedelta(days=1)}) |
| 59 | assert materialize_valuation([record],[next_price],now=NOW+timedelta(days=1))['trailingPE'].value==21 |
| 60 | assert values['trailingPE'].as_of_date==NOW |
| 61 | assert 'basisAsOf=' in values['trailingPE'].calculation_basis |
| 62 | |
| 63 | |
| 64 | @pytest.mark.parametrize('problem',['currency','future','expired_basis','conflicting_price']) |
| 65 | def test_valuation_rejects_incoherent_evidence(problem): |
| 66 | record,price=valuation_fixture(); prices=[price] |
| 67 | if problem=='currency': record.currency='USD' |
| 68 | if problem=='future': price.observed_at=NOW+timedelta(days=1) |
| 69 | if problem=='expired_basis': |
| 70 | for k in ('trailingEps','bookValue'): record.snapshot.facts[k].as_of_date=NOW-timedelta(days=121) |
| 71 | if problem=='conflicting_price': prices.append(price.model_copy(update={'provider':'NSE','price':Decimal('300')})) |
| 72 | assert materialize_valuation([record],prices,now=NOW)=={} |
| 73 | |
| 74 | |
| 75 | def test_stale_price_does_not_get_current_timestamp(): |
| 76 | record,price=valuation_fixture() |
| 77 | price.observed_at=NOW-timedelta(days=8) |
| 78 | values=materialize_valuation([record],[price],now=NOW) |
| 79 | assert values['trailingPE'].as_of_date==price.observed_at |
| 80 | |
| 81 | |
| 82 | @pytest.mark.parametrize('days_after,expected',[(0,'READY_FRESH'),(1,'READY_FRESH'),(5,'READY_STALE')]) |
| 83 | def test_valuation_readiness_overnight_and_stale_price(days_after,expected): |
| 84 | record,price=valuation_fixture() |
| 85 | profile=_profile().model_copy(update={'instrument_id':record.instrument_id}) |
| 86 | repo=DurableRepositoryFixture(profile,complete=False) |
| 87 | # Keep only the actual reusable basis; a provider ratio need not refresh daily. |
| 88 | record.snapshot.facts={k:v for k,v in record.snapshot.facts.items() if k in {'trailingEps','bookValue'}} |
| 89 | evaluation=NOW+timedelta(days=days_after) |
| 90 | if days_after<=1: |
| 91 | price.observed_at=evaluation; price.retrieved_at=evaluation |
| 92 | repo.structured=[record]; repo.observations=[price] |
| 93 | adapter=RepositoryResearchReadinessAdapter(repo); adapter._evaluation_times[profile.instrument_id]=evaluation |
| 94 | result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=evaluation) |
| 95 | assert result.for_requirement('VALUATION_INPUTS').status==expected |
| 96 | |
| 97 | |
| 98 | @pytest.mark.parametrize('provider_state,expected,coverage',[('SUCCESS_EMPTY','READY_FRESH',100),('FAILED','FAILED',0),('PARTIAL','PARTIAL',0)]) |
| 99 | def test_persisted_search_coverage_readiness_without_provider(provider_state,expected,coverage): |
| 100 | profile=_profile(); repo=DurableRepositoryFixture(profile,complete=False) |
| 101 | search=run([outcome(state=provider_state)]).model_copy(update={'instrument_id':profile.instrument_id}) |
| 102 | repo.news_records_for=lambda *a,**k:[search] |
| 103 | adapter=RepositoryResearchReadinessAdapter(repo); adapter._evaluation_times[profile.instrument_id]=NOW |
| 104 | result=ResearchReadinessService(adapter).assess(profile.instrument_id,jurisdiction='INDIA',now=NOW) |
| 105 | news=result.for_requirement('CURRENT_NEWS') |
| 106 | assert news.status==expected |
| 107 | assert news.coverage_pct==coverage |
| 108 | assert not news.mandatory |
| 109 | assert repo.provider_calls==0 |