Spaces:
Running
Running
| """ | |
| File-based cache for price data with collision-proof keys. | |
| Provides: | |
| - Parquet-based storage for efficiency | |
| - Collision-proof cache keys via fetch_params fingerprint | |
| - Cache invalidation and cleanup | |
| """ | |
| import hashlib | |
| import json | |
| import logging | |
| from pathlib import Path | |
| from datetime import datetime, timezone | |
| from typing import Optional | |
| import pandas as pd | |
| from screener.core.fingerprint import compute_file_fingerprint | |
| logger = logging.getLogger(__name__) | |
| class FetchParams: | |
| """ | |
| Immutable fetch parameters that define a unique cache entry. | |
| COLLISION-PROOF GUARANTEE: | |
| Any change to these parameters produces a different cache key. | |
| """ | |
| def __init__( | |
| self, | |
| symbol: str, | |
| start_date: str, | |
| end_date: str, | |
| interval: str = "1d", | |
| resample_rule: str = "W-FRI", | |
| price_field: str = "Close", | |
| auto_adjust: bool = True, | |
| actions: bool = True | |
| ): | |
| self.symbol = symbol | |
| self.start_date = start_date | |
| self.end_date = end_date | |
| self.interval = interval | |
| self.resample_rule = resample_rule | |
| self.price_field = price_field | |
| self.auto_adjust = auto_adjust | |
| self.actions = actions | |
| def to_dict(self) -> dict: | |
| """Convert to ordered dict for hashing.""" | |
| return { | |
| "symbol": self.symbol, | |
| "start_date": self.start_date, | |
| "end_date": self.end_date, | |
| "interval": self.interval, | |
| "resample_rule": self.resample_rule, | |
| "price_field": self.price_field, | |
| "auto_adjust": self.auto_adjust, | |
| "actions": self.actions | |
| } | |
| def fingerprint(self) -> str: | |
| """ | |
| Compute deterministic fingerprint of fetch params. | |
| Returns 8-char hex string (collision probability ~1 in 4 billion). | |
| """ | |
| data = json.dumps(self.to_dict(), sort_keys=True).encode("utf-8") | |
| return hashlib.sha256(data).hexdigest()[:8] | |
| def cache_key(self) -> str: | |
| """ | |
| Generate collision-proof cache key. | |
| Format: {safe_symbol}_{start}_{end}_{params_fingerprint} | |
| Example: HG_F_2018-01-01_2024-01-19_a3f2c891 | |
| """ | |
| safe_symbol = self.symbol.replace("=", "_").replace("^", "_").replace(".", "_") | |
| return f"{safe_symbol}_{self.start_date}_{self.end_date}_{self.fingerprint()}" | |
| class PriceCache: | |
| """ | |
| File-based cache for raw price data with collision-proof keys. | |
| Stores DataFrames as Parquet files. | |
| Tracks fingerprints for integrity verification. | |
| CACHE KEY CONTRACT (COLLISION-PROOF): | |
| Cache key = f"{symbol}_{start}_{end}_{params_fingerprint}" | |
| The params_fingerprint is a SHA256 hash of: | |
| - symbol, start_date, end_date | |
| - interval (e.g., "1d") | |
| - resample_rule (e.g., "W-FRI") | |
| - price_field (e.g., "Close") | |
| - auto_adjust, actions | |
| This GUARANTEES that any change to fetch parameters produces | |
| a different cache file, eliminating collision risk. | |
| """ | |
| def __init__(self, cache_dir: str | Path, namespace: Optional[str] = None): | |
| """ | |
| Initialize cache. | |
| Args: | |
| cache_dir: Base directory for cache files | |
| namespace: Optional namespace subdirectory (e.g., config_hash[:8]) | |
| """ | |
| self.base_dir = Path(cache_dir) | |
| # Use namespace subdirectory if provided | |
| if namespace: | |
| self.cache_dir = self.base_dir / namespace | |
| else: | |
| self.cache_dir = self.base_dir | |
| self.cache_dir.mkdir(parents=True, exist_ok=True) | |
| self.checksums_file = self.cache_dir / "checksums.json" | |
| self._checksums: dict[str, dict] = {} | |
| self._load_checksums() | |
| def _load_checksums(self): | |
| """Load existing checksums from file.""" | |
| if self.checksums_file.exists(): | |
| with open(self.checksums_file, "r") as f: | |
| self._checksums = json.load(f) | |
| def _save_checksums(self): | |
| """Save checksums to file.""" | |
| with open(self.checksums_file, "w") as f: | |
| json.dump(self._checksums, f, indent=2, sort_keys=True) | |
| def _make_path(self, cache_key: str) -> Path: | |
| """Get file path for cache key.""" | |
| return self.cache_dir / f"{cache_key}.parquet" | |
| def get(self, params: FetchParams) -> Optional[pd.DataFrame]: | |
| """ | |
| Get cached DataFrame for fetch params. | |
| Args: | |
| params: FetchParams object defining the cache entry | |
| Returns: | |
| DataFrame if cached and valid, None otherwise | |
| """ | |
| cache_key = params.cache_key() | |
| path = self._make_path(cache_key) | |
| # Debug log for cache lookup | |
| exists = path.exists() | |
| logger.info(f"Cache lookup: {params.symbol} -> {path.name} (exists={exists})") | |
| if not exists: | |
| return None | |
| try: | |
| df = pd.read_parquet(path) | |
| logger.info(f"Cache hit: {params.symbol} ({len(df)} rows)") | |
| return df | |
| except Exception as e: | |
| logger.warning(f"Cache read failed for {params.symbol}: {e}") | |
| return None | |
| def put(self, params: FetchParams, df: pd.DataFrame) -> str: | |
| """ | |
| Store DataFrame in cache. | |
| Args: | |
| params: FetchParams object defining the cache entry | |
| df: DataFrame to cache | |
| Returns: | |
| Fingerprint of stored file | |
| """ | |
| cache_key = params.cache_key() | |
| path = self._make_path(cache_key) | |
| df.to_parquet(path, index=True) | |
| fingerprint = compute_file_fingerprint(path) | |
| # Store metadata with checksum | |
| self._checksums[cache_key] = { | |
| "sha256": fingerprint, | |
| "params": params.to_dict(), | |
| "rows": len(df), | |
| "cached_at": datetime.now(timezone.utc).isoformat() | |
| } | |
| self._save_checksums() | |
| logger.debug(f"Cached: {params.symbol} ({len(df)} rows)") | |
| return fingerprint | |
| def has(self, params: FetchParams) -> bool: | |
| """Check if params are cached.""" | |
| cache_key = params.cache_key() | |
| return self._make_path(cache_key).exists() | |
| def get_metadata(self, params: FetchParams) -> Optional[dict]: | |
| """Get metadata for cached entry.""" | |
| cache_key = params.cache_key() | |
| return self._checksums.get(cache_key) | |
| def clear(self, older_than_days: Optional[int] = None): | |
| """ | |
| Clear cache files. | |
| Args: | |
| older_than_days: Only clear files older than N days (None = all) | |
| """ | |
| import os | |
| from datetime import timedelta | |
| cutoff = None | |
| if older_than_days is not None: | |
| cutoff = datetime.now(timezone.utc) - timedelta(days=older_than_days) | |
| cleared = 0 | |
| for path in self.cache_dir.glob("*.parquet"): | |
| should_clear = True | |
| if cutoff is not None: | |
| mtime = datetime.fromtimestamp( | |
| os.path.getmtime(path), | |
| tz=timezone.utc | |
| ) | |
| should_clear = mtime < cutoff | |
| if should_clear: | |
| path.unlink() | |
| cleared += 1 | |
| # Rebuild checksums from remaining files | |
| remaining_keys = set() | |
| for path in self.cache_dir.glob("*.parquet"): | |
| remaining_keys.add(path.stem) | |
| self._checksums = {k: v for k, v in self._checksums.items() if k in remaining_keys} | |
| self._save_checksums() | |
| logger.info(f"Cleared {cleared} cache files") | |
| # Backward compatibility wrapper | |
| def create_fetch_params( | |
| symbol: str, | |
| start_date: str, | |
| end_date: str, | |
| **kwargs | |
| ) -> FetchParams: | |
| """Helper to create FetchParams with defaults.""" | |
| return FetchParams( | |
| symbol=symbol, | |
| start_date=start_date, | |
| end_date=end_date, | |
| **kwargs | |
| ) | |