'"High-performance caching layer for market data with low latency and
Scanned 6/12/2026
Install via CLI
Skills Directoryopenskills install paulpas/agent-skill-router---
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/)
No comments yet. Be the first to comment!
