""" AkDevz AI Market Intelligence — Intraday Quantitative Backtester Strict zero-lookahead, walk-forward compatible simulator with realistic Indian exchange brokerage, STT, turnover charges, GST, and slippage. """ from typing import List, Dict, Any, Optional import numpy as np from python.intelligence.risk_engine import RiskEngine class Backtester: def __init__( self, brokerage_per_order: float = 20.0, slippage_pts: float = 1.5, lot_size: int = 25 ): self.brokerage = brokerage_per_order self.slippage_pts = slippage_pts self.lot_size = lot_size def run_backtest( self, candles: List[Dict[str, Any]], strategy_name: str = "VWAP_PULLBACK", risk_reward_target: float = 2.0 ) -> Dict[str, Any]: """ Executes event-driven backtest over historical candles with zero lookahead bias. """ if len(candles) < 30: return { "strategy": strategy_name, "total_trades": 0, "win_rate_pct": 0.0, "gross_pnl": 0.0, "net_pnl": 0.0, "profit_factor": 0.0, "max_drawdown_pct": 0.0, "trades": [], "message": "Insufficient candle data for backtesting (minimum 30 bars required)." } trades = [] in_trade = False entry_price = 0.0 stop_loss = 0.0 target_price = 0.0 direction = "BUY" entry_time = None trade_high = 0.0 trade_low = 0.0 # Run through candles bar by bar (no future lookahead) for i in range(25, len(candles)): curr_bar = candles[i] prev_bar = candles[i - 1] c = float(curr_bar["close"]) h = float(curr_bar["high"]) l = float(curr_bar["low"]) t = curr_bar.get("timestamp", i) # Manage existing open trade if in_trade: trade_high = max(trade_high, h) trade_low = min(trade_low, l) exit_type = None exit_price = 0.0 if direction == "BUY": if l <= stop_loss: exit_type = "STOP_LOSS" exit_price = stop_loss - self.slippage_pts elif h >= target_price: exit_type = "TARGET" exit_price = target_price - self.slippage_pts else: # SELL if h >= stop_loss: exit_type = "STOP_LOSS" exit_price = stop_loss + self.slippage_pts elif l <= target_price: exit_type = "TARGET" exit_price = target_price + self.slippage_pts # End of day mandatory square-off at last bar if not exit_type and i == len(candles) - 1: exit_type = "EOD_SQUAREOFF" exit_price = c if exit_type: gross_pts = (exit_price - entry_price) if direction == "BUY" else (entry_price - exit_price) gross_pnl = gross_pts * self.lot_size # Calculate exact taxes & fees cost_info = RiskEngine.calculate_rr( entry_price=entry_price, stop_loss=stop_loss, target_price=target_price, lot_size=self.lot_size, slippage_pts=self.slippage_pts ) total_costs = cost_info["total_estimated_costs"] net_pnl = gross_pnl - total_costs # Max Favorable / Adverse Excursions mfe = (trade_high - entry_price) if direction == "BUY" else (entry_price - trade_low) mae = (entry_price - trade_low) if direction == "BUY" else (trade_high - entry_price) trades.append({ "entry_time": entry_time, "exit_time": t, "direction": direction, "entry_price": round(entry_price, 2), "exit_price": round(exit_price, 2), "exit_reason": exit_type, "gross_pnl": round(gross_pnl, 2), "costs": round(total_costs, 2), "net_pnl": round(net_pnl, 2), "is_win": net_pnl > 0, "mfe_pts": round(mfe, 2), "mae_pts": round(mae, 2) }) in_trade = False # Check new setup triggers only if not in trade if not in_trade: # Example rule: Close crosses above 20-bar High with momentum prev_highs = [float(candles[k]["high"]) for k in range(i - 20, i)] breakout_level = max(prev_highs) if c > breakout_level and float(prev_bar["close"]) <= breakout_level: in_trade = True direction = "BUY" entry_price = c + self.slippage_pts sl_dist = max(15.0, (entry_price - float(prev_bar["low"]))) stop_loss = entry_price - sl_dist target_price = entry_price + (sl_dist * risk_reward_target) entry_time = t trade_high = h trade_low = l # Performance summary metrics total_trades = len(trades) if total_trades == 0: return { "strategy": strategy_name, "total_trades": 0, "win_rate_pct": 0.0, "gross_pnl": 0.0, "net_pnl": 0.0, "profit_factor": 0.0, "max_drawdown_pct": 0.0, "trades": [], "message": "Strategy triggered 0 trades over historical period." } wins = [t for t in trades if t["is_win"]] losses = [t for t in trades if not t["is_win"]] win_rate = (len(wins) / total_trades) * 100.0 gross_pnl_total = sum(t["gross_pnl"] for t in trades) net_pnl_total = sum(t["net_pnl"] for t in trades) gross_wins = sum(t["net_pnl"] for t in wins) gross_losses = abs(sum(t["net_pnl"] for t in losses)) profit_factor = (gross_wins / gross_losses) if gross_losses > 0 else (99.0 if gross_wins > 0 else 0.0) # Drawdown calculation equity_curve = np.cumsum([t["net_pnl"] for t in trades]) peak = np.maximum.accumulate(equity_curve) drawdowns = peak - equity_curve max_dd = float(np.max(drawdowns)) if len(drawdowns) > 0 else 0.0 expectancy = (net_pnl_total / total_trades) if total_trades > 0 else 0.0 return { "strategy": strategy_name, "total_trades": total_trades, "winning_trades": len(wins), "losing_trades": len(losses), "win_rate_pct": round(win_rate, 1), "gross_pnl": round(gross_pnl_total, 2), "net_pnl": round(net_pnl_total, 2), "profit_factor": round(profit_factor, 2), "net_expectancy_per_trade": round(expectancy, 2), "max_drawdown_pnl": round(max_dd, 2), "trades": trades }