| 1 | """Explicit global job. No provider acquisition; canonical catalog reads precede ranking.""" |
| 2 | from datetime import datetime, timezone |
| 3 | from uuid import uuid4, UUID |
| 4 | from app.global_opportunity_orchestration import GlobalOpportunityOrchestrator |
| 5 | from app.recommendation_engine import RecommendationEngineV1, lifecycle, area_scores |
| 6 | |
| 7 | |
| 8 | def nse_equities(rows): |
| 9 | return [r for r in rows if (r.get('exchange') or r.get('primaryExchange')) == 'NSE' |
| 10 | and r.get('status') == 'ACTIVE' and r.get('assetType') == 'EQUITY'] |
| 11 | |
| 12 | |
| 13 | def snapshot_from_entry(entry, cycle_id, now): |
| 14 | s = entry.model_dump(mode='json') |
| 15 | technical = s['evidence_state'].get('technical', {}) |
| 16 | s.update(snapshot_id=str(uuid4()), cycle_id=cycle_id, market='NSE', generated_at=now, |
| 17 | scanner_version='GLOBAL_PRE_SCORE_V1', rank_position=s.pop('rank'), |
| 18 | technical_version=s['technical_feature_version'], sector_version=s['sector_feature_version'], |
| 19 | ranker_version=s['opportunity_ranker_version'], current_price=technical.get('latest_price'), |
| 20 | price_as_of=technical.get('history_end')) |
| 21 | areas = area_scores(s['evidence_state'].get('rule', {})) |
| 22 | for area in ('valuation', 'quality', 'growth', 'balance_sheet', 'quarterly', 'catalyst', 'news', 'shareholding', 'governance'): |
| 23 | s[area + '_score'] = areas.get(area.upper()) |
| 24 | s['evidence_state']['as_of'] = now |
| 25 | return s |
| 26 | |
| 27 | |
| 28 | def prepare_cycle(persistence, snapshots, *, cycle_id, now, top_n, diagnostics=None): |
| 29 | engine = RecommendationEngineV1() |
| 30 | old_states = {s['global_instrument_id']: s for s in persistence.recommendation_states()} |
| 31 | histories = {r['recommendation_id']: r for r in persistence.recommendation_history()} |
| 32 | new_history, states, cards = [], [], [] |
| 33 | for s in snapshots: |
| 34 | key = s['global_instrument_id'] |
| 35 | old = old_states.get(key) |
| 36 | previous = histories.get(old['latest_recommendation_id']) if old else None |
| 37 | anchor = histories.get(old.get('anchor_recommendation_id')) if old else None |
| 38 | recommendation = engine.evaluate(s, anchor or previous) |
| 39 | anchor = anchor or previous or recommendation |
| 40 | state = lifecycle(anchor, recommendation, s['current_price'], old) |
| 41 | if previous and previous['fingerprint'] == recommendation['fingerprint']: |
| 42 | latest = previous |
| 43 | else: |
| 44 | recommendation.update(recommendation_id=str(uuid4()), snapshot_id=s['snapshot_id']) |
| 45 | latest = recommendation |
| 46 | new_history.append(latest) |
| 47 | state.update(global_instrument_id=key, latest_recommendation_id=latest['recommendation_id'], |
| 48 | anchor_recommendation_id=(old.get('anchor_recommendation_id') or old['latest_recommendation_id']) if old else latest['recommendation_id'], |
| 49 | updated_at=now, symbol=s.get('symbol'), company_name=s.get('company_name')) |
| 50 | states.append(state) |
| 51 | evidence = s['evidence_state'] |
| 52 | stale = sorted(set(evidence.get('stale_inputs', []) + evidence.get('technical', {}).get('stale_inputs', []))) |
| 53 | missing = sorted(set(evidence.get('missing_inputs', []) + evidence.get('rule', {}).get('missing_inputs', []) + evidence.get('technical', {}).get('missing_inputs', []))) |
| 54 | cards.append({**s, **latest, **state, 'opportunity_score': s['opportunity_score'], |
| 55 | 'confidence': s['opportunity_confidence'], 'coverage': s['score_coverage'], |
| 56 | 'missing_areas': missing, 'stale_areas': stale, |
| 57 | 'data_state': 'STALE' if stale else 'PARTIAL' if missing or s['score_coverage'] < 100 else 'FRESH', |
| 58 | 'short_horizon': '1 week–3 months', 'long_horizon': '6–12 months'}) |
| 59 | eligible = [c for c in cards if c['rank_eligible']] |
| 60 | def order(c): |
| 61 | return (-c['opportunity_score'], -c['opportunity_confidence'], -c['score_coverage'], |
| 62 | -(c['rule_engine_score'] if c['rule_engine_score'] is not None else -1), c['global_instrument_id']) |
| 63 | short = sorted([c for c in eligible if c['current_short_action'] == 'BUY'], key=order)[:top_n] |
| 64 | long = sorted([c for c in eligible if c['current_long_action'] in {'ACCUMULATE', 'TOP_UP'}], key=order)[:top_n] |
| 65 | buys = sorted({c['global_instrument_id']: c for c in short + long}.values(), key=order) |
| 66 | # Unseen prior recommendations remain visible, explicitly marked as not evaluated this cycle. |
| 67 | unseen = [{**s, 'evaluation_status': 'NOT_EVALUATED_THIS_CYCLE'} for key, s in old_states.items() |
| 68 | if key not in {r['global_instrument_id'] for r in states}] |
| 69 | selection = dict(cycle_id=cycle_id, generated_at=now, market='NSE', top_n=top_n, |
| 70 | best_buy_today=buys[0] if buys else None, top_short_term=short, top_long_term=long, |
| 71 | previous_recommendations=[c for c in cards if c['global_instrument_id'] in old_states] + unseen, |
| 72 | diagnostics=diagnostics or []) |
| 73 | return new_history, states, selection |
| 74 | |
| 75 | |
| 76 | async def run_global_opportunity_cycle(repository, canonical_source, *, top_n=4, shortlist_limit=25, |
| 77 | candidate_ids=None, identity_headers=None): |
| 78 | if type(top_n) is not int or not 2 <= top_n <= 4: |
| 79 | raise ValueError('TOP_N_MUST_BE_2_TO_4') |
| 80 | now = datetime.now(timezone.utc) |
| 81 | rows = nse_equities(await canonical_source.active_global_equities(identity_headers=identity_headers)) |
| 82 | if candidate_ids is not None: |
| 83 | allowed = {str(k) for k in candidate_ids} |
| 84 | rows = [r for r in rows if str(r['globalInstrumentId']) in allowed] |
| 85 | ids = {UUID(str(r['globalInstrumentId'])) for r in rows} |
| 86 | contexts = await canonical_source.sector_benchmark_contexts(ids, identity_headers=identity_headers) |
| 87 | previous_states = sorted(repository.persistence.recommendation_states(), |
| 88 | key=lambda s: (s['updated_at'], s['global_instrument_id'])) |
| 89 | review_ids = [UUID(s['global_instrument_id']) for s in previous_states if UUID(s['global_instrument_id']) in ids][:25] |
| 90 | orchestrator = GlobalOpportunityOrchestrator(repository, repository.persistence, |
| 91 | profile_hydrator=canonical_source.register_global_profile_metadata) |
| 92 | ranking = await orchestrator.run(rows, as_of=now, sector_contexts=contexts, |
| 93 | shortlist_limit=shortlist_limit, top_n=100, review_ids=review_ids) |
| 94 | cycle_id, timestamp = str(uuid4()), datetime.now(timezone.utc).isoformat() |
| 95 | snapshots = [snapshot_from_entry(e, cycle_id, timestamp) for e in ranking.evaluated_entries] |
| 96 | def build(persisted): |
| 97 | history, states, selection = prepare_cycle(repository.persistence, persisted, cycle_id=cycle_id, |
| 98 | now=timestamp, top_n=top_n, diagnostics=[d.model_dump(mode='json') for d in ranking.diagnostics]) |
| 99 | selection['universe_count'] = ranking.universe_count |
| 100 | selection['controlled_candidate_set'] = candidate_ids is not None |
| 101 | return history, states, selection |
| 102 | # Other research jobs use this lock for the same shared connection. Prevent their |
| 103 | # commits from publishing a half-built cycle while its projections are being assembled. |
| 104 | with repository._persistence_worker_lock: |
| 105 | return repository.persistence.publish_opportunity_cycle(snapshots, [], [], {'cycle_id': cycle_id}, build=build) |