Data Time Series Database

---




name: data-time-series-database
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Provides Time-series database queries and optimization for financial
  data"'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: ai-order-flow-analysis, data-alternative-data
  role: implementation
  scope: implementation
  triggers: data time series database, data-time-series-database, optimization, queries,
    time-series, performance, speed
  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 query time-series data with support for financial data patterns

**Philosophy:** Time-series queries are the bread and butter of trading; optimization enables faster decisions

## Key Principles

1. **Time-Ordered Indexing**: Optimize for sequential time-based queries
2. **Aggregation Primitives**: Support common financial aggregations (OHLCV, VWAP, etc.)
3. **Downsampling**: Efficiently reduce resolution for longer time ranges
4. **Gap Handling**: Handle missing data points in financial series
5. **Partitioning by Time**: Automatically partition data by time ranges

## Implementation Guidelines

### Structure
- Core logic: tsdb/tsdb.py
- Aggregations: tsdb/aggregations.py
- Tests: tests/test_tsdb.py

### Patterns to Follow
- Use time-indexed data structures
- Implement efficient downsampling
- Support interval queries
- Track query performance

## Adherence Checklist
Before completing your task, verify:
- [ ] Queries use time-based indexing
- [ ] Aggregations are computed efficiently
- [ ] Downsampling preserves data integrity
- [ ] Gap detection and interpolation is supported
- [ ] Query performance is monitored


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, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from collections import OrderedDict
import bisect
import logging

@dataclass
class TimeSeriesPoint:
    """Single point in time series."""
    timestamp: float
    value: float
    metadata: Dict[str, Any] = field(default_factory=dict)

@dataclass
class TimeSeriesQuery:
    """Query parameters for time series."""
    symbol: str
    start_time: float
    end_time: float
    resolution_seconds: Optional[float] = None
    aggregation: Optional[str] = None  # "mean", "first", "last", "ohlc"
    limit: Optional[int] = None

class TimeSeriesStorage:
    """Storage for time series data with time-indexed access."""
    
    def __init__(self, max_points_per_series: int = 1000000):
        self.max_points = max_points_per_series
        self._data: Dict[str, OrderedDict[float, float]] = {}
        self._metadata: Dict[str, Dict[float, Dict]] = {}
        self._lock = None  # Would use threading.Lock
    
    def insert(
        self,
        symbol: str,
        timestamp: float,
        value: float,
        metadata: Dict[str, Any] = None
    ):
        """Insert a data point."""
        if symbol not in self._data:
            self._data[symbol] = OrderedDict()
            self._metadata[symbol] = {}
        
        # Remove if exists (update)
        if timestamp in self._data[symbol]:
            del self._data[symbol][timestamp]
        
        # Insert at correct position
        self._data[symbol][timestamp] = value
        
        if metadata:
            self._metadata[symbol][timestamp] = metadata
        
        # Enforce max size (remove oldest)
        while len(self._data[symbol]) > self.max_points:
            oldest_key = next(iter(self._data[symbol]))
            del self._data[symbol][oldest_key]
            if oldest_key in self._metadata.get(symbol, {}):
                del self._metadata[symbol][oldest_key]
    
    def query(
        self,
        symbol: str,
        start_time: float,
        end_time: float,
        limit: Optional[int] = None
    ) -> List[TimeSeriesPoint]:
        """Query time series by time range."""
        if symbol not in self._data:
            return []
        
        points = []
        series = self._data[symbol]
        
        # Find start position using binary search
        timestamps = list(series.keys())
        start_idx = bisect.bisect_left(timestamps, start_time)
        
        count = 0
        for i in range(start_idx, len(timestamps)):
            timestamp = timestamps[i]
            if timestamp > end_time:
                break
            
            if limit and count >= limit:
                break
            
            value = series[timestamp]
            metadata = self._metadata.get(symbol, {}).get(timestamp, {})
            
            points.append(TimeSeriesPoint(
                timestamp=timestamp,
                value=value,
                metadata=metadata
            ))
            count += 1
        
        return points
    
    def get_latest(self, symbol: str) -> Optional[TimeSeriesPoint]:
        """Get most recent data point."""
        if symbol not in self._data or not self._data[symbol]:
            return None
        
        timestamp, value = next(reversed(self._data[symbol].items()))
        metadata = self._metadata.get(symbol, {}).get(timestamp, {})
        
        return TimeSeriesPoint(
            timestamp=timestamp,
            value=value,
            metadata=metadata
        )
    
    def delete_range(
        self,
        symbol: str,
        start_time: float,
        end_time: float
    ) -> int:
        """Delete data points in range. Returns count deleted."""
        if symbol not in self._data:
            return 0
        
        timestamps = list(self._data[symbol].keys())
        start_idx = bisect.bisect_left(timestamps, start_time)
        end_idx = bisect.bisect_right(timestamps, end_time)
        
        deleted = 0
        for i in range(start_idx, end_idx):
            ts = timestamps[i]
            del self._data[symbol][ts]
            if ts in self._metadata.get(symbol, {}):
                del self._metadata[symbol][ts]
            deleted += 1
        
        return deleted

