bot / hf-data-engine /ai_models.py
King-bay's picture
Deploy: Crypto Intelligence Hub with code review fixes
b5017fb
#!/usr/bin/env python3
"""
AI Models Module for Crypto Data Aggregator
HuggingFace local inference for sentiment analysis, summarization, and market trend analysis
NO API calls - all inference runs locally using transformers library
"""
import logging
from typing import Dict, List, Optional, Any
from functools import lru_cache
import warnings
# Suppress HuggingFace warnings
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", category=UserWarning)
try:
import torch
from transformers import (
pipeline,
AutoModelForSequenceClassification,
AutoTokenizer,
)
TRANSFORMERS_AVAILABLE = True
except ImportError:
TRANSFORMERS_AVAILABLE = False
logging.warning("transformers library not available. AI features will be disabled.")
import config
# ==================== LOGGING SETUP ====================
logging.basicConfig(
level=getattr(logging, config.LOG_LEVEL),
format=config.LOG_FORMAT,
handlers=[
logging.FileHandler(config.LOG_FILE),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# ==================== GLOBAL MODEL STORAGE ====================
# Lazy loading - models loaded only when first called
_models_initialized = False
_sentiment_twitter_pipeline = None
_sentiment_financial_pipeline = None
_summarization_pipeline = None
# Model loading lock to prevent concurrent initialization
_models_loading = False
# ==================== MODEL INITIALIZATION ====================
def initialize_models() -> Dict[str, Any]:
"""
Initialize all HuggingFace models for local inference.
Loads sentiment and summarization models using pipeline().
Returns:
Dict with status, success flag, and loaded models info
"""
global _models_initialized, _sentiment_twitter_pipeline
global _sentiment_financial_pipeline, _summarization_pipeline, _models_loading
if _models_initialized:
logger.info("Models already initialized")
return {
"success": True,
"status": "Models already loaded",
"models": {
"sentiment_twitter": _sentiment_twitter_pipeline is not None,
"sentiment_financial": _sentiment_financial_pipeline is not None,
"summarization": _summarization_pipeline is not None,
}
}
if _models_loading:
logger.warning("Models are currently being loaded by another process")
return {"success": False, "status": "Models loading in progress", "models": {}}
if not TRANSFORMERS_AVAILABLE:
logger.error("transformers library not available. Cannot initialize models.")
return {
"success": False,
"status": "transformers library not installed",
"models": {},
"error": "Install transformers: pip install transformers torch"
}
_models_loading = True
loaded_models = {}
errors = []
try:
logger.info("Starting model initialization...")
# Load Twitter sentiment model
try:
logger.info(f"Loading sentiment_twitter model: {config.HUGGINGFACE_MODELS['sentiment_twitter']}")
_sentiment_twitter_pipeline = pipeline(
"sentiment-analysis",
model=config.HUGGINGFACE_MODELS["sentiment_twitter"],
tokenizer=config.HUGGINGFACE_MODELS["sentiment_twitter"],
truncation=True,
max_length=512
)
loaded_models["sentiment_twitter"] = True
logger.info("Twitter sentiment model loaded successfully")
except Exception as e:
logger.error(f"Failed to load Twitter sentiment model: {str(e)}")
loaded_models["sentiment_twitter"] = False
errors.append(f"sentiment_twitter: {str(e)}")
# Load Financial sentiment model
try:
logger.info(f"Loading sentiment_financial model: {config.HUGGINGFACE_MODELS['sentiment_financial']}")
_sentiment_financial_pipeline = pipeline(
"sentiment-analysis",
model=config.HUGGINGFACE_MODELS["sentiment_financial"],
tokenizer=config.HUGGINGFACE_MODELS["sentiment_financial"],
truncation=True,
max_length=512
)
loaded_models["sentiment_financial"] = True
logger.info("Financial sentiment model loaded successfully")
except Exception as e:
logger.error(f"Failed to load Financial sentiment model: {str(e)}")
loaded_models["sentiment_financial"] = False
errors.append(f"sentiment_financial: {str(e)}")
# Load Summarization model
try:
logger.info(f"Loading summarization model: {config.HUGGINGFACE_MODELS['summarization']}")
_summarization_pipeline = pipeline(
"summarization",
model=config.HUGGINGFACE_MODELS["summarization"],
tokenizer=config.HUGGINGFACE_MODELS["summarization"],
truncation=True
)
loaded_models["summarization"] = True
logger.info("Summarization model loaded successfully")
except Exception as e:
logger.error(f"Failed to load Summarization model: {str(e)}")
loaded_models["summarization"] = False
errors.append(f"summarization: {str(e)}")
# Check if at least one model loaded successfully
success = any(loaded_models.values())
_models_initialized = success
result = {
"success": success,
"status": "Models loaded" if success else "All models failed to load",
"models": loaded_models
}
if errors:
result["errors"] = errors
logger.info(f"Model initialization complete. Success: {success}")
return result
except Exception as e:
logger.error(f"Unexpected error during model initialization: {str(e)}")
return {
"success": False,
"status": "Initialization failed",
"models": loaded_models,
"error": str(e)
}
finally:
_models_loading = False
def _ensure_models_loaded() -> bool:
"""
Internal function to ensure models are loaded (lazy loading).
Returns:
bool: True if at least one model is loaded, False otherwise
"""
global _models_initialized
if not _models_initialized:
result = initialize_models()
return result.get("success", False)
return True
# ==================== SENTIMENT ANALYSIS ====================
def analyze_sentiment(text: str) -> Dict[str, Any]:
"""
Analyze sentiment of text using both Twitter and Financial sentiment models.
Averages the scores and maps to sentiment labels.
Args:
text: Input text to analyze (will be truncated to 512 chars)
Returns:
Dict with:
- label: str (positive/negative/neutral/very_positive/very_negative)
- score: float (averaged sentiment score from -1 to 1)
- confidence: float (confidence in the prediction 0-1)
- details: Dict with individual model results
"""
try:
# Input validation
if not text or not isinstance(text, str):
logger.warning("Invalid text input for sentiment analysis")
return {
"label": "neutral",
"score": 0.0,
"confidence": 0.0,
"error": "Invalid input text"
}
# Truncate text to model limit
original_length = len(text)
text = text[:512].strip()
if len(text) < 10:
logger.warning("Text too short for meaningful sentiment analysis")
return {
"label": "neutral",
"score": 0.0,
"confidence": 0.0,
"warning": "Text too short"
}
# Ensure models are loaded
if not _ensure_models_loaded():
logger.error("Models not available for sentiment analysis")
return {
"label": "neutral",
"score": 0.0,
"confidence": 0.0,
"error": "Models not initialized"
}
scores = []
confidences = []
model_results = {}
# Analyze with Twitter sentiment model
if _sentiment_twitter_pipeline is not None:
try:
twitter_result = _sentiment_twitter_pipeline(text)[0]
# Convert label to score (-1 to 1)
label = twitter_result['label'].lower()
confidence = twitter_result['score']
# Map label to numeric score
if 'positive' in label:
score = confidence
elif 'negative' in label:
score = -confidence
else: # neutral
score = 0.0
scores.append(score)
confidences.append(confidence)
model_results["twitter"] = {
"label": label,
"score": score,
"confidence": confidence
}
logger.debug(f"Twitter sentiment: {label} (score: {score:.3f})")
except Exception as e:
logger.error(f"Twitter sentiment analysis failed: {str(e)}")
model_results["twitter"] = {"error": str(e)}
# Analyze with Financial sentiment model
if _sentiment_financial_pipeline is not None:
try:
financial_result = _sentiment_financial_pipeline(text)[0]
# Convert label to score (-1 to 1)
label = financial_result['label'].lower()
confidence = financial_result['score']
# Map FinBERT labels to score
if 'positive' in label:
score = confidence
elif 'negative' in label:
score = -confidence
else: # neutral
score = 0.0
scores.append(score)
confidences.append(confidence)
model_results["financial"] = {
"label": label,
"score": score,
"confidence": confidence
}
logger.debug(f"Financial sentiment: {label} (score: {score:.3f})")
except Exception as e:
logger.error(f"Financial sentiment analysis failed: {str(e)}")
model_results["financial"] = {"error": str(e)}
# Check if we got any results
if not scores:
logger.error("All sentiment models failed")
return {
"label": "neutral",
"score": 0.0,
"confidence": 0.0,
"error": "All models failed",
"details": model_results
}
# Average the scores
avg_score = sum(scores) / len(scores)
avg_confidence = sum(confidences) / len(confidences)
# Map score to sentiment label based on config.SENTIMENT_LABELS
sentiment_label = "neutral"
for label, (min_score, max_score) in config.SENTIMENT_LABELS.items():
if min_score <= avg_score < max_score:
sentiment_label = label
break
result = {
"label": sentiment_label,
"score": round(avg_score, 4),
"confidence": round(avg_confidence, 4),
"details": model_results
}
if original_length > 512:
result["warning"] = f"Text truncated from {original_length} to 512 characters"
logger.info(f"Sentiment analysis complete: {sentiment_label} (score: {avg_score:.3f})")
return result
except Exception as e:
logger.error(f"Unexpected error in sentiment analysis: {str(e)}")
return {
"label": "neutral",
"score": 0.0,
"confidence": 0.0,
"error": f"Analysis failed: {str(e)}"
}
# ==================== TEXT SUMMARIZATION ====================
def summarize_text(text: str, max_length: int = 130, min_length: int = 30) -> str:
"""
Summarize text using HuggingFace summarization model.
Returns original text if it's too short or if summarization fails.
Args:
text: Input text to summarize
max_length: Maximum length of summary (default: 130)
min_length: Minimum length of summary (default: 30)
Returns:
str: Summarized text or original text if summarization fails
"""
try:
# Input validation
if not text or not isinstance(text, str):
logger.warning("Invalid text input for summarization")
return ""
text = text.strip()
# Return as-is if text is too short
if len(text) < 100:
logger.debug("Text too short for summarization, returning original")
return text
# Ensure models are loaded
if not _ensure_models_loaded():
logger.error("Models not available for summarization")
return text
# Check if summarization model is available
if _summarization_pipeline is None:
logger.warning("Summarization model not loaded, returning original text")
return text
try:
# Perform summarization
logger.debug(f"Summarizing text of length {len(text)}")
# Adjust max_length based on input length
input_length = len(text.split())
if input_length < max_length:
max_length = max(min_length, int(input_length * 0.7))
summary_result = _summarization_pipeline(
text,
max_length=max_length,
min_length=min_length,
do_sample=False,
truncation=True
)
if summary_result and len(summary_result) > 0:
summary_text = summary_result[0]['summary_text']
logger.info(f"Text summarized: {len(text)} -> {len(summary_text)} chars")
return summary_text
else:
logger.warning("Summarization returned empty result")
return text
except Exception as e:
logger.error(f"Summarization failed: {str(e)}")
return text
except Exception as e:
logger.error(f"Unexpected error in summarization: {str(e)}")
return text if isinstance(text, str) else ""
# ==================== MARKET TREND ANALYSIS ====================
def analyze_market_trend(price_history: List[Dict]) -> Dict[str, Any]:
"""
Analyze market trends using technical indicators (MA, RSI) and price history.
Generates predictions and support/resistance levels.
Args:
price_history: List of dicts with 'price', 'timestamp', 'volume' keys
Format: [{"price": 50000.0, "timestamp": 1234567890, "volume": 1000}, ...]
Returns:
Dict with:
- trend: str (Bullish/Bearish/Neutral)
- ma7: float (7-day moving average)
- ma30: float (30-day moving average)
- rsi: float (Relative Strength Index)
- support_level: float (recent price minimum)
- resistance_level: float (recent price maximum)
- prediction: str (market prediction for next 24-72h)
- confidence: float (confidence score 0-1)
"""
try:
# Input validation
if not price_history or not isinstance(price_history, list):
logger.warning("Invalid price_history input")
return {
"trend": "Neutral",
"support_level": 0.0,
"resistance_level": 0.0,
"prediction": "Insufficient data for analysis",
"confidence": 0.0,
"error": "Invalid input"
}
if len(price_history) < 2:
logger.warning("Insufficient price history for analysis")
return {
"trend": "Neutral",
"support_level": 0.0,
"resistance_level": 0.0,
"prediction": "Need at least 2 data points",
"confidence": 0.0,
"error": "Insufficient data"
}
# Extract prices from history
prices = []
for item in price_history:
if isinstance(item, dict) and 'price' in item:
try:
price = float(item['price'])
if price > 0:
prices.append(price)
except (ValueError, TypeError):
continue
elif isinstance(item, (int, float)):
if item > 0:
prices.append(float(item))
if len(prices) < 2:
logger.warning("No valid prices found in price_history")
return {
"trend": "Neutral",
"support_level": 0.0,
"resistance_level": 0.0,
"prediction": "No valid price data",
"confidence": 0.0,
"error": "No valid prices"
}
# Calculate support and resistance levels
support_level = min(prices[-30:]) if len(prices) >= 30 else min(prices)
resistance_level = max(prices[-30:]) if len(prices) >= 30 else max(prices)
# Calculate Moving Averages
ma7 = None
ma30 = None
if len(prices) >= 7:
ma7 = sum(prices[-7:]) / 7
else:
ma7 = sum(prices) / len(prices)
if len(prices) >= 30:
ma30 = sum(prices[-30:]) / 30
else:
ma30 = sum(prices) / len(prices)
# Calculate RSI (Relative Strength Index)
rsi = _calculate_rsi(prices, period=config.RSI_PERIOD)
# Determine trend based on MA crossover and current price
current_price = prices[-1]
trend = "Neutral"
if ma7 > ma30 and current_price > ma7:
trend = "Bullish"
elif ma7 < ma30 and current_price < ma7:
trend = "Bearish"
elif abs(ma7 - ma30) / ma30 < 0.02: # Within 2% = neutral
trend = "Neutral"
else:
# Additional checks
if current_price > ma30:
trend = "Bullish"
elif current_price < ma30:
trend = "Bearish"
# Generate prediction based on trend and RSI
prediction = _generate_market_prediction(
trend=trend,
rsi=rsi,
current_price=current_price,
ma7=ma7,
ma30=ma30,
support_level=support_level,
resistance_level=resistance_level
)
# Calculate confidence score based on data quality
confidence = _calculate_confidence(
data_points=len(prices),
rsi=rsi,
trend=trend,
price_volatility=_calculate_volatility(prices)
)
result = {
"trend": trend,
"ma7": round(ma7, 2),
"ma30": round(ma30, 2),
"rsi": round(rsi, 2),
"support_level": round(support_level, 2),
"resistance_level": round(resistance_level, 2),
"current_price": round(current_price, 2),
"prediction": prediction,
"confidence": round(confidence, 4),
"data_points": len(prices)
}
logger.info(f"Market analysis complete: {trend} trend, RSI: {rsi:.2f}, Confidence: {confidence:.2f}")
return result
except Exception as e:
logger.error(f"Unexpected error in market trend analysis: {str(e)}")
return {
"trend": "Neutral",
"support_level": 0.0,
"resistance_level": 0.0,
"prediction": "Analysis failed",
"confidence": 0.0,
"error": f"Analysis error: {str(e)}"
}
# ==================== HELPER FUNCTIONS ====================
def _calculate_rsi(prices: List[float], period: int = 14) -> float:
"""
Calculate Relative Strength Index (RSI).
Args:
prices: List of prices
period: RSI period (default: 14)
Returns:
float: RSI value (0-100)
"""
try:
if len(prices) < period + 1:
# Not enough data, use available data
period = max(2, len(prices) - 1)
# Calculate price changes
deltas = [prices[i] - prices[i-1] for i in range(1, len(prices))]
# Separate gains and losses
gains = [delta if delta > 0 else 0 for delta in deltas]
losses = [-delta if delta < 0 else 0 for delta in deltas]
# Calculate average gains and losses
if len(gains) >= period:
avg_gain = sum(gains[-period:]) / period
avg_loss = sum(losses[-period:]) / period
else:
avg_gain = sum(gains) / len(gains) if gains else 0
avg_loss = sum(losses) / len(losses) if losses else 0
# Avoid division by zero
if avg_loss == 0:
return 100.0 if avg_gain > 0 else 50.0
# Calculate RS and RSI
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
except Exception as e:
logger.error(f"RSI calculation error: {str(e)}")
return 50.0 # Return neutral RSI on error
def _generate_market_prediction(
trend: str,
rsi: float,
current_price: float,
ma7: float,
ma30: float,
support_level: float,
resistance_level: float
) -> str:
"""
Generate market prediction based on technical indicators.
Returns:
str: Detailed prediction for next 24-72 hours
"""
try:
predictions = []
# RSI-based predictions
if rsi > 70:
predictions.append("overbought conditions suggest potential correction")
elif rsi < 30:
predictions.append("oversold conditions suggest potential bounce")
elif 40 <= rsi <= 60:
predictions.append("neutral momentum")
# Trend-based predictions
if trend == "Bullish":
if current_price < resistance_level * 0.95:
predictions.append(f"upward movement toward resistance at ${resistance_level:.2f}")
else:
predictions.append("potential breakout above resistance if momentum continues")
elif trend == "Bearish":
if current_price > support_level * 1.05:
predictions.append(f"downward pressure toward support at ${support_level:.2f}")
else:
predictions.append("potential breakdown below support if selling continues")
else: # Neutral
predictions.append(f"consolidation between ${support_level:.2f} and ${resistance_level:.2f}")
# MA crossover signals
if ma7 > ma30 * 1.02:
predictions.append("strong bullish crossover signal")
elif ma7 < ma30 * 0.98:
predictions.append("strong bearish crossover signal")
# Combine predictions
if predictions:
prediction_text = f"Next 24-72h: Expect {', '.join(predictions)}."
else:
prediction_text = "Next 24-72h: Insufficient signals for reliable prediction."
# Add price range estimate
price_range = resistance_level - support_level
if price_range > 0:
expected_low = current_price - (price_range * 0.1)
expected_high = current_price + (price_range * 0.1)
prediction_text += f" Price likely to range between ${expected_low:.2f} and ${expected_high:.2f}."
return prediction_text
except Exception as e:
logger.error(f"Prediction generation error: {str(e)}")
return "Unable to generate prediction due to data quality issues."
def _calculate_volatility(prices: List[float]) -> float:
"""
Calculate price volatility (standard deviation).
Args:
prices: List of prices
Returns:
float: Volatility as percentage
"""
try:
if len(prices) < 2:
return 0.0
mean_price = sum(prices) / len(prices)
variance = sum((p - mean_price) ** 2 for p in prices) / len(prices)
std_dev = variance ** 0.5
# Return as percentage of mean
volatility = (std_dev / mean_price) * 100 if mean_price > 0 else 0.0
return volatility
except Exception as e:
logger.error(f"Volatility calculation error: {str(e)}")
return 0.0
def _calculate_confidence(
data_points: int,
rsi: float,
trend: str,
price_volatility: float
) -> float:
"""
Calculate confidence score for market analysis.
Args:
data_points: Number of price data points
rsi: RSI value
trend: Market trend
price_volatility: Price volatility percentage
Returns:
float: Confidence score (0-1)
"""
try:
confidence = 0.0
# Data quality score (0-0.4)
if data_points >= 30:
data_score = 0.4
elif data_points >= 14:
data_score = 0.3
elif data_points >= 7:
data_score = 0.2
else:
data_score = 0.1
confidence += data_score
# RSI confidence (0-0.3)
# Extreme RSI values (very high or very low) give higher confidence
if rsi > 70 or rsi < 30:
rsi_score = 0.3
elif rsi > 60 or rsi < 40:
rsi_score = 0.2
else:
rsi_score = 0.1
confidence += rsi_score
# Trend clarity (0-0.2)
if trend in ["Bullish", "Bearish"]:
trend_score = 0.2
else:
trend_score = 0.1
confidence += trend_score
# Volatility penalty (0-0.1)
# Lower volatility = higher confidence
if price_volatility < 5:
volatility_score = 0.1
elif price_volatility < 10:
volatility_score = 0.05
else:
volatility_score = 0.0
confidence += volatility_score
# Ensure confidence is between 0 and 1
confidence = max(0.0, min(1.0, confidence))
return confidence
except Exception as e:
logger.error(f"Confidence calculation error: {str(e)}")
return 0.5 # Return medium confidence on error
# ==================== CACHE DECORATORS ====================
@lru_cache(maxsize=100)
def _cached_sentiment(text_hash: int) -> Dict[str, Any]:
"""Cache wrapper for sentiment analysis (internal use only)."""
# This would be called by analyze_sentiment with hash(text)
# Not exposed directly to avoid cache invalidation issues
pass
# ==================== MODULE INFO ====================
def get_model_info() -> Dict[str, Any]:
"""
Get information about loaded models and their status.
Returns:
Dict with model information
"""
return {
"transformers_available": TRANSFORMERS_AVAILABLE,
"models_initialized": _models_initialized,
"models_loading": _models_loading,
"loaded_models": {
"sentiment_twitter": _sentiment_twitter_pipeline is not None,
"sentiment_financial": _sentiment_financial_pipeline is not None,
"summarization": _summarization_pipeline is not None,
},
"model_names": config.HUGGINGFACE_MODELS,
"device": "cuda" if TRANSFORMERS_AVAILABLE and torch.cuda.is_available() else "cpu"
}
if __name__ == "__main__":
# Test the module
print("="*60)
print("AI Models Module Test")
print("="*60)
# Get model info
info = get_model_info()
print(f"\nTransformers available: {info['transformers_available']}")
print(f"Models initialized: {info['models_initialized']}")
print(f"Device: {info['device']}")
# Initialize models
print("\n" + "="*60)
print("Initializing models...")
print("="*60)
result = initialize_models()
print(f"Success: {result['success']}")
print(f"Status: {result['status']}")
print(f"Loaded models: {result['models']}")
if result['success']:
# Test sentiment analysis
print("\n" + "="*60)
print("Testing Sentiment Analysis")
print("="*60)
test_text = "Bitcoin shows strong bullish momentum with increasing adoption and positive market sentiment."
sentiment = analyze_sentiment(test_text)
print(f"Text: {test_text}")
print(f"Sentiment: {sentiment['label']}")
print(f"Score: {sentiment['score']}")
print(f"Confidence: {sentiment['confidence']}")
# Test summarization
print("\n" + "="*60)
print("Testing Summarization")
print("="*60)
long_text = """
Bitcoin, the world's largest cryptocurrency by market capitalization, has experienced
significant growth over the past decade. Initially created as a peer-to-peer electronic
cash system, Bitcoin has evolved into a store of value and investment asset. Institutional
adoption has increased dramatically, with major companies adding Bitcoin to their balance
sheets. The cryptocurrency market has matured, with improved infrastructure, regulatory
clarity, and growing mainstream acceptance. However, volatility remains a characteristic
feature of the market, presenting both opportunities and risks for investors.
"""
summary = summarize_text(long_text)
print(f"Original length: {len(long_text)} chars")
print(f"Summary length: {len(summary)} chars")
print(f"Summary: {summary}")
# Test market trend analysis
print("\n" + "="*60)
print("Testing Market Trend Analysis")
print("="*60)
# Simulated price history (bullish trend)
test_prices = [
{"price": 45000, "timestamp": 1000000, "volume": 100},
{"price": 45500, "timestamp": 1000001, "volume": 120},
{"price": 46000, "timestamp": 1000002, "volume": 110},
{"price": 46500, "timestamp": 1000003, "volume": 130},
{"price": 47000, "timestamp": 1000004, "volume": 140},
{"price": 47500, "timestamp": 1000005, "volume": 150},
{"price": 48000, "timestamp": 1000006, "volume": 160},
{"price": 48500, "timestamp": 1000007, "volume": 170},
]
trend = analyze_market_trend(test_prices)
print(f"Trend: {trend['trend']}")
print(f"RSI: {trend['rsi']}")
print(f"MA7: {trend['ma7']}")
print(f"MA30: {trend['ma30']}")
print(f"Support: ${trend['support_level']}")
print(f"Resistance: ${trend['resistance_level']}")
print(f"Prediction: {trend['prediction']}")
print(f"Confidence: {trend['confidence']}")
print("\n" + "="*60)
print("Test complete!")
print("="*60)