""" ========================================= CV6 AI Trading OS — Backtest Broker Simulates realistic order execution: • Market / Limit / SL orders • Slippage (volume-weighted impact) • Partial fills (low-volume stocks) • Rejected orders (circuit / capital) • Full Indian market charges: Brokerage ₹20/trade flat STT 0.025% intraday sell | 0.1% delivery Exchange 0.00345% (NSE) GST 18% on services Stamp 0.015% on buy SEBI ₹10 per crore ========================================= """ from __future__ import annotations import uuid from dataclasses import dataclass, field from typing import Optional from loguru import logger # ── Charges breakdown ───────────────────────────────────────────────────────── @dataclass class BrokerCharges: brokerage: float = 0.0 stt: float = 0.0 exchange: float = 0.0 gst: float = 0.0 stamp: float = 0.0 sebi: float = 0.0 @property def total(self) -> float: return self.brokerage + self.stt + self.exchange + self.gst + self.stamp + self.sebi def to_dict(self) -> dict: return { "brokerage": round(self.brokerage, 2), "stt": round(self.stt, 2), "exchange": round(self.exchange, 2), "gst": round(self.gst, 2), "stamp": round(self.stamp, 2), "sebi": round(self.sebi, 4), "total": round(self.total, 2), } # ── Order fill result ───────────────────────────────────────────────────────── @dataclass class BrokerFill: order_id: str symbol: str side: str # BUY | SELL qty: int # requested filled_qty: int # actually filled avg_price: float # fill price (after slippage) order_price: float # requested price status: str # FILLED | PARTIAL | REJECTED reason: str = "" slippage_pct: float = 0.0 charges: BrokerCharges = field(default_factory=BrokerCharges) @property def turnover(self) -> float: return self.filled_qty * self.avg_price @property def gross_value(self) -> float: """Value without charges.""" return self.filled_qty * self.avg_price @property def net_cost(self) -> float: """Buy: turnover + charges. Sell: turnover - charges.""" return self.turnover + self.charges.total if self.side == "BUY" else self.turnover - self.charges.total def to_dict(self) -> dict: return { "order_id": self.order_id, "symbol": self.symbol, "side": self.side, "qty": self.qty, "filled_qty": self.filled_qty, "avg_price": round(self.avg_price, 2), "order_price": round(self.order_price, 2), "status": self.status, "reason": self.reason, "turnover": round(self.turnover, 2), "slippage_pct":round(self.slippage_pct * 100, 4), "charges": self.charges.to_dict(), } # ── BacktestBroker ──────────────────────────────────────────────────────────── class BacktestBroker: """ Simulates broker execution for backtesting. All Indian market charges are computed per SEBI/NSE schedule (2024): - Brokerage: ₹20/order OR 0.03% of turnover (lower of the two) - STT: 0.025% on SELL turnover (intraday) / 0.1% both (delivery) - Exchange: 0.00345% of turnover (NSE equity) - GST: 18% on (brokerage + exchange) - Stamp duty: 0.015% on BUY turnover (max ₹10,000) - SEBI: ₹10 per crore of turnover """ # Brokerage (discount broker) BROKERAGE_FLAT = 20.0 # ₹20 per order BROKERAGE_MAX_PCT = 0.0003 # 0.03% cap # STT STT_INTRADAY_SELL = 0.00025 # 0.025% sell side only STT_DELIVERY = 0.001 # 0.1% both sides # NSE exchange transaction charge EXCHANGE_RATE = 0.0000345 # 0.00345% # GST on (brokerage + exchange charge) GST_RATE = 0.18 # Stamp duty (buy side) STAMP_RATE = 0.00015 # 0.015% STAMP_MAX = 10_000.0 # ₹10,000 per day per instrument # SEBI SEBI_RATE = 10.0 / 1_00_00_000 # ₹10 per crore def __init__(self, base_slippage: float = 0.0003): """ base_slippage: fraction (0.0003 = 0.03% default) """ self.base_slippage = base_slippage # ── Public API ──────────────────────────────────────────────────────────── def place_order( self, symbol: str, side: str, # BUY | SELL qty: int, order_price: float, order_type: str, # MARKET | LIMIT | SL candle_volume: int, # current candle volume for impact calc candle_high: float, # for SL BUY / LIMIT SELL checks candle_low: float, # for SL SELL / LIMIT BUY checks style: str = "INTRADAY", available_capital: float = 0.0, ) -> BrokerFill: """ Simulate a single order. Returns BrokerFill with charges. """ order_id = f"BT-{uuid.uuid4().hex[:8].upper()}" # ── Hard rejections ─────────────────────────────────────────────── if qty <= 0: return self._rejected(order_id, symbol, side, qty, order_price, "Invalid qty (≤0)") if order_price <= 0: return self._rejected(order_id, symbol, side, qty, order_price, "Invalid price (≤0)") if available_capital > 0 and qty * order_price > available_capital * 1.1: return self._rejected(order_id, symbol, side, qty, order_price, f"Insufficient margin (need ₹{qty*order_price:.0f}, have ₹{available_capital:.0f})") # ── LIMIT order: check if price was touched this candle ─────────── if order_type == "LIMIT": if side == "BUY" and candle_low > order_price: return self._rejected(order_id, symbol, side, qty, order_price, f"Limit BUY ₹{order_price} not touched (low={candle_low:.2f})") if side == "SELL" and candle_high < order_price: return self._rejected(order_id, symbol, side, qty, order_price, f"Limit SELL ₹{order_price} not touched (high={candle_high:.2f})") # ── SL order: check if SL was triggered ─────────────────────────── if order_type == "SL": if side == "BUY" and candle_high < order_price: return self._rejected(order_id, symbol, side, qty, order_price, f"SL-BUY ₹{order_price} not triggered (high={candle_high:.2f})") if side == "SELL" and candle_low > order_price: return self._rejected(order_id, symbol, side, qty, order_price, f"SL-SELL ₹{order_price} not triggered (low={candle_low:.2f})") # ── Partial fill simulation ──────────────────────────────────────── filled_qty = qty if candle_volume > 0 and qty * 50 > candle_volume: fill_ratio = min(1.0, candle_volume / (qty * 50)) filled_qty = max(1, int(qty * fill_ratio)) # ── Slippage ────────────────────────────────────────────────────── slip = self.base_slippage if order_type == "MARKET": slip *= 1.5 # market order has more slippage if candle_volume > 0 and qty > candle_volume * 0.005: slip *= 1.8 # large-order market impact direction = +1.0 if side == "BUY" else -1.0 fill_price = round(order_price * (1 + direction * slip), 2) fill_price = max(0.01, fill_price) # ── Charges ─────────────────────────────────────────────────────── charges = self._compute_charges(side, fill_price, filled_qty, style) status = "FILLED" if filled_qty >= qty else "PARTIAL" logger.debug( f"[Broker] {order_id} {side} {filled_qty}/{qty} {symbol} @ ₹{fill_price:.2f} " f"slip={slip*100:.3f}% charges=₹{charges.total:.2f}" ) return BrokerFill( order_id=order_id, symbol=symbol, side=side, qty=qty, filled_qty=filled_qty, avg_price=fill_price, order_price=order_price, status=status, slippage_pct=slip, charges=charges, ) # ── Charge computation ──────────────────────────────────────────────────── def _compute_charges( self, side: str, price: float, qty: int, style: str, ) -> BrokerCharges: turnover = price * qty # Brokerage — flat ₹20 or 0.03% (lower) brokerage = min(self.BROKERAGE_FLAT, turnover * self.BROKERAGE_MAX_PCT) # STT is_delivery = style in ("SWING", "LONG_TERM") if is_delivery: stt = turnover * self.STT_DELIVERY else: stt = (turnover * self.STT_INTRADAY_SELL) if side == "SELL" else 0.0 # Exchange exchange = turnover * self.EXCHANGE_RATE # GST on services gst = (brokerage + exchange) * self.GST_RATE # Stamp duty (buy only) stamp = 0.0 if side == "BUY": stamp = min(turnover * self.STAMP_RATE, self.STAMP_MAX) # SEBI sebi = turnover * self.SEBI_RATE return BrokerCharges( brokerage=round(brokerage, 4), stt=round(stt, 4), exchange=round(exchange, 4), gst=round(gst, 4), stamp=round(stamp, 4), sebi=round(sebi, 6), ) def _rejected( self, order_id, symbol, side, qty, price, reason ) -> BrokerFill: logger.debug(f"[Broker] REJECTED {order_id} {side} {qty} {symbol} — {reason}") return BrokerFill( order_id=order_id, symbol=symbol, side=side, qty=qty, filled_qty=0, avg_price=0.0, order_price=price, status="REJECTED", reason=reason, charges=BrokerCharges(), ) # Singleton backtest_broker = BacktestBroker()