"""
Simple caching layer to reduce API calls and improve performance.
"""
from typing import Any, Optional
from datetime import datetime, timedelta
import json
import hashlib
class Cache:
"""In-memory cache with TTL (Time To Live) support."""
def __init__(self, default_ttl: int = 60):
"""
Initialize cache.
Args:
default_ttl: Default time-to-live in seconds
"""
self._cache = {}
self.default_ttl = default_ttl
def _generate_key(self, *args, **kwargs) -> str:
"""
Generate a unique cache key from arguments.
Args:
*args: Positional arguments
**kwargs: Keyword arguments
Returns:
Hash string to use as cache key
"""
# Create a string representation of all arguments
key_data = json.dumps({
'args': args,
'kwargs': sorted(kwargs.items())
}, sort_keys=True)
# Return hash of the data
return hashlib.md5(key_data.encode()).hexdigest()
def get(self, key: str) -> Optional[Any]:
"""
Get value from cache if it exists and hasn't expired.
Args:
key: Cache key
Returns:
Cached value or None if not found/expired
"""
if key in self._cache:
entry = self._cache[key]
# Check if entry has expired
if datetime.now() < entry['expires_at']:
return entry['value']
else:
# Remove expired entry
del self._cache[key]
return None
def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
"""
Set value in cache with TTL.
Args:
key: Cache key
value: Value to cache
ttl: Time-to-live in seconds (uses default if None)
"""
ttl = ttl if ttl is not None else self.default_ttl
expires_at = datetime.now() + timedelta(seconds=ttl)
self._cache[key] = {
'value': value,
'expires_at': expires_at,
'created_at': datetime.now()
}
def delete(self, key: str) -> bool:
"""
Delete a key from cache.
Args:
key: Cache key
Returns:
True if key existed, False otherwise
"""
if key in self._cache:
del self._cache[key]
return True
return False
def clear(self) -> None:
"""Clear all cached entries."""
self._cache.clear()
def cleanup_expired(self) -> int:
"""
Remove all expired entries from cache.
Returns:
Number of entries removed
"""
now = datetime.now()
expired_keys = [
key for key, entry in self._cache.items()
if now >= entry['expires_at']
]
for key in expired_keys:
del self._cache[key]
return len(expired_keys)
def get_stats(self) -> dict:
"""
Get cache statistics.
Returns:
Dictionary with cache statistics
"""
now = datetime.now()
valid_entries = sum(
1 for entry in self._cache.values()
if now < entry['expires_at']
)
return {
'total_entries': len(self._cache),
'valid_entries': valid_entries,
'expired_entries': len(self._cache) - valid_entries
}
class CachedCryptoAPI:
"""
Wrapper around CryptoAPI that adds caching functionality.
"""
def __init__(self, crypto_api, cache_ttl: int = 60):
"""
Initialize cached API wrapper.
Args:
crypto_api: CryptoAPI instance to wrap
cache_ttl: Default cache TTL in seconds
"""
self.api = crypto_api
self.cache = Cache(default_ttl=cache_ttl)
def get_current_price(self, symbol: str, use_cache: bool = True) -> dict:
"""
Get current price with caching.
Args:
symbol: Trading pair
use_cache: Whether to use cache
Returns:
Price data dictionary
"""
cache_key = f"price:{symbol}"
if use_cache:
cached = self.cache.get(cache_key)
if cached is not None:
cached['from_cache'] = True
return cached
# Fetch from API
data = self.api.get_current_price(symbol)
data['from_cache'] = False
# Cache the result (30 seconds for real-time prices)
self.cache.set(cache_key, data, ttl=30)
return data
def get_historical_ohlcv(
self,
symbol: str,
timeframe: str = '1d',
limit: int = 100,
use_cache: bool = True
) -> list:
"""
Get historical data with caching.
Args:
symbol: Trading pair
timeframe: Timeframe
limit: Number of data points
use_cache: Whether to use cache
Returns:
List of OHLCV data
"""
cache_key = f"ohlcv:{symbol}:{timeframe}:{limit}"
if use_cache:
cached = self.cache.get(cache_key)
if cached is not None:
return cached
# Fetch from API
data = self.api.get_historical_ohlcv(symbol, timeframe, limit)
# Cache for longer (5 minutes for historical data)
self.cache.set(cache_key, data, ttl=300)
return data
def get_market_summary(self, symbol: str, use_cache: bool = True) -> dict:
"""
Get market summary with caching.
Args:
symbol: Trading pair
use_cache: Whether to use cache
Returns:
Market summary dictionary
"""
cache_key = f"summary:{symbol}"
if use_cache:
cached = self.cache.get(cache_key)
if cached is not None:
cached['from_cache'] = True
return cached
# Fetch from API
data = self.api.get_market_summary(symbol)
data['from_cache'] = False
# Cache for 1 minute
self.cache.set(cache_key, data, ttl=60)
return data
def clear_cache(self) -> None:
"""Clear all cached data."""
self.cache.clear()
def get_cache_stats(self) -> dict:
"""Get cache statistics."""
return self.cache.get_stats()
