| 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 | from app.technical_features import TechnicalFeatureEngine, TechnicalFeatureSnapshot |
| 21 | from app.sector_relative_strength import SectorContext, SectorRelativeStrengthEngine, SectorRelativeStrengthSnapshot |
| 22 | |
| 23 | |
| 24 | class EvidenceState(StrEnum): |
| 25 | AVAILABLE = "AVAILABLE" |
| 26 | PARTIAL = "PARTIAL" |
| 27 | STALE = "STALE" |
| 28 | MISSING = "MISSING" |
| 29 | CONFLICTING = "CONFLICTING" |
| 30 | NOT_APPLICABLE = "NOT_APPLICABLE" |
| 31 | |
| 32 | |
| 33 | class PreScoreDimension(ResearchBaseModel): |
| 34 | state: EvidenceState |
| 35 | score: float | None = None |
| 36 | coverage: float = 0 |
| 37 | |
| 38 | |
| 39 | class GlobalScanCandidate(ResearchBaseModel): |
| 40 | global_instrument_id: UUID |
| 41 | market: str | None = None |
| 42 | region: str | None = None |
| 43 | country: str | None = None |
| 44 | exchange: str | None = None |
| 45 | currency: str | None = None |
| 46 | asset_type: str | None = None |
| 47 | status: str | None = None |
| 48 | symbol: str | None = None |
| 49 | company_name: str | None = None |
| 50 | verified_provider_mapping_status: str |
| 51 | pre_score: float | None |
| 52 | confidence: float |
| 53 | critical_completeness: float |
| 54 | price_data_available: bool |
| 55 | financial_data_available: bool |
| 56 | technical_history_available: bool |
| 57 | sector_data_available: bool |
| 58 | dimensions: dict[str, PreScoreDimension] |
| 59 | missing_inputs: list[str] |
| 60 | stale_inputs: list[str] |
| 61 | eligible_for_deep_analysis: bool |
| 62 | exclusion_reasons: list[str] |
| 63 | |
| 64 | |
| 65 | class GlobalScanResult(ResearchBaseModel): |
| 66 | as_of: datetime |
| 67 | pre_score_version: str = "GLOBAL_PRE_SCORE_V1" |
| 68 | total_canonical_active_equities: int |
| 69 | eligible_candidates: int |
| 70 | excluded_candidates_by_reason: dict[str, int] |
| 71 | deep_analysis_eligible_count: int |
| 72 | candidates: list[GlobalScanCandidate] |
| 73 | top_candidates: list[GlobalScanCandidate] |
| 74 | deep_analysis_candidate_ids: list[UUID] |
| 75 | |
| 76 | |
| 77 | class StageBCandidate(ResearchBaseModel): |
| 78 | global_instrument_id: UUID |
| 79 | symbol: str | None |
| 80 | pre_score: float | None |
| 81 | feature_version: str = "GLOBAL_STAGE_B_V1" |
| 82 | technical_feature_snapshot: TechnicalFeatureSnapshot |
| 83 | sector_relative_strength_snapshot: SectorRelativeStrengthSnapshot |
| 84 | technical_score: float | None |
| 85 | sector_score: float | None |
| 86 | stage_b_score: float | None |
| 87 | confidence: float |
| 88 | score_coverage: float |
| 89 | score_weights: dict[str, float] |
| 90 | |
| 91 | |
| 92 | class EquityUniverse(Protocol): |
| 93 | async def active_global_equities(self, **kwargs) -> list[dict]: ... |
| 94 | |
| 95 | |
| 96 | class CanonicalEquityUniverse: |
| 97 | """Read canonical metadata only; no portfolio ownership or provider acquisition.""" |
| 98 | |
| 99 | def __init__(self, client: httpx.AsyncClient, base_url: str): |
| 100 | self.client, self.base_url = client, base_url.rstrip("/") |
| 101 | |
| 102 | async def active_global_equities(self, *, correlation_id=None, identity_headers=None): |
| 103 | headers = {k: v for k, v in (identity_headers or {}).items() if v} |
| 104 | if correlation_id: |
| 105 | headers["X-Correlation-Id"] = correlation_id |
| 106 | values, page = [], 0 |
| 107 | while True: |
| 108 | response = await self.client.get( |
| 109 | f"{self.base_url}/api/v1/instruments", |
| 110 | params={"status": "ACTIVE", "assetType": "EQUITY", "page": page, "size": 500}, |
| 111 | headers=headers or None, |
| 112 | ) |
| 113 | response.raise_for_status() |
| 114 | payload = response.json() |
| 115 | if not isinstance(payload, dict) or not isinstance(payload.get("instruments"), list): |
| 116 | raise ValueError("Invalid canonical universe response") |
| 117 | batch = payload["instruments"] |
| 118 | total = int(payload.get("totalElements", len(values) + len(batch))) |
| 119 | values.extend(batch) |
| 120 | if len(values) >= total: |
| 121 | return values |
| 122 | if not batch: |
| 123 | raise ValueError("Incomplete canonical universe pagination") |
| 124 | page += 1 |
| 125 | |
| 126 | |
| 127 | def _utc(value): |
| 128 | return value.replace(tzinfo=timezone.utc) if value.tzinfo is None else value.astimezone(timezone.utc) |
| 129 | |
| 130 | |
| 131 | def _number(value): |
| 132 | if value is None or isinstance(value, bool): |
| 133 | return None |
| 134 | try: |
| 135 | number = Decimal(str(value)) |
| 136 | return number if number.is_finite() else None |
| 137 | except (InvalidOperation, ValueError): |
| 138 | return None |
| 139 | |
| 140 | |
| 141 | class GlobalPreScore: |
| 142 | """Cheap readiness and sign heuristics; no valuation targets or V1 formulas. |
| 143 | |
| 144 | Critical inputs are a fresh trusted price and at least one fresh profitability |
| 145 | metric. Optional dimensions never supply a synthetic zero. Quality metrics |
| 146 | use only sign (positive=100, actual zero=50, negative=0), avoiding unit-dependent |
| 147 | thresholds. No financial ratios or technical indicators are derived here. |
| 148 | """ |
| 149 | |
| 150 | def __init__(self, *, price_max_age=timedelta(days=7), financial_max_age=timedelta(days=550), history_points=50): |
| 151 | if price_max_age.total_seconds() <= 0 or financial_max_age.total_seconds() <= 0 or history_points < 2: |
| 152 | raise ValueError("Readiness limits must be positive; history_points must be >= 2") |
| 153 | self.price_max_age = price_max_age |
| 154 | self.financial_max_age = financial_max_age |
| 155 | self.history_points = history_points |
| 156 | |
| 157 | def score(self, instrument, snapshots, facts, prices, *, as_of): |
| 158 | as_of = _utc(as_of) |
| 159 | missing, stale = set(), set() |
| 160 | evidence = defaultdict(list) |
| 161 | currency = instrument.get("currency") |
| 162 | mappings = [m for m in instrument.get("providerMappings", []) if isinstance(m, dict) |
| 163 | and m.get("status") == "VERIFIED" and m.get("provider") |
| 164 | and (m.get("providerInstrumentId") or m.get("providerSymbol")) |
| 165 | and m.get("resolutionSource") != "BROKER_IMPORT_IDENTITY"] |
| 166 | providers = {m["provider"] for m in mappings} |
| 167 | for record in snapshots: |
| 168 | mapped_ids = {str(m.get(key)) for m in mappings if m["provider"] == record.provider |
| 169 | for key in ("providerInstrumentId", "providerSymbol") if m.get(key)} |
| 170 | if (record.provider not in providers or _utc(record.retrieved_at) > as_of |
| 171 | or record.provider_instrument_id not in mapped_ids or record.currency != currency): |
| 172 | continue |
| 173 | for key, value in record.snapshot.facts.items(): |
| 174 | if _utc(value.retrieved_at) > as_of: |
| 175 | continue |
| 176 | stamp = value.as_of_date or value.published_at or record.market_as_of or record.retrieved_at |
| 177 | if _utc(stamp) <= as_of: |
| 178 | evidence[key].append((value.value, _utc(stamp))) |
| 179 | financial_evidence = defaultdict(list) |
| 180 | for fact in facts: |
| 181 | if fact.source_tier not in SUPPORTED_FINANCIAL_SOURCE_TIERS or str(fact.source_mode) != "REAL" or _utc(fact.value.retrieved_at) > as_of: |
| 182 | continue |
| 183 | stamp = fact.value.as_of_date or fact.value.published_at or fact.value.retrieved_at |
| 184 | if fact.key.period_end: |
| 185 | try: |
| 186 | stamp = datetime.fromisoformat(fact.key.period_end) |
| 187 | except ValueError: |
| 188 | continue |
| 189 | if _utc(stamp) <= as_of: |
| 190 | metric = {"profit_margin": "profitMargin", "return_on_equity": "roe", |
| 191 | "net_income": "pat", "net_profit": "pat", "revenue_growth": "revenueGrowth", |
| 192 | "earnings_growth": "earningsGrowth", "equity": "total_equity"}.get(fact.key.metric, fact.key.metric) |
| 193 | # Annual and quarterly amounts (or standalone and consolidated |
| 194 | # statements) are not conflicting observations of one metric. |
| 195 | basis_rank = {"CONSOLIDATED": 2, "UNKNOWN": 1}.get(fact.key.reporting_basis, 0) |
| 196 | period_rank = {"ANNUAL": 2, "QUARTERLY": 1}.get(fact.key.period_type, 0) |
| 197 | financial_evidence[metric].append((fact.value.value, _utc(stamp), basis_rank, period_rank)) |
| 198 | for metric, values in financial_evidence.items(): |
| 199 | selected = max((stamp, basis, period) for _, stamp, basis, period in values) |
| 200 | evidence[metric].extend((value, stamp) for value, stamp, basis, period in values |
| 201 | if (stamp, basis, period) == selected) |
| 202 | |
| 203 | def dimension(keys, *, quality=False, text=False, max_age=None): |
| 204 | scores, ready, conflicts, old = [], 0, False, False |
| 205 | for key in keys: |
| 206 | values = evidence.get(key, []) |
| 207 | valid = [(str(v).strip() if text and v is not None else _number(v), t) for v, t in values] |
| 208 | valid = [(v, t) for v, t in valid if v is not None and v != ""] |
| 209 | if not valid: |
| 210 | missing.add(key) |
| 211 | continue |
| 212 | latest = max(t for _, t in valid) |
| 213 | current = {v for v, t in valid if t == latest} |
| 214 | if len(current) > 1: |
| 215 | conflicts = True |
| 216 | continue |
| 217 | value = next(iter(current)) |
| 218 | is_stale = as_of - latest > (max_age or self.financial_max_age) |
| 219 | if is_stale: |
| 220 | stale.add(key) |
| 221 | old = True |
| 222 | else: |
| 223 | ready += 1 |
| 224 | scores.append((100 if value > 0 else 50 if value == 0 else 0) if quality else 100) |
| 225 | state = (EvidenceState.CONFLICTING if conflicts else EvidenceState.STALE if old else |
| 226 | EvidenceState.MISSING if not scores else EvidenceState.PARTIAL if len(scores) < len(keys) |
| 227 | else EvidenceState.AVAILABLE) |
| 228 | return PreScoreDimension(state=state, score=sum(scores)/len(scores) if scores and not conflicts else None, |
| 229 | coverage=ready/len(keys) if not conflicts else 0) |
| 230 | |
| 231 | usable = [p for p in prices if p.provider in providers and _number(p.price) is not None and p.price > 0 |
| 232 | and currency and p.currency == currency and _utc(p.observed_at) <= as_of and _utc(p.retrieved_at) <= as_of] |
| 233 | latest = max((_utc(p.observed_at) for p in usable), default=None) |
| 234 | price_conflict = len({p.price for p in usable if _utc(p.observed_at) == latest}) > 1 |
| 235 | fresh_price = latest is not None and as_of - latest <= self.price_max_age and not price_conflict |
| 236 | price_state = (EvidenceState.CONFLICTING if price_conflict else EvidenceState.MISSING if latest is None |
| 237 | else EvidenceState.AVAILABLE if fresh_price else EvidenceState.STALE) |
| 238 | if latest is None: |
| 239 | missing.add("price") |
| 240 | elif not fresh_price and not price_conflict: |
| 241 | stale.add("price") |
| 242 | days = len({_utc(p.observed_at).date() for p in usable}) |
| 243 | dimensions = { |
| 244 | "PRICE_DATA_QUALITY": PreScoreDimension(state=price_state, score=100 if usable and not price_conflict else None, coverage=float(fresh_price)), |
| 245 | "VALUATION_AVAILABILITY": dimension(["trailingPE", "priceToBook"], max_age=self.price_max_age), |
| 246 | "PROFITABILITY_QUALITY": dimension(["profitMargin", "roe", "pat"], quality=True), |
| 247 | "GROWTH_QUALITY": dimension(["revenueGrowth", "earningsGrowth"], quality=True), |
| 248 | "BALANCE_SHEET_QUALITY": dimension(["total_equity"], quality=True), |
| 249 | "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, |
| 250 | score=100 if days >= self.history_points else None, coverage=min(1, days/self.history_points) if fresh_price else 0), |
| 251 | "SECTOR_DATA_READINESS": dimension(["sector"], text=True), |
| 252 | } |
| 253 | if days < self.history_points: |
| 254 | missing.add("technicalHistory") |
| 255 | profitability = dimensions["PROFITABILITY_QUALITY"] |
| 256 | financial_available = profitability.score is not None |
| 257 | financial_ready = profitability.coverage > 0 and profitability.state in {"AVAILABLE", "PARTIAL"} |
| 258 | critical = (int(fresh_price) + int(financial_ready)) / 2 |
| 259 | readiness = sum(d.coverage for d in dimensions.values()) / len(dimensions) |
| 260 | dimensions["FRESHNESS"] = PreScoreDimension(state=EvidenceState.STALE if stale else EvidenceState.AVAILABLE, |
| 261 | score=None, coverage=readiness) |
| 262 | dimensions["CRITICAL_COMPLETENESS"] = PreScoreDimension(state=EvidenceState.AVAILABLE if critical == 1 else EvidenceState.PARTIAL if critical else EvidenceState.MISSING, |
| 263 | score=None, coverage=critical) |
| 264 | symbol = instrument.get("ticker") or instrument.get("primarySymbol") |
| 265 | name = instrument.get("canonicalName") |
| 266 | reasons = [] |
| 267 | if instrument.get("status") != "ACTIVE": reasons.append("INACTIVE") |
| 268 | if instrument.get("assetType") != "EQUITY": reasons.append("NON_EQUITY") |
| 269 | if not (mappings and symbol and name and instrument.get("exchange") and currency): reasons.append("UNTRUSTED_CANONICAL_IDENTITY") |
| 270 | if not usable: reasons.append("NO_USABLE_PRICE") |
| 271 | elif price_conflict: reasons.append("CONFLICTING_PRICE") |
| 272 | elif not fresh_price: reasons.append("STALE_PRICE") |
| 273 | if not financial_ready: reasons.append("CRITICAL_FINANCIAL_EVIDENCE_UNREADY") |
| 274 | scores = [d.score for d in dimensions.values() if d.score is not None and d.state in {"AVAILABLE", "PARTIAL"}] |
| 275 | return GlobalScanCandidate(global_instrument_id=instrument["globalInstrumentId"], market=instrument.get("exchange"), |
| 276 | region=instrument.get("region"), country=instrument.get("country"), exchange=instrument.get("exchange"), |
| 277 | currency=currency, asset_type=instrument.get("assetType"), status=instrument.get("status"), symbol=symbol, company_name=name, |
| 278 | verified_provider_mapping_status="VERIFIED" if mappings else "MISSING", pre_score=round(sum(scores)/len(scores), 6) if scores else None, |
| 279 | confidence=round(100*readiness*critical, 6), critical_completeness=critical*100, |
| 280 | price_data_available=bool(usable) and not price_conflict, financial_data_available=financial_available, |
| 281 | technical_history_available=days >= self.history_points, sector_data_available=dimensions["SECTOR_DATA_READINESS"].score is not None, |
| 282 | dimensions=dimensions, missing_inputs=sorted(missing), stale_inputs=sorted(stale), |
| 283 | eligible_for_deep_analysis=not reasons, exclusion_reasons=reasons) |
| 284 | |
| 285 | |
| 286 | class GlobalScanner: |
| 287 | def __init__(self, universe: EquityUniverse, persistence: ResearchPersistence, *, pre_score=None, batch_size=250): |
| 288 | if not 1 <= batch_size <= 500: |
| 289 | raise ValueError("batch_size must be between 1 and 500") |
| 290 | self.universe, self.persistence = universe, persistence |
| 291 | self.pre_score, self.batch_size = pre_score or GlobalPreScore(), batch_size |
| 292 | |
| 293 | def enrich_candidates(self, scan: GlobalScanResult, *, sector_contexts: dict[UUID, SectorContext] | None = None, |
| 294 | trusted_providers: dict[UUID, frozenset[str]] | None = None, |
| 295 | technical_engine: TechnicalFeatureEngine | None = None, |
| 296 | sector_engine: SectorRelativeStrengthEngine | None = None, |
| 297 | technical_weight: float = 0.7, sector_weight: float = 0.3) -> list[StageBCandidate]: |
| 298 | """Optional Stage B; no universe enumeration, providers, refresh or V1. |
| 299 | |
| 300 | Enrich all deep-eligible candidates (not the entire canonical universe). |
| 301 | Prices for candidates and explicitly mapped benchmarks share bounded SQL |
| 302 | batches. Default weights emphasize technical evidence; they are selection |
| 303 | defaults, not calibrated investment weights. Missing legs renormalize the |
| 304 | score while reducing reported coverage/confidence. Phase 1 is untouched. |
| 305 | """ |
| 306 | from math import isfinite |
| 307 | |
| 308 | if (not all(isfinite(v) and v >= 0 for v in (technical_weight, sector_weight)) |
| 309 | or technical_weight + sector_weight == 0): |
| 310 | raise ValueError("Stage-B weights must be finite, nonnegative and have positive sum") |
| 311 | technical_engine = technical_engine or TechnicalFeatureEngine() |
| 312 | sector_engine = sector_engine or SectorRelativeStrengthEngine() |
| 313 | contexts, providers = sector_contexts or {}, trusted_providers or {} |
| 314 | candidates = [c for c in scan.candidates if c.eligible_for_deep_analysis] |
| 315 | ids = {c.global_instrument_id for c in candidates} |
| 316 | daily_histories = defaultdict(list) |
| 317 | for candidate in candidates: |
| 318 | context = contexts.get(candidate.global_instrument_id, SectorContext()) |
| 319 | for reference in (context.sector_benchmark, context.market_benchmark): |
| 320 | if reference: |
| 321 | ids.add(reference.instrument_id) |
| 322 | histories = defaultdict(list) |
| 323 | ordered = sorted(ids, key=str) |
| 324 | for offset in range(0, len(ordered), self.batch_size): |
| 325 | batch = set(ordered[offset:offset+self.batch_size]) |
| 326 | for row in self.persistence.load_daily_market_bars(batch, provider="NSE"): |
| 327 | if row.global_instrument_id in batch: |
| 328 | daily_histories[row.global_instrument_id].append(row) |
| 329 | for row in self.persistence.load_market_price_observations(batch): |
| 330 | if row.instrument_id in batch: |
| 331 | histories[row.instrument_id].append(row) |
| 332 | output = [] |
| 333 | total_weight = technical_weight + sector_weight |
| 334 | for candidate in sorted(candidates, key=lambda c: str(c.global_instrument_id)): |
| 335 | key = candidate.global_instrument_id |
| 336 | technical = technical_engine.compute(key, histories[key], as_of=scan.as_of, currency=candidate.currency, |
| 337 | trusted_providers=providers.get(key), daily_bar_history=daily_histories[key]) |
| 338 | sector = sector_engine.compute(key, histories[key], as_of=scan.as_of, currency=candidate.currency, |
| 339 | context=contexts.get(key), benchmark_histories=histories, trusted_providers=providers.get(key), |
| 340 | daily_bar_histories=daily_histories) |
| 341 | scores = [(technical.technical_score, technical_weight), (sector.relative_strength_score, sector_weight)] |
| 342 | available = [(score, weight) for score, weight in scores if score is not None and weight > 0] |
| 343 | available_weight = sum(weight for _, weight in available) |
| 344 | score = sum(value * weight for value, weight in available) / available_weight if available_weight else None |
| 345 | output.append(StageBCandidate(global_instrument_id=key, symbol=candidate.symbol, pre_score=candidate.pre_score, |
| 346 | technical_feature_snapshot=technical, sector_relative_strength_snapshot=sector, |
| 347 | technical_score=technical.technical_score, sector_score=sector.relative_strength_score, |
| 348 | stage_b_score=round(score, 8) if score is not None else None, |
| 349 | confidence=round((technical.confidence * technical_weight + sector.confidence * sector_weight) / total_weight, 6), |
| 350 | score_coverage=round(available_weight / total_weight * 100, 6), |
| 351 | score_weights={"technical": technical_weight, "sector": sector_weight})) |
| 352 | output.sort(key=lambda c: (-(c.stage_b_score if c.stage_b_score is not None else -1), -c.confidence, str(c.global_instrument_id))) |
| 353 | return output |
| 354 | |
| 355 | async def scan(self, *, as_of: datetime, top_n: int, **universe_context) -> GlobalScanResult: |
| 356 | if top_n < 0: |
| 357 | raise ValueError("top_n must be nonnegative") |
| 358 | instruments = await self.universe.active_global_equities(**universe_context) |
| 359 | by_id = {} |
| 360 | invalid = 0 |
| 361 | for item in instruments: |
| 362 | try: |
| 363 | key = UUID(str(item.get("globalInstrumentId"))) |
| 364 | except (ValueError, TypeError): |
| 365 | invalid += 1 |
| 366 | continue |
| 367 | if key in by_id and by_id[key] != item: |
| 368 | raise ValueError(f"Conflicting canonical records for {key}") |
| 369 | by_id[key] = item |
| 370 | candidates = [] |
| 371 | ordered = sorted(by_id, key=str) |
| 372 | for offset in range(0, len(ordered), self.batch_size): |
| 373 | ids = set(ordered[offset:offset+self.batch_size]) |
| 374 | snapshots, facts, prices = defaultdict(list), defaultdict(list), defaultdict(list) |
| 375 | for row in self.persistence.load_structured_market_snapshots(ids): snapshots[row.instrument_id].append(row) |
| 376 | for row in self.persistence.load_financial_facts(ids): facts[row.key.instrument_id].append(row) |
| 377 | for row in self.persistence.load_market_price_observations(ids): prices[row.instrument_id].append(row) |
| 378 | for key in sorted(ids, key=str): |
| 379 | candidates.append(self.pre_score.score(by_id[key], snapshots[key], facts[key], prices[key], as_of=as_of)) |
| 380 | 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))) |
| 381 | eligible = [c for c in candidates if c.eligible_for_deep_analysis] |
| 382 | excluded = Counter(reason for c in candidates for reason in c.exclusion_reasons) |
| 383 | if invalid: excluded["UNTRUSTED_CANONICAL_IDENTITY"] += invalid |
| 384 | top = eligible[:top_n] |
| 385 | 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()), |
| 386 | eligible_candidates=len(eligible), excluded_candidates_by_reason=dict(sorted(excluded.items())), |
| 387 | deep_analysis_eligible_count=len(eligible), candidates=candidates, top_candidates=top, |
| 388 | deep_analysis_candidate_ids=[c.global_instrument_id for c in top]) |