The Caching Dilemma
Financial data caching is a balancing act. Cache too aggressively and your bot trades on stale data. Cache too little and you burn through API rate limits. The key insight: different data types have different staleness tolerances.
Live quote TTL
H1 indicator TTL
H4/D1 indicator TTL
Redis Setup
import redis
import json
from typing import Optional
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)
def cache_get(key: str) -> Optional[dict]:
"""Get cached data, returns None if expired or missing."""
data = r.get(key)
if data:
return json.loads(data)
return None
def cache_set(key: str, data: dict, ttl_seconds: int):
"""Store data with a TTL."""
r.setex(key, ttl_seconds, json.dumps(data))TTL Strategy by Data Type
The instinct is to tie the TTL to the candle length — cache an H1 value for an hour. That is the wrong clock. What actually decides how long a cached value can still be correct is how often the data is republished. A cached value only goes stale when a newer one exists upstream, so the TTL should track the publish interval, then add a margin so the entry survives until the next push lands.
That is how TickAtlas sets its own numbers. The price feeder publishes M1 and M5 every 60 seconds, M15/M30/H1 every 600 seconds, and H4 and above every 1800 seconds. Each server-side TTL is roughly 1.5x its publish interval: 180s for M1/M5, 900s for M15/M30/H1, 2700s for H4/D1, and 7200s for W1/MN1. Live quotes use 120s (twice the 60s base TTL). When a snapshot is detected as stale, it is re-cached for only 30s instead of the full TTL, so bad data cannot be masked for the whole window.
Mirror those numbers client-side and your cache expires at roughly the moment new data becomes available — maximum hit rate, no extra staleness. Pick anything longer and you are trading on data you could already have replaced.
# TTL configuration based on data type and timeframe
# TTL tracks the PUBLISH interval of the timeframe, not the candle length.
# These are the server-side values, so a client cache that mirrors them
# expires at about the moment fresher data becomes available.
TTL_CONFIG = {
# Indicators and OHLCV share the same clock: both come from the same push.
"indicator": {
"M1": 180, # published every 60s
"M5": 180, # published every 60s
"M15": 900, # published every 600s
"M30": 900, # published every 600s
"H1": 900, # published every 600s
"H4": 2700, # published every 1800s
"D1": 2700, # published every 1800s
"W1": 7200,
"MN1": 7200,
},
# Live quotes tick continuously but are cached at twice the 60s base TTL.
"quote": 120,
# Anything the API does not publish on a fixed clock is your own call.
"symbols": 86400, # 24 hours -- the instrument list rarely changes
}
# When you detect a stale upstream snapshot, re-cache it briefly instead of
# for the full TTL, so one bad read cannot be masked for the whole window.
STALE_FALLBACK_TTL = 30
def get_ttl(data_type: str, timeframe: str = None) -> int:
ttl = TTL_CONFIG.get(data_type)
if isinstance(ttl, dict):
return ttl.get(timeframe, 120) # Default to the base quote TTL
return ttl or 120Cached API Client
import requests
API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"
class CachedClient:
"""API client with Redis caching layer."""
def __init__(self, api_key: str):
self.headers = {"X-API-Key": api_key}
def get_indicator(self, symbol: str, indicator: str, timeframe: str) -> dict:
cache_key = f"ind:{symbol}:{indicator}:{timeframe}"
cached = cache_get(cache_key)
if cached:
return cached
resp = requests.get(
f"{BASE_URL}/indicator",
headers=self.headers,
params={"symbol": symbol, "indicator": indicator, "timeframe": timeframe},
)
resp.raise_for_status()
data = resp.json()["data"]
ttl = get_ttl("indicator", timeframe)
cache_set(cache_key, data, ttl)
return data
def get_ohlcv(self, symbol: str, timeframe: str, limit: int = 50) -> list:
cache_key = f"ohlcv:{symbol}:{timeframe}:{limit}"
cached = cache_get(cache_key)
if cached:
return cached
resp = requests.get(
f"{BASE_URL}/ohlc",
headers=self.headers,
params={"symbol": symbol, "timeframe": timeframe, "limit": limit},
)
data = resp.json()["data"]["candles"]
ttl = get_ttl("indicator", timeframe)
cache_set(cache_key, data, ttl)
return data
def get_spread(self, symbol: str) -> dict:
cache_key = f"spread:{symbol}"
cached = cache_get(cache_key)
if cached:
return cached
resp = requests.get(
f"{BASE_URL}/spread",
headers=self.headers,
params={"symbol": symbol},
)
data = resp.json()["data"]
cache_set(cache_key, data, get_ttl("quote"))
return data
# Usage -- automatic caching
client = CachedClient(API_KEY)
rsi = client.get_indicator("EURUSD", "RSI_14", "H4") # API call
rsi = client.get_indicator("EURUSD", "RSI_14", "H4") # Cache hitCache Warming
For critical symbols, pre-fetch data before your strategy loop runs:
WATCHLIST = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD", "BTCUSD"]
INDICATORS = ["RSI_14", "MACD_hist", "EMA_50", "ATR_14"]
def warm_cache(client: CachedClient, timeframe: str):
"""Pre-fetch all indicators for the watchlist."""
for symbol in WATCHLIST:
for indicator in INDICATORS:
try:
client.get_indicator(symbol, indicator, timeframe)
except Exception as e:
print(f"Failed to warm {symbol}/{indicator}: {e}")
# Warm cache before the strategy loop starts
warm_cache(client, "H4")Monitoring Cache Performance
class CacheStats:
def __init__(self):
self.hits = 0
self.misses = 0
def record_hit(self):
self.hits += 1
def record_miss(self):
self.misses += 1
@property
def hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0
def report(self):
print(f"Cache hits: {self.hits}, misses: {self.misses}")
print(f"Hit rate: {self.hit_rate:.1%}")
print(f"API calls saved: {self.hits}")Run this against live data.
Every account starts pay-as-you-go with $2.50 of credit and no card. Paste the key into the samples above and the requests work unchanged.