class TimeSeriesDB:
    """Time-series database with aggregation support."""
    
    def __init__(self):
        self.storage = TimeSeriesStorage()
        self._initialized = False
    
    def init(self):
        """Initialize database (create indexes, etc.)."""
        self._initialized = True
    
    def insert_candles(
        self,
        symbol: str,
        candles: List[Dict[str, Any]]
    ):
        """Insert candle data."""
        for candle in candles:
            # Insert each metric separately
            for metric in ["open", "high", "low", "close", "volume"]:
                if metric in candle:
                    self.storage.insert(
                        symbol=f"{symbol}.{metric}",
                        timestamp=candle["timestamp"],
                        value=float(candle[metric])
                    )
    
    def query_candles(
        self,
        symbol: str,
        start_time: float,
        end_time: float,
        resolution_seconds: float = 3600
    ) -> List[Dict[str, Any]]:
        """Query candle data with optional resampling."""
        candles = []
        
        # Query all metrics
        metrics = ["open", "high", "low", "close", "volume"]
        metric_data = {}
        
        for metric in metrics:
            points = self.storage.query(
                symbol=f"{symbol}.{metric}",
                start_time=start_time,
                end_time=end_time
            )
            if points:
                metric_data[metric] = points
        
        # Combine into candles
        min_length = min(len(data) for data in metric_data.values()) if metric_data else 0
        
        for i in range(min_length):
            candle = {"timestamp": metric_data["open"][i].timestamp}
            for metric in metrics:
                candle[metric] = metric_data[metric][i].value
            candles.append(candle)
        
        return candles
    
    def query_ohlc(
        self,
        symbol: str,
        start_time: float,
        end_time: float
    ) -> List[Dict[str, Any]]:
        """Query OHLC data."""
        candles = []
        
        for metric in ["open", "high", "low", "close"]:
            points = self.storage.query(
                symbol=f"{symbol}.{metric}",
                start_time=start_time,
                end_time=end_time
            )
            if points:
                if len(candles) == 0:
                    candles = [{} for _ in points]
                
                for i, point in enumerate(points):
                    candles[i][metric] = point.value
                    candles[i]["timestamp"] = point.timestamp
        
        return candles
    
    def compute_aggregates(
        self,
        symbol: str,
        start_time: float,
        end_time: float,
        window_seconds: float
    ) -> List[Dict[str, Any]]:
        """Compute time-windowed aggregates."""
        points = self.storage.query(symbol, start_time, end_time)
        
        if not points:
            return []
        
        aggregates = []
        current_start = points[0].timestamp
        current_values = []
        
        for point in points:
            if point.timestamp > current_start + window_seconds:
                if current_values:
                    aggregates.append({
                        "timestamp": current_start,
                        "open": current_values[0],
                        "high": max(current_values),
                        "low": min(current_values),
                        "close": current_values[-1],
                        "count": len(current_values)
                    })
                current_start += window_seconds
                current_values = []
            
            current_values.append(point.value)
        
        # Final bucket
        if current_values:
            aggregates.append({
                "timestamp": current_start,
                "open": current_values[0],
                "high": max(current_values),
                "low": min(current_values),
                "close": current_values[-1],
                "count": len(current_values)
            })
        
        return aggregates

class GapDetector:
    """Detects gaps in time series data."""
    
    def __init__(self, max_gap_seconds: float = 60.0):
        self.max_gap = max_gap_seconds
    
    def detect_gaps(
        self,
        points: List[TimeSeriesPoint]
    ) -> List[Dict[str, Any]]:
        """Detect gaps between consecutive points."""
        gaps = []
        
        for i in range(1, len(points)):
            prev_time = points[i-1].timestamp
            curr_time = points[i].timestamp
            gap_seconds = curr_time - prev_time
            
            if gap_seconds > self.max_gap:
                gaps.append({
                    "start": prev_time,
                    "end": curr_time,
                    "duration_seconds": gap_seconds,
                    "missing_points": int(gap_seconds / 60)  # Assume 1min data
                })
        
        return gaps

