#!/usr/bin/env python3 """ Database module for Crypto Data Aggregator Complete CRUD operations with the exact schema specified """ import sqlite3 import threading import json from datetime import datetime, timedelta from typing import List, Dict, Optional, Any, Tuple from contextlib import contextmanager import logging import config # Setup logging logging.basicConfig( level=getattr(logging, config.LOG_LEVEL), format=config.LOG_FORMAT, handlers=[ logging.FileHandler(config.LOG_FILE), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class CryptoDatabase: """ Database manager for cryptocurrency data with full CRUD operations Thread-safe implementation using context managers """ def __init__(self, db_path: str = None): """Initialize database with connection pooling""" self.db_path = str(db_path or config.DATABASE_PATH) self._local = threading.local() self._init_database() logger.info(f"Database initialized at {self.db_path}") @contextmanager def get_connection(self): """Get thread-safe database connection""" if not hasattr(self._local, 'conn'): self._local.conn = sqlite3.connect( self.db_path, check_same_thread=False, timeout=30.0 ) self._local.conn.row_factory = sqlite3.Row try: yield self._local.conn except Exception as e: self._local.conn.rollback() logger.error(f"Database error: {e}") raise def _init_database(self): """Initialize all database tables with exact schema""" with self.get_connection() as conn: cursor = conn.cursor() # ==================== PRICES TABLE ==================== cursor.execute(""" CREATE TABLE IF NOT EXISTS prices ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, name TEXT, price_usd REAL NOT NULL, volume_24h REAL, market_cap REAL, percent_change_1h REAL, percent_change_24h REAL, percent_change_7d REAL, rank INTEGER, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # ==================== NEWS TABLE ==================== cursor.execute(""" CREATE TABLE IF NOT EXISTS news ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, summary TEXT, url TEXT UNIQUE, source TEXT, sentiment_score REAL, sentiment_label TEXT, related_coins TEXT, published_date DATETIME, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # ==================== MARKET ANALYSIS TABLE ==================== cursor.execute(""" CREATE TABLE IF NOT EXISTS market_analysis ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, timeframe TEXT, trend TEXT, support_level REAL, resistance_level REAL, prediction TEXT, confidence REAL, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # ==================== USER QUERIES TABLE ==================== cursor.execute(""" CREATE TABLE IF NOT EXISTS user_queries ( id INTEGER PRIMARY KEY AUTOINCREMENT, query TEXT, result_count INTEGER, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP ) """) # ==================== CREATE INDEXES ==================== cursor.execute("CREATE INDEX IF NOT EXISTS idx_prices_symbol ON prices(symbol)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_prices_timestamp ON prices(timestamp)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_prices_rank ON prices(rank)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_news_url ON news(url)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_news_published ON news(published_date)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_news_sentiment ON news(sentiment_label)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_analysis_symbol ON market_analysis(symbol)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_analysis_timestamp ON market_analysis(timestamp)") conn.commit() logger.info("Database tables and indexes created successfully") # ==================== PRICES CRUD OPERATIONS ==================== def save_price(self, price_data: Dict[str, Any]) -> bool: """ Save a single price record Args: price_data: Dictionary containing price information Returns: bool: True if successful, False otherwise """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO prices (symbol, name, price_usd, volume_24h, market_cap, percent_change_1h, percent_change_24h, percent_change_7d, rank) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( price_data.get('symbol'), price_data.get('name'), price_data.get('price_usd', 0.0), price_data.get('volume_24h'), price_data.get('market_cap'), price_data.get('percent_change_1h'), price_data.get('percent_change_24h'), price_data.get('percent_change_7d'), price_data.get('rank') )) conn.commit() return True except Exception as e: logger.error(f"Error saving price: {e}") return False def save_prices_batch(self, prices: List[Dict[str, Any]]) -> int: """ Save multiple price records in batch (minimum 100 records for efficiency) Args: prices: List of price dictionaries Returns: int: Number of records saved """ saved_count = 0 try: with self.get_connection() as conn: cursor = conn.cursor() for price_data in prices: try: cursor.execute(""" INSERT INTO prices (symbol, name, price_usd, volume_24h, market_cap, percent_change_1h, percent_change_24h, percent_change_7d, rank) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( price_data.get('symbol'), price_data.get('name'), price_data.get('price_usd', 0.0), price_data.get('volume_24h'), price_data.get('market_cap'), price_data.get('percent_change_1h'), price_data.get('percent_change_24h'), price_data.get('percent_change_7d'), price_data.get('rank') )) saved_count += 1 except Exception as e: logger.warning(f"Error saving individual price: {e}") continue conn.commit() logger.info(f"Batch saved {saved_count} price records") except Exception as e: logger.error(f"Error in batch save: {e}") return saved_count def get_latest_prices(self, limit: int = 100) -> List[Dict[str, Any]]: """ Get latest prices for top cryptocurrencies Args: limit: Maximum number of records to return Returns: List of price dictionaries """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT DISTINCT ON (symbol) * FROM prices WHERE timestamp >= datetime('now', '-1 hour') ORDER BY symbol, timestamp DESC, rank ASC LIMIT ? """, (limit,)) # SQLite doesn't support DISTINCT ON, use subquery instead cursor.execute(""" SELECT p1.* FROM prices p1 INNER JOIN ( SELECT symbol, MAX(timestamp) as max_ts FROM prices WHERE timestamp >= datetime('now', '-1 hour') GROUP BY symbol ) p2 ON p1.symbol = p2.symbol AND p1.timestamp = p2.max_ts ORDER BY p1.rank ASC, p1.market_cap DESC LIMIT ? """, (limit,)) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error getting latest prices: {e}") return [] def get_price_history(self, symbol: str, hours: int = 24) -> List[Dict[str, Any]]: """ Get price history for a specific symbol Args: symbol: Cryptocurrency symbol hours: Number of hours to look back Returns: List of price dictionaries """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM prices WHERE symbol = ? AND timestamp >= datetime('now', '-' || ? || ' hours') ORDER BY timestamp ASC """, (symbol, hours)) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error getting price history: {e}") return [] def get_top_gainers(self, limit: int = 10) -> List[Dict[str, Any]]: """Get top gaining cryptocurrencies in last 24h""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT p1.* FROM prices p1 INNER JOIN ( SELECT symbol, MAX(timestamp) as max_ts FROM prices WHERE timestamp >= datetime('now', '-1 hour') GROUP BY symbol ) p2 ON p1.symbol = p2.symbol AND p1.timestamp = p2.max_ts WHERE p1.percent_change_24h IS NOT NULL ORDER BY p1.percent_change_24h DESC LIMIT ? """, (limit,)) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error getting top gainers: {e}") return [] def delete_old_prices(self, days: int = 30) -> int: """ Delete price records older than specified days Args: days: Number of days to keep Returns: Number of deleted records """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" DELETE FROM prices WHERE timestamp < datetime('now', '-' || ? || ' days') """, (days,)) conn.commit() deleted = cursor.rowcount logger.info(f"Deleted {deleted} old price records") return deleted except Exception as e: logger.error(f"Error deleting old prices: {e}") return 0 # ==================== NEWS CRUD OPERATIONS ==================== def save_news(self, news_data: Dict[str, Any]) -> bool: """ Save a single news record Args: news_data: Dictionary containing news information Returns: bool: True if successful, False otherwise """ try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT OR IGNORE INTO news (title, summary, url, source, sentiment_score, sentiment_label, related_coins, published_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( news_data.get('title'), news_data.get('summary'), news_data.get('url'), news_data.get('source'), news_data.get('sentiment_score'), news_data.get('sentiment_label'), json.dumps(news_data.get('related_coins', [])), news_data.get('published_date') )) conn.commit() return True except Exception as e: logger.error(f"Error saving news: {e}") return False def get_latest_news(self, limit: int = 50, sentiment: Optional[str] = None) -> List[Dict[str, Any]]: """ Get latest news articles Args: limit: Maximum number of articles sentiment: Filter by sentiment label (optional) Returns: List of news dictionaries """ try: with self.get_connection() as conn: cursor = conn.cursor() if sentiment: cursor.execute(""" SELECT * FROM news WHERE sentiment_label = ? ORDER BY published_date DESC, timestamp DESC LIMIT ? """, (sentiment, limit)) else: cursor.execute(""" SELECT * FROM news ORDER BY published_date DESC, timestamp DESC LIMIT ? """, (limit,)) results = [] for row in cursor.fetchall(): news_dict = dict(row) if news_dict.get('related_coins'): try: news_dict['related_coins'] = json.loads(news_dict['related_coins']) except: news_dict['related_coins'] = [] results.append(news_dict) return results except Exception as e: logger.error(f"Error getting latest news: {e}") return [] def get_news_by_coin(self, coin: str, limit: int = 20) -> List[Dict[str, Any]]: """Get news related to a specific coin""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM news WHERE related_coins LIKE ? ORDER BY published_date DESC LIMIT ? """, (f'%{coin}%', limit)) results = [] for row in cursor.fetchall(): news_dict = dict(row) if news_dict.get('related_coins'): try: news_dict['related_coins'] = json.loads(news_dict['related_coins']) except: news_dict['related_coins'] = [] results.append(news_dict) return results except Exception as e: logger.error(f"Error getting news by coin: {e}") return [] def update_news_sentiment(self, news_id: int, sentiment_score: float, sentiment_label: str) -> bool: """Update sentiment for a news article""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" UPDATE news SET sentiment_score = ?, sentiment_label = ? WHERE id = ? """, (sentiment_score, sentiment_label, news_id)) conn.commit() return True except Exception as e: logger.error(f"Error updating news sentiment: {e}") return False def delete_old_news(self, days: int = 30) -> int: """Delete news older than specified days""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" DELETE FROM news WHERE timestamp < datetime('now', '-' || ? || ' days') """, (days,)) conn.commit() deleted = cursor.rowcount logger.info(f"Deleted {deleted} old news records") return deleted except Exception as e: logger.error(f"Error deleting old news: {e}") return 0 # ==================== MARKET ANALYSIS CRUD OPERATIONS ==================== def save_analysis(self, analysis_data: Dict[str, Any]) -> bool: """Save market analysis""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO market_analysis (symbol, timeframe, trend, support_level, resistance_level, prediction, confidence) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( analysis_data.get('symbol'), analysis_data.get('timeframe'), analysis_data.get('trend'), analysis_data.get('support_level'), analysis_data.get('resistance_level'), analysis_data.get('prediction'), analysis_data.get('confidence') )) conn.commit() return True except Exception as e: logger.error(f"Error saving analysis: {e}") return False def get_latest_analysis(self, symbol: str) -> Optional[Dict[str, Any]]: """Get latest analysis for a symbol""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM market_analysis WHERE symbol = ? ORDER BY timestamp DESC LIMIT 1 """, (symbol,)) row = cursor.fetchone() return dict(row) if row else None except Exception as e: logger.error(f"Error getting latest analysis: {e}") return None def get_all_analyses(self, limit: int = 100) -> List[Dict[str, Any]]: """Get all market analyses""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM market_analysis ORDER BY timestamp DESC LIMIT ? """, (limit,)) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error getting all analyses: {e}") return [] # ==================== USER QUERIES CRUD OPERATIONS ==================== def log_user_query(self, query: str, result_count: int) -> bool: """Log a user query""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO user_queries (query, result_count) VALUES (?, ?) """, (query, result_count)) conn.commit() return True except Exception as e: logger.error(f"Error logging user query: {e}") return False def get_recent_queries(self, limit: int = 50) -> List[Dict[str, Any]]: """Get recent user queries""" try: with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM user_queries ORDER BY timestamp DESC LIMIT ? """, (limit,)) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error getting recent queries: {e}") return [] # ==================== UTILITY OPERATIONS ==================== def execute_safe_query(self, query: str, params: Tuple = ()) -> List[Dict[str, Any]]: """ Execute a safe read-only query Args: query: SQL query (must start with SELECT) params: Query parameters Returns: List of result dictionaries """ try: # Security: Only allow SELECT queries if not query.strip().upper().startswith('SELECT'): logger.warning(f"Attempted non-SELECT query: {query}") return [] with self.get_connection() as conn: cursor = conn.cursor() cursor.execute(query, params) return [dict(row) for row in cursor.fetchall()] except Exception as e: logger.error(f"Error executing safe query: {e}") return [] def get_database_stats(self) -> Dict[str, Any]: """Get database statistics""" try: with self.get_connection() as conn: cursor = conn.cursor() stats = {} # Count records in each table for table in ['prices', 'news', 'market_analysis', 'user_queries']: cursor.execute(f"SELECT COUNT(*) as count FROM {table}") stats[f'{table}_count'] = cursor.fetchone()['count'] # Get unique symbols cursor.execute("SELECT COUNT(DISTINCT symbol) as count FROM prices") stats['unique_symbols'] = cursor.fetchone()['count'] # Get latest price update cursor.execute("SELECT MAX(timestamp) as latest FROM prices") stats['latest_price_update'] = cursor.fetchone()['latest'] # Get latest news update cursor.execute("SELECT MAX(timestamp) as latest FROM news") stats['latest_news_update'] = cursor.fetchone()['latest'] # Database file size import os if os.path.exists(self.db_path): stats['database_size_bytes'] = os.path.getsize(self.db_path) stats['database_size_mb'] = stats['database_size_bytes'] / (1024 * 1024) return stats except Exception as e: logger.error(f"Error getting database stats: {e}") return {} def vacuum_database(self) -> bool: """Vacuum database to reclaim space""" try: with self.get_connection() as conn: conn.execute("VACUUM") logger.info("Database vacuumed successfully") return True except Exception as e: logger.error(f"Error vacuuming database: {e}") return False def backup_database(self, backup_path: Optional[str] = None) -> bool: """Create database backup""" try: import shutil if backup_path is None: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = config.DATABASE_BACKUP_DIR / f"backup_{timestamp}.db" shutil.copy2(self.db_path, backup_path) logger.info(f"Database backed up to {backup_path}") return True except Exception as e: logger.error(f"Error backing up database: {e}") return False def close(self): """Close database connection""" if hasattr(self._local, 'conn'): self._local.conn.close() delattr(self._local, 'conn') logger.info("Database connection closed") # Singleton instance _db_instance = None def get_database() -> CryptoDatabase: """Get database singleton instance""" global _db_instance if _db_instance is None: _db_instance = CryptoDatabase() return _db_instance