| 1 | from datetime import datetime, timezone |
| 2 | from decimal import Decimal |
| 3 | from uuid import UUID |
| 4 | |
| 5 | import pytest |
| 6 | |
| 7 | from app.fact_precedence import FactSourceTier, FinancialFact, FinancialFactKey, merge_fact |
| 8 | from app.models import ProvenancedValue |
| 9 | from app.persistence import SqliteResearchPersistence |
| 10 | |
| 11 | |
| 12 | INSTRUMENT = UUID("aaaaaaaa-1111-1111-1111-aaaaaaaaaaaa") |
| 13 | |
| 14 | |
| 15 | def fact(metric: str, value, tier: FactSourceTier, *, period="2026-06-30", period_type="QUARTERLY", basis="CONSOLIDATED", provider="NSE"): |
| 16 | return FinancialFact( |
| 17 | FinancialFactKey(INSTRUMENT, metric, period, period_type, basis), |
| 18 | ProvenancedValue(value=value, unit="INR", source_url=f"https://example.invalid/{provider}", source_name=provider, |
| 19 | source_type="EXCHANGE" if tier == FactSourceTier.OFFICIAL_NSE else "STRUCTURED_MARKET_PROVIDER", |
| 20 | retrieved_at=datetime.now(timezone.utc)), |
| 21 | tier, provider, f"{provider}:{metric}:{period}", |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | def test_nse_wins_conflicts_and_yahoo_only_fills_missing() -> None: |
| 26 | nse_revenue = fact("revenue", Decimal("100"), FactSourceTier.OFFICIAL_NSE) |
| 27 | yahoo_revenue = fact("revenue", Decimal("98"), FactSourceTier.YAHOO, provider="YAHOO_FINANCE") |
| 28 | yahoo_debt = fact("debt", Decimal("40"), FactSourceTier.YAHOO, provider="YAHOO_FINANCE") |
| 29 | assert merge_fact(nse_revenue, yahoo_revenue) is nse_revenue |
| 30 | assert merge_fact(None, yahoo_debt) is yahoo_debt |
| 31 | |
| 32 | |
| 33 | def test_search_fills_only_absent_and_cannot_overwrite_yahoo_or_nse() -> None: |
| 34 | yahoo = fact("debt", Decimal("40"), FactSourceTier.YAHOO, provider="YAHOO_FINANCE") |
| 35 | search = fact("debt", Decimal("42"), FactSourceTier.SEARCH, provider="SEARCH") |
| 36 | absent = fact("capex", Decimal("8"), FactSourceTier.SEARCH, provider="SEARCH") |
| 37 | assert merge_fact(yahoo, search) is yahoo |
| 38 | assert merge_fact(None, absent) is absent |
| 39 | assert merge_fact(fact("pat", 12, FactSourceTier.OFFICIAL_NSE), fact("pat", 11, FactSourceTier.SEARCH, provider="SEARCH")).source_provider == "NSE" |
| 40 | |
| 41 | |
| 42 | def test_missing_fallback_never_erases_valid_value() -> None: |
| 43 | nse = fact("eps", Decimal("4.5"), FactSourceTier.OFFICIAL_NSE) |
| 44 | assert merge_fact(nse, fact("eps", None, FactSourceTier.YAHOO, provider="YAHOO_FINANCE")) is nse |
| 45 | |
| 46 | |
| 47 | def test_period_type_and_basis_are_collision_boundaries() -> None: |
| 48 | quarterly = fact("revenue", 100, FactSourceTier.OFFICIAL_NSE) |
| 49 | annual = fact("revenue", 400, FactSourceTier.YAHOO, period_type="ANNUAL", provider="YAHOO_FINANCE") |
| 50 | standalone = fact("revenue", 90, FactSourceTier.YAHOO, basis="STANDALONE", provider="YAHOO_FINANCE") |
| 51 | with pytest.raises(ValueError): merge_fact(quarterly, annual) |
| 52 | with pytest.raises(ValueError): merge_fact(quarterly, standalone) |
| 53 | |
| 54 | |
| 55 | def test_explicit_official_same_fact_correction_preserves_winning_provenance() -> None: |
| 56 | original = fact("revenue", 100, FactSourceTier.OFFICIAL_NSE, provider="NSE_RECORD_1") |
| 57 | correction = fact("revenue", 101, FactSourceTier.OFFICIAL_NSE, provider="NSE_RECORD_2") |
| 58 | assert merge_fact(original, correction) is original |
| 59 | accepted = merge_fact(original, correction, allow_same_tier_correction=True) |
| 60 | assert accepted is correction |
| 61 | assert accepted.value.source_name == "NSE_RECORD_2" |
| 62 | structured = fact("revenue", 102, FactSourceTier.STRUCTURED_FUNDAMENTALS, provider="EODHD") |
| 63 | assert merge_fact(structured, fact("revenue", 103, FactSourceTier.STRUCTURED_FUNDAMENTALS, provider="EODHD_CORRECTION"), allow_same_tier_correction=True) is structured |
| 64 | |
| 65 | |
| 66 | def test_durable_fact_upsert_round_trip_and_canonical_boundaries() -> None: |
| 67 | store = SqliteResearchPersistence() |
| 68 | q1 = fact("revenue", Decimal("100"), FactSourceTier.OFFICIAL_NSE) |
| 69 | q2 = fact("revenue", Decimal("110"), FactSourceTier.OFFICIAL_NSE, period="2026-09-30") |
| 70 | annual = fact("revenue", Decimal("400"), FactSourceTier.OFFICIAL_NSE, period_type="ANNUAL") |
| 71 | standalone = fact("revenue", Decimal("90"), FactSourceTier.OFFICIAL_NSE, basis="STANDALONE") |
| 72 | assert store.upsert_financial_fact(q1) |
| 73 | assert not store.upsert_financial_fact(fact("revenue", None, FactSourceTier.YAHOO, provider="YAHOO_FINANCE")) |
| 74 | assert store.upsert_financial_fact(q2) |
| 75 | assert store.upsert_financial_fact(annual) |
| 76 | assert store.upsert_financial_fact(standalone) |
| 77 | loaded = store.load_financial_facts() |
| 78 | assert len(loaded) == 4 |
| 79 | assert next(value for value in loaded if value.key == q1.key).value.value == Decimal("100") |
| 80 | assert next(value for value in loaded if value.key == q1.key).source_tier == FactSourceTier.OFFICIAL_NSE |
| 81 | |
| 82 | |
| 83 | def test_persisted_tier_three_is_official_nse_and_outranks_structured_fundamentals() -> None: |
| 84 | store = SqliteResearchPersistence() |
| 85 | nse = fact("revenue", Decimal("100"), FactSourceTier.OFFICIAL_NSE, basis="UNKNOWN") |
| 86 | assert int(FactSourceTier.OFFICIAL_NSE) == 3 |
| 87 | assert store.upsert_financial_fact(nse) |
| 88 | loaded = store.load_financial_facts()[0] |
| 89 | assert loaded.source_tier == FactSourceTier.OFFICIAL_NSE |
| 90 | assert not store.upsert_financial_fact(fact("revenue", Decimal("90"), FactSourceTier.STRUCTURED_FUNDAMENTALS, basis="UNKNOWN", provider="EODHD")) |
| 91 | |
| 92 | |
| 93 | def test_yahoo_round_trip_fills_only_same_unknown_identity_and_never_replaces_official() -> None: |
| 94 | store = SqliteResearchPersistence() |
| 95 | yahoo = fact("revenue", Decimal("98"), FactSourceTier.YAHOO, basis="UNKNOWN", provider="YAHOO_FINANCE") |
| 96 | official = fact("revenue", Decimal("100"), FactSourceTier.OFFICIAL_NSE, basis="UNKNOWN", provider="NSE") |
| 97 | annual = fact("revenue", Decimal("390"), FactSourceTier.YAHOO, period_type="ANNUAL", basis="UNKNOWN", provider="YAHOO_FINANCE") |
| 98 | assert store.upsert_financial_fact(yahoo) |
| 99 | assert store.upsert_financial_fact(official) |
| 100 | assert not store.upsert_financial_fact(fact("revenue", Decimal("99"), FactSourceTier.YAHOO, basis="UNKNOWN", provider="YAHOO_FINANCE")) |
| 101 | assert store.upsert_financial_fact(annual) |
| 102 | loaded = store.load_financial_facts() |
| 103 | assert len(loaded) == 2 |
| 104 | assert next(value for value in loaded if value.key.period_type == "QUARTERLY").source_provider == "NSE" |