| 1 | from __future__ import annotations |
| 2 | |
| 3 | from dataclasses import replace |
| 4 | from datetime import datetime, timedelta, timezone |
| 5 | from decimal import Decimal |
| 6 | from uuid import UUID, uuid4 |
| 7 | |
| 8 | import pytest |
| 9 | from fastapi.testclient import TestClient |
| 10 | |
| 11 | from app import main |
| 12 | |
| 13 | from app.fact_precedence import FinancialFact, FinancialFactKey, FactSourceTier |
| 14 | from app.models import ( |
| 15 | CompanyResearchProfile, |
| 16 | EventImpact, |
| 17 | MarketPriceObservation, |
| 18 | ProvenancedValue, |
| 19 | ReliabilityLevel, |
| 20 | ResearchEvent, |
| 21 | ResearchEventType, |
| 22 | ResearchLifecycleStatus, |
| 23 | ShareholdingCategory, |
| 24 | ShareholdingSnapshot, |
| 25 | ShareholdingSnapshotValue, |
| 26 | SourceClassification, |
| 27 | SourceMode, |
| 28 | SourceType, |
| 29 | StructuredInstrumentResolution, |
| 30 | StructuredMarketSnapshot, |
| 31 | StructuredMarketSnapshotRecord, |
| 32 | TimeHorizon, |
| 33 | ) |
| 34 | from app.persistence import SqliteResearchPersistence |
| 35 | from app.research_readiness import ( |
| 36 | FreshnessPolicyRegistry, |
| 37 | ResearchDataConfidence, |
| 38 | ResearchReadinessResult, |
| 39 | ResearchRequirementReadiness, |
| 40 | ResearchRequirementRegistry, |
| 41 | ResearchRequirementStatus, |
| 42 | ResearchSourceTier, |
| 43 | RuleEngineArea, |
| 44 | ) |
| 45 | from app.stock_rule_engine import ( |
| 46 | CURRENT_NEWS_WINDOW_DAYS, |
| 47 | STOCK_RULE_ENGINE_AREA_WEIGHTS, |
| 48 | STOCK_RULE_ENGINE_VERSION, |
| 49 | AreaScoreStatus, |
| 50 | ConfidenceLevel, |
| 51 | DecisionSignal, |
| 52 | StockRuleEngineInput, |
| 53 | StockRuleEngineService, |
| 54 | StockRuleEngineV1, |
| 55 | _structured_data, |
| 56 | ) |
| 57 | |
| 58 | |
| 59 | NOW = datetime(2026, 9, 10, 12, 0, tzinfo=timezone.utc) |
| 60 | INSTRUMENT_ID = UUID("00000000-0000-0000-0000-000000000145") |
| 61 | COMPANY_ID = UUID("00000000-0000-0000-0000-000000000245") |
| 62 | |
| 63 | |
| 64 | def _profile(*, country: str = "IN", exchange: str = "NSE") -> CompanyResearchProfile: |
| 65 | return CompanyResearchProfile( |
| 66 | instrument_id=INSTRUMENT_ID, |
| 67 | company_id=COMPANY_ID, |
| 68 | company_name="Deterministic Industries", |
| 69 | ticker="DET", |
| 70 | exchange=exchange, |
| 71 | mic="XNSE" if country == "IN" else "XNAS", |
| 72 | country=country, |
| 73 | currency="INR" if country == "IN" else "USD", |
| 74 | ) |
| 75 | |
| 76 | |
| 77 | def _readiness( |
| 78 | overrides: dict[str, ResearchRequirementStatus] | None = None, |
| 79 | *, |
| 80 | unsupported: set[str] | None = None, |
| 81 | critical_pct: int = 100, |
| 82 | overall_pct: int = 100, |
| 83 | ) -> ResearchReadinessResult: |
| 84 | overrides = overrides or {} |
| 85 | unsupported = unsupported or set() |
| 86 | registry = ResearchRequirementRegistry.default() |
| 87 | freshness = FreshnessPolicyRegistry.default() |
| 88 | items = [] |
| 89 | for requirement in registry.requirements: |
| 90 | status = ( |
| 91 | ResearchRequirementStatus.UNSUPPORTED |
| 92 | if requirement.requirement_id in unsupported |
| 93 | else overrides.get(requirement.requirement_id, ResearchRequirementStatus.READY_FRESH) |
| 94 | ) |
| 95 | covered = tuple(item.input_id for item in requirement.inputs) if status in { |
| 96 | ResearchRequirementStatus.READY_FRESH, |
| 97 | ResearchRequirementStatus.READY_STALE, |
| 98 | } else () |
| 99 | items.append( |
| 100 | ResearchRequirementReadiness( |
| 101 | requirement_id=requirement.requirement_id, |
| 102 | rule_engine_area=requirement.rule_engine_area, |
| 103 | mandatory=requirement.mandatory, |
| 104 | status=status, |
| 105 | source=None if status == ResearchRequirementStatus.UNSUPPORTED else "NSE", |
| 106 | source_tier=None if status == ResearchRequirementStatus.UNSUPPORTED else ResearchSourceTier.OFFICIAL, |
| 107 | as_of=None if status == ResearchRequirementStatus.UNSUPPORTED else NOW, |
| 108 | retrieved_at=None if status == ResearchRequirementStatus.UNSUPPORTED else NOW, |
| 109 | age=None if status == ResearchRequirementStatus.UNSUPPORTED else timedelta(0), |
| 110 | freshness_policy=freshness.get(requirement.freshness_policy_id), |
| 111 | evidence_ids=() if status == ResearchRequirementStatus.UNSUPPORTED else (f"evidence:{requirement.requirement_id}",), |
| 112 | missing_reason="MISSING" if status == ResearchRequirementStatus.MISSING else None, |
| 113 | conflict_reason="CONFLICT" if status == ResearchRequirementStatus.CONFLICTING else None, |
| 114 | supported_actions=(), |
| 115 | importance=requirement.importance, |
| 116 | source_url=None if status == ResearchRequirementStatus.UNSUPPORTED else "https://www.nseindia.com/filing.pdf", |
| 117 | covered_input_ids=covered, |
| 118 | missing_input_ids=tuple(item.input_id for item in requirement.inputs if item.input_id not in covered), |
| 119 | coverage_pct=100 if covered else 0, |
| 120 | critical_coverage_pct=100 if covered else 0, |
| 121 | ) |
| 122 | ) |
| 123 | return ResearchReadinessResult( |
| 124 | global_instrument_id=INSTRUMENT_ID, |
| 125 | requirements=tuple(items), |
| 126 | generated_at=NOW, |
| 127 | overall_status=ResearchRequirementStatus.READY_FRESH if not overrides else ResearchRequirementStatus.PARTIAL, |
| 128 | overall_completeness_pct=overall_pct, |
| 129 | critical_completeness_pct=critical_pct, |
| 130 | confidence=ResearchDataConfidence.HIGH, |
| 131 | confidence_pct=100, |
| 132 | ) |
| 133 | |
| 134 | |
| 135 | def _pv(value, *, source="NSE", url="https://www.nseindia.com/filing.pdf", as_of=NOW): |
| 136 | return ProvenancedValue( |
| 137 | value=value, |
| 138 | unit="INR", |
| 139 | as_of_date=as_of, |
| 140 | source_url=url, |
| 141 | source_name=source, |
| 142 | source_type="EXCHANGE_FILING", |
| 143 | published_at=as_of, |
| 144 | retrieved_at=NOW, |
| 145 | confidence=0.95, |
| 146 | ) |
| 147 | |
| 148 | |
| 149 | def _fact(metric: str, value, period: datetime, period_type: str = "ANNUAL") -> FinancialFact: |
| 150 | return FinancialFact( |
| 151 | FinancialFactKey(INSTRUMENT_ID, metric, period.date().isoformat(), period_type, "CONSOLIDATED"), |
| 152 | _pv(value, as_of=period), |
| 153 | FactSourceTier.OFFICIAL_NSE, |
| 154 | "NSE", |
| 155 | f"filing:{period.date()}:{metric}", |
| 156 | SourceMode.REAL, |
| 157 | ) |
| 158 | |
| 159 | |
| 160 | def _structured(**updates) -> StructuredMarketSnapshotRecord: |
| 161 | facts = { |
| 162 | "trailingPE": _pv(Decimal("16"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 163 | "forwardPE": _pv(Decimal("14"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 164 | "priceToBook": _pv(Decimal("2.2"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 165 | "evToEbitda": _pv(Decimal("8"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 166 | "pegRatio": _pv(Decimal("1.1"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 167 | "marketCap": _pv(Decimal("10000"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 168 | "freeCashFlow": _pv(Decimal("650"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 169 | "roe": _pv(Decimal("0.22"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 170 | "roce": _pv(Decimal("0.24"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 171 | "operatingMargin": _pv(Decimal("0.18"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 172 | "profitMargin": _pv(Decimal("0.12"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 173 | "debtToEquity": _pv(Decimal("35"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 174 | "totalDebt": _pv(Decimal("900"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 175 | "totalCash": _pv(Decimal("400"), source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 176 | "sector": _pv("Industrials", source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS"), |
| 177 | } |
| 178 | for key, raw in updates.items(): |
| 179 | if raw is None: |
| 180 | facts.pop(key, None) |
| 181 | else: |
| 182 | facts[key] = _pv(raw, source="Yahoo Finance", url="https://finance.yahoo.com/quote/DET.NS") |
| 183 | resolution = StructuredInstrumentResolution( |
| 184 | instrument_id=INSTRUMENT_ID, |
| 185 | provider="YAHOO", |
| 186 | provider_ticker="DET.NS", |
| 187 | company_name="Deterministic Industries", |
| 188 | exchange="NSE", |
| 189 | currency="INR", |
| 190 | confidence=1, |
| 191 | resolved_at=NOW, |
| 192 | ) |
| 193 | snapshot = StructuredMarketSnapshot( |
| 194 | resolution=resolution, |
| 195 | status="SUCCESS", |
| 196 | retrieved_at=NOW, |
| 197 | market_as_of=NOW, |
| 198 | source_name="Yahoo Finance", |
| 199 | source_url="https://finance.yahoo.com/quote/DET.NS", |
| 200 | facts=facts, |
| 201 | ) |
| 202 | return StructuredMarketSnapshotRecord( |
| 203 | instrument_id=INSTRUMENT_ID, |
| 204 | provider="YAHOO_FINANCE", |
| 205 | provider_instrument_id="DET.NS", |
| 206 | exchange="NSE", |
| 207 | mic="XNSE", |
| 208 | currency="INR", |
| 209 | source_url=snapshot.source_url, |
| 210 | source_name=snapshot.source_name, |
| 211 | source_type="STRUCTURED_MARKET_PROVIDER", |
| 212 | source_identity="DET.NS", |
| 213 | market_as_of=NOW, |
| 214 | retrieved_at=NOW, |
| 215 | persisted_at=NOW, |
| 216 | last_success_at=NOW, |
| 217 | snapshot=snapshot, |
| 218 | ) |
| 219 | |
| 220 | |
| 221 | def _prices(count: int = 180, *, falling: bool = False) -> tuple[MarketPriceObservation, ...]: |
| 222 | values = [] |
| 223 | for index in range(count): |
| 224 | price = Decimal("200") - Decimal(index) if falling else Decimal("100") + Decimal(index) / 2 |
| 225 | values.append( |
| 226 | MarketPriceObservation( |
| 227 | instrument_id=INSTRUMENT_ID, |
| 228 | observed_at=NOW - timedelta(days=count - 1 - index), |
| 229 | price=max(Decimal("1"), price), |
| 230 | currency="INR", |
| 231 | provider="YAHOO_FINANCE", |
| 232 | source_url="https://finance.yahoo.com/quote/DET.NS/history", |
| 233 | retrieved_at=NOW, |
| 234 | ) |
| 235 | ) |
| 236 | return tuple(values) |
| 237 | |
| 238 | |
| 239 | def _base_facts(*, declining: bool = False, leverage: Decimal = Decimal("900")) -> tuple[FinancialFact, ...]: |
| 240 | facts = [] |
| 241 | annual_dates = [datetime(year, 3, 31, tzinfo=timezone.utc) for year in (2023, 2024, 2025, 2026)] |
| 242 | revenues = [1000, 1200, 1450, 1750] if not declining else [1750, 1450, 1200, 950] |
| 243 | earnings = [90, 115, 145, 190] if not declining else [190, 140, 100, 60] |
| 244 | for period, revenue, pat in zip(annual_dates, revenues, earnings): |
| 245 | facts.extend((_fact("revenue", revenue, period), _fact("pat", pat, period))) |
| 246 | latest = annual_dates[-1] |
| 247 | facts.extend( |
| 248 | ( |
| 249 | _fact("equity", 2500, latest), |
| 250 | _fact("debt_or_borrowings", leverage, latest), |
| 251 | _fact("cash_and_cash_equivalents", 400, latest), |
| 252 | _fact("current_assets", 1800, latest), |
| 253 | _fact("current_liabilities", 1000, latest), |
| 254 | _fact("ebitda", 320, latest), |
| 255 | _fact("finance_cost", 40, latest), |
| 256 | _fact("operating_cash_flow", 230, latest), |
| 257 | _fact("free_cash_flow", 150, latest), |
| 258 | ) |
| 259 | ) |
| 260 | quarter_dates = [ |
| 261 | datetime(2024, 6, 30, tzinfo=timezone.utc), datetime(2024, 9, 30, tzinfo=timezone.utc), |
| 262 | datetime(2024, 12, 31, tzinfo=timezone.utc), datetime(2025, 3, 31, tzinfo=timezone.utc), |
| 263 | datetime(2025, 6, 30, tzinfo=timezone.utc), datetime(2025, 9, 30, tzinfo=timezone.utc), |
| 264 | datetime(2025, 12, 31, tzinfo=timezone.utc), datetime(2026, 3, 31, tzinfo=timezone.utc), |
| 265 | ] |
| 266 | for index, period in enumerate(quarter_dates): |
| 267 | revenue = Decimal(250 + index * 12) |
| 268 | pat = Decimal(22 + index * 2) |
| 269 | facts.extend( |
| 270 | ( |
| 271 | _fact("revenue", revenue, period, "QUARTERLY"), |
| 272 | _fact("pat", pat, period, "QUARTERLY"), |
| 273 | _fact("eps", pat / 10, period, "QUARTERLY"), |
| 274 | _fact("ebitda", revenue * Decimal("0.18"), period, "QUARTERLY"), |
| 275 | ) |
| 276 | ) |
| 277 | return tuple(facts) |
| 278 | |
| 279 | |
| 280 | def _event( |
| 281 | *, |
| 282 | event_type=ResearchEventType.MAJOR_CONTRACT, |
| 283 | impact=EventImpact.POSITIVE, |
| 284 | at=NOW - timedelta(days=3), |
| 285 | title="Material contract awarded", |
| 286 | summary="Exchange filing confirms a major customer contract.", |
| 287 | classification=SourceClassification.EXCHANGE, |
| 288 | reliability=ReliabilityLevel.LEVEL_A, |
| 289 | status=ResearchLifecycleStatus.VALIDATED, |
| 290 | confidence=0.95, |
| 291 | monetary=Decimal("500"), |
| 292 | ) -> ResearchEvent: |
| 293 | document_id = uuid4() |
| 294 | return ResearchEvent( |
| 295 | instrument_id=INSTRUMENT_ID, |
| 296 | company_id=COMPANY_ID, |
| 297 | event_type=event_type, |
| 298 | event_date=at, |
| 299 | detected_at=NOW, |
| 300 | title=title, |
| 301 | summary=summary, |
| 302 | source_document_id=document_id, |
| 303 | source_url=f"https://www.nseindia.com/event/{document_id}", |
| 304 | source_type=SourceType.EXCHANGE_ANNOUNCEMENT, |
| 305 | source_classification=classification, |
| 306 | reliability=reliability, |
| 307 | source_mode=SourceMode.REAL, |
| 308 | confidence=confidence, |
| 309 | impact=impact, |
| 310 | time_horizon=TimeHorizon.MEDIUM_TERM, |
| 311 | monetary_value=monetary, |
| 312 | counterparty="Material Customer" if monetary is not None else None, |
| 313 | status=status, |
| 314 | raw_evidence_reference=title, |
| 315 | published_at=at, |
| 316 | retrieved_at=NOW, |
| 317 | independence_key=str(document_id), |
| 318 | ) |
| 319 | |
| 320 | |
| 321 | def _shareholding() -> tuple[ShareholdingSnapshot, ...]: |
| 322 | snapshots = [] |
| 323 | for period, promoter, fii, pledge in ( |
| 324 | (datetime(2025, 12, 31, tzinfo=timezone.utc), "51", "12", "4"), |
| 325 | (datetime(2026, 3, 31, tzinfo=timezone.utc), "52", "13", "2"), |
| 326 | ): |
| 327 | snapshots.append( |
| 328 | ShareholdingSnapshot( |
| 329 | instrument_id=INSTRUMENT_ID, |
| 330 | period_end=period, |
| 331 | source_provider="NSE", |
| 332 | source_type="EXCHANGE_FILING", |
| 333 | source_identity_key=f"shareholding:{period.date()}", |
| 334 | source_url=f"https://www.nseindia.com/shareholding/{period.date()}", |
| 335 | published_at=period + timedelta(days=20), |
| 336 | retrieved_at=NOW, |
| 337 | confidence=Decimal("0.99"), |
| 338 | reliability_level=ReliabilityLevel.LEVEL_A, |
| 339 | source_mode=SourceMode.REAL, |
| 340 | values=[ |
| 341 | ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER, percentage=Decimal(promoter)), |
| 342 | ShareholdingSnapshotValue(category=ShareholdingCategory.FII_FPI, percentage=Decimal(fii)), |
| 343 | ShareholdingSnapshotValue(category=ShareholdingCategory.PROMOTER_PLEDGE, percentage=Decimal(pledge), metric_basis="PERCENT_OF_PROMOTER_HOLDING"), |
| 344 | ], |
| 345 | ) |
| 346 | ) |
| 347 | return tuple(snapshots) |
| 348 | |
| 349 | |
| 350 | def _inputs( |
| 351 | *, |
| 352 | readiness=None, |
| 353 | facts=None, |
| 354 | structured=None, |
| 355 | prices=None, |
| 356 | events=None, |
| 357 | shareholding=None, |
| 358 | profile=None, |
| 359 | sector="Industrials", |
| 360 | ) -> StockRuleEngineInput: |
| 361 | return StockRuleEngineInput( |
| 362 | profile=profile or _profile(), |
| 363 | readiness=readiness or _readiness(), |
| 364 | financial_facts=tuple(_base_facts() if facts is None else facts), |
| 365 | structured_snapshots=tuple((_structured(),) if structured is None else structured), |
| 366 | market_prices=tuple(_prices() if prices is None else prices), |
| 367 | events=tuple((_event(),) if events is None else events), |
| 368 | shareholding=tuple(_shareholding() if shareholding is None else shareholding), |
| 369 | canonical_metadata={"canonicalSector": sector, "updatedAt": NOW.isoformat()}, |
| 370 | evaluated_at=NOW, |
| 371 | ) |
| 372 | |
| 373 | |
| 374 | def _area(result, area: RuleEngineArea): |
| 375 | return next(item for item in result.area_scores if item.area == area) |
| 376 | |
| 377 | |
| 378 | def test_nonpositive_structured_latest_price_is_excluded_from_rule_inputs() -> None: |
| 379 | structured = _structured(latestPrice=Decimal("0")) |
| 380 | data = _structured_data(_inputs(structured=(structured,), prices=())) |
| 381 | |
| 382 | assert not any(key.replace("_", "").lower() == "latestprice" for key in data) |
| 383 | |
| 384 | |
| 385 | def test_v1_top_level_weights_total_exactly_100_and_cover_all_areas(): |
| 386 | assert STOCK_RULE_ENGINE_VERSION == "STOCK_RULE_ENGINE_V1" |
| 387 | assert sum(STOCK_RULE_ENGINE_AREA_WEIGHTS.values()) == 100 |
| 388 | assert set(STOCK_RULE_ENGINE_AREA_WEIGHTS) == set(RuleEngineArea) |
| 389 | |
| 390 | |
| 391 | def test_each_area_contribution_uses_configured_weight_and_run_is_deterministic(): |
| 392 | engine = StockRuleEngineV1() |
| 393 | inputs = _inputs() |
| 394 | first = engine.evaluate(inputs, allow_partial=False) |
| 395 | second = engine.evaluate(inputs, allow_partial=False) |
| 396 | assert first.model_dump() == second.model_dump() |
| 397 | for area in first.area_scores: |
| 398 | assert area.weight == STOCK_RULE_ENGINE_AREA_WEIGHTS[RuleEngineArea(area.area)] |
| 399 | if area.raw_score is not None: |
| 400 | assert area.weighted_contribution == pytest.approx(area.raw_score * area.weight / 100, abs=0.01) |
| 401 | |
| 402 | |
| 403 | def test_critical_missing_blocks_full_analysis_and_strong_buy(): |
| 404 | readiness = _readiness({"LATEST_PRICE": ResearchRequirementStatus.MISSING}, critical_pct=70, overall_pct=90) |
| 405 | result = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=False) |
| 406 | assert result.decision_signal == DecisionSignal.INSUFFICIENT_DATA |
| 407 | assert result.overall_score is None |
| 408 | assert not result.eligibility.full_analysis_allowed |
| 409 | |
| 410 | |
| 411 | def test_partial_analysis_requires_explicit_flag_and_is_capped_at_hold(): |
| 412 | readiness = _readiness({"GROWTH_FACTS": ResearchRequirementStatus.MISSING}, critical_pct=90, overall_pct=82) |
| 413 | denied = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=False) |
| 414 | allowed = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=True) |
| 415 | assert denied.decision_signal == DecisionSignal.INSUFFICIENT_DATA |
| 416 | assert denied.overall_score is None |
| 417 | assert allowed.overall_score is not None |
| 418 | assert allowed.decision_signal not in {DecisionSignal.STRONG_BUY, DecisionSignal.BUY, DecisionSignal.ACCUMULATE} |
| 419 | |
| 420 | |
| 421 | def test_unsupported_optional_regional_requirement_is_not_a_bad_score(): |
| 422 | readiness = _readiness(unsupported={"SHAREHOLDING"}) |
| 423 | usa = _inputs(readiness=readiness, profile=_profile(country="US", exchange="NASDAQ"), shareholding=(), sector="Technology") |
| 424 | result = StockRuleEngineV1().evaluate(usa, allow_partial=False) |
| 425 | shareholding = _area(result, RuleEngineArea.SHAREHOLDING) |
| 426 | assert not shareholding.applicable |
| 427 | assert shareholding.status == AreaScoreStatus.UNSUPPORTED |
| 428 | assert shareholding.raw_score is None |
| 429 | |
| 430 | |
| 431 | def test_cheaper_otherwise_equal_valuation_scores_better_and_optional_missing_is_omitted(): |
| 432 | engine = StockRuleEngineV1() |
| 433 | cheap = engine._valuation(_inputs(structured=(_structured(trailingPE=8, forwardPE=7),))) |
| 434 | expensive = engine._valuation(_inputs(structured=(_structured(trailingPE=45, forwardPE=40),))) |
| 435 | optional_missing = engine._valuation(_inputs(structured=(_structured(pegRatio=None, evToEbitda=None),))) |
| 436 | assert cheap.raw_score > expensive.raw_score |
| 437 | assert optional_missing.raw_score is not None |
| 438 | assert all(item.metric not in {"PEG", "EV_EBITDA"} for item in optional_missing.metrics) |
| 439 | |
| 440 | |
| 441 | def test_valuation_sector_applicability_changes_pb_and_excludes_financial_ev_ebitda(): |
| 442 | industrial = StockRuleEngineV1()._valuation(_inputs(sector="Industrials")) |
| 443 | financial = StockRuleEngineV1()._valuation(_inputs(sector="Financial Services")) |
| 444 | assert any(item.rule == "PRICE_TO_BOOK_GENERAL_V1" for item in industrial.metrics) |
| 445 | assert any(item.rule == "PRICE_TO_BOOK_FINANCIAL_V1" for item in financial.metrics) |
| 446 | assert not any(item.metric == "EV_EBITDA" for item in financial.metrics) |
| 447 | |
| 448 | |
| 449 | def test_stronger_returns_and_margins_score_better_and_cash_conversion_penalizes(): |
| 450 | engine = StockRuleEngineV1() |
| 451 | strong = engine._quality(_inputs()) |
| 452 | weak = engine._quality(_inputs(structured=(_structured(roe=Decimal("0.03"), roce=Decimal("0.04"), operatingMargin=Decimal("0.03"), profitMargin=Decimal("0.01")),))) |
| 453 | cash_poor_facts = tuple(fact for fact in _base_facts() if fact.key.metric != "operating_cash_flow") + (_fact("operating_cash_flow", 5, datetime(2026, 3, 31, tzinfo=timezone.utc)),) |
| 454 | cash_poor = engine._quality(_inputs(facts=cash_poor_facts)) |
| 455 | assert strong.raw_score > weak.raw_score |
| 456 | assert next(item for item in cash_poor.metrics if item.metric == "CASH_CONVERSION").score < 40 |
| 457 | |
| 458 | |
| 459 | def test_sustained_growth_beats_decline_and_qoq_has_limited_weight(): |
| 460 | engine = StockRuleEngineV1() |
| 461 | growing = engine._growth(_inputs(facts=_base_facts())) |
| 462 | declining = engine._growth(_inputs(facts=_base_facts(declining=True))) |
| 463 | assert growing.raw_score > declining.raw_score |
| 464 | qoq = next(item for item in growing.metrics if item.metric == "RECENT_REVENUE_QOQ") |
| 465 | structural = next(item for item in growing.metrics if item.metric == "REVENUE_CAGR") |
| 466 | assert qoq.configured_subrule_weight == 6 |
| 467 | assert structural.configured_subrule_weight == 27 |
| 468 | |
| 469 | |
| 470 | def test_unsustainable_leverage_scores_worse_but_bank_skips_industrial_debt_rule(): |
| 471 | engine = StockRuleEngineV1() |
| 472 | sound = engine._balance_sheet(_inputs()) |
| 473 | stressed = engine._balance_sheet(_inputs(structured=(_structured(debtToEquity=250),))) |
| 474 | bank = engine._balance_sheet(_inputs(sector="Banks")) |
| 475 | assert sound.raw_score > stressed.raw_score |
| 476 | assert not any(item.rule == "INDUSTRIAL_DEBT_TO_EQUITY_V1" for item in bank.metrics) |
| 477 | |
| 478 | |
| 479 | def test_quarterly_uses_comparable_yoy_and_labels_sequential_as_limited_weight(): |
| 480 | area = StockRuleEngineV1()._quarterly(_inputs()) |
| 481 | revenue_yoy = next(item for item in area.metrics if item.metric == "REVENUE_YOY") |
| 482 | revenue_qoq = next(item for item in area.metrics if item.metric == "REVENUE_QOQ") |
| 483 | assert revenue_yoy.rule == "COMPARABLE_QUARTER_REVENUE_YOY_V1" |
| 484 | assert revenue_qoq.rule == "SEQUENTIAL_REVENUE_QOQ_LIMITED_WEIGHT_V1" |
| 485 | assert "nseindia.com" in (revenue_yoy.source_url or "") |
| 486 | |
| 487 | |
| 488 | def test_material_structured_catalyst_scores_but_keyword_only_document_cannot_enter_engine(): |
| 489 | engine = StockRuleEngineV1() |
| 490 | material = engine._catalysts(_inputs()) |
| 491 | keyword_only = engine._catalysts(_inputs(events=())) |
| 492 | immaterial_event = _event(monetary=None, classification=SourceClassification.REPUTABLE_NEWS, title="capacity mentioned", summary="A generic article says capacity.") |
| 493 | irrelevant = engine._catalysts(_inputs(events=(immaterial_event,))) |
| 494 | assert material.raw_score is not None |
| 495 | assert keyword_only.raw_score is None |
| 496 | assert irrelevant.raw_score is None |
| 497 | |
| 498 | |
| 499 | def test_technical_uses_durable_prices_and_preserves_50_150_median_names(): |
| 500 | area = StockRuleEngineV1()._technical(_inputs()) |
| 501 | rules = {item.rule for item in area.metrics} |
| 502 | assert "PRICE_VS_50_OBSERVATION_MEDIAN_V1" in rules |
| 503 | assert "PRICE_VS_150_OBSERVATION_MEDIAN_V1" in rules |
| 504 | assert all("AVERAGE" not in rule and "YAHOO_FETCH" not in rule for rule in rules) |
| 505 | |
| 506 | |
| 507 | @pytest.mark.parametrize( |
| 508 | ("age_days", "eligible"), |
| 509 | [(CURRENT_NEWS_WINDOW_DAYS, True), (CURRENT_NEWS_WINDOW_DAYS + 1, False)], |
| 510 | ) |
| 511 | def test_current_news_inclusive_30_day_boundary(age_days, eligible): |
| 512 | event = _event(at=NOW - timedelta(days=age_days)) |
| 513 | area = StockRuleEngineV1()._news(_inputs(events=(event,))) |
| 514 | assert (area.raw_score is not None) is eligible |
| 515 | |
| 516 | |
| 517 | def test_irrelevant_geopolitical_event_excluded_but_relevant_exposure_changes_news_score(): |
| 518 | unrelated = _event(event_type=ResearchEventType.OTHER, impact=EventImpact.STRONG_NEGATIVE, title="War disrupts oil supply", summary="Oil prices rise.", classification=SourceClassification.REPUTABLE_NEWS) |
| 519 | unrelated_area = StockRuleEngineV1()._news(_inputs(events=(unrelated,), sector="Software")) |
| 520 | airline_area = StockRuleEngineV1()._news(_inputs(events=(unrelated,), sector="Airlines")) |
| 521 | assert unrelated_area.raw_score is None |
| 522 | assert airline_area.raw_score is not None |
| 523 | assert airline_area.raw_score < 50 |
| 524 | |
| 525 | |
| 526 | def test_positive_and_negative_relevant_current_events_move_score_and_duplicates_deduplicate(): |
| 527 | positive = _event(impact=EventImpact.STRONG_POSITIVE) |
| 528 | negative = positive.model_copy(update={"impact": EventImpact.STRONG_NEGATIVE}) |
| 529 | engine = StockRuleEngineV1() |
| 530 | assert engine._news(_inputs(events=(positive,))).raw_score > engine._news(_inputs(events=(negative,))).raw_score |
| 531 | duplicate = positive.model_copy(update={"event_id": uuid4()}) |
| 532 | assert len(engine._news(_inputs(events=(positive, duplicate))).metrics) == 1 |
| 533 | |
| 534 | |
| 535 | def test_india_shareholding_applies_while_non_india_is_unsupported_without_penalty(): |
| 536 | engine = StockRuleEngineV1() |
| 537 | india = engine._shareholding(_inputs()) |
| 538 | usa = engine._shareholding(_inputs(profile=_profile(country="US", exchange="NASDAQ"), shareholding=())) |
| 539 | assert india.applicable and india.raw_score is not None |
| 540 | assert not usa.applicable and usa.status == AreaScoreStatus.UNSUPPORTED |
| 541 | |
| 542 | |
| 543 | def test_old_unresolved_governance_remains_scored_and_resolved_is_handled_differently(): |
| 544 | old = _event(event_type=ResearchEventType.REGULATORY_EVENT, impact=EventImpact.STRONG_NEGATIVE, at=NOW - timedelta(days=500), title="Regulatory enforcement", summary="Official enforcement remains unresolved.") |
| 545 | resolved = old.model_copy(update={"event_id": uuid4(), "title": "Regulatory matter resolved", "summary": "Matter remediated and closed."}) |
| 546 | engine = StockRuleEngineV1() |
| 547 | unresolved_area = engine._governance(_inputs(events=(old,))) |
| 548 | resolved_area = engine._governance(_inputs(events=(resolved,))) |
| 549 | assert unresolved_area.raw_score < resolved_area.raw_score |
| 550 | assert old.event_id.hex in "".join(unresolved_area.evidence_references).replace("-", "") |
| 551 | |
| 552 | |
| 553 | def test_authoritative_severe_governance_can_override_but_weak_news_cannot(): |
| 554 | severe = _event(event_type=ResearchEventType.REGULATORY_EVENT, impact=EventImpact.STRONG_NEGATIVE, title="Confirmed accounting fraud", summary="Exchange confirms accounting fraud.") |
| 555 | weak = severe.model_copy(update={"event_id": uuid4(), "source_classification": SourceClassification.REPUTABLE_NEWS, "reliability": ReliabilityLevel.LEVEL_C, "confidence": 0.55}) |
| 556 | engine = StockRuleEngineV1() |
| 557 | severe_result = engine.evaluate(_inputs(events=(severe,)), allow_partial=False) |
| 558 | weak_result = engine.evaluate(_inputs(events=(weak,)), allow_partial=False) |
| 559 | assert any(item.code == "CONFIRMED_FRAUD_OR_ACCOUNTING_CRISIS" for item in severe_result.risk_overrides) |
| 560 | assert severe_result.decision_signal == DecisionSignal.EXIT_REVIEW |
| 561 | assert not weak_result.risk_overrides |
| 562 | |
| 563 | |
| 564 | def test_confidence_is_separate_from_raw_score_conflicts_lower_it_and_authority_raises_it(): |
| 565 | engine = StockRuleEngineV1() |
| 566 | high = engine.evaluate(_inputs(), allow_partial=False) |
| 567 | conflict_readiness = _readiness({"VALUATION_INPUTS": ResearchRequirementStatus.CONFLICTING}, critical_pct=80, overall_pct=90) |
| 568 | low = engine.evaluate(_inputs(readiness=conflict_readiness), allow_partial=True) |
| 569 | assert high.confidence == ConfidenceLevel.HIGH |
| 570 | assert low.confidence_score < high.confidence_score |
| 571 | assert high.overall_score != high.confidence_score |
| 572 | |
| 573 | |
| 574 | def test_same_inputs_have_same_fingerprint_and_relevant_change_invalidates_it(): |
| 575 | engine = StockRuleEngineV1() |
| 576 | inputs = _inputs() |
| 577 | first = engine.input_fingerprint(inputs, allow_partial=False) |
| 578 | same = engine.input_fingerprint(inputs, allow_partial=False) |
| 579 | changed = engine.input_fingerprint(_inputs(structured=(_structured(trailingPE=17),)), allow_partial=False) |
| 580 | changed_readiness = replace( |
| 581 | inputs.readiness, |
| 582 | requirements=tuple( |
| 583 | replace(item, source_url="https://www.nseindia.com/revised-filing.pdf") |
| 584 | if item.requirement_id == "QUARTERLY_FINANCIALS" |
| 585 | else item |
| 586 | for item in inputs.readiness.requirements |
| 587 | ), |
| 588 | ) |
| 589 | provenance_changed = engine.input_fingerprint( |
| 590 | _inputs(readiness=changed_readiness), allow_partial=False |
| 591 | ) |
| 592 | assert first == same |
| 593 | assert first != changed |
| 594 | assert first != provenance_changed |
| 595 | |
| 596 | |
| 597 | def test_global_score_persistence_cache_and_private_portfolio_field_guard(): |
| 598 | persistence = SqliteResearchPersistence() |
| 599 | result = StockRuleEngineV1().evaluate(_inputs(), allow_partial=False).model_dump(mode="json", by_alias=False) |
| 600 | persistence.upsert_stock_rule_engine_result(result) |
| 601 | loaded = persistence.load_stock_rule_engine_result(INSTRUMENT_ID, STOCK_RULE_ENGINE_VERSION, result["input_fingerprint"]) |
| 602 | assert loaded["input_fingerprint"] == result["input_fingerprint"] |
| 603 | assert "portfolio_id" not in loaded |
| 604 | with pytest.raises(ValueError, match="PRIVATE_PORTFOLIO_FIELD"): |
| 605 | persistence.upsert_stock_rule_engine_result({**result, "portfolio_id": str(uuid4())}) |
| 606 | |
| 607 | |
| 608 | class _CacheRepository: |
| 609 | def __init__(self, inputs): |
| 610 | self.inputs = inputs |
| 611 | self.cache = {} |
| 612 | self.provider_calls = 0 |
| 613 | self.portfolio_mutations = 0 |
| 614 | self.watchlist_mutations = 0 |
| 615 | |
| 616 | async def financial_facts_for_instruments(self, ids): return {INSTRUMENT_ID: list(self.inputs.financial_facts)} |
| 617 | async def structured_market_snapshots_for_instruments(self, ids): return {INSTRUMENT_ID: list(self.inputs.structured_snapshots)} |
| 618 | async def market_price_observations_for_instruments(self, ids): return {INSTRUMENT_ID: list(self.inputs.market_prices)} |
| 619 | def events_for(self, instrument_id, source_mode=None): return list(self.inputs.events) |
| 620 | def shareholding_for(self, instrument_id, limit=8): return list(self.inputs.shareholding) |
| 621 | async def stock_rule_engine_result(self, instrument_id, version, fingerprint): return self.cache.get((str(instrument_id), version, fingerprint)) |
| 622 | async def persist_stock_rule_engine_result(self, result): self.cache[(str(result["global_instrument_id"]), result["rule_engine_version"], result["input_fingerprint"])] = result |
| 623 | async def fetch_yahoo(self): self.provider_calls += 1; raise AssertionError("provider called") |
| 624 | async def mutate_portfolio(self): self.portfolio_mutations += 1; raise AssertionError("portfolio mutated") |
| 625 | async def mutate_watchlist(self): self.watchlist_mutations += 1; raise AssertionError("watchlist mutated") |
| 626 | |
| 627 | |
| 628 | class _Metadata: |
| 629 | def __init__(self): self.metadata = {"canonicalSector": "Industrials", "updatedAt": NOW.isoformat()} |
| 630 | def canonical_metadata_for(self, instrument_id): return dict(self.metadata) |
| 631 | def remember_canonical_metadata(self, instrument_id, metadata): self.metadata = dict(metadata) |
| 632 | |
| 633 | |
| 634 | @pytest.mark.asyncio |
| 635 | async def test_analysis_service_is_provider_free_non_held_safe_and_reuses_exact_cache(): |
| 636 | inputs = _inputs() |
| 637 | repository = _CacheRepository(inputs) |
| 638 | service = StockRuleEngineService(repository, _Metadata()) |
| 639 | first = await service.analyze(inputs.profile, inputs.readiness, allow_partial=False, now=NOW) |
| 640 | second = await service.analyze(inputs.profile, inputs.readiness, allow_partial=False, now=NOW) |
| 641 | assert not first.cache_hit |
| 642 | assert second.cache_hit |
| 643 | assert first.input_fingerprint == second.input_fingerprint |
| 644 | assert first.overall_score == second.overall_score |
| 645 | assert repository.provider_calls == repository.portfolio_mutations == repository.watchlist_mutations == 0 |
| 646 | |
| 647 | |
| 648 | def test_score_record_has_explainable_metrics_sources_and_no_portfolio_fields(): |
| 649 | result = StockRuleEngineV1().evaluate(_inputs(), allow_partial=False) |
| 650 | payload = result.model_dump(mode="json", by_alias=True) |
| 651 | text = str(payload).casefold() |
| 652 | assert payload["ruleEngineVersion"] == STOCK_RULE_ENGINE_VERSION |
| 653 | assert payload["evidenceReferences"] |
| 654 | assert any(metric["sourceUrl"] for area in payload["areaScores"] for metric in area["metrics"]) |
| 655 | quarterly = next( |
| 656 | area for area in payload["areaScores"] |
| 657 | if area["area"] == "QUARTERLY_EARNINGS_TREND" |
| 658 | ) |
| 659 | assert any( |
| 660 | source["sourceProvider"] == "NSE" |
| 661 | and "nseindia.com" in source["sourceUrl"] |
| 662 | for source in quarterly["sourceReferences"] |
| 663 | ) |
| 664 | assert any(reference.startswith("evidence:") for reference in quarterly["evidenceReferences"]) |
| 665 | assert all(term not in text for term in ("portfolioid", "averagecost", "costbasis", "quantity", "allocation")) |
| 666 | |
| 667 | |
| 668 | class _ApiRepository: |
| 669 | def __init__(self, profile): |
| 670 | self.value = profile |
| 671 | self.portfolio_mutations = 0 |
| 672 | self.watchlist_mutations = 0 |
| 673 | self.provider_calls = 0 |
| 674 | |
| 675 | def profile(self, instrument_id): |
| 676 | if instrument_id != self.value.instrument_id: |
| 677 | raise StopIteration |
| 678 | return self.value |
| 679 | |
| 680 | |
| 681 | class _ApiOrchestrator: |
| 682 | def __init__(self, profile): |
| 683 | self.profile = profile |
| 684 | self.identity_reads = 0 |
| 685 | self.portfolio_mutations = 0 |
| 686 | self.watchlist_mutations = 0 |
| 687 | self.provider_calls = 0 |
| 688 | |
| 689 | async def global_instrument_metadata(self, instrument_id, **kwargs): |
| 690 | self.identity_reads += 1 |
| 691 | return { |
| 692 | "globalInstrumentId": str(instrument_id), |
| 693 | "canonicalSector": "Industrials", |
| 694 | "updatedAt": NOW.isoformat(), |
| 695 | } |
| 696 | |
| 697 | def register_global_profile_metadata(self, instrument_id, metadata): |
| 698 | return instrument_id == self.profile.instrument_id |
| 699 | |
| 700 | |
| 701 | class _ApiReadinessRuntime: |
| 702 | def __init__(self, readiness): self.value = readiness; self.provider_calls = 0 |
| 703 | async def read(self, instrument_id, jurisdiction="GLOBAL"): return self.value |
| 704 | |
| 705 | |
| 706 | class _ApiAnalysisService: |
| 707 | def __init__(self, result): self.value = result; self.calls = [] |
| 708 | async def analyze(self, profile, readiness, *, allow_partial, now=None): |
| 709 | self.calls.append((profile.instrument_id, allow_partial)) |
| 710 | return self.value |
| 711 | |
| 712 | |
| 713 | def test_analysis_api_accepts_global_identity_and_performs_no_provider_or_private_mutation(monkeypatch): |
| 714 | profile = _profile() |
| 715 | readiness = _readiness() |
| 716 | result = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=False) |
| 717 | repository = _ApiRepository(profile) |
| 718 | orchestrator = _ApiOrchestrator(profile) |
| 719 | runtime = _ApiReadinessRuntime(readiness) |
| 720 | service = _ApiAnalysisService(result) |
| 721 | metadata = _Metadata() |
| 722 | monkeypatch.setattr(main, "repository", repository) |
| 723 | monkeypatch.setattr(main, "portfolio_orchestrator", orchestrator) |
| 724 | monkeypatch.setattr(main, "research_readiness_runtime", runtime) |
| 725 | monkeypatch.setattr(main, "research_readiness_adapter", metadata) |
| 726 | monkeypatch.setattr(main, "stock_rule_engine_service", service) |
| 727 | telemetry = [] |
| 728 | monkeypatch.setattr(main.logger, "info", lambda message, *args: telemetry.append(message % args)) |
| 729 | |
| 730 | response = TestClient(main.app).post( |
| 731 | f"/api/v1/research/analysis/{INSTRUMENT_ID}", |
| 732 | json={"allowPartial": False}, |
| 733 | headers={ |
| 734 | "X-AIP-User-Id": "held-user", |
| 735 | "X-AIP-User-Issuer": "gateway", |
| 736 | "X-AIP-User-Subject": "subject", |
| 737 | }, |
| 738 | ) |
| 739 | |
| 740 | assert response.status_code == 200 |
| 741 | assert response.json()["globalInstrumentId"] == str(INSTRUMENT_ID) |
| 742 | assert response.json()["ruleEngineVersion"] == STOCK_RULE_ENGINE_VERSION |
| 743 | assert service.calls == [(INSTRUMENT_ID, False)] |
| 744 | assert runtime.provider_calls == repository.provider_calls == orchestrator.provider_calls == 0 |
| 745 | assert repository.portfolio_mutations == repository.watchlist_mutations == 0 |
| 746 | assert orchestrator.portfolio_mutations == orchestrator.watchlist_mutations == 0 |
| 747 | assert any("operation=RULE_ENGINE_ANALYSIS" in entry for entry in telemetry) |
| 748 | |
| 749 | |
| 750 | def test_held_and_non_held_callers_receive_same_global_company_score(monkeypatch): |
| 751 | profile = _profile() |
| 752 | readiness = _readiness() |
| 753 | result = StockRuleEngineV1().evaluate(_inputs(readiness=readiness), allow_partial=False) |
| 754 | monkeypatch.setattr(main, "repository", _ApiRepository(profile)) |
| 755 | monkeypatch.setattr(main, "portfolio_orchestrator", _ApiOrchestrator(profile)) |
| 756 | monkeypatch.setattr(main, "research_readiness_runtime", _ApiReadinessRuntime(readiness)) |
| 757 | monkeypatch.setattr(main, "research_readiness_adapter", _Metadata()) |
| 758 | monkeypatch.setattr(main, "stock_rule_engine_service", _ApiAnalysisService(result)) |
| 759 | client = TestClient(main.app) |
| 760 | |
| 761 | held = client.post( |
| 762 | f"/api/v1/research/analysis/{INSTRUMENT_ID}", |
| 763 | headers={"X-AIP-User-Id": "held", "X-AIP-User-Issuer": "gateway", "X-AIP-User-Subject": "held"}, |
| 764 | json={"allowPartial": False}, |
| 765 | ).json() |
| 766 | non_held = client.post( |
| 767 | f"/api/v1/research/analysis/{INSTRUMENT_ID}", |
| 768 | headers={"X-AIP-User-Id": "watchlist", "X-AIP-User-Issuer": "gateway", "X-AIP-User-Subject": "watchlist"}, |
| 769 | json={"allowPartial": False}, |
| 770 | ).json() |
| 771 | |
| 772 | for key in ("inputFingerprint", "overallScore", "qualityScore", "opportunityScore", "riskScore", "decisionSignal", "areaScores"): |
| 773 | assert held[key] == non_held[key] |
| 774 | |
| 775 | def test_domain_not_applicable_has_no_zero_or_neutral_score(): |
| 776 | readiness = _readiness({"ORDER_BOOK_CAPEX_GUIDANCE": ResearchRequirementStatus.NOT_APPLICABLE}) |
| 777 | result = StockRuleEngineV1()._catalysts(_inputs(readiness=readiness)) |
| 778 | assert result.status == AreaScoreStatus.NOT_APPLICABLE |
| 779 | assert result.applicable is False |
| 780 | assert result.raw_score is None |
| 781 | assert result.metrics == [] |