engine.py

"""
Kronos Prediction Engine
Performs autoregressive financial time series prediction with probabilistic forecasts.
"""

import pandas as pd
import numpy as np
import torch
from typing import Dict, Tuple, Optional
from pathlib import Path
import warnings

from model import Kronos, KronosTokenizer, KronosPredictor
from data_fetcher import fetch_hourly_klines, get_data_info

# Suppress warnings for cleaner output
warnings.filterwarnings('ignore')

# Global model cache to avoid reloading the same model multiple times
loaded_models = {}

# Model configuration mapping
MODEL_CONFIG = {
    'NeoQuasar/Kronos-mini': {
        'name': 'Kronos-Mini',
        'tokenizer': 'NeoQuasar/Kronos-Tokenizer-2k',
        'context_length': 2048,
        'params': '4.1M'
    },
    'NeoQuasar/Kronos-small': {
        'name': 'Kronos-Small',
        'tokenizer': 'NeoQuasar/Kronos-Tokenizer-base',
        'context_length': 512,
        'params': '24.7M'
    },
    'NeoQuasar/Kronos-base': {
        'name': 'Kronos-Base',
        'tokenizer': 'NeoQuasar/Kronos-Tokenizer-base',
        'context_length': 512,
        'params': '102.3M'
    }
}


