@prakhar-gitlab / ai-investment-platform / commits / 8597516

feat: add global equity scanner foundation

prakhar82 committed Sep 13, 2026 at 17:21 UTC 85975161f2d1fa9ec8f02616ce34787c135b527e
7 files changed +707 -40
ai/research-engine/app/global_scanner.py new
+309
@@ -0,0 +1,309 @@
1 +"""Provider-free candidate selection. This is not an investment score or V1 analysis.
2 +
3 +Scores use equal-weight available dimensions, renormalized over evidence only.
4 +Confidence measures coverage/readiness, independently of positive/negative values.
5 +The explicit as_of clock is part of the reproducible input state.
6 +"""
7 +from __future__ import annotations
8 +
9 +from collections import Counter, defaultdict
10 +from datetime import datetime, timedelta, timezone
11 +from decimal import Decimal, InvalidOperation
12 +from enum import StrEnum
13 +from typing import Protocol
14 +from uuid import UUID
15 +
16 +import httpx
17 +from app.models import ResearchBaseModel
18 +from app.fact_precedence import SUPPORTED_FINANCIAL_SOURCE_TIERS
19 +from app.persistence import ResearchPersistence
20 +
21 +
22 +class EvidenceState(StrEnum):
23 + AVAILABLE = "AVAILABLE"
24 + PARTIAL = "PARTIAL"
25 + STALE = "STALE"
26 + MISSING = "MISSING"
27 + CONFLICTING = "CONFLICTING"
28 + NOT_APPLICABLE = "NOT_APPLICABLE"
29 +
30 +
31 +class PreScoreDimension(ResearchBaseModel):
32 + state: EvidenceState
33 + score: float | None = None
34 + coverage: float = 0
35 +
36 +
37 +class GlobalScanCandidate(ResearchBaseModel):
38 + global_instrument_id: UUID
39 + market: str | None = None
40 + region: str | None = None
41 + country: str | None = None
42 + exchange: str | None = None
43 + currency: str | None = None
44 + asset_type: str | None = None
45 + status: str | None = None
46 + symbol: str | None = None
47 + company_name: str | None = None
48 + verified_provider_mapping_status: str
49 + pre_score: float | None
50 + confidence: float
51 + critical_completeness: float
52 + price_data_available: bool
53 + financial_data_available: bool
54 + technical_history_available: bool
55 + sector_data_available: bool
56 + dimensions: dict[str, PreScoreDimension]
57 + missing_inputs: list[str]
58 + stale_inputs: list[str]
59 + eligible_for_deep_analysis: bool
60 + exclusion_reasons: list[str]
61 +
62 +
63 +class GlobalScanResult(ResearchBaseModel):
64 + as_of: datetime
65 + pre_score_version: str = "GLOBAL_PRE_SCORE_V1"
66 + total_canonical_active_equities: int
67 + eligible_candidates: int
68 + excluded_candidates_by_reason: dict[str, int]
69 + deep_analysis_eligible_count: int
70 + candidates: list[GlobalScanCandidate]
71 + top_candidates: list[GlobalScanCandidate]
72 + deep_analysis_candidate_ids: list[UUID]
73 +
74 +
75 +class EquityUniverse(Protocol):
76 + async def active_global_equities(self, **kwargs) -> list[dict]: ...
77 +
78 +
79 +class CanonicalEquityUniverse:
80 + """Read canonical metadata only; no portfolio ownership or provider acquisition."""
81 +
82 + def __init__(self, client: httpx.AsyncClient, base_url: str):
83 + self.client, self.base_url = client, base_url.rstrip("/")
84 +
85 + async def active_global_equities(self, *, correlation_id=None, identity_headers=None):
86 + headers = {k: v for k, v in (identity_headers or {}).items() if v}
87 + if correlation_id:
88 + headers["X-Correlation-Id"] = correlation_id
89 + values, page = [], 0
90 + while True:
91 + response = await self.client.get(
92 + f"{self.base_url}/api/v1/instruments",
93 + params={"status": "ACTIVE", "assetType": "EQUITY", "page": page, "size": 500},
94 + headers=headers or None,
95 + )
96 + response.raise_for_status()
97 + payload = response.json()
98 + if not isinstance(payload, dict) or not isinstance(payload.get("instruments"), list):
99 + raise ValueError("Invalid canonical universe response")
100 + batch = payload["instruments"]
101 + total = int(payload.get("totalElements", len(values) + len(batch)))
102 + values.extend(batch)
103 + if len(values) >= total:
104 + return values
105 + if not batch:
106 + raise ValueError("Incomplete canonical universe pagination")
107 + page += 1
108 +
109 +
110 +def _utc(value):
111 + return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc)
112 +
113 +
114 +def _number(value):
115 + if value is None or isinstance(value, bool):
116 + return None
117 + try:
118 + number = Decimal(str(value))
119 + return number if number.is_finite() else None
120 + except (InvalidOperation, ValueError):
121 + return None
122 +
123 +
124 +class GlobalPreScore:
125 + """Cheap readiness and sign heuristics; no valuation targets or V1 formulas.
126 +
127 + Critical inputs are a fresh trusted price and at least one fresh profitability
128 + metric. Optional dimensions never supply a synthetic zero. Quality metrics
129 + use only sign (positive=100, actual zero=50, negative=0), avoiding unit-dependent
130 + thresholds. No financial ratios or technical indicators are derived here.
131 + """
132 +
133 + def __init__(self, *, price_max_age=timedelta(days=7), financial_max_age=timedelta(days=550), history_points=50):
134 + if price_max_age.total_seconds() <= 0 or financial_max_age.total_seconds() <= 0 or history_points < 2:
135 + raise ValueError("Readiness limits must be positive; history_points must be >= 2")
136 + self.price_max_age = price_max_age
137 + self.financial_max_age = financial_max_age
138 + self.history_points = history_points
139 +
140 + def score(self, instrument, snapshots, facts, prices, *, as_of):
141 + as_of = _utc(as_of)
142 + missing, stale = set(), set()
143 + evidence = defaultdict(list)
144 + currency = instrument.get("currency")
145 + mappings = [m for m in instrument.get("providerMappings", []) if isinstance(m, dict)
146 + and m.get("status") == "VERIFIED" and m.get("provider")
147 + and (m.get("providerInstrumentId") or m.get("providerSymbol"))
148 + and m.get("resolutionSource") != "BROKER_IMPORT_IDENTITY"]
149 + providers = {m["provider"] for m in mappings}
150 + for record in snapshots:
151 + mapped_ids = {str(m.get(key)) for m in mappings if m["provider"] == record.provider
152 + for key in ("providerInstrumentId", "providerSymbol") if m.get(key)}
153 + if (record.provider not in providers or _utc(record.retrieved_at) > as_of
154 + or record.provider_instrument_id not in mapped_ids or record.currency != currency):
155 + continue
156 + for key, value in record.snapshot.facts.items():
157 + if _utc(value.retrieved_at) > as_of:
158 + continue
159 + stamp = value.as_of_date or value.published_at or record.market_as_of or record.retrieved_at
160 + if _utc(stamp) <= as_of:
161 + evidence[key].append((value.value, _utc(stamp)))
162 + financial_evidence = defaultdict(list)
163 + for fact in facts:
164 + if fact.source_tier not in SUPPORTED_FINANCIAL_SOURCE_TIERS or str(fact.source_mode) != "REAL" or _utc(fact.value.retrieved_at) > as_of:
165 + continue
166 + stamp = fact.value.as_of_date or fact.value.published_at or fact.value.retrieved_at
167 + if fact.key.period_end:
168 + try:
169 + stamp = datetime.fromisoformat(fact.key.period_end)
170 + except ValueError:
171 + continue
172 + if _utc(stamp) <= as_of:
173 + metric = {"profit_margin": "profitMargin", "return_on_equity": "roe",
174 + "net_income": "pat", "net_profit": "pat", "revenue_growth": "revenueGrowth",
175 + "earnings_growth": "earningsGrowth", "equity": "total_equity"}.get(fact.key.metric, fact.key.metric)
176 + # Annual and quarterly amounts (or standalone and consolidated
177 + # statements) are not conflicting observations of one metric.
178 + basis_rank = {"CONSOLIDATED": 2, "UNKNOWN": 1}.get(fact.key.reporting_basis, 0)
179 + period_rank = {"ANNUAL": 2, "QUARTERLY": 1}.get(fact.key.period_type, 0)
180 + financial_evidence[metric].append((fact.value.value, _utc(stamp), basis_rank, period_rank))
181 + for metric, values in financial_evidence.items():
182 + selected = max((stamp, basis, period) for _, stamp, basis, period in values)
183 + evidence[metric].extend((value, stamp) for value, stamp, basis, period in values
184 + if (stamp, basis, period) == selected)
185 +
186 + def dimension(keys, *, quality=False, text=False, max_age=None):
187 + scores, ready, conflicts, old = [], 0, False, False
188 + for key in keys:
189 + values = evidence.get(key, [])
190 + valid = [(str(v).strip() if text and v is not None else _number(v), t) for v, t in values]
191 + valid = [(v, t) for v, t in valid if v is not None and v != ""]
192 + if not valid:
193 + missing.add(key)
194 + continue
195 + latest = max(t for _, t in valid)
196 + current = {v for v, t in valid if t == latest}
197 + if len(current) > 1:
198 + conflicts = True
199 + continue
200 + value = next(iter(current))
201 + is_stale = as_of - latest > (max_age or self.financial_max_age)
202 + if is_stale:
203 + stale.add(key)
204 + old = True
205 + else:
206 + ready += 1
207 + scores.append((100 if value > 0 else 50 if value == 0 else 0) if quality else 100)
208 + state = (EvidenceState.CONFLICTING if conflicts else EvidenceState.STALE if old else
209 + EvidenceState.MISSING if not scores else EvidenceState.PARTIAL if len(scores) < len(keys)
210 + else EvidenceState.AVAILABLE)
211 + return PreScoreDimension(state=state, score=sum(scores)/len(scores) if scores and not conflicts else None,
212 + coverage=ready/len(keys) if not conflicts else 0)
213 +
214 + usable = [p for p in prices if p.provider in providers and _number(p.price) is not None and p.price > 0
215 + and currency and p.currency == currency and _utc(p.observed_at) <= as_of and _utc(p.retrieved_at) <= as_of]
216 + latest = max((_utc(p.observed_at) for p in usable), default=None)
217 + price_conflict = len({p.price for p in usable if _utc(p.observed_at) == latest}) > 1
218 + fresh_price = latest is not None and as_of - latest <= self.price_max_age and not price_conflict
219 + price_state = (EvidenceState.CONFLICTING if price_conflict else EvidenceState.MISSING if latest is None
220 + else EvidenceState.AVAILABLE if fresh_price else EvidenceState.STALE)
221 + if latest is None:
222 + missing.add("price")
223 + elif not fresh_price and not price_conflict:
224 + stale.add("price")
225 + days = len({_utc(p.observed_at).date() for p in usable})
226 + dimensions = {
227 + "PRICE_DATA_QUALITY": PreScoreDimension(state=price_state, score=100 if usable and not price_conflict else None, coverage=float(fresh_price)),
228 + "VALUATION_AVAILABILITY": dimension(["trailingPE", "priceToBook"], max_age=self.price_max_age),
229 + "PROFITABILITY_QUALITY": dimension(["profitMargin", "roe", "pat"], quality=True),
230 + "GROWTH_QUALITY": dimension(["revenueGrowth", "earningsGrowth"], quality=True),
231 + "BALANCE_SHEET_QUALITY": dimension(["total_equity"], quality=True),
232 + "TECHNICAL_DATA_READINESS": PreScoreDimension(state=EvidenceState.MISSING if not days else EvidenceState.STALE if not fresh_price else EvidenceState.AVAILABLE if days >= self.history_points else EvidenceState.PARTIAL,
233 + score=100 if days >= self.history_points else None, coverage=min(1, days/self.history_points) if fresh_price else 0),
234 + "SECTOR_DATA_READINESS": dimension(["sector"], text=True),
235 + }
236 + if days < self.history_points:
237 + missing.add("technicalHistory")
238 + profitability = dimensions["PROFITABILITY_QUALITY"]
239 + financial_available = profitability.score is not None
240 + financial_ready = profitability.coverage > 0 and profitability.state in {"AVAILABLE", "PARTIAL"}
241 + critical = (int(fresh_price) + int(financial_ready)) / 2
242 + readiness = sum(d.coverage for d in dimensions.values()) / len(dimensions)
243 + dimensions["FRESHNESS"] = PreScoreDimension(state=EvidenceState.STALE if stale else EvidenceState.AVAILABLE,
244 + score=None, coverage=readiness)
245 + dimensions["CRITICAL_COMPLETENESS"] = PreScoreDimension(state=EvidenceState.AVAILABLE if critical == 1 else EvidenceState.PARTIAL if critical else EvidenceState.MISSING,
246 + score=None, coverage=critical)
247 + symbol = instrument.get("ticker") or instrument.get("primarySymbol")
248 + name = instrument.get("canonicalName")
249 + reasons = []
250 + if instrument.get("status") != "ACTIVE": reasons.append("INACTIVE")
251 + if instrument.get("assetType") != "EQUITY": reasons.append("NON_EQUITY")
252 + if not (mappings and symbol and name and instrument.get("exchange") and currency): reasons.append("UNTRUSTED_CANONICAL_IDENTITY")
253 + if not usable: reasons.append("NO_USABLE_PRICE")
254 + elif price_conflict: reasons.append("CONFLICTING_PRICE")
255 + elif not fresh_price: reasons.append("STALE_PRICE")
256 + if not financial_ready: reasons.append("CRITICAL_FINANCIAL_EVIDENCE_UNREADY")
257 + scores = [d.score for d in dimensions.values() if d.score is not None and d.state in {"AVAILABLE", "PARTIAL"}]
258 + return GlobalScanCandidate(global_instrument_id=instrument["globalInstrumentId"], market=instrument.get("exchange"),
259 + region=instrument.get("region"), country=instrument.get("country"), exchange=instrument.get("exchange"),
260 + currency=currency, asset_type=instrument.get("assetType"), status=instrument.get("status"), symbol=symbol, company_name=name,
261 + verified_provider_mapping_status="VERIFIED" if mappings else "MISSING", pre_score=round(sum(scores)/len(scores), 6) if scores else None,
262 + confidence=round(100*readiness*critical, 6), critical_completeness=critical*100,
263 + price_data_available=bool(usable) and not price_conflict, financial_data_available=financial_available,
264 + technical_history_available=days >= self.history_points, sector_data_available=dimensions["SECTOR_DATA_READINESS"].score is not None,
265 + dimensions=dimensions, missing_inputs=sorted(missing), stale_inputs=sorted(stale),
266 + eligible_for_deep_analysis=not reasons, exclusion_reasons=reasons)
267 +
268 +
269 +class GlobalScanner:
270 + def __init__(self, universe: EquityUniverse, persistence: ResearchPersistence, *, pre_score=None, batch_size=250):
271 + if not 1 <= batch_size <= 500:
272 + raise ValueError("batch_size must be between 1 and 500")
273 + self.universe, self.persistence = universe, persistence
274 + self.pre_score, self.batch_size = pre_score or GlobalPreScore(), batch_size
275 +
276 + async def scan(self, *, as_of: datetime, top_n: int, **universe_context) -> GlobalScanResult:
277 + if top_n < 0:
278 + raise ValueError("top_n must be nonnegative")
279 + instruments = await self.universe.active_global_equities(**universe_context)
280 + by_id = {}
281 + invalid = 0
282 + for item in instruments:
283 + try:
284 + key = UUID(str(item.get("globalInstrumentId")))
285 + except (ValueError, TypeError):
286 + invalid += 1
287 + continue
288 + if key in by_id and by_id[key] != item:
289 + raise ValueError(f"Conflicting canonical records for {key}")
290 + by_id[key] = item
291 + candidates = []
292 + ordered = sorted(by_id, key=str)
293 + for offset in range(0, len(ordered), self.batch_size):
294 + ids = set(ordered[offset:offset+self.batch_size])
295 + snapshots, facts, prices = defaultdict(list), defaultdict(list), defaultdict(list)
296 + for row in self.persistence.load_structured_market_snapshots(ids): snapshots[row.instrument_id].append(row)
297 + for row in self.persistence.load_financial_facts(ids): facts[row.key.instrument_id].append(row)
298 + for row in self.persistence.load_market_price_observations(ids): prices[row.instrument_id].append(row)
299 + for key in sorted(ids, key=str):
300 + candidates.append(self.pre_score.score(by_id[key], snapshots[key], facts[key], prices[key], as_of=as_of))
301 + candidates.sort(key=lambda c: (-(c.pre_score if c.pre_score is not None else -1), -c.confidence, -c.critical_completeness, str(c.global_instrument_id)))
302 + eligible = [c for c in candidates if c.eligible_for_deep_analysis]
303 + excluded = Counter(reason for c in candidates for reason in c.exclusion_reasons)
304 + if invalid: excluded["UNTRUSTED_CANONICAL_IDENTITY"] += invalid
305 + top = eligible[:top_n]
306 + return GlobalScanResult(as_of=_utc(as_of), total_canonical_active_equities=sum(i.get("status") == "ACTIVE" and i.get("assetType") == "EQUITY" for i in by_id.values()),
307 + eligible_candidates=len(eligible), excluded_candidates_by_reason=dict(sorted(excluded.items())),
308 + deep_analysis_eligible_count=len(eligible), candidates=candidates, top_candidates=top,
309 + deep_analysis_candidate_ids=[c.global_instrument_id for c in top])
ai/research-engine/app/persistence.py
+35 -16
@@ -49,7 +49,7 @@ class ResearchPersistence(Protocol):
49 def load_documents(self) -> list[ResearchDocument]:
50 ...
51
52 - def load_events(self) -> list[ResearchEvent]:
52 + def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]:
53 ...
54
55 def upsert_document(self, document: ResearchDocument) -> bool:
@@ -67,7 +67,7 @@ class ResearchPersistence(Protocol):
67 def upsert_event(self, event: ResearchEvent) -> bool:
68 ...
69
70 - def load_shareholding_snapshots(self) -> list[ShareholdingSnapshot]:
70 + def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]:
71 ...
72
73 def upsert_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool:
@@ -99,7 +99,7 @@ class ResearchPersistence(Protocol):
99 ) -> None:
100 ...
101
102 - def load_financial_facts(self) -> list[FinancialFact]: ...
102 + def load_financial_facts(self, instrument_ids: set[UUID] | None = None) -> list[FinancialFact]: ...
103 def upsert_financial_fact(self, fact: FinancialFact, *, allow_same_tier_correction: bool = False) -> bool: ...
104 def load_structured_market_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[StructuredMarketSnapshotRecord]: ...
105 def upsert_structured_market_snapshot(self, record: StructuredMarketSnapshotRecord) -> None: ...
@@ -121,7 +121,7 @@ class DisabledResearchPersistence:
121 def load_documents(self) -> list[ResearchDocument]:
122 return []
123
124 - def load_events(self) -> list[ResearchEvent]:
124 + def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]:
125 return []
126
127 def upsert_document(self, document: ResearchDocument) -> bool:
@@ -130,7 +130,7 @@ class DisabledResearchPersistence:
130 def upsert_event(self, event: ResearchEvent) -> bool:
131 return True
132
133 - def load_shareholding_snapshots(self) -> list[ShareholdingSnapshot]:
133 + def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]:
134 return []
135
136 def upsert_shareholding_snapshot(self, snapshot: ShareholdingSnapshot) -> bool:
@@ -149,7 +149,7 @@ class DisabledResearchPersistence:
149 def complete_refresh_run(self, run: RefreshRun, **kwargs) -> None:
150 return None
151
152 - def load_financial_facts(self): return []
152 + def load_financial_facts(self, instrument_ids=None): return []
153 def upsert_financial_fact(self, fact, **kwargs): return True
154 def load_structured_market_snapshots(self, instrument_ids=None): return []
155 def upsert_structured_market_snapshot(self, record): return None
@@ -186,8 +186,25 @@ class SqliteResearchPersistence:
186 self._connection.executescript(_sqlite_schema())
187 self._connection.commit()
188
189 - def load_financial_facts(self) -> list[FinancialFact]:
190 - return [_financial_fact_from_row(row) for row in self._connection.execute("SELECT * FROM global_financial_facts").fetchall()]
189 + def _filtered_rows(self, table, ids, *, column="instrument_id", order=None):
190 + """Internal identifiers only; values are parameterized in bounded batches."""
191 + if ids is not None:
192 + ordered = sorted({str(value) for value in ids})
193 + rows = []
194 + for offset in range(0, len(ordered), 500):
195 + batch = ordered[offset:offset + 500]
196 + sql = f"SELECT * FROM {table} WHERE {column} IN ({','.join('?' for _ in batch)})"
197 + if order:
198 + sql += f" ORDER BY {order}"
199 + rows.extend(self._connection.execute(sql, batch).fetchall())
200 + return rows
201 + sql = f"SELECT * FROM {table}"
202 + if order:
203 + sql += f" ORDER BY {order}"
204 + return self._connection.execute(sql).fetchall()
205 +
206 + def load_financial_facts(self, instrument_ids: set[UUID] | None = None) -> list[FinancialFact]:
207 + return [_financial_fact_from_row(row) for row in self._filtered_rows("global_financial_facts", instrument_ids)]
208
209 def upsert_financial_fact(self, fact: FinancialFact, *, allow_same_tier_correction: bool = False) -> bool:
210 key = fact.key
@@ -258,6 +275,8 @@ class SqliteResearchPersistence:
275 (str(key.instrument_id), key.metric, key.period_end or "", key.period_type, key.reporting_basis or "", str(fact.value.value), fact.value.unit, fact.source_provider, fact.source_identity, fact.value.source_url, fact.value.source_name, fact.value.source_type, _dt(fact.value.published_at), _dt(fact.value.retrieved_at), fact.value.confidence, str(fact.source_mode), int(fact.source_tier)))
276
277 def load_structured_market_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[StructuredMarketSnapshotRecord]:
278 + if instrument_ids is not None and not instrument_ids:
279 + return []
280 params: list[str] = []
281 sql = "SELECT * FROM global_structured_market_snapshots"
282 if instrument_ids:
@@ -292,6 +311,8 @@ class SqliteResearchPersistence:
311 observation.currency, observation.provider, observation.source_url, _dt(observation.retrieved_at)))
312
313 def load_market_price_observations(self, instrument_ids: set[UUID] | None = None) -> list[MarketPriceObservation]:
314 + if instrument_ids is not None and not instrument_ids:
315 + return []
316 params: list[str] = []
317 sql = "SELECT * FROM global_market_price_observations"
318 if instrument_ids:
@@ -405,11 +426,11 @@ class SqliteResearchPersistence:
426 rows = self._connection.execute("SELECT * FROM research_documents ORDER BY retrieved_at").fetchall()
427 return [_document_from_row(row) for row in rows]
428
408 - def load_events(self) -> list[ResearchEvent]:
409 - rows = self._connection.execute("SELECT * FROM research_events ORDER BY detected_at").fetchall()
429 + def load_events(self, instrument_ids: set[UUID] | None = None) -> list[ResearchEvent]:
430 + rows = self._filtered_rows("research_events", instrument_ids, order="detected_at")
431 events = [_event_from_row(row) for row in rows]
432 sources_by_event: dict[UUID, list[ResearchEvidenceSource]] = {}
412 - for row in self._connection.execute("SELECT * FROM research_event_sources ORDER BY created_at").fetchall():
433 + for row in self._filtered_rows("research_event_sources", {row["event_id"] for row in rows}, column="event_id", order="created_at"):
434 event_id = _parse_uuid(row["event_id"])
435 if event_id is None:
436 continue
@@ -418,12 +439,10 @@ class SqliteResearchPersistence:
439 event.supporting_sources = sources_by_event.get(event.event_id, [])
440 return events
441
421 - def load_shareholding_snapshots(self) -> list[ShareholdingSnapshot]:
422 - rows = self._connection.execute(
423 - "SELECT * FROM global_shareholding_snapshots ORDER BY period_end DESC, retrieved_at DESC"
424 - ).fetchall()
442 + def load_shareholding_snapshots(self, instrument_ids: set[UUID] | None = None) -> list[ShareholdingSnapshot]:
443 + rows = self._filtered_rows("global_shareholding_snapshots", instrument_ids, order="period_end DESC, retrieved_at DESC")
444 values_by_snapshot: dict[UUID, list[ShareholdingSnapshotValue]] = {}
426 - for row in self._connection.execute("SELECT * FROM global_shareholding_snapshot_values ORDER BY created_at").fetchall():
445 + for row in self._filtered_rows("global_shareholding_snapshot_values", {row["id"] for row in rows}, column="snapshot_id", order="created_at"):
446 snapshot_id = _required_uuid(row["snapshot_id"], "global_shareholding_snapshot_values.snapshot_id")
447 values_by_snapshot.setdefault(snapshot_id, []).append(ShareholdingSnapshotValue(
448 id=_required_uuid(row["id"], "global_shareholding_snapshot_values.id"), category=ShareholdingCategory(row["category"]),
ai/research-engine/app/portfolio_orchestration.py
+6 -17
@@ -82,24 +82,13 @@ class PortfolioResearchOrchestrator:
82 identity_headers: dict[str, str | None] | None = None,
83 ) -> list[dict]:
84 """Read the portfolio-service owned active-equity universe; never mutate it."""
85 + from app.global_scanner import CanonicalEquityUniverse
86 +
87 try:
86 - page, values, total = 0, [], None
87 - headers = {key: value for key, value in (identity_headers or {}).items() if value}
88 - if correlation_id:
89 - headers["X-Correlation-Id"] = correlation_id
90 - while total is None or len(values) < total:
91 - response = await self._client.get(
92 - f"{self.settings.portfolio_service_base_url}/api/v1/instruments",
93 - params={"status": "ACTIVE", "assetType": "EQUITY", "page": page, "size": 500},
94 - headers=headers or None,
95 - )
96 - response.raise_for_status(); payload = response.json()
97 - batch = payload.get("instruments", []) if isinstance(payload, dict) else []
98 - values.extend(batch); total = payload.get("totalElements", len(values)) if isinstance(payload, dict) else len(values)
99 - if not batch: break
100 - page += 1
101 - return values
102 - except httpx.HTTPError as exc:
88 + return await CanonicalEquityUniverse(self._client, self.settings.portfolio_service_base_url).active_global_equities(
89 + correlation_id=correlation_id, identity_headers=identity_headers,
90 + )
91 + except (httpx.HTTPError, ValueError) as exc:
92 raise PortfolioServiceUnavailableError("Portfolio service unavailable for instrument enumeration") from exc
93
94 async def india_nifty500_universe(self, *, correlation_id: str | None = None, identity_headers: dict[str, str | None] | None = None) -> list[dict]:
ai/research-engine/app/repository.py
+2 -2
@@ -206,11 +206,11 @@ class ResearchRepository:
206 return self.etf_profiles
207
208 def financial_facts_for(self, instrument_id: UUID):
209 - return [fact for fact in self._persistence.load_financial_facts() if fact.key.instrument_id == instrument_id]
209 + return self._persistence.load_financial_facts({instrument_id})
210
211 async def financial_facts_for_instruments(self, instrument_ids: set[UUID]) -> dict[UUID, list[FinancialFact]]:
212 started = time.perf_counter()
213 - facts = await self._run_blocking_persistence(self._persistence.load_financial_facts)
213 + facts = await self._run_blocking_persistence(self._persistence.load_financial_facts, instrument_ids)
214 grouped: dict[UUID, list[FinancialFact]] = {instrument_id: [] for instrument_id in instrument_ids}
215 for fact in facts:
216 if fact.key.instrument_id in grouped:
ai/research-engine/tests/test_global_scanner.py new
+253
@@ -0,0 +1,253 @@
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"
ai/research-engine/tests/test_official_nse_financial_parsing.py
+10 -5
@@ -793,7 +793,8 @@ def test_direct_aligned_diluted_eps_outranks_malformed_fallback_when_basic_is_un
793
794 class _FactRecorder:
795 def __init__(self): self.facts = []
796 - def load_financial_facts(self): return self.facts
796 + def load_financial_facts(self, instrument_ids=None):
797 + return [fact for fact in self.facts if instrument_ids is None or fact.key.instrument_id in instrument_ids]
798 def upsert_financial_fact(self, fact, **_kwargs): self.facts.append(fact); return True
799
800
@@ -1069,11 +1070,15 @@ def test_reused_trusted_nse_document_reconciles_same_document_facts_without_fetc
1070
1071 asyncio.run(repository._fetch_official_filings(profile, [DiscoveryResult("FINANCIAL_RESULTS", source)], set()))
1072 facts = repository.financial_facts_for(profile.instrument_id)
1072 - eps = next(fact for fact in facts if fact.key.metric == "eps")
1073 + # Filtered SQL may use a different index; assert the reconciled period,
1074 + # independently of database row order and comparative-period facts.
1075 + current = {fact.key.metric: fact for fact in facts
1076 + if fact.key.period_end == "2026-06-30" and fact.key.period_type == "QUARTERLY"}
1077 + eps = current["eps"]
1078 assert eps.value.value == Decimal("1.47")
1079 assert eps.value.unit == "INR per share"
1075 - assert next(fact for fact in facts if fact.key.metric == "revenue").value.value == Decimal("8261.11")
1076 - assert next(fact for fact in facts if fact.key.metric == "pat").value.value == Decimal("1927.21")
1080 + assert current["revenue"].value.value == Decimal("8261.11")
1081 + assert current["pat"].value.value == Decimal("1927.21")
1082 assert len(repository.events) == events_before
1083 assert len(repository.documents) == documents_before
1084
@@ -1081,7 +1086,7 @@ def test_reused_trusted_nse_document_reconciles_same_document_facts_without_fetc
1086 assert len(repository.financial_facts_for(profile.instrument_id)) == len(facts)
1087 yahoo = FinancialFact(eps.key, ProvenancedValue(value=Decimal("9"), source_url="https://yahoo.test", source_name="Yahoo", source_type="YAHOO", retrieved_at=document.retrieved_at), FactSourceTier.YAHOO, "YAHOO_FINANCE", "yahoo", SourceMode.REAL)
1088 assert not repository._persistence.upsert_financial_fact(yahoo)
1084 - assert next(fact for fact in repository.financial_facts_for(profile.instrument_id) if fact.key.metric == "eps").value.value == Decimal("1.47")
1089 + assert next(fact for fact in repository.financial_facts_for(profile.instrument_id) if fact.key == eps.key).value.value == Decimal("1.47")
1090
1091
1092 def test_already_persisted_official_document_recovers_zero_facts_without_redownload() -> None:
ai/research-engine/tests/test_scanner_batch_persistence.py new
+92
@@ -0,0 +1,92 @@
1 +from dataclasses import replace
2 +from uuid import UUID
3 +
4 +import pytest
5 +
6 +from app.persistence import SqliteResearchPersistence, DisabledResearchPersistence
7 +from app.postgres_persistence import PostgresResearchPersistence, _PostgresConnectionAdapter
8 +from app.repository import ResearchRepository
9 +from test_fact_precedence import fact
10 +from app.fact_precedence import FactSourceTier
11 +from test_shareholding import _snapshot
12 +from test_structured_market_persistence import _record
13 +
14 +
15 +@pytest.mark.parametrize("method", ["load_financial_facts", "load_events", "load_shareholding_snapshots",
16 + "load_structured_market_snapshots", "load_market_price_observations"])
17 +def test_empty_filter_never_reads_all_rows(method):
18 + store = SqliteResearchPersistence()
19 + queries = []
20 + store._connection.set_trace_callback(queries.append)
21 + assert getattr(store, method)(set()) == []
22 + assert queries == []
23 + assert getattr(DisabledResearchPersistence(), method)(set()) == []
24 +
25 +
26 +def test_filtered_facts_snapshots_prices_and_shareholding_values():
27 + store = SqliteResearchPersistence()
28 + for n in [1, 2]:
29 + key = UUID(int=n)
30 + value = fact("pat", n, FactSourceTier.OFFICIAL_NSE)
31 + store.upsert_financial_fact(replace(value, key=replace(value.key, instrument_id=key)))
32 + store.upsert_shareholding_snapshot(_snapshot(key, source=f"source-{n}"))
33 + store.upsert_structured_market_snapshot(_record(key))
34 + queries = []
35 + store._connection.set_trace_callback(queries.append)
36 + ids = {UUID(int=1)}
37 + assert [f.key.instrument_id for f in store.load_financial_facts(ids)] == list(ids)
38 + snapshots = store.load_shareholding_snapshots(ids)
39 + assert [s.instrument_id for s in snapshots] == list(ids)
40 + assert len(snapshots[0].values) == 1
41 + assert [s.instrument_id for s in store.load_structured_market_snapshots(ids)] == list(ids)
42 + assert [p.instrument_id for p in store.load_market_price_observations(ids)] == list(ids)
43 + assert all(" WHERE " in q and " IN (" in q for q in queries)
44 + assert len(store.load_financial_facts()) == 2
45 +
46 +
47 +@pytest.mark.asyncio
48 +async def test_repository_financial_reads_pass_ids_to_persistence():
49 + class FilterRequired(DisabledResearchPersistence):
50 + def load_financial_facts(self, instrument_ids=None):
51 + assert instrument_ids == {UUID(int=1)}
52 + return []
53 + repository = ResearchRepository.__new__(ResearchRepository)
54 + repository._persistence = FilterRequired()
55 + import threading
56 + repository._persistence_worker_lock = threading.RLock()
57 + assert repository.financial_facts_for(UUID(int=1)) == []
58 + assert await repository.financial_facts_for_instruments({UUID(int=1)}) == {UUID(int=1): []}
59 +
60 +
61 +def test_postgres_inherits_parameterized_bounded_batch_reads():
62 + queries = []
63 + class Connection:
64 + def execute(self, sql, params):
65 + queries.append((sql, params))
66 + return self
67 + def fetchall(self): return []
68 + store = PostgresResearchPersistence.__new__(PostgresResearchPersistence)
69 + store._connection = _PostgresConnectionAdapter(Connection())
70 + ids = {UUID(int=n) for n in range(1, 1002)}
71 + assert store.load_financial_facts(ids) == []
72 + assert [len(params) for _, params in queries] == [500, 500, 1]
73 + assert all("WHERE instrument_id IN (%s" in sql and "?" not in sql for sql, _ in queries)
74 + assert set(p for _, params in queries for p in params) == {str(i) for i in ids}
75 +
76 +
77 +def test_event_and_supporting_source_reads_are_filtered():
78 + # Existing domain fixtures supply valid source documents and events.
79 + repository = ResearchRepository(persistence=DisabledResearchPersistence())
80 + store = SqliteResearchPersistence()
81 + events = list(repository.events.values())
82 + assert events
83 + for document in repository.documents.values(): store.upsert_document(document)
84 + for event in events: store.upsert_event(event)
85 + target = events[0].instrument_id
86 + queries = []
87 + store._connection.set_trace_callback(queries.append)
88 + loaded = store.load_events({target})
89 + assert loaded and {e.instrument_id for e in loaded} == {target}
90 + assert len(loaded) == sum(e.instrument_id == target for e in events)
91 + assert all(" WHERE " in q and " IN (" in q for q in queries)
92 + assert any("research_event_sources" in q and "event_id IN" in q for q in queries)