File size: 4,049 Bytes
c49b21b |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
"""
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}")
|