""" Backtesting engine for evaluating strategies on historical data. """ import logging from datetime import datetime from typing import Optional import pandas as pd from src.models import CoinConfig, Position, Side, Signal, SignalType, TradeRecord from src.strategy import EnvelopeStrategy logger = logging.getLogger("crypto_bot") class Backtester: """ Backtesting engine that simulates trading on historical OHLCV data. Produces trade logs, equity curves, and performance metrics. """ def __init__(self, config: CoinConfig, initial_balance: float = 1000, leverage: int = 1, open_fee_rate: float = 0.0002, close_fee_rate: float = 0.0006): self.config = config self.initial_balance = initial_balance self.leverage = leverage self.open_fee_rate = open_fee_rate self.close_fee_rate = close_fee_rate self.strategy = EnvelopeStrategy(config) self.balance = initial_balance self.position: Optional[Position] = None self.trades: list = [] self.equity_records: list = [] def run(self, df: pd.DataFrame) -> dict: """ Run backtest on historical data. Args: df: OHLCV DataFrame with datetime index Returns: Dict with metrics and trade data """ logger.info(f"Running backtest on {self.config.symbol} ({len(df)} candles)") self.balance = self.initial_balance self.position = None self.trades = [] self.equity_records = [] min_period = self.config.average_period + 2 for i in range(min_period, len(df)): window = df.iloc[:i + 1] current_time = df.index[i] current_close = df["close"].iloc[i] # Generate signals signals = self.strategy.generate_signals(window, self.position) # Process signals for signal in signals: signal.timestamp = current_time self._process_signal(signal, current_close, current_time) # Record equity periodically (every 6 candles) if i % 6 == 0: equity = self._compute_equity(current_close) self.equity_records.append({ "timestamp": current_time, "equity": equity, "price": current_close, "balance": self.balance, }) # Close any remaining position at last price if self.position and self.position.is_open: last_price = df["close"].iloc[-1] self._close_position(last_price, df.index[-1], "End of backtest") # Final equity record last_price = df["close"].iloc[-1] self.equity_records.append({ "timestamp": df.index[-1], "equity": self.balance, "price": last_price, "balance": self.balance, }) return self._compute_metrics(df) def _process_signal(self, signal: Signal, current_price: float, timestamp: datetime): """Process a trading signal.""" if signal.type == SignalType.ENTRY: self._open_entry(signal, timestamp) elif signal.type in (SignalType.EXIT, SignalType.STOP_LOSS, SignalType.CLOSE_ALL): if self.position and self.position.is_open: self._close_position(current_price, timestamp, signal.reason) # Block re-entry after stop-loss or close-all if signal.type in (SignalType.STOP_LOSS, SignalType.CLOSE_ALL): if self.position is None: self.position = Position( symbol=self.config.symbol, side=signal.side, blocked_reentry=True, ) else: self.position.blocked_reentry = True def _open_entry(self, signal: Signal, timestamp: datetime): """Open or add to a position.""" capital = self.config.capital_allocation if self.balance < capital: capital = self.balance size_per_envelope = (capital / self.config.average_period if self.strategy.num_envelopes == 0 else capital / self.strategy.num_envelopes) # Apply leverage effective_size = size_per_envelope * self.leverage # Entry fee fee = effective_size * self.open_fee_rate if self.position is None or not self.position.is_open: self.position = Position( symbol=self.config.symbol, side=signal.side, open_time=timestamp, ) self.position.add_entry(signal.price, effective_size) self.balance -= fee def _close_position(self, close_price: float, timestamp: datetime, reason: str): """Close the current position and record the trade.""" if not self.position or not self.position.is_open: return # Create trade record trade = TradeRecord( symbol=self.position.symbol, side=self.position.side, open_time=self.position.open_time, close_time=timestamp, entry_prices=list(self.position.entry_prices), entry_sizes=list(self.position.entry_sizes), exit_price=close_price, close_reason=reason, ) # Calculate close fee close_fee = self.position.total_size * self.close_fee_rate trade.fees = ( sum(s * self.open_fee_rate for s in self.position.entry_sizes) + close_fee ) trade.calculate_pnl() self.balance += trade.pnl + sum(self.position.entry_sizes) # Return capital + pnl self.trades.append(trade) # Reset position self.position = Position( symbol=self.config.symbol, side=self.position.side, blocked_reentry=self.position.blocked_reentry, ) def _compute_equity(self, current_price: float) -> float: """Compute total portfolio value including unrealized P&L.""" equity = self.balance if self.position and self.position.is_open: self.position.update_unrealized_pnl(current_price) equity += self.position.unrealized_pnl + sum(self.position.entry_sizes) return equity def _compute_metrics(self, df: pd.DataFrame) -> dict: """Compute backtest performance metrics.""" if not self.trades: return {"error": "No trades executed"} pnls = [t.pnl for t in self.trades] pnl_pcts = [t.pnl_pct for t in self.trades] wins = [t for t in self.trades if t.is_win] losses = [t for t in self.trades if not t.is_win] # Equity curve for Sharpe & drawdown eq_df = pd.DataFrame(self.equity_records) eq_df.set_index("timestamp", inplace=True) # Returns for Sharpe ratio eq_returns = eq_df["equity"].pct_change().dropna() sharpe = 0.0 if len(eq_returns) > 1 and eq_returns.std() > 0: # Annualized (assuming 6-hour equity updates → ~1460 periods per year) sharpe = (eq_returns.mean() / eq_returns.std()) * (1460 ** 0.5) # Max drawdown eq_peak = eq_df["equity"].expanding().max() drawdown = (eq_df["equity"] - eq_peak) / eq_peak max_drawdown = drawdown.min() * 100 # Trade durations durations = [] for t in self.trades: if t.open_time and t.close_time: dur = (t.close_time - t.open_time).total_seconds() / 3600 durations.append(dur) # Hodling performance first_price = df["close"].iloc[self.config.average_period + 2] last_price = df["close"].iloc[-1] hodl_return = ((last_price - first_price) / first_price) * 100 total_fees = sum(t.fees for t in self.trades) metrics = { "symbol": self.config.symbol, "period": f"{df.index[0]} to {df.index[-1]}", "initial_balance": self.initial_balance, "final_balance": round(self.balance, 2), "total_return_pct": round(((self.balance - self.initial_balance) / self.initial_balance) * 100, 2), "hodl_return_pct": round(hodl_return, 2), "leverage": self.leverage, "total_trades": len(self.trades), "winning_trades": len(wins), "losing_trades": len(losses), "win_rate_pct": round(len(wins) / len(self.trades) * 100, 2) if self.trades else 0, "avg_pnl": round(sum(pnls) / len(pnls), 2), "avg_pnl_pct": round(sum(pnl_pcts) / len(pnl_pcts), 2), "avg_win_pnl": round(sum(t.pnl for t in wins) / len(wins), 2) if wins else 0, "avg_loss_pnl": round(sum(t.pnl for t in losses) / len(losses), 2) if losses else 0, "best_trade_pnl": round(max(pnls), 2), "worst_trade_pnl": round(min(pnls), 2), "max_drawdown_pct": round(max_drawdown, 2), "sharpe_ratio": round(sharpe, 2), "avg_trade_duration_hours": round(sum(durations) / len(durations), 1) if durations else 0, "total_fees": round(total_fees, 2), "profit_factor": round( abs(sum(t.pnl for t in wins)) / abs(sum(t.pnl for t in losses)), 2 ) if losses and sum(t.pnl for t in losses) != 0 else float("inf"), } return metrics def get_equity_dataframe(self) -> pd.DataFrame: """Return equity curve as a DataFrame.""" return pd.DataFrame(self.equity_records).set_index("timestamp") def get_trades_dataframe(self) -> pd.DataFrame: """Return trade log as a DataFrame.""" records = [] for t in self.trades: records.append({ "open_time": t.open_time, "close_time": t.close_time, "side": t.side.value, "avg_entry_price": round(t.avg_entry_price, 4), "exit_price": round(t.exit_price, 4), "total_size": round(t.total_size, 4), "pnl": round(t.pnl, 2), "pnl_pct": round(t.pnl_pct, 2), "fees": round(t.fees, 4), "close_reason": t.close_reason, "is_win": t.is_win, "envelopes_hit": len(t.entry_prices), }) return pd.DataFrame(records) def print_metrics(self, metrics: dict): """Pretty-print backtest metrics.""" print("\n" + "=" * 60) print(f" BACKTEST RESULTS — {metrics['symbol']}") print("=" * 60) print(f" Period: {metrics['period']}") print(f" Initial Balance: ${metrics['initial_balance']:,.2f}") print(f" Final Balance: ${metrics['final_balance']:,.2f}") print(f" Total Return: {metrics['total_return_pct']:+.2f}%") print(f" HODL Return: {metrics['hodl_return_pct']:+.2f}%") print(f" Leverage: {metrics['leverage']}x") print("-" * 60) print(f" Total Trades: {metrics['total_trades']}") print(f" Win Rate: {metrics['win_rate_pct']:.1f}%") print(f" Avg P&L/Trade: ${metrics['avg_pnl']:+.2f} ({metrics['avg_pnl_pct']:+.2f}%)") print(f" Best Trade: ${metrics['best_trade_pnl']:+.2f}") print(f" Worst Trade: ${metrics['worst_trade_pnl']:+.2f}") print("-" * 60) print(f" Max Drawdown: {metrics['max_drawdown_pct']:.2f}%") print(f" Sharpe Ratio: {metrics['sharpe_ratio']:.2f}") print(f" Profit Factor: {metrics['profit_factor']:.2f}") print(f" Avg Duration: {metrics['avg_trade_duration_hours']:.1f} hours") print(f" Total Fees: ${metrics['total_fees']:.2f}") print("=" * 60 + "\n")