class KronosPredictionEngine:
    """
    Prediction engine for Kronos model.
    Handles model loading, data preparation, and probabilistic forecasting.
    """
    
    def __init__(self, 
                 tokenizer_id: str = "NeoQuasar/Kronos-Tokenizer-base",
                 model_id: str = "NeoQuasar/Kronos-small",
                 model_path: Optional[str] = None,
                 device: str = "cpu",
                 max_context: int = 512,
                 lookback: int = 400):
        """
        Initialize the prediction engine.
        
        Args:
            tokenizer_id (str): HuggingFace tokenizer model ID (deprecated if model_path provided)
            model_id (str): HuggingFace model ID (deprecated if model_path provided)
            model_path (str): Model path (e.g., 'NeoQuasar/Kronos-small'). Overrides model_id if provided.
            device (str): Device to run on ('cpu', 'cuda', 'mps')
            max_context (int): Maximum context length for the model
            lookback (int): Lookback window for historical data (default: 400)
        """
        # Use model_path if provided, otherwise use model_id
        if model_path:
            model_id = model_path
        
        # Get model configuration
        if model_id in MODEL_CONFIG:
            config = MODEL_CONFIG[model_id]
            tokenizer_id = config['tokenizer']
            max_context = config['context_length']
            model_name = config['name']
        else:
            model_name = model_id
        
        print(f"🤖 Preparing Kronos models...")
        print(f"   Model: {model_name} ({model_id})")
        print(f"   Tokenizer: {tokenizer_id}")
        
        self.device = device
        self.lookback = lookback
        self.max_context = max_context  # Store for use in prepare_data truncation
        self.pred_len = 24
        self.model_id = model_id
        self.tokenizer_id = tokenizer_id
        
        try:
            # Check if model is already loaded
            if model_id in loaded_models:
                print(f"   ♻️  Using cached model instance...")
                cached = loaded_models[model_id]
                self.tokenizer = cached['tokenizer']
                self.model = cached['model']
                self.predictor = cached['predictor']
                print(f"✅ Models loaded from cache")
            else:
                print(f"   📥 Loading model from HuggingFace (this may take a minute)...")
                # Load tokenizer
                tokenizer = KronosTokenizer.from_pretrained(tokenizer_id)
                
                # Load model with OOM error handling
                try:
                    model = Kronos.from_pretrained(model_id)
                except RuntimeError as e:
                    if 'out of memory' in str(e).lower() or 'cuda out of memory' in str(e).lower():
                        print(f"❌ Out of Memory Error: The {model_name} model is too large for your system.")
                        print(f"   💡 Try a smaller model:")
                        print(f"      - NeoQuasar/Kronos-mini (4.1M) - Most memory efficient")
                        print(f"      - NeoQuasar/Kronos-small (24.7M) - Balanced")
                        if device == 'cuda':
                            print(f"   💡 Or switch to CPU mode (slower but uses less GPU memory)")
                        raise RuntimeError(
                            f"Out of Memory: {model_name} is too large. Try a smaller model (Kronos-mini or Kronos-small) "
                            f"or switch to CPU device."
                        )
                    else:
                        raise
                
                # Create predictor
                predictor = KronosPredictor(
                    model,
                    tokenizer,
                    device=device,
                    max_context=max_context
                )
                
                # Cache the loaded models
                loaded_models[model_id] = {
                    'tokenizer': tokenizer,
                    'model': model,
                    'predictor': predictor
                }
                
                self.tokenizer = tokenizer
                self.model = model
                self.predictor = predictor
                print(f"✅ Models loaded successfully on {device}")
                
        except RuntimeError as e:
            if 'Out of Memory' in str(e):
                raise e
            print(f"❌ Failed to load models: {str(e)}")
            raise
    
    def prepare_data(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Series, pd.Series]:
        """
        Prepare data for prediction.
        Automatically pads DataFrame to 400 rows if insufficient data.
        
        Args:
            df (pd.DataFrame): Input data with columns: timestamps, open, high, low, close, volume
        
        Returns:
            Tuple[pd.DataFrame, pd.Series, pd.Series]: (x_df, x_timestamp, y_timestamp)
        """
        min_lookback = 50  # Minimum data points for model to work
        target_lookback = 400  # Target context window
        
        if len(df) < min_lookback:
            raise ValueError(
                f"Insufficient data: need at least {min_lookback} rows, got {len(df)}"
            )
        
        # Pad DataFrame to target_lookback if insufficient
        if len(df) < target_lookback:
            print(f"⚠️  Data has {len(df)} rows, padding to {target_lookback}...")
            df = self._pad_dataframe(df, target_lookback)
            print(f"✅ DataFrame padded to {len(df)} rows")
        
        # Truncate to max_context (384 tokens) — the model only attends to this window anyway.
        # Using fewer tokens dramatically speeds up the attention computation.
        truncate_to = min(self.lookback, self.max_context, len(df) - self.pred_len)
        
        if truncate_to < min_lookback:
            raise ValueError(
                f"Insufficient data: need at least {min_lookback + self.pred_len} rows for lookback + prediction, got {len(df)}"
            )
        
        # Use last truncate_to points as input
        x_df = df[['open', 'high', 'low', 'close', 'volume']].iloc[-truncate_to:].copy()
        x_timestamp = df['timestamps'].iloc[-truncate_to:].copy()
        
        # Generate future timestamps for prediction
        last_timestamp = df['timestamps'].iloc[-1]
        if len(df) > 1:
            # Use the minimum positive time diff across all rows to avoid
            # overnight/weekend gaps skewing the forecast frequency
            all_diffs = df['timestamps'].diff().dropna()
            positive_diffs = all_diffs[all_diffs > pd.Timedelta(0)]
            time_diff = positive_diffs.min() if len(positive_diffs) > 0 else pd.Timedelta(hours=1)
        else:
            time_diff = pd.Timedelta(hours=1)
        
        y_timestamp = pd.date_range(
            start=last_timestamp + time_diff,
            periods=self.pred_len,
            freq=time_diff
        )
        
        return x_df, x_timestamp, y_timestamp
    
    def _pad_dataframe(self, df: pd.DataFrame, target_rows: int = 400) -> pd.DataFrame:
        """
        Pad DataFrame to target_rows by duplicating the earliest row.
        
        Args:
            df (pd.DataFrame): Original DataFrame
            target_rows (int): Target number of rows
        
        Returns:
            pd.DataFrame: Padded DataFrame
        """
        if len(df) >= target_rows:
            return df
        
        rows_needed = target_rows - len(df)
        
        # Get the earliest row for padding
        earliest_row = df.iloc[0].copy()
        
        # Calculate timestamp interval
        if len(df) > 1:
            time_diff = df.iloc[1]['timestamps'] - df.iloc[0]['timestamps']
        else:
            time_diff = pd.Timedelta(hours=1)
        
        # Create padding rows
        padding_rows = []
        for i in range(rows_needed):
            padded_row = earliest_row.copy()
            padded_row['timestamps'] = earliest_row['timestamps'] - (time_diff * (rows_needed - i))
            padding_rows.append(padded_row)
        
        # Combine padding with original data
        padding_df = pd.DataFrame(padding_rows)
        result = pd.concat([padding_df, df], ignore_index=True)
        
        return result
    
    def predict(self, 
                df: pd.DataFrame,
                sample_count: int = 30,
                temperature: float = 1.0,
                top_p: float = 0.9) -> Dict:
        """
        Generate probabilistic predictions.
        
        Args:
            df (pd.DataFrame): Historical OHLCV data
            sample_count (int): Number of sample paths (default: 30)
            temperature (float): Sampling temperature (default: 1.0)
            top_p (float): Nucleus sampling parameter (default: 0.9)
        
        Returns:
            Dict: Prediction results including mean, std, percentiles, and all samples
        """
        print(f"\n🔮 Generating {sample_count} sample paths for {self.pred_len}-hour forecast...")
        
        # Prepare data
        x_df, x_timestamp, y_timestamp = self.prepare_data(df)
        
        # Ensure timestamps are Series, not DatetimeIndex
        if isinstance(x_timestamp, pd.DatetimeIndex):
            x_timestamp = pd.Series(x_timestamp.values, name='timestamps')
        if isinstance(y_timestamp, pd.DatetimeIndex):
            y_timestamp = pd.Series(y_timestamp.values, name='timestamps')
        
        # Each call with sample_count=1 draws an independent stochastic sample.
        # auto_regressive_inference averages internally when sample_count>1, so
        # calling once with sample_count=N would collapse all variance → std=0.
        # We need independent calls to preserve the distribution for confidence intervals.
        predictions_list = []
        print(f"   Generating samples: ", end="", flush=True)
        for i in range(sample_count):
            if (i + 1) % max(1, sample_count // 5) == 0:
                print(f"{i+1}...", end="", flush=True)
            try:
                pred_df = self.predictor.predict(
                    df=x_df,
                    x_timestamp=x_timestamp,
                    y_timestamp=y_timestamp,
                    pred_len=self.pred_len,
                    T=temperature,
                    top_p=top_p,
                    sample_count=1,
                    verbose=False
                )
                predictions_list.append(pred_df)
            except Exception as e:
                print(f"\n⚠️  Sample {i+1} failed: {str(e)}, skipping...")
                continue
        print("✅")
        
        if not predictions_list:
            raise RuntimeError("All predictions failed")
        
        print(f"✅ Successfully generated {len(predictions_list)} samples")
        results = self._aggregate_predictions(predictions_list, y_timestamp)
        
        return results
    
    def _aggregate_predictions(self, 
                               predictions_list: list,
                               y_timestamp: pd.Series) -> Dict:
        """
        Aggregate multiple sample predictions into probabilistic forecast.
        
        Args:
            predictions_list (list): List of prediction DataFrames
            y_timestamp (pd.Series): Future timestamps
        
        Returns:
            Dict: Aggregated statistics and forecasts
        """
        # Stack all predictions
        samples = {}
        for col in predictions_list[0].columns:
            samples[col] = np.array([pred[col].values for pred in predictions_list])
        
        # Calculate statistics
        results = {
            'timestamps': np.array([ts.isoformat() if hasattr(ts, 'isoformat') else str(ts) 
                                   for ts in y_timestamp]),
            'samples': {}
        }
        
        for col in samples.keys():
            data = samples[col]
            
            results[col] = {
                'mean': np.mean(data, axis=0),
                'std': np.std(data, axis=0),
                'median': np.median(data, axis=0),
                'q5': np.percentile(data, 5, axis=0),    # 5th percentile
                'q25': np.percentile(data, 25, axis=0),  # 25th percentile
                'q75': np.percentile(data, 75, axis=0),  # 75th percentile
                'q95': np.percentile(data, 95, axis=0),  # 95th percentile
            }
            
            results['samples'][col] = data
        
        # Create summary DataFrame
        summary_df = pd.DataFrame({
            'timestamps': results['timestamps'],
            'open_mean': results['open']['mean'],
            'open_std': results['open']['std'],
            'high_mean': results['high']['mean'],
            'high_std': results['high']['std'],
            'low_mean': results['low']['mean'],
            'low_std': results['low']['std'],
            'close_mean': results['close']['mean'],
            'close_std': results['close']['std'],
            'close_q5': results['close']['q5'],
            'close_q25': results['close']['q25'],
            'close_q75': results['close']['q75'],
            'close_q95': results['close']['q95'],
            'volume_mean': results['volume']['mean'],
            'volume_std': results['volume']['std'],
        })
        
        results['summary_df'] = summary_df
        
        return results
    
    def print_forecast(self, results: Dict) -> None:
        """
        Print formatted forecast results.
        
        Args:
            results (Dict): Prediction results from predict()
        """
        df = results['summary_df']
        
        print("\n📊 Probabilistic Forecast Summary:")
        print("=" * 100)
        print(f"{'Time':<22} {'Close (Mean)':<12} {'±Std':<10} {'[5%, 95%]':<20}")
        print("-" * 100)
        
        for idx, row in df.iterrows():
            ts = row['timestamps'][:16] if isinstance(row['timestamps'], str) else str(row['timestamps'])[:16]
            close = row['close_mean']
            std = row['close_std']
            q5 = row['close_q5']
            q95 = row['close_q95']
            
            print(f"{ts:<22} ${close:>10.2f} ±{std:>8.2f} [{q5:>8.2f}, {q95:>8.2f}]")
        
        print("=" * 100)


def get_prediction(symbol: str = None,
                   data_path: str = None,
                   periods: int = 500,
                   sample_count: int = 30,
                   temperature: float = 1.0,
                   top_p: float = 0.9,
                   save_results: bool = True,
                   lookback: int = 400) -> Dict:
    """
    Main function to get prediction for a given ticker symbol or data file.
    
    Args:
        symbol (str): Stock ticker (e.g., 'AAPL', 'BTC-USD'). Either symbol or data_path required.
        data_path (str): Path to CSV file with OHLCV data. Either symbol or data_path required.
        periods (int): Number of historical periods to use (default: 500). Ignored if data_path provided.
        sample_count (int): Number of sample paths (default: 30)
        temperature (float): Sampling temperature (default: 1.0)
        top_p (float): Nucleus sampling parameter (default: 0.9)
        save_results (bool): Whether to save results to CSV (default: True)
        lookback (int): Lookback window for historical data (default: 400). Auto-adjusted based on data availability.
    
    Returns:
        Dict: Prediction results with mean, std, and confidence intervals
    
    Example:
        >>> results = get_prediction(symbol='AAPL')
        >>> results = get_prediction(data_path='examples/data/XSHG_5min_600977.csv', sample_count=30)
        >>> results = get_prediction(symbol='BTC-USD', sample_count=50, lookback=100)
    """
    if not symbol and not data_path:
        raise ValueError("Either 'symbol' or 'data_path' must be provided")
    
    if symbol and data_path:
        raise ValueError("Provide only one of 'symbol' or 'data_path', not both")
    
    print(f"\n🚀 Kronos Prediction Engine")
    print(f"{'='*60}")
    
    # Fetch or load data
    print(f"\n1️⃣  Loading historical data...")
    try:
        if data_path:
            # Load from CSV file
            df = pd.read_csv(data_path)
            df['timestamps'] = pd.to_datetime(df['timestamps'])
            df = df.sort_values('timestamps').reset_index(drop=True)
            data_source = f"file: {data_path}"
        else:
            # Fetch from yfinance
            df = fetch_hourly_klines(symbol, periods=periods)
            data_source = f"ticker: {symbol}"
        
        info = get_data_info(df)
        print(f"   ✅ Loaded {info['total_rows']} records from {data_source}")
        print(f"   📅 Date range: {info['start_date']} to {info['end_date']}")
        print(f"   💰 Price range: ${info['price_range_min']:.2f} - ${info['price_range_max']:.2f}")
    except Exception as e:
        print(f"   ❌ Failed to load data: {str(e)}")
        raise
    
    # Initialize engine with configurable lookback
    print(f"\n2️⃣  Initializing Kronos prediction engine...")
    try:
        engine = KronosPredictionEngine(lookback=lookback)
    except Exception as e:
        print(f"   ❌ Failed to initialize engine: {str(e)}")
        raise
    
    # Generate predictions
    print(f"\n3️⃣  Generating probabilistic forecast...")
    try:
        results = engine.predict(
            df,
            sample_count=sample_count,
            temperature=temperature,
            top_p=top_p
        )
    except Exception as e:
        print(f"   ❌ Prediction failed: {str(e)}")
        raise
    
    # Print summary
    print(f"\n4️⃣  Forecast Summary")
    engine.print_forecast(results)
    
    # Save results
    if save_results:
        print(f"\n5️⃣  Saving results...")
        output_name = symbol if symbol else Path(data_path).stem
        output_path = Path('predictions') / f"{output_name}_forecast.csv"
        output_path.parent.mkdir(parents=True, exist_ok=True)
        results['summary_df'].to_csv(output_path, index=False)
        print(f"   💾 Results saved to: {output_path}")
        
        # Also save full sample paths
        samples_path = output_path.parent / f"{output_name}_samples.npz"
        np.savez(samples_path, **results['samples'])
        print(f"   💾 Sample paths saved to: {samples_path}")
    
    print(f"\n✅ Prediction complete!")
    print(f"{'='*60}\n")
    
    return results


if __name__ == "__main__":
    import sys
    
    # Get symbol from command line or use default
    symbol = sys.argv[1].upper() if len(sys.argv) > 1 else "AAPL"
    sample_count = int(sys.argv[2]) if len(sys.argv) > 2 else 30
    
    try:
        results = get_prediction(symbol, sample_count=sample_count)
    except Exception as e:
        print(f"\n❌ Error: {str(e)}")
        sys.exit(1)