By the HolySheep AI Technical Writing Team | Published January 2026
Case Study: How a Singapore Quantitative Fund Cut Latency by 57% and Reduced Costs by 84%
A Series-A quantitative hedge fund based in Singapore approached HolySheep AI with a critical infrastructure challenge. Their trading systems were ingesting real-time market data from multiple exchange APIs—including Binance spot and futures markets—while running complex backtesting pipelines that processed millions of historical data points daily. The existing architecture relied on direct exchange API calls with a major cloud provider middleware layer, resulting in P99 latencies of 420ms during peak trading hours and monthly infrastructure bills exceeding $4,200.
The team's pain points were multidimensional: inconsistent rate limiting across different exchange endpoints caused sporadic data gaps during backtesting runs, authentication token refresh cycles introduced unpredictable delays, and the legacy system lacked proper circuit breaker patterns—resulting in cascade failures during volatile market conditions. More critically, the engineering team spent approximately 15 hours weekly managing API quirks and debugging data inconsistency issues that directly impacted their trading strategy development velocity.
After evaluating three alternative providers, the fund's engineering lead chose HolySheep AI for three decisive reasons: sub-50ms average latency through their optimized relay infrastructure, unified authentication across all supported exchanges (Binance, Bybit, OKX, and Deribit), and their proprietary data normalization layer that eliminated the persistent timestamp alignment issues the team had struggled with for eight months.
The migration involved three phases over a two-week sprint. Phase one replaced the authentication layer with HolySheep's unified key management system, requiring a simple base_url swap from their legacy endpoint to https://api.holysheep.ai/v1 and rotation of API keys through HolySheep's dashboard. Phase two implemented a canary deployment pattern where 10% of production traffic routed through the new integration while monitoring latency percentiles and error rates. Phase three involved a full traffic cutover with zero-downtime deployment using blue-green infrastructure patterns.
Thirty days post-launch, the results validated the migration investment: P99 latency dropped from 420ms to 180ms (57% improvement), monthly infrastructure costs fell from $4,200 to $680 (84% reduction), and the engineering team's API-related debugging hours decreased from 15 hours weekly to under 2 hours. The fund's head of quantitative research reported that strategy iteration cycles shortened by approximately 40% due to more reliable backtesting infrastructure.
Understanding Binance Market Data Architecture
Spot vs. Futures: Structural Differences
Binance operates two distinct market structures that quantitative traders must understand before building data pipelines. Spot markets involve immediate asset transfer at current prices, with trading pairs like BTC/USDT representing actual cryptocurrency ownership exchange. Futures markets, conversely, involve contractual agreements to buy or sell assets at predetermined future prices, with quarterly and perpetual contract variants that introduce basis risk and funding rate dynamics into trading strategies.
The API architecture reflects these structural differences. Spot market data streams provide trade executions, order book snapshots, and kline (candlestick) data with relatively straightforward timestamp alignment. Futures APIs add complexity through funding rate feeds, position-related endpoints, and mark price streams that track the theoretical fair value of perpetual contracts. For quantitative backtesting purposes, this distinction matters significantly—strategy logic that works on spot data may require fundamental redesign when applied to futures due to leverage, liquidation mechanics, and funding payment flows.
Data Types and Their Applications in Backtesting
Effective quantitative backtesting requires four primary data types, each serving distinct analytical purposes:
- AggTrade Data (Individual Trades): Raw transaction-level data capturing every execution on the exchange order book. Essential for slippage modeling, order flow analysis, and high-frequency strategy development where individual trade patterns matter.
- Depth/Order Book Data: Snapshots of pending orders at various price levels, typically represented as bid-ask ladder structures. Critical for liquidity analysis, market impact estimation, and spread evolution modeling during backtesting.
- Kline/Candlestick Data: Aggregated OHLCV (Open, High, Low, Close, Volume) data at various timeframes from 1-minute to 1-month granularity. The foundation for most technical indicator calculations and price-pattern-based strategies.
- Liquidation and Funding Rate Data: Exchange-specific signals indicating forced position closures (liquidations) and periodic funding payments between long and short position holders. Particularly relevant for futures strategy backtesting where these factors directly impact profitability.
Building a Unified Data Acquisition Pipeline
Architecture Overview
A production-grade quantitative data pipeline requires three architectural layers: data ingestion, data normalization, and data storage. The ingestion layer handles connection management, authentication, rate limiting, and reconnection logic for exchange WebSocket and REST endpoints. The normalization layer transforms exchange-specific data formats into a standardized schema that downstream systems consume. The storage layer provides both real-time access for live trading and historical retrieval for backtesting workloads.
HolySheep AI's relay infrastructure provides significant advantages at the ingestion layer through their Tardis.dev-powered market data relay. This relay aggregates normalized data streams from Binance, Bybit, OKX, and Deribit with consistent timestamp formatting, automatic reconnection handling, and unified authentication. For teams building quantitative strategies that backtest across multiple exchanges, this eliminates the most tedious integration work while delivering sub-50ms latency on WebSocket streams.
REST API Integration with HolySheep
The following implementation demonstrates a complete Binance spot and futures data fetching system using HolySheep AI's unified API. This example fetches historical kline data for both spot and futures markets, normalizes the response formats, and prepares data for backtesting pipelines.
#!/usr/bin/env python3
"""
Binance Spot and Futures Data Fetcher using HolySheep AI
Supports historical kline extraction for backtesting pipelines
"""
import requests
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
from dataclasses import dataclass
import time
@dataclass
class KlineData:
"""Normalized candlestick data structure"""
symbol: str
open_time: datetime
open: float
high: float
low: float
close: float
volume: float
close_time: datetime
quote_volume: float
market_type: str # 'spot' or 'futures'
class HolySheepBinanceClient:
"""HolySheep AI unified client for Binance spot and futures data"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"User-Agent": "HolySheep-Quant-Pipeline/1.0"
})
# Rate tracking for HolySheep tier limits
self.request_count = 0
self.window_start = time.time()
def _rate_limit_check(self):
"""Ensure we stay within HolySheep rate limits"""
elapsed = time.time() - self.window_start
if elapsed > 60: # Reset window every 60 seconds
self.request_count = 0
self.window_start = time.time()
# HolySheep free tier: 60 requests/minute
if self.request_count >= 55: # Leave buffer
sleep_time = 60 - elapsed
if sleep_time > 0:
time.sleep(sleep_time)
self.request_count += 1
def _make_request(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
"""Execute authenticated request to HolySheep relay"""
self._rate_limit_check()
url = f"{self.BASE_URL}{endpoint}"
try:
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
raise Exception("Rate limit exceeded - upgrade HolySheep tier or wait")
elif response.status_code == 401:
raise Exception("Invalid API key - check HolySheep dashboard credentials")
else:
raise Exception(f"HTTP {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
raise Exception("Request timeout - HolySheep latency may be elevated")
def get_spot_klines(self, symbol: str, interval: str = "1h",
start_time: Optional[int] = None, limit: int = 500) -> List[KlineData]:
"""
Fetch historical klines for Binance spot market
Args:
symbol: Trading pair (e.g., 'BTCUSDT')
interval: Kline interval ('1m', '5m', '1h', '4h', '1d')
start_time: Unix timestamp in milliseconds
limit: Number of klines (max 1000 per request)
Returns:
List of normalized KlineData objects
"""
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": min(limit, 1000)
}
if start_time:
params["startTime"] = start_time
# HolySheep unified endpoint: /binance/spot/klines
data = self._make_request("/binance/spot/klines", params)
return self._parse_klines(data, "spot")
def get_futures_klines(self, symbol: str, interval: str = "1h",
start_time: Optional[int] = None, limit: int = 500) -> List[KlineData]:
"""
Fetch historical klines for Binance USD-M futures
Args:
symbol: Futures contract symbol (e.g., 'BTCUSDT')
interval: Kline interval
start_time: Unix timestamp in milliseconds
limit: Number of klines (max 1500 per request for futures)
"""
params = {
"symbol": symbol.upper(),
"interval": interval,
"limit": min(limit, 1500)
}
if start_time:
params["startTime"] = start_time
# HolySheep unified endpoint: /binance/futures/klines
data = self._make_request("/binance/futures/klines", params)
return self._parse_klines(data, "futures")
def get_funding_rates(self, symbol: str, start_time: Optional[int] = None,
limit: int = 200) -> List[Dict]:
"""
Fetch historical funding rates for futures contracts
Critical for perpetual futures strategy backtesting
"""
params = {
"symbol": symbol.upper(),
"limit": min(limit, 200)
}
if start_time:
params["startTime"] = start_time
# Endpoint: /binance/futures/funding-rate
return self._make_request("/binance/futures/funding-rate", params)
def get_order_book(self, symbol: str, market_type: str = "spot",
limit: int = 100) -> Dict:
"""
Fetch current order book depth for liquidity analysis
"""
params = {
"symbol": symbol.upper(),
"limit": limit
}
endpoint = f"/binance/{market_type}/depth"
return self._make_request(endpoint, params)
def _parse_klines(self, data: List, market_type: str) -> List[KlineData]:
"""Normalize exchange-specific kline format to standard schema"""
result = []
for kline in data:
# HolySheep normalizes response format regardless of exchange
kline_data = KlineData(
symbol=kline.get("symbol", ""),
open_time=datetime.fromtimestamp(kline["openTime"] / 1000),
open=float(kline["open"]),
high=float(kline["high"]),
low=float(kline["low"]),
close=float(kline["close"]),
volume=float(kline["volume"]),
close_time=datetime.fromtimestamp(kline["closeTime"] / 1000),
quote_volume=float(kline["quoteVolume"]),
market_type=market_type
)
result.append(kline_data)
return result
def fetch_historical_backtest_data(
client: HolySheepBinanceClient,
symbols: List[str],
start_date: datetime,
end_date: datetime,
interval: str = "1h"
) -> Dict[str, List[KlineData]]:
"""
High-level function to fetch complete historical dataset for backtesting
Handles pagination automatically to respect API limits
"""
all_data = {}
start_ts = int(start_date.timestamp() * 1000)
end_ts = int(end_date.timestamp() * 1000)
for symbol in symbols:
print(f"Fetching {symbol} data...")
spot_data = []
futures_data = []
current_start = start_ts
# Binance kline endpoint returns max 1000 (spot) or 1500 (futures) per call
while current_start < end_ts:
try:
klines = client.get_spot_klines(
symbol=symbol,
interval=interval,
start_time=current_start,
limit=1000
)
if not klines:
break
spot_data.extend(klines)
current_start = int(klines[-1].close_time.timestamp() * 1000) + 1
print(f" Retrieved {len(spot_data)} spot klines...")
# Respect rate limits between requests
time.sleep(0.2)
except Exception as e:
print(f" Error fetching {symbol} spot data: {e}")
break
all_data[f"{symbol}_spot"] = spot_data
# Fetch futures data for the same symbol
current_start = start_ts
while current_start < end_ts:
try:
klines = client.get_futures_klines(
symbol=symbol,
interval=interval,
start_time=current_start,
limit=1500
)
if not klines:
break
futures_data.extend(klines)
current_start = int(futures_data[-1].close_time.timestamp() * 1000) + 1
time.sleep(0.2)
except Exception as e:
print(f" Error fetching {symbol} futures data: {e}")
break
all_data[f"{symbol}_futures"] = futures_data
return all_data
Example usage
if __name__ == "__main__":
# Initialize HolySheep client with your API key
# Sign up at https://www.holysheep.ai/register for free credits
client = HolySheepBinanceClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Fetch 30 days of hourly data for backtesting
end = datetime.now()
start = end - timedelta(days=30)
historical_data = fetch_historical_backtest_data(
client=client,
symbols=["BTCUSDT", "ETHUSDT", "BNBUSDT"],
start_date=start,
end_date=end,
interval="1h"
)
print(f"\nTotal datasets fetched:")
for key, data in historical_data.items():
print(f" {key}: {len(data)} candles")WebSocket Real-Time Data Streaming
For live trading and real-time strategy monitoring, WebSocket connections provide lower-latency data delivery compared to REST polling. The following implementation demonstrates a robust WebSocket client that handles spot and futures streams through HolySheep's unified relay infrastructure.
#!/usr/bin/env python3
"""
Real-time WebSocket data streaming for Binance spot and futures
Using HolySheep AI relay for unified authentication and normalization
"""
import asyncio
import json
import websockets
from datetime import datetime
from typing import Callable, Dict, Set
import time
import hmac
import hashlib
from dataclasses import dataclass, field
from collections import deque
import threading
@dataclass
class TradeTick:
"""Normalized individual trade tick"""
symbol: str
price: float
quantity: float
timestamp: int
is_buyer_maker: bool # True if aggressive seller initiated
market_type: str
@dataclass
class OrderBookUpdate:
"""Normalized order book delta update"""
symbol: str
bids: list # [(price, quantity), ...]
asks: list
timestamp: int
market_type: str
class HolySheepWebSocketClient:
"""
HolySheep unified WebSocket client for real-time market data
Features:
- Unified authentication across all exchanges
- Automatic reconnection with exponential backoff
- Message buffering for backpressure handling
- Configurable subscription management
"""
WS_BASE_URL = "wss://stream.holysheep.ai/v1/ws"
def __init__(self, api_key: str, secret_key: str = None):
self.api_key = api_key
self.secret_key = secret_key
self.websocket = None
self.running = False
self.subscriptions: Set[str] = set()
# Message queues for different stream types
self.trade_queue = deque(maxlen=10000)
self.orderbook_queue = deque(maxlen=5000)
# Callbacks for real-time data processing
self.trade_callbacks: list = []
self.orderbook_callbacks: list = []
# Connection state management
self.reconnect_attempts = 0
self.max_reconnect_attempts = 10
self.last_ping_time = time.time()
self.ping_interval = 30 # seconds
# Thread safety
self.lock = threading.Lock()
def _generate_signature(self, timestamp: int) -> str:
"""Generate HMAC signature for authenticated endpoints"""
if not self.secret_key:
return ""
message = f"{timestamp}".encode()
signature = hmac.new(
self.secret_key.encode(),
message,
hashlib.sha256
).hexdigest()
return signature
async def connect(self):
"""Establish WebSocket connection with HolySheep relay"""
timestamp = int(time.time() * 1000)
signature = self._generate_signature(timestamp)
# HolySheep authentication via URL parameters
auth_params = f"?api_key={self.api_key}×tamp={timestamp}"
if signature:
auth_params += f"&signature={signature}"
ws_url = f"{self.WS_BASE_URL}{auth_params}"
try:
self.websocket = await websockets.connect(ws_url, ping_interval=None)
self.running = True
self.reconnect_attempts = 0
print(f"Connected to HolySheep WebSocket relay")
return True
except websockets.exceptions.InvalidURI:
raise Exception("Invalid WebSocket URL - check HolySheep endpoint configuration")
except websockets.exceptions.ConnectionClosed as e:
print(f"Connection closed: {e}")
return False
async def subscribe(self, streams: list):
"""
Subscribe to market data streams
Stream format: "{market}:{symbol}@{stream_type}"
Examples:
- "spot:btcusdt@trade" - BTCUSDT spot trades
- "futures:btcusdt@depth@100ms" - BTCUSDT futures order book at 100ms
- "spot:ethusdt@kline_1h" - ETHUSDT spot hourly candles
"""
subscribe_msg = {
"method": "SUBSCRIBE",
"params": streams,
"id": int(time.time() * 1000)
}
if self.websocket and self.running:
await self.websocket.send(json.dumps(subscribe_msg))
self.subscriptions.update(streams)
print(f"Subscribed to {len(streams)} streams")
async def unsubscribe(self, streams: list):
"""Unsubscribe from market data streams"""
unsubscribe_msg = {
"method": "UNSUBSCRIBE",
"params": streams,
"id": int(time.time() * 1000)
}
if self.websocket and self.running:
await self.websocket.send(json.dumps(unsubscribe_msg))
self.subscriptions.difference_update(streams)
async def listen(self):
"""
Main message processing loop
Handles message parsing, reconnection logic, and callback dispatch
"""
message_count = 0
last_stats_print = time.time()
while self.running:
try:
async for message in self.websocket:
message_count += 1
data = json.loads(message)
# Print statistics every 60 seconds
if time.time() - last_stats_print > 60:
elapsed = time.time() - last_stats_print
rate = message_count / elapsed
print(f"Messages/sec: {rate:.2f}, Total queues - Trades: {len(self.trade_queue)}, Orderbook: {len(self.orderbook_queue)}")
message_count = 0
last_stats_print = time.time()
# Route message to appropriate handler
await self._process_message(data)
except websockets.exceptions.ConnectionClosed as e:
print(f"WebSocket disconnected: {e}")
await self._handle_reconnection()
break
except Exception as e:
print(f"Message processing error: {e}")
continue
async def _process_message(self, data: Dict):
"""Parse and dispatch market data messages"""
# HolySheep uses normalized message format across exchanges
stream_type = data.get("stream_type", "")
payload = data.get("data", {})
if "trade" in stream_type:
tick = TradeTick(
symbol=payload["symbol"],
price=float(payload["price"]),
quantity=float(payload["quantity"]),
timestamp=payload["timestamp"],
is_buyer_maker=payload["isBuyerMaker"],
market_type=payload.get("market_type", "spot")
)
with self.lock:
self.trade_queue.append(tick)
# Invoke registered callbacks
for callback in self.trade_callbacks:
try:
callback(tick)
except Exception as e:
print(f"Trade callback error: {e}")
elif "depth" in stream_type or "orderbook" in stream_type:
update = OrderBookUpdate(
symbol=payload["symbol"],
bids=[[float(p), float(q)] for p, q in payload.get("bids", [])],
asks=[[float(p), float(q)] for p, q in payload.get("asks", [])],
timestamp=payload["timestamp"],
market_type=payload.get("market_type", "spot")
)
with self.lock:
self.orderbook_queue.append(update)
for callback in self.orderbook_callbacks:
try:
callback(update)
except Exception as e:
print(f"Orderbook callback error: {e}")
elif "kline" in stream_type:
# Real-time candle updates for strategy execution
kline_data = payload.get("kline", {})
print(f"Kline update: {kline_data['symbol']} O:{kline_data['open']} H:{kline_data['high']} L:{kline_data['low']} C:{kline_data['close']}")
async def _handle_reconnection(self):
"""Exponential backoff reconnection strategy"""
self.running = False
if self.reconnect_attempts >= self.max_reconnect_attempts:
print("Max reconnection attempts reached - manual intervention required")
return
delay = min(2 ** self.reconnect_attempts, 60) # Cap at 60 seconds
print(f"Reconnecting in {delay} seconds (attempt {self.reconnect_attempts + 1})")
await asyncio.sleep(delay)
self.reconnect_attempts += 1
if await self.connect():
# Resubscribe to previous streams
if self.subscriptions:
await self.subscribe(list(self.subscriptions))
# Resume listening
await self.listen()
def on_trade(self, callback: Callable[[TradeTick], None]):
"""Register trade tick callback"""
self.trade_callbacks.append(callback)
def on_orderbook(self, callback: Callable[[OrderBookUpdate], None]):
"""Register order book callback"""
self.orderbook_callbacks.append(callback)
async def disconnect(self):
"""Graceful connection shutdown"""
self.running = False
if self.websocket:
await self.websocket.close()
print("WebSocket connection closed")
Example: Real-time momentum strategy signal generation
async def momentum_signal_example():
"""Demonstrates real-time signal generation from trade flow"""
def analyze_trade_flow(tick: TradeTick):
"""Simple volume-weighted momentum analysis"""
# In production, accumulate over lookback window
if tick.is_buyer_maker:
direction = -1 # Selling pressure
else:
direction = 1 # Buying pressure
momentum = tick.quantity * direction * tick.price
print(f" {tick.symbol}: Price={tick.price}, Qty={tick.quantity}, Momentum={momentum:.2f}")
def analyze_orderbook_imbalance(update: OrderBookUpdate):
"""Calculate order book bid-ask imbalance"""
bid_volume = sum(q for p, q in update.bids[:10])
ask_volume = sum(q for p, q in update.asks[:10])
if ask_volume > 0:
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
print(f" Orderbook imbalance {update.symbol}: {imbalance:.3f}")
# Initialize client
client = HolySheepWebSocketClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Register analysis callbacks
client.on_trade(analyze_trade_flow)
client.on_orderbook(analyze_orderbook_imbalance)
# Connect and subscribe
await client.connect()
streams = [
"spot:btcusdt@trade",
"futures:btcusdt@depth@100ms",
"spot:ethusdt@trade",
"futures:ethusdt@depth@100ms"
]
await client.subscribe(streams)
# Listen for 60 seconds
print("Streaming market data for 60 seconds...")
await asyncio.sleep(60)
await client.disconnect()
if __name__ == "__main__":
asyncio.run(momentum_signal_example())Building a Quantitative Backtesting Engine
Backtesting Architecture Principles
Quantitative backtesting transforms historical market data into strategy performance estimates through simulation of trading logic against realistic market conditions. A robust backtesting architecture must address five critical dimensions: data fidelity (ensuring historical data accurately represents market microstructure), execution modeling (simulating order fills with realistic slippage and latency), risk management (applying position sizing and drawdown controls), statistical validity (preventing overfitting and lookahead bias), and computational efficiency (enabling rapid iteration over parameter spaces).
The most common failure modes in retail quantitative backtesting stem from three sources: data leakage where future information inadvertently influences historical decisions, survivorship bias where delisted or failed assets are excluded from historical analysis, and overfitting where strategy parameters are tuned to noise rather than signal. Professional-grade backtesting systems address these through strict temporal data separation, comprehensive historical universe construction, and out-of-sample validation frameworks.
Vectorized Backtesting Implementation
For strategies operating on daily or hourly timeframes, vectorized backtesting provides 10-100x speed improvements over event-driven simulation by processing entire price series as numpy arrays rather than iterating through individual bars. The following implementation demonstrates a complete vectorized backtesting framework with realistic execution modeling.
#!/usr/bin/env python3
"""
Vectorized Quantitative Backtesting Engine
Supports spot and futures strategy testing with execution modeling
"""
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import warnings
class PositionSide(Enum):
FLAT = 0
LONG = 1
SHORT = -1
@dataclass
class BacktestConfig:
"""Backtesting parameters"""
initial_capital: float = 100000.0
commission_rate: float = 0.0004 # 0.04% per side (Binance spot)
futures_commission_rate: float = 0.0002 # 0.02% for USD-M futures
slippage_bps: float = 2.0 # 2 basis points execution slippage
leverage: float = 1.0 # 1x for spot, up to 125x for futures
funding_rate: float = 0.0001 # 0.01% per 8 hours for perpetual
max_position_size: float = 0.3 # Max 30% of capital per position
@dataclass
class StrategySignal:
"""Generated trading signal"""
timestamp: datetime
symbol: str
position: PositionSide
confidence: float # 0.0 to 1.0
metadata: Dict
@dataclass
class BacktestResult:
"""Backtest performance metrics"""
total_return: float
sharpe_ratio: float
max_drawdown: float
win_rate: float
profit_factor: float
avg_trade_return: float
num_trades: int
avg_trade_duration: timedelta
equity_curve: pd.Series
trades: pd.DataFrame
class VectorizedBacktester:
"""
High-performance vectorized backtesting engine
Features:
- Numpy-accelerated signal generation
- Realistic execution modeling with slippage
- Multi-asset portfolio simulation
- Comprehensive performance analytics
"""
def __init__(self, config: BacktestConfig = None):
self.config = config or BacktestConfig()
self.data: Dict[str, pd.DataFrame] = {}
self.signals: List[StrategySignal] = []
self.positions: Dict[str, PositionSide] = {}
self.entry_prices: Dict[str, float] = {}
self.trades: List[Dict] = []
def load_data(self, symbol: str, data: pd.DataFrame):
"""
Load historical price data for backtesting
Required columns: open, high, low, close, volume
Index: datetime
"""
df = data.copy()
df = df.sort_index()
# Ensure required columns exist
required = ['open', 'high', 'low', 'close', 'volume']
for col in required:
if col not in df.columns:
raise ValueError(f"Missing required column: {col}")
self.data[symbol] = df
self.positions[symbol] = PositionSide.FLAT
self.entry_prices[symbol] = 0.0
def compute_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Compute technical indicators for signal generation
Override this method for custom strategies
"""
df = df.copy()
# Simple Moving Averages
df['sma_20'] = df['close'].rolling(window=20).mean()
df['sma_50'] = df['close'].rolling(window=50).mean()
# Exponential Moving Averages
df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
# MACD
df['macd'] = df['ema_12'] - df['ema_26']
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
df['macd_hist'] = df['macd'] - df['macd_signal']
# RSI (14-period)
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['rsi'] = 100 - (100 / (1 + rs))
# Bollinger Bands
df['bb_middle'] = df['close'].rolling(window=20).mean()
bb_std = df['close'].rolling(window=20).std()
df['bb_upper'] = df['bb_middle'] + (bb_std * 2)
df['bb_lower'] = df['bb_middle'] - (bb_std * 2)
df['bb_position'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower'])
# ATR for volatility-adjusted position sizing
high_low = df['high'] - df['low']
high_close = np.abs(df['high'] - df['close'].shift())
low_close = np.abs(df['low'] - df['close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
df['atr'] = tr.rolling(window=14).mean()
return df
def generate_signals(self, symbol: str) -> pd.DataFrame:
"""
Generate trading signals based on strategy logic
Returns DataFrame with 'signal' column: 1 (long), -1 (short), 0 (flat)
Override this method for custom strategy implementation
"""
df = self.compute_indicators(self.data[symbol])
# Trend-following strategy: SMA crossover with RSI filter
df['signal'] = 0
# Long signal: SMA 20 > SMA 50 AND RSI in oversold territory, recovering
long_condition = (
(df['sma_20'] > df['sma_50']) &
(df['rsi'] > 30) &
(df['rsi'] < 70) &
(df['macd_hist'] > 0)
)
# Short signal: SMA 20 < SMA 50 AND RSI in overbought territory, declining
short_condition = (
(df['sma_20'] < df['sma_50']) &
(df['rsi'] > 30) &
(df['rsi'] < 70) &
(df['macd_hist'] < 0)
)
df.loc[long_condition, 'signal'] = 1
df.loc[short_condition, 'signal'] = -1
# Signal smoothing: require consecutive signals
df['signal_raw'] = df['signal']
df['signal'] = df['signal'].where(
df['signal'].rolling(window=3).count() == 3, 0
)
return df
def simulate_execution(self, symbol: str, signal: int, price: float,
timestamp: datetime, capital: float) -> Tuple[float, float, float]:
"""
Simulate order execution with realistic slippage and fees
Returns: (executed_price, commission, slippage_cost)
"""
# Slippage model: random slippage within configured bps
slippage_factor = 1 + (np.random.uniform(-0.5, 0.5) * self.config.slippage_bps / 10000)
executed_price = price * slippage_factor
# Commission based on market type
commission = price * self.config.commission_rate
return executed_price, commission, abs(price - executed_price)
def run_backtest(self, symbol: str, leverage: float = 1.0) -> BacktestResult:
"""
Execute vectorized backtest for single symbol
Returns comprehensive performance metrics
"""
df = self.generate_signals(symbol)
df = df.dropna()
# Initialize tracking variables
capital = self.config.initial_capital
position = 0
entry_price = 0
entry_time = None
equity_curve = []
trade_log = []
trade