import sys import os import csv import time import json import logging from decimal import Decimal from unittest.mock import MagicMock, patch # Add project root to path current_dir = os.path.dirname(os.path.abspath(__file__)) project_root = os.path.dirname(os.path.dirname(current_dir)) sys.path.append(project_root) from tests.backtest.mocks import MockExchangeState, MockWeb3, MockExchangeAPI, MockInfo, MockContract # Setup Logging logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(name)s] %(message)s') logger = logging.getLogger("BACKTESTER") class Backtester: def __init__(self, book_file, trades_file, config_overrides=None): self.book_file = book_file self.trades_file = trades_file self.config_overrides = config_overrides or {} self.events = [] self.state = MockExchangeState() # Mocks self.mock_web3 = MockWeb3(self.state) self.mock_hl_api = MockExchangeAPI(self.state) self.mock_hl_info = MockInfo(self.state) # Components (Lazy loaded) self.manager = None self.hedger = None def load_data(self): logger.info("Loading Market Data...") # Load Book with open(self.book_file, 'r') as f: reader = csv.DictReader(f) for row in reader: self.events.append({ "type": "BOOK", "ts": int(row['timestamp_ms']), "data": row }) # Load Trades # (Optional: Trades are useful for market impact, but for basic PnL tracking # based on mid-price, Book is sufficient. Loading trades just to advance time) with open(self.trades_file, 'r') as f: reader = csv.DictReader(f) for row in reader: self.events.append({ "type": "TRADE", "ts": int(row['timestamp_ms']), "data": row }) # Sort by Timestamp self.events.sort(key=lambda x: x['ts']) logger.info(f"Loaded {len(self.events)} events.") def patch_and_init(self): logger.info("Initializing Logic...") # --- PATCH MANAGER --- # We need to patch clp_manager.Web3 to return our MockWeb3 # And os.environ for config with patch.dict(os.environ, { "TARGET_DEX": "PANCAKESWAP_BNB", # Example "MAIN_WALLET_PRIVATE_KEY": "0x0000000000000000000000000000000000000000000000000000000000000001", "BNB_RPC_URL": "http://mock", "HEDGER_PRIVATE_KEY": "0x0000000000000000000000000000000000000000000000000000000000000001", "MAIN_WALLET_ADDRESS": "0xMyWallet" }): import clp_manager import clp_hedger # Apply Config Overrides if self.config_overrides: logger.info(f"Applying Config Overrides: {self.config_overrides}") for k, v in self.config_overrides.items(): # Patch Manager if hasattr(clp_manager, k): setattr(clp_manager, k, v) # Patch Hedger if hasattr(clp_hedger, k): setattr(clp_hedger, k, v) # 1. Init Manager # clp_manager.main() connects to Web3. We need to inject our mock. # Since clp_manager creates w3 inside main(), we can't inject easily without patching Web3 class. self.manager_module = clp_manager self.hedger_module = clp_hedger def run(self): self.load_data() self.patch_and_init() # MOCK TIME start_time = self.events[0]['ts'] / 1000.0 # STATUS FILE MOCK self.status_memory = [] # List[Dict] def mock_load_status(): logger.info(f"MOCK LOAD STATUS: Found {len(self.status_memory)} items") return self.status_memory def mock_save_status(data): logger.info(f"MOCK SAVE STATUS: Saving {len(data)} items") self.status_memory = data def mock_hedger_scan(): return [] # We need to globally patch time.time and the Libraries web3_class_mock = MagicMock(return_value=self.mock_web3) web3_class_mock.to_wei = self.mock_web3.to_wei web3_class_mock.from_wei = self.mock_web3.from_wei web3_class_mock.is_address = self.mock_web3.is_address web3_class_mock.to_checksum_address = lambda x: x # Mock Web3.keccak to return correct topics def mock_keccak(text=None, hexstr=None): # Known Topics if text == "Transfer(address,address,uint256)": return bytes.fromhex("ddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") if text == "IncreaseLiquidity(uint256,uint128,uint256,uint256)": return bytes.fromhex("7a53080ba414158be7ec69b987b5fb7d07dee101fe85488f0853ae16239d0bde") if text == "DecreaseLiquidity(uint256,uint128,uint256,uint256)": # 0x26f6a048ee9138f2c0ce266f322cb99228e8d619ae2bff30c67f8dcf9d2377b4 return bytes.fromhex("26f6a048ee9138f2c0ce266f322cb99228e8d619ae2bff30c67f8dcf9d2377b4") if text == "Collect(uint256,address,uint256,uint256)": # 0x70935338e69775456a85ddef226c395fb668b63fa0115f5f206227278f746d4d return bytes.fromhex("70935338e69775456a85ddef226c395fb668b63fa0115f5f206227278f746d4d") return b'\x00'*32 web3_class_mock.keccak = MagicMock(side_effect=mock_keccak) # Ensure environment is patched during the whole run env_patch = { "TARGET_DEX": "PANCAKESWAP_BNB", "MAIN_WALLET_PRIVATE_KEY": "0x0000000000000000000000000000000000000000000000000000000000000001", "BNB_RPC_URL": "http://mock", "HEDGER_PRIVATE_KEY": "0x0000000000000000000000000000000000000000000000000000000000000001", "MAIN_WALLET_ADDRESS": "0xMyWallet" } with patch.dict(os.environ, env_patch), \ patch('time.time', side_effect=lambda: self.state.current_time_ms / 1000.0), \ patch('clp_manager.Web3', web3_class_mock), \ patch('clp_hedger.Account.from_key', return_value=MagicMock(address="0xMyWallet")), \ patch('clp_hedger.Exchange', return_value=self.mock_hl_api), \ patch('clp_hedger.Info', return_value=self.mock_hl_info), \ patch('clp_manager.load_status_data', side_effect=mock_load_status), \ patch('clp_manager.save_status_data', side_effect=mock_save_status), \ patch('clp_manager.clean_address', side_effect=lambda x: x), \ patch('clp_hedger.glob.glob', return_value=[]): # Initialize Hedger (It creates the classes in __init__) self.hedger = self.hedger_module.UnifiedHedger() # Initialize Manager Components manually (simulate main setup) w3 = self.mock_web3 account = MagicMock(address="0xMyWallet") npm = w3.eth.contract("0xNPM", []) factory = w3.eth.contract("0xFactory", []) router = w3.eth.contract("0xRouter", []) # --- SIMULATION LOOP --- last_manager_tick = 0 manager_interval = 60 * 1000 trade_count = len([e for e in self.events if e['type'] == "TRADE"]) book_count = len([e for e in self.events if e['type'] == "BOOK"]) logger.info(f"SIMULATION START: {len(self.events)} total events ({book_count} BOOK, {trade_count} TRADE)") for event in self.events: self.state.current_time_ms = event['ts'] # Update Market if event['type'] == "BOOK": row = event['data'] mid = Decimal(row['mid_price']) self.state.update_price("BNB", mid) if event['type'] == "TRADE": self.state.process_trade(event['data']) # Run Logic # 1. Manager (Every X seconds) if self.state.current_time_ms - last_manager_tick > manager_interval: self.manager_module.run_tick(w3, account, npm, factory, router) last_manager_tick = self.state.current_time_ms # SYNC MANAGER STATUS TO HEDGER for pos in self.status_memory: if pos.get('status') == 'OPEN' and pos.get('type') == 'AUTOMATIC': key = ("MOCK", pos['token_id']) if key not in self.hedger.strategies: self.hedger._init_single_strategy(key, pos, "BNB") else: self.hedger.strategy_states[key]['status'] = pos.get('status', 'OPEN') elif pos.get('status') == 'CLOSED': key = ("MOCK", pos['token_id']) if key in self.hedger.strategies: self.hedger.strategy_states[key]['status'] = 'CLOSED' # 2. Hedger (Every Tick/Event) self.hedger.run_tick() # Finalize: Collect accrued fees from open positions logger.info(f"Finalizing... Checking {len(self.state.uni_positions)} open positions.") for token_id, pos in self.state.uni_positions.items(): raw_owed0 = pos.get('tokensOwed0', 0) logger.info(f"DEBUG: Position {token_id} Raw TokensOwed0: {raw_owed0}") owed0 = Decimal(raw_owed0) / Decimal(10**18) owed1 = Decimal(pos.get('tokensOwed1', 0)) / Decimal(10**18) # Convert to USD price = self.state.prices.get("BNB", Decimal("0")) # Fee0 is USDT (USD), Fee1 is WBNB usd_val = owed0 + (owed1 * price) if usd_val > 0: self.state.uni_fees_collected += usd_val logger.info(f"Finalizing Open Position {token_id}: Accrued Fees ${usd_val:.2f}") logger.info("Backtest Complete.") logger.info(f"Final Uni Fees: {self.state.uni_fees_collected}") logger.info(f"Final HL PnL: {self.state.hl_realized_pnl - self.state.hl_fees_paid}") def calculate_final_nav(self): """Calculates total Net Asset Value (USD) at the end of simulation.""" total_usd = Decimal("0") # 1. Wallet Balances # We assume T0=USDT, T1=WBNB for this profile price = self.state.prices.get("BNB", Decimal("0")) for sym, bal in self.state.wallet_balances.items(): if sym in ["USDC", "USDT"]: total_usd += bal elif sym in ["BNB", "WBNB", "NATIVE"]: total_usd += bal * price elif sym in ["ETH", "WETH"]: # If ETH price available? We mocked update_price("BNB") only. # Assuming ETH price static or 0 if not tracked eth_price = self.state.prices.get("ETH", Decimal("0")) total_usd += bal * eth_price # 2. Uniswap Positions (Liquidity Value) # Value = Amount0 * Price0 + Amount1 * Price1 # We need to calculate amounts from liquidity & current price import math # Helper to get amounts from liquidity def get_amounts(liquidity, sqrt_price_x96, tick_lower, tick_upper): # Simplified: Use the amounts we stored at mint time? # No, that's initial. We need current value. # But calculating precise amounts from liquidity/sqrtPrice requires complex math. # For approximation, we can look at what the manager logged as "Deposited" # if price hasn't moved much, or implement full liquidity math. # Since implementing full math here is complex, let's use a simplified approach: # If we are in range, we have a mix. # If out of range, we have 100% of one token. # Better: The Mock 'mint' stored initial amounts. # We can adjust by price ratio? No, IL is non-linear. # Let's use the 'decrease_liquidity' logic mock if available? # Or just assume Liquidity Value = Initial Value + PnL (Fees) - IL. # For this MVP, let's just count the Fees collected (Realized) + Initial Capital (Wallet). # BUT we spent wallet funds to open LP. # So Wallet is LOW. LP has Value. # We MUST value the LP. # Let's approximate: # Value = Liquidity / (something) ... # Actually, `clp_manager.py` calculates `actual_value` on entry. # We can track `entry_value` in the position state. return Decimal("0") # Placeholder if we can't calc easily # 3. Hyperliquid Positions (Unrealized PnL + Margin) hl_equity = self.state.hl_balances.get("USDC", 0) # Margin for sym, pos in self.state.hl_positions.items(): hl_equity += pos['unrealized_pnl'] total_usd += hl_equity return total_usd if __name__ == "__main__": # Example usage: # python tests/backtest/backtester.py market_data/BNB_raw_20251230_book.csv market_data/BNB_raw_20251230_trades.csv if len(sys.argv) < 3: print("Usage: python backtester.py ") else: bt = Backtester(sys.argv[1], sys.argv[2]) bt.run()