Upload folder using huggingface_hub

services/cache_manager.py

@@ -0,0 +1,170 @@
+"""
+Cache manager for storing predictions and uploaded data
+"""
+
+import logging
+from typing import Dict, Optional
+from datetime import datetime, timedelta
+import pandas as pd
+
+from config.constants import MAX_PREDICTION_HISTORY
+
+logger = logging.getLogger(__name__)
+
+
+class CacheManager:
+    """
+    Manages caching of predictions and data to improve performance
+    """
+
+    def __init__(self):
+        self.predictions = []  # List of prediction results
+        self.uploaded_data = {}  # Dict of uploaded datasets
+        self.max_predictions = MAX_PREDICTION_HISTORY
+
+    def store_prediction(
+        self,
+        data_hash: str,
+        horizon: int,
+        confidence_levels: list,
+        result: Dict
+    ):
+        """
+        Store a prediction result
+
+        Args:
+            data_hash: Hash of the input data
+            horizon: Forecast horizon used
+            confidence_levels: Confidence levels used
+            result: Prediction result dictionary
+        """
+        prediction_entry = {
+            'data_hash': data_hash,
+            'horizon': horizon,
+            'confidence_levels': confidence_levels,
+            'result': result,
+            'timestamp': datetime.now()
+        }
+
+        self.predictions.append(prediction_entry)
+
+        # Keep only the most recent predictions
+        if len(self.predictions) > self.max_predictions:
+            self.predictions = self.predictions[-self.max_predictions:]
+
+        logger.debug(f"Stored prediction, cache size: {len(self.predictions)}")
+
+    def get_prediction(
+        self,
+        data_hash: str,
+        horizon: int,
+        confidence_levels: list
+    ) -> Optional[Dict]:
+        """
+        Retrieve a cached prediction if available
+
+        Args:
+            data_hash: Hash of the input data
+            horizon: Forecast horizon
+            confidence_levels: Confidence levels
+
+        Returns:
+            Cached prediction result or None
+        """
+        for entry in reversed(self.predictions):
+            if (entry['data_hash'] == data_hash and
+                entry['horizon'] == horizon and
+                entry['confidence_levels'] == confidence_levels):
+
+                logger.info("Cache hit for prediction")
+                return entry['result']
+
+        logger.debug("Cache miss for prediction")
+        return None
+
+    def store_data(self, filename: str, data: pd.DataFrame):
+        """
+        Store uploaded data
+
+        Args:
+            filename: Name of the uploaded file
+            data: DataFrame containing the data
+        """
+        self.uploaded_data[filename] = {
+            'data': data,
+            'timestamp': datetime.now()
+        }
+
+        logger.info(f"Stored data for {filename}")
+
+    def get_data(self, filename: str) -> Optional[pd.DataFrame]:
+        """
+        Retrieve uploaded data
+
+        Args:
+            filename: Name of the file
+
+        Returns:
+            DataFrame or None
+        """
+        if filename in self.uploaded_data:
+            return self.uploaded_data[filename]['data']
+        return None
+
+    def clear_old_data(self, max_age_hours: int = 24):
+        """
+        Clear data older than specified hours
+
+        Args:
+            max_age_hours: Maximum age in hours
+        """
+        cutoff = datetime.now() - timedelta(hours=max_age_hours)
+
+        # Clear old uploaded data
+        old_files = [
+            filename for filename, entry in self.uploaded_data.items()
+            if entry['timestamp'] < cutoff
+        ]
+
+        for filename in old_files:
+            del self.uploaded_data[filename]
+
+        if old_files:
+            logger.info(f"Cleared {len(old_files)} old data entries")
+
+    def clear_all(self):
+        """Clear all cached data"""
+        self.predictions.clear()
+        self.uploaded_data.clear()
+        logger.info("Cleared all cache")
+
+    def get_stats(self) -> Dict:
+        """Get cache statistics"""
+        return {
+            'num_predictions': len(self.predictions),
+            'num_datasets': len(self.uploaded_data),
+            'total_memory_mb': self._estimate_memory()
+        }
+
+    def _estimate_memory(self) -> float:
+        """Estimate memory usage in MB (rough estimate)"""
+        try:
+            total_bytes = 0
+
+            # Estimate prediction cache size
+            for entry in self.predictions:
+                if 'forecast' in entry['result']:
+                    total_bytes += entry['result']['forecast'].memory_usage(deep=True).sum()
+
+            # Estimate data cache size
+            for entry in self.uploaded_data.values():
+                total_bytes += entry['data'].memory_usage(deep=True).sum()
+
+            return total_bytes / (1024 * 1024)
+        except Exception as e:
+            logger.warning(f"Failed to estimate memory: {str(e)}")
+            return 0.0
+
+
+# Global cache instance
+cache_manager = CacheManager()

