| 1 | """Explicit engineering ranking over already-read canonical and persisted evidence. |
| 2 | |
| 3 | No HTTP universe adapter, readiness ensure, provider refresh, or membership reads. |
| 4 | Pass PortfolioResearchOrchestrator.register_global_profile_metadata as the |
| 5 | profile_hydrator: this existing synchronous method consumes supplied metadata |
| 6 | only. The repository must already have hydrated its persisted public evidence |
| 7 | (ResearchRepository does this at initialization). Existing V1 cache writes are |
| 8 | preserved; this operation does not persist ranking snapshots. |
| 9 | |
| 10 | Use a fixed timezone-aware as_of to reproduce evidence selection/fingerprints. |
| 11 | generated_at is the completion clock; cache_hit diagnostics may change on a warm |
| 12 | run, but ranking content does not. All limits are explicit and at most 100. |
| 13 | """ |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | from copy import deepcopy |
| 17 | from datetime import datetime, timezone |
| 18 | from typing import Callable, Iterable, Mapping |
| 19 | from uuid import UUID |
| 20 | from pydantic import Field |
| 21 | |
| 22 | from app.global_scanner import GlobalScanner |
| 23 | from app.global_opportunity_ranker import GlobalOpportunityRanker |
| 24 | from app.models import ResearchBaseModel |
| 25 | from app.research_readiness_runtime import RepositoryResearchReadinessAdapter, ResearchReadinessRuntime, jurisdiction_for_profile |
| 26 | from app.sector_relative_strength import SectorContext |
| 27 | from app.stock_rule_engine import StockRuleEngineService |
| 28 | from app.news_intelligence import EventImpactFeature, latest_known_features |
| 29 | |
| 30 | |
| 31 | class OpportunityEntry(ResearchBaseModel): |
| 32 | rank: int |
| 33 | global_instrument_id: UUID |
| 34 | symbol: str | None |
| 35 | company_name: str | None |
| 36 | sector: str | None |
| 37 | market: str | None |
| 38 | pre_score: float | None |
| 39 | stage_b_score: float | None |
| 40 | technical_score: float | None |
| 41 | technical_state: str |
| 42 | sector_score: float | None |
| 43 | sector_state: str |
| 44 | rule_engine_score: float | None |
| 45 | rule_engine_confidence: float |
| 46 | opportunity_score: float | None |
| 47 | opportunity_confidence: float |
| 48 | score_coverage: float |
| 49 | top_positive_reasons: list[str] |
| 50 | top_negative_reasons: list[str] |
| 51 | rule_engine_version: str |
| 52 | technical_feature_version: str |
| 53 | sector_feature_version: str |
| 54 | opportunity_ranker_version: str |
| 55 | evidence_state: dict = Field(default_factory=dict) |
| 56 | rank_eligible: bool = True |
| 57 | suppression_reasons: list[str] = Field(default_factory=list) |
| 58 | |
| 59 | |
| 60 | class CandidateDiagnostic(ResearchBaseModel): |
| 61 | global_instrument_id: UUID |
| 62 | status: str |
| 63 | failure_reason: str | None = None |
| 64 | cache_hit: bool | None = None |
| 65 | rank_eligible: bool = False |
| 66 | suppression_reasons: list[str] = Field(default_factory=list) |
| 67 | |
| 68 | |
| 69 | class OpportunityRanking(ResearchBaseModel): |
| 70 | generated_at: datetime |
| 71 | as_of: datetime |
| 72 | universe_count: int |
| 73 | phase1_eligible_count: int |
| 74 | stage_b_count: int |
| 75 | shortlist_count: int |
| 76 | deep_evaluated_count: int |
| 77 | rank_eligible_count: int |
| 78 | top_n: list[OpportunityEntry] |
| 79 | diagnostics: list[CandidateDiagnostic] |
| 80 | evaluated_entries: list[OpportunityEntry] = Field(default_factory=list) |
| 81 | |
| 82 | |
| 83 | class _PersistedUniverse: |
| 84 | """The scanner's existing universe protocol, backed solely by supplied rows.""" |
| 85 | def __init__(self, rows): |
| 86 | self.rows = rows |
| 87 | |
| 88 | async def active_global_equities(self, **unused): |
| 89 | return self.rows |
| 90 | |
| 91 | |
| 92 | class GlobalOpportunityOrchestrator: |
| 93 | def __init__(self, repository, persistence, *, profile_hydrator: Callable[[UUID, dict], bool], clock=None): |
| 94 | self.repository, self.persistence = repository, persistence |
| 95 | self.profile_hydrator = profile_hydrator |
| 96 | self.clock = clock or (lambda: datetime.now(timezone.utc)) |
| 97 | self.readiness_adapter = RepositoryResearchReadinessAdapter(repository) |
| 98 | self.readiness = ResearchReadinessRuntime(repository, self.readiness_adapter, executor=None) |
| 99 | self.rule_engine = StockRuleEngineService(repository, self.readiness_adapter) |
| 100 | self.ranker = GlobalOpportunityRanker() |
| 101 | |
| 102 | async def run(self, canonical_instruments: Iterable[dict], *, as_of: datetime, |
| 103 | sector_contexts: Mapping[UUID, SectorContext] | None = None, |
| 104 | shortlist_limit: int = 25, top_n: int = 10, |
| 105 | review_ids: Iterable[UUID] = ()) -> OpportunityRanking: |
| 106 | if (type(shortlist_limit) is not int or not 1 <= shortlist_limit <= 100 |
| 107 | or type(top_n) is not int or not 0 <= top_n <= 100): |
| 108 | raise ValueError('INVALID_OPPORTUNITY_LIMIT') |
| 109 | if as_of.tzinfo is None or as_of.utcoffset() is None: |
| 110 | raise ValueError('AWARE_AS_OF_REQUIRED') |
| 111 | as_of = as_of.astimezone(timezone.utc) |
| 112 | rows = deepcopy(list(canonical_instruments)) |
| 113 | # Support both existing canonical enumeration and master-detail shapes. |
| 114 | for row in rows: |
| 115 | row.setdefault('exchange', row.get('primaryExchange')) |
| 116 | row.setdefault('ticker', row.get('primarySymbol')) |
| 117 | scanner = GlobalScanner(_PersistedUniverse(rows), self.persistence) |
| 118 | scan = await scanner.scan(as_of=as_of, top_n=0) |
| 119 | phase1 = {c.global_instrument_id:c for c in scan.candidates if c.eligible_for_deep_analysis} |
| 120 | stage_b = scanner.enrich_candidates(scan, sector_contexts=dict(sector_contexts or {})) |
| 121 | def desc(value): return -value if value is not None else float('inf') |
| 122 | shortlist = sorted((c for c in stage_b if c.global_instrument_id in phase1), key=lambda c: ( |
| 123 | desc(c.stage_b_score), -c.confidence, desc(phase1[c.global_instrument_id].pre_score), |
| 124 | -phase1[c.global_instrument_id].confidence, str(c.global_instrument_id)))[:shortlist_limit] |
| 125 | # Prior public recommendations have a separate bounded review budget. They never |
| 126 | # affect scores or the scanner shortlist. Rotate oldest projections in the caller. |
| 127 | review_ids = list(review_ids)[:25] |
| 128 | stage_by_id = {c.global_instrument_id: c for c in stage_b} |
| 129 | selected = {c.global_instrument_id for c in shortlist} |
| 130 | reviews = [stage_by_id[key] for key in review_ids if key in stage_by_id and key not in selected] |
| 131 | evaluation_candidates = shortlist + reviews |
| 132 | initial_by_id = {c.global_instrument_id: c for c in scan.candidates} |
| 133 | metadata = {str(row.get('globalInstrumentId')):row for row in rows} |
| 134 | diagnostics, rules, successful, temporal_evidence = [], {}, [], {} |
| 135 | evaluated = 0 |
| 136 | for candidate in evaluation_candidates: |
| 137 | key = candidate.global_instrument_id |
| 138 | step = 'PUBLIC_EVIDENCE_UNAVAILABLE' |
| 139 | cache_hit = None |
| 140 | try: |
| 141 | payload = dict(metadata[str(key)]) |
| 142 | payload.setdefault('primaryExchange', payload.get('exchange')) |
| 143 | payload.setdefault('primarySymbol', payload.get('ticker')) |
| 144 | if not self.profile_hydrator(key, payload): |
| 145 | raise ValueError('UNRESOLVED_PROFILE') |
| 146 | profile = self.repository.profile(key) |
| 147 | if profile.instrument_id != key: |
| 148 | raise ValueError('PROFILE_IDENTITY_MISMATCH') |
| 149 | self.readiness_adapter.remember_canonical_metadata(key, payload) |
| 150 | step = 'READINESS_UNAVAILABLE' |
| 151 | readiness = await self.readiness.read(key, jurisdiction=jurisdiction_for_profile(profile), now=as_of) |
| 152 | step = 'RULE_ENGINE_UNAVAILABLE' |
| 153 | result = await self.rule_engine.analyze(profile, readiness, allow_partial=False, now=as_of) |
| 154 | evaluated += 1 |
| 155 | cache_hit = result.cache_hit |
| 156 | step = 'RANKER_INPUT_UNAVAILABLE' |
| 157 | ranked = self.ranker.score(candidate, result) |
| 158 | loader = getattr(self.repository, 'news_records_for', None) |
| 159 | features = loader(key, EventImpactFeature, as_of=as_of) if callable(loader) else [] |
| 160 | temporal_evidence[key] = [f.model_dump(mode='json') for f in latest_known_features(features, as_of)] |
| 161 | diagnostics.append(CandidateDiagnostic(global_instrument_id=key, |
| 162 | status='RANK_ELIGIBLE' if ranked.rank_eligible else 'SUPPRESSED', cache_hit=cache_hit, |
| 163 | rank_eligible=ranked.rank_eligible, suppression_reasons=ranked.eligibility_reasons)) |
| 164 | rules[key] = result |
| 165 | successful.append(candidate) |
| 166 | except Exception: |
| 167 | # Never return exception messages, headers, raw evidence or recommendations. |
| 168 | diagnostics.append(CandidateDiagnostic(global_instrument_id=key, status='FAILED', |
| 169 | failure_reason=step, cache_hit=cache_hit)) |
| 170 | ranked = [r for r in self.ranker.rank(successful, rules) if r.rank_eligible] |
| 171 | eligible_ids = {r.global_instrument_id for r in ranked} |
| 172 | suppressed = [self.ranker.score(c, rules[c.global_instrument_id]) for c in successful |
| 173 | if c.global_instrument_id not in eligible_ids] |
| 174 | by_id = {c.global_instrument_id:c for c in successful} |
| 175 | entries = [] |
| 176 | for position, result in enumerate([*ranked, *suppressed], 1): |
| 177 | key = result.global_instrument_id |
| 178 | initial, enriched, rule = initial_by_id[key], by_id[key], rules[key] |
| 179 | entries.append(OpportunityEntry(rank=position, global_instrument_id=key, |
| 180 | symbol=initial.symbol, company_name=initial.company_name, |
| 181 | sector=enriched.sector_relative_strength_snapshot.sector, market=initial.market, |
| 182 | pre_score=initial.pre_score, stage_b_score=enriched.stage_b_score, |
| 183 | technical_score=result.technical_score, technical_state=result.technical_state, |
| 184 | sector_score=result.sector_score, sector_state=result.sector_state, |
| 185 | rule_engine_score=result.rule_engine_score, rule_engine_confidence=rule.confidence_score, |
| 186 | opportunity_score=result.opportunity_score, opportunity_confidence=result.opportunity_confidence, |
| 187 | score_coverage=result.score_coverage, top_positive_reasons=result.top_positive_reasons, |
| 188 | top_negative_reasons=result.top_negative_reasons, rule_engine_version=rule.rule_engine_version, |
| 189 | technical_feature_version=enriched.technical_feature_snapshot.feature_version, |
| 190 | sector_feature_version=enriched.sector_relative_strength_snapshot.feature_version, |
| 191 | opportunity_ranker_version=result.ranker_version, |
| 192 | rank_eligible=result.rank_eligible, suppression_reasons=result.eligibility_reasons, |
| 193 | evidence_state={'technical': enriched.technical_feature_snapshot.model_dump(mode='json'), |
| 194 | 'sector': enriched.sector_relative_strength_snapshot.model_dump(mode='json'), |
| 195 | 'rule': rule.model_dump(mode='json'), 'missing_inputs': initial.missing_inputs, |
| 196 | 'stale_inputs': initial.stale_inputs, 'news_features': temporal_evidence.get(key, [])})) |
| 197 | return OpportunityRanking(generated_at=self.clock(), as_of=as_of, |
| 198 | universe_count=scan.total_canonical_active_equities, phase1_eligible_count=len(phase1), |
| 199 | stage_b_count=len(stage_b), shortlist_count=len(shortlist), deep_evaluated_count=evaluated, |
| 200 | rank_eligible_count=len(ranked), top_n=[e for e in entries if e.rank_eligible][:top_n], |
| 201 | diagnostics=diagnostics, evaluated_entries=entries) |