When I first built my crypto quant trading system in 2024, I spent three weeks debugging a ConnectionError: timeout error that was actually caused by incorrect historical data timestamp configuration. The Binance OHLCV endpoint kept returning empty datasets because my UTC offset was misconfigured. That single mistake cost me valuable backtesting time and taught me why proper historical data configuration is the foundation of any quantitative trading framework. In this guide, I will walk you through everything you need to know about configuring historical data for cryptocurrency quantitative backtesting, including working code examples, pricing comparisons, and solutions to the most common errors you will encounter.
Understanding Historical Data in Quantitative Backtesting
Quantitative backtesting requires historical market data that is accurate, complete, and properly formatted. The quality of your backtesting results is directly tied to the quality of your historical data configuration. Cryptocurrency markets operate 24/7, which means there are no market holidays or gaps to consider, but this also means you need robust data pipelines that can handle continuous market activity without interruption.
For quantitative strategies, you typically need several types of historical data: OHLCV (Open, High, Low, Close, Volume) price data for technical analysis, order book snapshots for market microstructure studies, funding rate history for perpetual futures strategies, and trade-level data for high-frequency analysis. Each data type has specific configuration requirements that we will explore in detail.
HolySheep AI Data Relay Integration
If you are building a sophisticated quant system, you will likely need AI-powered data analysis capabilities for strategy optimization. Sign up here for HolySheep AI, which offers crypto market data relay including trades, order book, liquidations, and funding rates for major exchanges like Binance, Bybit, OKX, and Deribit. The platform delivers sub-50ms latency with pricing at just $0.42 per million tokens for DeepSeek V3.2, compared to GPT-4.1 at $8 per million tokens—a savings of over 85%.
Setting Up Your Data Configuration Environment
Before diving into code, you need to establish a proper environment for historical data configuration. Your Python environment should include essential libraries for data retrieval, manipulation, and storage. Install the following packages to get started:
pip install pandas numpy requests asyncio aiohttp ccxt pandas-ta
pip install pyarrow fastparquet sqlalchemy redis-cache python-dotenvCreate a configuration file to manage your data sources, API credentials, and backtesting parameters. This separation of configuration from logic makes your backtesting framework maintainable and adaptable to different market conditions.
# config/historical_data_config.py
import os
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime, timedelta
@dataclass
class DataSourceConfig:
"""Configuration for cryptocurrency data sources."""
exchange: str = "binance"
symbols: List[str] = None # e.g., ["BTC/USDT", "ETH/USDT"]
timeframes: List[str] = None # e.g., ["1m", "5m", "1h", "1d"]
start_date: datetime = None
end_date: datetime = None
include_orderbook: bool = False
include_trades: bool = False
include_funding: bool = True # For perpetual futures
def __post_init__(self):
if self.symbols is None:
self.symbols = ["BTC/USDT"]
if self.timeframes is None:
self.timeframes = ["1h"]
if self.start_date is None:
self.start_date = datetime.utcnow() - timedelta(days=365)
if self.end_date is None:
self.end_date = datetime.utcnow()
@dataclass
class StorageConfig:
"""Configuration for data storage backends."""
use_parquet: bool = True
use_redis_cache: bool = True
data_directory: str = "./data/historical"
cache_ttl_seconds: int = 3600
@dataclass
class BacktestConfig:
"""Configuration for backtesting engine."""
initial_capital: float = 10000.0
commission_rate: float = 0.001 # 0.1% per trade
slippage_model: str = "fixed" # "fixed", "volume", "volatility"
slippage_bps: float = 5.0 # Basis points
max_position_size: float = 0.2 # 20% of capital per position
class HistoricalDataConfig:
"""Main configuration class for historical data in backtesting."""
def __init__(
self,
api_key: str = None,
data_source: str = "ccxt",
holy_sheep_api_key: str = None
):
self.api_key = api_key or os.getenv("EXCHANGE_API_KEY")
self.holy_sheep_api_key = holy_sheep_api_key or os.getenv("HOLYSHEEP_API_KEY")
self.data_source = data_source
self.data_source_config = DataSourceConfig()
self.storage_config = StorageConfig()
self.backtest_config = BacktestConfig()
self.base_url = "https://api.holysheep.ai/v1" # HolySheep API endpoint
def validate_configuration(self) -> List[str]:
"""Validate configuration and return list of issues."""
issues = []
if not self.data_source_config.symbols:
issues.append("No symbols configured for data retrieval")
if self.data_source_config.start_date >= self.data_source_config.end_date:
issues.append("Start date must be before end date")
if self.data_source == "holy_sheep" and not self.holy_sheep_api_key:
issues.append("HolySheep API key required when using holy_sheep data source")
# Check for data freshness requirements
data_age = datetime.utcnow() - self.data_source_config.end_date
if data_age.days > 30:
issues.append("Warning: End date is more than 30 days in the past")
return issues
def to_dict(self) -> dict:
"""Export configuration as dictionary for logging."""
return {
"data_source": self.data_source,
"symbols": self.data_source_config.symbols,
"timeframes": self.data_source_config.timeframes,
"date_range": {
"start": self.data_source_config.start_date.isoformat(),
"end": self.data_source_config.end_date.isoformat()
},
"storage": {
"parquet": self.storage_config.use_parquet,
"redis_cache": self.storage_config.use_redis_cache
},
"backtest": {
"initial_capital": self.backtest_config.initial_capital,
"commission_rate": f"{self.backtest_config.commission_rate * 100}%"
}
}Fetching Historical OHLCV Data with CCXT
The CCXT library is the industry standard for accessing cryptocurrency exchange APIs. It provides a unified interface for over 100 exchanges, making it ideal for multi-exchange backtesting. The key to successful data retrieval is proper pagination handling and rate limit management.
# src/data_fetchers/ccxt_fetcher.py
import ccxt
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import time
import asyncio
class CCXTHistoricalDataFetcher:
"""
Fetches historical OHLCV data using CCXT library.
Handles rate limiting, pagination, and data validation.
"""
def __init__(self, exchange_id: str = "binance", api_key: str = None,
api_secret: str = None, config = None):
self.exchange_id = exchange_id
# Initialize exchange with credentials
exchange_class = getattr(ccxt, exchange_id)
init_params = {
'enableRateLimit': True,
'options': {'defaultType': 'spot'} # or 'future', 'margin'
}
if api_key and api_secret:
init_params.update({'apiKey': api_key, 'secret': api_secret})
self.exchange = exchange_class(init_params)
self.config = config
# Rate limit tracking
self.request_count = 0
self.last_request_time = time.time()
def _rate_limit_check(self, calls_per_second: int = 10):
"""Implement rate limiting to avoid exchange bans."""
elapsed = time.time() - self.last_request_time
min_interval = 1.0 / calls_per_second
if elapsed < min_interval:
time.sleep(min_interval - elapsed)
self.last_request_time = time.time()
async def fetch_ohlcv_async(
self,
symbol: str,
timeframe: str,
start_time: datetime,
end_time: Optional[datetime] = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Fetch OHLCV data asynchronously using CCXT.
Args:
symbol: Trading pair (e.g., "BTC/USDT")
timeframe: Candle timeframe ("1m", "5m", "1h", "1d")
start_time: Start of data range
end_time: End of data range (defaults to now)
limit: Maximum candles per request (exchange-specific)
Returns:
DataFrame with OHLCV data and timestamp index
"""
all_candles = []
current_start = start_time
# CCXT requires milliseconds
since_ms = int(start_time.timestamp() * 1000)
end_ms = int(end_time.timestamp() * 1000) if end_time else None
print(f"Fetching {symbol} {timeframe} data from {start_time}")
while True:
self._rate_limit_check(calls_per_second=5)
try:
# Fetch candles
candles = await asyncio.to_thread(
self.exchange.fetch_ohlcv,
symbol,
timeframe=timeframe,
since=since_ms,
limit=limit
)
if not candles:
break
all_candles.extend(candles)
# Update start time for next request
last_candle_time = candles[-1][0]
since_ms = last_candle_time + 1
# Check if we've reached the end
if end_ms and since_ms >= end_ms:
break
# CCXT returns data in ascending order
# Break if we're past the end time
if candles[-1][0] >= (end_ms or float('inf')):
break
print(f" Retrieved {len(candles)} candles, "
f"latest timestamp: {pd.Timestamp(last_candle_time, unit='ms')}")
except ccxt.RateLimitExceeded:
print("Rate limit hit, waiting 60 seconds...")
await asyncio.sleep(60)
except ccxt.NetworkError as e:
print(f"Network error: {e}, retrying in 10 seconds...")
await asyncio.sleep(10)
except Exception as e:
print(f"Error fetching data: {e}")
break
# Convert to DataFrame
df = pd.DataFrame(
all_candles,
columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
# Remove duplicates and sort
df = df[~df.index.duplicated(keep='last')]
df = df.sort_index()
# Filter to exact date range
if end_time:
df = df[df.index <= end_time]
print(f"Total candles retrieved: {len(df)}")
return df
def fetch_funding_rates(self, symbol: str, start_time: datetime,
end_time: datetime) -> pd.DataFrame:
"""
Fetch funding rate history for perpetual futures.
Critical for funding rate arbitrage strategies.
"""
all_funding = []
since_ms = int(start_time.timestamp() * 1000)
end_ms = int(end_time.timestamp() * 1000)
# Set exchange to use futures
if hasattr(self.exchange, 'options'):
self.exchange.options['defaultType'] = 'future'
while since_ms < end_ms:
self._rate_limit_check(calls_per_second=2)
try:
funding = self.exchange.fetch_funding_history(
symbol, since=since_ms, limit=200
)
if not funding:
break
all_funding.extend(funding)
since_ms = funding[-1]['timestamp'] + 1
except Exception as e:
print(f"Error fetching funding rates: {e}")
break
df = pd.DataFrame(all_funding)
if not df.empty:
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
def validate_data_quality(self, df: pd.DataFrame) -> Dict[str, any]:
"""
Validate data quality and return quality report.
Checks for gaps, anomalies, and completeness.
"""
report = {
"total_candles": len(df),
"date_range": {
"start": df.index.min(),
"end": df.index.max()
},
"missing_candles": 0,
"price_anomalies": [],
"volume_anomalies": []
}
# Check for expected candle count
if len(df) > 1:
expected_interval = df.index[1] - df.index[0]
full_range = df.index.max() - df.index.min()
expected_candles = full_range / expected_interval + 1
report["missing_candles"] = expected_candles - len(df)
report["expected_vs_actual"] = f"{len(df)}/{int(expected_candles)}"
# Check for price anomalies (zero or negative)
anomalies = df[(df['close'] <= 0) | (df['high'] < df['low'])]
report["price_anomalies"] = len(anomalies)
# Check for volume anomalies
volume_pct = df['volume'].quantile([0.01, 0.99])
extreme_volume = df[
(df['volume'] < volume_pct[0.01]) |
(df['volume'] > volume_pct[0.99])
]
report["volume_anomalies"] = len(extreme_volume)
return report
Usage example
async def main():
config = HistoricalDataConfig()
fetcher = CCXTHistoricalDataFetcher(
exchange_id="binance",
config=config
)
# Fetch one year of daily BTC data
btc_daily = await fetcher.fetch_ohlcv_async(
symbol="BTC/USDT",
timeframe="1d",
start_time=datetime(2024, 1, 1),
end_time=datetime(2025, 1, 1)
)
# Validate data quality
quality_report = fetcher.validate_data_quality(btc_daily)
print(f"Data quality report: {quality_report}")
# Save to parquet for efficient storage
btc_daily.to_parquet("./data/btc_daily_2024.parquet")
if __name__ == "__main__":
asyncio.run(main())HolySheep AI Integration for Advanced Analysis
For sophisticated quant strategies, you may want to leverage AI for pattern recognition, strategy optimization, or market regime detection. HolySheep AI provides a cost-effective solution with $0.42 per million tokens for DeepSeek V3.2, compared to $8.00 for GPT-4.1 and $15.00 for Claude Sonnet 4.5. This pricing makes large-scale AI analysis economically viable for retail traders and small hedge funds.
# src/data_fetchers/holy_sheep_analyzer.py
import requests
import json
import pandas as pd
from typing import List, Dict, Optional
from datetime import datetime
class HolySheepQuantAnalyzer:
"""
Integrates with HolySheep AI for quantitative analysis.
Uses market data relay for trades, order book, liquidations, and funding rates.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def _make_request(self, endpoint: str, method: str = "GET",
data: dict = None) -> dict:
"""Make authenticated request to HolySheep API."""
url = f"{self.base_url}/{endpoint}"
try:
if method == "GET":
response = requests.get(url, headers=self.headers, params=data)
else:
response = requests.post(url, headers=self.headers, json=data)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise Exception("Invalid API key. Check your HolySheep credentials.")
elif response.status_code == 429:
raise Exception("Rate limit exceeded. Wait before retrying.")
else:
raise Exception(f"HTTP error {response.status_code}: {e}")
except requests.exceptions.ConnectionError:
raise Exception("Connection error: Unable to reach HolySheep API. "
"Check your internet connection.")
def analyze_market_regime(self, symbol: str, timeframe: str,
price_data: List[Dict]) -> Dict:
"""
Use AI to analyze current market regime.
Returns regime classification and recommended strategy parameters.
"""
prompt = f"""Analyze this {symbol} {timeframe} price data and identify the current market regime.
Price data is provided as a list of candles with 'open', 'high', 'low', 'close', 'volume'.
Return a JSON response with:
{{
"regime": "trending_up|trending_down|ranging|volatile|unknown",
"confidence": 0.0-1.0,
"recommended_strategy": "mean_reversion|momentum|breakout|neutral",
"risk_level": "low|medium|high",
"key_support_levels": [price_levels],
"key_resistance_levels": [price_levels],
"volatility_assessment": "low|medium|high",
"momentum_direction": "bullish|bearish|neutral",
"market_sentiment": "fear|greed|neutral"
}}
Price data sample (last 20 candles):
{json.dumps(price_data[-20:], indent=2)}
"""
payload = {
"model": "deepseek-chat", # Cost-effective: $0.42/MTok vs $8.00 for GPT-4.1
"messages": [
{"role": "system", "content": "You are an expert quantitative analyst specializing in cryptocurrency markets."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
result = self._make_request("chat/completions", method="POST", data=payload)
# Parse AI response
content = result['choices'][0]['message']['content']
# Extract JSON from response
try:
# Find JSON in the response
start = content.find('{')
end = content.rfind('}') + 1
analysis = json.loads(content[start:end])
return analysis
except:
return {"error": "Failed to parse AI response", "raw": content}
def optimize_strategy_parameters(self, strategy_type: str,
historical_results: Dict) -> Dict:
"""
Use AI to optimize strategy parameters based on backtest results.
Identifies parameter combinations that maximize Sharpe ratio.
"""
prompt = f"""Optimize parameters for a {strategy_type} trading strategy based on backtest results.
Backtest Results:
- Total Return: {historical_results.get('total_return', 'N/A')}%
- Sharpe Ratio: {historical_results.get('sharpe_ratio', 'N/A')}
- Max Drawdown: {historical_results.get('max_drawdown', 'N/A')}%
- Win Rate: {historical_results.get('win_rate', 'N/A')}%
- Total Trades: {historical_results.get('total_trades', 'N/A')}
- Average Trade Duration: {historical_results.get('avg_duration', 'N/A')}
Parameter Space Tested:
{json.dumps(historical_results.get('parameter_grid', {}), indent=2)}
Return optimized parameters and explain why they perform better.
"""
payload = {
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": "You are an expert algorithmic trading strategist."},
{"role": "user", "content": prompt}
],
"temperature": 0.2,
"max_tokens": 800
}
result = self._make_request("chat/completions", method="POST", data=payload)
return result['choices'][0]['message']['content']
def detect_patterns(self, ohlcv_df: pd.DataFrame,
symbol: str) -> List[Dict]:
"""
Use AI to detect chart patterns and technical formations.
"""
# Prepare data summary
recent_data = ohlcv_df.tail(50).to_dict('records')
prompt = f"""Analyze this {symbol} price chart and identify any notable chart patterns or technical formations.
Recent price data (50 candles):
{json.dumps(recent_data[:20], indent=2)} # Send first 20 as sample
Identify any of the following patterns:
- Technical: double_top, double_bottom, head_shoulders, cup_handle, wedge
- Candlestick: doji, hammer, engulfing, morning_star, evening_star
- Trend: ascending_triangle, descending_triangle, symmetrical_triangle
Return patterns found with confidence scores and price targets.
"""
payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 600
}
result = self._make_request("chat/completions", method="POST", data=payload)
return {"patterns": result['choices'][0]['message']['content']}
def generate_trading_signals(self, market_data: Dict) -> Dict:
"""
Generate trading signals using HolySheep AI analysis.
Combines technical analysis with market microstructure data.
"""
prompt = f"""Generate a comprehensive trading signal analysis for {market_data.get('symbol', 'UNKNOWN')}.
Market Data:
{json.dumps(market_data, indent=2)}
Provide:
1. Overall signal: BUY, SELL, or NEUTRAL
2. Entry price suggestion
3. Stop loss level (with rationale)
4. Take profit levels (multiple targets)
5. Position sizing recommendation
6. Risk/reward ratio
7. Key risks to monitor
8. Time horizon recommendation
"""
payload = {
"model": "deepseek-chat",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.4,
"max_tokens": 700
}
result = self._make_request("chat/completions", method="POST", data=payload)
return {"signal": result['choices'][0]['message']['content']}
Example usage with cost tracking
def main():
# Initialize analyzer
analyzer = HolySheepQuantAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
# Create sample price data
sample_prices = [
{"open": 42000, "high": 42500, "low": 41800, "close": 42300, "volume": 25000},
{"open": 42300, "high": 42800, "low": 42100, "close": 42650, "volume": 28000},
# ... more data
] * 20 # Replicate to get 20 candles
# Analyze market regime
try:
regime_analysis = analyzer.analyze_market_regime(
symbol="BTC/USDT",
timeframe="1h",
price_data=sample_prices
)
print(f"Market Regime: {regime_analysis.get('regime', 'unknown')}")
print(f"Confidence: {regime_analysis.get('confidence', 0):.2%}")
print(f"Recommended Strategy: {regime_analysis.get('recommended_strategy', 'N/A')}")
except Exception as e:
print(f"Analysis error: {e}")
if __name__ == "__main__":
main()Data Storage and Retrieval Optimization
Efficient data storage is crucial for backtesting performance. Parquet format provides excellent compression and fast read/write speeds, making it ideal for large historical datasets. A typical one-year dataset for 10 trading pairs across 5 timeframes can consume several gigabytes in CSV format but only 200-500MB in Parquet.
# src/data_storage/optimized_storage.py
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os
from pathlib import Path
from typing import List, Dict, Optional
from datetime import datetime
import hashlib
class HistoricalDataStore:
"""
Optimized storage system for historical market data.
Uses Parquet format with partitioning for efficient querying.
"""
def __init__(self, base_path: str = "./data/historical"):
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
def _get_partition_path(self, exchange: str, symbol: str,
timeframe: str) -> Path:
"""Generate partition path for data organization."""
# Path format: base_path/exchange/symbol/timeframe/
symbol_clean = symbol.replace('/', '_')
return self.base_path / exchange / symbol_clean / timeframe
def save_ohlcv(self, df: pd.DataFrame, exchange: str, symbol: str,
timeframe: str, partition: str = "daily"):
"""
Save OHLCV data with intelligent partitioning.
Partition strategies:
- "daily": One file per day (best for large datasets, frequent updates)
- "monthly": One file per month (balanced)
- "static": Single file for entire dataset (fastest reads)
"""
partition_path = self._get_partition_path(exchange, symbol, timeframe)
if partition == "daily":
# Partition by date
df['date'] = df.index.date
for date, group_df in df.groupby('date'):
date_path = partition_path / f"date={date}"
date_path.mkdir(parents=True, exist_ok=True)
file_path = date_path / "data.parquet"
group_df.drop(columns=['date']).to_parquet(
file_path,
engine='pyarrow',
compression='snappy' # Good balance of speed and compression
)
else:
# Single file storage
partition_path.mkdir(parents=True, exist_ok=True)
file_path = partition_path / "data.parquet"
df.to_parquet(file_path, engine='pyarrow', compression='snappy')
print(f"Saved {len(df)} candles to {partition_path}")
def load_ohlcv(self, exchange: str, symbol: str, timeframe: str,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None) -> pd.DataFrame:
"""
Load OHLCV data with date range filtering.
Uses predicate pushdown for efficient Parquet queries.
"""
partition_path = self._get_partition_path(exchange, symbol, timeframe)
if not partition_path.exists():
raise FileNotFoundError(f"No data found at {partition_path}")
# Check if partitioned data exists
daily_files = list(partition_path.glob("date=*/data.parquet"))
if daily_files:
# Load partitioned data
tables = []
for file_path in daily_files:
date_str = file_path.parent.name.replace("date=", "")
file_date = datetime.strptime(date_str, "%Y-%m-%d").date()
# Filter by date range
if start_date and file_date < start_date.date():
continue
if end_date and file_date > end_date.date():
continue
df = pd.read_parquet(file_path)
tables.append(df)
if tables:
result = pd.concat(tables, ignore_index=False)
else:
result = pd.DataFrame()
else:
# Load single file
file_path = partition_path / "data.parquet"
result = pd.read_parquet(file_path)
# Apply date filter
if start_date:
result = result[result.index >= start_date]
if end_date:
result = result[result.index <= end_date]
return result.sort_index()
def get_data_info(self, exchange: str, symbol: str,
timeframe: str) -> Dict:
"""Get information about stored data without loading it."""
partition_path = self._get_partition_path(exchange, symbol, timeframe)
if not partition_path.exists():
return {"exists": False}
daily_files = list(partition_path.glob("date=*/data.parquet"))
if daily_files:
# Get metadata from first and last files
first_df = pd.read_parquet(daily_files[0])
last_df = pd.read_parquet(daily_files[-1])
return {
"exists": True,
"partition_type": "daily",
"file_count": len(daily_files),
"date_range": {
"start": daily_files[0].parent.name.replace("date=", ""),
"end": daily_files[-1].parent.name.replace("date=", "")
},
"row_count": len(first_df) + len(last_df), # Approximate
"sample_columns": list(first_df.columns)
}
else:
file_path = partition_path / "data.parquet"
if file_path.exists():
pf = pq.ParquetFile(file_path)
schema = pf.schema_arrow
return {
"exists": True,
"partition_type": "static",
"file_size_mb": file_path.stat().st_size / (1024 * 1024),
"row_count": pf.metadata.num_rows,
"columns": [s.name for s in schema]
}
return {"exists": False}
def compress_existing_data(self, exchange: str, symbol: str,
timeframe: str):
"""Compress existing CSV data to Parquet format."""
symbol_clean = symbol.replace('/', '_')
csv_path = self.base_path / f"{exchange}_{symbol_clean}_{timeframe}.csv"
if csv_path.exists():
df = pd.read_csv(csv_path, parse_dates=True, index_col=0)
partition_path = self._get_partition_path(exchange, symbol, timeframe)
partition_path.mkdir(parents=True, exist_ok=True)
output_path = partition_path / "data.parquet"
df.to_parquet(output_path, compression='snappy')
csv_size = csv_path.stat().st_size / (1024 * 1024)
parquet_size = output_path.stat().st_size / (1024 * 1024)
print(f"Compressed {csv_path} ({csv_size:.2f}MB) to "
f"{output_path} ({parquet_size:.2f}MB) - "
f"{100 * (1 - parquet_size/csv_size):.1f}% reduction")
Backtesting data pipeline
class BacktestDataPipeline:
"""
Complete pipeline for preparing data for backtesting.
Handles data fetching, cleaning, feature engineering, and storage.
"""
def __init__(self, store: HistoricalDataStore, fetcher = None):
self.store = store
self.fetcher = fetcher
def prepare_backtest_data(
self,
symbols: List[str],
timeframes: List[str],
start_date: datetime,
end_date: datetime,
exchanges: List[str] = None,
compute_features: bool = True
) -> Dict[str, Dict[str, pd.DataFrame]]:
"""
Prepare complete dataset for backtesting.
Returns nested dictionary: {exchange: {symbol_timeframe: DataFrame}}
"""
if exchanges is None:
exchanges = ["binance"]
all_data = {}
for exchange in exchanges:
exchange_data = {}
for symbol in symbols:
for timeframe in timeframes:
key = f"{symbol}_{timeframe}"
try:
# Try to load from cache
df = self.store.load_ohlcv(
exchange, symbol, timeframe,
start_date, end_date
)
print(f"Loaded cached data for {exchange}/{symbol}/{timeframe}")
except FileNotFoundError:
# Fetch from exchange
if self.fetcher:
print(f"Fetching data for {exchange}/{symbol}/{timeframe}")
df = self.fetcher.fetch_ohlcv(
symbol=symbol,
timeframe=timeframe,
start_time=start_date,
end_time=end_date
)
# Cache the data
self.store.save_ohlcv(
df, exchange, symbol, timeframe
)
else:
raise Exception(f"No data and no fetcher configured")
if compute_features:
df = self._add_technical_features(df)
exchange_data[key] = df
all_data[exchange] = exchange_data
return all_data
def _add_technical_features(self, df: pd.DataFrame) -> pd.DataFrame:
"""Add common technical indicators as features."""
df = df.copy()
# Price-based features
df['returns'] = df['close'].pct_change()
df['log_returns'] = np.log(df['close'] / df['close'].shift(1))
# Moving averages
for window in [5, 10, 20, 50]:
df[f'sma_{window}'] = df['close'].rolling(window).mean()
df[f'ema_{window}'] = df['close'].ewm(span=window).mean()
# Volatility
df['volatility_20'] = df['returns'].rolling(20).std()
df['volatility_50'] = df['returns'].rolling(50).std()
# Volume features
df['volume_sma_20'] = df['volume'].rolling(20).mean()
df['volume_ratio'] = df['volume'] / df['volume_sma_20']
# High-Low range
df['hl_range'] = (df['high'] - df['low']) / df['close']
# Drop NaN rows created by rolling calculations
df = df.dropna()
return df
Usage
import numpy as np
store = HistoricalDataStore("./data/historical")
pipeline = BacktestDataPipeline(store)
data = pipeline.prepare_backtest_data(
symbols=["BTC/USDT",