Exchange Market Data Cache

---




name: exchange-market-data-cache
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"High-performance caching layer for market data with low latency and
  high" throughput'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: ai-order-flow-analysis, data-alternative-data
  role: implementation
  scope: implementation
  triggers: caching, exchange market data cache, exchange-market-data-cache, high-performance,
    layer, optimization, elasticache, performance
  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:** Cache market data efficiently to reduce API calls and improve data access speed

**Philosophy:** Market data changes constantly; caching must balance freshness with performance while ensuring data integrity

## Key Principles

1. **Time-Based TTL**: Data expires based on market activity frequency
2. **Cache Invalidation**: Use WebSocket updates to maintain cache consistency
3. **Hierarchical Caching**: L1 (in-memory) and L2 (Redis) layers
4. **Delta Updates**: Only transmit changed data to reduce bandwidth
5. **Read-Through Pattern**: Cache handles both reads and refreshes transparently

## Implementation Guidelines

### Structure
- Core logic: data_cache/market_data_cache.py
- L1 cache: data_cache/l1_cache.py
- L2 cache: data_cache/l2_cache.py
- Tests: tests/test_market_cache.py

### Patterns to Follow
- UseLRU eviction for memory-constrained caches
- Implement batch operations for efficiency
- Support both synchronous and asynchronous access
- Track cache hit/miss rates

## Adherence Checklist
Before completing your task, verify:
- [ ] Cache hit rate is monitored and alerting configured
- [ ] TTL values are configurable per data type
- [ ] Cache invalidation propagates to all layers
- [ ] Batch operations reduce round trips
- [ ] Stale data is detected and handled


Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.

## Python Implementation

```python
import time
import threading
from typing import Dict, List, Optional, Any, Callable
from dataclasses import dataclass, field
from collections import OrderedDict
from datetime import datetime, timedelta
import logging

@dataclass
class CacheEntry:
    """Represents a cached market data entry with metadata."""
    value: Any
    created_at: float
    expires_at: float
    hit_count: int = 0
    last_accessed: float = field(default_factory=time.time)
    
    @property
    def is_expired(self) -> bool:
        return time.time() > self.expires_at

class L1Cache:
    """In-memory L1 cache with LRU eviction."""
    
    def __init__(self, max_size: int = 10000, default_ttl: float = 60.0):
        self.max_size = max_size
        self.default_ttl = default_ttl
        self._cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._lock = threading.RLock()
        self.hits = 0
        self.misses = 0
    
    def get(self, key: str) -> Optional[Any]:
        """Get value from cache, returning None if not found or expired."""
        with self._lock:
            if key not in self._cache:
                self.misses += 1
                return None
            
            entry = self._cache[key]
            
            if entry.is_expired:
                self._remove(key)
                self.misses += 1
                return None
            
            entry.hit_count += 1
            entry.last_accessed = time.time()
            self.hits += 1
            
            # Move to end for LRU (most recently used)
            self._cache.move_to_end(key)
            return entry.value
    
    def set(self, key: str, value: Any, ttl: Optional[float] = None):
        """Set value in cache with optional TTL."""
        ttl = ttl or self.default_ttl
        expires_at = time.time() + ttl
        
        with self._lock:
            if key in self._cache:
                self._cache[key] = CacheEntry(value, time.time(), expires_at)
                self._cache.move_to_end(key)
            else:
                # Evict if at capacity
                while len(self._cache) >= self.max_size:
                    self._cache.popitem(last=False)
                
                self._cache[key] = CacheEntry(value, time.time(), expires_at)
    
    def delete(self, key: str):
        """Remove entry from cache."""
        with self._lock:
            self._remove(key)
    
    def _remove(self, key: str):
        """Remove key from cache (must hold lock)."""
        if key in self._cache:
            del self._cache[key]
    
    def invalidate_pattern(self, pattern: str):
        """Invalidate all keys matching pattern."""
        with self._lock:
            keys_to_remove = [k for k in self._cache if pattern in k]
            for key in keys_to_remove:
                self._remove(key)
    
    def stats(self) -> Dict[str, Any]:
        """Return cache statistics."""
        with self._lock:
            total = self.hits + self.misses
            hit_rate = self.hits / total if total > 0 else 0
            return {
                "size": len(self._cache),
                "hits": self.hits,
                "misses": self.misses,
                "hit_rate": hit_rate
            }

class MarketDataCache:
    """Market data cache with hierarchical storage layers."""
    
    def __init__(
        self,
        l1_max_size: int = 10000,
        l1_ttl: float = 60.0,
        l2_client: Optional[Any] = None
    ):
        self.l1 = L1Cache(max_size=l1_max_size, default_ttl=l1_ttl)
        self.l2 = l2_client
        self.snapshot_cache: OrderedDict[str, CacheEntry] = OrderedDict()
        self._lock = threading.RLock()
    
    def get_ticker(self, symbol: str) -> Optional[Dict]:
        """Get latest ticker data for symbol."""
        key = f"ticker:{symbol}"
        value = self.l1.get(key)
        
        if value is None and self.l2:
            value = self.l2.get(key)
            if value:
                self.l1.set(key, value)
        
        return value
    
    def set_ticker(self, symbol: str, data: Dict, ttl: float = 5.0):
        """Set ticker data for symbol (short TTL for real-time updates)."""
        key = f"ticker:{symbol}"
        self.l1.set(key, data, ttl=ttl)
    
    def get_orderbook(self, symbol: str) -> Optional[Dict]:
        """Get full order book for symbol."""
        key = f"orderbook:{symbol}"
        value = self.l1.get(key)
        
        if value is None and self.l2:
            value = self.l2.get(key)
            if value:
                self.l1.set(key, value)
        
        return value
    
    def set_orderbook(self, symbol: str, book: Dict, ttl: float = 2.0):
        """Set order book data (very short TTL for real-time)."""
        key = f"orderbook:{symbol}"
        self.l1.set(key, book, ttl=ttl)
    
    def get_candles(self, symbol: str, timeframe: str, count: int = 100) -> List[Dict]:
        """Get candle data with fallback to L2."""
        key = f"candles:{symbol}:{timeframe}:{count}"
        
        # Try L1
        cached = self.l1.get(key)
        if cached:
            return cached
        
        # Try L2
        if self.l2:
            cached = self.l2.get(key)
            if cached:
                self.l1.set(key, cached, ttl=300)  # Longer TTL for historical
                return cached
        
        return []
    
    def set_candles(self, symbol: str, timeframe: str, candles: List[Dict]):
        """Store candle data."""
        key = f"candles:{symbol}:{timeframe}:{len(candles)}"
        self.l1.set(key, candles, ttl=300)  # Historical data persists longer
    
    def invalidate_symbol(self, symbol: str):
        """Invalidate all cache entries for a symbol."""
        self.l1.invalidate_pattern(f"ticker:{symbol}")
        self.l1.invalidate_pattern(f"orderbook:{symbol}")
        self.l1.invalidate_pattern(f"candles:{symbol}")

class DeltaCache:
    """Optimized cache that tracks and transmits only changes."""
    
    def __init__(self, max_items: int = 1000):
        self._cache: Dict[str, Any] = {}
        self._versions: Dict[str, int] = {}
        self._changes: Dict[str, Any] = {}
        self.max_items = max_items
    
    def update(self, key: str, value: Any) -> Optional[Dict[str, Any]]:
        """Update value and return delta if changed."""
        current_version = self._versions.get(key, 0)
        self._cache[key] = value
        self._versions[key] = current_version + 1
        
        # Track change for batch transmission
        self._changes[key] = {
            "key": key,
            "value": value,
            "version": self._versions[key],
            "timestamp": time.time()
        }
        
        # Return all accumulated changes
        if len(self._changes) >= 100:  # Batch size
            changes = self._changes.copy()
            self._changes.clear()
            return changes
        
        return None
    
    def get_delta(self, since_version: int = 0) -> Dict[str, Any]:
        """Get all changes since given version."""
        return {
            k: {"value": v, "version": self._versions[k]}
            for k, v in self._cache.items()
            if self._versions.get(k, 0) > since_version
        }
```

