"""Backtesting simulation engine.""" import logging import sys import os import numpy as np import pandas as pd # Add parent dir to path so we can import project modules sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import config import indicators import risk_manager from backtest import signal_generator logger = logging.getLogger(__name__) WARMUP_BARS = 100 FEE_PCT = 0.003 # 0.2% fee + 0.1% slippage # Backtest-specific overrides (tighter for ranging markets) BT_ATR_SL_MIN_PCT = 0.02 # 3% → 2% BT_ATR_SL_MAX_PCT = 0.05 # 8% → 5% # Trailing stop thresholds TRAILING_BREAKEVEN_PCT = 0.03 # ≥3% profit: move stop to entry (breakeven) TRAILING_ATR_PCT = 0.05 # ≥5% profit: trailing stop = peak - 2×ATR # Position management MAX_CONCURRENT_SYMBOLS = 3 # max 3 symbols at once MAX_ADDON_PER_SYMBOL = 1 # max 1 add-on per symbol BUY_COOLDOWN_BARS = 48 # 4 hours cooldown after buy TIMEOUT_UNCONDITIONAL_BARS = 864 # 72h: close unconditionally MIN_PROFIT_FOR_SIGNAL_SELL = 0.03 # 3% min profit for signal sell MAX_DAILY_BUYS_PER_SYMBOL = 1 # max 1 new position per symbol per day def _calc_atr_stop_price(entry_price: float, atr: float) -> float: atr_pct = (atr * config.ATR_SL_MULTIPLIER) / entry_price bounded_pct = max(BT_ATR_SL_MIN_PCT, min(BT_ATR_SL_MAX_PCT, atr_pct)) return round(entry_price * (1 - bounded_pct), 8) def _calc_all_daily_pivots(htf_df: pd.DataFrame) -> dict: """Calculate pivot points for each day from 1h candles. Returns {date_str: {"pivot": ..., "r1": ..., "s1": ..., ...}}. """ if htf_df.empty or len(htf_df) < 24: return {} df = htf_df.copy() df["date"] = df["timestamp"].dt.date pivots_by_date = {} dates = sorted(df["date"].unique()) for i in range(1, len(dates)): prev_day = df[df["date"] == dates[i - 1]] if prev_day.empty: continue h = prev_day["high"].max() l = prev_day["low"].min() c = prev_day["close"].iloc[-1] pivot = (h + l + c) / 3 pivots_by_date[str(dates[i])] = { "pivot": pivot, "r1": 2 * pivot - l, "r2": pivot + (h - l), "r3": h + 2 * (pivot - l), "s1": 2 * pivot - h, "s2": pivot - (h - l), "s3": l - 2 * (pivot - l), } return pivots_by_date class BacktestEngine: def __init__(self, starting_capital: float = 10000.0): self.starting_capital = starting_capital self.portfolio = { "total_balance_usdt": starting_capital, "available_usdt": starting_capital, "positions": {}, "trade_history": [], "initial_capital": starting_capital, } self.stop_orders = {} # symbol -> {"stop_price", "entry_price", "amount"} self.trade_log = [] # all executed trades self.equity_curve = [] # (timestamp, equity) self.buy_cooldown = {} # symbol -> bar_index of last buy self.addon_count = {} # symbol -> number of add-ons self.entry_bar = {} # symbol -> bar_index of first entry self.trailing_stop = {} # symbol -> peak_price for trailing stop self.daily_buys = {} # (symbol, date_str) -> count def run(self, data: dict[str, dict], btc_perp_1h: pd.DataFrame | None = None, use_context: bool = True) -> dict: """Run backtest on historical data. Args: data: {symbol: {"candles_5m": DataFrame, "candles_1h": DataFrame}} btc_perp_1h: BTC perpetual 1h candles for funding/basis calculation. use_context: If False, disable all context filters (for A/B testing). Returns: Results dict with metrics, trades, equity curve. """ # Step 1: Pre-compute indicators for all symbols ind_5m = {} ind_1h = {} pivots_by_sym = {} for sym, d in data.items(): df_5m = d["candles_5m"] df_1h = d["candles_1h"] if df_5m.empty or len(df_5m) < WARMUP_BARS: logger.warning("Skipping %s: insufficient 5m data (%d bars)", sym, len(df_5m)) continue ind_5m[sym] = indicators.calculate_indicators(df_5m) if not df_1h.empty and len(df_1h) >= 30: ind_1h[sym] = indicators.calculate_htf_indicators(df_1h) pivots_by_sym[sym] = _calc_all_daily_pivots(df_1h) else: ind_1h[sym] = pd.DataFrame() pivots_by_sym[sym] = {} if not ind_5m: logger.error("No symbols with sufficient data") return self._build_results() # Step 2: Build unified 5m timeline (use longest series) ref_sym = max(ind_5m, key=lambda s: len(ind_5m[s])) timeline = ind_5m[ref_sym]["timestamp"].values # Pre-extract 1h timestamps for searchsorted htf_ts = {} for sym in ind_1h: if not ind_1h[sym].empty: htf_ts[sym] = ind_1h[sym]["timestamp"].values # Pre-extract BTC 1h and perp timestamps for context btc_sym = "tBTCUST" btc_htf_ts = htf_ts.get(btc_sym) btc_htf_df = ind_1h.get(btc_sym) perp_ts = None perp_df = btc_perp_1h if perp_df is not None and not perp_df.empty: perp_ts = perp_df["timestamp"].values # Step 3: Iterate total_bars = len(timeline) log_interval = max(1, total_bars // 20) for i in range(WARMUP_BARS, total_bars): ts = timeline[i] if i % log_interval == 0: pct = i / total_bars * 100 logger.info("Progress: %.0f%% (%s)", pct, pd.Timestamp(ts)) current_prices = {} signals = [] # --- Build context dict (before symbol loop) --- context = None if use_context: context = {} # BTC trend context if btc_htf_ts is not None and btc_htf_df is not None and len(btc_htf_ts) > 0: btc_idx = np.searchsorted(btc_htf_ts, ts, side="right") - 1 if btc_idx >= 0: btc_row = btc_htf_df.iloc[btc_idx] btc_ema9 = btc_row.get("ema9", 0) btc_ema21 = btc_row.get("ema21", 0) btc_adx = btc_row.get("adx", 0) ema9_gt_ema21 = btc_ema9 > btc_ema21 if pd.notna(btc_ema9) and pd.notna(btc_ema21) else True btc_adx_val = btc_adx if pd.notna(btc_adx) else 0 context["btc_trend"] = { "ema9_gt_ema21": ema9_gt_ema21, "adx": btc_adx_val, "bearish_confirmed": (not ema9_gt_ema21) and btc_adx_val > 20, } # Funding sentiment (BTC perp basis) if perp_ts is not None and btc_htf_ts is not None: perp_idx = np.searchsorted(perp_ts, ts, side="right") - 1 btc_spot_idx = np.searchsorted(btc_htf_ts, ts, side="right") - 1 if perp_idx >= 0 and btc_spot_idx >= 0: perp_close = perp_df.iloc[perp_idx].get("close", 0) spot_close = btc_htf_df.iloc[btc_spot_idx].get("close", 0) if spot_close > 0 and perp_close > 0: basis_pct = (perp_close - spot_close) / spot_close if basis_pct > 0.0005: signal_label = "overleveraged_long" elif basis_pct < -0.0005: signal_label = "fearful" else: signal_label = "neutral" context["funding_sentiment"] = { "basis_pct": basis_pct, "signal": signal_label, } for sym in ind_5m: df = ind_5m[sym] if i >= len(df): continue row = df.iloc[i] close = row.get("close", 0) if close <= 0: continue current_prices[sym] = close # Check stop-loss (including trailing stop) if sym in self.stop_orders: low = row.get("low", close) stop = self.stop_orders[sym] if low <= stop["stop_price"]: # Determine if this was a trailing stop exit exit_type = "stop_loss" if sym in self.trailing_stop: exit_type = "trailing_stop" self._execute_stop_loss(sym, stop, ts, exit_type) continue # Update trailing stop for open positions pos = self.portfolio.get("positions", {}).get(sym, {}) if pos.get("amount", 0) > 0 and sym in self.stop_orders: entry = pos.get("entry_price", 0) high = row.get("high", close) atr = row.get("atr", 0) if entry > 0: # Track peak price if sym not in self.trailing_stop: self.trailing_stop[sym] = high else: self.trailing_stop[sym] = max(self.trailing_stop[sym], high) peak = self.trailing_stop[sym] profit_pct = (peak - entry) / entry new_stop = self.stop_orders[sym]["stop_price"] if profit_pct >= TRAILING_ATR_PCT and atr > 0: # ≥5% profit: trailing stop = peak - 2×ATR trail_stop = peak - 2 * atr new_stop = max(new_stop, trail_stop) elif profit_pct >= TRAILING_BREAKEVEN_PCT: # ≥3% profit: move stop to entry (breakeven) new_stop = max(new_stop, entry) if new_stop > self.stop_orders[sym]["stop_price"]: self.stop_orders[sym]["stop_price"] = round(new_stop, 8) # Check position timeout (72h unconditional) if pos.get("amount", 0) > 0 and sym in self.entry_bar: bars_held = i - self.entry_bar[sym] if bars_held >= TIMEOUT_UNCONDITIONAL_BARS: self._execute_timeout(sym, pos, close, ts, "timeout_72h") continue # Generate signal if i < 1: continue prev_row = df.iloc[i - 1] # Find latest 1h context htf_last = None if sym in htf_ts and len(htf_ts[sym]) > 0: idx = np.searchsorted(htf_ts[sym], ts, side="right") - 1 if idx >= 0: htf_last = ind_1h[sym].iloc[idx] # Find today's pivots ts_date = str(pd.Timestamp(ts).date()) pivots = pivots_by_sym.get(sym, {}).get(ts_date) has_pos = self.portfolio.get("positions", {}).get(sym, {}).get("amount", 0) > 0 # Add per-symbol buy_pressure to context sym_context = context if context is not None: sym_context = dict(context) sym_context["buy_pressure"] = row.get("buy_pressure", 0.5) sig = signal_generator.generate_signal(sym, row, prev_row, htf_last, pivots, has_position=has_pos, context=sym_context) if sig["action"] != "HOLD": # Attach ATR for stop-loss calculation sig["atr"] = row.get("atr", 0) # Apply cooldown filter for BUY signals if sig["action"] == "BUY" and sym in self.buy_cooldown: if i - self.buy_cooldown[sym] < BUY_COOLDOWN_BARS: continue # Apply min profit filter for signal SELL if sig["action"] == "SELL" and has_pos: entry = pos.get("entry_price", 0) pnl_pct = (close - entry) / entry if entry > 0 else 0 # Allow sell if 1h strong bearish, otherwise require min profit if pnl_pct < MIN_PROFIT_FOR_SIGNAL_SELL: htf_last_check = None if sym in htf_ts and len(htf_ts[sym]) > 0: idx = np.searchsorted(htf_ts[sym], ts, side="right") - 1 if idx >= 0: htf_last_check = ind_1h[sym].iloc[idx] is_strong_bearish = (htf_last_check is not None and pd.notna(htf_last_check.get("ema9")) and htf_last_check["ema9"] < htf_last_check["ema21"] and htf_last_check.get("adx", 0) > 25) if not is_strong_bearish: continue signals.append(sig) # Sort signals by confidence (high first) to prioritize capital allocation signals.sort(key=lambda s: s.get("confidence", 0), reverse=True) for sig in signals: self._process_signal(sig, current_prices, ts, i) # Record equity equity = self.portfolio["available_usdt"] for sym, pos in self.portfolio.get("positions", {}).items(): if pos.get("amount", 0) > 0: px = current_prices.get(sym, pos.get("entry_price", 0)) equity += pos["amount"] * px self.equity_curve.append((ts, equity)) return self._build_results() def _execute_stop_loss(self, sym: str, stop: dict, ts, exit_type: str = "stop_loss"): amount = stop["amount"] stop_price = stop["stop_price"] sell_price = stop_price * (1 - FEE_PCT) entry = stop["entry_price"] realized_pnl = amount * sell_price - amount * entry realized_pnl_pct = (sell_price - entry) / entry * 100 if entry > 0 else 0 # Update portfolio self.portfolio["available_usdt"] += amount * sell_price self.portfolio["positions"].pop(sym, None) del self.stop_orders[sym] self.entry_bar.pop(sym, None) self.addon_count.pop(sym, None) self.trailing_stop.pop(sym, None) self.trade_log.append({ "timestamp": str(pd.Timestamp(ts)), "symbol": sym, "action": "SELL", "type": exit_type, "amount": amount, "price": stop_price, "effective_price": sell_price, "amount_usdt": amount * sell_price, "entry_price": entry, "realized_pnl": realized_pnl, "realized_pnl_pct": realized_pnl_pct, }) def _execute_timeout(self, sym: str, pos: dict, close: float, ts, timeout_type: str): amount = pos["amount"] entry = pos["entry_price"] sell_price = close * (1 - FEE_PCT) realized_pnl = amount * sell_price - amount * entry realized_pnl_pct = (sell_price - entry) / entry * 100 if entry > 0 else 0 self.portfolio["available_usdt"] += amount * sell_price self.portfolio["positions"].pop(sym, None) self.stop_orders.pop(sym, None) self.entry_bar.pop(sym, None) self.addon_count.pop(sym, None) self.trailing_stop.pop(sym, None) self.trade_log.append({ "timestamp": str(pd.Timestamp(ts)), "symbol": sym, "action": "SELL", "type": timeout_type, "amount": amount, "price": close, "effective_price": sell_price, "amount_usdt": amount * sell_price, "entry_price": entry, "realized_pnl": realized_pnl, "realized_pnl_pct": realized_pnl_pct, }) def _execute_take_profit(self, sym: str, pos: dict, tp_price: float, ts): amount = pos["amount"] entry = pos["entry_price"] sell_price = tp_price * (1 - FEE_PCT) realized_pnl = amount * sell_price - amount * entry realized_pnl_pct = (sell_price - entry) / entry * 100 if entry > 0 else 0 self.portfolio["available_usdt"] += amount * sell_price self.portfolio["positions"].pop(sym, None) self.stop_orders.pop(sym, None) self.entry_bar.pop(sym, None) self.addon_count.pop(sym, None) self.trailing_stop.pop(sym, None) self.trade_log.append({ "timestamp": str(pd.Timestamp(ts)), "symbol": sym, "action": "SELL", "type": "take_profit", "amount": amount, "price": tp_price, "effective_price": sell_price, "amount_usdt": amount * sell_price, "entry_price": entry, "realized_pnl": realized_pnl, "realized_pnl_pct": realized_pnl_pct, }) def _process_signal(self, signal: dict, current_prices: dict, ts, bar_index: int): action = signal["action"] sym = signal["symbol"] # Build indicators_df stub for risk_manager (it reads iloc[-1] for ATR/close) price = current_prices.get(sym, 0) if price <= 0: return validated, reject_reason = risk_manager.validate_trade(signal, self.portfolio, None, current_prices) if validated is None: return if action == "BUY": # Enforce daily buy limit per symbol ts_date = str(pd.Timestamp(ts).date()) daily_key = (sym, ts_date) if self.daily_buys.get(daily_key, 0) >= MAX_DAILY_BUYS_PER_SYMBOL: return # Enforce max concurrent symbols active_positions = {s for s, p in self.portfolio.get("positions", {}).items() if p.get("amount", 0) > 0} if sym not in active_positions and len(active_positions) >= MAX_CONCURRENT_SYMBOLS: return # Enforce max add-on per symbol if sym in active_positions and self.addon_count.get(sym, 0) >= MAX_ADDON_PER_SYMBOL: return amount_usdt = validated.get("amount_usdt", 0) buy_price = price * (1 + FEE_PCT) amount = amount_usdt / buy_price # Check minimum order amount min_amt = config.MIN_ORDER_AMOUNT.get(sym, 0) if min_amt > 0 and amount < min_amt: return if amount_usdt > self.portfolio["available_usdt"]: return # Update portfolio pos = self.portfolio.get("positions", {}).get(sym, {}) old_amt = pos.get("amount", 0) old_entry = pos.get("entry_price", 0) new_amt = old_amt + amount new_entry = ((old_amt * old_entry) + (amount * buy_price)) / new_amt if new_amt > 0 else buy_price self.portfolio.setdefault("positions", {})[sym] = { "amount": new_amt, "entry_price": new_entry, "value_usdt": new_amt * new_entry, "last_update": float(pd.Timestamp(ts).timestamp()), } self.portfolio["available_usdt"] -= amount_usdt # Set ATR-based stop-loss atr = signal.get("atr", 0) if atr > 0: stop_price = _calc_atr_stop_price(new_entry, atr) else: stop_price = round(new_entry * (1 - BT_ATR_SL_MAX_PCT), 8) self.stop_orders[sym] = { "stop_price": stop_price, "entry_price": new_entry, "amount": new_amt, } # Reset trailing stop tracking on new/add-on entry self.trailing_stop.pop(sym, None) # Track cooldown, addon count, entry bar, daily buys self.buy_cooldown[sym] = bar_index self.daily_buys[daily_key] = self.daily_buys.get(daily_key, 0) + 1 if old_amt > 0: self.addon_count[sym] = self.addon_count.get(sym, 0) + 1 else: self.addon_count[sym] = 0 self.entry_bar[sym] = bar_index self.trade_log.append({ "timestamp": str(pd.Timestamp(ts)), "symbol": sym, "action": "BUY", "type": "signal", "amount": amount, "price": price, "effective_price": buy_price, "amount_usdt": amount_usdt, "confidence": signal.get("confidence", 0), "reason": signal.get("reason", ""), "stop_price": stop_price, }) elif action == "SELL": pos = self.portfolio.get("positions", {}).get(sym, {}) amount = pos.get("amount", 0) if amount <= 0: return entry = pos.get("entry_price", 0) sell_price = price * (1 - FEE_PCT) realized_pnl = amount * sell_price - amount * entry realized_pnl_pct = (sell_price - entry) / entry * 100 if entry > 0 else 0 self.portfolio["available_usdt"] += amount * sell_price self.portfolio["positions"].pop(sym, None) self.stop_orders.pop(sym, None) self.entry_bar.pop(sym, None) self.addon_count.pop(sym, None) self.trailing_stop.pop(sym, None) self.trade_log.append({ "timestamp": str(pd.Timestamp(ts)), "symbol": sym, "action": "SELL", "type": "signal", "amount": amount, "price": price, "effective_price": sell_price, "amount_usdt": amount * sell_price, "entry_price": entry, "realized_pnl": realized_pnl, "realized_pnl_pct": realized_pnl_pct, "confidence": signal.get("confidence", 0), "reason": signal.get("reason", ""), }) def _build_results(self) -> dict: """Calculate performance metrics.""" equity_df = pd.DataFrame(self.equity_curve, columns=["timestamp", "equity"]) trades_df = pd.DataFrame(self.trade_log) if self.trade_log else pd.DataFrame() final_equity = equity_df["equity"].iloc[-1] if not equity_df.empty else self.starting_capital total_return = (final_equity - self.starting_capital) / self.starting_capital * 100 # Max drawdown max_dd = 0.0 if not equity_df.empty: peak = equity_df["equity"].expanding().max() drawdown = (equity_df["equity"] - peak) / peak * 100 max_dd = drawdown.min() # Trade statistics sells = [t for t in self.trade_log if t["action"] == "SELL"] buys = [t for t in self.trade_log if t["action"] == "BUY"] wins = [t for t in sells if t.get("realized_pnl", 0) > 0] losses = [t for t in sells if t.get("realized_pnl", 0) <= 0] win_rate = len(wins) / len(sells) * 100 if sells else 0 gross_profit = sum(t["realized_pnl"] for t in wins) if wins else 0 gross_loss = abs(sum(t["realized_pnl"] for t in losses)) if losses else 0 profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf") if gross_profit > 0 else 0 avg_win = gross_profit / len(wins) if wins else 0 avg_loss = gross_loss / len(losses) if losses else 0 # Trade type breakdown sl_trades = [t for t in sells if t.get("type") == "stop_loss"] tp_trades = [t for t in sells if t.get("type") == "take_profit"] sig_sells = [t for t in sells if t.get("type") == "signal"] timeout_trades = [t for t in sells if t.get("type", "").startswith("timeout")] trailing_trades = [t for t in sells if t.get("type") == "trailing_stop"] # Days with exposure days = (equity_df["timestamp"].iloc[-1] - equity_df["timestamp"].iloc[0]).days if len(equity_df) > 1 else 1 annualized = total_return * (365 / days) if days > 0 else 0 return { "starting_capital": self.starting_capital, "final_equity": round(final_equity, 2), "total_return_pct": round(total_return, 2), "annualized_return_pct": round(annualized, 2), "max_drawdown_pct": round(max_dd, 2), "total_buys": len(buys), "total_sells": len(sells), "stop_loss_count": len(sl_trades), "take_profit_count": len(tp_trades), "signal_sell_count": len(sig_sells), "timeout_count": len(timeout_trades), "trailing_stop_count": len(trailing_trades), "win_rate_pct": round(win_rate, 2), "profit_factor": round(profit_factor, 2), "avg_win_usdt": round(avg_win, 2), "avg_loss_usdt": round(avg_loss, 2), "gross_profit": round(gross_profit, 2), "gross_loss": round(gross_loss, 2), "days": days, "trades": self.trade_log, "equity_curve": self.equity_curve, }