TickAtlas
Developer 10 min read · March 28, 2026

Caching Financial Data: Redis Patterns for Trading Applications

Learn smart caching strategies for financial data using Redis. Reduce API costs, improve latency, and maintain data freshness with TTL-based patterns.

CG
By the TickAtlas team

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.

2s

Real-time price TTL

5m

Indicator values TTL

6h

Daily OHLCV 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

python 
# TTL configuration based on data type and timeframe
TTL_CONFIG = {
    # Indicator TTL matches the timeframe
    "indicator": {
        "M1": 60,       # 1 minute
        "M5": 300,      # 5 minutes
        "M15": 600,     # 10 minutes (half the candle)
        "M30": 900,     # 15 minutes
        "H1": 1800,     # 30 minutes
        "H4": 3600,     # 1 hour
        "D1": 21600,    # 6 hours
    },
    # OHLCV data -- cache longer since historical bars don't change
    "ohlcv": {
        "M1": 120,
        "M5": 600,
        "M15": 900,
        "H1": 3600,
        "H4": 7200,
        "D1": 43200,    # 12 hours
    },
    # Spread data -- very short TTL, always want fresh data
    "spread": 5,         # 5 seconds
    # Symbol list -- rarely changes
    "symbols": 86400,    # 24 hours
}

def get_ttl(data_type: str, timeframe: str = None) -> int:
    ttl = TTL_CONFIG.get(data_type)
    if isinstance(ttl, dict):
        return ttl.get(timeframe, 300)  # Default 5 min
    return ttl or 300

Cached API Client

python 
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}/ohlcv",
            headers=self.headers,
            params={"symbol": symbol, "timeframe": timeframe, "limit": limit},
        )
        data = resp.json()["data"]["candles"]

        ttl = get_ttl("ohlcv", 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("spread"))
        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 hit

Cache Warming

For critical symbols, pre-fetch data before your strategy loop runs:

python 
WATCHLIST = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD", "BTCUSD"]
INDICATORS = ["RSI_14", "MACD", "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

python 
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}")

Related Reading

REST API Best Practices

Retry logic and error handling

Data Pipeline Architecture

End-to-end data flow design

Try this with live data

Every account gets $2.50 in free PAYG credits. No card required — paste your API key and run the code above against live broker data.

Get API Key Read API docs

Keep reading

All articles