data-backfill-strategy

---

name: data-backfill-strategy compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Provides Strategic data backfill for populating historical data in trading systems"' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: ai-order-flow-analysis, data-alternative-data, data-candle-data role: implementation scope: implementation triggers: data backfill strategy, data-backfill-strategy, historical, populating, strategic archetypes:
  • tactical anti_triggers:
  • brainstorming
  • vague ideation
  • no risk management response_profile: verbosity: low directive_strength: high abstraction_level: operational version: "1.0.0"

---

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

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

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
• Data Gap Handling in Backtests
• Market Data Quality Best Practices
• Historical Data Recovery Methods
• Data Validation and Cleaning