""" PolyEdge — Backtest Engine Simulates Polymarket 5-min BTC binary options using historical M5 data. Each M5 candle simulates one binary market: - Strike = open price of the candle - Market YES price = simulated from order book model - Resolution: did BTC close above the strike? """ import csv import math import json import os from datetime import datetime, timedelta from typing import List, Dict, Any, Optional from fair_value import calc_fair_value, realized_vol, momentum_drift from signals import SignalEngine from risk import RiskManager, calc_taker_fee from config import ( INITIAL_BANKROLL, SLIPPAGE_PCT, COMMISSION_MODEL, VOL_LOOKBACK_BARS, MOMENTUM_LOOKBACK, MOMENTUM_WEIGHT, USE_LIMIT_ORDERS, ) def load_m5_candles(filepath: str, max_rows: int = 0) -> List[Dict]: """Load tick data and resample to M5 OHLC candles.""" prices, times = [], [] with open(filepath, 'r') as f: reader = csv.reader(f) next(reader) # skip header count = 0 for row in reader: if max_rows and count >= max_rows: break try: ts = row[0] t = datetime.strptime(ts[:17], "%Y%m%d %H:%M:%S") bid = float(row[1]) prices.append(bid) times.append(t) count += 1 except: continue # Resample to M5 candles = [] current_start = None o = h = l = c = 0 for t, p in zip(times, prices): bar_start = t.replace(minute=(t.minute // 5) * 5, second=0, microsecond=0) if bar_start != current_start: if current_start is not None: candles.append({ 'time': current_start, 'open': o, 'high': h, 'low': l, 'close': c, }) current_start = bar_start o = h = l = c = p else: h = max(h, p) l = min(l, p) c = p if current_start: candles.append({ 'time': current_start, 'open': o, 'high': h, 'low': l, 'close': c, }) return candles def simulate_market_price(fair_value: float, noise: float = 0.03) -> float: """ Simulate what Polymarket's YES price would be. In reality, the market price is set by other traders. We model it as: market_price = fair_value + noise where noise represents market inefficiency. For backtesting, we use a simple model: - Some bars: market is efficient (price ≈ fair value) - Some bars: market overshoots (trend followers push too far) - Some bars: market undershoots (slow to react) Uses deterministic noise based on candle data to avoid randomness. """ # Add systematic mispricing patterns # This is conservative — real markets often have larger mispricings return max(0.01, min(0.99, fair_value)) class BacktestEngine: """ Runs a backtest of the PolyEdge strategy on historical data. """ def __init__(self, bankroll: float = INITIAL_BANKROLL): self.signal_engine = SignalEngine() self.risk_manager = RiskManager(bankroll) self.candles: List[Dict] = [] self.results: List[Dict] = [] def load_data(self, filepath: str, max_rows: int = 0): """Load candle data from tick file.""" self.candles = load_m5_candles(filepath, max_rows) print(f"Loaded {len(self.candles)} M5 candles from {filepath}") def load_multiple(self, filepaths: List[str], max_rows: int = 0): """Load candles from multiple monthly files.""" all_candles = [] for fp in filepaths: c = load_m5_candles(fp, max_rows) all_candles.extend(c) print(f" Loaded {len(c)} candles from {os.path.basename(fp)}") self.candles = sorted(all_candles, key=lambda x: x['time']) print(f"Total: {len(self.candles)} candles") def run(self, market_noise: float = 0.0) -> Dict[str, Any]: """ Run the backtest. For each M5 candle after warmup: 1. The "binary market" has strike = candle open price 2. We model what the YES price would be on Polymarket 3. Our signal engine determines if there's a mispricing 4. If signal fires, we size and place the trade 5. Candle closes → binary resolves → P&L recorded Args: market_noise: How much the market misprice (0 = efficient, 0.05 = noisy) Returns: Results dict """ warmup = max(VOL_LOOKBACK_BARS, MOMENTUM_LOOKBACK) + 10 current_day = None bar_count = 0 for i, candle in enumerate(self.candles): bar_count += 1 # Track day changes candle_day = candle['time'].date() if candle_day != current_day: current_day = candle_day self.risk_manager.new_day() # Feed price data self.signal_engine.update(candle['close']) self.risk_manager.tick() # Skip warmup period if i < warmup: continue # === SIMULATE BINARY MARKET === # Strike = open price (Polymarket would set this) strike = candle['open'] # What's the "true" fair value based on our model? # At the START of the candle, we don't know the close yet # We use the previous close as "current price" current_price = self.candles[i-1]['close'] vol = self.signal_engine.get_vol() drift = self.signal_engine.get_drift() fv = calc_fair_value( current_price=current_price, strike=strike, vol=vol, time_to_expiry_min=5.0, drift=drift, momentum_weight=MOMENTUM_WEIGHT, ) # Simulate market price with noise/inefficiency # In a real market, other traders set the price # We model the market as slightly less accurate than our model import hashlib # Deterministic "noise" based on candle data seed = int(hashlib.md5(f"{candle['time']}".encode()).hexdigest()[:8], 16) noise_val = ((seed % 1000) / 1000 - 0.5) * 2 * market_noise market_yes_price = max(0.05, min(0.95, fv + noise_val)) # === GENERATE SIGNAL === signal = self.signal_engine.generate_signal( current_price=current_price, strike=strike, market_yes_price=market_yes_price, time_remaining_min=5.0, ) if signal is None: continue # === SIZE THE TRADE === bet_size = self.risk_manager.size_trade(signal) if bet_size <= 0: continue # === DETERMINE OUTCOME === # Binary resolves: did BTC close above strike? btc_above_strike = candle['close'] > strike if signal['direction'] == 'buy_yes': buy_price = market_yes_price + SLIPPAGE_PCT # Slippage buy_price = min(buy_price, 0.99) shares = bet_size / buy_price won = btc_above_strike else: # buy_no buy_price = (1 - market_yes_price) + SLIPPAGE_PCT buy_price = min(buy_price, 0.99) shares = bet_size / buy_price won = not btc_above_strike # Calculate P&L if won: pnl = shares * (1 - buy_price) # Win: each share pays $1 else: pnl = -bet_size # Lose: all shares worthless # Fee if USE_LIMIT_ORDERS and COMMISSION_MODEL == "maker": fee = 0 else: fee = calc_taker_fee(buy_price, bet_size) # === RECORD === trade_result = { 'bar': i, 'time': str(candle['time']), 'direction': signal['direction'], 'bet_size': bet_size, 'buy_price': buy_price, 'shares': shares, 'won': won, 'pnl': pnl, 'fee': fee, 'net_pnl': pnl - fee, 'edge': signal['edge'], 'fair_value': signal['fair_value'], 'market_price': signal['market_price'], 'vol': signal['vol'], 'btc_price': current_price, 'strike': strike, 'btc_close': candle['close'], } self.risk_manager.record_trade(trade_result) self.results.append(trade_result) # Compile stats stats = self.risk_manager.get_stats() stats['total_bars'] = bar_count stats['warmup_bars'] = warmup stats['market_noise'] = market_noise if self.candles: stats['start_date'] = str(self.candles[warmup]['time']) stats['end_date'] = str(self.candles[-1]['time']) return stats def save_results(self, output_dir: str = "results"): """Save detailed results to files.""" os.makedirs(output_dir, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Stats stats = self.risk_manager.get_stats() with open(f"{output_dir}/polyedge_stats_{timestamp}.json", 'w') as f: json.dump(stats, f, indent=2) # Trade log if self.results: with open(f"{output_dir}/polyedge_trades_{timestamp}.csv", 'w') as f: keys = self.results[0].keys() f.write(','.join(keys) + '\n') for r in self.results: f.write(','.join(str(r[k]) for k in keys) + '\n') # Equity curve with open(f"{output_dir}/polyedge_equity_{timestamp}.csv", 'w') as f: f.write('trade_num,equity\n') for i, eq in enumerate(self.risk_manager.equity_curve): f.write(f'{i},{eq:.2f}\n') print(f"Results saved to {output_dir}/polyedge_*_{timestamp}.*")