#!/usr/bin/env python3 """ Redis Cache Service for QuantEdge AI Replaces MySQL NBACacheEntry with high-performance Redis caching """ import redis import json import hashlib import logging from typing import Any, Optional, Dict from datetime import datetime, timedelta logger = logging.getLogger(__name__) class CacheService: """Redis-based caching service""" def __init__(self, host: str = 'localhost', port: int = 6379, db: int = 0, password: Optional[str] = None): try: self.redis = redis.Redis( host=host, port=port, db=db, password=password, decode_responses=True, socket_timeout=5, socket_connect_timeout=5 ) # Test connection self.redis.ping() logger.info("✅ Redis cache service connected") self.enabled = True except redis.ConnectionError: logger.warning("❌ Redis not available, caching disabled") self.enabled = False self.redis = None def _make_key(self, prefix: str, data: Any) -> str: """Create a consistent cache key""" if isinstance(data, dict): # Sort keys for consistent hashing data_str = json.dumps(data, sort_keys=True) else: data_str = str(data) # Create hash for key hash_obj = hashlib.md5(data_str.encode()) return f"{prefix}:{hash_obj.hexdigest()[:16]}" def get(self, key: str) -> Optional[Any]: """Get value from cache""" if not self.enabled or not self.redis: return None try: data = self.redis.get(key) if data: return json.loads(data) except Exception as e: logger.error(f"Cache get error: {e}") return None def set(self, key: str, value: Any, ttl_seconds: int = 3600) -> bool: """Set value in cache with TTL""" if not self.enabled or not self.redis: return False try: data = json.dumps(value) return self.redis.setex(key, ttl_seconds, data) except Exception as e: logger.error(f"Cache set error: {e}") return False def delete(self, key: str) -> bool: """Delete key from cache""" if not self.enabled or not self.redis: return False try: return bool(self.redis.delete(key)) except Exception as e: logger.error(f"Cache delete error: {e}") return False def clear_pattern(self, pattern: str) -> int: """Clear all keys matching pattern""" if not self.enabled or not self.redis: return 0 try: keys = self.redis.keys(pattern) if keys: return self.redis.delete(*keys) except Exception as e: logger.error(f"Cache clear pattern error: {e}") return 0 # NBA-specific cache methods def get_nba_games(self, season: str = "2023-24", limit: int = 100) -> Optional[list]: """Get cached NBA games data""" key = self._make_key("nba_games", {"season": season, "limit": limit}) return self.get(key) def set_nba_games(self, games: list, season: str = "2023-24", limit: int = 100, ttl_hours: int = 24) -> bool: """Cache NBA games data""" key = self._make_key("nba_games", {"season": season, "limit": limit}) return self.set(key, games, ttl_hours * 3600) def get_nba_players(self, limit: int = 100) -> Optional[list]: """Get cached NBA players data""" key = self._make_key("nba_players", {"limit": limit}) return self.get(key) def set_nba_players(self, players: list, limit: int = 100, ttl_hours: int = 24) -> bool: """Cache NBA players data""" key = self._make_key("nba_players", {"limit": limit}) return self.set(key, players, ttl_hours * 3600) def get_nba_teams(self) -> Optional[list]: """Get cached NBA teams data""" key = "nba_teams" return self.get(key) def set_nba_teams(self, teams: list, ttl_hours: int = 168) -> bool: """Cache NBA teams data (longer TTL since teams change infrequently)""" key = "nba_teams" return self.set(key, teams, ttl_hours * 3600) # Backtest result caching def get_backtest_result(self, strategy_name: str, market: str, params: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Get cached backtest result""" key = self._make_key("backtest", { "strategy": strategy_name, "market": market, "params": params }) return self.get(key) def set_backtest_result(self, strategy_name: str, market: str, params: Dict[str, Any], result: Dict[str, Any], ttl_minutes: int = 60) -> bool: """Cache backtest result""" key = self._make_key("backtest", { "strategy": strategy_name, "market": market, "params": params }) return self.set(key, result, ttl_minutes * 60) # Strategy caching def get_strategy_signals(self, strategy_code: str, market: str, params: Dict[str, Any]) -> Optional[Dict[str, Any]]: """Get cached strategy signals""" key = self._make_key("strategy_signals", { "code_hash": hashlib.md5(strategy_code.encode()).hexdigest()[:16], "market": market, "params": params }) return self.get(key) def set_strategy_signals(self, strategy_code: str, market: str, params: Dict[str, Any], signals: Dict[str, Any], ttl_minutes: int = 30) -> bool: """Cache strategy signals""" key = self._make_key("strategy_signals", { "code_hash": hashlib.md5(strategy_code.encode()).hexdigest()[:16], "market": market, "params": params }) return self.set(key, signals, ttl_minutes * 60) # Cache statistics def get_stats(self) -> Dict[str, Any]: """Get cache statistics""" if not self.enabled or not self.redis: return {"enabled": False} try: info = self.redis.info() keys = self.redis.keys("*") return { "enabled": True, "total_keys": len(keys), "memory_used": info.get('used_memory_human', 'unknown'), "connected_clients": info.get('connected_clients', 0), "uptime_days": info.get('uptime_in_days', 0) } except Exception as e: logger.error(f"Cache stats error: {e}") return {"enabled": False, "error": str(e)} def clear_nba_cache(self) -> int: """Clear all NBA-related cache""" patterns = ["nba_games:*", "nba_players:*", "nba_teams"] total_deleted = 0 for pattern in patterns: total_deleted += self.clear_pattern(pattern) return total_deleted def clear_backtest_cache(self) -> int: """Clear all backtest-related cache""" return self.clear_pattern("backtest:*") def clear_all_cache(self) -> int: """Clear all cache data""" if not self.enabled or not self.redis: return 0 try: return self.redis.flushdb() except Exception as e: logger.error(f"Clear all cache error: {e}") return 0 # Global cache instance cache_service = CacheService() # Export for use in main.py __all__ = ['CacheService', 'cache_service']