""" Tick-based backtesting engine with real-time metrics tracking. Designed for processing real tick data with comprehensive analysis and multi-threaded execution support. """ import time import threading import queue from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import List, Optional, Dict, Callable, Tuple from concurrent.futures import ThreadPoolExecutor, as_completed import json import sys from config import ( SYMBOL, INITIAL_BALANCE, COMMISSION_PERCENT, SLIPPAGE_PIPS, BASE_MULTIPLIER, RISK_PERCENT, AGGRESSION_LEVEL, ) from settings import ( CUSTOM_LEVELS, CUSTOM_MULTIPLIERS, MAX_SPREAD, REENTRY_ENABLED, ) from data_loader import TickDataLoader, TickData from normalization import get_spread, normalize_price, get_pip_size, pips_to_price from thread_manager import ThreadManager, ThreadData, ThreadStatus from order_engine import BacktestOrderEngine, OrderSide, Position from risk_manager import RiskManager from averaging_engine import AveragingEngine, AveragingLevel from utils import setup_logging, get_logger, timestamp_to_str @dataclass class LiveMetrics: """Real-time metrics during backtest.""" tick_count: int = 0 current_time: Optional[datetime] = None current_price: float = 0.0 current_spread: float = 0.0 # Balance & Equity balance: float = 0.0 equity: float = 0.0 unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 # Drawdown tracking peak_equity: float = 0.0 current_drawdown: float = 0.0 current_drawdown_pct: float = 0.0 max_drawdown: float = 0.0 max_drawdown_pct: float = 0.0 max_drawdown_time: Optional[datetime] = None # Trade metrics open_positions: int = 0 total_trades: int = 0 winning_trades: int = 0 losing_trades: int = 0 # Thread metrics active_threads: int = 0 total_threads: int = 0 current_avg_level: int = 0 max_avg_level: int = 0 # Performance ticks_per_second: float = 0.0 @dataclass class DrawdownEvent: """Record of a drawdown event.""" start_time: datetime end_time: Optional[datetime] peak_equity: float trough_equity: float drawdown_amount: float drawdown_pct: float recovery_time: Optional[timedelta] = None recovered: bool = False @dataclass class TradeRecord: """Detailed trade record.""" trade_id: int thread_magic: int order_type: str # 'main' or 'averaging' averaging_level: int entry_time: datetime entry_price: float exit_time: Optional[datetime] exit_price: Optional[float] lots: float profit: float is_winner: bool @dataclass class ThreadEvent: """Single event in a thread's lifecycle.""" timestamp: datetime event_type: str # OPEN, AVG, TP_HIT, REENTRY, CLOSE, ABANDON level: int price: float lots: float tp_price: float profit: float = 0.0 balance: float = 0.0 equity: float = 0.0 details: str = "" @dataclass class ThreadLog: """Complete log of a trading thread.""" thread_id: int start_time: datetime end_time: Optional[datetime] = None events: List[ThreadEvent] = field(default_factory=list) max_level: int = 0 total_profit: float = 0.0 total_trades: int = 0 winning_trades: int = 0 entry_price: float = 0.0 def add_event(self, event: ThreadEvent): self.events.append(event) if event.level > self.max_level: self.max_level = event.level def summary(self) -> str: """Generate thread summary.""" duration = (self.end_time - self.start_time).total_seconds() if self.end_time else 0 hours = int(duration // 3600) mins = int((duration % 3600) // 60) return ( f"Thread #{self.thread_id}: " f"Entry ${self.entry_price:,.2f} | " f"MaxLvl {self.max_level} | " f"Trades {self.total_trades} | " f"Wins {self.winning_trades} | " f"P/L ${self.total_profit:+,.2f} | " f"Duration {hours}h {mins}m" ) @dataclass class PendingReentry: """Tracks a level waiting for price to return for re-entry.""" level: int entry_price: float # Price at which we should reopen lots: float # Lot size for this level tp_distance: float # TP distance from entry magic: int = 0 # Thread magic number @dataclass class TickBacktestResult: """Comprehensive backtest results.""" # Run info symbol: str start_time: datetime end_time: datetime total_ticks: int run_duration_seconds: float # Performance initial_balance: float final_balance: float final_equity: float total_return: float total_return_pct: float # Trade stats total_trades: int winning_trades: int losing_trades: int win_rate: float gross_profit: float gross_loss: float net_profit: float profit_factor: float avg_win: float avg_loss: float largest_win: float largest_loss: float avg_trade: float # Drawdown analysis max_drawdown: float max_drawdown_pct: float max_drawdown_duration: Optional[timedelta] avg_drawdown: float drawdown_events: List[DrawdownEvent] # Thread/Averaging stats total_threads: int avg_thread_profit: float max_averaging_level: int avg_averaging_level: float total_averaging_orders: int # Risk metrics sharpe_ratio: float sortino_ratio: float calmar_ratio: float # Commission total_commission: float # Curves (sampled) equity_curve: List[Tuple[datetime, float]] balance_curve: List[Tuple[datetime, float]] drawdown_curve: List[Tuple[datetime, float]] # Trade log trades: List[TradeRecord] # New features tracking (must come after non-default fields) spread_filtered: int = 0 reentries_abandoned: int = 0 def print_summary(self): """Print formatted summary.""" print("\n" + "=" * 70) print("TICK BACKTEST RESULTS") print("=" * 70) print(f"\n{'RUN INFO':=^50}") print(f"Symbol: {self.symbol}") print(f"Period: {self.start_time} to {self.end_time}") print(f"Total Ticks: {self.total_ticks:,}") print(f"Run Duration: {self.run_duration_seconds:.1f} seconds") print(f"Ticks/Second: {self.total_ticks / self.run_duration_seconds:,.0f}") print(f"\n{'PERFORMANCE':=^50}") print(f"Initial Balance: ${self.initial_balance:,.2f}") print(f"Final Balance: ${self.final_balance:,.2f}") print(f"Final Equity: ${self.final_equity:,.2f}") print(f"Net Profit: ${self.net_profit:,.2f}") print(f"Total Return: {self.total_return_pct:+.2f}%") print(f"\n{'TRADE STATISTICS':=^50}") print(f"Total Trades: {self.total_trades}") print(f"Winning Trades: {self.winning_trades} ({self.win_rate:.1f}%)") print(f"Losing Trades: {self.losing_trades}") print(f"Gross Profit: ${self.gross_profit:,.2f}") print(f"Gross Loss: ${self.gross_loss:,.2f}") print(f"Profit Factor: {self.profit_factor:.2f}") print(f"Avg Win: ${self.avg_win:,.2f}") print(f"Avg Loss: ${self.avg_loss:,.2f}") print(f"Largest Win: ${self.largest_win:,.2f}") print(f"Largest Loss: ${self.largest_loss:,.2f}") print(f"Avg Trade: ${self.avg_trade:,.2f}") print(f"\n{'DRAWDOWN ANALYSIS':=^50}") print(f"Max Drawdown: ${self.max_drawdown:,.2f} ({self.max_drawdown_pct:.2f}%)") if self.max_drawdown_duration: print(f"Max DD Duration: {self.max_drawdown_duration}") print(f"Avg Drawdown: ${self.avg_drawdown:,.2f}") print(f"DD Events (>1%): {len([d for d in self.drawdown_events if d.drawdown_pct > 1])}") print(f"\n{'AVERAGING STATISTICS':=^50}") print(f"Total Threads: {self.total_threads}") print(f"Avg Thread Profit: ${self.avg_thread_profit:,.2f}") print(f"Max Avg Level: {self.max_averaging_level}") print(f"Avg Avg Level: {self.avg_averaging_level:.2f}") print(f"Total Avg Orders: {self.total_averaging_orders}") print(f"\n{'RISK METRICS':=^50}") print(f"Sharpe Ratio: {self.sharpe_ratio:.2f}") print(f"Sortino Ratio: {self.sortino_ratio:.2f}") print(f"Calmar Ratio: {self.calmar_ratio:.2f}") print(f"\n{'COSTS':=^50}") print(f"Total Commission: ${self.total_commission:,.2f}") print(f"Spread Filtered: {self.spread_filtered}") print(f"Reentries Abandoned: {self.reentries_abandoned}") print("\n" + "=" * 70) def to_json(self, filepath: str): """Save results to JSON file.""" data = { "symbol": self.symbol, "start_time": str(self.start_time), "end_time": str(self.end_time), "total_ticks": self.total_ticks, "run_duration_seconds": self.run_duration_seconds, "initial_balance": self.initial_balance, "final_balance": self.final_balance, "final_equity": self.final_equity, "total_return": self.total_return, "total_return_pct": self.total_return_pct, "total_trades": self.total_trades, "winning_trades": self.winning_trades, "losing_trades": self.losing_trades, "win_rate": self.win_rate, "gross_profit": self.gross_profit, "gross_loss": self.gross_loss, "net_profit": self.net_profit, "profit_factor": self.profit_factor, "max_drawdown": self.max_drawdown, "max_drawdown_pct": self.max_drawdown_pct, "total_threads": self.total_threads, "max_averaging_level": self.max_averaging_level, "total_commission": self.total_commission, } with open(filepath, 'w') as f: json.dump(data, f, indent=2) class TickBacktestEngine: """ Tick-based backtesting engine with real-time metrics. """ def __init__( self, symbol: str = SYMBOL, initial_balance: float = INITIAL_BALANCE, commission_percent: float = COMMISSION_PERCENT, slippage_pips: float = SLIPPAGE_PIPS, risk_percent: float = RISK_PERCENT, base_multiplier: float = BASE_MULTIPLIER, report_interval: int = 10000, ): """ Initialize tick backtest engine. Args: symbol: Trading symbol initial_balance: Starting balance commission_percent: Commission per trade slippage_pips: Slippage in pips risk_percent: Risk percentage base_multiplier: TP/Entry distance multiplier report_interval: Ticks between status reports """ self.symbol = symbol self.initial_balance = initial_balance self.commission_percent = commission_percent self.slippage_pips = slippage_pips self.risk_percent = risk_percent self.base_multiplier = base_multiplier self.report_interval = report_interval # Components self.order_engine = BacktestOrderEngine( symbol=symbol, initial_balance=initial_balance, commission_percent=commission_percent, slippage_pips=slippage_pips ) self.thread_manager = ThreadManager() self.risk_manager = RiskManager(risk_percent=risk_percent) self.averaging_engine = AveragingEngine( thread_manager=self.thread_manager, risk_manager=self.risk_manager, symbol=symbol ) # Metrics tracking self.metrics = LiveMetrics() self.metrics.balance = initial_balance self.metrics.equity = initial_balance self.metrics.peak_equity = initial_balance # Curves (sampled) self.equity_samples: List[Tuple[datetime, float]] = [] self.balance_samples: List[Tuple[datetime, float]] = [] self.drawdown_samples: List[Tuple[datetime, float]] = [] self._sample_interval = 1000 # Sample every N ticks # Drawdown tracking self.drawdown_events: List[DrawdownEvent] = [] self._current_dd_event: Optional[DrawdownEvent] = None self._in_drawdown = False # Trade records self.trade_records: List[TradeRecord] = [] self._trade_counter = 0 # Thread profit tracking self.thread_profits: List[float] = [] self.averaging_levels: List[int] = [] # Timing self._start_time: Optional[float] = None self._last_report_time: float = 0 self.logger = get_logger() # === NEW FEATURES === # Fibonacci sequence for default averaging distances self.fib = [1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610] # Custom martingale levels (from CustomMartingaleEA) self.custom_levels = CUSTOM_LEVELS self.custom_multipliers = CUSTOM_MULTIPLIERS # Spread safety self.max_spread = MAX_SPREAD self.spread_filtered_count = 0 # Unified TP tracking (weighted average) self.thread_total_lots = 0.0 self.thread_weighted_entry = 0.0 self.thread_tp_dist = 0.0 self.thread_entry_dist = 0.0 self.thread_entry_price = 0.0 self.thread_base_lots = 0.0 # Re-entry / TP Cascade tracking self.reentry_enabled = REENTRY_ENABLED self.pending_reentries: Dict[int, PendingReentry] = {} self.reentry_tolerance = 1.0 # Price tolerance for re-entry trigger self.reentry_abandon_distance = 500.0 # Abandon if price moves too far self.reentries_abandoned = 0 # Grid-based order management self.grid_size = 50.0 # Grid size in dollars for BTC self.occupied_grid_levels: set = set() # Thread logging self.thread_logs: List[ThreadLog] = [] self.current_thread_log: Optional[ThreadLog] = None self.enable_logging = True self._thread_log_counter = 0 def _calculate_tp_entry_distances(self, spread: float) -> Tuple[float, float]: """Calculate TP and entry distances based on spread.""" tp_distance = spread * self.base_multiplier entry_distance = spread * self.base_multiplier return tp_distance, entry_distance def _get_fib(self, n: int) -> int: """Get Fibonacci number, extend if needed.""" while n >= len(self.fib): self.fib.append(self.fib[-1] + self.fib[-2]) return self.fib[n] def _get_grid_index(self, price: float) -> int: """Convert price to grid level index.""" return int(round(price / self.grid_size)) def _is_grid_level_occupied(self, price: float) -> bool: """Check if there's already an order at this grid level.""" grid_idx = self._get_grid_index(price) return grid_idx in self.occupied_grid_levels def _calc_avg_lot(self, base_lot: float, level: int) -> float: """Calculate averaging lot size with custom multipliers support.""" if self.custom_multipliers and level > 0 and level <= len(self.custom_multipliers): return base_lot * self.custom_multipliers[level - 1] else: # Default: exponential aggression aggression = AGGRESSION_LEVEL / 100 return base_lot * ((1 + aggression) ** level) def _calc_avg_distance(self, base_dist: float, level: int) -> float: """Calculate averaging distance with custom levels support.""" if self.custom_levels and level < len(self.custom_levels): return base_dist * self.custom_levels[level] else: # Default: Fibonacci return base_dist * self._get_fib(level) def _update_unified_tp(self, thread: 'ThreadData'): """Update all positions to use unified TP based on weighted average.""" if self.thread_total_lots > 0: avg_entry = self.thread_weighted_entry / self.thread_total_lots unified_tp = avg_entry + self.thread_tp_dist # Update all open positions in this thread for pos_id, pos in self.order_engine.positions.items(): order = self.order_engine.orders.get(pos_id) if order and order.magic == thread.magic_number: pos.tp_price = unified_tp def _update_metrics(self, tick: TickData): """Update real-time metrics.""" equity = self.order_engine.get_equity() balance = self.order_engine.balance self.metrics.current_time = tick.timestamp self.metrics.current_price = tick.mid self.metrics.current_spread = tick.spread self.metrics.balance = balance self.metrics.equity = equity self.metrics.unrealized_pnl = equity - balance self.metrics.open_positions = len(self.order_engine.positions) # Update drawdown if equity > self.metrics.peak_equity: self.metrics.peak_equity = equity # End drawdown event if we were in one if self._in_drawdown and self._current_dd_event: self._current_dd_event.end_time = tick.timestamp self._current_dd_event.recovered = True if self._current_dd_event.start_time: self._current_dd_event.recovery_time = ( tick.timestamp - self._current_dd_event.start_time ) self.drawdown_events.append(self._current_dd_event) self._current_dd_event = None self._in_drawdown = False dd_amount = self.metrics.peak_equity - equity dd_pct = (dd_amount / self.metrics.peak_equity * 100) if self.metrics.peak_equity > 0 else 0 self.metrics.current_drawdown = dd_amount self.metrics.current_drawdown_pct = dd_pct if dd_amount > self.metrics.max_drawdown: self.metrics.max_drawdown = dd_amount self.metrics.max_drawdown_pct = dd_pct self.metrics.max_drawdown_time = tick.timestamp # Track drawdown events (threshold: 0.5%) if dd_pct > 0.5 and not self._in_drawdown: self._in_drawdown = True self._current_dd_event = DrawdownEvent( start_time=tick.timestamp, end_time=None, peak_equity=self.metrics.peak_equity, trough_equity=equity, drawdown_amount=dd_amount, drawdown_pct=dd_pct ) elif self._in_drawdown and self._current_dd_event: if equity < self._current_dd_event.trough_equity: self._current_dd_event.trough_equity = equity self._current_dd_event.drawdown_amount = dd_amount self._current_dd_event.drawdown_pct = dd_pct # Thread metrics active = self.thread_manager.get_active_threads() self.metrics.active_threads = len(active) stats = self.thread_manager.get_statistics() self.metrics.total_threads = stats["total_threads"] self.metrics.max_avg_level = max(self.metrics.max_avg_level, stats["max_averaging_level"]) if active: self.metrics.current_avg_level = max(t.get_highest_averaging_level() for t in active) def _sample_curves(self, tick: TickData): """Sample curves at interval.""" if self.metrics.tick_count % self._sample_interval == 0: self.equity_samples.append((tick.timestamp, self.metrics.equity)) self.balance_samples.append((tick.timestamp, self.metrics.balance)) self.drawdown_samples.append((tick.timestamp, self.metrics.current_drawdown)) def _print_live_status(self): """Print live status update.""" elapsed = time.time() - self._start_time if self._start_time else 0 tps = self.metrics.tick_count / elapsed if elapsed > 0 else 0 self.metrics.ticks_per_second = tps # Build status line status = ( f"\r[{self.metrics.tick_count:>10,} ticks | {tps:>8,.0f} t/s] " f"Bal: ${self.metrics.balance:>10,.2f} | " f"Eq: ${self.metrics.equity:>10,.2f} | " f"DD: ${self.metrics.current_drawdown:>8,.2f} ({self.metrics.current_drawdown_pct:>5.2f}%) | " f"MaxDD: ${self.metrics.max_drawdown:>8,.2f} ({self.metrics.max_drawdown_pct:>5.2f}%) | " f"Pos: {self.metrics.open_positions:>2} | " f"Thrd: {self.metrics.active_threads:>2} | " f"AvgL: {self.metrics.current_avg_level:>2}" ) print(status, end="", flush=True) def _should_open_thread(self) -> bool: """Check if new thread should be opened.""" return len(self.thread_manager.get_active_threads()) == 0 def _open_thread(self, tick: TickData) -> Optional[ThreadData]: """Open a new trading thread.""" tp_dist, entry_dist = self._calculate_tp_entry_distances(tick.spread) lot_size = self.risk_manager.calculate_lot_size( self.order_engine.balance, self.symbol, tick.ask ) tp_price = tick.ask + tp_dist # Clear state for new thread self.pending_reentries.clear() self.occupied_grid_levels.clear() order = self.order_engine.place_buy_order( lots=lot_size, tp_price=tp_price, magic=0, comment="MAIN" ) if order: thread = self.thread_manager.create_thread( symbol=self.symbol, entry_price=order.price, entry_time=tick.timestamp, lot_size=lot_size, tp_distance=tp_dist, entry_distance=entry_dist, order_id=order.order_id ) order.magic = thread.magic_number if order.order_id in self.order_engine.positions: self.order_engine.positions[order.order_id].magic = thread.magic_number # Track grid level grid_idx = self._get_grid_index(order.price) self.occupied_grid_levels.add(grid_idx) # Initialize unified TP tracking self.thread_entry_price = order.price self.thread_tp_dist = tp_dist self.thread_entry_dist = entry_dist self.thread_base_lots = lot_size self.thread_total_lots = lot_size self.thread_weighted_entry = order.price * lot_size # Record trade self._trade_counter += 1 self.trade_records.append(TradeRecord( trade_id=self._trade_counter, thread_magic=thread.magic_number, order_type='main', averaging_level=0, entry_time=tick.timestamp, entry_price=order.price, exit_time=None, exit_price=None, lots=lot_size, profit=0.0, is_winner=False )) # Start thread log if self.enable_logging: self._thread_log_counter += 1 self.current_thread_log = ThreadLog( thread_id=self._thread_log_counter, start_time=tick.timestamp, entry_price=order.price ) self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="OPEN", level=0, price=order.price, lots=lot_size, tp_price=tp_price, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"Spread=${tick.spread:.2f}, TP_dist=${tp_dist:.2f}" )) return thread return None def _process_averaging(self, thread: ThreadData, tick: TickData): """Process averaging for a thread with custom levels/multipliers and unified TP.""" # Calculate next averaging level using custom distances current_level = thread.get_highest_averaging_level() next_level = current_level + 1 # Calculate trigger distance and price trigger_dist = self._calc_avg_distance(self.thread_entry_dist, next_level) trigger_price = self.thread_entry_price - trigger_dist # Check if we should place averaging order if tick.bid <= trigger_price: # Grid-based duplicate prevention if self._is_grid_level_occupied(tick.ask): return # Skip - already have order at this grid level # Calculate lot size using custom multipliers lot_size = self._calc_avg_lot(self.thread_base_lots, next_level) # Update weighted average tracking self.thread_total_lots += lot_size self.thread_weighted_entry += tick.ask * lot_size # Calculate unified TP avg_entry = self.thread_weighted_entry / self.thread_total_lots unified_tp = avg_entry + self.thread_tp_dist order = self.order_engine.place_buy_order( lots=lot_size, tp_price=unified_tp, magic=thread.magic_number, comment=f"AVG_L{next_level}" ) if order: self.thread_manager.add_averaging_order( thread.magic_number, tick.ask, lot_size, unified_tp, tick.timestamp, next_level, order.order_id ) # Track grid level grid_idx = self._get_grid_index(tick.ask) self.occupied_grid_levels.add(grid_idx) # Update ALL existing positions to use unified TP self._update_unified_tp(thread) self.averaging_levels.append(next_level) # Record trade self._trade_counter += 1 self.trade_records.append(TradeRecord( trade_id=self._trade_counter, thread_magic=thread.magic_number, order_type='averaging', averaging_level=next_level, entry_time=tick.timestamp, entry_price=order.price, exit_time=None, exit_price=None, lots=lot_size, profit=0.0, is_winner=False )) # Log averaging event if self.enable_logging and self.current_thread_log: drop_from_entry = self.thread_entry_price - tick.ask self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="AVG", level=next_level, price=tick.ask, lots=lot_size, tp_price=unified_tp, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"Drop=${drop_from_entry:.2f}, AvgEntry=${avg_entry:.2f}, TotalLots={self.thread_total_lots:.4f}" )) def _check_tp_hits(self, tick: TickData): """Check and process TP hits with re-entry logic.""" for position_id in list(self.order_engine.positions.keys()): position = self.order_engine.positions[position_id] if position.tp_price and tick.bid >= position.tp_price: # Get position info before closing entry_price = position.entry_price lots = position.lots is_reentry = getattr(position, 'is_reentry', False) profit = self.order_engine.close_position(position_id, position.tp_price) if profit is not None: order = self.order_engine.orders.get(position_id) if order: magic = order.magic thread = self.thread_manager.get_thread(magic) avg_level = getattr(order, 'avg_level', 0) # Update weighted average tracking (remove this position's contribution) self.thread_total_lots -= lots self.thread_weighted_entry -= entry_price * lots # Remove grid level from occupied set grid_idx = self._get_grid_index(entry_price) self.occupied_grid_levels.discard(grid_idx) if thread: self.thread_manager.close_order( magic, position_id, position.tp_price, tick.timestamp ) # Update trade record for tr in reversed(self.trade_records): if tr.thread_magic == magic and tr.exit_time is None: tr.exit_time = tick.timestamp tr.exit_price = position.tp_price tr.profit = profit tr.is_winner = profit > 0 avg_level = tr.averaging_level break self.metrics.total_trades += 1 if profit > 0: self.metrics.winning_trades += 1 else: self.metrics.losing_trades += 1 # Log TP hit if self.enable_logging and self.current_thread_log: event_type = "TP_REENTRY" if is_reentry else "TP_HIT" self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type=event_type, level=avg_level, price=position.tp_price, lots=lots, tp_price=position.tp_price, profit=profit, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"Entry=${entry_price:.2f}, Profit=${profit:.2f}" )) self.current_thread_log.total_profit += profit self.current_thread_log.total_trades += 1 if profit > 0: self.current_thread_log.winning_trades += 1 # TP Cascade: mark for re-entry if enabled and not already a re-entry if self.reentry_enabled and not is_reentry: self.pending_reentries[avg_level] = PendingReentry( level=avg_level, entry_price=entry_price, lots=lots, tp_distance=self.thread_tp_dist, magic=magic ) # Check if thread complete (no positions AND no pending re-entries) if not thread.get_open_orders() and not self.pending_reentries: self.thread_manager.close_thread(magic) self.thread_profits.append(thread.total_profit) # Finalize thread log if self.enable_logging and self.current_thread_log: self.current_thread_log.end_time = tick.timestamp self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="THREAD_COMPLETE", level=self.current_thread_log.max_level, price=tick.bid, lots=0, tp_price=0, profit=self.current_thread_log.total_profit, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"TotalProfit=${self.current_thread_log.total_profit:.2f}" )) self.thread_logs.append(self.current_thread_log) self.current_thread_log = None def _check_reentries(self, tick: TickData, thread: ThreadData): """Check if price has returned to any pending re-entry levels.""" levels_to_reopen = [] for level, reentry in list(self.pending_reentries.items()): # Re-enter when price returns to the entry level if tick.ask <= reentry.entry_price + self.reentry_tolerance: if not self._is_grid_level_occupied(tick.ask): levels_to_reopen.append(level) for level in sorted(levels_to_reopen): reentry = self.pending_reentries[level] if self._is_grid_level_occupied(tick.ask): continue # Update weighted average tracking self.thread_total_lots += reentry.lots self.thread_weighted_entry += tick.ask * reentry.lots # Calculate unified TP if self.thread_total_lots > 0: avg_entry = self.thread_weighted_entry / self.thread_total_lots unified_tp = avg_entry + reentry.tp_distance else: avg_entry = tick.ask unified_tp = tick.ask + reentry.tp_distance # Reopen position order = self.order_engine.place_buy_order( lots=reentry.lots, tp_price=unified_tp, magic=reentry.magic, comment=f"REENTRY_L{level}" ) if order: # Mark as re-entry if order.order_id in self.order_engine.positions: self.order_engine.positions[order.order_id].is_reentry = True # Track grid level grid_idx = self._get_grid_index(tick.ask) self.occupied_grid_levels.add(grid_idx) # Update all positions to unified TP self._update_unified_tp(thread) # Record trade self._trade_counter += 1 self.trade_records.append(TradeRecord( trade_id=self._trade_counter, thread_magic=reentry.magic, order_type='reentry', averaging_level=level, entry_time=tick.timestamp, entry_price=tick.ask, exit_time=None, exit_price=None, lots=reentry.lots, profit=0.0, is_winner=False )) # Log re-entry if self.enable_logging and self.current_thread_log: self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="REENTRY", level=level, price=tick.ask, lots=reentry.lots, tp_price=unified_tp, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"OrigEntry=${reentry.entry_price:.2f}, AvgEntry=${avg_entry:.2f}" )) # Remove from pending del self.pending_reentries[level] def _cleanup_stale_reentries(self, tick: TickData): """Abandon pending re-entries if price has moved too far away.""" levels_to_abandon = [] for level, reentry in self.pending_reentries.items(): distance = abs(tick.ask - reentry.entry_price) if distance > self.reentry_abandon_distance: levels_to_abandon.append((level, reentry, distance)) for level, reentry, distance in levels_to_abandon: # Log abandonment if self.enable_logging and self.current_thread_log: self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="ABANDON", level=level, price=tick.ask, lots=reentry.lots, tp_price=0, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"Distance=${distance:.2f}, OrigEntry=${reentry.entry_price:.2f}" )) del self.pending_reentries[level] self.reentries_abandoned += 1 # If all re-entries abandoned and no positions, complete thread if levels_to_abandon and not self.pending_reentries: active_threads = self.thread_manager.get_active_threads() for thread in active_threads: if not thread.get_open_orders(): self.thread_manager.close_thread(thread.magic_number) self.thread_profits.append(thread.total_profit) # Finalize thread log if self.enable_logging and self.current_thread_log: self.current_thread_log.end_time = tick.timestamp self.current_thread_log.add_event(ThreadEvent( timestamp=tick.timestamp, event_type="THREAD_ABANDONED", level=self.current_thread_log.max_level, price=tick.ask, lots=0, tp_price=0, profit=self.current_thread_log.total_profit, balance=self.order_engine.balance, equity=self.order_engine.get_equity(), details=f"AllReentriesAbandoned, TotalProfit=${self.current_thread_log.total_profit:.2f}" )) self.thread_logs.append(self.current_thread_log) self.current_thread_log = None def process_tick(self, tick: TickData): """Process a single tick with all new features.""" self.metrics.tick_count += 1 self.order_engine.set_time(tick.timestamp) self.order_engine.update_prices(tick.bid, tick.ask) # Spread safety check spread_too_high = tick.spread > self.max_spread # Check TPs (always, even during high spread) self._check_tp_hits(tick) # Check for re-entries (only if spread OK) active_threads = self.thread_manager.get_active_threads() if active_threads and self.pending_reentries and not spread_too_high: self._check_reentries(tick, active_threads[0]) # Cleanup stale re-entries if self.pending_reentries and not active_threads: self._cleanup_stale_reentries(tick) elif self.pending_reentries and active_threads: # Check if no positions but pending re-entries exist thread = active_threads[0] if not thread.get_open_orders(): self._cleanup_stale_reentries(tick) # Open new thread if needed (only if spread OK) if self._should_open_thread() and not self.pending_reentries: if spread_too_high: self.spread_filtered_count += 1 else: self._open_thread(tick) # Process averaging (only if spread OK) if not spread_too_high: for thread in self.thread_manager.get_active_threads(): self._process_averaging(thread, tick) # Update metrics self._update_metrics(tick) self._sample_curves(tick) # Live reporting if self.metrics.tick_count % self.report_interval == 0: self._print_live_status() def run( self, data_loader: TickDataLoader, max_ticks: Optional[int] = None, start_idx: int = 0 ) -> TickBacktestResult: """ Run backtest on tick data. Args: data_loader: TickDataLoader with data max_ticks: Maximum ticks to process start_idx: Starting index Returns: TickBacktestResult """ print(f"\nStarting tick backtest on {self.symbol}") print(f"Initial balance: ${self.initial_balance:,.2f}") print(f"Risk: {self.risk_percent}% | Multiplier: {self.base_multiplier}") print(f"Processing {max_ticks if max_ticks else len(data_loader):,} ticks...") print("-" * 80) self._start_time = time.time() data_start, data_end = data_loader.get_date_range() end_idx = start_idx + max_ticks if max_ticks else None for tick in data_loader.iterate_ticks(start_idx, end_idx): self.process_tick(tick) # Final status print() # New line after status updates print("-" * 80) # Close remaining positions if self.order_engine.positions: print(f"Closing {len(self.order_engine.positions)} remaining positions...") last_tick = data_loader.get_tick(min(end_idx - 1 if end_idx else len(data_loader) - 1, len(data_loader) - 1)) if last_tick: for pos_id in list(self.order_engine.positions.keys()): self.order_engine.close_position(pos_id, last_tick.bid) run_duration = time.time() - self._start_time result = self._compile_results(data_start, data_end, run_duration) # Write thread logs if logging is enabled if self.enable_logging and self.thread_logs: self.write_thread_logs("tick_thread_logs.txt") return result def write_thread_logs(self, filename: str = "tick_thread_logs.txt"): """Write detailed thread logs to a file.""" with open(filename, 'w') as f: f.write("=" * 100 + "\n") f.write("THREAD-BY-THREAD TRADING LOG (Tick Backtest)\n") f.write("=" * 100 + "\n\n") # Summary statistics total_threads = len(self.thread_logs) if total_threads == 0: f.write("No threads completed.\n") return profitable_threads = sum(1 for t in self.thread_logs if t.total_profit > 0) losing_threads = sum(1 for t in self.thread_logs if t.total_profit <= 0) total_profit = sum(t.total_profit for t in self.thread_logs) f.write(f"SUMMARY\n") f.write(f"-" * 50 + "\n") f.write(f"Total Threads: {total_threads}\n") f.write(f"Profitable: {profitable_threads} ({profitable_threads/total_threads*100:.1f}%)\n") f.write(f"Losing: {losing_threads}\n") f.write(f"Total P/L: ${total_profit:+,.2f}\n") f.write(f"Spread Filtered: {self.spread_filtered_count}\n") f.write(f"Reentries Abandoned:{self.reentries_abandoned}\n") f.write(f"\n") # Max level distribution level_dist = {} for t in self.thread_logs: level_dist[t.max_level] = level_dist.get(t.max_level, 0) + 1 f.write(f"MAX LEVEL DISTRIBUTION\n") f.write(f"-" * 50 + "\n") for level in sorted(level_dist.keys()): count = level_dist[level] pct = count / total_threads * 100 bar = "#" * int(pct / 2) f.write(f"Level {level:>2}: {count:>4} ({pct:>5.1f}%) {bar}\n") f.write(f"\n") # Detailed thread logs f.write("=" * 100 + "\n") f.write("DETAILED THREAD LOGS\n") f.write("=" * 100 + "\n\n") for thread in self.thread_logs: f.write(f"\n{'='*80}\n") f.write(f"{thread.summary()}\n") f.write(f"{'='*80}\n") f.write(f"{'Time':<25} {'Event':<15} {'Lvl':>4} {'Price':>12} {'Lots':>10} {'TP':>12} {'Profit':>12} {'Balance':>12} {'Details'}\n") f.write(f"{'-'*130}\n") for event in thread.events: time_str = event.timestamp.strftime("%Y-%m-%d %H:%M:%S") profit_str = f"${event.profit:+,.2f}" if event.profit != 0 else "" f.write( f"{time_str:<25} {event.event_type:<15} {event.level:>4} " f"${event.price:>10,.2f} {event.lots:>10.4f} " f"${event.tp_price:>10,.2f} {profit_str:>12} " f"${event.balance:>10,.2f} {event.details}\n" ) f.write(f"\n") # Top 10 most profitable threads f.write("\n" + "=" * 100 + "\n") f.write("TOP 10 MOST PROFITABLE THREADS\n") f.write("=" * 100 + "\n") sorted_by_profit = sorted(self.thread_logs, key=lambda t: t.total_profit, reverse=True)[:10] for i, thread in enumerate(sorted_by_profit, 1): f.write(f"{i:>2}. {thread.summary()}\n") # Threads with highest averaging levels f.write("\n" + "=" * 100 + "\n") f.write("THREADS WITH HIGHEST AVERAGING LEVELS\n") f.write("=" * 100 + "\n") sorted_by_level = sorted(self.thread_logs, key=lambda t: t.max_level, reverse=True)[:20] for i, thread in enumerate(sorted_by_level, 1): f.write(f"{i:>2}. {thread.summary()}\n") print(f"\nThread logs written to: {filename}") def _compile_results( self, data_start: datetime, data_end: datetime, run_duration: float ) -> TickBacktestResult: """Compile final results.""" stats = self.order_engine.get_statistics() # Calculate derived metrics winners = [t for t in self.trade_records if t.is_winner and t.exit_time] losers = [t for t in self.trade_records if not t.is_winner and t.exit_time] gross_profit = sum(t.profit for t in winners) if winners else 0 gross_loss = abs(sum(t.profit for t in losers)) if losers else 0 net_profit = gross_profit - gross_loss profit_factor = gross_profit / gross_loss if gross_loss > 0 else float('inf') avg_win = gross_profit / len(winners) if winners else 0 avg_loss = gross_loss / len(losers) if losers else 0 all_profits = [t.profit for t in self.trade_records if t.exit_time] largest_win = max(all_profits) if all_profits else 0 largest_loss = min(all_profits) if all_profits else 0 avg_trade = sum(all_profits) / len(all_profits) if all_profits else 0 # Averaging stats avg_avg_level = sum(self.averaging_levels) / len(self.averaging_levels) if self.averaging_levels else 0 avg_thread_profit = sum(self.thread_profits) / len(self.thread_profits) if self.thread_profits else 0 # Max drawdown duration max_dd_duration = None if self.drawdown_events: durations = [d.recovery_time for d in self.drawdown_events if d.recovery_time] if durations: max_dd_duration = max(durations) # Avg drawdown avg_dd = sum(d.drawdown_amount for d in self.drawdown_events) / len(self.drawdown_events) if self.drawdown_events else 0 # Risk metrics (simplified) returns = [t.profit / self.initial_balance * 100 for t in self.trade_records if t.exit_time] if returns: import statistics avg_return = statistics.mean(returns) std_return = statistics.stdev(returns) if len(returns) > 1 else 1 downside_returns = [r for r in returns if r < 0] downside_std = statistics.stdev(downside_returns) if len(downside_returns) > 1 else 1 sharpe = avg_return / std_return if std_return > 0 else 0 sortino = avg_return / downside_std if downside_std > 0 else 0 else: sharpe = sortino = 0 annual_return = (self.metrics.equity / self.initial_balance - 1) * 100 calmar = annual_return / self.metrics.max_drawdown_pct if self.metrics.max_drawdown_pct > 0 else 0 return TickBacktestResult( symbol=self.symbol, start_time=data_start, end_time=data_end, total_ticks=self.metrics.tick_count, run_duration_seconds=run_duration, initial_balance=self.initial_balance, final_balance=self.order_engine.balance, final_equity=self.metrics.equity, total_return=self.metrics.equity - self.initial_balance, total_return_pct=(self.metrics.equity / self.initial_balance - 1) * 100, total_trades=self.metrics.total_trades, winning_trades=self.metrics.winning_trades, losing_trades=self.metrics.losing_trades, win_rate=self.metrics.winning_trades / self.metrics.total_trades * 100 if self.metrics.total_trades > 0 else 0, gross_profit=gross_profit, gross_loss=gross_loss, net_profit=net_profit, profit_factor=profit_factor, avg_win=avg_win, avg_loss=avg_loss, largest_win=largest_win, largest_loss=largest_loss, avg_trade=avg_trade, max_drawdown=self.metrics.max_drawdown, max_drawdown_pct=self.metrics.max_drawdown_pct, max_drawdown_duration=max_dd_duration, avg_drawdown=avg_dd, drawdown_events=self.drawdown_events, total_threads=self.metrics.total_threads, avg_thread_profit=avg_thread_profit, max_averaging_level=self.metrics.max_avg_level, avg_averaging_level=avg_avg_level, total_averaging_orders=len(self.averaging_levels), sharpe_ratio=sharpe, sortino_ratio=sortino, calmar_ratio=calmar, total_commission=stats["total_commission"], spread_filtered=self.spread_filtered_count, reentries_abandoned=self.reentries_abandoned, equity_curve=self.equity_samples, balance_curve=self.balance_samples, drawdown_curve=self.drawdown_samples, trades=self.trade_records ) def run_tick_backtest( csv_file: str, max_ticks: Optional[int] = None, skip_ticks: int = 0 ) -> TickBacktestResult: """ Run tick backtest from CSV file. Args: csv_file: Path to tick data CSV max_ticks: Maximum ticks to process skip_ticks: Ticks to skip from start Returns: TickBacktestResult """ from settings import ( SYMBOL, INITIAL_BALANCE, COMMISSION_PERCENT, SLIPPAGE_PIPS, RISK_PERCENT, BASE_MULTIPLIER ) setup_logging(console=False, file=True) # Load data loader = TickDataLoader(SYMBOL) loader.load_csv(csv_file, max_rows=max_ticks, skip_rows=skip_ticks) # Print data info print(f"\nData loaded: {len(loader):,} ticks") price_range = loader.get_price_range() print(f"Price range: ${price_range[0]:,.2f} - ${price_range[1]:,.2f}") spread_stats = loader.get_spread_stats() print(f"Spread: avg=${spread_stats['avg_spread']:.2f}, " f"min=${spread_stats['min_spread']:.2f}, max=${spread_stats['max_spread']:.2f}") # Run backtest engine = TickBacktestEngine( symbol=SYMBOL, initial_balance=INITIAL_BALANCE, commission_percent=COMMISSION_PERCENT, slippage_pips=SLIPPAGE_PIPS, risk_percent=RISK_PERCENT, base_multiplier=BASE_MULTIPLIER, report_interval=10000 ) result = engine.run(loader, max_ticks=max_ticks) result.print_summary() return result if __name__ == "__main__": import sys from settings import TICK_DATA_FILE, TICK_MAX_ROWS csv_file = TICK_DATA_FILE max_ticks = TICK_MAX_ROWS # None for all ticks if len(sys.argv) > 1: csv_file = sys.argv[1] if len(sys.argv) > 2: max_ticks = int(sys.argv[2]) result = run_tick_backtest(csv_file, max_ticks=max_ticks)