""" Santiment Frequency Controller ============================= This module provides frequency control for Santiment API calls to preserve API limits. It tracks execution frequency and limits runs to avoid exceeding API quotas. """ import json import os from datetime import datetime, timedelta from pathlib import Path class SantimentFrequencyController: """Controls the frequency of Santiment API calls to preserve API limits""" def __init__(self, state_file: str = None): """Initialize the frequency controller Args: state_file: Path to the state file. If None, uses default location. """ if state_file is None: # Try to find the state file in data/santiment directory try: from src.config import DATA_DIR state_file = os.path.join(DATA_DIR, "santiment", "frequency_state.json") except Exception: # Fallback to local directory state_file = "data/santiment/frequency_state.json" self.state_file = Path(state_file) self.state_file.parent.mkdir(parents=True, exist_ok=True) self._load_state() def _load_state(self): """Load the current state from file""" if self.state_file.exists(): try: with open(self.state_file, 'r') as f: self.state = json.load(f) except Exception: self.state = {} else: self.state = {} # Ensure required fields exist if 'last_run' not in self.state: self.state['last_run'] = None if 'runs_today' not in self.state: self.state['runs_today'] = 0 if 'date' not in self.state: self.state['date'] = None def _save_state(self): """Save the current state to file""" try: with open(self.state_file, 'w') as f: json.dump(self.state, f, indent=2) except Exception as e: print(f"[WARN] Failed to save frequency state: {e}") def should_run_santiment(self, max_runs_per_day: int = 2) -> bool: """Check if Santiment should be allowed to run Args: max_runs_per_day: Maximum number of runs allowed per day Returns: True if Santiment should run, False otherwise """ today = datetime.now().strftime("%Y-%m-%d") # Reset counter if it's a new day if self.state.get('date') != today: self.state['date'] = today self.state['runs_today'] = 0 self._save_state() # Check if we've exceeded the daily limit return self.state['runs_today'] < max_runs_per_day def record_run(self): """Record that Santiment has been run""" today = datetime.now().strftime("%Y-%m-%d") now = datetime.now().isoformat() # Update state self.state['last_run'] = now self.state['date'] = today self.state['runs_today'] = self.state.get('runs_today', 0) + 1 # Save state self._save_state() print(f"[SANTIMENT] Recorded run #{self.state['runs_today']} for {today}") def get_status(self) -> dict: """Get the current status of the frequency controller Returns: Dictionary with current status information """ return { 'last_run': self.state.get('last_run'), 'runs_today': self.state.get('runs_today', 0), 'date': self.state.get('date'), 'state_file': str(self.state_file) } def reset_daily_count(self): """Reset the daily run count (for testing or manual reset)""" today = datetime.now().strftime("%Y-%m-%d") self.state['date'] = today self.state['runs_today'] = 0 self._save_state() print(f"[SANTIMENT] Reset daily count for {today}")