hf-data-engine/ai_models.py

#!/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)