crypto-mcp/cache/__init__.py

"""TTL-based in-memory cache."""

import threading
import time
from typing import Any, Optional

from config import PRICE_TTL, OHLCV_TTL


class TTLCache:
    def __init__(self):
        self._store: dict[str, tuple[Any, float]] = {}
        self._lock = threading.Lock()

    def get(self, key: str) -> Optional[Any]:
        with self._lock:
            entry = self._store.get(key)
            if entry is None:
                return None
            value, expires_at = entry
            if time.time() > expires_at:
                del self._store[key]
                return None
            return value

    def set(self, key: str, value: Any, ttl: int) -> None:
        with self._lock:
            self._store[key] = (value, time.time() + ttl)

    def delete(self, key: str) -> None:
        with self._lock:
            self._store.pop(key, None)

    def stats(self) -> dict:
        with self._lock:
            now = time.time()
            alive = sum(1 for _, (_, exp) in self._store.items() if exp > now)
            return {"total_keys": len(self._store), "alive_keys": alive}


_cache = TTLCache()


def cache_key_price(symbol: str) -> str:
    return f"price:{symbol.upper()}"


def cache_key_ohlcv(symbol: str, timeframe: str) -> str:
    return f"ohlcv:{symbol.upper()}:{timeframe}"


def get_cached_price(symbol: str) -> Optional[dict]:
    return _cache.get(cache_key_price(symbol))


def set_cached_price(symbol: str, data: dict) -> None:
    _cache.set(cache_key_price(symbol), data, ttl=PRICE_TTL)


def get_cached_ohlcv(symbol: str, timeframe: str) -> Optional[dict]:
    return _cache.get(cache_key_ohlcv(symbol, timeframe))


def set_cached_ohlcv(symbol: str, timeframe: str, data: dict) -> None:
    _cache.set(cache_key_ohlcv(symbol, timeframe), data, ttl=OHLCV_TTL.get(timeframe, 60))


def get_cache_stats() -> dict:
    return _cache.stats()