#!/usr/bin/env python3 """ Sports Data Cache Layer Persistent SQLite caching for SportsData.io responses Benefits: - Reduce API calls and costs - Faster response times - Work offline with cached data - 30-day TTL for historical data """ import sqlite3 import hashlib import json import os from datetime import datetime, timedelta from typing import Any, Optional, List, Dict class SportsDataCache: """ SQLite-based persistent cache for sports data. Reduces API calls and provides faster lookups. """ def __init__(self, db_path: str = './data/sports_cache.db'): """Initialize cache with SQLite database""" # Ensure data directory exists os.makedirs(os.path.dirname(db_path), exist_ok=True) self.db_path = db_path self.conn = sqlite3.connect(db_path, check_same_thread=False) self._init_db() def _init_db(self): """Create cache tables if they don't exist""" self.conn.executescript(''' CREATE TABLE IF NOT EXISTS sports_data ( cache_key TEXT PRIMARY KEY, sport TEXT NOT NULL, endpoint TEXT NOT NULL, season TEXT, data_json TEXT NOT NULL, record_count INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP, last_accessed TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_sport_endpoint ON sports_data(sport, endpoint); CREATE INDEX IF NOT EXISTS idx_expires ON sports_data(expires_at); CREATE TABLE IF NOT EXISTS backtest_results ( id TEXT PRIMARY KEY, user_id TEXT, sport TEXT NOT NULL, strategy_name TEXT NOT NULL, strategy_code TEXT, parameters_json TEXT, results_json TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, win_rate REAL, total_profit REAL, total_bets INTEGER ); CREATE INDEX IF NOT EXISTS idx_user_sport ON backtest_results(user_id, sport); ''') self.conn.commit() def _get_cache_key(self, sport: str, endpoint: str, season: str = None, **kwargs) -> str: """Generate consistent cache key from parameters""" key_parts = [sport.lower(), endpoint.lower()] if season: key_parts.append(str(season)) # Include any additional params for k, v in sorted(kwargs.items()): if v is not None: key_parts.append(f"{k}={v}") return hashlib.md5('_'.join(key_parts).encode()).hexdigest() def get(self, sport: str, endpoint: str, season: str = None, max_age_hours: int = 24 * 30, **kwargs) -> Optional[List[Dict]]: """ Get cached data if fresh enough. Args: sport: Sport code (nba, nfl, etc.) endpoint: API endpoint (games, teams, etc.) season: Season identifier max_age_hours: Maximum cache age (default 30 days) Returns: Cached data or None if not found/expired """ cache_key = self._get_cache_key(sport, endpoint, season, **kwargs) cursor = self.conn.execute(''' SELECT data_json, expires_at FROM sports_data WHERE cache_key = ? AND expires_at > ? ''', (cache_key, datetime.now().isoformat())) row = cursor.fetchone() if row: # Update last accessed time self.conn.execute(''' UPDATE sports_data SET last_accessed = ? WHERE cache_key = ? ''', (datetime.now().isoformat(), cache_key)) self.conn.commit() return json.loads(row[0]) return None def set(self, sport: str, endpoint: str, data: Any, season: str = None, ttl_hours: int = 24 * 30, **kwargs) -> None: """ Cache data with TTL. Args: sport: Sport code endpoint: API endpoint data: Data to cache (will be JSON serialized) season: Season identifier ttl_hours: Time to live in hours (default 30 days) """ cache_key = self._get_cache_key(sport, endpoint, season, **kwargs) expires_at = datetime.now() + timedelta(hours=ttl_hours) record_count = len(data) if isinstance(data, list) else 1 self.conn.execute(''' INSERT OR REPLACE INTO sports_data (cache_key, sport, endpoint, season, data_json, record_count, created_at, expires_at, last_accessed) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( cache_key, sport, endpoint, season, json.dumps(data), record_count, datetime.now().isoformat(), expires_at.isoformat(), datetime.now().isoformat() )) self.conn.commit() def save_backtest_result(self, result_id: str, user_id: str, sport: str, strategy_name: str, parameters: Dict, results: Dict, strategy_code: str = None) -> None: """Save backtest results for user history""" self.conn.execute(''' INSERT OR REPLACE INTO backtest_results (id, user_id, sport, strategy_name, strategy_code, parameters_json, results_json, created_at, win_rate, total_profit, total_bets) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( result_id, user_id, sport, strategy_name, strategy_code, json.dumps(parameters), json.dumps(results), datetime.now().isoformat(), results.get('win_rate'), results.get('total_profit'), results.get('total_bets') )) self.conn.commit() def get_user_history(self, user_id: str, limit: int = 20) -> List[Dict]: """Get user's backtest history""" cursor = self.conn.execute(''' SELECT id, sport, strategy_name, created_at, win_rate, total_profit, total_bets FROM backtest_results WHERE user_id = ? ORDER BY created_at DESC LIMIT ? ''', (user_id, limit)) return [ { 'id': row[0], 'sport': row[1], 'strategy_name': row[2], 'created_at': row[3], 'win_rate': row[4], 'total_profit': row[5], 'total_bets': row[6] } for row in cursor.fetchall() ] def cleanup_expired(self) -> int: """Remove expired cache entries. Returns count deleted.""" cursor = self.conn.execute(''' DELETE FROM sports_data WHERE expires_at < ? ''', (datetime.now().isoformat(),)) self.conn.commit() return cursor.rowcount def get_cache_stats(self) -> Dict: """Get cache statistics""" cursor = self.conn.execute(''' SELECT COUNT(*) as total_entries, SUM(record_count) as total_records, SUM(LENGTH(data_json)) as total_bytes, COUNT(DISTINCT sport) as sports_cached FROM sports_data WHERE expires_at > ? ''', (datetime.now().isoformat(),)) row = cursor.fetchone() return { 'total_entries': row[0] or 0, 'total_records': row[1] or 0, 'total_bytes': row[2] or 0, 'total_mb': round((row[2] or 0) / (1024 * 1024), 2), 'sports_cached': row[3] or 0 } def close(self): """Close database connection""" self.conn.close() # Singleton instance _cache_instance: Optional[SportsDataCache] = None def get_cache() -> SportsDataCache: """Get or create the singleton cache instance""" global _cache_instance if _cache_instance is None: _cache_instance = SportsDataCache() return _cache_instance # Test the cache if __name__ == '__main__': cache = get_cache() # Test set/get test_data = [{'game_id': '123', 'home_team': 'Lakers', 'away_team': 'Celtics'}] cache.set('nba', 'games', test_data, season='2024') retrieved = cache.get('nba', 'games', season='2024') print(f"Cache test: {'āœ… PASS' if retrieved == test_data else 'āŒ FAIL'}") # Stats stats = cache.get_cache_stats() print(f"Cache stats: {stats}") print("\nāœ… SportsDataCache is working!")