#!/usr/bin/env python3 """ Data Collection Module for Crypto Data Aggregator Collects price data, news, and sentiment from various sources """ import requests import aiohttp import asyncio import json import logging import time import threading from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Tuple import re # Try to import optional dependencies try: import feedparser FEEDPARSER_AVAILABLE = True except ImportError: FEEDPARSER_AVAILABLE = False logging.warning("feedparser not installed. RSS feed parsing will be limited.") try: from bs4 import BeautifulSoup BS4_AVAILABLE = True except ImportError: BS4_AVAILABLE = False logging.warning("beautifulsoup4 not installed. HTML parsing will be limited.") # Import local modules import config import database # Setup logging using config settings logging.basicConfig( level=getattr(logging, config.LOG_LEVEL), format=config.LOG_FORMAT, handlers=[ logging.FileHandler(config.LOG_FILE), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Get database instance db = database.get_database() # Collection state tracking _collection_timers = [] _is_collecting = False # ==================== AI MODEL STUB FUNCTIONS ==================== # These provide fallback functionality when ai_models.py is not available def analyze_sentiment(text: str) -> Dict[str, Any]: """ Simple sentiment analysis based on keyword matching Returns sentiment score and label Args: text: Text to analyze Returns: Dict with 'score' and 'label' """ if not text: return {'score': 0.0, 'label': 'neutral'} text_lower = text.lower() # Positive keywords positive_words = [ 'bullish', 'moon', 'rally', 'surge', 'gain', 'profit', 'up', 'green', 'buy', 'long', 'growth', 'rise', 'pump', 'ATH', 'breakthrough', 'adoption', 'positive', 'optimistic', 'upgrade', 'partnership' ] # Negative keywords negative_words = [ 'bearish', 'crash', 'dump', 'drop', 'loss', 'down', 'red', 'sell', 'short', 'decline', 'fall', 'fear', 'scam', 'hack', 'vulnerability', 'negative', 'pessimistic', 'concern', 'warning', 'risk' ] # Count occurrences positive_count = sum(1 for word in positive_words if word in text_lower) negative_count = sum(1 for word in negative_words if word in text_lower) # Calculate score (-1 to 1) total = positive_count + negative_count if total == 0: score = 0.0 label = 'neutral' else: score = (positive_count - negative_count) / total # Determine label if score <= -0.6: label = 'very_negative' elif score <= -0.2: label = 'negative' elif score <= 0.2: label = 'neutral' elif score <= 0.6: label = 'positive' else: label = 'very_positive' return {'score': score, 'label': label} def summarize_text(text: str, max_length: int = 150) -> str: """ Simple text summarization - takes first sentences up to max_length Args: text: Text to summarize max_length: Maximum length of summary Returns: Summarized text """ if not text: return "" # Remove extra whitespace text = ' '.join(text.split()) # If already short enough, return as is if len(text) <= max_length: return text # Try to break at sentence boundary sentences = re.split(r'[.!?]+', text) summary = "" for sentence in sentences: sentence = sentence.strip() if not sentence: continue if len(summary) + len(sentence) + 2 <= max_length: summary += sentence + ". " else: break # If no complete sentences fit, truncate if not summary: summary = text[:max_length-3] + "..." return summary.strip() # Try to import AI models if available try: import ai_models # Override stub functions with real AI models if available analyze_sentiment = ai_models.analyze_sentiment summarize_text = ai_models.summarize_text logger.info("Using AI models for sentiment analysis and summarization") except ImportError: logger.info("AI models not available, using simple keyword-based analysis") # ==================== HELPER FUNCTIONS ==================== def safe_api_call(url: str, timeout: int = 10, headers: Optional[Dict] = None) -> Optional[Dict]: """ Make HTTP GET request with error handling and retry logic Args: url: URL to fetch timeout: Request timeout in seconds headers: Optional request headers Returns: Response JSON or None on failure """ if headers is None: headers = {'User-Agent': config.USER_AGENT} for attempt in range(config.MAX_RETRIES): try: logger.debug(f"API call attempt {attempt + 1}/{config.MAX_RETRIES}: {url}") response = requests.get(url, timeout=timeout, headers=headers) response.raise_for_status() return response.json() except requests.exceptions.HTTPError as e: logger.warning(f"HTTP error on attempt {attempt + 1}: {e}") if response.status_code == 429: # Rate limit wait_time = (attempt + 1) * 5 logger.info(f"Rate limited, waiting {wait_time}s...") time.sleep(wait_time) elif response.status_code >= 500: # Server error time.sleep(attempt + 1) else: break # Don't retry on 4xx errors except requests.exceptions.Timeout: logger.warning(f"Timeout on attempt {attempt + 1}") time.sleep(attempt + 1) except requests.exceptions.RequestException as e: logger.warning(f"Request error on attempt {attempt + 1}: {e}") time.sleep(attempt + 1) except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") break except Exception as e: logger.error(f"Unexpected error on attempt {attempt + 1}: {e}") break logger.error(f"All retry attempts failed for {url}") return None def extract_mentioned_coins(text: str) -> List[str]: """ Extract cryptocurrency symbols/names mentioned in text Args: text: Text to search for coin mentions Returns: List of coin symbols mentioned """ if not text: return [] text_upper = text.upper() mentioned = [] # Check for common symbols common_symbols = { 'BTC': 'bitcoin', 'ETH': 'ethereum', 'BNB': 'binancecoin', 'XRP': 'ripple', 'ADA': 'cardano', 'SOL': 'solana', 'DOT': 'polkadot', 'DOGE': 'dogecoin', 'AVAX': 'avalanche-2', 'MATIC': 'polygon', 'LINK': 'chainlink', 'UNI': 'uniswap', 'LTC': 'litecoin', 'ATOM': 'cosmos', 'ALGO': 'algorand' } # Check coin symbols for symbol, coin_id in common_symbols.items(): # Look for symbol as whole word or with $ prefix pattern = r'\b' + symbol + r'\b|\$' + symbol + r'\b' if re.search(pattern, text_upper): mentioned.append(symbol) # Check for full coin names (case insensitive) coin_names = { 'bitcoin': 'BTC', 'ethereum': 'ETH', 'binance': 'BNB', 'ripple': 'XRP', 'cardano': 'ADA', 'solana': 'SOL', 'polkadot': 'DOT', 'dogecoin': 'DOGE' } text_lower = text.lower() for name, symbol in coin_names.items(): if name in text_lower and symbol not in mentioned: mentioned.append(symbol) return list(set(mentioned)) # Remove duplicates # ==================== PRICE DATA COLLECTION ==================== def collect_price_data() -> Tuple[bool, int]: """ Fetch price data from CoinGecko API, fallback to CoinCap if needed Returns: Tuple of (success: bool, count: int) """ logger.info("Starting price data collection...") try: # Try CoinGecko first url = f"{config.COINGECKO_BASE_URL}{config.COINGECKO_ENDPOINTS['coins_markets']}" params = { 'vs_currency': 'usd', 'order': 'market_cap_desc', 'per_page': config.TOP_COINS_LIMIT, 'page': 1, 'sparkline': 'false', 'price_change_percentage': '1h,24h,7d' } # Add params to URL param_str = '&'.join([f"{k}={v}" for k, v in params.items()]) full_url = f"{url}?{param_str}" data = safe_api_call(full_url, timeout=config.REQUEST_TIMEOUT) if data is None: logger.warning("CoinGecko API failed, trying CoinCap backup...") return collect_price_data_coincap() # Parse and validate data prices = [] for item in data: try: price = item.get('current_price', 0) # Validate price if not config.MIN_PRICE <= price <= config.MAX_PRICE: logger.warning(f"Invalid price for {item.get('symbol')}: {price}") continue price_data = { 'symbol': item.get('symbol', '').upper(), 'name': item.get('name', ''), 'price_usd': price, 'volume_24h': item.get('total_volume', 0), 'market_cap': item.get('market_cap', 0), 'percent_change_1h': item.get('price_change_percentage_1h_in_currency'), 'percent_change_24h': item.get('price_change_percentage_24h'), 'percent_change_7d': item.get('price_change_percentage_7d'), 'rank': item.get('market_cap_rank', 999) } # Validate market cap and volume if price_data['market_cap'] and price_data['market_cap'] < config.MIN_MARKET_CAP: continue if price_data['volume_24h'] and price_data['volume_24h'] < config.MIN_VOLUME: continue prices.append(price_data) except Exception as e: logger.error(f"Error parsing price data item: {e}") continue # Save to database if prices: count = db.save_prices_batch(prices) logger.info(f"Successfully collected and saved {count} price records from CoinGecko") return True, count else: logger.warning("No valid price data to save") return False, 0 except Exception as e: logger.error(f"Error in collect_price_data: {e}") return False, 0 def collect_price_data_coincap() -> Tuple[bool, int]: """ Backup function using CoinCap API Returns: Tuple of (success: bool, count: int) """ logger.info("Starting CoinCap price data collection...") try: url = f"{config.COINCAP_BASE_URL}{config.COINCAP_ENDPOINTS['assets']}" params = { 'limit': config.TOP_COINS_LIMIT } param_str = '&'.join([f"{k}={v}" for k, v in params.items()]) full_url = f"{url}?{param_str}" response = safe_api_call(full_url, timeout=config.REQUEST_TIMEOUT) if response is None or 'data' not in response: logger.error("CoinCap API failed") return False, 0 data = response['data'] # Parse and validate data prices = [] for idx, item in enumerate(data): try: price = float(item.get('priceUsd', 0)) # Validate price if not config.MIN_PRICE <= price <= config.MAX_PRICE: logger.warning(f"Invalid price for {item.get('symbol')}: {price}") continue price_data = { 'symbol': item.get('symbol', '').upper(), 'name': item.get('name', ''), 'price_usd': price, 'volume_24h': float(item.get('volumeUsd24Hr', 0)) if item.get('volumeUsd24Hr') else None, 'market_cap': float(item.get('marketCapUsd', 0)) if item.get('marketCapUsd') else None, 'percent_change_1h': None, # CoinCap doesn't provide 1h change 'percent_change_24h': float(item.get('changePercent24Hr', 0)) if item.get('changePercent24Hr') else None, 'percent_change_7d': None, # CoinCap doesn't provide 7d change 'rank': int(item.get('rank', idx + 1)) } # Validate market cap and volume if price_data['market_cap'] and price_data['market_cap'] < config.MIN_MARKET_CAP: continue if price_data['volume_24h'] and price_data['volume_24h'] < config.MIN_VOLUME: continue prices.append(price_data) except Exception as e: logger.error(f"Error parsing CoinCap data item: {e}") continue # Save to database if prices: count = db.save_prices_batch(prices) logger.info(f"Successfully collected and saved {count} price records from CoinCap") return True, count else: logger.warning("No valid price data to save from CoinCap") return False, 0 except Exception as e: logger.error(f"Error in collect_price_data_coincap: {e}") return False, 0 # ==================== NEWS DATA COLLECTION ==================== def collect_news_data() -> int: """ Parse RSS feeds and Reddit posts, analyze sentiment and save to database Returns: Count of articles collected """ logger.info("Starting news data collection...") articles_collected = 0 # Collect from RSS feeds if FEEDPARSER_AVAILABLE: articles_collected += _collect_rss_feeds() else: logger.warning("Feedparser not available, skipping RSS feeds") # Collect from Reddit articles_collected += _collect_reddit_posts() logger.info(f"News collection completed. Total articles: {articles_collected}") return articles_collected def _collect_rss_feeds() -> int: """Collect articles from RSS feeds""" count = 0 for source_name, feed_url in config.RSS_FEEDS.items(): try: logger.debug(f"Parsing RSS feed: {source_name}") feed = feedparser.parse(feed_url) for entry in feed.entries[:20]: # Limit to 20 most recent per feed try: # Extract article data title = entry.get('title', '') url = entry.get('link', '') # Skip if no URL if not url: continue # Get published date published_date = None if hasattr(entry, 'published_parsed') and entry.published_parsed: try: published_date = datetime(*entry.published_parsed[:6]).isoformat() except: pass # Get summary/description summary = entry.get('summary', '') or entry.get('description', '') if summary and BS4_AVAILABLE: # Strip HTML tags soup = BeautifulSoup(summary, 'html.parser') summary = soup.get_text() # Combine title and summary for analysis full_text = f"{title} {summary}" # Extract mentioned coins related_coins = extract_mentioned_coins(full_text) # Analyze sentiment sentiment_result = analyze_sentiment(full_text) # Summarize text summary_text = summarize_text(summary or title, max_length=200) # Prepare news data news_data = { 'title': title, 'summary': summary_text, 'url': url, 'source': source_name, 'sentiment_score': sentiment_result['score'], 'sentiment_label': sentiment_result['label'], 'related_coins': related_coins, 'published_date': published_date } # Save to database if db.save_news(news_data): count += 1 except Exception as e: logger.error(f"Error processing RSS entry from {source_name}: {e}") continue except Exception as e: logger.error(f"Error parsing RSS feed {source_name}: {e}") continue logger.info(f"Collected {count} articles from RSS feeds") return count def _collect_reddit_posts() -> int: """Collect posts from Reddit""" count = 0 for subreddit_name, endpoint_url in config.REDDIT_ENDPOINTS.items(): try: logger.debug(f"Fetching Reddit posts from r/{subreddit_name}") # Reddit API requires .json extension if not endpoint_url.endswith('.json'): endpoint_url = endpoint_url.rstrip('/') + '.json' headers = {'User-Agent': config.USER_AGENT} data = safe_api_call(endpoint_url, headers=headers) if not data or 'data' not in data or 'children' not in data['data']: logger.warning(f"Invalid response from Reddit: {subreddit_name}") continue posts = data['data']['children'] for post_data in posts[:15]: # Limit to 15 posts per subreddit try: post = post_data.get('data', {}) # Extract post data title = post.get('title', '') url = post.get('url', '') permalink = f"https://reddit.com{post.get('permalink', '')}" selftext = post.get('selftext', '') # Skip if no title if not title: continue # Use permalink as primary URL (actual Reddit post) article_url = permalink # Get timestamp created_utc = post.get('created_utc') published_date = None if created_utc: try: published_date = datetime.fromtimestamp(created_utc).isoformat() except: pass # Combine title and text for analysis full_text = f"{title} {selftext}" # Extract mentioned coins related_coins = extract_mentioned_coins(full_text) # Analyze sentiment sentiment_result = analyze_sentiment(full_text) # Summarize text summary_text = summarize_text(selftext or title, max_length=200) # Prepare news data news_data = { 'title': title, 'summary': summary_text, 'url': article_url, 'source': f"reddit_{subreddit_name}", 'sentiment_score': sentiment_result['score'], 'sentiment_label': sentiment_result['label'], 'related_coins': related_coins, 'published_date': published_date } # Save to database if db.save_news(news_data): count += 1 except Exception as e: logger.error(f"Error processing Reddit post from {subreddit_name}: {e}") continue except Exception as e: logger.error(f"Error fetching Reddit posts from {subreddit_name}: {e}") continue logger.info(f"Collected {count} posts from Reddit") return count # ==================== SENTIMENT DATA COLLECTION ==================== def collect_sentiment_data() -> Optional[Dict[str, Any]]: """ Fetch Fear & Greed Index from Alternative.me Returns: Sentiment data or None on failure """ logger.info("Starting sentiment data collection...") try: # Fetch Fear & Greed Index data = safe_api_call(config.ALTERNATIVE_ME_URL, timeout=config.REQUEST_TIMEOUT) if data is None or 'data' not in data: logger.error("Failed to fetch Fear & Greed Index") return None # Parse response fng_data = data['data'][0] if data['data'] else {} value = fng_data.get('value') classification = fng_data.get('value_classification', 'Unknown') timestamp = fng_data.get('timestamp') if value is None: logger.warning("No value in Fear & Greed response") return None # Convert to sentiment score (-1 to 1) # Fear & Greed is 0-100, convert to -1 to 1 sentiment_score = (int(value) - 50) / 50.0 # Determine label if int(value) <= 25: sentiment_label = 'extreme_fear' elif int(value) <= 45: sentiment_label = 'fear' elif int(value) <= 55: sentiment_label = 'neutral' elif int(value) <= 75: sentiment_label = 'greed' else: sentiment_label = 'extreme_greed' sentiment_data = { 'value': int(value), 'classification': classification, 'sentiment_score': sentiment_score, 'sentiment_label': sentiment_label, 'timestamp': timestamp } # Save to news table as market-wide sentiment news_data = { 'title': f"Market Sentiment: {classification}", 'summary': f"Fear & Greed Index: {value}/100 - {classification}", 'url': config.ALTERNATIVE_ME_URL, 'source': 'alternative_me', 'sentiment_score': sentiment_score, 'sentiment_label': sentiment_label, 'related_coins': ['BTC', 'ETH'], # Market-wide 'published_date': datetime.now().isoformat() } db.save_news(news_data) logger.info(f"Sentiment collected: {classification} ({value}/100)") return sentiment_data except Exception as e: logger.error(f"Error in collect_sentiment_data: {e}") return None # ==================== SCHEDULING ==================== def schedule_data_collection(): """ Schedule periodic data collection using threading.Timer Runs collection tasks in background at configured intervals """ global _is_collecting, _collection_timers if _is_collecting: logger.warning("Data collection already running") return _is_collecting = True logger.info("Starting scheduled data collection...") def run_price_collection(): """Wrapper for price collection with rescheduling""" try: collect_price_data() except Exception as e: logger.error(f"Error in scheduled price collection: {e}") finally: # Reschedule if _is_collecting: timer = threading.Timer( config.COLLECTION_INTERVALS['price_data'], run_price_collection ) timer.daemon = True timer.start() _collection_timers.append(timer) def run_news_collection(): """Wrapper for news collection with rescheduling""" try: collect_news_data() except Exception as e: logger.error(f"Error in scheduled news collection: {e}") finally: # Reschedule if _is_collecting: timer = threading.Timer( config.COLLECTION_INTERVALS['news_data'], run_news_collection ) timer.daemon = True timer.start() _collection_timers.append(timer) def run_sentiment_collection(): """Wrapper for sentiment collection with rescheduling""" try: collect_sentiment_data() except Exception as e: logger.error(f"Error in scheduled sentiment collection: {e}") finally: # Reschedule if _is_collecting: timer = threading.Timer( config.COLLECTION_INTERVALS['sentiment_data'], run_sentiment_collection ) timer.daemon = True timer.start() _collection_timers.append(timer) # Initial run immediately logger.info("Running initial data collection...") # Run initial collections in separate threads threading.Thread(target=run_price_collection, daemon=True).start() time.sleep(2) # Stagger starts threading.Thread(target=run_news_collection, daemon=True).start() time.sleep(2) threading.Thread(target=run_sentiment_collection, daemon=True).start() logger.info("Scheduled data collection started successfully") logger.info(f"Price data: every {config.COLLECTION_INTERVALS['price_data']}s") logger.info(f"News data: every {config.COLLECTION_INTERVALS['news_data']}s") logger.info(f"Sentiment data: every {config.COLLECTION_INTERVALS['sentiment_data']}s") def stop_scheduled_collection(): """Stop all scheduled collection tasks""" global _is_collecting, _collection_timers logger.info("Stopping scheduled data collection...") _is_collecting = False # Cancel all timers for timer in _collection_timers: try: timer.cancel() except: pass _collection_timers.clear() logger.info("Scheduled data collection stopped") # ==================== ASYNC COLLECTION (BONUS) ==================== async def collect_price_data_async() -> Tuple[bool, int]: """ Async version of price data collection using aiohttp Returns: Tuple of (success: bool, count: int) """ logger.info("Starting async price data collection...") try: url = f"{config.COINGECKO_BASE_URL}{config.COINGECKO_ENDPOINTS['coins_markets']}" params = { 'vs_currency': 'usd', 'order': 'market_cap_desc', 'per_page': config.TOP_COINS_LIMIT, 'page': 1, 'sparkline': 'false', 'price_change_percentage': '1h,24h,7d' } async with aiohttp.ClientSession() as session: async with session.get(url, params=params, timeout=config.REQUEST_TIMEOUT) as response: if response.status != 200: logger.error(f"API returned status {response.status}") return False, 0 data = await response.json() # Parse and validate data (same as sync version) prices = [] for item in data: try: price = item.get('current_price', 0) if not config.MIN_PRICE <= price <= config.MAX_PRICE: continue price_data = { 'symbol': item.get('symbol', '').upper(), 'name': item.get('name', ''), 'price_usd': price, 'volume_24h': item.get('total_volume', 0), 'market_cap': item.get('market_cap', 0), 'percent_change_1h': item.get('price_change_percentage_1h_in_currency'), 'percent_change_24h': item.get('price_change_percentage_24h'), 'percent_change_7d': item.get('price_change_percentage_7d'), 'rank': item.get('market_cap_rank', 999) } if price_data['market_cap'] and price_data['market_cap'] < config.MIN_MARKET_CAP: continue if price_data['volume_24h'] and price_data['volume_24h'] < config.MIN_VOLUME: continue prices.append(price_data) except Exception as e: logger.error(f"Error parsing price data item: {e}") continue # Save to database if prices: count = db.save_prices_batch(prices) logger.info(f"Async collected and saved {count} price records") return True, count else: return False, 0 except Exception as e: logger.error(f"Error in collect_price_data_async: {e}") return False, 0 # ==================== MAIN ENTRY POINT ==================== if __name__ == "__main__": logger.info("=" * 60) logger.info("Crypto Data Collector - Manual Test Run") logger.info("=" * 60) # Test price collection logger.info("\n--- Testing Price Collection ---") success, count = collect_price_data() print(f"Price collection: {'SUCCESS' if success else 'FAILED'} - {count} records") # Test news collection logger.info("\n--- Testing News Collection ---") news_count = collect_news_data() print(f"News collection: {news_count} articles collected") # Test sentiment collection logger.info("\n--- Testing Sentiment Collection ---") sentiment = collect_sentiment_data() if sentiment: print(f"Sentiment: {sentiment['classification']} ({sentiment['value']}/100)") else: print("Sentiment collection: FAILED") logger.info("\n" + "=" * 60) logger.info("Manual test run completed") logger.info("=" * 60)