@prakhar-gitlab / ai-investment-platform
Code Commits Issues Pulls Sessions CI/CD
main
main
py 253 lines 12.1 KB
Raw
1 from datetime import datetime, timedelta, timezone
2 from decimal import Decimal
3 from uuid import UUID
4
5 import httpx
6 import pytest
7
8 from app.global_scanner import CanonicalEquityUniverse, GlobalPreScore, GlobalScanner
9 from app.models import MarketPriceObservation, ProvenancedValue
10 from app.persistence import SqliteResearchPersistence
11 from app.fact_precedence import FinancialFact, FinancialFactKey, FactSourceTier
12 from test_structured_market_persistence import _record
13
14 NOW = datetime(2026, 9, 13, tzinfo=timezone.utc)
15
16
17 def instrument(n=1, **updates):
18 return dict(globalInstrumentId=str(UUID(int=n)), canonicalName=f"Company {n}", ticker=f"C{n}",
19 exchange="NSE", country="IN", currency="INR", status="ACTIVE", assetType="EQUITY",
20 providerMappings=[dict(provider="YAHOO_FINANCE", providerSymbol="ABC.NS", status="VERIFIED")], **updates)
21
22
23 def persisted(store, item, metrics=None, age=0, price=True):
24 key = UUID(item["globalInstrumentId"])
25 stamp = NOW - timedelta(days=age)
26 record = _record(key)
27 record.currency = item["currency"]
28 record.retrieved_at = record.market_as_of = stamp
29 record.snapshot.facts = {k: ProvenancedValue(value=v, source_url="https://example.test", source_name="test", retrieved_at=stamp, as_of_date=stamp)
30 for k, v in (metrics if metrics is not None else {"profitMargin": 10, "roe": 5}).items()}
31 store.upsert_structured_market_snapshot(record)
32 if price:
33 store.upsert_market_price_observation(MarketPriceObservation(instrument_id=key, observed_at=stamp, retrieved_at=stamp,
34 price=Decimal(100), currency=item["currency"], provider="YAHOO_FINANCE", source_url="https://example.test"))
35
36
37 async def scan(items, store, **kwargs):
38 def handler(request):
39 assert request.method == "GET" and request.url.path == "/api/v1/instruments"
40 assert request.url.params["status"] == "ACTIVE" and request.url.params["assetType"] == "EQUITY"
41 return httpx.Response(200, json={"instruments": items, "totalElements": len(items)})
42 async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
43 return await GlobalScanner(CanonicalEquityUniverse(client, "http://canonical"), store).scan(as_of=NOW, top_n=kwargs.get("top_n", 10))
44
45
46 @pytest.mark.asyncio
47 @pytest.mark.parametrize("change,reason", [({}, None), ({"status": "INACTIVE"}, "INACTIVE"),
48 ({"assetType": "ETF"}, "NON_EQUITY"), ({"providerMappings": []}, "UNTRUSTED_CANONICAL_IDENTITY"),
49 ({"canonicalName": None}, "UNTRUSTED_CANONICAL_IDENTITY")])
50 async def test_eligibility(change, reason):
51 store = SqliteResearchPersistence()
52 item = instrument() | change
53 persisted(store, item)
54 result = await scan([item], store)
55 candidate = result.candidates[0]
56 assert candidate.eligible_for_deep_analysis == (reason is None)
57 if reason: assert reason in candidate.exclusion_reasons
58
59
60 @pytest.mark.asyncio
61 async def test_no_price_and_stale_price_fail_critical_gate():
62 store = SqliteResearchPersistence()
63 items = [instrument(n) for n in range(1, 4)]
64 persisted(store, items[0])
65 persisted(store, items[1], price=False)
66 persisted(store, items[2], age=20)
67 result = await scan(items, store)
68 by_id = {c.global_instrument_id.int: c for c in result.candidates}
69 assert "NO_USABLE_PRICE" in by_id[2].exclusion_reasons
70 assert "STALE_PRICE" in by_id[3].exclusion_reasons
71 assert by_id[3].confidence < by_id[1].confidence
72 assert by_id[3].stale_inputs == ["price"]
73
74
75 @pytest.mark.asyncio
76 async def test_missing_optional_renormalizes_and_actual_zero_survives():
77 store = SqliteResearchPersistence()
78 items = [instrument(n) for n in range(1, 4)]
79 persisted(store, items[0], {"profitMargin": 10, "roe": 5, "revenueGrowth": None})
80 persisted(store, items[1], {"profitMargin": 10, "roe": 5, "revenueGrowth": 0})
81 persisted(store, items[2], {"profitMargin": 10, "roe": 5, "revenueGrowth": -10})
82 result = await scan(items, store)
83 a, b, c = result.candidates
84 assert a.dimensions["GROWTH_QUALITY"].score is None
85 assert b.dimensions["GROWTH_QUALITY"].score == 50
86 assert c.dimensions["GROWTH_QUALITY"].score == 0
87 assert a.pre_score > b.pre_score > c.pre_score
88
89
90 @pytest.mark.asyncio
91 async def test_determinism_ties_top_n_and_membership_independence():
92 store = SqliteResearchPersistence()
93 items = [instrument(n) for n in range(1, 13)]
94 for item in items: persisted(store, item)
95 first = await scan(items, store, top_n=7)
96 private_items = [item | {"portfolioId": "private", "quantity": 900, "averageCost": 1, "PnL": -500,
97 "allocation": .9, "watchlistMembership": True} for item in reversed(items)]
98 second = await scan(private_items, store, top_n=7)
99 assert first == second
100 assert [c.global_instrument_id.int for c in first.candidates] == list(range(1, 13))
101 assert len(first.deep_analysis_candidate_ids) == 7
102
103
104 @pytest.mark.asyncio
105 @pytest.mark.parametrize("country,exchange,currency", [("IN", "NSE", "INR"), ("US", "XNAS", "USD"), ("DE", "XETR", "EUR")])
106 async def test_region_neutral_public_evidence(country, exchange, currency):
107 store = SqliteResearchPersistence()
108 item = instrument() | dict(country=country, exchange=exchange, currency=currency)
109 persisted(store, item)
110 candidate = (await scan([item], store)).candidates[0]
111 assert candidate.eligible_for_deep_analysis
112 assert candidate.market == exchange and candidate.country == country
113
114
115 @pytest.mark.asyncio
116 async def test_provider_adapters_and_v1_never_invoked(monkeypatch):
117 def forbidden(*args, **kwargs): raise AssertionError("Provider or V1 invoked")
118 from app.structured_market import YahooFinanceProvider
119 from app.stock_rule_engine import StockRuleEngineService
120 monkeypatch.setattr(YahooFinanceProvider, "__init__", forbidden)
121 monkeypatch.setattr(StockRuleEngineService, "analyze", forbidden)
122 monkeypatch.setattr(httpx.Client, "send", forbidden)
123 store = SqliteResearchPersistence()
124 item = instrument()
125 persisted(store, item)
126 assert (await scan([item], store)).eligible_candidates == 1
127
128
129 @pytest.mark.asyncio
130 async def test_conflicting_financial_evidence_and_future_prices():
131 store = SqliteResearchPersistence()
132 item = instrument()
133 persisted(store, item, {"profitMargin": 10})
134 store.upsert_financial_fact(FinancialFact(FinancialFactKey(UUID(int=1), "profitMargin", None, "ANNUAL"),
135 ProvenancedValue(value=-10, source_url="https://example.test", source_name="official", retrieved_at=NOW),
136 FactSourceTier.OFFICIAL_REGULATORY, "OFFICIAL", "fact"))
137 candidate = (await scan([item], store)).candidates[0]
138 assert candidate.dimensions["PROFITABILITY_QUALITY"].state == "CONFLICTING"
139 assert not candidate.eligible_for_deep_analysis
140 store = SqliteResearchPersistence()
141 persisted(store, item, age=-1)
142 assert "NO_USABLE_PRICE" in (await scan([item], store)).candidates[0].exclusion_reasons
143
144
145 @pytest.mark.asyncio
146 async def test_stale_financial_evidence_cannot_be_revived_by_retrieval():
147 store = SqliteResearchPersistence()
148 item = instrument()
149 persisted(store, item, {"profitMargin": 5}, age=600)
150 record = store.load_structured_market_snapshots({UUID(int=1)})[0]
151 record.retrieved_at = NOW
152 store.upsert_structured_market_snapshot(record)
153 candidate = (await scan([item], store)).candidates[0]
154 assert candidate.dimensions["PROFITABILITY_QUALITY"].state == "STALE"
155 assert "profitMargin" in candidate.stale_inputs
156
157
158 @pytest.mark.asyncio
159 async def test_empty_universe_performs_no_persistence_reads():
160 class NoReads:
161 def __getattr__(self, name): raise AssertionError(name)
162 assert (await scan([], NoReads())).candidates == []
163
164
165 def test_history_counts_distinct_days_and_rejects_currency_mismatch():
166 store = SqliteResearchPersistence()
167 item = instrument()
168 persisted(store, item)
169 prices = store.load_market_price_observations({UUID(int=1)})
170 scorer = GlobalPreScore(history_points=2)
171 candidate = scorer.score(item, [], [], prices*50, as_of=NOW)
172 assert not candidate.technical_history_available
173 prices[0].currency = "USD"
174 assert not scorer.score(item, [], [], prices, as_of=NOW).price_data_available
175
176
177 @pytest.mark.asyncio
178 async def test_equal_prescore_orders_confidence_before_id():
179 store = SqliteResearchPersistence()
180 items = [instrument(1), instrument(2)]
181 persisted(store, items[0], {"profitMargin": 10})
182 persisted(store, items[1], {"profitMargin": 10, "roe": 5, "revenueGrowth": 10})
183 candidates = (await scan(items, store)).candidates
184 assert candidates[0].pre_score == candidates[1].pre_score
185 assert candidates[0].confidence > candidates[1].confidence
186 assert candidates[0].global_instrument_id.int == 2
187
188
189 @pytest.mark.asyncio
190 async def test_scanner_batches_requested_instruments_only():
191 from app.persistence import DisabledResearchPersistence
192 calls = []
193 class Store(DisabledResearchPersistence):
194 def load_financial_facts(self, ids): calls.append(("facts", ids)); return []
195 def load_market_price_observations(self, ids): calls.append(("prices", ids)); return []
196 def load_structured_market_snapshots(self, ids): calls.append(("snapshots", ids)); return []
197 result = await scan([instrument(n) for n in range(1, 502)], Store())
198 assert len(result.candidates) == 501
199 assert [len(ids) for name, ids in calls if name == "facts"] == [250, 250, 1]
200 assert len(calls) == 9
201
202
203 @pytest.mark.asyncio
204 async def test_untrusted_mapping_and_conflicting_price_fail_closed():
205 store = SqliteResearchPersistence()
206 item = instrument()
207 persisted(store, item)
208 price = store.load_market_price_observations({UUID(int=1)})[0]
209 item["providerMappings"].append(dict(provider="OTHER", providerSymbol="ABC", status="VERIFIED"))
210 store.upsert_market_price_observation(price.model_copy(update={"provider": "OTHER", "price": Decimal(200)}))
211 candidate = (await scan([item], store)).candidates[0]
212 assert "CONFLICTING_PRICE" in candidate.exclusion_reasons
213 assert candidate.dimensions["PRICE_DATA_QUALITY"].score is None
214 item["providerMappings"] = [dict(provider="YAHOO_FINANCE", providerSymbol="ABC.NS", status="VERIFIED", resolutionSource="BROKER_IMPORT_IDENTITY")]
215 assert "UNTRUSTED_CANONICAL_IDENTITY" in (await scan([item], store)).candidates[0].exclusion_reasons
216
217
218 @pytest.mark.asyncio
219 async def test_official_persisted_profit_and_equity_support_india_without_structured_financials():
220 store = SqliteResearchPersistence()
221 item = instrument()
222 persisted(store, item, {})
223 for metric in ["pat", "equity"]:
224 store.upsert_financial_fact(FinancialFact(FinancialFactKey(UUID(int=1), metric, "2026-06-30", "QUARTERLY"),
225 ProvenancedValue(value=10, source_url="https://example.test", source_name="NSE", retrieved_at=NOW),
226 FactSourceTier.OFFICIAL_NSE, "NSE", metric))
227 candidate = (await scan([item], store)).candidates[0]
228 assert candidate.eligible_for_deep_analysis
229 assert candidate.dimensions["BALANCE_SHEET_QUALITY"].score == 100
230
231
232 @pytest.mark.asyncio
233 async def test_nonfinite_optional_values_stay_missing():
234 store = SqliteResearchPersistence()
235 item = instrument()
236 persisted(store, item, {"profitMargin": 10, "revenueGrowth": "NaN", "earningsGrowth": "Infinity"})
237 candidate = (await scan([item], store)).candidates[0]
238 assert candidate.dimensions["GROWTH_QUALITY"].state == "MISSING"
239 assert candidate.dimensions["GROWTH_QUALITY"].score is None
240
241
242 @pytest.mark.asyncio
243 async def test_different_financial_period_types_are_not_false_conflicts():
244 store = SqliteResearchPersistence()
245 item = instrument()
246 persisted(store, item, {})
247 for period_type, value in [("ANNUAL", 100), ("QUARTERLY", 25)]:
248 store.upsert_financial_fact(FinancialFact(FinancialFactKey(UUID(int=1), "pat", "2026-03-31", period_type),
249 ProvenancedValue(value=value, source_url="https://example.test", source_name="official", retrieved_at=NOW),
250 FactSourceTier.OFFICIAL_REGULATORY, "OFFICIAL", period_type))
251 candidate = (await scan([item], store)).candidates[0]
252 assert candidate.eligible_for_deep_analysis
253 assert candidate.dimensions["PROFITABILITY_QUALITY"].state == "PARTIAL"