""" CV6 AI Trading OS — Backtest Manager (Legacy compatibility layer) Used by the /backtest/run legacy endpoint. Uses REAL historical data via BacktestDataFeed. Same logic as the new BacktestEngine — only the API shape differs. """ from __future__ import annotations import math from datetime import datetime, timedelta from typing import List from loguru import logger from app.backtest.backtest_models import BacktestRequest, BacktestResponse, TradeLog from app.backtest.backtest_data_feed import BacktestDataFeed, HistoricalCandle from app.backtest.backtest_broker import BacktestBroker # ── Indicator helpers (same as BacktestEngine._Indicators) ──────────────────── 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) v = sum(closes[:period]) / period for c in closes[period:]: v = c * k + v * (1 - k) return v def _ema_series(closes: List[float], period: int) -> List[float]: if len(closes) < period: return closes[:] k = 2.0 / (period + 1) result = [sum(closes[:period]) / period] for c in closes[period:]: result.append(c * k + result[-1] * (1 - k)) return [closes[0]] * (period - 1) + result 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) def _compute_signal(strategy: str, candles: List[HistoricalCandle], idx: int) -> str: """Compute signal at candle index `idx` using only candles 0..idx (no peeking).""" window = candles[:idx + 1] closes = [c.close for c in window] n = len(closes) if strategy in ("EMA", "EMA_CROSSOVER", "EMA CROSSOVER"): if n < 25: return "HOLD" e9_prev = _ema(closes[:-1], 9) e21_prev = _ema(closes[:-1], 21) e9_cur = _ema(closes, 9) e21_cur = _ema(closes, 21) if e9_cur > e21_cur and e9_prev <= e21_prev: return "BUY" if e9_cur < e21_cur and e9_prev >= e21_prev: return "SELL" elif strategy in ("RSI", "RSI_REVERSAL", "RSI REVERSAL"): if n < 16: return "HOLD" rsi_prev = _rsi(closes[:-1]) rsi_cur = _rsi(closes) if rsi_prev < 30 and rsi_cur >= 30: return "BUY" if rsi_prev > 70 and rsi_cur <= 70: return "SELL" elif strategy in ("MACD", "MACD_MOMENTUM", "MACD MOMENTUM"): if n < 35: return "HOLD" ema12 = _ema_series(closes, 12) ema26 = _ema_series(closes, 26) macd = [e12 - e26 for e12, e26 in zip(ema12, ema26)] sig = _ema_series(macd, 9) if len(macd) < 2: return "HOLD" if macd[-1] > sig[-1] and macd[-2] <= sig[-2]: return "BUY" if macd[-1] < sig[-1] and macd[-2] >= sig[-2]: return "SELL" elif strategy in ("VWAP",): if n < 2: return "HOLD" cum_pv = sum((c.high + c.low + c.close) / 3 * c.volume for c in window) cum_v = sum(c.volume for c in window) vwap = cum_pv / cum_v if cum_v > 0 else closes[-1] prev_tp = (window[-2].high + window[-2].low + window[-2].close) / 3 if closes[-1] > vwap and prev_tp <= vwap: return "BUY" if closes[-1] < vwap and prev_tp >= vwap: return "SELL" elif strategy in ("SUPERTREND", "AI", "AI STRATEGY"): if n < 10: return "HOLD" period = 7 mult = 3.0 highs = [c.high for c in window] lows = [c.low for c in window] tr = [highs[0] - lows[0]] for i in range(1, n): 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 lower = hl2 - mult * atr trend_cur = "UP" if closes[-1] >= lower else "DOWN" # previous if n >= 2: atr_p = sum(tr[-(period+1):-1]) / period hl2_p = (highs[-2] + lows[-2]) / 2 lower_p = hl2_p - mult * atr_p trend_prev = "UP" if closes[-2] >= lower_p else "DOWN" else: trend_prev = trend_cur if trend_cur == "UP" and trend_prev == "DOWN": return "BUY" if trend_cur == "DOWN" and trend_prev == "UP": return "SELL" return "HOLD" # ── Legacy BacktestManager ──────────────────────────────────────────────────── class BacktestManager: """ Legacy backtest manager. Uses REAL historical data from yfinance via BacktestDataFeed. Returns BacktestResponse for backward-compatible /backtest/run endpoint. """ def run_backtest(self, request: BacktestRequest) -> BacktestResponse: symbol = (request.symbol or "RELIANCE").upper() strategy = (request.strategy or request.strategy_name or "EMA_CROSSOVER").upper() capital = float(request.initial_capital or request.capital or 100_000.0) risk_pct = float(request.risk_percent or 1.0) # Date range try: end_dt = datetime.strptime(request.to_date or request.end_date or "", "%Y-%m-%d") except (ValueError, TypeError): end_dt = datetime.now() try: start_dt = datetime.strptime(request.from_date or request.start_date or "", "%Y-%m-%d") except (ValueError, TypeError): start_dt = end_dt - timedelta(days=365) from_date = start_dt.strftime("%Y-%m-%d") to_date = end_dt.strftime("%Y-%m-%d") # Fetch real candles feed = BacktestDataFeed() broker = BacktestBroker() try: candles = feed.fetch_candles(symbol, from_date, to_date, "1d") except Exception as e: logger.error(f"[LegacyBT] Data fetch failed: {e} — returning empty result") candles = [] if not candles: return BacktestResponse( symbol=symbol, strategy=strategy, total_trades=0, winning_trades=0, losing_trades=0, win_rate=0.0, win_rate_pct=0.0, net_profit=0.0, net_pnl=0.0, total_pnl=0.0, return_pct=0.0, max_drawdown=0.0, sharpe_ratio=0.0, cagr=0.0, initial_capital=capital, final_capital=capital, profit_factor=None, sortino_ratio=None, expectancy=None, avg_win=None, avg_loss=None, largest_win=None, largest_loss=None, recovery_factor=None, total_charges=0.0, ) # ── Simulation ──────────────────────────────────────────────────── cash = capital position = 0 entry_price = 0.0 entry_candle = None trades: List[TradeLog] = [] equity_curve = [capital] total_charges = 0.0 for idx, candle in enumerate(candles): sig = _compute_signal(strategy, candles, idx) if sig == "BUY" and position == 0: risk_amount = cash * (risk_pct / 100) sl = candle.close * 0.985 sl_dist = abs(candle.close - sl) qty = max(1, int(risk_amount / sl_dist)) if sl_dist > 0 else 1 qty = min(qty, int(cash * 0.3 / candle.close)) # max 30% per trade qty = max(1, qty) fill = broker.place_order( symbol=symbol, side="BUY", qty=qty, order_price=candle.close, order_type="MARKET", candle_volume=candle.volume, candle_high=candle.high, candle_low=candle.low, ) if fill.status != "REJECTED": cost = fill.filled_qty * fill.avg_price + fill.charges.total if cost <= cash: cash -= cost position = fill.filled_qty entry_price = fill.avg_price entry_candle = candle total_charges += fill.charges.total elif sig == "SELL" and position > 0: fill = broker.place_order( symbol=symbol, side="SELL", qty=position, order_price=candle.close, order_type="MARKET", candle_volume=candle.volume, candle_high=candle.high, candle_low=candle.low, ) if fill.status != "REJECTED": proceeds = fill.filled_qty * fill.avg_price - fill.charges.total pnl = proceeds - (entry_price * fill.filled_qty) pnl_pct = (pnl / (entry_price * fill.filled_qty)) * 100 if entry_price > 0 else 0 cash += proceeds trades.append(TradeLog( date=candle.date, side="LONG", entry=round(entry_price, 2), exit=round(fill.avg_price, 2), qty=fill.filled_qty, pnl=round(pnl, 2), pnl_pct=round(pnl_pct, 2), )) total_charges += fill.charges.total position = 0 entry_price = 0.0 equity_curve.append(cash + position * candle.close) # Close open position at last candle if position > 0 and candles: last = candles[-1] fill = broker.place_order( symbol=symbol, side="SELL", qty=position, order_price=last.close, order_type="MARKET", candle_volume=last.volume, candle_high=last.high, candle_low=last.low, ) proceeds = fill.filled_qty * fill.avg_price - fill.charges.total pnl = proceeds - (entry_price * fill.filled_qty) pnl_pct = (pnl / (entry_price * fill.filled_qty)) * 100 if entry_price > 0 else 0 cash += proceeds trades.append(TradeLog( date=last.date, side="LONG", entry=round(entry_price, 2), exit=round(fill.avg_price, 2), qty=fill.filled_qty, pnl=round(pnl, 2), pnl_pct=round(pnl_pct, 2), )) total_charges += fill.charges.total equity_curve.append(cash) final_capital = cash net_pnl = final_capital - capital return_pct = (net_pnl / capital) * 100 if capital > 0 else 0.0 winning = [t for t in trades if t.pnl > 0] losing = [t for t in trades if t.pnl < 0] win_rate = len(winning) / len(trades) * 100 if trades else 0.0 # Max drawdown peak = equity_curve[0] max_dd = 0.0 for val in equity_curve: peak = max(peak, val) dd = (peak - val) / peak * 100 if peak > 0 else 0 max_dd = max(max_dd, dd) # Sharpe (annualised) sharpe = 0.0 if len(equity_curve) > 1: dr = [(equity_curve[i] - equity_curve[i-1]) / equity_curve[i-1] for i in range(1, len(equity_curve)) if equity_curve[i-1] > 0] if dr: mean_r = sum(dr) / len(dr) var_r = sum((r - mean_r) ** 2 for r in dr) / len(dr) std_r = math.sqrt(var_r) if var_r > 0 else 1e-9 sharpe = round((mean_r / std_r) * math.sqrt(252), 4) # Sortino (uses downside deviation only) sortino = None if len(equity_curve) > 1: dr = [(equity_curve[i] - equity_curve[i-1]) / equity_curve[i-1] for i in range(1, len(equity_curve)) if equity_curve[i-1] > 0] if dr: mean_r = sum(dr) / len(dr) down = [r for r in dr if r < 0] if down: dd_var = sum(r ** 2 for r in down) / len(down) dd_std = math.sqrt(dd_var) if dd_var > 0 else 1e-9 sortino = round((mean_r / dd_std) * math.sqrt(252), 4) # CAGR n_days = max(len(candles), 1) years = max(n_days / 252, 0.01) cagr = ((final_capital / capital) ** (1 / years) - 1) * 100 if capital > 0 else 0.0 # Profit factor = gross wins / |gross losses| profit_factor = None gross_win = sum(t.pnl for t in winning) gross_loss = abs(sum(t.pnl for t in losing)) if gross_loss > 0: profit_factor = round(gross_win / gross_loss, 4) elif gross_win > 0: profit_factor = 999.0 # all wins, no losses # Per-trade averages avg_win_val = round(gross_win / len(winning), 2) if winning else None avg_loss_val = round(sum(t.pnl for t in losing) / len(losing), 2) if losing else None largest_win_val = round(max((t.pnl for t in winning), default=None) or 0, 2) if winning else None largest_loss_val= round(min((t.pnl for t in losing), default=None) or 0, 2) if losing else None # Expectancy = (win_rate * avg_win) + (loss_rate * avg_loss) expectancy = None if trades: wr = len(winning) / len(trades) lr = 1.0 - wr aw = avg_win_val or 0.0 al = avg_loss_val or 0.0 expectancy = round(wr * aw + lr * al, 2) # Recovery factor = net_pnl / max_drawdown_in_rupees recovery_factor = None if max_dd > 0 and capital > 0: max_dd_inr = capital * max_dd / 100 if max_dd_inr > 0: recovery_factor = round(net_pnl / max_dd_inr, 4) return BacktestResponse( symbol=symbol, strategy=strategy, total_trades=len(trades), winning_trades=len(winning), losing_trades=len(losing), win_rate=round(win_rate, 2), win_rate_pct=round(win_rate, 2), net_profit=round(net_pnl, 2), net_pnl=round(net_pnl, 2), total_pnl=round(net_pnl, 2), return_pct=round(return_pct, 2), max_drawdown=round(max_dd, 2), sharpe_ratio=round(sharpe, 2), cagr=round(cagr, 2), initial_capital=capital, final_capital=round(final_capital, 2), total_charges=round(total_charges, 2), profit_factor=profit_factor, sortino_ratio=sortino, expectancy=expectancy, avg_win=avg_win_val, avg_loss=avg_loss_val, largest_win=largest_win_val, largest_loss=largest_loss_val, recovery_factor=recovery_factor, trades_list=trades[-50:], )