In this hands-on engineering tutorial, I walk through building a complete orderbook reconstruction pipeline using Tardis.dev historical market data and implementing a simulation matching engine in Python. Whether you are backtesting algorithmic trading strategies, building market microstructure models, or training machine learning systems on historical liquidity patterns, this guide delivers production-ready code you can deploy today.
What You Will Build
- A Python pipeline that fetches historical Level 2 orderbook snapshots from Tardis.dev
- An orderbook reconstruction engine that replays incremental updates into a full depth-of-market
- A simulation matching engine that processes historical trades against reconstructed orderbook states
- Cost-optimized AI inference using HolySheep AI relay for downstream analysis pipelines
Why Tardis.dev + HolySheep AI
Tardis.dev provides tick-level historical market data from major crypto exchanges including Binance, Bybit, OKX, and Deribit. Their normalized API structure simplifies multi-exchange backtesting. HolySheep AI relay adds a critical cost layer: their 2026 pricing (DeepSeek V3.2 at $0.42/MTok output) enables massive-scale analysis workloads at a fraction of OpenAI/Anthropic costs.
2026 AI Provider Pricing Comparison
Before diving into code, here is the pricing landscape that directly impacts your backtesting infrastructure costs:
| Provider / Model | Output Price ($/MTok) | 10M Tokens/Month Cost | vs DeepSeek V3.2 |
|---|---|---|---|
| GPT-4.1 (OpenAI) | $8.00 | $80.00 | 19× more expensive |
| Claude Sonnet 4.5 (Anthropic) | $15.00 | $150.00 | 35.7× more expensive |
| Gemini 2.5 Flash (Google) | $2.50 | $25.00 | 5.9× more expensive |
| DeepSeek V3.2 (via HolySheep) | $0.42 | $4.20 | Baseline |
For a typical quant team running 10 million tokens monthly through AI-assisted strategy analysis, switching from GPT-4.1 to DeepSeek V3.2 via HolySheep relay saves $75.80/month — a 94.75% cost reduction. At scale (100M tokens), the savings reach $758/month.
Prerequisites
- Python 3.9+ with
pip - Tardis.dev API key (free tier available)
- HolySheep AI API key (Sign up here for free credits)
pip install aiohttp asyncio-socks-proxy pandas numpy
Step 1: Fetching Historical Orderbook Data from Tardis.dev
The following async Python client fetches historical orderbook snapshots. Tardis.dev uses a WebSocket-first API for live data and REST endpoints for historical replay.
# tardis_client.py
import aiohttp
import asyncio
import json
from datetime import datetime, timedelta
from typing import AsyncIterator, Dict, List, Optional
class TardisOrderbookClient:
"""
Async client for fetching historical orderbook snapshots from Tardis.dev.
Supports Binance, Bybit, OKX, and Deribit normalized data format.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
async def fetch_orderbook_snapshots(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> AsyncIterator[Dict]:
"""
Fetch historical orderbook snapshots for a given exchange/symbol pair.
Args:
exchange: 'binance', 'bybit', 'okx', 'deribit'
symbol: Trading pair, e.g., 'BTCUSDT'
start_date: Start of historical window
end_date: End of historical window
"""
# Convert dates to Unix timestamps (milliseconds)
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)
url = f"{self.base_url}/historical/orderbooks/{exchange}/{symbol}"
params = {
"from": start_ts,
"to": end_ts,
"format": "json"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
async with aiohttp.ClientSession() as session:
page = 1
while True:
params["page"] = page
async with session.get(
url,
params=params,
headers=headers,
timeout=aiohttp.ClientTimeout(total=60)
) as response:
if response.status == 429:
retry_after = int(response.headers.get("Retry-After", 5))
await asyncio.sleep(retry_after)
continue
response.raise_for_status()
data = await response.json()
if not data or len(data.get("data", [])) == 0:
break
for item in data["data"]:
# Normalize to internal format
yield self._normalize_orderbook(item, exchange, symbol)
if not data.get("hasMore", False):
break
page += 1
await asyncio.sleep(0.1) # Rate limiting
def _normalize_orderbook(
self,
raw: Dict,
exchange: str,
symbol: str
) -> Dict:
"""Convert exchange-specific format to normalized internal schema."""
return {
"exchange": exchange,
"symbol": symbol,
"timestamp": raw["timestamp"],
"localTimestamp": raw.get("localTimestamp"),
"bids": [[float(p), float(q)] for p, q in raw.get("bids", [])],
"asks": [[float(p), float(q)] for p, q in raw.get("asks", [])],
"type": raw.get("type", "snapshot") # 'snapshot' or 'update'
}
Usage example
async def main():
client = TardisOrderbookClient(api_key="YOUR_TARDIS_API_KEY")
async for orderbook in client.fetch_orderbook_snapshots(
exchange="binance",
symbol="BTCUSDT",
start_date=datetime(2024, 1, 1),
end_date=datetime(2024, 1, 2)
):
print(f"[{orderbook['timestamp']}] "
f"Bids: {len(orderbook['bids'])}, "
f"Asks: {len(orderbook['asks'])}")
if __name__ == "__main__":
asyncio.run(main())Step 2: Orderbook Reconstruction Engine
The reconstruction engine processes incremental orderbook updates, maintaining a full depth-of-market (DOM) state. This is critical for accurate slippage simulation and liquidity analysis.
# orderbook_engine.py
from dataclasses import dataclass, field
from sortedcontainers import SortedDict
from typing import Dict, List, Tuple, Optional
import heapq
import time
@dataclass
class OrderbookState:
"""
Reconstructed orderbook state with O(log n) price level operations.
Uses SortedDict for automatic price ordering.
"""
bids: SortedDict = field(default_factory=SortedDict) # price -> quantity
asks: SortedDict = field(default_factory=SortedDict) # price -> quantity
last_update_ts: int = 0
sequence: int = 0
def get_best_bid(self) -> Optional[Tuple[float, float]]:
"""Returns (price, quantity) of best bid or None."""
if not self.bids:
return None
best_price = self.bids.keys()[-1] # SortedDict is ascending, bids need descending
return (best_price, self.bids[best_price])
def get_best_ask(self) -> Optional[Tuple[float, float]]:
"""Returns (price, quantity) of best ask or None."""
if not self.asks:
return None
return (self.asks.keys()[0], self.asks[self.asks.keys()[0]])
def spread(self) -> Optional[float]:
"""Calculate bid-ask spread."""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return best_ask[0] - best_bid[0]
return None
def mid_price(self) -> Optional[float]:
"""Calculate mid-market price."""
best_bid = self.get_best_bid()
best_ask = self.get_best_ask()
if best_bid and best_ask:
return (best_bid[0] + best_ask[0]) / 2
return None
def depth_at_level(self, levels: int = 10) -> Dict[str, List[Tuple[float, float]]]:
"""Get cumulative depth for top N levels."""
result = {"bids": [], "asks": []}
# Bids: reverse iterate (highest price first)
bid_prices = list(self.bids.keys())
cumsum = 0.0
for price in reversed(bid_prices[-levels:]):
cumsum += self.bids[price]
result["bids"].append((price, cumsum))
# Asks: forward iterate (lowest price first)
ask_prices = list(self.asks.keys())
cumsum = 0.0
for price in ask_prices[:levels]:
cumsum += self.asks[price]
result["asks"].append((price, cumsum))
return result
class OrderbookReconstructor:
"""
Reconstructs full orderbook state from incremental updates.
Handles out-of-order messages, gaps, and replay synchronization.
"""
def __init__(self, symbol: str):
self.symbol = symbol
self.state = OrderbookState()
self.message_log: List[Dict] = []
self._pending_updates: List[Dict] = []
self._sequence_heap: List[Tuple[int, int]] = [] # (sequence, index)
def apply_snapshot(self, snapshot: Dict) -> None:
"""Apply full orderbook snapshot, replacing current state."""
self.state.bids.clear()
self.state.asks.clear()
# Apply bids (price, quantity pairs)
for price, qty in snapshot["bids"]:
if qty > 0:
self.state.bids[price] = qty
elif price in self.state.bids:
del self.state.bids[price]
# Apply asks
for price, qty in snapshot["asks"]:
if qty > 0:
self.state.asks[price] = qty
elif price in self.state.asks:
del self.state.asks[price]
self.state.last_update_ts = snapshot.get("timestamp", 0)
self.state.sequence = snapshot.get("sequence", 0)
def apply_update(self, update: Dict) -> None:
"""
Apply incremental orderbook update.
Supports both full replacement and delta updates.
"""
ts = update.get("timestamp", 0)
seq = update.get("sequence", self.state.sequence + 1)
if seq <= self.state.sequence:
# Stale or duplicate message
return
# Process delta updates
for price, qty in update.get("bids", []):
if qty <= 0:
self.state.bids.pop(price, None)
else:
self.state.bids[price] = qty
for price, qty in update.get("asks", []):
if qty <= 0:
self.state.asks.pop(price, None)
else:
self.state.asks[price] = qty
self.state.last_update_ts = ts
self.state.sequence = seq
self.message_log.append({
"timestamp": ts,
"sequence": seq,
"state_snapshot": self.get_snapshot()
})
def get_snapshot(self) -> Dict:
"""Export current state as dictionary."""
return {
"symbol": self.symbol,
"timestamp": self.state.last_update_ts,
"sequence": self.state.sequence,
"best_bid": self.get_best_bid(),
"best_ask": self.get_best_ask(),
"spread": self.spread(),
"mid_price": self.mid_price(),
"bid_levels": len(self.state.bids),
"ask_levels": len(self.state.asks)
}
# Delegate methods to state
def get_best_bid(self): return self.state.get_best_bid()
def get_best_ask(self): return self.state.get_best_ask()
def spread(self): return self.state.spread()
def mid_price(self): return self.state.mid_price()
def simulate_execution(
self,
side: str,
quantity: float,
order_type: str = "market"
) -> Dict:
"""
Simulate executing a trade against current orderbook.
Returns execution details including average price and slippage.
"""
if side.lower() == "buy":
book_side = self.state.asks # We hit the asks when buying
else:
book_side = self.state.bids # We hit the bids when selling
executed_qty = 0.0
total_cost = 0.0
levels_used = []
for price in (list(self.state.asks.keys()) if side.lower() == "buy"
else list(reversed(self.state.bids.keys()))):
available = book_side.get(price, 0)
fill_qty = min(available, quantity - executed_qty)
if fill_qty <= 0:
break
levels_used.append((price, fill_qty))
total_cost += price * fill_qty
executed_qty += fill_qty
if executed_qty == 0:
return {"success": False, "reason": "Insufficient liquidity"}
avg_price = total_cost / executed_qty
mid = self.mid_price() or avg_price
slippage_bps = abs(avg_price - mid) / mid * 10000
return {
"success": True,
"side": side,
"requested_quantity": quantity,
"executed_quantity": executed_qty,
"fill_percentage": executed_qty / quantity * 100,
"average_price": avg_price,
"slippage_bps": round(slippage_bps, 2),
"levels_used": len(levels_used),
"execution_details": levels_used
}
Example usage with reconstructed state
if __name__ == "__main__":
import random
reconstructor = OrderbookReconstructor("BTCUSDT")
# Simulate receiving a snapshot
snapshot = {
"timestamp": 1704067200000,
"sequence": 1,
"bids": [
(42150.0, 2.5), (42148.0, 1.8), (42145.0, 3.2),
(42140.0, 5.0), (42135.0, 2.0)
],
"asks": [
(42155.0, 1.5), (42158.0, 2.0), (42160.0, 4.5),
(42165.0, 3.0), (42170.0, 2.5)
]
}
reconstructor.apply_snapshot(snapshot)
print(f"Initial State: Bid={reconstructor.get_best_bid()}, "
f"Ask={reconstructor.get_best_ask()}, "
f"Spread={reconstructor.spread()}")
# Simulate market buy
result = reconstructor.simulate_execution("buy", 5.0)
print(f"Market Buy 5 BTC: Avg Price={result['average_price']}, "
f"Slippage={result['slippage_bps']} bps")Step 3: Simulation Matching Engine
The matching engine replays historical trades against reconstructed orderbook states, enabling realistic backtesting with accurate fill simulation.
# matching_engine.py
import asyncio
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Tuple
from orderbook_engine import OrderbookReconstructor, OrderbookState
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP_LOSS = "stop_loss"
STOP_LIMIT = "stop_limit"
class OrderStatus(Enum):
PENDING = "pending"
FILLED = "filled"
PARTIALLY_FILLED = "partially_filled"
CANCELLED = "cancelled"
REJECTED = "rejected"
@dataclass
class Order:
order_id: str
symbol: str
side: OrderSide
order_type: OrderType
quantity: float
price: Optional[float] = None # For limit orders
stop_price: Optional[float] = None
filled_quantity: float = 0.0
avg_fill_price: float = 0.0
status: OrderStatus = OrderStatus.PENDING
created_at: int = 0
trades: List[Dict] = field(default_factory=list)
@dataclass
class Trade:
trade_id: str
order_id: str
symbol: str
side: OrderSide
price: float
quantity: float
fee: float = 0.0
fee_currency: str = "USDT"
timestamp: int = 0
liquidity_type: str = "taker" # 'maker' or 'taker'
class SimulationMatchingEngine:
"""
High-fidelity simulation matching engine for backtesting.
Processes orders against reconstructed historical orderbook states.
"""
def __init__(
self,
maker_fee: float = 0.0002, # 0.02% maker fee
taker_fee: float = 0.0004, # 0.04% taker fee
latency_ms: int = 50 # Simulated network latency
):
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.latency_ms = latency_ms
self.orders: Dict[str, Order] = {}
self.orderbook: OrderbookReconstructor
self.trades: List[Trade] = []
self.position: Dict[str, float] = {} # symbol -> quantity
self.cash: float = 1_000_000.0 # Starting cash in USDT
self.trade_counter: int = 0
def initialize_orderbook(self, symbol: str):
"""Initialize empty orderbook for given symbol."""
self.orderbook = OrderbookReconstructor(symbol)
def apply_orderbook_state(self, snapshot: Dict):
"""Apply historical orderbook snapshot from replay."""
self.orderbook.apply_snapshot(snapshot)
def submit_order(self, order: Order) -> Order:
"""
Submit order for simulation.
Market orders execute immediately.
Limit orders are added to the book if they cross the spread.
"""
self.orders[order.order_id] = order
order.created_at = self.orderbook.state.last_update_ts
if order.order_type == OrderType.MARKET:
return self._execute_market_order(order)
elif order.order_type == OrderType.LIMIT:
return self._execute_limit_order(order)
return order
def _execute_market_order(self, order: Order) -> Order:
"""Execute market order against current orderbook state."""
exec_result = self.orderbook.simulate_execution(
side=order.side.value,
quantity=order.quantity
)
if not exec_result["success"]:
order.status = OrderStatus.REJECTED
return order
# Calculate costs
total_cost = exec_result["average_price"] * exec_result["executed_quantity"]
fees = total_cost * self.taker_fee
# Update cash and position
if order.side == OrderSide.BUY:
self.cash -= (total_cost + fees)
self.position[order.symbol] = (
self.position.get(order.symbol, 0) + exec_result["executed_quantity"]
)
else:
self.cash += (total_cost - fees)
self.position[order.symbol] = (
self.position.get(order.symbol, 0) - exec_result["executed_quantity"]
)
# Record trade
trade = Trade(
trade_id=f"T{self.trade_counter}",
order_id=order.order_id,
symbol=order.symbol,
side=order.side,
price=exec_result["average_price"],
quantity=exec_result["executed_quantity"],
fee=fees,
timestamp=order.created_at,
liquidity_type="taker"
)
self.trades.append(trade)
self.trade_counter += 1
order.filled_quantity = exec_result["executed_quantity"]
order.avg_fill_price = exec_result["average_price"]
order.status = OrderStatus.FILLED if (
order.filled_quantity >= order.quantity * 0.999
) else OrderStatus.PARTIALLY_FILLED
order.trades.append({
"trade_id": trade.trade_id,
"price": trade.price,
"quantity": trade.quantity,
"fee": trade.fee
})
return order
def _execute_limit_order(self, order: Order) -> Order:
"""Execute limit order with crossing-the-spread logic."""
best_bid = self.orderbook.get_best_bid()
best_ask = self.orderbook.get_best_ask()
if not best_bid or not best_ask:
return order # No liquidity, order stays pending
should_execute = False
if order.side == OrderSide.BUY and order.price >= best_ask[0]:
should_execute = True
elif order.side == OrderSide.SELL and order.price <= best_bid[0]:
should_execute = True
if should_execute:
# Cross spread: execute as taker
order.order_type = OrderType.MARKET
return self._execute_market_order(order)
# Order rests in book (maker)
return order
def get_portfolio_value(self, current_price: float) -> float:
"""Calculate total portfolio value."""
position_value = self.position.get(self.orderbook.symbol, 0) * current_price
return self.cash + position_value
def generate_execution_report(self) -> Dict:
"""Generate comprehensive execution report for backtesting."""
if not self.trades:
return {"error": "No trades to report"}
total_fees = sum(t.fee for t in self.trades)
buy_trades = [t for t in self.trades if t.side == OrderSide.BUY]
sell_trades = [t for t in self.trades if t.side == OrderSide.SELL]
return {
"total_trades": len(self.trades),
"buy_trades": len(buy_trades),
"sell_trades": len(sell_trades),
"total_volume": sum(t.quantity for t in self.trades),
"total_fees": total_fees,
"final_cash": self.cash,
"final_position": self.position,
"avg_trade_size": sum(t.quantity for t in self.trades) / len(self.trades),
"avg_slippage_bps": self._calculate_avg_slippage()
}
def _calculate_avg_slippage(self) -> float:
"""Calculate average slippage across all trades."""
slippages = []
for trade in self.trades:
mid = self.orderbook.mid_price()
if mid:
slippage_bps = abs(trade.price - mid) / mid * 10000
slippages.append(slippage_bps)
return sum(slippages) / len(slippages) if slippages else 0.0
Backtest runner
async def run_backtest(
historical_data: List[Dict],
strategy_fn,
initial_capital: float = 1_000_000.0
):
"""
Run strategy backtest over historical data.
Args:
historical_data: List of orderbook snapshots and trades
strategy_fn: User-defined strategy function(orderbook, portfolio) -> Order
initial_capital: Starting capital in quote currency
"""
engine = SimulationMatchingEngine()
engine.cash = initial_capital
for tick in historical_data:
tick_type = tick.get("type", "snapshot")
if tick_type == "snapshot":
engine.initialize_orderbook(tick["symbol"])
engine.apply_orderbook_state(tick)
elif tick_type == "update":
engine.orderbook.apply_update(tick)
elif tick_type == "trade":
# Generate order signal from strategy
order = strategy_fn(engine.orderbook, engine)
if order:
engine.submit_order(order)
# Periodically log portfolio state
if tick.get("timestamp", 0) % 3600000 == 0: # Hourly
print(f"[{datetime.fromtimestamp(tick['timestamp']/1000)}] "
f"Portfolio: ${engine.cash:.2f}")
return engine.generate_execution_report()
Example strategy
def simple_momentum_strategy(orderbook, engine) -> Optional[Order]:
"""Example: Buy on price uptick, sell on downtick."""
# Implementation placeholder
return NoneStep 4: Integrating HolySheep AI for Strategy Analysis
After running backtests, use HolySheep AI relay to analyze results at scale. Their free tier includes credits for initial testing, and their DeepSeek V3.2 integration costs just $0.42/MTok output — ideal for processing large backtest reports.
# strategy_analysis.py
import aiohttp
import asyncio
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class HolySheepClient:
"""
Async client for HolySheep AI relay API.
Supports DeepSeek V3.2 at $0.42/MTok output — 95% cheaper than GPT-4.1.
Rate: ¥1 = $1 USD (saves 85%+ vs ¥7.3 domestic pricing)
Payment: WeChat, Alipay, international cards
Latency: <50ms p99
"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
model: str = "deepseek-chat-v3.2"
async def analyze_backtest_results(
self,
backtest_report: Dict,
prompt_override: Optional[str] = None
) -> str:
"""
Analyze backtest results using DeepSeek V3.2.
Generates performance summary, identifies issues, and provides
optimization suggestions at $0.42/MTok output cost.
"""
system_prompt = """You are a quantitative trading analyst reviewing
backtest results. Provide actionable insights including:
1. Performance summary (Sharpe, max drawdown, win rate)
2. Risk analysis
3. Strategy optimization recommendations
4. Implementation priorities
Be specific and data-driven in your analysis."""
user_prompt = prompt_override or f"""Analyze this backtest report:
{json.dumps(backtest_report, indent=2)}
Provide a structured analysis with specific recommendations."""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3, # Lower for analytical tasks
"max_tokens": 2048
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 429:
raise Exception("Rate limit exceeded. Upgrade plan or wait.")
response.raise_for_status()
result = await response.json()
return result["choices"][0]["message"]["content"]
async def batch_analyze_trades(
self,
trades: List[Dict],
batch_size: int = 50
) -> List[str]:
"""
Batch analyze trade patterns using HolySheep AI relay.
Processes trades in batches for cost efficiency.
At $0.42/MTok, analyzing 10,000 trades costs ~$0.50.
"""
analyses = []
for i in range(0, len(trades), batch_size):
batch = trades[i:i+batch_size]
prompt = f"""Analyze these {len(batch)} trades for patterns:
- Identify common execution price levels
- Detect anomalous slippage patterns
- Flag potential liquidity issues
Trades: {json.dumps(batch[:10], indent=2)} # First 10 for context"""
result = await self.analyze_backtest_results(
{"trades": batch, "batch_index": i // batch_size},
prompt_override=prompt
)
analyses.append(result)
# Rate limiting to stay within API limits
await asyncio.sleep(0.5)
return analyses
async def main():
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Example backtest report
sample_report = {
"total_trades": 1247,
"total_volume": 456.8,
"total_fees": 234.56,
"final_cash": 1245678.90,
"avg_trade_size": 0.366,
"sharpe_ratio": 1.45,
"max_drawdown": -12.3,
"win_rate": 0.58
}
analysis = await client.analyze_backtest_results(sample_report)
print("Strategy Analysis:")
print(analysis)
# Cost estimate
output_tokens = len(analysis.split()) * 1.3 # Rough token estimate
cost = output_tokens / 1_000_000 * 0.42
print(f"\nEstimated cost: ${cost:.4f} ({output_tokens:.0f} output tokens)")
if __name__ == "__main__":
asyncio.run(main())End-to-End Pipeline Integration
Combine all components into a complete backtesting pipeline that fetches historical data, reconstructs orderbooks, runs simulations, and generates AI-powered analysis.
# complete_pipeline.py
import asyncio
import json
from datetime import datetime, timedelta
from tardis_client import TardisOrderbookClient
from orderbook_engine import OrderbookReconstructor
from matching_engine import SimulationMatchingEngine, OrderSide, OrderType, Order
from strategy_analysis import HolySheepClient
class BacktestPipeline:
"""
Complete backtesting pipeline:
1. Fetch historical orderbook data from Tardis.dev
2. Reconstruct orderbook state
3. Run simulation with order matching
4. Generate AI-powered analysis via HolySheep
"""
def __init__(
self,
tardis_api_key: str,
holysheep_api_key: str,
exchange: str = "binance",
symbol: str = "BTCUSDT"
):
self.tardis_client = TardisOrderbookClient(tardis_api_key)
self.holysheep = HolySheepClient(api_key=holysheep_api_key)
self.exchange = exchange
self.symbol = symbol
self.engine = SimulationMatchingEngine()
self.orderbook = OrderbookReconstructor(symbol)
async def run(
self,
start_date: datetime,
end_date: datetime,
strategy_fn,
initial_capital: float = 1_000_000.0
) -> Dict:
"""
Execute complete backtest with AI analysis.
Cost breakdown (via HolySheep):
- Strategy analysis: ~$0.50 per backtest (10K output tokens)
- Trade pattern analysis: ~$0.20 per 1K trades
- Total AI cost: <$1.00 per backtest run
"""
print(f"Starting backtest: {self.exchange}/{self.symbol}")
print(f"Period: {start_date} to {end_date}")
print(f"Initial capital: ${initial_capital:,.2f}")
self.engine.cash = initial_capital
historical_data = []
# Step 1: Fetch historical data
print("Fetching historical orderbook data...")
async for orderbook_data in self.tardis_client.fetch_orderbook_snapshots(
exchange=self.exchange,
symbol=self.symbol,
start_date=start_date,
end_date=end_date
):
historical_data.append(orderbook_data)
# Apply to reconstruction engine
if orderbook_data.get("type") == "snapshot":
self.orderbook.apply_snapshot(orderbook_data)
self.engine.orderbook = self.orderbook
else:
self.orderbook.apply_update(orderbook_data)
# Step 2: Generate and execute strategy signals
order = strategy