Spaces:
Running on Zero
Running on Zero
deveshc23
feat: Add Free Gradio Space support, Git LFS tracking for data artifacts, and Quant Allocator
3e33e2e | """Event-driven long-only portfolio simulation with next-bar-open execution.""" | |
| from __future__ import annotations | |
| import math | |
| from collections.abc import Callable | |
| from dataclasses import asdict, dataclass | |
| from datetime import datetime | |
| from statistics import fmean | |
| from typing import Any | |
| from uuid import uuid4 | |
| from drift_quant.backtesting.strategies import Strategy | |
| from drift_quant.nse_data.loader import LoadedNseDataset | |
| class BacktestConfig: | |
| """Visible simulation assumptions; costs are configurable research assumptions.""" | |
| initial_capital: float = 1_000_000.0 | |
| max_gross_exposure: float = 0.95 | |
| transaction_cost_bps: float = 10.0 | |
| slippage_bps: float = 5.0 | |
| allow_overnight: bool = True | |
| force_close_at_end: bool = True | |
| def __post_init__(self) -> None: | |
| if self.initial_capital <= 0: | |
| raise ValueError("Initial capital must be positive.") | |
| if not 0 < self.max_gross_exposure <= 1: | |
| raise ValueError("Maximum gross exposure must be in (0, 1].") | |
| if self.transaction_cost_bps < 0 or self.slippage_bps < 0: | |
| raise ValueError("Cost assumptions must not be negative.") | |
| class Trade: | |
| symbol: str | |
| entry_time: datetime | |
| exit_time: datetime | |
| quantity: int | |
| entry_price: float | |
| exit_price: float | |
| gross_pnl: float | |
| costs: float | |
| net_pnl: float | |
| exit_reason: str | |
| class EquityPoint: | |
| timestamp: datetime | |
| equity: float | |
| gross_exposure: float | |
| class BacktestResult: | |
| experiment_id: str | |
| strategy_name: str | |
| strategy_version: str | |
| dataset_id: str | |
| config: BacktestConfig | |
| trades: tuple[Trade, ...] | |
| equity_curve: tuple[EquityPoint, ...] | |
| metrics: dict[str, float | int | None] | |
| def to_dict(self, max_equity_points: int | None = 2500) -> dict[str, Any]: | |
| points = list(self.equity_curve) | |
| if max_equity_points is not None and len(points) > max_equity_points: | |
| step = (len(points) - 1) / (max_equity_points - 1) | |
| indices = {round(i * step) for i in range(max_equity_points)} | |
| points = [p for i, p in enumerate(points) if i in indices] | |
| return { | |
| "experiment_id": self.experiment_id, | |
| "strategy_name": self.strategy_name, | |
| "strategy_version": self.strategy_version, | |
| "dataset_id": self.dataset_id, | |
| "config": asdict(self.config), | |
| "trades": [ | |
| { | |
| **asdict(trade), | |
| "entry_time": trade.entry_time.isoformat(sep=" "), | |
| "exit_time": trade.exit_time.isoformat(sep=" "), | |
| } | |
| for trade in self.trades | |
| ], | |
| "equity_curve": [ | |
| { | |
| **asdict(point), | |
| "timestamp": point.timestamp.isoformat(sep=" "), | |
| } | |
| for point in points | |
| ], | |
| "metrics": self.metrics, | |
| } | |
| class _Position: | |
| quantity: int | |
| entry_time: datetime | |
| entry_price: float | |
| entry_cost: float | |
| class _PendingOrder: | |
| symbol: str | |
| target_long: bool | |
| decision_time: datetime | |
| def _side_adjusted_price(open_price: float, is_buy: bool, slippage_bps: float) -> float: | |
| adjustment = slippage_bps / 10_000 | |
| return open_price * (1 + adjustment if is_buy else 1 - adjustment) | |
| def _commission(notional: float, cost_bps: float) -> float: | |
| return notional * cost_bps / 10_000 | |
| def _mark_equity( | |
| cash: float, positions: dict[str, _Position], latest_prices: dict[str, float] | |
| ) -> tuple[float, float]: | |
| gross_exposure = sum( | |
| position.quantity * latest_prices[symbol] | |
| for symbol, position in positions.items() | |
| if symbol in latest_prices | |
| ) | |
| return cash + gross_exposure, gross_exposure | |
| def _daily_equity(points: tuple[EquityPoint, ...]) -> list[float]: | |
| last_by_date: dict[object, float] = {} | |
| for point in points: | |
| last_by_date[point.timestamp.date()] = point.equity | |
| return [last_by_date[key] for key in sorted(last_by_date)] | |
| def _metrics( | |
| initial_capital: float, points: tuple[EquityPoint, ...], trades: tuple[Trade, ...], costs: float | |
| ) -> dict[str, float | int | None]: | |
| if not points: | |
| return { | |
| "cumulative_return": 0.0, | |
| "annualized_return": None, | |
| "annualized_volatility": None, | |
| "sharpe_ratio": None, | |
| "sortino_ratio": None, | |
| "maximum_drawdown": 0.0, | |
| "trade_count": 0, | |
| "transaction_costs": costs, | |
| } | |
| final_equity = points[-1].equity | |
| cumulative_return = final_equity / initial_capital - 1 | |
| daily_equity = _daily_equity(points) | |
| daily_returns = [ | |
| daily_equity[index] / daily_equity[index - 1] - 1 | |
| for index in range(1, len(daily_equity)) | |
| if daily_equity[index - 1] > 0 | |
| ] | |
| annualized_return: float | None = None | |
| annualized_volatility: float | None = None | |
| sharpe: float | None = None | |
| sortino: float | None = None | |
| if len(daily_equity) >= 2 and daily_equity[0] > 0: | |
| years = max((points[-1].timestamp - points[0].timestamp).days / 365.25, 1 / 365.25) | |
| annualized_return = (final_equity / initial_capital) ** (1 / years) - 1 | |
| if len(daily_returns) >= 2: | |
| mean_return = fmean(daily_returns) | |
| variance = fmean([(value - mean_return) ** 2 for value in daily_returns]) | |
| annualized_volatility = math.sqrt(variance) * math.sqrt(252) | |
| if annualized_volatility > 0: | |
| sharpe = mean_return / math.sqrt(variance) * math.sqrt(252) | |
| downside = [min(value, 0.0) for value in daily_returns] | |
| downside_deviation = math.sqrt(fmean([value**2 for value in downside])) | |
| if downside_deviation > 0: | |
| sortino = mean_return / downside_deviation * math.sqrt(252) | |
| peak = initial_capital | |
| maximum_drawdown = 0.0 | |
| for point in points: | |
| peak = max(peak, point.equity) | |
| maximum_drawdown = min(maximum_drawdown, point.equity / peak - 1) | |
| wins = [trade.net_pnl for trade in trades if trade.net_pnl > 0] | |
| losses = [trade.net_pnl for trade in trades if trade.net_pnl < 0] | |
| gross_profit = sum(wins) | |
| gross_loss = abs(sum(losses)) | |
| holding_minutes = [ | |
| (trade.exit_time - trade.entry_time).total_seconds() / 60 for trade in trades | |
| ] | |
| return { | |
| "cumulative_return": cumulative_return, | |
| "annualized_return": annualized_return, | |
| "annualized_volatility": annualized_volatility, | |
| "sharpe_ratio": sharpe, | |
| "sortino_ratio": sortino, | |
| "maximum_drawdown": maximum_drawdown, | |
| "trade_count": len(trades), | |
| "win_rate": len(wins) / len(trades) if trades else None, | |
| "average_win": fmean(wins) if wins else None, | |
| "average_loss": fmean(losses) if losses else None, | |
| "profit_factor": gross_profit / gross_loss if gross_loss else None, | |
| "average_holding_minutes": fmean(holding_minutes) if holding_minutes else None, | |
| "transaction_costs": costs, | |
| "final_equity": final_equity, | |
| } | |
| def run_backtest( | |
| dataset: LoadedNseDataset, | |
| strategy: Strategy, | |
| config: BacktestConfig, | |
| progress_callback: Callable[[float, datetime, float, int, Trade | None], None] | None = None, | |
| ) -> BacktestResult: | |
| """Run completed-bar decisions with fills only at the next available bar open. | |
| The engine allocates the allowed gross exposure equally across the configured | |
| symbols. It does not use current-bar close to fill an order. | |
| """ | |
| all_events = sorted( | |
| (bar for bars in dataset.bars_by_symbol.values() for bar in bars), | |
| key=lambda bar: (bar.timestamp, bar.symbol), | |
| ) | |
| if not all_events: | |
| raise ValueError("Dataset has no valid bars for the selected universe.") | |
| allocation_per_symbol = ( | |
| config.initial_capital * config.max_gross_exposure / len(dataset.symbols) | |
| ) | |
| cash = config.initial_capital | |
| positions: dict[str, _Position] = {} | |
| pending: dict[str, _PendingOrder] = {} | |
| latest_prices: dict[str, float] = {} | |
| trades: list[Trade] = [] | |
| equity_curve: list[EquityPoint] = [] | |
| total_costs = 0.0 | |
| total_events = len(all_events) | |
| report_interval = max(200, total_events // 60) | |
| current_ts: datetime | None = None | |
| last_trade: Trade | None = None | |
| for idx, bar in enumerate(all_events): | |
| latest_prices[bar.symbol] = bar.close | |
| order = pending.pop(bar.symbol, None) | |
| if order is not None: | |
| if order.target_long and bar.symbol not in positions: | |
| fill_price = _side_adjusted_price(bar.open, True, config.slippage_bps) | |
| quantity = int(allocation_per_symbol // fill_price) | |
| notional = quantity * fill_price | |
| cost = _commission(notional, config.transaction_cost_bps) | |
| if quantity > 0 and cash >= notional + cost: | |
| cash -= notional + cost | |
| total_costs += cost | |
| positions[bar.symbol] = _Position( | |
| quantity=quantity, | |
| entry_time=bar.timestamp, | |
| entry_price=fill_price, | |
| entry_cost=cost, | |
| ) | |
| elif not order.target_long and bar.symbol in positions: | |
| position = positions.pop(bar.symbol) | |
| fill_price = _side_adjusted_price(bar.open, False, config.slippage_bps) | |
| notional = position.quantity * fill_price | |
| exit_cost = _commission(notional, config.transaction_cost_bps) | |
| cash += notional - exit_cost | |
| total_costs += exit_cost | |
| gross_pnl = position.quantity * (fill_price - position.entry_price) | |
| costs = position.entry_cost + exit_cost | |
| new_trade = Trade( | |
| symbol=bar.symbol, | |
| entry_time=position.entry_time, | |
| exit_time=bar.timestamp, | |
| quantity=position.quantity, | |
| entry_price=position.entry_price, | |
| exit_price=fill_price, | |
| gross_pnl=gross_pnl, | |
| costs=costs, | |
| net_pnl=gross_pnl - costs, | |
| exit_reason="strategy_signal", | |
| ) | |
| trades.append(new_trade) | |
| last_trade = new_trade | |
| desired = strategy.target(bar.symbol, bar) | |
| if desired is not None: | |
| is_long = bar.symbol in positions | |
| existing_pending = pending.get(bar.symbol) | |
| if desired != is_long and ( | |
| existing_pending is None or existing_pending.target_long != desired | |
| ): | |
| pending[bar.symbol] = _PendingOrder( | |
| symbol=bar.symbol, target_long=desired, decision_time=bar.timestamp | |
| ) | |
| if current_ts is None or bar.timestamp != current_ts: | |
| equity, exposure = _mark_equity(cash, positions, latest_prices) | |
| equity_curve.append( | |
| EquityPoint(timestamp=bar.timestamp, equity=equity, gross_exposure=exposure) | |
| ) | |
| current_ts = bar.timestamp | |
| if progress_callback is not None and (idx % report_interval == 0): | |
| pct = round((idx + 1) / total_events * 100, 1) | |
| progress_callback(pct, bar.timestamp, equity, len(trades), last_trade) | |
| if config.force_close_at_end: | |
| for symbol, position in list(positions.items()): | |
| bar = dataset.bars_by_symbol[symbol][-1] | |
| # Explicit end-of-data liquidation assumption; it is never treated as a next-bar fill. | |
| fill_price = _side_adjusted_price(bar.close, False, config.slippage_bps) | |
| notional = position.quantity * fill_price | |
| exit_cost = _commission(notional, config.transaction_cost_bps) | |
| cash += notional - exit_cost | |
| total_costs += exit_cost | |
| gross_pnl = position.quantity * (fill_price - position.entry_price) | |
| costs = position.entry_cost + exit_cost | |
| new_trade = Trade( | |
| symbol=symbol, | |
| entry_time=position.entry_time, | |
| exit_time=bar.timestamp, | |
| quantity=position.quantity, | |
| entry_price=position.entry_price, | |
| exit_price=fill_price, | |
| gross_pnl=gross_pnl, | |
| costs=costs, | |
| net_pnl=gross_pnl - costs, | |
| exit_reason="end_of_data_close_assumption", | |
| ) | |
| trades.append(new_trade) | |
| last_trade = new_trade | |
| del positions[symbol] | |
| final_time = max(bar.timestamp for bar in all_events) | |
| equity_curve.append(EquityPoint(timestamp=final_time, equity=cash, gross_exposure=0.0)) | |
| if progress_callback is not None: | |
| final_eq = equity_curve[-1].equity if equity_curve else cash | |
| progress_callback(100.0, all_events[-1].timestamp, final_eq, len(trades), last_trade) | |
| immutable_points = tuple(equity_curve) | |
| immutable_trades = tuple(trades) | |
| return BacktestResult( | |
| experiment_id=str(uuid4()), | |
| strategy_name=strategy.name, | |
| strategy_version=strategy.version, | |
| dataset_id=dataset.dataset_id, | |
| config=config, | |
| trades=immutable_trades, | |
| equity_curve=immutable_points, | |
| metrics=_metrics(config.initial_capital, immutable_points, immutable_trades, total_costs), | |
| ) | |