---

---



### Pattern 2: Risk-Managed Trading Logic with Validation

```python
from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TradeSignal:
    """Immutable trade signal with all required validation constraints."""
    symbol: str
    side: str  # "buy" or "sell"
    price: float
    quantity: float
    confidence: float  # 0.0 to 1.0
    reason: str

    def validate(self) -> bool:
        """Validate that the trade signal meets all business constraints."""
        if self.quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self.quantity}")
        if self.price <= 0:
            raise ValueError(f"Price must be positive, got {self.price}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
        return True


def generate_trade_signal(
    symbol: str,
    side: str,
    price: float,
    quantity: float,
    confidence: float,
    reason: str,
) -> TradeSignal:
    """Generate a validated trade signal with guard clause checks."""
    if side not in ("buy", "sell"):
        raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")

    signal = TradeSignal(
        symbol=symbol,
        side=side,
        price=price,
        quantity=quantity,
        confidence=confidence,
        reason=reason,
    )
    signal.validate()
    logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
                 symbol, side, quantity, price, confidence)
    return signal


def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
    """Execute a trade signal after applying risk management checks."""
    adjusted_quantity = signal.quantity
    if signal.side == "buy" and signal.quantity > max_position_pct:
        logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
                        signal.symbol, max_position_pct * 100, max_position_pct)
        adjusted_quantity = max_position_pct

    return {
        "symbol": signal.symbol,
        "side": signal.side,
        "price": signal.price,
        "quantity": adjusted_quantity,
        "capped": adjusted_quantity < signal.quantity,
        "confidence": signal.confidence,
        "status": "submitted",
    }
```

## Constraints

### MUST DO
- Implement a unified adapter interface across all exchange integrations to standardize order placement, cancellation, and querying
- Handle rate limiting proactively with token bucket or leaky bucket algorithms — never wait for 429 responses before slowing down
- Maintain local order state as the source of truth; reconcile with exchange state periodically via webhook events and polling
- Implement heartbeat monitoring per exchange connection with automatic failover to a secondary data feed on timeout
- Log all API interactions including request/response IDs, timing, and status codes for audit and debugging

### MUST NOT DO
- Do not trust exchange-reported order states without local confirmation — always reconcile after every state change
- Avoid sending multiple orders for the same position simultaneously across different adapters or sessions
- Never store API keys or secrets in code — use environment variables or a secrets manager with automatic rotation
- Do not assume all exchanges support the same order types — implement graceful degradation with clear capability negotiation
- Avoid polling-based price updates when WebSocket/streaming APIs are available — polling creates unnecessary load and latency


## 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 Cache Metrics](https://docs.influxdata.com/influxdb/v2/monitor/metrics/system/#cache-stats)
- [Redis Caching for Market Data](https://redis.io/docs/latest/operate/rc/during-deployment/caching-strategies/)
- [Market Data Latency Optimization](https://docs.quantconnect.com/tutorials/data-sources-and-format)
- [Caching Strategies for Real-Time Trading](https://www.investopedia.com/terms/c/cache.asp)
- [Time-Series Cache Architecture](https://docs.timescale.com/latest/about/how-timescaledb-works/hypertables/)