# Exchange Market Data Cache > "High-performance caching layer for market data with low latency and high" throughput - Skill: `paulpas/exchange-market-data-cache` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/exchange-market-data-cache` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/exchange-market-data-cache/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - 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/exchange-market-data-cache --- **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/)