feat: add global recommendation lifecycle and backtesting
prakhar82 committed
Sep 14, 2026 at 20:23 UTC
ca3f49d3dc319bb98999ac0f951ecec6f6e64000
19 files changed
+1385
-16
ai/research-engine/GLOBAL_RECOMMENDATIONS_V1.md
new
+173
@@ -0,0 +1,173 @@
1
+# Global recommendation lifecycle MVP
2
+
3
+## Architecture
4
+
5
+Java research-service Flyway owns additive migration **V13**. Python shares that schema through
6
+`SqliteResearchPersistence` / `PostgresResearchPersistence`; SQLite mirrors it for deterministic tests.
7
+No acquisition, migration backfill, Rule Engine weight changes, or provider calls were added.
8
+
9
+The explicit `run_global_opportunity_cycle()` job enumerates the canonical ACTIVE EQUITY catalog,
10
+filters NSE, then calls the existing `GlobalOpportunityOrchestrator`. It reuses GlobalScanner,
11
+Stage-B technical/sector enrichment, `STOCK_RULE_ENGINE_V1`, and `GLOBAL_OPPORTUNITY_RANKER_V1`.
12
+Catalog and benchmark metadata reads occur before ranking; portfolio/watchlist/broker holdings
13
+never supply candidates or affect any public score.
14
+
15
+The default deep shortlist is 25 (configurable 1–100). Up to 25 previous public recommendations,
16
+ordered by oldest projection update, have an additional review budget. This does not change
17
+the scanner shortlist or scoring. Reviews that cannot be evaluated remain visible with
18
+`NOT_EVALUATED_THIS_CYCLE`; the application does not claim their thesis was refreshed.
19
+
20
+Within one database transaction, the job inserts immutable opportunity snapshots, **reads those
21
+persisted rows back**, evaluates recommendations, appends changed recommendations, updates current
22
+state, and publishes the Top-N projection. A failure rolls back the publication. The POST route
23
+serializes cycles in the existing single-worker application. Multiple application processes would
24
+need a distributed job coordinator before concurrent scheduling is enabled.
25
+
26
+## Tables
27
+
28
+* `global_opportunity_snapshot`: one immutable instrument row per cycle; versions, scores,
29
+ eligibility/suppression, rank, price/date, reasons, and full technical/sector/rule evidence.
30
+* `stock_recommendation_history`: append-only decisions, snapshot FK, fingerprint, reference price,
31
+ short/long levels, reasons, and frozen evidence.
32
+* `recommendation_current_state`: latest history FK, original recommendation anchor, separate
33
+ short/long lifecycle/action, price, target distances, and reasons.
34
+* `global_opportunity_top_selection`: immutable cycle publication with independent horizon picks.
35
+* `recommendation_backtest_run`: immutable inputs, metrics, examples, and per-horizon outcome status.
36
+
37
+Both PostgreSQL and SQLite reject UPDATE/DELETE of the immutable tables. Current-state updates
38
+do not mutate history. Complete JSON documents are stored as text alongside queryable columns,
39
+following a shared serialization contract; NULL numeric evidence remains SQL NULL / JSON null.
40
+
41
+## Recommendation policy: RECOMMENDATION_ENGINE_V1
42
+
43
+These are explicit starting rules, not calibrated investment forecasts or a second scoring engine.
44
+
45
+* A positive candidate requires existing rank eligibility, opportunity score ≥65, confidence ≥60,
46
+ and coverage ≥60. A qualified score ≥80 is a strong-buy candidate.
47
+* Short BUY also requires UPTREND, BREAKOUT, PULLBACK_IN_UPTREND, or REVERSAL_CANDIDATE technical state.
48
+* Long ACCUMULATE additionally requires available quality and valuation dimensions ≥60.
49
+* Unavailable dimensions are not zero and do not constitute deterioration.
50
+* Critical existing risk overrides, governance ≤20, or two available weak core dimensions ≤30
51
+ prompt long EXIT_REVIEW. One weak core dimension prompts REDUCE.
52
+* New investors and holders receive separate actions. Holder actions also reflect partial-profit,
53
+ reduction, and exit-review lifecycle conditions. No LLM participates.
54
+
55
+The fingerprint covers instrument/version, all actions, a rounded reference-price bucket,
56
+rounded score/confidence/coverage state, precise stored levels, and deterministic evidence content.
57
+Completion/cache clocks alone do not change it. Comparison with the latest fingerprint avoids
58
+identical history rows while allowing a genuine A→B→A change to be recorded.
59
+
60
+## Price ranges
61
+
62
+All levels are in the persisted technical price's currency. V1 uses only explicit evidence:
63
+
64
+* When `0 < support ≤ price < resistance`, short entry is support through
65
+ `min(price, support × 1.03)` and target 1 is resistance.
66
+* With positive ATR and support greater than ATR, target 2 is resistance + ATR;
67
+ invalidation is support − ATR.
68
+* An explicitly persisted positive fair value supports long accumulation at 80–90% of fair value.
69
+ Bull target and invalidation require their own valid evidence. A valuation score or analyst
70
+ consensus is never relabeled as fair value. The current upstream pipeline does not synthesize
71
+ intrinsic value, so these long levels can legitimately remain unavailable.
72
+* Unavailable levels are NULL and produce `PRICE_RANGE_EVIDENCE_INSUFFICIENT`.
73
+
74
+## Lifecycle
75
+
76
+Original recommendation levels remain anchored independently of later price/score changes.
77
+Within 3% of a target is TARGET_APPROACHING; hitting target 1 triggers SHORT PARTIAL_PROFIT and
78
+risk/reward-compression reasons. Target 2 proximity/reach has a separate reason. A valid long
79
+thesis can remain HOLD simultaneously. Crossing invalidation prompts SHORT EXIT / LONG EXIT_REVIEW;
80
+within 3% above invalidation is INVALIDATION_APPROACHING. Entry bands and proximity have separate
81
+states. Thesis deterioration takes precedence over profit conditions.
82
+
83
+No position membership is inferred from a recommendation's lifecycle: holder actions are a
84
+public recommendation for someone who already holds it, not an instruction executed on an account.
85
+
86
+## Top-N and dashboard
87
+
88
+`top_n` defaults to 4 and accepts 2–4. Short BUY and long ACCUMULATE/TOP_UP lists are selected
89
+separately after existing suppression gates. Each uses the existing opportunity ordering:
90
+score, confidence, coverage, Rule Engine score, then canonical UUID. Lists may contain fewer
91
+than N when fewer candidates qualify; they are never padded with fabricated picks.
92
+
93
+Dashboard reads persisted selection, snapshot, history, and current-state rows. It does not run
94
+the scanner, Rule Engine, recommendation engine, readiness ensure, or providers. UI membership
95
+annotations use the already-loaded portfolio/watchlist state **after** ranking. The generation
96
+timestamp and missing/stale areas expose the cycle's data status.
97
+
98
+## APIs and controlled DEV execution
99
+
100
+* `GET /api/v1/research/opportunities/current`
101
+* `GET /api/v1/research/opportunities/history/{instrumentId}`
102
+* `POST /api/v1/research/opportunities/cycles`
103
+* `GET /api/v1/research/backtesting/runs`
104
+* `POST /api/v1/research/backtesting/runs`
105
+
106
+The existing gateway already routes `/api/v1/research/**` to the research engine. GET routes are
107
+read-only. With persistence disabled, reads return empty results and write endpoints return 503.
108
+
109
+After research-service applies V13, submit a small canonical candidate set first:
110
+
111
+```json
112
+{"top_n": 2, "shortlist_limit": 2, "candidate_ids": ["<existing-canonical-NSE-UUID>"]}
113
+```
114
+
115
+Omit `candidate_ids` for a global canonical-universe run. The job does not fetch missing stock data;
116
+it can report no eligible results when the persisted evidence is insufficient. It is intentionally
117
+not invoked from a dashboard effect or GET. No broad live cycle was required for validation.
118
+
119
+Example backtest POST (UTC instants; end must not be in the future):
120
+
121
+```json
122
+{"start":"2026-01-01T00:00:00Z","end":"2026-09-01T00:00:00Z","market":"NSE","horizon":"SHORT_TERM"}
123
+```
124
+
125
+## Backtesting V1 and temporal guards
126
+
127
+This MVP evaluates **recorded historical recommendations**, not hypothetical recommendations for
128
+dates before the system had history. The selected date range defines the historical recommendation
129
+cohort; each row's generation timestamp is its evaluation date T. No current event/exposure tables
130
+are queried during a backtest.
131
+
132
+Frozen evidence must have publication/public-availability/discovery/retrieval/computation timestamps
133
+no later than T. News features use V12's `latest_known_features` selection and typed model validation,
134
+in addition to the recursive timestamp checks. Persisted price normalization checks both observation
135
+and retrieval timestamps for the entry. Daily closes retain their original retrieval clock, so later
136
+corrections cannot be used as an earlier entry. Future prices are outcome data only.
137
+
138
+Horizons are 7, 30, 91, 182, and 365 calendar days (1W/1M/3M/6M/1Y). The first available persisted
139
+close on/after the horizon, within seven days, supplies the outcome. Missing entry/future prices
140
+have explicit statuses and are excluded from return/hit-rate denominators. Metrics include cohort
141
+count, evaluated/missing counts, hit rate (>0), mean/median/best/worst returns, close-based maximum
142
+adverse excursion, and benchmark excess when explicit benchmark evidence exists. The optional
143
+`benchmark_id` selects a canonical persisted benchmark; no benchmark is guessed.
144
+
145
+Returns are unadjusted price returns, not simulated executions, dividend-adjusted total returns,
146
+or a transaction-cost model. The UI provides dates, market, horizon, run/selection, all five horizon
147
+metrics, and winner/loser examples without complex charts.
148
+
149
+## Validation notes
150
+
151
+Run focused Python tests in `tests/test_recommendation_lifecycle.py` with existing scanner,
152
+ranker, and orchestration tests. `test_recommendation_postgres.py` is opt-in using
153
+`RECOMMENDATION_TEST_PG_PORT` against an isolated `recommendation_validation` database already
154
+migrated by Flyway. The complete research-engine suite is also required.
155
+
156
+V12 uses PostgreSQL types and PL/pgSQL triggers, which the existing H2 test profile cannot execute.
157
+Run Maven/Flyway validation with disposable PostgreSQL datasource overrides and a second database
158
+for `dailyBarsUpgradeJdbcUrl`. The upgrade assertion now expects V10→V13 (three migrations).
159
+V1–V12 are unchanged. Frontend production builds and rendering tests run in Docker.
160
+
161
+The broader frontend suite has three existing source-assertion failures also reproduced with
162
+unchanged HEAD sources: visible-stock readiness navigation, shareholding drawer company lookup,
163
+and drawer provider/sector/industry projection. The new recommendation rendering tests pass.
164
+
165
+## Known follow-ups
166
+
167
+* Freshness tuning.
168
+* SHAREHOLDING behavior in global “Find required data”; acquisition policy is unchanged.
169
+* Recommendation threshold calibration.
170
+* Entry/target calibration and explicit fair-value evidence availability.
171
+* Prediction/ML.
172
+* Benchmark calibration, adjusted-return/corporate-action handling, and execution costs.
173
+* Distributed job coordination before scheduling concurrent application instances.
ai/research-engine/app/global_opportunity_cycle.py
new
+105
@@ -0,0 +1,105 @@
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)
ai/research-engine/app/global_opportunity_orchestration.py
+36
-11
@@ -25,6 +25,7 @@ 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):
@@ -51,6 +52,9 @@ class OpportunityEntry(ResearchBaseModel):
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):
@@ -73,6 +77,7 @@ class OpportunityRanking(ResearchBaseModel):
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:
@@ -96,7 +101,8 @@ class GlobalOpportunityOrchestrator:
101
102
async def run(self, canonical_instruments: Iterable[dict], *, as_of: datetime,
103
sector_contexts: Mapping[UUID, SectorContext] | None = None,
99
- shortlist_limit: int = 25, top_n: int = 10) -> OpportunityRanking:
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')
@@ -116,10 +122,18 @@ class GlobalOpportunityOrchestrator:
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}
120
- diagnostics, rules, successful = [], {}, []
134
+ diagnostics, rules, successful, temporal_evidence = [], {}, [], {}
135
evaluated = 0
122
- for candidate in shortlist:
136
+ for candidate in evaluation_candidates:
137
key = candidate.global_instrument_id
138
step = 'PUBLIC_EVIDENCE_UNAVAILABLE'
139
cache_hit = None
@@ -141,22 +155,27 @@ class GlobalOpportunityOrchestrator:
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))
147
- if ranked.rank_eligible:
148
- rules[key] = result
149
- successful.append(candidate)
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))
154
- ranked = self.ranker.rank(successful, rules)
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 = []
157
- for position, result in enumerate(ranked[:top_n], 1):
176
+ for position, result in enumerate([*ranked, *suppressed], 1):
177
key = result.global_instrument_id
159
- initial, enriched, rule = phase1[key], by_id[key], rules[key]
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,
@@ -169,8 +188,14 @@ class GlobalOpportunityOrchestrator:
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,
172
- opportunity_ranker_version=result.ranker_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,
176
- rank_eligible_count=len(ranked), top_n=entries, diagnostics=diagnostics)
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)
ai/research-engine/app/main.py
+63
@@ -79,6 +79,69 @@ app = FastAPI(title="Research Engine", version="0.3.0")
79
logger = logging.getLogger(__name__)
80
81
82
+@app.get('/api/v1/research/opportunities/current')
83
+async def opportunity_radar():
84
+ with repository._persistence_worker_lock:
85
+ return repository.persistence.opportunity_current()
86
+
87
+
88
+@app.get('/api/v1/research/opportunities/history/{instrument_id}')
89
+async def opportunity_history(instrument_id: UUID):
90
+ with repository._persistence_worker_lock:
91
+ return repository.persistence.recommendation_history(instrument_id)
92
+
93
+
94
+class OpportunityCycleRequest(BaseModel):
95
+ top_n: int = Field(default=4, ge=2, le=4)
96
+ shortlist_limit: int = Field(default=25, ge=1, le=100)
97
+ candidate_ids: list[UUID] | None = Field(default=None, max_length=100)
98
+
99
+
100
+@app.post('/api/v1/research/opportunities/cycles')
101
+async def opportunity_cycle(body: OpportunityCycleRequest, request: Request):
102
+ from app.global_opportunity_cycle import run_global_opportunity_cycle
103
+ if not hasattr(repository.persistence, 'publish_opportunity_cycle'):
104
+ raise HTTPException(503, 'RECOMMENDATION_PERSISTENCE_REQUIRED')
105
+ # One in-process cycle at a time; no awaits between history read and atomic publish.
106
+ import asyncio
107
+ if not hasattr(app.state, 'opportunity_cycle_lock'):
108
+ app.state.opportunity_cycle_lock = asyncio.Lock()
109
+ async with app.state.opportunity_cycle_lock:
110
+ return await run_global_opportunity_cycle(repository, portfolio_orchestrator,
111
+ top_n=body.top_n, shortlist_limit=body.shortlist_limit, candidate_ids=body.candidate_ids,
112
+ identity_headers={k: v for k, v in request.headers.items()
113
+ if k.lower() in {'authorization', 'x-user-id', 'x-correlation-id'}})
114
+
115
+
116
+class BacktestRequest(BaseModel):
117
+ start: str
118
+ end: str
119
+ market: str = 'NSE'
120
+ horizon: str = 'SHORT_TERM'
121
+ benchmark_id: UUID | None = None
122
+
123
+
124
+@app.get('/api/v1/research/backtesting/runs')
125
+async def backtest_runs():
126
+ with repository._persistence_worker_lock:
127
+ return repository.persistence.backtests()
128
+
129
+
130
+@app.post('/api/v1/research/backtesting/runs')
131
+async def create_backtest(body: BacktestRequest):
132
+ from app.recommendation_backtesting import run_backtest
133
+ if not hasattr(repository.persistence, 'save_backtest'):
134
+ raise HTTPException(503, 'RECOMMENDATION_PERSISTENCE_REQUIRED')
135
+ if body.market != 'NSE':
136
+ raise HTTPException(422, 'UNSUPPORTED_MARKET')
137
+ try:
138
+ with repository._persistence_worker_lock:
139
+ return run_backtest(repository.persistence, start=body.start, end=body.end,
140
+ horizon=body.horizon, benchmark_id=body.benchmark_id)
141
+ except ValueError as exc:
142
+ raise HTTPException(422, str(exc)) from exc
143
+
144
+
145
@app.middleware("http")
146
async def correlation_id_middleware(request: Request, call_next):
147
candidate = request.headers.get("X-Request-ID") or request.headers.get("X-Correlation-Id")
ai/research-engine/app/opportunity_persistence.py
new
+147
@@ -0,0 +1,147 @@
1
+"""V13 projection storage. All cycle writes publish atomically on the shared DB.
2
+
3
+History and snapshot methods only INSERT; projection methods alone use UPDATE.
4
+The payload preserves complete evidence alongside queryable identity/version columns.
5
+"""
6
+import json
7
+
8
+SNAPSHOT_FIELDS = {
9
+ **dict.fromkeys('scanner_version rule_engine_version technical_version sector_version ranker_version price_as_of'.split(), 'TEXT'),
10
+ **dict.fromkeys(('opportunity_score opportunity_confidence score_coverage rule_engine_score rank_position '
11
+ 'technical_score sector_score valuation_score quality_score growth_score balance_sheet_score '
12
+ 'quarterly_score catalyst_score news_score shareholding_score governance_score current_price').split(), 'REAL'),
13
+ 'rank_eligible': 'INTEGER',
14
+ **dict.fromkeys('suppression_reasons top_positive_reasons top_negative_reasons evidence_state'.split(), 'JSON')}
15
+HISTORY_FIELDS = {
16
+ **dict.fromkeys('rule_engine_version ranker_version new_investor_action existing_holder_action short_term_action long_term_action'.split(), 'TEXT'),
17
+ **dict.fromkeys(('price_at_recommendation opportunity_score confidence coverage short_entry_low short_entry_high '
18
+ 'short_target_1 short_target_2 short_invalidation long_entry_low long_entry_high '
19
+ 'long_fair_value long_target long_invalidation').split(), 'REAL'),
20
+ **dict.fromkeys('top_positive_reasons top_negative_reasons evidence_snapshot'.split(), 'JSON')}
21
+STATE_FIELDS = {
22
+ **dict.fromkeys('short_term_state long_term_state current_short_action current_long_action lifecycle_status'.split(), 'TEXT'),
23
+ **dict.fromkeys('price_at_recommendation current_price short_target_distance_pct long_target_distance_pct'.split(), 'REAL')}
24
+EXTRA_FIELDS = {'global_opportunity_snapshot': SNAPSHOT_FIELDS, 'stock_recommendation_history': HISTORY_FIELDS,
25
+ 'recommendation_current_state': STATE_FIELDS}
26
+
27
+SCHEMA = '''
28
+CREATE TABLE IF NOT EXISTS global_opportunity_snapshot (
29
+ snapshot_id TEXT PRIMARY KEY, cycle_id TEXT NOT NULL, global_instrument_id TEXT NOT NULL,
30
+ market TEXT NOT NULL, generated_at TEXT NOT NULL, payload TEXT NOT NULL,
31
+ UNIQUE(cycle_id, global_instrument_id)
32
+);
33
+CREATE INDEX IF NOT EXISTS opportunity_snapshot_instrument ON global_opportunity_snapshot(global_instrument_id, generated_at);
34
+CREATE TABLE IF NOT EXISTS stock_recommendation_history (
35
+ recommendation_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES global_opportunity_snapshot(snapshot_id),
36
+ global_instrument_id TEXT NOT NULL, generated_at TEXT NOT NULL,
37
+ recommendation_engine_version TEXT NOT NULL, fingerprint TEXT NOT NULL, payload TEXT NOT NULL
38
+);
39
+CREATE INDEX IF NOT EXISTS recommendation_history_instrument ON stock_recommendation_history(global_instrument_id, generated_at);
40
+CREATE TABLE IF NOT EXISTS recommendation_current_state (
41
+ global_instrument_id TEXT PRIMARY KEY,
42
+ latest_recommendation_id TEXT NOT NULL REFERENCES stock_recommendation_history(recommendation_id),
43
+ updated_at TEXT NOT NULL, payload TEXT NOT NULL
44
+);
45
+CREATE TABLE IF NOT EXISTS global_opportunity_top_selection (
46
+ cycle_id TEXT PRIMARY KEY, generated_at TEXT NOT NULL, market TEXT NOT NULL, payload TEXT NOT NULL
47
+);
48
+CREATE TABLE IF NOT EXISTS recommendation_backtest_run (
49
+ backtest_id TEXT PRIMARY KEY, generated_at TEXT NOT NULL, payload TEXT NOT NULL
50
+);
51
+'''
52
+
53
+for _table, _fields in EXTRA_FIELDS.items():
54
+ _start = SCHEMA.index('CREATE TABLE IF NOT EXISTS ' + _table)
55
+ _payload = SCHEMA.index('payload TEXT NOT NULL', _start)
56
+ SCHEMA = SCHEMA[:_payload] + ', '.join(k + ' ' + ('TEXT' if v == 'JSON' else v) for k, v in _fields.items()) + ', ' + SCHEMA[_payload:]
57
+
58
+for _table in ('global_opportunity_snapshot', 'stock_recommendation_history', 'global_opportunity_top_selection', 'recommendation_backtest_run'):
59
+ for _operation in ('UPDATE', 'DELETE'):
60
+ SCHEMA += f'''CREATE TRIGGER IF NOT EXISTS immutable_{_table}_{_operation.lower()}
61
+ BEFORE {_operation} ON {_table} BEGIN SELECT RAISE(ABORT, 'RECOMMENDATION_HISTORY_IS_IMMUTABLE'); END;\n'''
62
+
63
+
64
+def decode(row):
65
+ if row is None:
66
+ return None
67
+ value = row['payload']
68
+ return json.loads(value) if isinstance(value, str) else value
69
+
70
+
71
+class OpportunityPersistenceMixin:
72
+ def recommendation_history(self, instrument_id=None):
73
+ sql = 'SELECT payload FROM stock_recommendation_history'
74
+ params = ()
75
+ if instrument_id is not None:
76
+ sql += ' WHERE global_instrument_id = ?'
77
+ params = (str(instrument_id),)
78
+ return [decode(r) for r in self._connection.execute(sql + ' ORDER BY generated_at, recommendation_id', params).fetchall()]
79
+
80
+ def recommendation_states(self):
81
+ return [decode(r) for r in self._connection.execute(
82
+ 'SELECT payload FROM recommendation_current_state ORDER BY global_instrument_id').fetchall()]
83
+
84
+ def opportunity_current(self):
85
+ selection = decode(self._connection.execute(
86
+ 'SELECT payload FROM global_opportunity_top_selection ORDER BY generated_at DESC, cycle_id DESC LIMIT 1').fetchone())
87
+ if selection is None:
88
+ return dict(generated_at=None, best_buy_today=None, top_short_term=[], top_long_term=[], previous_recommendations=[])
89
+ snapshots = {s['global_instrument_id']: s for s in self.opportunity_snapshots(selection['cycle_id'])}
90
+ states = {s['global_instrument_id']: s for s in self.recommendation_states()}
91
+ def hydrate(card):
92
+ key = card['global_instrument_id']
93
+ state = states.get(key, {})
94
+ latest_id = state.get('latest_recommendation_id')
95
+ history = decode(self._connection.execute(
96
+ 'SELECT payload FROM stock_recommendation_history WHERE recommendation_id = ?', (latest_id,)).fetchone()) if latest_id else None
97
+ # Current public score/price come from the snapshot/projection, not an older recommendation.
98
+ current = snapshots.get(key, {})
99
+ scores = {k: current[k] for k in ('opportunity_score', 'opportunity_confidence', 'score_coverage', 'rule_engine_score') if k in current}
100
+ return {**card, **current, **(history or {}), **scores, **state}
101
+ for field in ('top_short_term', 'top_long_term', 'previous_recommendations'):
102
+ selection[field] = [hydrate(c) for c in selection[field]]
103
+ if selection['best_buy_today']:
104
+ selection['best_buy_today'] = hydrate(selection['best_buy_today'])
105
+ return selection
106
+
107
+ def opportunity_snapshots(self, cycle_id):
108
+ return [decode(r) for r in self._connection.execute(
109
+ 'SELECT payload FROM global_opportunity_snapshot WHERE cycle_id = ? ORDER BY global_instrument_id', (cycle_id,)).fetchall()]
110
+
111
+ def _insert_opportunity(self, table, columns, value):
112
+ # Table/column names are internal constants, never caller SQL.
113
+ extra = EXTRA_FIELDS.get(table, {})
114
+ columns = (*columns, *extra)
115
+ def stored(c):
116
+ v = value.get(c)
117
+ return json.dumps(v, sort_keys=True) if extra.get(c) == 'JSON' else int(v) if isinstance(v, bool) else v
118
+ self._connection.execute(f"INSERT INTO {table} ({','.join(columns)},payload) VALUES ({','.join('?' for _ in range(len(columns)+1))})",
119
+ tuple(stored(c) for c in columns) + (json.dumps(value, sort_keys=True, allow_nan=False),))
120
+
121
+ def publish_opportunity_cycle(self, snapshots, recommendations, states, selection, *, build=None):
122
+ with self._connection:
123
+ for s in snapshots:
124
+ self._insert_opportunity('global_opportunity_snapshot',
125
+ ('snapshot_id', 'cycle_id', 'global_instrument_id', 'market', 'generated_at'), s)
126
+ if build is not None:
127
+ persisted = self.opportunity_snapshots(selection['cycle_id'])
128
+ recommendations, states, selection = build(persisted)
129
+ for r in recommendations:
130
+ self._insert_opportunity('stock_recommendation_history',
131
+ ('recommendation_id', 'snapshot_id', 'global_instrument_id', 'generated_at', 'recommendation_engine_version', 'fingerprint'), r)
132
+ for s in states:
133
+ cols = ('global_instrument_id', 'latest_recommendation_id', 'updated_at', *STATE_FIELDS, 'payload')
134
+ self._connection.execute(f"INSERT INTO recommendation_current_state ({','.join(cols)}) VALUES ({','.join('?' for _ in cols)}) "
135
+ + 'ON CONFLICT(global_instrument_id) DO UPDATE SET ' + ','.join(f'{c}=excluded.{c}' for c in cols[1:]),
136
+ tuple(s.get(c) for c in cols[:-1]) + (json.dumps(s, sort_keys=True),))
137
+ self._insert_opportunity('global_opportunity_top_selection', ('cycle_id', 'generated_at', 'market'), selection)
138
+ return selection
139
+
140
+ def save_backtest(self, result):
141
+ with self._connection:
142
+ self._insert_opportunity('recommendation_backtest_run', ('backtest_id', 'generated_at'), result)
143
+ return result
144
+
145
+ def backtests(self):
146
+ return [decode(r) for r in self._connection.execute(
147
+ 'SELECT payload FROM recommendation_backtest_run ORDER BY generated_at DESC, backtest_id DESC').fetchall()]
ai/research-engine/app/persistence.py
+11
-1
@@ -121,6 +121,13 @@ class ResearchPersistence(Protocol):
121
122
123
class DisabledResearchPersistence:
124
+ def opportunity_current(self):
125
+ return dict(generated_at=None, best_buy_today=None, top_short_term=[], top_long_term=[], previous_recommendations=[])
126
+
127
+ def recommendation_history(self, instrument_id=None): return []
128
+ def recommendation_states(self): return []
129
+ def backtests(self): return []
130
+
131
def __init__(self) -> None:
132
self._stock_rule_engine_results: dict[tuple[str, str, str], dict[str, Any]] = {}
133
def load_documents(self) -> list[ResearchDocument]:
@@ -183,9 +190,10 @@ class DisabledResearchPersistence:
190
191
192
from app.news_persistence import NewsPersistenceMixin, sqlite_schema as news_sqlite_schema
193
+from app.opportunity_persistence import OpportunityPersistenceMixin, SCHEMA as OPPORTUNITY_SCHEMA
194
195
188
-class SqliteResearchPersistence(NewsPersistenceMixin):
196
+class SqliteResearchPersistence(NewsPersistenceMixin, OpportunityPersistenceMixin):
197
def __init__(self, database_path: str | Path = ":memory:") -> None:
198
self.database_path = str(database_path)
199
self._connection = sqlite3.connect(self.database_path)
@@ -197,6 +205,8 @@ class SqliteResearchPersistence(NewsPersistenceMixin):
205
self._connection.executescript(_sqlite_schema())
206
self._connection.commit()
207
news_sqlite_schema(self._connection)
208
+ self._connection.executescript(OPPORTUNITY_SCHEMA)
209
+ self._connection.commit()
210
211
def upsert_daily_market_bar(self, bar: DailyMarketBar) -> None:
212
self.upsert_daily_market_bars([bar])
ai/research-engine/app/postgres_persistence.py
+5
@@ -48,6 +48,11 @@ class PostgresResearchPersistence(SqliteResearchPersistence):
48
self._connection.execute("SELECT 1 FROM company_business_exposure_profiles LIMIT 0")
49
self._connection.execute("SELECT 1 FROM research_news_search_runs LIMIT 0")
50
self._connection.execute("SELECT 1 FROM research_event_impact_features LIMIT 0")
51
+ self._connection.execute("SELECT 1 FROM global_opportunity_snapshot LIMIT 0")
52
+ self._connection.execute("SELECT 1 FROM stock_recommendation_history LIMIT 0")
53
+ self._connection.execute("SELECT 1 FROM recommendation_current_state LIMIT 0")
54
+ self._connection.execute("SELECT 1 FROM global_opportunity_top_selection LIMIT 0")
55
+ self._connection.execute("SELECT 1 FROM recommendation_backtest_run LIMIT 0")
56
self._connection.execute("SELECT 1 FROM global_stock_rule_engine_results LIMIT 0")
57
self._connection.execute("SELECT 1 FROM market_trading_schedules LIMIT 0")
58
self._connection.execute("SELECT 1 FROM market_trading_calendar_exceptions LIMIT 0")
ai/research-engine/app/recommendation_backtesting.py
new
+135
@@ -0,0 +1,135 @@
1
+"""Recorded-recommendation cohort backtesting; no retrospective evidence reconstruction.
2
+
3
+Entry is the first persisted price known at evaluation T. An immutable recommendation
4
+must itself have existed by T. This conservative boundary also excludes V12 features
5
+computed/discovered after T; current exposure/event tables are never queried.
6
+"""
7
+from datetime import datetime, timedelta, timezone, time
8
+from statistics import mean, median
9
+from uuid import UUID, uuid4
10
+from zoneinfo import ZoneInfo
11
+from app.models import MarketPriceObservation
12
+from app.technical_features import normalize_price_history
13
+from app.news_intelligence import EventImpactFeature, latest_known_features
14
+
15
+HORIZONS = {'1W': 7, '1M': 30, '3M': 91, '6M': 182, '1Y': 365}
16
+
17
+
18
+def utc(value):
19
+ result = datetime.fromisoformat(value.replace('Z', '+00:00')) if isinstance(value, str) else value
20
+ if result.tzinfo is None:
21
+ raise ValueError('AWARE_DATE_REQUIRED')
22
+ return result.astimezone(timezone.utc)
23
+
24
+
25
+def evidence_available(value, at):
26
+ """Conservative recursive guard for persisted availability/provenance timestamps."""
27
+ if isinstance(value, dict):
28
+ if 'news_features' in value:
29
+ try:
30
+ features = [EventImpactFeature.model_validate(f) for f in value['news_features']]
31
+ if len(latest_known_features(features, at)) != len(features):
32
+ return False
33
+ except (ValueError, TypeError):
34
+ return False
35
+ for key, item in value.items():
36
+ normalized = key.replace('_', '').lower()
37
+ if normalized in {'publishedat', 'publicationtime', 'publicavailabilityat', 'publicavailableat',
38
+ 'discoveredat', 'computedat', 'calculatedat', 'retrievedat', 'asof', 'inputasof'} and item:
39
+ try:
40
+ if utc(item) > at:
41
+ return False
42
+ except (ValueError, TypeError, AttributeError):
43
+ return False
44
+ if not evidence_available(item, at):
45
+ return False
46
+ elif isinstance(value, list):
47
+ return all(evidence_available(item, at) for item in value)
48
+ return True
49
+
50
+
51
+def metrics(samples):
52
+ returns = [s['return_pct'] for s in samples if s['return_pct'] is not None]
53
+ adverse = [s['max_adverse_excursion_pct'] for s in samples if s['max_adverse_excursion_pct'] is not None]
54
+ excess = [s['benchmark_excess_return_pct'] for s in samples if s['benchmark_excess_return_pct'] is not None]
55
+ return dict(recommendation_count=len(samples), evaluated_count=len(returns), missing_count=len(samples)-len(returns),
56
+ hit_rate=100 * sum(v > 0 for v in returns)/len(returns) if returns else None,
57
+ average_return=mean(returns) if returns else None, median_return=median(returns) if returns else None,
58
+ best_return=max(returns) if returns else None, worst_return=min(returns) if returns else None,
59
+ max_adverse_excursion=min(adverse) if adverse else None,
60
+ benchmark_excess_return=mean(excess) if excess else None)
61
+
62
+
63
+def evaluate_backtest(history, prices, *, start, end, horizon, now, benchmark_prices=None):
64
+ start, end, now = utc(start), utc(end), utc(now)
65
+ if start > end or end > now or horizon not in {'SHORT_TERM', 'LONG_TERM'}:
66
+ raise ValueError('INVALID_BACKTEST_PARAMETERS')
67
+ actions = {'BUY'} if horizon == 'SHORT_TERM' else {'ACCUMULATE', 'TOP_UP'}
68
+ field = 'short_term_action' if horizon == 'SHORT_TERM' else 'long_term_action'
69
+ cohort, excluded = [], 0
70
+ for r in history:
71
+ t = utc(r['generated_at'])
72
+ if not start <= t <= end or r.get('market') != 'NSE' or r[field] not in actions:
73
+ continue
74
+ if not evidence_available(r.get('evidence_snapshot', {}), t):
75
+ excluded += 1
76
+ continue
77
+ cohort.append(r)
78
+ output = {h: [] for h in HORIZONS}
79
+ for r in cohort:
80
+ t, key = utc(r['generated_at']), UUID(r['global_instrument_id'])
81
+ rows = prices.get(key, [])
82
+ known = normalize_price_history(key, rows, as_of=t)
83
+ entry = known.observations[-1] if known.observations and not known.current_conflict else None
84
+ observed = normalize_price_history(key, rows, as_of=now, currency=entry.currency if entry else None)
85
+ for h, days in HORIZONS.items():
86
+ target = t + timedelta(days=days)
87
+ future = [p for p in observed.observations if target <= utc(p.observed_at) <= target + timedelta(days=7)] if target <= now else []
88
+ exit_price = future[0] if future else None
89
+ value = (float(exit_price.price / entry.price) - 1) * 100 if entry and exit_price else None
90
+ window = [float(p.price / entry.price - 1) * 100 for p in observed.observations
91
+ if entry and exit_price and t < utc(p.observed_at) <= utc(exit_price.observed_at)]
92
+ excess = None
93
+ if benchmark_prices and value is not None:
94
+ bkey = benchmark_prices[0].instrument_id
95
+ before = normalize_price_history(bkey, benchmark_prices, as_of=t)
96
+ after = normalize_price_history(bkey, benchmark_prices, as_of=now)
97
+ matching = [p for p in after.observations if utc(p.observed_at).date() == utc(exit_price.observed_at).date()]
98
+ if before.observations and not before.current_conflict and matching:
99
+ excess = value - (float(matching[0].price / before.observations[-1].price) - 1) * 100
100
+ output[h].append(dict(recommendation_id=r['recommendation_id'], global_instrument_id=str(key),
101
+ symbol=r.get('symbol'), company_name=r.get('company_name'),
102
+ evaluation_date=t.isoformat(), return_pct=value,
103
+ max_adverse_excursion_pct=min([0, *window]) if window else None,
104
+ benchmark_excess_return_pct=excess,
105
+ status='AVAILABLE' if value is not None else 'ENTRY_PRICE_UNAVAILABLE' if not entry else 'FUTURE_PRICE_UNAVAILABLE'))
106
+ examples = sorted([s for s in output['1M' if horizon == 'SHORT_TERM' else '1Y'] if s['return_pct'] is not None], key=lambda s: s['return_pct'])
107
+ return dict(backtest_id=str(uuid4()), generated_at=now.isoformat(), engine_version='BACKTESTING_V1',
108
+ start=start.isoformat(), end=end.isoformat(), market='NSE', horizon=horizon,
109
+ recommendation_count=len(cohort), excluded_unavailable_evidence=excluded,
110
+ methodology='Recorded recommendations; calendar horizons, first available close within 7 days; unadjusted price returns; close-based MAE.',
111
+ metrics={h: metrics(v) for h, v in output.items()}, samples=output,
112
+ winners=[s for s in reversed(examples) if s['return_pct'] > 0][:4],
113
+ losers=[s for s in examples if s['return_pct'] <= 0][:4])
114
+
115
+
116
+def run_backtest(persistence, *, start, end, horizon, benchmark_id=None):
117
+ history = persistence.recommendation_history()
118
+ keys = {UUID(r['global_instrument_id']) for r in history}
119
+ rows = persistence.load_market_price_observations(keys) if keys else []
120
+ prices = {key: [] for key in keys}
121
+ for row in rows:
122
+ prices[row.instrument_id].append(row)
123
+ # Preserve each bar's retrieval clock: a later correction cannot become an entry at T.
124
+ # Daily and quote conflicts are still resolved by the existing history normalizer.
125
+ for bar in persistence.load_daily_market_bars(keys) if keys else []:
126
+ if bar.close is None or bar.close <= 0 or bar.source_mode != 'REAL':
127
+ continue
128
+ stamp = datetime.combine(bar.trading_date, time(15, 30), ZoneInfo('Asia/Kolkata')).astimezone(timezone.utc)
129
+ prices[bar.global_instrument_id].append(MarketPriceObservation(
130
+ instrument_id=bar.global_instrument_id, observed_at=stamp, retrieved_at=bar.retrieved_at,
131
+ price=bar.close, currency=bar.currency, provider=bar.provider, source_url=bar.source_url))
132
+ benchmark = persistence.load_market_price_observations({benchmark_id}) if benchmark_id else None
133
+ result = evaluate_backtest(history, prices, start=start, end=end, horizon=horizon,
134
+ now=datetime.now(timezone.utc), benchmark_prices=benchmark)
135
+ return persistence.save_backtest(result)
ai/research-engine/app/recommendation_engine.py
new
+180
@@ -0,0 +1,180 @@
1
+"""Deterministic recommendation policy, independent of membership and public scoring.
2
+
3
+V1 thresholds are deliberately explicit initial policy, not calibrated forecasts.
4
+All prices use the persisted technical snapshot's currency and unadjusted basis.
5
+"""
6
+from copy import deepcopy
7
+from hashlib import sha256
8
+import json
9
+from math import isfinite
10
+
11
+VERSION = 'RECOMMENDATION_ENGINE_V1'
12
+BUY_SCORE, STRONG_SCORE, MIN_CONFIDENCE, MIN_COVERAGE = 65, 80, 60, 60
13
+RANGE_KEYS = ('short_entry_low short_entry_high short_target_1 short_target_2 short_invalidation '
14
+ 'long_entry_low long_entry_high long_fair_value long_target long_invalidation').split()
15
+AREA_NAMES = {'QUALITY': 'FUNDAMENTAL_BUSINESS_QUALITY', 'QUARTERLY': 'QUARTERLY_EARNINGS_TREND',
16
+ 'CATALYST': 'ORDER_BOOK_CAPACITY_CATALYSTS', 'NEWS': 'NEWS_GEOPOLITICAL_EVENTS',
17
+ 'GOVERNANCE': 'MANAGEMENT_GOVERNANCE'}
18
+
19
+
20
+def area_scores(rule):
21
+ raw = {a['area']: a.get('raw_score') for a in rule.get('area_scores', [])}
22
+ return {**raw, **{k: raw.get(v) for k, v in AREA_NAMES.items()}}
23
+
24
+
25
+def number(value):
26
+ return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) and isfinite(value) else None
27
+
28
+
29
+def digest(value):
30
+ return sha256(json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()).hexdigest()
31
+
32
+
33
+def ranges(snapshot):
34
+ result = dict.fromkeys(RANGE_KEYS)
35
+ technical = snapshot['evidence_state'].get('technical', {})
36
+ p, support, resistance, atr = (number(v) for v in (
37
+ snapshot.get('current_price'), technical.get('support_level'),
38
+ technical.get('resistance_level'), technical.get('atr14')))
39
+ if p and support and resistance and 0 < support <= p < resistance:
40
+ result.update(short_entry_low=support, short_entry_high=min(p, support * 1.03),
41
+ short_target_1=resistance)
42
+ if atr and atr > 0 and support > atr:
43
+ result.update(short_target_2=resistance + atr, short_invalidation=support - atr)
44
+ # Only explicitly supplied fair-value evidence is usable; never turn a valuation score into a price.
45
+ valuation = snapshot['evidence_state'].get('valuation', {})
46
+ fair, bull, invalidation = (number(valuation.get(k)) for k in ('fair_value', 'bull_target', 'invalidation'))
47
+ if fair and fair > 0:
48
+ result.update(long_entry_low=fair * .8, long_entry_high=fair * .9, long_fair_value=fair)
49
+ if bull and bull >= fair:
50
+ result['long_target'] = bull
51
+ if invalidation and 0 < invalidation < fair * .8:
52
+ result['long_invalidation'] = invalidation
53
+ return {k: round(v, 4) if v is not None else None for k, v in result.items()}
54
+
55
+
56
+class RecommendationEngineV1:
57
+ def evaluate(self, snapshot, previous=None):
58
+ evidence = snapshot['evidence_state']
59
+ rule = evidence.get('rule', {})
60
+ dimensions = area_scores(rule)
61
+ critical = any(r.get('severity') == 'CRITICAL' for r in rule.get('risk_overrides', []))
62
+ weak = [a for a in ('QUALITY', 'GROWTH', 'BALANCE_SHEET', 'QUARTERLY')
63
+ if dimensions.get(a) is not None and dimensions[a] <= 30]
64
+ governance = dimensions.get('GOVERNANCE')
65
+ thesis_broken = critical or (governance is not None and governance <= 20) or len(weak) >= 2
66
+ score = snapshot.get('opportunity_score')
67
+ qualified = (snapshot['rank_eligible'] and score is not None and score >= BUY_SCORE
68
+ and snapshot['opportunity_confidence'] >= MIN_CONFIDENCE
69
+ and snapshot['score_coverage'] >= MIN_COVERAGE and not thesis_broken)
70
+ technical = evidence.get('technical', {}).get('technical_state')
71
+ short_buy = qualified and technical in {'UPTREND', 'BREAKOUT', 'PULLBACK_IN_UPTREND', 'REVERSAL_CANDIDATE'}
72
+ long_buy = qualified and all(dimensions.get(a) is not None and dimensions[a] >= 60
73
+ for a in ('QUALITY', 'VALUATION'))
74
+ result = {k: deepcopy(snapshot.get(k)) for k in (
75
+ 'global_instrument_id', 'market', 'symbol', 'company_name', 'generated_at', 'rule_engine_version', 'ranker_version',
76
+ 'opportunity_score', 'top_positive_reasons', 'top_negative_reasons')}
77
+ result.update(recommendation_engine_version=VERSION, price_at_recommendation=snapshot.get('current_price'),
78
+ confidence=snapshot['opportunity_confidence'], coverage=snapshot['score_coverage'],
79
+ new_investor_action='AVOID' if thesis_broken else ('STRONG_BUY_CANDIDATE' if score >= STRONG_SCORE else 'BUY_CANDIDATE') if qualified else 'WATCH_WAIT',
80
+ existing_holder_action='EXIT_REVIEW' if thesis_broken else 'TOP_UP' if long_buy else 'HOLD_NO_NEW_MONEY' if not qualified else 'HOLD',
81
+ short_term_action='EXIT' if thesis_broken else 'BUY' if short_buy else 'WAIT',
82
+ long_term_action='EXIT_REVIEW' if thesis_broken else 'ACCUMULATE' if long_buy else 'REDUCE' if weak else 'HOLD',
83
+ evidence_snapshot=deepcopy(evidence), **ranges(snapshot))
84
+ reasons = result['top_negative_reasons'] or []
85
+ if any(result[k] is None for k in RANGE_KEYS):
86
+ reasons = [*reasons, 'PRICE_RANGE_EVIDENCE_INSUFFICIENT']
87
+ if thesis_broken:
88
+ reasons = [*reasons, 'LONG_TERM_THESIS_BROKEN']
89
+ result['top_negative_reasons'] = sorted(set(reasons))
90
+ if previous:
91
+ comparison = lifecycle(previous, result, snapshot.get('current_price'), {'review': True})
92
+ result['short_term_action'] = comparison['current_short_action']
93
+ result['long_term_action'] = comparison['current_long_action']
94
+ if comparison['current_long_action'] == 'EXIT_REVIEW':
95
+ result['existing_holder_action'] = 'EXIT_REVIEW'
96
+ elif comparison['current_long_action'] == 'REDUCE' or comparison['current_short_action'] == 'EXIT':
97
+ result['existing_holder_action'] = 'REDUCE'
98
+ elif comparison['current_short_action'] == 'PARTIAL_PROFIT':
99
+ result['existing_holder_action'] = 'PARTIAL_PROFIT'
100
+ if result['long_term_action'] != 'TOP_UP':
101
+ result['new_investor_action'] = 'WATCH_WAIT'
102
+ result['top_negative_reasons'] = sorted(set(result['top_negative_reasons'] + comparison['lifecycle_reasons']))
103
+ # Cache clocks, completion timestamps and observed price drift are not evidence changes.
104
+ evidence_key = {'references': rule.get('evidence_references', []), 'dimensions': dimensions,
105
+ 'metrics': [a.get('metrics', []) for a in rule.get('area_scores', [])],
106
+ 'risk': rule.get('risk_overrides', []), 'technical_state': technical,
107
+ 'sector_state': evidence.get('sector', {}).get('sector_state'),
108
+ 'missing': evidence.get('missing_inputs', []), 'stale': evidence.get('stale_inputs', [])}
109
+ p = result['price_at_recommendation']
110
+ result['fingerprint'] = digest({
111
+ 'instrument': result['global_instrument_id'], 'engine': VERSION,
112
+ 'actions': [result[k] for k in ('new_investor_action', 'existing_holder_action', 'short_term_action', 'long_term_action')],
113
+ 'reference_bucket': round(p, -1) if p is not None else None,
114
+ 'ranges': {k: result[k] for k in RANGE_KEYS}, 'evidence': evidence_key,
115
+ 'score_state': [round(result[k] / 5) * 5 if result[k] is not None else None for k in ('opportunity_score', 'confidence', 'coverage')]})
116
+ return result
117
+
118
+
119
+def lifecycle(anchor, recommendation, price, previous_state=None):
120
+ """Compare with the original trade levels. Never overwrite the historical anchor."""
121
+ reasons = []
122
+ short, long = recommendation['short_term_action'], recommendation['long_term_action']
123
+ ss = ls = 'NEW'
124
+ p = number(price)
125
+ for horizon in ('short', 'long'):
126
+ invalidation = anchor.get(f'{horizon}_invalidation')
127
+ low, high = anchor.get(f'{horizon}_entry_low'), anchor.get(f'{horizon}_entry_high')
128
+ target = anchor.get('short_target_1' if horizon == 'short' else 'long_target')
129
+ state = 'HOLDING' if previous_state else 'NEW'
130
+ if p is not None:
131
+ if invalidation and p <= invalidation:
132
+ state = 'INVALIDATED'
133
+ elif invalidation and p <= invalidation * 1.03:
134
+ state = 'INVALIDATION_APPROACHING'
135
+ elif target and p >= target:
136
+ state = 'TARGET_REACHED'
137
+ elif target and p >= target * .97:
138
+ state = 'TARGET_APPROACHING'
139
+ elif low is not None and high is not None and low <= p <= high:
140
+ state = 'IN_ENTRY_ZONE'
141
+ elif high and high < p <= high * 1.03:
142
+ state = 'ENTRY_APPROACHING'
143
+ if horizon == 'short':
144
+ ss = state
145
+ if state == 'INVALIDATED':
146
+ short = 'EXIT'
147
+ reasons.append('SHORT_INVALIDATION_REACHED')
148
+ elif state == 'TARGET_REACHED':
149
+ ss, short = 'PARTIAL_PROFIT', 'PARTIAL_PROFIT'
150
+ reasons.extend(['TARGET_1_REACHED', 'SHORT_TERM_RISK_REWARD_COMPRESSED'])
151
+ if anchor.get('short_target_2') and p >= anchor['short_target_2'] * .97:
152
+ reasons.append('TARGET_2_REACHED' if p >= anchor['short_target_2'] else 'TARGET_2_APPROACHING')
153
+ elif (previous_state and short == 'WAIT'
154
+ and anchor.get('short_term_action') in {'BUY', 'HOLD', 'PARTIAL_PROFIT'}):
155
+ short = 'HOLD'
156
+ else:
157
+ ls = state
158
+ if state == 'INVALIDATED':
159
+ long, ls = 'EXIT_REVIEW', 'INVALIDATED'
160
+ elif state in {'TARGET_REACHED', 'TARGET_APPROACHING'} and long not in {'REDUCE', 'EXIT_REVIEW'}:
161
+ long = 'HOLD'
162
+ elif (previous_state and long == 'ACCUMULATE' and p is not None
163
+ and anchor.get('price_at_recommendation') and p <= anchor['price_at_recommendation'] * .97):
164
+ long = 'TOP_UP'
165
+ if recommendation['long_term_action'] == 'EXIT_REVIEW':
166
+ long, ls = 'EXIT_REVIEW', 'EXIT_REVIEW'
167
+ reasons.append('LONG_TERM_THESIS_BROKEN')
168
+ elif long == 'REDUCE':
169
+ ls = 'THESIS_WEAKENING'
170
+ if short == 'PARTIAL_PROFIT' and long in {'TOP_UP', 'ACCUMULATE'}:
171
+ long = 'HOLD'
172
+ if recommendation['short_term_action'] == 'EXIT':
173
+ short, ss = 'EXIT', 'EXIT_REVIEW'
174
+ def distance(key):
175
+ return round((anchor[key] / p - 1) * 100, 4) if p and anchor.get(key) else None
176
+ return dict(short_term_state=ss, long_term_state=ls, current_short_action=short,
177
+ current_long_action=long, lifecycle_status=ls if ls in {'EXIT_REVIEW', 'INVALIDATED', 'THESIS_WEAKENING'} else ss,
178
+ price_at_recommendation=anchor.get('price_at_recommendation'), current_price=p,
179
+ short_target_distance_pct=distance('short_target_2'), long_target_distance_pct=distance('long_target'),
180
+ lifecycle_reasons=reasons)
ai/research-engine/tests/test_recommendation_lifecycle.py
new
+260
@@ -0,0 +1,260 @@
1
+from copy import deepcopy
2
+from datetime import timedelta
3
+from decimal import Decimal
4
+from uuid import UUID, uuid4
5
+from unittest.mock import AsyncMock
6
+import pytest
7
+
8
+from app.persistence import SqliteResearchPersistence
9
+from app.recommendation_engine import RecommendationEngineV1, lifecycle, ranges, RANGE_KEYS
10
+from app.global_opportunity_cycle import snapshot_from_entry, prepare_cycle, nse_equities
11
+from app.recommendation_backtesting import evaluate_backtest, evidence_available
12
+from app.models import MarketPriceObservation
13
+from test_global_scanner import NOW, instrument
14
+from test_global_opportunity_orchestration import setup
15
+
16
+
17
+def snapshot(n=1):
18
+ return dict(snapshot_id=str(uuid4()), cycle_id=str(uuid4()), global_instrument_id=str(UUID(int=n)),
19
+ generated_at=NOW.isoformat(), market='NSE', current_price=1000., opportunity_score=80.,
20
+ opportunity_confidence=80., score_coverage=90., rule_engine_score=80., rank_eligible=True,
21
+ rule_engine_version='STOCK_RULE_ENGINE_V1', ranker_version='GLOBAL_OPPORTUNITY_RANKER_V1',
22
+ top_positive_reasons=['SUPPORT:QUALITY'], top_negative_reasons=[], symbol=f'C{n}',
23
+ evidence_state={'technical': {'technical_state': 'UPTREND', 'support_level': 980., 'resistance_level': 1150., 'atr14': 50.},
24
+ 'rule': {'area_scores': [{'area': a, 'raw_score': 80} for a in ('FUNDAMENTAL_BUSINESS_QUALITY', 'VALUATION', 'GROWTH', 'BALANCE_SHEET')]}})
25
+
26
+
27
+def publish(store, snapshots, top_n=4):
28
+ cid = str(uuid4())
29
+ for s in snapshots: s['cycle_id'] = cid
30
+ history, states, selection = prepare_cycle(store, snapshots, cycle_id=cid, now=snapshots[0]['generated_at'], top_n=top_n)
31
+ store.publish_opportunity_cycle(snapshots, history, states, selection)
32
+ return selection
33
+
34
+
35
+def test_actions_independent_and_missing_not_negative():
36
+ s = snapshot()
37
+ r = RecommendationEngineV1().evaluate(s)
38
+ assert (r['short_term_action'], r['long_term_action']) == ('BUY', 'ACCUMULATE')
39
+ s['evidence_state']['technical']['technical_state'] = 'OVEREXTENDED'
40
+ r = RecommendationEngineV1().evaluate(s)
41
+ assert (r['short_term_action'], r['long_term_action']) == ('WAIT', 'ACCUMULATE')
42
+ s['opportunity_score'] = None
43
+ r = RecommendationEngineV1().evaluate(s)
44
+ assert (r['new_investor_action'], r['existing_holder_action']) == ('WATCH_WAIT', 'HOLD_NO_NEW_MONEY')
45
+ s['evidence_state']['rule']['area_scores'] = []
46
+ assert RecommendationEngineV1().evaluate(s)['long_term_action'] == 'HOLD'
47
+ s['evidence_state']['rule']['area_scores'] = [{'area': a, 'raw_score': 0} for a in ('GROWTH', 'BALANCE_SHEET')]
48
+ assert RecommendationEngineV1().evaluate(s)['long_term_action'] == 'EXIT_REVIEW'
49
+
50
+
51
+def test_wait_recommendation_does_not_churn_history():
52
+ store = SqliteResearchPersistence()
53
+ s = snapshot()
54
+ s['opportunity_score'] = None
55
+ publish(store, [s])
56
+ s = deepcopy(s)
57
+ s.update(snapshot_id=str(uuid4()), generated_at=(NOW+timedelta(minutes=1)).isoformat())
58
+ publish(store, [s])
59
+ assert len(store.recommendation_history()) == 1
60
+
61
+
62
+def test_ranges_and_determinism():
63
+ s = snapshot()
64
+ r = RecommendationEngineV1().evaluate(s)
65
+ assert r == RecommendationEngineV1().evaluate(deepcopy(s))
66
+ assert r['short_target_1'] == 1150 and r['short_target_2'] == 1200
67
+ assert r['long_fair_value'] is None and 'PRICE_RANGE_EVIDENCE_INSUFFICIENT' in r['top_negative_reasons']
68
+ s['evidence_state']['technical'] = {}
69
+ assert all(v is None for v in ranges(s).values())
70
+ s['evidence_state']['valuation'] = {'fair_value': 1500., 'bull_target': 1800., 'invalidation': 900.}
71
+ assert ranges(s)['long_entry_low'] == 1200
72
+
73
+
74
+@pytest.mark.parametrize('price,state,action', [(1120, 'TARGET_APPROACHING', 'BUY'),
75
+ (1150, 'PARTIAL_PROFIT', 'PARTIAL_PROFIT'), (1185, 'PARTIAL_PROFIT', 'PARTIAL_PROFIT'),
76
+ (930, 'INVALIDATED', 'EXIT'), (950, 'INVALIDATION_APPROACHING', 'BUY'),
77
+ (990, 'IN_ENTRY_ZONE', 'BUY'), (1010, 'ENTRY_APPROACHING', 'BUY')])
78
+def test_lifecycle(price, state, action):
79
+ r = RecommendationEngineV1().evaluate(snapshot())
80
+ r['long_term_action'] = 'HOLD'
81
+ actual = lifecycle(r, r, price, {'short_term_state': 'NEW'})
82
+ assert actual['short_term_state'] == state and actual['current_short_action'] == action
83
+ assert actual['current_long_action'] == 'HOLD'
84
+ if price == 1185:
85
+ assert 'TARGET_2_APPROACHING' in actual['lifecycle_reasons']
86
+
87
+
88
+def test_thesis_break_exit_review_overrides_targets():
89
+ s = snapshot()
90
+ prior = RecommendationEngineV1().evaluate(s)
91
+ s['evidence_state']['rule']['risk_overrides'] = [{'severity': 'CRITICAL', 'code': 'VALIDATED_GOVERNANCE_RISK'}]
92
+ now = RecommendationEngineV1().evaluate(s)
93
+ state = lifecycle(prior, now, 1185, {})
94
+ assert state['current_long_action'] == 'EXIT_REVIEW' and state['current_short_action'] == 'EXIT'
95
+
96
+
97
+def test_dedupe_history_immutable_state_updates_and_restart(tmp_path):
98
+ path = tmp_path/'recommendations.db'
99
+ store = SqliteResearchPersistence(path)
100
+ s = snapshot()
101
+ publish(store, [s])
102
+ original = store.recommendation_history()
103
+ again = deepcopy(s)
104
+ again.update(snapshot_id=str(uuid4()), generated_at=(NOW+timedelta(hours=1)).isoformat())
105
+ publish(store, [again])
106
+ assert store.recommendation_history() == original
107
+ assert store.recommendation_states()[0]['updated_at'] == again['generated_at']
108
+ moved = deepcopy(again)
109
+ moved.update(snapshot_id=str(uuid4()), current_price=1185., generated_at=(NOW+timedelta(hours=2)).isoformat())
110
+ publish(store, [moved])
111
+ assert store.recommendation_states()[0]['current_short_action'] == 'PARTIAL_PROFIT'
112
+ assert store.recommendation_history()[0] == original[0]
113
+ assert SqliteResearchPersistence(path).opportunity_current()['previous_recommendations'][0]['current_short_action'] == 'PARTIAL_PROFIT'
114
+
115
+
116
+def test_database_rejects_history_mutation_and_failed_publish_rolls_back():
117
+ store = SqliteResearchPersistence()
118
+ s = snapshot()
119
+ selection = publish(store, [s])
120
+ for table in ('stock_recommendation_history', 'global_opportunity_snapshot'):
121
+ with pytest.raises(Exception, match='IMMUTABLE'):
122
+ with store._connection:
123
+ store._connection.execute(f'UPDATE {table} SET payload = payload')
124
+ with pytest.raises(Exception, match='IMMUTABLE'):
125
+ with store._connection:
126
+ store._connection.execute(f'DELETE FROM {table}')
127
+ failed = snapshot(2)
128
+ def build(persisted):
129
+ assert persisted[0]['global_instrument_id'] == failed['global_instrument_id']
130
+ raise ValueError('SIMULATED_FAILURE')
131
+ with pytest.raises(ValueError, match='SIMULATED_FAILURE'):
132
+ store.publish_opportunity_cycle([failed], [], [], {'cycle_id': failed['cycle_id']}, build=build)
133
+ assert store.opportunity_snapshots(failed['cycle_id']) == []
134
+ assert store.opportunity_current() == selection
135
+
136
+
137
+def test_global_top_order_membership_independence_and_suppression():
138
+ rows = [snapshot(i) for i in range(1, 7)]
139
+ rows[-1]['rank_eligible'] = False
140
+ first = publish(SqliteResearchPersistence(), rows, 4)
141
+ second = publish(SqliteResearchPersistence(), [dict(r, held=True, watchlisted=True) for r in reversed(rows)], 4)
142
+ keys = lambda r: [c['global_instrument_id'] for c in r['top_short_term']]
143
+ assert keys(first) == keys(second) == [str(UUID(int=n)) for n in range(1, 5)]
144
+ assert len(first['top_long_term']) == 4
145
+ assert len(publish(SqliteResearchPersistence(), [snapshot(i) for i in range(1, 6)], 2)['top_short_term']) == 2
146
+ assert nse_equities([instrument(), instrument(2) | {'exchange': 'NYSE'}, instrument(3) | {'assetType': 'ETF'}]) == [instrument()]
147
+
148
+
149
+@pytest.mark.asyncio
150
+async def test_real_scanner_ranker_snapshot_persisted(monkeypatch):
151
+ service, rows, pairs, store = setup(monkeypatch, 3)
152
+ ranking = await service.run(rows, as_of=NOW)
153
+ snapshots = [snapshot_from_entry(e, str(uuid4()), NOW.isoformat()) for e in ranking.evaluated_entries]
154
+ selection = publish(store, snapshots)
155
+ actual = store.opportunity_snapshots(selection['cycle_id'])
156
+ assert len(actual) == 3
157
+ assert actual[0]['opportunity_score'] == ranking.top_n[0].opportunity_score
158
+ assert actual[0]['evidence_state']['rule']['rule_engine_version'] == 'STOCK_RULE_ENGINE_V1'
159
+
160
+
161
+@pytest.mark.asyncio
162
+async def test_explicit_cycle_uses_canonical_rows_and_reads_back_snapshots(monkeypatch):
163
+ from app import global_opportunity_cycle as cycle
164
+ service, rows, pairs, store = setup(monkeypatch, 3)
165
+ class Clock:
166
+ @staticmethod
167
+ def now(*a): return NOW
168
+ monkeypatch.setattr(cycle, 'datetime', Clock)
169
+ monkeypatch.setattr(cycle, 'GlobalOpportunityOrchestrator', lambda *a, **k: service)
170
+ source = AsyncMock()
171
+ source.active_global_equities.return_value = rows
172
+ source.sector_benchmark_contexts.return_value = {}
173
+ original = RecommendationEngineV1.evaluate
174
+ def evaluate(self, s, previous=None):
175
+ assert any(r['snapshot_id'] == s['snapshot_id'] for r in store.opportunity_snapshots(s['cycle_id']))
176
+ return original(self, s, previous)
177
+ monkeypatch.setattr(RecommendationEngineV1, 'evaluate', evaluate)
178
+ result = await cycle.run_global_opportunity_cycle(service.repository, source, candidate_ids=[UUID(int=1)], top_n=2)
179
+ assert result['universe_count'] == 1 and result['controlled_candidate_set']
180
+ assert len(store.recommendation_history()) == 1
181
+ source.active_global_equities.assert_awaited_once()
182
+
183
+
184
+@pytest.mark.asyncio
185
+async def test_bounded_previous_review_does_not_change_scanner_shortlist(monkeypatch):
186
+ service, rows, pairs, store = setup(monkeypatch, 3)
187
+ result = await service.run(rows, as_of=NOW, shortlist_limit=1, review_ids=[UUID(int=3)])
188
+ assert result.shortlist_count == 1 and result.deep_evaluated_count == 2
189
+ assert {e.global_instrument_id for e in result.evaluated_entries} == {UUID(int=1), UUID(int=3)}
190
+
191
+
192
+def observation(day, price, retrieved=None):
193
+ return MarketPriceObservation(instrument_id=UUID(int=1), observed_at=NOW+timedelta(days=day),
194
+ retrieved_at=NOW+timedelta(days=retrieved if retrieved is not None else day),
195
+ price=Decimal(str(price)), currency='INR', provider='YAHOO_FINANCE', source_url='https://example.test')
196
+
197
+
198
+def test_backtest_point_in_time_returns_and_missing_future():
199
+ r = RecommendationEngineV1().evaluate(snapshot()) | {'recommendation_id': str(uuid4())}
200
+ prices = {UUID(int=1): [observation(0, 100), observation(7, 110), observation(30, 90), observation(3, 80)]}
201
+ result = evaluate_backtest([r], prices, start=NOW, end=NOW, horizon='SHORT_TERM', now=NOW+timedelta(days=40))
202
+ assert result['metrics']['1W']['average_return'] == pytest.approx(10)
203
+ assert result['metrics']['1M']['average_return'] == pytest.approx(-10)
204
+ assert result['metrics']['1M']['max_adverse_excursion'] == pytest.approx(-20)
205
+ assert result['metrics']['1Y']['average_return'] is None
206
+ assert result['samples']['1Y'][0]['status'] == 'FUTURE_PRICE_UNAVAILABLE'
207
+ prices[UUID(int=1)][0] = observation(0, 100, retrieved=1)
208
+ assert evaluate_backtest([r], prices, start=NOW, end=NOW, horizon='SHORT_TERM', now=NOW+timedelta(days=40))['samples']['1W'][0]['status'] == 'ENTRY_PRICE_UNAVAILABLE'
209
+
210
+
211
+@pytest.mark.parametrize('key', ['publishedAt', 'publicAvailabilityAt', 'discoveredAt', 'computedAt', 'retrieved_at', 'calculated_at'])
212
+def test_temporal_guards(key):
213
+ assert not evidence_available({'nested': [{key: (NOW+timedelta(days=1)).isoformat()}]}, NOW)
214
+ assert evidence_available({'nested': [{key: NOW.isoformat()}]}, NOW)
215
+
216
+
217
+def test_backtest_excludes_future_evidence_and_future_recommendation():
218
+ r = RecommendationEngineV1().evaluate(snapshot()) | {'recommendation_id': str(uuid4())}
219
+ r['evidence_snapshot']['computedAt'] = (NOW+timedelta(days=1)).isoformat()
220
+ result = evaluate_backtest([r], {}, start=NOW, end=NOW, horizon='SHORT_TERM', now=NOW+timedelta(days=40))
221
+ assert result['recommendation_count'] == 0 and result['excluded_unavailable_evidence'] == 1
222
+ r['generated_at'] = (NOW+timedelta(days=1)).isoformat()
223
+ assert evaluate_backtest([r], {}, start=NOW, end=NOW, horizon='SHORT_TERM', now=NOW+timedelta(days=40))['recommendation_count'] == 0
224
+
225
+
226
+def test_backtest_daily_closes_are_persisted_outcome_evidence(monkeypatch):
227
+ from app import recommendation_backtesting as backtesting
228
+ from app.models import DailyMarketBar
229
+ from datetime import datetime
230
+ store = SqliteResearchPersistence()
231
+ publish(store, [snapshot()])
232
+ for day, close in [(-1, 100), (7, 110)]:
233
+ store.upsert_daily_market_bar(DailyMarketBar(global_instrument_id=UUID(int=1),
234
+ trading_date=(NOW+timedelta(days=day)).date(), close=Decimal(close), currency='INR',
235
+ provider='NSE', source_mode='REAL', source_url='https://example.test/bars',
236
+ retrieved_at=NOW+timedelta(days=day, hours=12)))
237
+ class Clock(datetime):
238
+ @classmethod
239
+ def now(cls, *a): return NOW+timedelta(days=40)
240
+ monkeypatch.setattr(backtesting, 'datetime', Clock)
241
+ result = backtesting.run_backtest(store, start=NOW, end=NOW, horizon='SHORT_TERM')
242
+ assert result['metrics']['1W']['average_return'] == pytest.approx(10)
243
+ assert store.backtests()[0] == result
244
+
245
+
246
+@pytest.mark.asyncio
247
+async def test_dashboard_read_has_no_compute_or_providers(monkeypatch):
248
+ from app import main
249
+ store = SqliteResearchPersistence()
250
+ publish(store, [snapshot()])
251
+ monkeypatch.setattr(main.repository, '_persistence', store)
252
+ def forbidden(*a, **kw): raise AssertionError('Provider or computation in GET')
253
+ monkeypatch.setattr(RecommendationEngineV1, 'evaluate', forbidden)
254
+ monkeypatch.setattr(main.portfolio_orchestrator, 'active_global_equities', forbidden)
255
+ monkeypatch.setattr(main.stock_rule_engine_service, 'analyze', forbidden)
256
+ import httpx
257
+ monkeypatch.setattr(httpx.AsyncClient, 'request', forbidden)
258
+ assert len((await main.opportunity_radar())['top_short_term']) == 1
259
+ assert len(await main.opportunity_history(UUID(int=1))) == 1
260
+ assert await main.backtest_runs() == []
ai/research-engine/tests/test_recommendation_postgres.py
new
+22
@@ -0,0 +1,22 @@
1
+"""Opt-in adapter smoke against an isolated database migrated by Java Flyway."""
2
+import os
3
+import pytest
4
+from uuid import UUID
5
+from app.settings import Settings
6
+from app.postgres_persistence import PostgresResearchPersistence
7
+from test_recommendation_lifecycle import snapshot, publish
8
+
9
+
10
+@pytest.mark.skipif(not os.environ.get('RECOMMENDATION_TEST_PG_PORT'), reason='Requires disposable Flyway-migrated PostgreSQL')
11
+def test_postgres_publish_read_history_and_immutable_trigger():
12
+ store = PostgresResearchPersistence(Settings(research_database_host='127.0.0.1',
13
+ research_database_port=int(os.environ['RECOMMENDATION_TEST_PG_PORT']),
14
+ research_database_name='recommendation_validation', research_database_user='postgres',
15
+ research_database_password='', research_database_schema='research', research_database_ssl_mode='disable'))
16
+ s = snapshot(87123)
17
+ published = publish(store, [s])
18
+ assert store.opportunity_current()['cycle_id'] == published['cycle_id']
19
+ assert len(store.recommendation_history(UUID(int=87123))) == 1
20
+ with pytest.raises(Exception, match='RECOMMENDATION_HISTORY_IS_IMMUTABLE'):
21
+ with store._connection:
22
+ store._connection.execute('UPDATE stock_recommendation_history SET payload = payload')
frontend/app/components/backtesting.tsx
new
+54
@@ -0,0 +1,54 @@
1
+"use client";
2
+import { useEffect, useState } from "react";
3
+import { request } from "../lib/portfolio-api";
4
+
5
+type Metric = { recommendation_count: number; evaluated_count: number; missing_count: number;
6
+ hit_rate: number | null; average_return: number | null; median_return: number | null };
7
+type Example = { recommendation_id: string; global_instrument_id: string; symbol?: string; company_name?: string; return_pct: number };
8
+export type Backtest = { backtest_id: string; generated_at: string; recommendation_count: number;
9
+ metrics: Record<string, Metric>; winners: Example[]; losers: Example[]; methodology: string };
10
+const pct = (v: number | null) => v == null ? "Insufficient evidence" : `${v.toFixed(2)}%`;
11
+
12
+export function BacktestResults({ run }: { run: Backtest }) {
13
+ return <section aria-label="Backtest results"><h3>Recommendations tested: {run.recommendation_count}</h3>
14
+ <p>{run.methodology}</p><div style={{ overflowX: "auto" }}><table><thead><tr><th>Horizon</th><th>Evaluated</th><th>Missing</th><th>Hit rate</th><th>Average return</th><th>Median return</th></tr></thead>
15
+ <tbody>{["1W", "1M", "3M", "6M", "1Y"].map(h => <tr key={h}><th>{h}</th><td>{run.metrics[h].evaluated_count}</td><td>{run.metrics[h].missing_count}</td>
16
+ <td>{pct(run.metrics[h].hit_rate)}</td><td>{pct(run.metrics[h].average_return)}</td><td>{pct(run.metrics[h].median_return)}</td></tr>)}</tbody></table></div>
17
+ <h4>Winner examples</h4>{run.winners.length ? <ul>{run.winners.map(e => <li key={e.recommendation_id}>{e.company_name ?? e.symbol ?? "Company unavailable"}: {pct(e.return_pct)}</li>)}</ul> : <p>No winner evidence yet.</p>}
18
+ <h4>Loser examples</h4>{run.losers.length ? <ul>{run.losers.map(e => <li key={e.recommendation_id}>{e.company_name ?? e.symbol ?? "Company unavailable"}: {pct(e.return_pct)}</li>)}</ul> : <p>No loser evidence yet.</p>}
19
+ </section>;
20
+}
21
+
22
+export function Backtesting() {
23
+ const [start, setStart] = useState(""); const [end, setEnd] = useState("");
24
+ const [horizon, setHorizon] = useState("SHORT_TERM"); const [runs, setRuns] = useState<Backtest[]>([]);
25
+ const [selected, setSelected] = useState(""); const [busy, setBusy] = useState(false); const [error, setError] = useState("");
26
+ useEffect(() => { let active = true;
27
+ request<Backtest[]>("/api/v1/research/backtesting/runs").then(r => { if (active) { setRuns(r); setSelected(r[0]?.backtest_id ?? ""); } }).catch(() => { if (active) setError("Unable to load backtests."); });
28
+ return () => { active = false; };
29
+ }, []);
30
+ async function run() {
31
+ setBusy(true); setError("");
32
+ try {
33
+ const result = await request<Backtest>("/api/v1/research/backtesting/runs", { method: "POST", body: JSON.stringify({
34
+ start: `${start}T00:00:00Z`, end: new Date(Math.min(new Date(`${end}T23:59:59Z`).getTime(), Date.now())).toISOString(), market: "NSE", horizon }) });
35
+ setRuns(old => [result, ...old]); setSelected(result.backtest_id);
36
+ } catch { setError("Backtest could not run. Check the date range and try again."); }
37
+ finally { setBusy(false); }
38
+ }
39
+ const result = runs.find(r => r.backtest_id === selected);
40
+ return <section className="opportunity-radar"><h2>BACKTESTING</h2>
41
+ <p>Evaluate persisted recommendations using recorded evidence and subsequent prices.</p>
42
+ <form onSubmit={e => { e.preventDefault(); void run(); }}>
43
+ <label>Start date <input type="date" required value={start} onChange={e => setStart(e.target.value)} /></label>{" "}
44
+ <label>End date <input type="date" required min={start} value={end} onChange={e => setEnd(e.target.value)} /></label>{" "}
45
+ <label>Market <select><option>NSE</option></select></label>{" "}
46
+ <label>Horizon <select value={horizon} onChange={e => setHorizon(e.target.value)}><option value="SHORT_TERM">Short-term</option><option value="LONG_TERM">Long-term</option></select></label>{" "}
47
+ <button disabled={busy}>{busy ? "Running…" : "Run backtest"}</button>
48
+ </form>
49
+ {error && <p role="alert">{error}</p>}
50
+ <label>Persisted backtest <select value={selected} onChange={e => setSelected(e.target.value)}><option value="">Select a run</option>
51
+ {runs.map(r => <option key={r.backtest_id} value={r.backtest_id}>{r.generated_at} · {r.recommendation_count} recommendations</option>)}</select></label>
52
+ {result ? <BacktestResults run={result} /> : <p>No backtest selected.</p>}
53
+ </section>;
54
+}
frontend/app/components/investment-workspace.tsx
+6
-1
@@ -29,6 +29,8 @@ import {
29
} from "lucide-react";
30
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
31
import { createPortal } from "react-dom";
32
+import { OpportunityRadar } from "./opportunity-radar";
33
+import { Backtesting } from "./backtesting";
34
import { frontendConfig } from "../config";
35
import {
36
type AuthenticatedUser,
@@ -77,7 +79,7 @@ import {
79
} from "../lib/market-intelligence";
80
import { Badge, Button, Card, EmptyState, ErrorState, Field, MetricCard, Skeleton } from "./ui";
81
80
-type View = "dashboard" | "portfolio" | "research" | "brokers" | "settings";
82
+type View = "dashboard" | "portfolio" | "research" | "backtesting" | "brokers" | "settings";
83
type Theme = "system" | "light" | "dark";
84
type SortKey = "company" | "ticker" | "marketValue" | "profitLoss" | "allocation";
85
type ResearchSectionId = "overview" | "growth" | "orders" | "capex" | "customers" | "guidance" | "news" | "sources";
@@ -94,6 +96,7 @@ const navItems: Array<{ id: View; label: string; icon: typeof Gauge }> = [
96
{ id: "dashboard", label: "Dashboard", icon: Gauge },
97
{ id: "portfolio", label: "Portfolio", icon: BriefcaseBusiness },
98
{ id: "research", label: "Research", icon: Search },
99
+ { id: "backtesting", label: "Backtesting", icon: LineChart },
100
{ id: "brokers", label: "Brokers", icon: WalletCards },
101
{ id: "settings", label: "Settings", icon: Settings }
102
];
@@ -1510,6 +1513,8 @@ export function InvestmentWorkspace() {
1513
) : null}
1514
1515
{view !== "research" && watchlistActionError ? <p role="alert">{watchlistActionError}</p> : null}
1516
+ {view === "backtesting" ? <Backtesting /> : null}
1517
+ {view === "dashboard" ? <OpportunityRadar heldIds={positions.flatMap(p => p.instrument.globalInstrumentId ? [p.instrument.globalInstrumentId] : [])} watchlistedIds={Object.values(savedWatchlistIds).flat()} /> : null}
1518
{loading ? <LoadingView /> : null}
1519
1520
{!loading && !error ? (
frontend/app/components/opportunity-radar.tsx
new
+87
@@ -0,0 +1,87 @@
1
+"use client";
2
+
3
+import { useEffect, useState } from "react";
4
+import { request } from "../lib/portfolio-api";
5
+
6
+export type Opportunity = {
7
+ global_instrument_id: string; company_name?: string; symbol?: string; current_price: number | null;
8
+ opportunity_score: number | null; opportunity_confidence: number; score_coverage: number;
9
+ new_investor_action: string; existing_holder_action: string; current_short_action: string; current_long_action: string;
10
+ short_entry_low: number | null; short_entry_high: number | null; short_target_1: number | null;
11
+ short_target_2: number | null; short_invalidation: number | null; long_entry_low: number | null;
12
+ long_entry_high: number | null; long_fair_value: number | null; long_target: number | null; long_invalidation: number | null;
13
+ top_positive_reasons: string[]; top_negative_reasons: string[]; data_state: string;
14
+ missing_areas: string[]; stale_areas: string[]; short_horizon: string; long_horizon: string;
15
+ short_term_state: string; long_term_state: string; lifecycle_reasons?: string[]; evaluation_status?: string;
16
+};
17
+export type Radar = { generated_at: string | null; best_buy_today: Opportunity | null;
18
+ top_short_term: Opportunity[]; top_long_term: Opportunity[]; previous_recommendations: Opportunity[] };
19
+
20
+export function price(value: number | null | undefined): string {
21
+ return value == null ? "Insufficient evidence" : `₹${value.toLocaleString("en-IN", { maximumFractionDigits: 2 })}`;
22
+}
23
+function range(low: number | null, high: number | null): string {
24
+ return low == null || high == null ? "Insufficient evidence" : `${price(low)} – ${price(high)}`;
25
+}
26
+function label(value: string | undefined) { return value?.replaceAll("_", " ") ?? "Unavailable"; }
27
+
28
+export function OpportunityCard({ item, horizon, held, watchlisted }: {
29
+ item: Opportunity; horizon: "short" | "long"; held: boolean; watchlisted: boolean;
30
+}) {
31
+ return <article className="opportunity-card">
32
+ <h4>{item.company_name ?? item.symbol} <small>{item.symbol}</small></h4>
33
+ <p>{price(item.current_price)} {held ? " · Held" : ""}{watchlisted ? " · Watchlisted" : ""}</p>
34
+ <p>Opportunity {item.opportunity_score?.toFixed(1) ?? "Unavailable"} · Confidence {item.opportunity_confidence.toFixed(1)}% · Coverage {item.score_coverage.toFixed(1)}%</p>
35
+ <p>New investor: {label(item.new_investor_action)}<br />Existing holder: {label(item.existing_holder_action)}</p>
36
+ <strong>{label(horizon === "short" ? item.current_short_action : item.current_long_action)}</strong>
37
+ <dl>{horizon === "short" ? <>
38
+ <dt>Entry range</dt><dd>{range(item.short_entry_low, item.short_entry_high)}</dd>
39
+ <dt>Target 1</dt><dd>{price(item.short_target_1)}</dd><dt>Target 2</dt><dd>{price(item.short_target_2)}</dd>
40
+ <dt>Invalidation</dt><dd>{price(item.short_invalidation)}</dd><dt>Horizon</dt><dd>{item.short_horizon}</dd>
41
+ </> : <>
42
+ <dt>Accumulation range</dt><dd>{range(item.long_entry_low, item.long_entry_high)}</dd>
43
+ <dt>Fair value</dt><dd>{price(item.long_fair_value)}</dd><dt>Target</dt><dd>{price(item.long_target)}</dd>
44
+ <dt>Invalidation</dt><dd>{price(item.long_invalidation)}</dd><dt>Horizon</dt><dd>{item.long_horizon}</dd>
45
+ </>}</dl>
46
+ <p><strong>Why</strong></p><ul>{item.top_positive_reasons.slice(0, 4).map(r => <li key={r}>{label(r)}</li>)}</ul>
47
+ <p><strong>Risks</strong></p><ul>{item.top_negative_reasons.slice(0, 3).map(r => <li key={r}>{label(r)}</li>)}</ul>
48
+ <details><summary>Data: {label(item.data_state)}</summary>
49
+ <p>Missing: {item.missing_areas.join(", ") || "None reported"}</p><p>Stale: {item.stale_areas.join(", ") || "None reported"}</p>
50
+ </details>
51
+ </article>;
52
+}
53
+
54
+export function RadarContent({ data, heldIds = [], watchlistedIds = [] }: { data: Radar; heldIds?: string[]; watchlistedIds?: string[] }) {
55
+ const card = (item: Opportunity, horizon: "short" | "long") => <OpportunityCard key={item.global_instrument_id} item={item} horizon={horizon}
56
+ held={heldIds.includes(item.global_instrument_id)} watchlisted={watchlistedIds.includes(item.global_instrument_id)} />;
57
+ return <>
58
+ <p>{data.generated_at ? `Last cycle: ${new Date(data.generated_at).toLocaleString()}` : "No persisted opportunity cycle yet."}</p>
59
+ <h3>BEST BUY TODAY</h3>
60
+ {data.best_buy_today ? <p>{data.best_buy_today.company_name ?? data.best_buy_today.symbol} · {price(data.best_buy_today.current_price)} · {label(data.best_buy_today.new_investor_action)}</p> : <p>No qualifying buy candidate.</p>}
61
+ <h3>TOP SHORT-TERM OPPORTUNITIES</h3>
62
+ <div className="opportunity-cards">{data.top_short_term.map(item => card(item, "short"))}</div>
63
+ {!data.top_short_term.length && <p>No qualifying short-term opportunities.</p>}
64
+ <h3>TOP LONG-TERM OPPORTUNITIES</h3>
65
+ <div className="opportunity-cards">{data.top_long_term.map(item => card(item, "long"))}</div>
66
+ {!data.top_long_term.length && <p>No qualifying long-term opportunities.</p>}
67
+ <h3>PREVIOUS RECOMMENDATIONS</h3>
68
+ {!data.previous_recommendations.length ? <p>No previous recommendations.</p> : <ul>{data.previous_recommendations.map(item =>
69
+ <li key={item.global_instrument_id}>{item.company_name ?? item.symbol} · Short: {label(item.current_short_action)} ({label(item.short_term_state)}) · Long: {label(item.current_long_action)} ({label(item.long_term_state)})
70
+ <p>{item.lifecycle_reasons?.map(label).join(" · ")}{item.evaluation_status ? ` · ${label(item.evaluation_status)}` : ""}</p>
71
+ </li>)}</ul>}
72
+ </>;
73
+}
74
+
75
+export function OpportunityRadar({ heldIds, watchlistedIds }: { heldIds: string[]; watchlistedIds: string[] }) {
76
+ const [data, setData] = useState<Radar | null>(null);
77
+ const [error, setError] = useState(false);
78
+ useEffect(() => {
79
+ let active = true;
80
+ request<Radar>("/api/v1/research/opportunities/current").then(value => { if (active) setData(value); }).catch(() => { if (active) setError(true); });
81
+ return () => { active = false; };
82
+ }, []);
83
+ return <section className="opportunity-radar" aria-label="Global opportunity radar">
84
+ <h2>GLOBAL OPPORTUNITY RADAR</h2>
85
+ {error ? <p role="alert">Opportunity radar unavailable.</p> : data ? <RadarContent data={data} heldIds={heldIds} watchlistedIds={watchlistedIds} /> : <p>Loading persisted opportunities…</p>}
86
+ </section>;
87
+}
frontend/app/lib/portfolio-api.ts
+1
-1
@@ -825,7 +825,7 @@ function currentAuthenticatedApiToken(): string | null {
825
return typeof window === "undefined" ? null : window.localStorage.getItem("aip.accessToken");
826
}
827
828
-async function request<T>(path: string, init?: RequestInit): Promise<T> {
828
+export async function request<T>(path: string, init?: RequestInit): Promise<T> {
829
const token = currentAuthenticatedApiToken();
830
const response = await fetch(`${frontendConfig.apiBaseUrl}${path}`, {
831
...init,
frontend/app/styles.css
+8
@@ -2213,3 +2213,11 @@ tbody tr[class*="ownership-"] td {
2213
.research-saved-lists { display: flex; align-items: center; gap: 16px; flex-wrap: wrap; padding: 16px 20px; }
2214
.research-saved-lists h3 { margin: 0; }
2215
.research-saved-lists .research-context-select { margin-left: auto; }
2216
+/* Persisted global recommendations */
2217
+.opportunity-radar { padding: 24px; margin-bottom: 24px; border: 1px solid var(--border, #657080); border-radius: 16px; }
2218
+.opportunity-cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(260px, 1fr)); gap: 16px; }
2219
+.opportunity-card { padding: 18px; border: 1px solid var(--border, #657080); border-radius: 12px; }
2220
+.opportunity-card dl { display: grid; grid-template-columns: 1fr 1fr; gap: 8px; }
2221
+.opportunity-card dd { margin: 0; }
2222
+.opportunity-card p, .opportunity-card li { overflow-wrap: anywhere; }
2223
+.opportunity-radar table { width: 100%; text-align: left; border-spacing: 12px; }
frontend/tests/opportunity-radar.test.mjs
new
+49
@@ -0,0 +1,49 @@
1
+import test from 'node:test';
2
+import assert from 'node:assert/strict';
3
+import fs from 'node:fs';
4
+import { createRequire } from 'node:module';
5
+import ts from 'typescript';
6
+import React from 'react';
7
+import { renderToStaticMarkup } from 'react-dom/server';
8
+
9
+const require = createRequire(import.meta.url);
10
+function component(name) {
11
+ const source = fs.readFileSync(new URL(`../app/components/${name}.tsx`, import.meta.url), 'utf8');
12
+ const output = ts.transpileModule(source, { compilerOptions: { jsx: ts.JsxEmit.ReactJSX, module: ts.ModuleKind.CommonJS, target: ts.ScriptTarget.ES2022 } }).outputText;
13
+ const module = { exports: {} };
14
+ new Function('require', 'module', 'exports', output)(path => path.includes('portfolio-api') ? {} : require(path), module, module.exports);
15
+ return module.exports;
16
+}
17
+const { RadarContent, price } = component('opportunity-radar');
18
+const { Backtesting, BacktestResults } = component('backtesting');
19
+const item = i => ({ global_instrument_id: `${i}`, company_name: `Company ${i}`, symbol: `C${i}`, current_price: 100,
20
+ opportunity_score: 80, opportunity_confidence: 85, score_coverage: 90, new_investor_action: 'BUY_CANDIDATE',
21
+ existing_holder_action: 'HOLD', current_short_action: 'BUY', current_long_action: 'ACCUMULATE',
22
+ short_entry_low: null, short_entry_high: null, top_positive_reasons: ['QUALITY'], top_negative_reasons: ['MISSING_ATR'],
23
+ data_state: 'PARTIAL', missing_areas: ['ATR'], stale_areas: [], short_horizon: '1–3 months', long_horizon: '6–12 months' });
24
+const empty = { generated_at: null, best_buy_today: null, top_short_term: [], top_long_term: [], previous_recommendations: [] };
25
+
26
+test('radar renders persisted 2–4 cards per horizon and missing ranges', () => {
27
+ for (const count of [2, 3, 4]) {
28
+ const picks = Array.from({ length: count }, (_, i) => item(i));
29
+ const html = renderToStaticMarkup(React.createElement(RadarContent, { data: { ...empty, top_short_term: picks, top_long_term: picks }, heldIds: ['0'], watchlistedIds: ['1'] }));
30
+ assert.equal((html.match(/class="opportunity-card"/g) ?? []).length, count * 2);
31
+ assert.match(html, /Insufficient evidence/);
32
+ assert.match(html, /Held/); assert.match(html, /Watchlisted/);
33
+ assert.doesNotMatch(html, /₹0/);
34
+ }
35
+ assert.equal(price(null), 'Insufficient evidence');
36
+});
37
+test('empty and lifecycle render', () => {
38
+ assert.match(renderToStaticMarkup(React.createElement(RadarContent, { data: empty })), /No persisted opportunity cycle/);
39
+ const html = renderToStaticMarkup(React.createElement(RadarContent, { data: { ...empty, previous_recommendations: [{ ...item(1), current_short_action: 'PARTIAL_PROFIT', short_term_state: 'PARTIAL_PROFIT', current_long_action: 'HOLD', long_term_state: 'HOLDING' }] } }));
40
+ assert.match(html, /PARTIAL PROFIT/); assert.match(html, /Long: HOLD/);
41
+});
42
+test('backtesting page and all horizon results render', () => {
43
+ assert.match(renderToStaticMarkup(React.createElement(Backtesting)), /Run backtest/);
44
+ const metric = { evaluated_count: 1, missing_count: 0, hit_rate: 100, average_return: 10, median_return: 10 };
45
+ const run = { recommendation_count: 1, methodology: 'Recorded evidence', winners: [], losers: [], metrics: Object.fromEntries(['1W', '1M', '3M', '6M', '1Y'].map(h => [h, metric])) };
46
+ const html = renderToStaticMarkup(React.createElement(BacktestResults, { run }));
47
+ for (const h of ['1W', '1M', '3M', '6M', '1Y']) assert.ok(html.includes(h));
48
+ assert.match(html, /100.00%/);
49
+});
services/research-service/src/main/resources/db/migration/V13__global_recommendation_lifecycle.sql
new
+39
@@ -0,0 +1,39 @@
1
+-- Additive recommendation lifecycle; payloads retain complete versioned evidence. No backfill.
2
+
3
+CREATE TABLE global_opportunity_snapshot (
4
+ snapshot_id TEXT PRIMARY KEY, cycle_id TEXT NOT NULL, global_instrument_id TEXT NOT NULL,
5
+ market TEXT NOT NULL, generated_at TEXT NOT NULL, scanner_version TEXT, rule_engine_version TEXT, technical_version TEXT, sector_version TEXT, ranker_version TEXT, price_as_of TEXT, opportunity_score NUMERIC, opportunity_confidence NUMERIC, score_coverage NUMERIC, rule_engine_score NUMERIC, rank_position NUMERIC, technical_score NUMERIC, sector_score NUMERIC, valuation_score NUMERIC, quality_score NUMERIC, growth_score NUMERIC, balance_sheet_score NUMERIC, quarterly_score NUMERIC, catalyst_score NUMERIC, news_score NUMERIC, shareholding_score NUMERIC, governance_score NUMERIC, current_price NUMERIC, rank_eligible INTEGER, suppression_reasons TEXT, top_positive_reasons TEXT, top_negative_reasons TEXT, evidence_state TEXT, payload TEXT NOT NULL,
6
+ UNIQUE(cycle_id, global_instrument_id)
7
+);
8
+CREATE INDEX opportunity_snapshot_instrument ON global_opportunity_snapshot(global_instrument_id, generated_at);
9
+CREATE TABLE stock_recommendation_history (
10
+ recommendation_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL REFERENCES global_opportunity_snapshot(snapshot_id),
11
+ global_instrument_id TEXT NOT NULL, generated_at TEXT NOT NULL,
12
+ recommendation_engine_version TEXT NOT NULL, fingerprint TEXT NOT NULL, rule_engine_version TEXT, ranker_version TEXT, new_investor_action TEXT, existing_holder_action TEXT, short_term_action TEXT, long_term_action TEXT, price_at_recommendation NUMERIC, opportunity_score NUMERIC, confidence NUMERIC, coverage NUMERIC, short_entry_low NUMERIC, short_entry_high NUMERIC, short_target_1 NUMERIC, short_target_2 NUMERIC, short_invalidation NUMERIC, long_entry_low NUMERIC, long_entry_high NUMERIC, long_fair_value NUMERIC, long_target NUMERIC, long_invalidation NUMERIC, top_positive_reasons TEXT, top_negative_reasons TEXT, evidence_snapshot TEXT, payload TEXT NOT NULL
13
+);
14
+CREATE INDEX recommendation_history_instrument ON stock_recommendation_history(global_instrument_id, generated_at);
15
+CREATE TABLE recommendation_current_state (
16
+ global_instrument_id TEXT PRIMARY KEY,
17
+ latest_recommendation_id TEXT NOT NULL REFERENCES stock_recommendation_history(recommendation_id),
18
+ updated_at TEXT NOT NULL, short_term_state TEXT, long_term_state TEXT, current_short_action TEXT, current_long_action TEXT, lifecycle_status TEXT, price_at_recommendation NUMERIC, current_price NUMERIC, short_target_distance_pct NUMERIC, long_target_distance_pct NUMERIC, payload TEXT NOT NULL
19
+);
20
+CREATE TABLE global_opportunity_top_selection (
21
+ cycle_id TEXT PRIMARY KEY, generated_at TEXT NOT NULL, market TEXT NOT NULL, payload TEXT NOT NULL
22
+);
23
+CREATE TABLE recommendation_backtest_run (
24
+ backtest_id TEXT PRIMARY KEY, generated_at TEXT NOT NULL, payload TEXT NOT NULL
25
+);
26
+
27
+CREATE FUNCTION reject_recommendation_history_mutation() RETURNS trigger LANGUAGE plpgsql AS $$
28
+BEGIN
29
+ RAISE EXCEPTION 'RECOMMENDATION_HISTORY_IS_IMMUTABLE';
30
+END;
31
+$$;
32
+CREATE TRIGGER immutable_opportunity_snapshots BEFORE UPDATE OR DELETE ON global_opportunity_snapshot
33
+ FOR EACH ROW EXECUTE FUNCTION reject_recommendation_history_mutation();
34
+CREATE TRIGGER immutable_recommendation_history BEFORE UPDATE OR DELETE ON stock_recommendation_history
35
+ FOR EACH ROW EXECUTE FUNCTION reject_recommendation_history_mutation();
36
+CREATE TRIGGER immutable_top_selections BEFORE UPDATE OR DELETE ON global_opportunity_top_selection
37
+ FOR EACH ROW EXECUTE FUNCTION reject_recommendation_history_mutation();
38
+CREATE TRIGGER immutable_backtest_runs BEFORE UPDATE OR DELETE ON recommendation_backtest_run
39
+ FOR EACH ROW EXECUTE FUNCTION reject_recommendation_history_mutation();
services/research-service/src/test/java/com/aiinvestment/research/ResearchFlywayMigrationTest.java
+4
-2
@@ -94,7 +94,9 @@ class ResearchFlywayMigrationTest {
94
""",
95
String.class
96
);
97
- assertThat(version).isEqualTo("11");
97
+ assertThat(version).isEqualTo("13");
98
+ assertThat(tables).contains("global_opportunity_snapshot", "stock_recommendation_history",
99
+ "recommendation_current_state", "global_opportunity_top_selection", "recommendation_backtest_run");
100
101
Integer nseSessions = jdbcTemplate.queryForObject(
102
"SELECT count(*) FROM research.market_trading_schedules WHERE market_code = 'NSE'", Integer.class
@@ -161,7 +163,7 @@ class ResearchFlywayMigrationTest {
163
""");
164
Flyway upgrade = Flyway.configure().dataSource(url, username, password)
165
.schemas("research").defaultSchema("research").table("flyway_schema_history_research").load();
164
- assertThat(upgrade.migrate().migrationsExecuted).isEqualTo(1);
166
+ assertThat(upgrade.migrate().migrationsExecuted).isEqualTo(3);
167
assertThat(upgrade.migrate().migrationsExecuted).isZero();
168
assertThat(existing.queryForObject("SELECT COUNT(*) FROM research.global_daily_market_bars", Integer.class)).isZero();
169
assertThat(existing.queryForObject("SELECT COUNT(*) FROM research.global_market_price_observations", Integer.class)).isEqualTo(1);