| 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) |