| 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() == [] |