class Downsample:
    """Downsample time series data."""
    
    @staticmethod
    def mean(
        points: List[TimeSeriesPoint],
        window_seconds: float
    ) -> List[TimeSeriesPoint]:
        """Downsample by computing mean in windows."""
        if not points:
            return []
        
        result = []
        current_start = points[0].timestamp
        current_sum = 0.0
        current_count = 0
        
        for point in points:
            if point.timestamp > current_start + window_seconds:
                if current_count > 0:
                    result.append(TimeSeriesPoint(
                        timestamp=current_start,
                        value=current_sum / current_count
                    ))
                current_start += window_seconds
                current_sum = 0.0
                current_count = 0
            
            current_sum += point.value
            current_count += 1
        
        if current_count > 0:
            result.append(TimeSeriesPoint(
                timestamp=current_start,
                value=current_sum / current_count
            ))
        
        return result
    
    @staticmethod
    def first(
        points: List[TimeSeriesPoint],
        window_seconds: float
    ) -> List[TimeSeriesPoint]:
        """Downsample using first value in windows."""
        if not points:
            return []
        
        result = []
        current_start = points[0].timestamp
        current_value = points[0].value
        
        for point in points:
            if point.timestamp > current_start + window_seconds:
                result.append(TimeSeriesPoint(
                    timestamp=current_start,
                    value=current_value
                ))
                current_start += window_seconds
                current_value = point.value
        
        result.append(TimeSeriesPoint(
            timestamp=current_start,
            value=current_value
        ))
        
        return result
```

---

---


### Pattern 2: Time-Series Queries with InfluxDB-style Filter Expressions

```python
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass
class TSQueryFilter:
    """Declarative filter for time-series database queries."""
    symbol: str | None = None
    timeframe: str | None = None
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None
    min_volume: Optional[float] = None
    tags: dict[str, str] = field(default_factory=dict)

    def to_query_string(self, table: str = "candles") -> str:
        """Build a filtered query string suitable for InfluxDB or QuestDB."""
        parts = [f"FROM {table}"]

        conditions = []
        if self.symbol:
            conditions.append(f"symbol = '{self.symbol}'")
        if self.timeframe:
            conditions.append(f"timeframe = '{self.timeframe}'")
        if self.start_time:
            conditions.append(f"time >= '{self.start_time.isoformat()}'")
        if self.end_time:
            conditions.append(f"time < '{self.end_time.isoformat()}'")
        if self.min_volume:
            conditions.append(f"volume >= {self.min_volume}")
        for k, v in self.tags.items():
            conditions.append(f"{k} = '{v}'")

        if conditions:
            parts.append(f"WHERE {' AND '.join(conditions)}")

        return " ".join(parts)


class TimeSeriesEngine:
    """Abstraction over time-series databases for trading data queries."""

    def __init__(self, connection_url: str):
        self._url = connection_url

    def query_candles(self, filters: TSQueryFilter) -> list[dict]:
        """Execute a candle data query with filtering and aggregation.

        Args:
            filters: Declarative filter specifying symbol, time range, etc.

        Returns:
            List of candle dicts matching the query.
        """
        query_str = filters.to_query_string()
        logger.info("Executing TS query: %s", query_str)

        # Placeholder — actual implementation depends on backend (InfluxDB, QuestDB, TimescaleDB)
        # Example SQL for TimescaleDB:
        # SELECT time_bucket('1h', time) AS interval,
        #        first(open, time) AS open, max(high), min(low), last(close), sum(volume)
        # FROM candles WHERE symbol = 'BTC/USDT' AND time >= '2024-01-01'
        # GROUP BY interval ORDER BY interval

        return [
            {
                "interval": filters.start_time.isoformat(),
                "open": 67500.0,
                "high": 68200.0,
                "low": 67100.0,
                "close": 67900.0,
                "volume": 1523.45,
            }
        ]

    def aggregate_metrics(
        self,
        start: datetime,
        end: datetime,
        interval: str = "1h",
    ) -> dict:
        """Compute summary statistics over a time range.

        Returns volume-weighted average price, total volume, and trade count.
        """
        return {
            "start": start.isoformat(),
            "end": end.isoformat(),
            "interval": interval,
            "vwap": 67854.32,
            "total_volume": 145_678.90,
            "trade_count": 892_341,
            "num_candles": 168,  # 7 days * 24 hours
        }
```

## 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.

- [InfluxDB Documentation](https://docs.influxdata.com/influxdb/v2/)
- [Time Series Database Comparison](https://www.influxdata.com/time-series-database-comparison/)
- [TSDB Best Practices for Financial Data](https://docs.influxdata.com/influxdb/v2/write-data/)
- [Prometheus Metrics Storage](https://prometheus.io/docs/introduction/storage/)
- [Time-Series Data Modeling Patterns](https://docs.timescale.com/latest/about/concepts-time-series/)