services/data_processor.py

@@ -0,0 +1,393 @@
+"""
+Data preprocessing pipeline for time series data
+"""
+
+import logging
+from typing import Dict, List, Optional, Tuple, Any
+import pandas as pd
+import numpy as np
+from io import BytesIO
+
+from config.constants import (
+    DATE_FORMATS,
+    MAX_MISSING_PERCENT,
+    MIN_DATA_POINTS_MULTIPLIER,
+    ALLOWED_EXTENSIONS
+)
+
+logger = logging.getLogger(__name__)
+
+
+class DataProcessor:
+    """
+    Handles all data preprocessing tasks for time series forecasting
+    """
+
+    def __init__(self):
+        self.data = None
+        self.original_data = None
+        self.metadata = {}
+
+    def _timedelta_to_freq_string(self, td: pd.Timedelta) -> str:
+        """
+        Convert a Timedelta to a pandas frequency string
+
+        Args:
+            td: Timedelta object
+
+        Returns:
+            Frequency string (e.g., 'H', 'D', '5min', etc.)
+        """
+        total_seconds = td.total_seconds()
+
+        # Common time frequencies
+        if total_seconds == 0:
+            return 'D'  # Default to daily if zero
+        elif total_seconds % 604800 == 0:  # Weekly (7 days)
+            weeks = int(total_seconds / 604800)
+            return f'{weeks}W' if weeks > 1 else 'W'
+        elif total_seconds % 86400 == 0:  # Daily (24 hours)
+            days = int(total_seconds / 86400)
+            return f'{days}D' if days > 1 else 'D'
+        elif total_seconds % 3600 == 0:  # Hourly
+            hours = int(total_seconds / 3600)
+            return f'{hours}H' if hours > 1 else 'H'
+        elif total_seconds % 60 == 0:  # Minutes
+            minutes = int(total_seconds / 60)
+            return f'{minutes}min' if minutes > 1 else 'min'
+        elif total_seconds % 1 == 0:  # Seconds
+            seconds = int(total_seconds)
+            return f'{seconds}s' if seconds > 1 else 's'
+        else:
+            # For irregular frequencies, default to daily
+            logger.warning(f"Irregular frequency detected ({td}), defaulting to Daily")
+            return 'D'
+
+    def load_file(self, contents: bytes, filename: str) -> Dict[str, Any]:
+        """
+        Load data from uploaded file
+
+        Args:
+            contents: File contents as bytes
+            filename: Original filename
+
+        Returns:
+            Dictionary with status and data/error
+        """
+        try:
+            # Determine file type
+            extension = filename.split('.')[-1].lower()
+
+            if extension not in ALLOWED_EXTENSIONS:
+                return {
+                    'status': 'error',
+                    'error': f'Invalid file type. Allowed: {", ".join(ALLOWED_EXTENSIONS)}'
+                }
+
+            # Load data based on file type
+            if extension == 'csv':
+                self.data = pd.read_csv(BytesIO(contents))
+            elif extension in ['xlsx', 'xls']:
+                self.data = pd.read_excel(BytesIO(contents))
+
+            self.original_data = self.data.copy()
+
+            logger.info(f"Loaded file {filename} with shape {self.data.shape}")
+
+            # Generate initial metadata
+            self.metadata = {
+                'filename': filename,
+                'rows': len(self.data),
+                'columns': list(self.data.columns),
+                'dtypes': {col: str(dtype) for col, dtype in self.data.dtypes.items()}
+            }
+
+            return {
+                'status': 'success',
+                'data': self.data,
+                'metadata': self.metadata
+            }
+
+        except Exception as e:
+            logger.error(f"Failed to load file {filename}: {str(e)}", exc_info=True)
+            return {
+                'status': 'error',
+                'error': f'Failed to load file: {str(e)}'
+            }
+
+    def validate_data(
+        self,
+        date_column: str,
+        target_column: str,
+        id_column: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Validate the selected columns and data quality
+
+        Args:
+            date_column: Name of the date/time column
+            target_column: Name of the target variable column
+            id_column: Optional ID column for multivariate series
+
+        Returns:
+            Validation result dictionary
+        """
+        try:
+            issues = []
+            warnings = []
+
+            # Check if columns exist
+            if date_column not in self.data.columns:
+                issues.append(f"Date column '{date_column}' not found")
+            if target_column not in self.data.columns:
+                issues.append(f"Target column '{target_column}' not found")
+            if id_column and id_column not in self.data.columns:
+                issues.append(f"ID column '{id_column}' not found")
+
+            if issues:
+                return {'status': 'error', 'issues': issues}
+
+            # Check for missing values
+            missing_pct = (self.data[target_column].isna().sum() / len(self.data)) * 100
+            if missing_pct > MAX_MISSING_PERCENT:
+                warnings.append(
+                    f"Target column has {missing_pct:.1f}% missing values (>{MAX_MISSING_PERCENT}%)"
+                )
+
+            # Check data type of target
+            if not pd.api.types.is_numeric_dtype(self.data[target_column]):
+                issues.append(f"Target column must be numeric, found {self.data[target_column].dtype}")
+
+            # Try to parse date column
+            try:
+                _ = pd.to_datetime(self.data[date_column])
+            except Exception as e:
+                issues.append(f"Cannot parse date column: {str(e)}")
+
+            if issues:
+                return {'status': 'error', 'issues': issues, 'warnings': warnings}
+
+            return {
+                'status': 'success',
+                'warnings': warnings,
+                'missing_pct': missing_pct
+            }
+
+        except Exception as e:
+            logger.error(f"Validation failed: {str(e)}", exc_info=True)
+            return {'status': 'error', 'issues': [str(e)]}
+
+    def preprocess(
+        self,
+        date_column: str,
+        target_column: any,  # Can be string or list of strings
+        id_column: Optional[str] = None,
+        forecast_horizon: int = 30,
+        max_rows: int = 100000
+    ) -> Dict[str, Any]:
+        """
+        Complete preprocessing pipeline
+
+        Args:
+            date_column: Name of the date column
+            target_column: Name of the target column (string) or list of target columns for multivariate
+            id_column: Optional ID column
+            forecast_horizon: Number of periods to forecast
+
+        Returns:
+            Processed data and metadata
+        """
+        try:
+            logger.info("Starting preprocessing pipeline")
+
+            # Step 0: Handle very large datasets
+            original_row_count = len(self.data)
+            if original_row_count > max_rows:
+                logger.warning(f"Dataset has {original_row_count} rows, sampling to {max_rows} for performance")
+                # Keep the most recent data for forecasting
+                self.data = self.data.tail(max_rows).reset_index(drop=True)
+
+            # Step 1: Parse dates
+            logger.info("Parsing dates...")
+            self.data[date_column] = pd.to_datetime(self.data[date_column])
+
+            # Step 2: Sort by date and remove duplicate timestamps
+            self.data = self.data.sort_values(date_column).reset_index(drop=True)
+
+            # Check for and handle duplicate timestamps
+            duplicate_count = self.data[date_column].duplicated().sum()
+            if duplicate_count > 0:
+                logger.warning(f"Found {duplicate_count} duplicate timestamps, keeping first occurrence")
+                self.data = self.data.drop_duplicates(subset=[date_column], keep='first').reset_index(drop=True)
+
+            # Step 3: Detect frequency
+            logger.info("Detecting frequency...")
+            freq = pd.infer_freq(self.data[date_column])
+            if freq is None:
+                # Try to infer from differences
+                diffs = self.data[date_column].diff().dropna()
+                if len(diffs) > 0:
+                    # Get the most common time difference
+                    mode_diff = diffs.mode()
+                    if len(mode_diff) > 0 and mode_diff[0] != pd.Timedelta(0):
+                        # Convert Timedelta to frequency string
+                        td = mode_diff[0]
+                        freq = self._timedelta_to_freq_string(td)
+                        logger.warning(f"Could not auto-detect frequency, inferred from mode: {freq}")
+                    else:
+                        freq = 'D'
+                        logger.warning("Using default frequency: Daily")
+                else:
+                    freq = 'D'
+                    logger.warning("Using default frequency: Daily")
+
+            # Step 4: Handle missing values in target(s)
+            # Normalize target_column to list
+            target_columns = [target_column] if isinstance(target_column, str) else target_column
+            logger.info(f"Processing {len(target_columns)} target column(s): {target_columns}")
+
+            logger.info("Handling missing values...")
+            total_missing_count = 0
+
+            for tcol in target_columns:
+                missing_count = self.data[tcol].isna().sum()
+                total_missing_count += missing_count
+
+                if missing_count > 0:
+                    # Forward fill for small gaps
+                    self.data[tcol] = self.data[tcol].ffill(limit=5)
+
+                    # Linear interpolation for remaining
+                    self.data[tcol] = self.data[tcol].interpolate(method='linear')
+
+                    # Final fallback: backward fill
+                    self.data[tcol] = self.data[tcol].bfill()
+
+                    logger.info(f"Filled {missing_count} missing values in '{tcol}'")
+
+            # Step 5: Detect outliers (IQR method) - only for primary target
+            logger.info("Detecting outliers...")
+            primary_target = target_columns[0]
+            Q1 = self.data[primary_target].quantile(0.25)
+            Q3 = self.data[primary_target].quantile(0.75)
+            IQR = Q3 - Q1
+            outlier_mask = (
+                (self.data[primary_target] < (Q1 - 3 * IQR)) |
+                (self.data[primary_target] > (Q3 + 3 * IQR))
+            )
+            outlier_count = outlier_mask.sum()
+
+            # Step 6: Check if sufficient data
+            min_required = forecast_horizon * MIN_DATA_POINTS_MULTIPLIER
+            if len(self.data) < min_required:
+                return {
+                    'status': 'error',
+                    'error': f'Insufficient data. Need at least {min_required} points for {forecast_horizon}-period forecast.'
+                }
+
+            # Step 7: Prepare for Chronos 2 format
+            # Chronos 2 expects columns: ['id', 'timestamp', 'target']
+            # For multivariate: ['id', 'timestamp', 'target', 'covariate1', 'covariate2', ...]
+            processed_df = pd.DataFrame({
+                'id': self.data[id_column] if id_column else 'series_1',
+                'timestamp': self.data[date_column],
+                'target': self.data[target_columns[0]].astype(float)
+            })
+
+            # Add additional target columns as covariates
+            if len(target_columns) > 1:
+                logger.info(f"Adding {len(target_columns)-1} additional target column(s) as covariates")
+                for tcol in target_columns[1:]:
+                    processed_df[tcol] = self.data[tcol].astype(float)
+
+            # Generate quality report
+            quality_report = {
+                'total_points': len(processed_df),
+                'original_points': original_row_count,
+                'sampled': original_row_count > max_rows,
+                'date_range': {
+                    'start': processed_df['timestamp'].min().strftime('%Y-%m-%d'),
+                    'end': processed_df['timestamp'].max().strftime('%Y-%m-%d')
+                },
+                'frequency': str(freq),
+                'missing_filled': total_missing_count,
+                'outliers_detected': outlier_count,
+                'duplicates_removed': duplicate_count if duplicate_count > 0 else 0,
+                'target_columns': target_columns,
+                'statistics': {
+                    'mean': float(processed_df['target'].mean()),
+                    'std': float(processed_df['target'].std()),
+                    'min': float(processed_df['target'].min()),
+                    'max': float(processed_df['target'].max())
+                }
+            }
+
+            logger.info("Preprocessing completed successfully")
+
+            return {
+                'status': 'success',
+                'data': processed_df,
+                'quality_report': quality_report,
+                'frequency': freq
+            }
+
+        except Exception as e:
+            logger.error(f"Preprocessing failed: {str(e)}", exc_info=True)
+            return {
+                'status': 'error',
+                'error': str(e)
+            }
+
+    def get_column_info(self) -> Dict[str, List[str]]:
+        """
+        Get information about columns for UI dropdowns
+
+        Returns:
+            Dictionary with potential date and numeric columns
+        """
+        if self.data is None:
+            return {'date_columns': [], 'numeric_columns': [], 'all_columns': []}
+
+        date_columns = []
+        numeric_columns = []
+
+        for col in self.data.columns:
+            # Check if column could be a date
+            if self.data[col].dtype == 'object':
+                # Try to parse a sample
+                try:
+                    pd.to_datetime(self.data[col].iloc[:5])
+                    date_columns.append(col)
+                except:
+                    pass
+            elif pd.api.types.is_datetime64_any_dtype(self.data[col]):
+                date_columns.append(col)
+
+            # Check if column is numeric
+            if pd.api.types.is_numeric_dtype(self.data[col]):
+                numeric_columns.append(col)
+
+        return {
+            'date_columns': date_columns,
+            'numeric_columns': numeric_columns,
+            'all_columns': list(self.data.columns)
+        }
+
+    def get_preview(self, n_rows: int = 10) -> pd.DataFrame:
+        """
+        Get a preview of the data
+
+        Args:
+            n_rows: Number of rows to return
+
+        Returns:
+            DataFrame preview
+        """
+        if self.data is None:
+            return pd.DataFrame()
+
+        return self.data.head(n_rows)
+
+
+# Global data processor instance
+data_processor = DataProcessor()

services/model_service.py

@@ -0,0 +1,430 @@
+"""
+Chronos 2 Model Service
+Handles model loading, caching, and inference using Chronos2Pipeline
+"""
+
+import logging
+import time
+from typing import Dict, List, Optional, Tuple, Any
+import numpy as np
+import pandas as pd
+import torch
+from chronos import ChronosPipeline, Chronos2Pipeline
+
+from config.constants import CHRONOS2_MODEL, CONFIDENCE_LEVELS
+from config.settings import CONFIG, DEVICE, MODEL_CONFIG
+
+logger = logging.getLogger(__name__)
+
+
+class ChronosModelService:
+    """
+    Service for managing Chronos 2 model lifecycle and inference
+    Uses Chronos2Pipeline with DataFrame-based API
+    """
+
+    def __init__(self):
+        self.model = None
+        self.device = None
+        self.model_variant = None
+        self.is_loaded = False
+        self.load_time = None
+        self.is_chronos2 = False  # Track which pipeline type is loaded
+
+    def _get_device(self) -> str:
+        """Determine the best available device"""
+        if DEVICE == 'cuda':
+            if not torch.cuda.is_available():
+                logger.warning("CUDA requested but not available, falling back to CPU")
+                return 'cpu'
+            return 'cuda'
+        elif DEVICE == 'cpu':
+            return 'cpu'
+        else:  # auto
+            return 'cuda' if torch.cuda.is_available() else 'cpu'
+
+    def load_model(self) -> Dict[str, Any]:
+        """
+        Load the Chronos 2 model at startup
+
+        Returns:
+            Dictionary with loading status and metadata
+        """
+        try:
+            start_time = time.time()
+            logger.info("Loading Chronos 2 model from HuggingFace paper 2510.15821")
+
+            # Use the single Chronos-2 model
+            model_path = CHRONOS2_MODEL
+            self.model_variant = 'chronos-2'
+
+            # Determine device
+            self.device = self._get_device()
+            logger.info(f"Using device: {self.device}")
+
+            # Load model using Chronos2Pipeline
+            self.model = Chronos2Pipeline.from_pretrained(
+                model_path,
+                device_map=self.device,
+                torch_dtype=torch.bfloat16 if self.device == 'cuda' else torch.float32,
+            )
+            self.is_chronos2 = True
+
+            self.load_time = time.time() - start_time
+            self.is_loaded = True
+
+            logger.info(f"Model loaded successfully in {self.load_time:.2f}s")
+
+            # Warmup prediction
+            if MODEL_CONFIG['warmup_enabled']:
+                self._warmup()
+
+            return {
+                'status': 'success',
+                'model': 'chronos-2',
+                'device': self.device,
+                'load_time': self.load_time,
+                'model_name': model_path
+            }
+
+        except Exception as e:
+            logger.error(f"Failed to load model: {str(e)}", exc_info=True)
+            self.is_loaded = False
+            return {
+                'status': 'error',
+                'error': str(e)
+            }
+
+    def _warmup(self):
+        """Run a warmup prediction to initialize the model"""
+        try:
+            logger.info("Running warmup prediction")
+
+            # Create warmup DataFrame in Chronos 2 format
+            warmup_data = pd.DataFrame({
+                'id': ['warmup'] * MODEL_CONFIG['warmup_length'],
+                'timestamp': pd.date_range('2020-01-01', periods=MODEL_CONFIG['warmup_length'], freq='D'),
+                'target': np.random.randn(MODEL_CONFIG['warmup_length'])
+            })
+
+            self.predict(
+                warmup_data,
+                horizon=MODEL_CONFIG['warmup_horizon'],
+                confidence_levels=[80]
+            )
+            logger.info("Warmup completed successfully")
+
+        except Exception as e:
+            logger.warning(f"Warmup failed: {str(e)}")
+
+    def predict(
+        self,
+        data: pd.DataFrame,
+        horizon: int,
+        confidence_levels: List[int] = None,
+        future_df: Optional[pd.DataFrame] = None
+    ) -> Dict[str, Any]:
+        """
+        Generate forecasts using Chronos 2 model with DataFrame API
+
+        Args:
+            data: DataFrame with columns ['id', 'timestamp', 'target']
+                  Can also include covariates for multivariate forecasting
+            horizon: Number of periods to forecast
+            confidence_levels: List of confidence levels (e.g., [80, 90, 95])
+            future_df: Optional DataFrame with future covariate values
+
+        Returns:
+            Dictionary with predictions and metadata
+        """
+        logger.info("=" * 80)
+        logger.info("MODEL SERVICE: predict() - ENTRY")
+        logger.info(f"Data shape: {data.shape}")
+        logger.info(f"Data columns: {data.columns.tolist()}")
+        logger.info(f"Horizon: {horizon}")
+        logger.info(f"Confidence levels: {confidence_levels}")
+        logger.info(f"Is loaded: {self.is_loaded}")
+        logger.info("=" * 80)
+
+        if not self.is_loaded:
+            logger.error("✗ Model not loaded!")
+            raise RuntimeError("Model not loaded. Call load_model() first.")
+
+        try:
+            start_time = time.time()
+            logger.info("Starting prediction...")
+
+            # Use default confidence levels if not provided
+            if confidence_levels is None:
+                confidence_levels = CONFIDENCE_LEVELS
+
+            # Calculate quantile levels from confidence intervals
+            quantile_levels = []
+            for cl in sorted(confidence_levels):
+                lower = (100 - cl) / 200  # e.g., 80% -> 0.10
+                upper = 1 - lower  # e.g., 80% -> 0.90
+                quantile_levels.extend([lower, upper])
+
+            # Add median
+            quantile_levels.append(0.5)
+            quantile_levels = sorted(set(quantile_levels))
+
+            logger.info(f"Generating forecast for horizon={horizon}, quantiles={quantile_levels}")
+
+            # Ensure required columns exist
+            required_cols = ['id', 'timestamp', 'target']
+            logger.info(f"Checking for required columns: {required_cols}")
+            if not all(col in data.columns for col in required_cols):
+                error_msg = f"Data must contain columns: {required_cols}, but got: {data.columns.tolist()}"
+                logger.error(f"✗ {error_msg}")
+                raise ValueError(error_msg)
+            logger.info("✓ All required columns present")
+
+            # Generate forecast using appropriate API
+            if self.is_chronos2:
+                logger.info("Using Chronos2Pipeline.predict_df() method")
+                logger.info(f"Calling predict_df with prediction_length={horizon}, quantile_levels={quantile_levels}")
+                # Use Chronos 2 DataFrame API
+                pred_df = self.model.predict_df(
+                    df=data,
+                    future_df=future_df,
+                    prediction_length=horizon,
+                    quantile_levels=quantile_levels,
+                    id_column='id',
+                    timestamp_column='timestamp',
+                    target='target'
+                )
+                logger.info(f"✓ predict_df completed - result shape: {pred_df.shape}")
+            else:
+                # Use original Chronos tensor API
+                # Convert DataFrame to tensor
+                context_tensor = torch.tensor(data['target'].values, dtype=torch.float32).unsqueeze(0)
+
+                # Generate forecast
+                forecast_tensors = self.model.predict(
+                    context=context_tensor,
+                    prediction_length=horizon,
+                    num_samples=20,  # Number of sample paths
+                    limit_prediction_length=False
+                )
+
+                # Convert tensor output to DataFrame format
+                # forecast_tensors shape: [batch, num_samples, prediction_length]
+                quantiles_np = np.quantile(
+                    forecast_tensors.squeeze(0).numpy(),
+                    q=quantile_levels,
+                    axis=0
+                )
+
+                # Create prediction DataFrame in Chronos 2 format
+                last_timestamp = pd.to_datetime(data['timestamp'].iloc[-1])
+                freq = pd.infer_freq(pd.to_datetime(data['timestamp']))
+                if freq is None:
+                    freq = 'D'  # Default to daily
+
+                future_timestamps = pd.date_range(
+                    start=last_timestamp,
+                    periods=horizon + 1,
+                    freq=freq
+                )[1:]  # Exclude the last historical point
+
+                pred_df = pd.DataFrame({
+                    'id': [data['id'].iloc[0]] * horizon,
+                    'timestamp': future_timestamps
+                })
+
+                # Add quantile columns
+                for i, q in enumerate(quantile_levels):
+                    pred_df[f'{q:.2f}'] = quantiles_np[i, :]
+
+            # Process forecast results
+            # pred_df contains columns: id, timestamp, and quantile columns
+
+            # Extract forecast for the first series (if multiple)
+            series_ids = pred_df['id'].unique()
+            if len(series_ids) > 0:
+                series_pred = pred_df[pred_df['id'] == series_ids[0]].copy()
+            else:
+                series_pred = pred_df.copy()
+
+            # Create forecast dataframe with confidence intervals
+            forecast_df = pd.DataFrame({
+                'ds': series_pred['timestamp'],
+                'forecast': series_pred['0.5']  # Median forecast
+            })
+
+            # Add confidence intervals
+            for cl in confidence_levels:
+                lower = (100 - cl) / 200
+                upper = 1 - lower
+
+                lower_col = f'{lower:.2f}'
+                upper_col = f'{upper:.2f}'
+
+                if lower_col in series_pred.columns:
+                    forecast_df[f'lower_{cl}'] = series_pred[lower_col].values
+                if upper_col in series_pred.columns:
+                    forecast_df[f'upper_{cl}'] = series_pred[upper_col].values
+
+            inference_time = time.time() - start_time
+
+            logger.info(f"✓ Forecast generated successfully in {inference_time:.2f}s")
+            logger.info(f"Returning forecast DataFrame with {len(forecast_df)} rows")
+            logger.info("MODEL SERVICE: predict() - EXIT (success)")
+            logger.info("=" * 80)
+
+            return {
+                'status': 'success',
+                'forecast': forecast_df,
+                'inference_time': inference_time,
+                'horizon': horizon,
+                'confidence_levels': confidence_levels,
+                'full_prediction': pred_df  # Include full prediction for multivariate
+            }
+
+        except Exception as e:
+            logger.error(f"✗ EXCEPTION in predict(): {str(e)}", exc_info=True)
+            logger.info("MODEL SERVICE: predict() - EXIT (exception)")
+            logger.info("=" * 80)
+            return {
+                'status': 'error',
+                'error': str(e)
+            }
+
+    def backtest(
+        self,
+        data: pd.DataFrame,
+        test_size: int,
+        forecast_horizon: int,
+        confidence_levels: List[int] = None
+    ) -> Dict[str, Any]:
+        """
+        Perform backtesting on historical data to evaluate model performance
+
+        Args:
+            data: DataFrame with columns ['id', 'timestamp', 'target']
+            test_size: Number of periods to use for testing
+            forecast_horizon: Forecast horizon for each prediction
+            confidence_levels: List of confidence levels
+
+        Returns:
+            Dictionary with backtest results including predictions vs actuals
+        """
+        logger.info("=" * 80)
+        logger.info("MODEL SERVICE: backtest() - ENTRY")
+        logger.info(f"Data shape: {data.shape}")
+        logger.info(f"Test size: {test_size}")
+        logger.info(f"Forecast horizon: {forecast_horizon}")
+        logger.info("=" * 80)
+
+        if not self.is_loaded:
+            raise RuntimeError("Model not loaded. Call load_model() first.")
+
+        try:
+            start_time = time.time()
+
+            # Split data into train and test
+            train_size = len(data) - test_size
+            if train_size < forecast_horizon * 2:
+                raise ValueError(f"Insufficient training data. Need at least {forecast_horizon * 2} points.")
+
+            # Use rolling window approach
+            # We'll make predictions for the test period using the training data
+            train_data = data.iloc[:train_size].copy()
+            test_data = data.iloc[train_size:].copy()
+
+            logger.info(f"Train size: {len(train_data)}, Test size: {len(test_data)}")
+
+            # Make prediction on test period
+            forecast_result = self.predict(
+                data=train_data,
+                horizon=test_size,
+                confidence_levels=confidence_levels
+            )
+
+            if forecast_result['status'] == 'error':
+                return forecast_result
+
+            forecast_df = forecast_result['forecast']
+
+            # Align forecast with actual values
+            backtest_df = pd.DataFrame({
+                'timestamp': test_data['timestamp'].values,
+                'actual': test_data['target'].values,
+                'predicted': forecast_df['forecast'].values[:len(test_data)]
+            })
+
+            # Add confidence intervals if available
+            for cl in (confidence_levels or []):
+                lower_col = f'lower_{cl}'
+                upper_col = f'upper_{cl}'
+                if lower_col in forecast_df.columns:
+                    backtest_df[lower_col] = forecast_df[lower_col].values[:len(test_data)]
+                if upper_col in forecast_df.columns:
+                    backtest_df[upper_col] = forecast_df[upper_col].values[:len(test_data)]
+
+            # Calculate metrics
+            actual = backtest_df['actual'].values
+            predicted = backtest_df['predicted'].values
+
+            # Remove any NaN values
+            mask = ~(np.isnan(actual) | np.isnan(predicted))
+            actual = actual[mask]
+            predicted = predicted[mask]
+
+            if len(actual) == 0:
+                raise ValueError("No valid data points for metric calculation")
+
+            mae = np.mean(np.abs(actual - predicted))
+            rmse = np.sqrt(np.mean((actual - predicted) ** 2))
+            mape = np.mean(np.abs((actual - predicted) / (actual + 1e-10))) * 100
+
+            # R-squared
+            ss_res = np.sum((actual - predicted) ** 2)
+            ss_tot = np.sum((actual - np.mean(actual)) ** 2)
+            r2 = 1 - (ss_res / (ss_tot + 1e-10))
+
+            metrics = {
+                'MAE': float(mae),
+                'RMSE': float(rmse),
+                'MAPE': float(mape),
+                'R2': float(r2)
+            }
+
+            inference_time = time.time() - start_time
+
+            logger.info(f"✓ Backtest completed in {inference_time:.2f}s")
+            logger.info(f"Metrics: MAE={mae:.2f}, RMSE={rmse:.2f}, MAPE={mape:.2f}%, R2={r2:.4f}")
+            logger.info("MODEL SERVICE: backtest() - EXIT (success)")
+            logger.info("=" * 80)
+
+            return {
+                'status': 'success',
+                'backtest_data': backtest_df,
+                'metrics': metrics,
+                'inference_time': inference_time,
+                'train_size': train_size,
+                'test_size': test_size
+            }
+
+        except Exception as e:
+            logger.error(f"✗ EXCEPTION in backtest(): {str(e)}", exc_info=True)
+            logger.info("MODEL SERVICE: backtest() - EXIT (exception)")
+            logger.info("=" * 80)
+            return {
+                'status': 'error',
+                'error': str(e)
+            }
+
+    def get_status(self) -> Dict[str, Any]:
+        """Get current model status"""
+        return {
+            'is_loaded': self.is_loaded,
+            'variant': self.model_variant,
+            'device': self.device,
+            'load_time': self.load_time
+        }
+
+
+# Global model service instance
+model_service = ChronosModelService()