data-backfill-strategy"Provides Strategic data backfill for populating historical data in trading systems"
Independent GitHub snapshot · Automated classification · Original documentation governs
name: data-backfill-strategy compatibility: opencode completeness: 95 content-types:
Role: Efficiently populate and maintain historical data for analysis and backtesting
Philosophy: Historical data quality determines backtest accuracy; backfill must be comprehensive, efficient, and auditable
Before completing your task, verify:
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
import time
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
import logging
class BackfillStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
PAUSED = "paused"
@dataclass
class BackfillTask:
"""Represents a backfill task."""
task_id: str
symbol: str
start_time: float
end_time: float
status: BackfillStatus = BackfillStatus.PENDING
data_type: str = "candles"
progress: float = 0.0
created_at: float = field(default_factory=time.time)
started_at: Optional[float] = None
completed_at: Optional[float] = None
error: Optional[str] = None
@dataclass
class BackfillConfig:
"""Configuration for backfill."""
batch_size: int = 1000
parallel_workers: int = 4
max_retries: int = 3
retry_delay_seconds: float = 60.0
rate_limit_per_second: int = 10
class BackfillManager:
"""Manages historical data backfill operations."""
def __init__(self, config: BackfillConfig, data_source: Any):
self.config = config
self.data_source = data_source
self.tasks: Dict[str, BackfillTask] = {}
self.progress: Dict[str, Dict[str, Any]] = {}
self._lock = None
def create_backfill_task(
self,
symbol: str,
start_time: float,
end_time: float,
data_type: str = "candles"
) -> str:
"""Create a new backfill task."""
task_id = f"backfill_{symbol}_{int(start_time)}_{int(end_time)}"
task = BackfillTask(
task_id=task_id,
symbol=symbol,
start_time=start_time,
end_time=end_time,
data_type=data_type
)
self.tasks[task_id] = task
return task_id
async def execute_backfill(self, task_id: str) -> bool:
"""Execute a backfill task."""
task = self.tasks.get(task_id)
if not task:
return False
task.status = BackfillStatus.RUNNING
task.started_at = time.time()
try:
await self._execute_backfill_internal(task)
task.status = BackfillStatus.COMPLETED
task.completed_at = time.time()
return True
except Exception as e:
task.status = BackfillStatus.FAILED
task.error = str(e)
task.completed_at = time.time()
logging.error(f"Backfill failed for {task_id}: {e}")
return False
async def _execute_backfill_internal(self, task: BackfillTask):
"""Internal backfill execution."""
# Detect gaps first
gaps = await self._detect_gaps(task.symbol, task.start_time, task.end_time)
if not gaps:
task.progress = 100.0
return
# Process each gap
total_gaps = len(gaps)
processed = 0
for gap in gaps:
start, end = gap
await self._fetch_and_store(task.symbol, start, end, task.data_type)
processed += 1
task.progress = (processed / total_gaps) * 100
task.progress = 100.0
async def _detect_gaps(
self,
symbol: str,
start_time: float,
end_time: float
) -> List[Tuple[float, float]]:
"""Detect gaps in existing data."""
# This would query the data store in a real implementation
# For example, if we have data from 2020-01-01 to 2020-06-01
# and 2020-07-01 to 2020-12-01, gaps would be [(2020-06-01, 2020-07-01)]
# Simplified: return the requested range as gap
# In reality, would check existing data
return [(start_time, end_time)]
async def _fetch_and_store(
self,
symbol: str,
start_time: float,
end_time: float,
data_type: str
):
"""Fetch and store data for a time range."""
# Calculate required range based on data type
if data_type == "candles":
# Fetch hourly candles
interval = 3600 # 1 hour
elif data_type == "tick":
interval = 60 # 1 minute
else:
interval = 3600
# Fetch in batches
current_start = start_time
while current_start < end_time:
current_end = min(current_start + (self.config.batch_size * interval), end_time)
# Fetch data
data = await self._fetch_data(symbol, current_start, current_end, data_type)
if data:
# Store data
await self._store_data(symbol, data, data_type)
current_start = current_end
async def _fetch_data(
self,
symbol: str,
start_time: float,
end_time: float,
data_type: str
) -> List[Dict[str, Any]]:
"""Fetch data from external source."""
# This would call the actual data source API
# For now, return empty list (placeholder)
return []
async def _store_data(
self,
symbol: str,
data: List[Dict[str, Any]],
data_type: str
):
"""Store fetched data."""
# This would persist data to the data store
pass
def get_task(self, task_id: str) -> Optional[BackfillTask]:
"""Get backfill task by ID."""
return self.tasks.get(task_id)
def get_all_tasks(self) -> List[BackfillTask]:
"""Get all backfill tasks."""
return list(self.tasks.values())
def resume_task(self, task_id: str) -> bool:
"""Resume a paused or failed task."""
task = self.tasks.get(task_id)
if not task:
return False
if task.status in [BackfillStatus.PAUSED, BackfillStatus.FAILED]:
task.status = BackfillStatus.RUNNING
return True
return False
def pause_task(self, task_id: str) -> bool:
"""Pause a running task."""
task = self.tasks.get(task_id)
if not task:
return False
if task.status == BackfillStatus.RUNNING:
task.status = BackfillStatus.PAUSED
return True
return False
class GapFinder:
"""Finds gaps in historical data."""
def __init__(self, data_store: Any):
self.store = data_store
async def find_gaps(
self,
symbol: str,
start_time: float,
end_time: float,
interval_seconds: float
) -> List[Tuple[float, float]]:
"""Find gaps in data for a time range."""
gaps = []
# Get existing data timestamps
existing = await self._get_existing_timestamps(symbol, start_time, end_time)
if not existing:
return [(start_time, end_time)]
# Sort existing timestamps
existing_sorted = sorted(existing)
# Find gaps between consecutive timestamps
for i in range(len(existing_sorted) - 1):
current = existing_sorted[i]
next_timestamp = existing_sorted[i + 1]
expected_next = current + interval_seconds
if next_timestamp > expected_next:
gaps.append((expected_next, next_timestamp - 1))
# Check for gaps at the beginning
if existing_sorted[0] > start_time + interval_seconds:
gaps.insert(0, (start_time, existing_sorted[0] - 1))
# Check for gaps at the end
if existing_sorted[-1] < end_time - interval_seconds:
gaps.append((existing_sorted[-1] + interval_seconds, end_time))
return gaps
async def _get_existing_timestamps(
self,
symbol: str,
start_time: float,
end_time: float
) -> List[float]:
"""Get existing timestamps for symbol."""
# This would query the data store
return []
async def fill_gaps(
self,
symbol: str,
data_type: str,
interval_seconds: float
) -> int:
"""Fill all gaps for a symbol. Returns number of gaps filled."""
gaps = await self.find_gaps(symbol, 0, time.time(), interval_seconds)
filled = 0
for start, end in gaps:
# Fetch and store missing data
# Implementation would call backfill manager
filled += 1
return filled
class BackfillProgressTracker:
"""Tracks and persists backfill progress."""
def __init__(self, storage_path: str):
self.storage_path = storage_path
self._progress: Dict[str, Any] = {}
self._load_progress()
def _load_progress(self):
"""Load saved progress."""
# This would read from persistent storage
pass
def save_progress(self, task_id: str, progress: float, current_time: float):
"""Save backfill progress."""
self._progress[task_id] = {
"progress": progress,
"last_update": current_time
}
# Would persist to storage
def get_progress(self, task_id: str) -> Optional[Dict[str, Any]]:
"""Get saved progress for task."""
return self._progress.get(task_id)
def get_all_progress(self) -> Dict[str, Dict]:
"""Get progress for all tasks."""
return self._progress.copy()
class BatchBackfillWorker:
"""Worker for parallel backfill processing."""
def __init__(self, task_id: str, data_source: Any):
self.task_id = task_id
self.data_source = data_source
self.completed_ranges: List[Tuple[float, float]] = []
async def process_batch(
self,
symbol: str,
start_time: float,
end_time: float,
data_type: str
) -> bool:
"""Process a single batch."""
try:
data = await self._fetch_data(symbol, start_time, end_time, data_type)
if data:
await self._store_data(symbol, data, data_type)
self.completed_ranges.append((start_time, end_time))
return True
return False
except Exception as e:
logging.error(f"Batch processing failed: {e}")
return False
async def _fetch_data(
self,
symbol: str,
start_time: float,
end_time: float,
data_type: str
) -> List[Dict[str, Any]]:
"""Fetch data for batch."""
# Implementation
return []
async def _store_data(
self,
symbol: str,
data: List[Dict[str, Any]],
data_type: str
):
"""Store batch data."""
# Implementation
pass
def get_completed_ranges(self) -> List[Tuple[float, float]]:
"""Get list of completed time ranges."""
return self.completed_ranges
from __future__ import annotations
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Iterator
from uuid import UUID, uuid7
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class BackfillCheckpoint:
"""Tracks the progress of a backfill operation for resume capability."""
symbol: str
timeframe: str
source: str
last_processed_ts: datetime
total_records_fetched: int = 0
last_updated: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
def advance(self, new_ts: datetime, count: int) -> BackfillCheckpoint:
return BackfillCheckpoint(
symbol=self.symbol,
timeframe=self.timeframe,
source=self.source,
last_processed_ts=new_ts,
total_records_fetched=self.total_records_fetched + count,
)
class BackfillManager:
"""Manages incremental backfill operations with checkpoint persistence."""
def __init__(self, data_source, checkpoint_store):
self._source = data_source
self._store = checkpoint_store
def get_checkpoint(self, symbol: str, timeframe: str, source: str) -> BackfillCheckpoint | None:
"""Load the last backfill checkpoint for a given data stream."""
key = f"{symbol}:{timeframe}:{source}"
return self._store.get(key)
def save_checkpoint(self, cp: BackfillCheckpoint) -> None:
"""Persist a checkpoint after successfully processing data."""
key = f"{cp.symbol}:{cp.timeframe}:{cp.source}"
self._store.save(key, cp)
logger.info("Checkpoint saved: %s (processed %d records)", key, cp.total_records_fetched)
def backfill_range(
self,
symbol: str,
start: datetime,
end: datetime,
timeframe: str = "1h",
max_batch_size: int = 5000,
) -> dict:
"""Execute a full or incremental backfill between two timestamps.
If a checkpoint exists for this stream, starts from the last processed point.
Otherwise performs a full range backfill from `start`.
Args:
symbol: Trading pair identifier.
start: Start of the desired data range (inclusive).
end: End of the desired data range (exclusive).
timeframe: Data granularity ("1m", "5m", "1h", etc.).
max_batch_size: Maximum records per API call to avoid rate limits.
Returns:
Summary dict with total records, duration, and gaps detected.
"""
existing = self.get_checkpoint(symbol, timeframe, self._source)
if existing and existing.last_processed_ts > start:
logger.info("Resuming backfill from checkpoint: %s", existing.last_processed_ts)
actual_start = existing.last_processed_ts + timedelta(seconds=1)
else:
actual_start = start
total_fetched = 0
gaps_detected = 0
batch_start = actual_start
while batch_start < end:
batch_end = min(batch_start + timedelta(hours=24), end)
try:
records = self._source.fetch_ohlcv(
symbol=symbol,
start=batch_start,
end=batch_end,
timeframe=timeframe,
limit=max_batch_size,
)
except Exception as e:
logger.error("Failed to fetch batch %s → %s for %s: %s", batch_start, batch_end, symbol, e)
time.sleep(5)
continue
if records:
last_ts = max(r["timestamp"] for r in records)
total_fetched += len(records)
self.save_checkpoint(
BackfillCheckpoint(symbol=symbol, timeframe=timeframe, source=self._source,
last_processed_ts=last_ts, total_records_fetched=total_fetched),
)
expected_interval = {"1m": 60, "5m": 300, "1h": 3600}.get(timeframe, 3600)
if len(records) < (batch_end - batch_start).total_seconds() / expected_interval:
gaps_detected += 1
batch_start = batch_end
time.sleep(0.5) # Respect rate limits between batches
return {
"symbol": symbol,
"start": actual_start.isoformat(),
"end": end.isoformat(),
"timeframe": timeframe,
"records_fetched": total_fetched,
"gaps_detected": gaps_detected,
}
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.