# Data Backfill Strategy > "Provides Strategic data backfill for populating historical data in trading systems" - Skill: `paulpas/data-backfill-strategy` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/data-backfill-strategy` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/data-backfill-strategy/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/data-backfill-strategy --- **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 ## Key Principles 1. **Incremental Backfill**: Only fetch missing data to reduce API load 2. **Batch Processing**: Process data in optimized batches 3. **Gap Detection**: Identify and fill missing data periods 4. **Data Versioning**: Track backfill versions for reproducibility 5. **Progress Tracking**: Monitor and resume interrupted backfills ## Implementation Guidelines ### Structure - Core logic: backfill/backfill_manager.py - Gap finder: backfill/finder.py - Tests: tests/test_backfill.py ### Patterns to Follow - Use parallel processing for efficiency - Implement backfill checkpoints - Support multiple data sources - Track backfill performance metrics ## Adherence Checklist Before completing your task, verify: - [ ] Incremental backfill reduces API calls - [ ] Gaps are detected and filled - [ ] Backfill progress is persisted - [ ] Interrupted backfills can resume - [ ] Backfill quality is validated Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python 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 ``` --- --- ### Pattern 2: Checkpoint-Based Incremental Backfill with Deduplication ```python 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, } ``` ## Constraints ### MUST DO - Validate all incoming data against schema constraints (type, range, nullability) before processing or storage - Implement idempotent operations: re-processing the same data must produce identical results - Track data lineage and provenance with timestamps, source identifiers, and transformation history for every record - Handle out-of-order data by implementing a watermark-based ordering mechanism with configurable tolerance window - Log data quality metrics (completeness, freshness, accuracy) per source with automatic alerting on degradation ### MUST NOT DO - Do not silently drop records that fail validation — log them to a quarantine table for review - Avoid concatenating strings for timestamp comparison; use proper datetime/timedelta objects - Never assume data arrives in chronological order from any external feed without explicit ordering guarantees - Do not store raw and processed data in the same table without clear partitioning or separation strategy - Avoid blocking on slow data sources — implement async prefetch with timeout-based fallback to cached data ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [QuantConnect Data Overview](https://docs.quantconnect.com/dataset/overview) - [Data Gap Handling in Backtests](https://docs.quantconnect.com/tutorials/data-sources-and-format) - [Market Data Quality Best Practices](https://docs.quantconnect.com/tutorials/data-market-data) - [Historical Data Recovery Methods](https://www.investopedia.com/terms/h/historical-data.asp) - [Data Validation and Cleaning](https://greatexpectations.io/)