61. Ultra-Fast Strategy Backtesting Engine¶

services/ultra-fast-backtest-center/
├── src/
│   ├── main.py
│   ├── datasource/
│   │   ├── data_loader.py
│   │   ├── tick_loader.py
│   │   └── l2_loader.py
│   ├── engine/
│   │   ├── event_engine.py
│   │   ├── match_engine.py
│   │   ├── strategy_executor.py
│   │   └── account_simulator.py
│   ├── analyzer/
│   │   ├── backtest_analyzer.py
│   │   └── performance_calculator.py
│   ├── api/
│   │   ├── backtest_api.py
│   ├── config.py
│   └── requirements.txt
├── Dockerfile
└── tests/

class DataLoader:
    def __init__(self, tick_loader, l2_loader):
        self.tick_loader = tick_loader
        self.l2_loader = l2_loader
        self.data_cache = {}

    async def load_backtest_data(self, symbols: List[str], start_time: str, end_time: str, data_type: str = "tick") -> Dict:
        """Load backtest data for specified symbols and time period"""

        symbol_data = {}

        for symbol in symbols:
            try:
                if data_type == "tick":
                    data = await self.tick_loader.load_data(symbol, start_time, end_time)
                elif data_type == "l2":
                    data = await self.l2_loader.load_data(symbol, start_time, end_time)
                else:
                    raise ValueError(f"Unsupported data type: {data_type}")

                symbol_data[symbol] = data

            except Exception as e:
                print(f"Error loading data for {symbol}: {e}")
                continue

        return {
            "symbols": symbols,
            "data": symbol_data,
            "start_time": start_time,
            "end_time": end_time,
            "data_type": data_type
        }

class EventEngine:
    def __init__(self):
        self.event_queue = []
        self.current_time = None
        self.event_handlers = {}

    def schedule_event(self, timestamp: float, event_type: str, event_data: Dict):
        """Schedule an event for processing"""

        event = {
            "timestamp": timestamp,
            "type": event_type,
            "data": event_data
        }

        heapq.heappush(self.event_queue, (timestamp, event))

    async def run_simulation(self, start_time: float, end_time: float, event_handlers: Dict):
        """Run event-driven simulation"""

        self.current_time = start_time
        self.event_handlers = event_handlers

        while self.event_queue and self.current_time <= end_time:
            # Get next event
            event_time, event = heapq.heappop(self.event_queue)

            # Update current time
            self.current_time = event_time

            # Process event
            await self.process_event(event)

    async def process_event(self, event: Dict):
        """Process a single event"""

        event_type = event["type"]
        event_data = event["data"]

        # Get event handler
        handler = self.event_handlers.get(event_type)

        if handler:
            try:
                # Execute event handler
                await handler(event_data, self.current_time)
            except Exception as e:
                print(f"Error processing event: {e}")

class MatchEngine:
    def __init__(self):
        self.execution_stats = {
            "total_orders": 0,
            "filled_orders": 0,
            "rejected_orders": 0,
            "avg_slippage": 0.0
        }

    async def process_order(self, order: Dict, market_depth: Dict) -> Dict:
        """Process and match an order"""

        self.execution_stats["total_orders"] += 1

        # Validate order
        if not self.validate_order(order):
            self.execution_stats["rejected_orders"] += 1
            return {"status": "rejected", "reason": "invalid_order"}

        # Simulate execution delay
        execution_delay = self.simulate_execution_delay(order)
        await asyncio.sleep(execution_delay / 1000)

        # Match order against market depth
        execution_result = await self.match_against_market(order, market_depth)

        if execution_result["status"] == "filled":
            self.execution_stats["filled_orders"] += 1

            # Calculate slippage
            slippage = self.calculate_slippage(order, execution_result)
            execution_result["slippage"] = slippage

        return execution_result

    def simulate_execution_delay(self, order: Dict) -> float:
        """Simulate realistic execution delay"""

        base_delay = 5.0  # 5ms base delay
        random_delay = np.random.normal(0, 2.0)  # 2ms standard deviation
        size_delay = order["quantity"] * 0.001  # 0.001ms per unit

        return max(0, base_delay + random_delay + size_delay)

    def calculate_slippage(self, order: Dict, execution: Dict) -> float:
        """Calculate execution slippage"""

        order_price = order["price"]
        execution_price = execution["average_price"]

        if order["side"] == "buy":
            slippage = (execution_price - order_price) / order_price
        else:
            slippage = (order_price - execution_price) / order_price

        return slippage

class StrategyExecutor:
    def __init__(self):
        self.strategies = {}
        self.signal_history = []

    async def execute_strategy(self, strategy_id: str, market_event: Dict, current_time: float) -> List[Dict]:
        """Execute strategy and generate signals"""

        if strategy_id not in self.strategies:
            return []

        try:
            # Execute strategy logic
            signals = await self.run_strategy_logic(strategy_id, market_event, current_time)

            # Validate signals
            valid_signals = [signal for signal in signals if self.validate_signal(signal)]

            # Record signals
            self.signal_history.extend(valid_signals)

            return valid_signals

        except Exception as e:
            print(f"Error executing strategy {strategy_id}: {e}")
            return []

    def validate_signal(self, signal: Dict) -> bool:
        """Validate trading signal"""

        required_fields = ["strategy_id", "timestamp", "type", "symbol", "quantity", "price"]

        for field in required_fields:
            if field not in signal:
                return False

        return signal["type"] in ["buy", "sell"] and signal["quantity"] > 0 and signal["price"] > 0

@router.post("/backtest/run")
async def run_backtest(request: BacktestRequest):
    """Run a backtest"""

    # Load data
    data = await data_loader.load_backtest_data(
        request.symbols, 
        request.start_time, 
        request.end_time, 
        request.data_type
    )

    # Run simulation
    result = await run_backtest_simulation(request, data)

    return {
        "backtest_id": result["backtest_id"],
        "status": "completed",
        "result": result
    }

@router.get("/backtest/results/{backtest_id}")
async def get_backtest_results(backtest_id: str):
    """Get backtest results"""

    results = await get_backtest_results_by_id(backtest_id)

    return {
        "backtest_id": backtest_id,
        "results": results
    }

@router.post("/backtest/batch")
async def run_batch_backtest(request: BatchBacktestRequest):
    """Run batch backtest for multiple strategies"""

    results = await run_batch_backtest_simulation(request)

    return {
        "batch_id": results["batch_id"],
        "results": results
    }

const UltraFastBacktestView: React.FC = () => {
  const [backtests, setBacktests] = useState<Backtest[]>([]);
  const [currentBacktest, setCurrentBacktest] = useState<Backtest | null>(null);
  const [performanceMetrics, setPerformanceMetrics] = useState<PerformanceMetrics | null>(null);

  return (
    <div className="ultra-fast-backtest-dashboard">
      {/* Backtest Control Panel */}
      <BacktestControlPanel 
        onBacktestStart={handleBacktestStart}
        onBatchBacktest={handleBatchBacktest}
      />

      {/* Performance Charts */}
      <PerformanceChartsPanel 
        backtest={currentBacktest}
        metrics={performanceMetrics}
      />

      {/* Strategy Comparison */}
      <StrategyComparisonPanel 
        batchResults={batchResults}
      />

      {/* Execution Statistics */}
      <ExecutionStatisticsPanel 
        backtest={currentBacktest}
      />

      {/* Risk Analysis */}
      <RiskAnalysisPanel 
        backtest={currentBacktest}
      />
    </div>
  );
};