""" ========================================= CV6 AI Trading OS Professional Market Replay & Validation Engine ========================================= EXACT simulation of the Live Engine. Only difference: Historical Data instead of Live Data. LIVE FLOW: Market Data → Scanner → Strategy → Risk → AI → Execution → Position Monitor → Trade Journal → Portfolio → Dashboard BACKTEST FLOW (identical logic): Historical Candle → Scanner → Strategy → Risk → AI (optional) → Execution Simulation → Position Monitor → Trade Journal → Portfolio Update → Next Candle MODES: 1. QUICK — single stock, date range 2. PORTFOLIO — multiple symbols / watchlist / NSE 3. REPLAY — step-by-step with play/pause/speed 4. PAPER_REPLAY — historical treated as live (no future peek) REUSES (no duplication): BrokerAllocator — position size + target calculation CapitalPoolManager — capital per strategy style PositionMonitor — SL / Target / Trail SL RiskGuard — all risk filters AIConsensusEngine — optional AI confirmation ========================================= """ from __future__ import annotations import asyncio import math import threading import time import uuid from collections import defaultdict from datetime import datetime, timedelta from dataclasses import dataclass, field, asdict from enum import Enum from typing import Dict, List, Optional, Any from loguru import logger # ── Live engine modules (REUSED — no duplication) ───────────────────────────── from app.autonomous.capital_pool import AutonomousPosition, CapitalPoolManager from app.autonomous.position_monitor import PositionMonitor from app.autonomous.broker_allocator import BrokerAllocator from app.autonomous.autonomous_config import load_config # ── Backtest-specific modules ────────────────────────────────────────────────── from app.backtest.backtest_data_feed import BacktestDataFeed, HistoricalCandle from app.backtest.backtest_broker import BacktestBroker, BrokerFill from app.backtest.backtest_reporter import BacktestReporter # ═══════════════════════════════════════════════════════════════════════════════ # State enums # ═══════════════════════════════════════════════════════════════════════════════ class BacktestMode(str, Enum): QUICK = "QUICK" PORTFOLIO = "PORTFOLIO" REPLAY = "REPLAY" PAPER_REPLAY = "PAPER_REPLAY" class BacktestState(str, Enum): IDLE = "IDLE" LOADING = "LOADING" RUNNING = "RUNNING" PAUSED = "PAUSED" COMPLETED = "COMPLETED" ERROR = "ERROR" # ═══════════════════════════════════════════════════════════════════════════════ # Trade Journal entry # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class BacktestTrade: trade_id: str symbol: str side: str # BUY | SELL style: str # INTRADAY | SWING | ... strategy: str sector: str entry_price: float exit_price: float qty: int entry_time: str exit_time: str exit_reason: str # TARGET_HIT | STOPPED_OUT | EOD | MANUAL pnl: float pnl_pct: float charges: float # total broker charges charges_detail: dict slippage_pct: float entry_indicators: dict # RSI, EMA, MACD, etc. at entry ai_model: str ai_confidence: float ai_reason: str risk_score: float broker: str entry_order_id: str exit_order_id: str market_personality: str holding_hours: float def to_dict(self) -> dict: return asdict(self) # ═══════════════════════════════════════════════════════════════════════════════ # Replay step snapshot (for REPLAY mode navigation) # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class ReplayStep: candle_idx: int candle: dict scan_signals: List[dict] trades_opened: List[dict] trades_closed: List[dict] open_positions: List[dict] equity: float cumulative_pnl: float capital_pools: List[dict] ai_decisions: List[dict] risk_verdicts: List[dict] # ═══════════════════════════════════════════════════════════════════════════════ # Indicator computation (rolling window — NO future peeking) # ═══════════════════════════════════════════════════════════════════════════════ class _Indicators: """Compute technical indicators on a rolling window of candles.""" @staticmethod def ema(closes: List[float], period: int) -> float: if len(closes) < period: return closes[-1] if closes else 0.0 k = 2.0 / (period + 1) val = sum(closes[:period]) / period for c in closes[period:]: val = c * k + val * (1 - k) return round(val, 4) @staticmethod def rsi(closes: List[float], period: int = 14) -> float: if len(closes) < period + 1: return 50.0 gains, losses = [], [] for i in range(1, len(closes)): d = closes[i] - closes[i - 1] gains.append(max(d, 0)) losses.append(max(-d, 0)) ag = sum(gains[:period]) / period al = sum(losses[:period]) / period for i in range(period, len(gains)): ag = (ag * (period - 1) + gains[i]) / period al = (al * (period - 1) + losses[i]) / period rs = ag / al if al != 0 else 100 return round(100 - 100 / (1 + rs), 2) @staticmethod def macd(closes: List[float]) -> tuple: """Returns (macd_line, signal_line).""" if len(closes) < 26: return 0.0, 0.0 ema12 = _Indicators._ema_series(closes, 12) ema26 = _Indicators._ema_series(closes, 26) macd_line = [e12 - e26 for e12, e26 in zip(ema12, ema26)] sig = _Indicators._ema_series(macd_line, 9) return round(macd_line[-1], 4), round(sig[-1], 4) @staticmethod def vwap(candles: List[HistoricalCandle]) -> float: cum_pv = cum_v = 0.0 for c in candles: tp = (c.high + c.low + c.close) / 3 cum_pv += tp * c.volume cum_v += c.volume return round(cum_pv / cum_v, 2) if cum_v > 0 else candles[-1].close if candles else 0.0 @staticmethod def supertrend_direction(candles: List[HistoricalCandle], period: int = 7, mult: float = 3.0) -> str: if len(candles) < period + 1: return "UP" closes = [c.close for c in candles] highs = [c.high for c in candles] lows = [c.low for c in candles] tr = [highs[0] - lows[0]] for i in range(1, len(candles)): tr.append(max(highs[i]-lows[i], abs(highs[i]-closes[i-1]), abs(lows[i]-closes[i-1]))) atr = sum(tr[-period:]) / period hl2 = (highs[-1] + lows[-1]) / 2 upper = hl2 + mult * atr lower = hl2 - mult * atr return "DOWN" if closes[-1] < lower else "UP" @staticmethod def atr(candles: List[HistoricalCandle], period: int = 14) -> float: if len(candles) < 2: return candles[-1].close * 0.01 if candles else 1.0 tr = [] for i in range(1, len(candles)): h, l, pc = candles[i].high, candles[i].low, candles[i-1].close tr.append(max(h - l, abs(h - pc), abs(l - pc))) return round(sum(tr[-period:]) / min(period, len(tr)), 4) @staticmethod def _ema_series(values: List[float], period: int) -> List[float]: if len(values) < period: return values k = 2.0 / (period + 1) result = [sum(values[:period]) / period] for v in values[period:]: result.append(v * k + result[-1] * (1 - k)) return ([values[0]] * (period - 1)) + result # ═══════════════════════════════════════════════════════════════════════════════ # Strategy signal engine (identical logic to live scanner) # ═══════════════════════════════════════════════════════════════════════════════ class _StrategySignal: """ Computes strategy signal from rolling candle history. Same indicator logic as the live scanner — only data source changes. """ # Minimum candles needed per strategy MIN_CANDLES = { "EMA_CROSSOVER": 25, "VWAP": 5, "SUPERTREND": 10, "MACD": 35, "RSI_REVERSAL": 16, } def compute( self, strategy: str, candles: List[HistoricalCandle], # ALL candles up to current (no future) ) -> dict: """ Returns {"signal": BUY|SELL|HOLD, "indicators": {...}} candles[-1] is the CURRENT candle (most recent). """ closes = [c.close for c in candles] result = {"signal": "HOLD", "indicators": {}} if len(candles) < self.MIN_CANDLES.get(strategy, 5): return result if strategy == "EMA_CROSSOVER": return self._ema_crossover(candles, closes, result) elif strategy == "VWAP": return self._vwap(candles, closes, result) elif strategy == "SUPERTREND": return self._supertrend(candles, closes, result) elif strategy == "MACD": return self._macd(candles, closes, result) elif strategy == "RSI_REVERSAL": return self._rsi_reversal(candles, closes, result) return result def _ema_crossover(self, candles, closes, result): ema9_prev = _Indicators.ema(closes[:-1], 9) ema21_prev = _Indicators.ema(closes[:-1], 21) ema9_cur = _Indicators.ema(closes, 9) ema21_cur = _Indicators.ema(closes, 21) rsi = _Indicators.rsi(closes) result["indicators"] = { "ema9": round(ema9_cur, 2), "ema21": round(ema21_cur, 2), "rsi": rsi, } if ema9_cur > ema21_cur and ema9_prev <= ema21_prev: result["signal"] = "BUY" elif ema9_cur < ema21_cur and ema9_prev >= ema21_prev: result["signal"] = "SELL" return result def _vwap(self, candles, closes, result): vwap = _Indicators.vwap(candles) cur = candles[-1] prev = candles[-2] if len(candles) >= 2 else cur prev_tp = (prev.high + prev.low + prev.close) / 3 result["indicators"] = {"vwap": vwap, "ltp": cur.close} if cur.close > vwap and prev_tp <= vwap: result["signal"] = "BUY" elif cur.close < vwap and prev_tp >= vwap: result["signal"] = "SELL" return result def _supertrend(self, candles, closes, result): trend_cur = _Indicators.supertrend_direction(candles) trend_prev = _Indicators.supertrend_direction(candles[:-1]) if len(candles) > 1 else "UP" rsi = _Indicators.rsi(closes) result["indicators"] = {"supertrend": trend_cur, "rsi": rsi} if trend_cur == "UP" and trend_prev == "DOWN": result["signal"] = "BUY" elif trend_cur == "DOWN" and trend_prev == "UP": result["signal"] = "SELL" return result def _macd(self, candles, closes, result): macd_cur, sig_cur = _Indicators.macd(closes) macd_prev, sig_prev = _Indicators.macd(closes[:-1]) if len(closes) > 1 else (macd_cur, sig_cur) result["indicators"] = {"macd": macd_cur, "signal": sig_cur} if macd_cur > sig_cur and macd_prev <= sig_prev: result["signal"] = "BUY" elif macd_cur < sig_cur and macd_prev >= sig_prev: result["signal"] = "SELL" return result def _rsi_reversal(self, candles, closes, result): rsi_cur = _Indicators.rsi(closes) rsi_prev = _Indicators.rsi(closes[:-1]) if len(closes) > 1 else rsi_cur result["indicators"] = {"rsi": rsi_cur} if rsi_prev < 30 and rsi_cur >= 30: result["signal"] = "BUY" elif rsi_prev > 70 and rsi_cur <= 70: result["signal"] = "SELL" return result # ═══════════════════════════════════════════════════════════════════════════════ # BacktestSession — one complete run # ═══════════════════════════════════════════════════════════════════════════════ class BacktestSession: """ One backtest session. Manages state, runs simulation in a background thread, stores all steps for replay navigation. """ def __init__(self, session_id: str, request: dict): self.session_id = session_id self.request = request self.mode = BacktestMode(request.get("mode", "QUICK")) self.state = BacktestState.IDLE self.error = "" self.progress = 0 # 0–100 # ── Live engine modules (reused) ───────────────────────────────── cfg = load_config() self.pool_manager = CapitalPoolManager() self.position_monitor = PositionMonitor() self.allocator = BrokerAllocator(cfg.broker_allocation) self.broker = BacktestBroker( base_slippage=request.get("slippage_pct", 0.0003) ) # ── Config from request ────────────────────────────────────────── self.initial_capital = float(request.get("capital", cfg.total_capital)) self.risk_pct = float(request.get("risk_pct", cfg.risk_pct)) self.strategies = request.get("strategies", cfg.strategies_enabled) self.use_ai = bool(request.get("use_ai", False)) self.ai_threshold = float(request.get("ai_threshold", cfg.consensus_threshold)) self.trading_end = request.get("trading_end", cfg.trading_end) self.max_daily_loss = float(request.get("max_daily_loss", cfg.max_daily_loss)) self.max_open_pos = int(request.get("max_open_positions", cfg.max_open_positions)) self.max_trades_day = int(request.get("max_trades_per_day", cfg.max_trades_per_day)) self.interval = request.get("interval", "1d") alloc_pct = { "INTRADAY": 30.0, "SWING": 25.0, "FNO": 20.0, "SCALPING": 10.0, "LONG_TERM": 15.0, } self.pool_manager.initialize(self.initial_capital, alloc_pct, cfg.broker_allocation) # ── Simulation state ───────────────────────────────────────────── self.trades: List[BacktestTrade] = [] self.equity_curve: List[dict] = [] self.replay_steps: List[ReplayStep] = [] self.current_step_idx = 0 self.report: dict = {} # ── Threading (QUICK / PORTFOLIO / PAPER_REPLAY run async) ─────── self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._pause_event = threading.Event() self._pause_event.set() # starts unpaused self._speed = 1.0 # candles/second for replay # ── Helpers ────────────────────────────────────────────────────── self._signal_engine = _StrategySignal() self._data_feed = BacktestDataFeed() self._reporter = BacktestReporter() self._indicator = _Indicators() self._candle_history: Dict[str, List[HistoricalCandle]] = {} # symbol → candles seen so far # ── Control ─────────────────────────────────────────────────────────────── def start(self): """Launch the simulation in a background thread.""" self.state = BacktestState.LOADING self._thread = threading.Thread(target=self._run, daemon=True) self._thread.start() def pause(self): self._pause_event.clear() self.state = BacktestState.PAUSED def resume(self): self._pause_event.set() self.state = BacktestState.RUNNING def stop(self): self._stop_event.set() self._pause_event.set() # unblock any wait def set_speed(self, speed: float): self._speed = max(0.1, float(speed)) def next_step(self) -> Optional[dict]: """Advance one step in REPLAY mode.""" idx = self.current_step_idx + 1 if idx < len(self.replay_steps): self.current_step_idx = idx return asdict(self.replay_steps[idx]) return None def prev_step(self) -> Optional[dict]: """Go back one step in REPLAY mode.""" idx = self.current_step_idx - 1 if idx >= 0: self.current_step_idx = idx return asdict(self.replay_steps[idx]) return None def current_state(self) -> dict: """Return current replay step state.""" if self.replay_steps and 0 <= self.current_step_idx < len(self.replay_steps): step = self.replay_steps[self.current_step_idx] return { "step": asdict(step), "step_idx": self.current_step_idx, "total_steps": len(self.replay_steps), "progress_pct": round(self.current_step_idx / max(len(self.replay_steps), 1) * 100, 1), } return {} def status(self) -> dict: trades_today = len(self.trades) equity = self.equity_curve[-1]["equity"] if self.equity_curve else self.initial_capital net_pnl = equity - self.initial_capital return { "session_id": self.session_id, "state": self.state.value, "mode": self.mode.value, "progress": self.progress, "total_trades": trades_today, "net_pnl": round(net_pnl, 2), "equity": round(equity, 2), "initial_capital": self.initial_capital, "candle_idx": self.current_step_idx, "total_candles": len(self.replay_steps), "error": self.error, "open_positions": len(self.pool_manager.all_positions()), } # ── Main simulation loop ────────────────────────────────────────────────── def _run(self): try: symbols = self._get_symbols() from_date = self.request.get("from_date") or self.request.get("start_date") to_date = self.request.get("to_date") or self.request.get("end_date") if not from_date or not to_date: raise ValueError("from_date and to_date required") logger.info( f"[Backtest:{self.session_id}] {self.mode} | " f"{len(symbols)} symbol(s) | {from_date}→{to_date} | {self.interval}" ) # ── Load historical data ────────────────────────────────────── self.state = BacktestState.LOADING all_candles = self._data_feed.fetch_multi(symbols, from_date, to_date, self.interval) # ── Build unified time-sorted event list ─────────────────────── # Each "event" = (timestamp, symbol, candle) events: List[tuple] = [] for sym, candles in all_candles.items(): for c in candles: events.append((c.timestamp, sym, c)) events.sort(key=lambda x: x[0]) if not events: raise RuntimeError("No historical data loaded for any symbol") total_events = len(events) self.state = BacktestState.RUNNING # Track daily equity snapshot dates last_equity_date = "" for idx, (ts, sym, candle) in enumerate(events): if self._stop_event.is_set(): break # Pause support self._pause_event.wait() # Replay speed throttle if self.mode in (BacktestMode.REPLAY, BacktestMode.PAPER_REPLAY): time.sleep(1.0 / self._speed) # ── Accumulate rolling history for this symbol ───────────── if sym not in self._candle_history: self._candle_history[sym] = [] self._candle_history[sym].append(candle) # ── Strategy signal (rolling window — NO future peeking) ─── step_scan: List[dict] = [] step_ai: List[dict] = [] step_risk: List[dict] = [] step_trades_open: List[dict] = [] step_trades_close: List[dict] = [] for strategy in self.strategies: history = self._candle_history[sym] sig_data = self._signal_engine.compute(strategy, history) signal = sig_data["signal"] indicators = sig_data["indicators"] indicators.update({ "atr": _Indicators.atr(history), "vwap": _Indicators.vwap(history), "rsi": _Indicators.rsi([c.close for c in history]), }) # Compute opportunity score (same formula as live scanner) chg_pct = candle.change_pct momentum_score = min(50.0, max(0.0, (abs(chg_pct) / 3.0) * 50.0)) vol_ratio = candle.volume / max(sum(c.volume for c in history[-5:]) / 5, 1) vol_score = min(30.0, vol_ratio * 15.0) if candle.high > candle.low: pos = (candle.close - candle.low) / (candle.high - candle.low) else: pos = 0.5 breakout_score = pos * 20.0 opp_score = min(100.0, momentum_score + vol_score + breakout_score) scan_entry = { "symbol": sym, "strategy": strategy, "signal": signal, "score": round(opp_score, 1), "ltp": candle.close, "change_pct": round(chg_pct, 2), "indicators": indicators, } step_scan.append(scan_entry) if signal not in ("BUY", "SELL"): continue # ── AI confirmation (optional) ───────────────────────── ai_signal = signal ai_confidence = opp_score ai_reason = "Technical signal" ai_model = "Technical" if self.use_ai: ai_result = self._call_ai(sym, candle, indicators) if ai_result: ai_signal = ai_result.get("signal", signal) ai_confidence = ai_result.get("confidence", opp_score) ai_reason = ai_result.get("reason", ai_reason) ai_model = ai_result.get("model", "AI") step_ai.append({ "symbol": sym, "strategy": strategy, "signal": ai_signal, "confidence": ai_confidence, "reason": ai_reason, "model": ai_model, }) if ai_confidence < self.ai_threshold: continue if ai_signal in ("HOLD", "NEUTRAL"): continue # ── Map strategy → pool style ────────────────────────── style = self._map_style(strategy) # ── Skip if already in this symbol ───────────────────── open_syms = {p.symbol for p in self.pool_manager.all_positions()} if sym in open_syms: continue # ── Daily trade limit ────────────────────────────────── daily_trades = sum( p.daily_trades for p in self.pool_manager.pools.values() ) if daily_trades >= self.max_trades_day: continue pool = self.pool_manager.get_pool(style) if pool is None: continue # ── Position size + targets (SAME as live engine) ────── sl, tp = self.allocator.calculate_targets(candle.close, ai_signal, style) qty, val = self.allocator.calculate_position_size( pool.available_capital, candle.close, self.risk_pct, sl, style ) if qty <= 0 or val <= 0: continue ok, reason = pool.can_trade(val, self.max_daily_loss, self.max_open_pos) if not ok: continue # ── Risk guard (same as live engine, market hours OFF) ── verdict = self._check_risk(sym, candle, val) step_risk.append({ "symbol": sym, "allowed": verdict["allowed"], "reason": verdict["reason"], }) if not verdict["allowed"]: continue # ── Place entry order ────────────────────────────────── broker = self.allocator.get_broker(style) fill = self.broker.place_order( symbol=sym, side=ai_signal, qty=qty, order_price=candle.close, order_type="MARKET", candle_volume=candle.volume, candle_high=candle.high, candle_low=candle.low, style=style, available_capital=pool.available_capital, ) if fill.status == "REJECTED": logger.debug(f"[BT] Entry rejected: {fill.reason}") continue # ── Create position (SAME as live engine) ────────────── pos = AutonomousPosition( symbol=sym, style=style, side=ai_signal, qty=fill.filled_qty, entry_price=fill.avg_price, stop_loss=sl, target_price=tp, broker=broker, order_id=fill.order_id, ) pos.ai_confidence = ai_confidence pos.strategy = strategy pos.opened_at = self._ts_to_epoch(candle.timestamp) pool.add_position(pos) # Track charges against realized PnL pool.realized_pnl -= fill.charges.total entry_dict = { "symbol": sym, "side": ai_signal, "qty": fill.filled_qty, "entry_price": fill.avg_price, "sl": sl, "tp": tp, "strategy": strategy, "style": style, "broker": broker, "charges": fill.charges.total, "order_id": fill.order_id, "ai_confidence": ai_confidence, "indicators": indicators, } step_trades_open.append(entry_dict) logger.debug( f"[BT] OPEN {ai_signal} {qty} {sym} @ ₹{fill.avg_price:.2f} " f"SL={sl:.2f} TP={tp:.2f} [{strategy}]" ) # ── Position monitoring (SL / Target / Trail / EOD) ─────── price_map = {sym: candle.close} # Check SL/Target using candle low (for BUY stop) / high (for SELL stop) price_map_sl = {sym: candle.low} price_map_tgt = {sym: candle.high} for pool in self.pool_manager.pools.values(): # First check using intrabar extreme prices for pos in list(pool.open_positions): if pos.symbol != sym: continue # BUY position: SL check on low, target on high if pos.side == "BUY": pool.update_ltp(sym, candle.low) actions_sl = self.position_monitor.monitor_pool( pool, price_map_sl, self.trading_end ) pool.update_ltp(sym, candle.high) actions_tgt = self.position_monitor.monitor_pool( pool, price_map_tgt, self.trading_end ) else: pool.update_ltp(sym, candle.high) actions_sl = self.position_monitor.monitor_pool( pool, price_map_sl, self.trading_end ) pool.update_ltp(sym, candle.low) actions_tgt = self.position_monitor.monitor_pool( pool, price_map_tgt, self.trading_end ) for action in actions_sl + actions_tgt: if action["action"] == "CLOSE": closed_pos = self._find_closed_pos(pool, action["pos_id"]) if closed_pos: trade = self._build_trade( pos_snapshot=None, pool=pool, sym=sym, candle=candle, action=action, entry_indicators=indicators if sym == action["symbol"] else {}, ai_model=ai_model if sym == action["symbol"] else "Technical", ai_confidence=ai_confidence, ai_reason=ai_reason, fill_entry=None, ) if trade: self.trades.append(trade) step_trades_close.append(trade.to_dict()) # ── Equity snapshot ──────────────────────────────────────── total_eq = self.initial_capital + \ self.pool_manager.total_realized_pnl() + \ self.pool_manager.total_unrealized_pnl() # Daily equity point if candle.date != last_equity_date: self.equity_curve.append({ "date": candle.date, "equity": round(total_eq, 2), "pnl": round(total_eq - self.initial_capital, 2), }) last_equity_date = candle.date # ── Replay step snapshot ─────────────────────────────────── step = ReplayStep( candle_idx=idx, candle=candle.to_dict(), scan_signals=step_scan, trades_opened=step_trades_open, trades_closed=step_trades_close, open_positions=[p.to_dict() for p in self.pool_manager.all_positions()], equity=round(total_eq, 2), cumulative_pnl=round(total_eq - self.initial_capital, 2), capital_pools=self.pool_manager.summary(), ai_decisions=step_ai, risk_verdicts=step_risk, ) self.replay_steps.append(step) self.current_step_idx = idx # Progress update self.progress = round((idx + 1) / total_events * 100) # ── Close all remaining positions at last price ──────────────── self._close_all_positions_eod() # Final equity point total_eq = self.initial_capital + \ self.pool_manager.total_realized_pnl() + \ self.pool_manager.total_unrealized_pnl() if self.equity_curve: self.equity_curve[-1]["equity"] = round(total_eq, 2) else: self.equity_curve.append({"date": to_date, "equity": round(total_eq, 2), "pnl": 0.0}) # ── Generate analytics report ───────────────────────────────── self.report = self._reporter.generate( trades=[t.to_dict() for t in self.trades], equity_curve=self.equity_curve, initial_capital=self.initial_capital, config=self.request, ) self.state = BacktestState.COMPLETED self.progress = 100 logger.info( f"[Backtest:{self.session_id}] COMPLETED — " f"{len(self.trades)} trades | " f"PnL ₹{self.report.get('net_pnl', 0):.2f}" ) except Exception as e: self.state = BacktestState.ERROR self.error = str(e) logger.error(f"[Backtest:{self.session_id}] ERROR: {e}") # ── Helper: get symbol list ─────────────────────────────────────────────── def _get_symbols(self) -> List[str]: mode = self.mode req = self.request if mode == BacktestMode.QUICK: sym = req.get("symbol") or req.get("symbols") if isinstance(sym, list): return [s.strip() for s in sym if s.strip()] return [str(sym).strip()] if sym else ["RELIANCE"] elif mode in (BacktestMode.PORTFOLIO, BacktestMode.REPLAY, BacktestMode.PAPER_REPLAY): syms = req.get("symbols", []) if isinstance(syms, str): syms = [s.strip() for s in syms.split(",") if s.strip()] if syms: return syms # Fallback: Nifty 50 top 10 liquid stocks return [ "RELIANCE", "TCS", "HDFCBANK", "INFY", "ICICIBANK", "SBIN", "WIPRO", "AXISBANK", "KOTAKBANK", "LT", ] return ["RELIANCE"] # ── Helper: map strategy → style ───────────────────────────────────────── def _map_style(self, strategy: str) -> str: mapping = { "EMA_CROSSOVER": "INTRADAY", "VWAP": "INTRADAY", "SUPERTREND": "SWING", "MACD": "SWING", "RSI_REVERSAL": "INTRADAY", } return mapping.get(strategy, "INTRADAY") # ── Helper: risk check (market-hours disabled for backtest) ─────────────── def _check_risk(self, sym: str, candle: HistoricalCandle, val: float) -> dict: try: from app.risk_guard.risk_guard import risk_guard, RiskConfig # Temporarily disable market-hours enforcement for backtest orig = risk_guard.config.enforce_market_hours risk_guard.config.enforce_market_hours = False open_pos_dicts = [ {"sector": getattr(p, "sector", "Other")} for p in self.pool_manager.all_positions() ] verdict = risk_guard.check_entry( symbol = sym, sector = "Other", ltp = candle.close, change_pct = candle.change_pct, vix = 0.0, news_sentiment = "NEUTRAL", open_positions = open_pos_dicts, capital = self.initial_capital, deployed = val, mode = "PAPER", ) risk_guard.config.enforce_market_hours = orig return {"allowed": verdict.allowed, "reason": verdict.reason} except Exception: return {"allowed": True, "reason": "Risk guard skipped"} # ── Helper: optional AI call ────────────────────────────────────────────── def _call_ai(self, sym: str, candle: HistoricalCandle, indicators: dict) -> Optional[dict]: try: from app.ai.ai_consensus import consensus_engine loop = asyncio.new_event_loop() result = loop.run_until_complete( consensus_engine.consensus( symbol=sym, market_data={ "ltp": candle.close, "open": candle.open, "high": candle.high, "low": candle.low, "change_pct": candle.change_pct, "volume": candle.volume, **indicators, }, ) ) loop.close() d = result.to_dict() return { "signal": d.get("signal", "HOLD"), "confidence": d.get("confidence", 0), "reason": d.get("reason", ""), "model": "AI-Consensus", } except Exception as e: logger.debug(f"[BT] AI call skipped: {e}") return None # ── Helper: find closed position by id ─────────────────────────────────── def _find_closed_pos(self, pool, pos_id: str) -> Optional[AutonomousPosition]: for p in pool._closed_positions: if p.id == pos_id: return p return None # ── Helper: build BacktestTrade from close action ───────────────────────── def _build_trade( self, pos_snapshot, pool, sym: str, candle: HistoricalCandle, action: dict, entry_indicators: dict, ai_model: str, ai_confidence: float, ai_reason: str, fill_entry, ) -> Optional[BacktestTrade]: try: closed = self._find_closed_pos(pool, action["pos_id"]) if not closed: return None exit_price = action["exit_price"] entry_price = closed.entry_price qty = closed.qty pnl = action["pnl"] pnl_pct = (pnl / (entry_price * qty)) * 100 if entry_price * qty > 0 else 0 # Exit fill charges exit_fill = self.broker.place_order( symbol=sym, side="SELL" if closed.side == "BUY" else "BUY", qty=qty, order_price=exit_price, order_type="MARKET", candle_volume=candle.volume, candle_high=candle.high, candle_low=candle.low, style=closed.style, ) total_charges = exit_fill.charges.total # subtract exit charges from realized PnL pool.realized_pnl -= exit_fill.charges.total entry_time = datetime.fromtimestamp(closed.opened_at).isoformat() exit_time = candle.timestamp[:19] try: e1 = datetime.fromisoformat(entry_time[:19]) e2 = datetime.fromisoformat(exit_time[:19]) hold_h = (e2 - e1).total_seconds() / 3600 except Exception: hold_h = 0.0 return BacktestTrade( trade_id = f"BT-{uuid.uuid4().hex[:8]}", symbol = sym, side = closed.side, style = closed.style, strategy = closed.strategy, sector = "Other", entry_price = round(entry_price, 2), exit_price = round(exit_price, 2), qty = qty, entry_time = entry_time, exit_time = exit_time, exit_reason = action["reason"], pnl = round(pnl - total_charges, 2), pnl_pct = round(pnl_pct, 3), charges = round(total_charges, 2), charges_detail = exit_fill.charges.to_dict(), slippage_pct = exit_fill.slippage_pct, entry_indicators = entry_indicators, ai_model = ai_model, ai_confidence = ai_confidence, ai_reason = ai_reason, risk_score = 0.0, broker = closed.broker, entry_order_id = closed.order_id, exit_order_id = exit_fill.order_id, market_personality = "UNKNOWN", holding_hours = round(hold_h, 2), ) except Exception as e: logger.warning(f"[BT] _build_trade error: {e}") return None # ── Helper: EOD close ───────────────────────────────────────────────────── def _close_all_positions_eod(self): """Close all remaining open positions at last known price.""" for pool in self.pool_manager.pools.values(): for pos in list(pool.open_positions): pnl = pool.close_position(pos.id, pos.ltp, "EOD_FINAL") if pnl is not None and self.trades: # Update last trade if it matches pass # ── Helper: timestamp string to epoch float ─────────────────────────────── @staticmethod def _ts_to_epoch(ts: str) -> float: try: return datetime.fromisoformat(ts[:19]).timestamp() except Exception: return time.time() # ═══════════════════════════════════════════════════════════════════════════════ # BacktestManager — manages multiple concurrent sessions # ═══════════════════════════════════════════════════════════════════════════════ class BacktestManager: """ Manages multiple concurrent backtest sessions. Provides start / control / report / compare API. """ def __init__(self): self._sessions: Dict[str, BacktestSession] = {} self._lock = threading.Lock() def start(self, request: dict) -> dict: session_id = f"BT-{uuid.uuid4().hex[:12].upper()}" session = BacktestSession(session_id, request) with self._lock: self._sessions[session_id] = session session.start() return {"success": True, "session_id": session_id, "message": "Backtest started"} def pause(self, session_id: str) -> dict: s = self._get(session_id) s.pause() return {"success": True, "session_id": session_id, "state": "PAUSED"} def resume(self, session_id: str) -> dict: s = self._get(session_id) s.resume() return {"success": True, "session_id": session_id, "state": "RUNNING"} def stop(self, session_id: str) -> dict: s = self._get(session_id) s.stop() return {"success": True, "session_id": session_id} def set_speed(self, session_id: str, speed: float) -> dict: s = self._get(session_id) s.set_speed(speed) return {"success": True, "speed": speed} def next_candle(self, session_id: str) -> dict: s = self._get(session_id) step = s.next_step() return {"success": True, "step": step} def prev_candle(self, session_id: str) -> dict: s = self._get(session_id) step = s.prev_step() return {"success": True, "step": step} def current_state(self, session_id: str) -> dict: s = self._get(session_id) return {"success": True, **s.current_state()} def status(self, session_id: str) -> dict: s = self._get(session_id) return {"success": True, **s.status()} def get_report(self, session_id: str) -> dict: s = self._get(session_id) if s.state != BacktestState.COMPLETED: return {"success": False, "message": f"Not completed yet: {s.state.value}"} return {"success": True, "report": s.report} def get_journal(self, session_id: str, limit: int = 200) -> dict: s = self._get(session_id) trades = [t.to_dict() for t in s.trades[-limit:]] return {"success": True, "trades": trades, "count": len(s.trades)} def list_sessions(self) -> dict: with self._lock: sessions = [s.status() for s in self._sessions.values()] return {"success": True, "sessions": sessions} def compare(self, session_id_a: str, session_id_b: str) -> dict: """ Compare two backtest sessions (or compare backtest vs paper/live). Returns side-by-side metric comparison. """ sa = self._get(session_id_a) sb = self._get(session_id_b) def _metrics(s: BacktestSession) -> dict: r = s.report if s.report else {} return { "session_id": s.session_id, "mode": s.mode.value, "state": s.state.value, "total_trades": r.get("total_trades", len(s.trades)), "win_rate": r.get("win_rate", 0), "net_pnl": r.get("net_pnl", 0), "return_pct": r.get("return_pct", 0), "sharpe_ratio": r.get("sharpe_ratio", 0), "sortino_ratio": r.get("sortino_ratio", 0), "max_drawdown": r.get("max_drawdown_pct",0), "profit_factor": r.get("profit_factor", 0), "expectancy": r.get("expectancy", 0), "total_charges": r.get("total_charges", 0), } ma = _metrics(sa) mb = _metrics(sb) diff = {} for key in ("net_pnl", "win_rate", "return_pct", "sharpe_ratio", "max_drawdown", "profit_factor", "expectancy"): diff[key] = round(ma.get(key, 0) - mb.get(key, 0), 3) return { "success": True, "session_a": ma, "session_b": mb, "difference": diff, "verdict": "Session A outperforms" if ma["net_pnl"] > mb["net_pnl"] else "Session B outperforms", } def delete_session(self, session_id: str) -> dict: with self._lock: if session_id in self._sessions: self._sessions[session_id].stop() del self._sessions[session_id] return {"success": True} def _get(self, session_id: str) -> BacktestSession: s = self._sessions.get(session_id) if not s: raise KeyError(f"Session {session_id} not found") return s # ─── Singleton ───────────────────────────────────────────────────────────────── backtest_manager = BacktestManager()