""" External Sources Manager - Handles custom API-based signal sources. Allows users to define external data sources (like CoinMarketCap Fear & Greed Index) that can be used as signal sources alongside technical indicators. """ import json import logging import uuid import datetime as dt from dataclasses import dataclass, asdict from typing import Any, Dict, List, Optional import requests from jsonpath_ng import parse as jsonpath_parse from DataCache_v3 import DataCache logger = logging.getLogger(__name__) @dataclass class ExternalSource: """Represents an external API data source.""" name: str url: str auth_header: str = "" # e.g., "X-CMC_PRO_API_KEY" auth_key: str = "" # The actual API key (stored encrypted ideally) json_path: str = "$.data[0].value" # JSONPath to extract value refresh_interval: int = 300 # Seconds between fetches (default 5 min) last_value: float = None last_fetch_time: float = None # Unix timestamp creator_id: int = None enabled: bool = True tbl_key: str = None def to_dict(self) -> Dict[str, Any]: """Convert to dictionary for JSON serialization.""" return { 'name': self.name, 'url': self.url, 'auth_header': self.auth_header, 'json_path': self.json_path, 'refresh_interval': self.refresh_interval, 'last_value': self.last_value, 'last_fetch_time': self.last_fetch_time, 'creator_id': self.creator_id, 'enabled': self.enabled, 'tbl_key': self.tbl_key, # Note: auth_key intentionally excluded for security } class ExternalSources: """Manages external API data sources for signals.""" def __init__(self, data_cache: DataCache): """ Initialize the ExternalSources manager. :param data_cache: DataCache instance for database operations. """ self.data_cache = data_cache self._ensure_table_exists() # Create cache for external sources self.data_cache.create_cache( name='external_sources', cache_type='table', size_limit=100, eviction_policy='deny', default_expiration=dt.timedelta(hours=24), columns=[ "creator_id", "name", "url", "auth_header", "auth_key", "json_path", "refresh_interval", "last_value", "last_fetch_time", "enabled", "tbl_key" ] ) # In-memory cache of ExternalSource objects self.sources: Dict[str, ExternalSource] = {} # Load existing sources self._load_sources_from_db() def _ensure_table_exists(self) -> None: """Create the external_sources table if it doesn't exist.""" try: if not self.data_cache.db.table_exists('external_sources'): create_sql = """ CREATE TABLE IF NOT EXISTS external_sources ( id INTEGER PRIMARY KEY AUTOINCREMENT, creator_id INTEGER, name TEXT NOT NULL, url TEXT NOT NULL, auth_header TEXT, auth_key TEXT, json_path TEXT DEFAULT '$.data[0].value', refresh_interval INTEGER DEFAULT 300, last_value REAL, last_fetch_time REAL, enabled INTEGER DEFAULT 1, tbl_key TEXT UNIQUE ) """ self.data_cache.db.execute_sql(create_sql, params=[]) logger.info("Created external_sources table in database") except Exception as e: logger.error(f"Error creating external_sources table: {e}", exc_info=True) def _load_sources_from_db(self) -> None: """Load all external sources from database into memory.""" try: sources_df = self.data_cache.get_all_rows_from_datacache(cache_name='external_sources') if sources_df is not None and not sources_df.empty: for _, row in sources_df.iterrows(): source = ExternalSource( name=row.get('name', ''), url=row.get('url', ''), auth_header=row.get('auth_header', ''), auth_key=row.get('auth_key', ''), json_path=row.get('json_path', '$.data[0].value'), refresh_interval=int(row.get('refresh_interval', 300)), last_value=row.get('last_value'), last_fetch_time=row.get('last_fetch_time'), creator_id=row.get('creator_id'), enabled=bool(row.get('enabled', True)), tbl_key=row.get('tbl_key') ) if source.tbl_key: self.sources[source.tbl_key] = source logger.info(f"Loaded {len(self.sources)} external sources from database") except Exception as e: logger.error(f"Error loading external sources: {e}", exc_info=True) def create_source(self, name: str, url: str, auth_header: str, auth_key: str, json_path: str, refresh_interval: int, creator_id: int) -> Dict[str, Any]: """ Create a new external source. :param name: Display name for the source. :param url: API endpoint URL. :param auth_header: Header name for authentication (e.g., "X-CMC_PRO_API_KEY"). :param auth_key: API key value. :param json_path: JSONPath expression to extract value from response. :param refresh_interval: Seconds between data fetches. :param creator_id: User ID of the creator. :return: Dict with success status and source data. """ try: tbl_key = str(uuid.uuid4()) source = ExternalSource( name=name, url=url, auth_header=auth_header, auth_key=auth_key, json_path=json_path, refresh_interval=refresh_interval, creator_id=creator_id, enabled=True, tbl_key=tbl_key ) # Test the source by fetching data fetch_result = self._fetch_value(source) if not fetch_result['success']: return { 'success': False, 'message': f"Failed to fetch data: {fetch_result['error']}" } source.last_value = fetch_result['value'] source.last_fetch_time = dt.datetime.now().timestamp() # Save to database self._save_source_to_db(source) # Add to memory cache self.sources[tbl_key] = source logger.info(f"Created external source: {name} (value: {source.last_value})") return { 'success': True, 'source': source.to_dict(), 'message': f"Created source '{name}' with initial value: {source.last_value}" } except Exception as e: logger.error(f"Error creating external source: {e}", exc_info=True) return {'success': False, 'message': str(e)} def _save_source_to_db(self, source: ExternalSource) -> None: """Save an external source to the database (insert or update).""" # Check if source already exists existing = self.data_cache.db.execute_sql( "SELECT id FROM external_sources WHERE tbl_key = ?", params=(source.tbl_key,), fetch_one=True ) if existing: # Update existing record update_sql = """ UPDATE external_sources SET creator_id = ?, name = ?, url = ?, auth_header = ?, auth_key = ?, json_path = ?, refresh_interval = ?, last_value = ?, last_fetch_time = ?, enabled = ? WHERE tbl_key = ? """ self.data_cache.db.execute_sql(update_sql, params=( source.creator_id, source.name, source.url, source.auth_header, source.auth_key, source.json_path, source.refresh_interval, source.last_value, source.last_fetch_time, 1 if source.enabled else 0, source.tbl_key )) else: # Insert new record columns = ( 'creator_id', 'name', 'url', 'auth_header', 'auth_key', 'json_path', 'refresh_interval', 'last_value', 'last_fetch_time', 'enabled', 'tbl_key' ) values = ( source.creator_id, source.name, source.url, source.auth_header, source.auth_key, source.json_path, source.refresh_interval, source.last_value, source.last_fetch_time, 1 if source.enabled else 0, source.tbl_key ) self.data_cache.insert_row_into_datacache( cache_name='external_sources', columns=columns, values=values ) def _fetch_value(self, source: ExternalSource) -> Dict[str, Any]: """ Fetch the current value from an external API. :param source: The ExternalSource to fetch from. :return: Dict with success status, value, and optional error. """ try: headers = {'Accept': 'application/json'} # Add authentication header if provided if source.auth_header and source.auth_key: headers[source.auth_header] = source.auth_key response = requests.get(source.url, headers=headers, timeout=10) # Parse JSON first to check for API error messages try: data = response.json() except json.JSONDecodeError: return {'success': False, 'error': f'Invalid JSON response (HTTP {response.status_code})'} # Check for API-level errors in response if not response.ok: # Try to extract error message from common API error formats error_msg = None if isinstance(data, dict): # CoinMarketCap format: {"status": {"error_message": "..."}} if 'status' in data and isinstance(data['status'], dict): error_msg = data['status'].get('error_message') # Generic formats elif 'error' in data: error_msg = data.get('error') elif 'message' in data: error_msg = data.get('message') elif 'error_message' in data: error_msg = data.get('error_message') if error_msg: return {'success': False, 'error': f'API Error: {error_msg}'} else: return {'success': False, 'error': f'HTTP {response.status_code}: {response.reason}'} # Extract value using JSONPath jsonpath_expr = jsonpath_parse(source.json_path) matches = jsonpath_expr.find(data) if not matches: # Show a snippet of the response to help debug response_preview = str(data)[:200] return { 'success': False, 'error': f"JSONPath '{source.json_path}' found no matches. Response: {response_preview}..." } value = matches[0].value # Try to convert to float try: value = float(value) except (TypeError, ValueError): pass # Keep original value if not numeric return {'success': True, 'value': value} except requests.exceptions.Timeout: return {'success': False, 'error': 'Request timed out (10s)'} except requests.exceptions.ConnectionError: return {'success': False, 'error': 'Connection failed - check URL'} except requests.exceptions.RequestException as e: return {'success': False, 'error': f'Request failed: {str(e)}'} except Exception as e: logger.error(f"Error fetching external source: {e}", exc_info=True) return {'success': False, 'error': f'Error: {str(e)}'} def refresh_source(self, tbl_key: str) -> Dict[str, Any]: """ Refresh the value for a specific source. :param tbl_key: The unique key of the source to refresh. :return: Dict with success status and new value. """ source = self.sources.get(tbl_key) if not source: return {'success': False, 'error': 'Source not found'} result = self._fetch_value(source) if result['success']: source.last_value = result['value'] source.last_fetch_time = dt.datetime.now().timestamp() self._save_source_to_db(source) return {'success': True, 'value': result['value']} return result def refresh_all_sources(self) -> Dict[str, Any]: """ Refresh all sources that need updating based on their refresh interval. :return: Dict with results for each source. """ results = {} current_time = dt.datetime.now().timestamp() for tbl_key, source in self.sources.items(): if not source.enabled: continue # Check if refresh is needed if source.last_fetch_time: time_since_fetch = current_time - source.last_fetch_time if time_since_fetch < source.refresh_interval: continue result = self.refresh_source(tbl_key) results[source.name] = result return results def get_source(self, tbl_key: str) -> Optional[ExternalSource]: """Get a source by its tbl_key.""" return self.sources.get(tbl_key) def get_source_by_name(self, name: str) -> Optional[ExternalSource]: """Get a source by its name.""" for source in self.sources.values(): if source.name == name: return source return None def get_sources_for_user(self, user_id: int) -> List[Dict[str, Any]]: """ Get all sources created by a specific user. :param user_id: The user ID. :return: List of source dictionaries. """ return [ source.to_dict() for source in self.sources.values() if source.creator_id == user_id ] def get_all_sources(self) -> List[Dict[str, Any]]: """Get all sources as dictionaries.""" return [source.to_dict() for source in self.sources.values()] def delete_source(self, tbl_key: str, user_id: int) -> Dict[str, Any]: """ Delete an external source. :param tbl_key: The unique key of the source. :param user_id: The user requesting deletion (must be creator). :return: Dict with success status. """ source = self.sources.get(tbl_key) if not source: return {'success': False, 'message': 'Source not found'} if source.creator_id != user_id: return {'success': False, 'message': 'Not authorized to delete this source'} try: # Remove from database self.data_cache.db.execute_sql( "DELETE FROM external_sources WHERE tbl_key = ?", params=(tbl_key,) ) # Remove from memory del self.sources[tbl_key] logger.info(f"Deleted external source: {source.name}") return {'success': True, 'message': f"Deleted source '{source.name}'"} except Exception as e: logger.error(f"Error deleting external source: {e}", exc_info=True) return {'success': False, 'message': str(e)} def get_source_value(self, name: str) -> Optional[float]: """ Get the current value of a source by name. Used by the signals system. :param name: The source name. :return: The current value or None. """ source = self.get_source_by_name(name) if source: return source.last_value return None def get_sources_as_indicators(self, user_id: int) -> Dict[str, Dict[str, Any]]: """ Get sources formatted like indicators for use in signals. :param user_id: The user ID to get sources for. :return: Dict mapping source names to indicator-like data. """ result = {} for source in self.sources.values(): if source.creator_id == user_id and source.enabled: result[source.name] = { 'type': 'EXTERNAL', 'value': source.last_value, 'source_type': 'external', 'tbl_key': source.tbl_key, 'last_fetch_time': source.last_fetch_time } return result