Update app.py

app.py

@@ -1,50 +1,62 @@
+import os
 import torch
 import numpy as np
-import pandas as pd
 from chronos import BaseChronosPipeline
 
+# Fix: Point HF cache to a writable path for the non-root HF Spaces user
+os.environ.setdefault("HF_HOME", "/home/user/.cache/huggingface")
+os.environ.setdefault("TRANSFORMERS_CACHE", "/home/user/.cache/huggingface")
+
+
 class SMCPredictionEngine:
     def __init__(self, model_name="amazon/chronos-bolt-base"):
         """
         Initializes the Hugging Face Time-Series Transformer.
-        Defaults to CPU mode for rock-solid stability inside standard containers.
+        Runs on CPU for compatibility with standard HF Spaces containers.
         """
         print(f"Loading pre-trained AI weights: {model_name}...")
         self.pipeline = BaseChronosPipeline.from_pretrained(
             model_name,
             device_map="cpu",
-            torch_dtype=torch.float32
+            torch_dtype=torch.float32,
         )
+        print("Model loaded successfully.")
 
-    def calculate_forecast(self, history_array, prediction_length=12):
+    def calculate_forecast(self, history_array: list | np.ndarray, prediction_length: int = 12) -> dict:
         """
-        Takes an array of historical prices, converts them into a PyTorch tensor,
-        and generates high-probability prediction arrays.
+        Takes an array of historical prices and generates probabilistic forecasts.
+
+        BaseChronosPipeline.predict() returns a tensor of shape:
+            [batch_size, num_samples, prediction_length]
+
+        We derive low / median / high by taking quantiles across the sample axis (axis=1).
         """
-        # Ensure input data is formatted as a 2D float32 tensor: [Batch Size, Sequence Length]
-        context_tensor = torch.tensor([history_array], dtype=torch.float32)
-        
-        # Run inference via Chronos-Bolt
+        context_tensor = torch.tensor(
+            np.array(history_array, dtype=np.float32).reshape(1, -1),  # [1, seq_len]
+            dtype=torch.float32,
+        )
+
+        # forecast shape: [1, num_samples, prediction_length]
         forecast = self.pipeline.predict(
             context=context_tensor,
-            prediction_length=prediction_length
+            prediction_length=prediction_length,
         )
-        
-        # Extract sample paths [Quantiles: 10th percentile, 50th percentile, 90th percentile]
-        quantiles = forecast[0].numpy()
-        
-        low_bounds = quantiles[:, 0]
-        median_predictions = quantiles[:, 1]
-        high_bounds = quantiles[:, 2]
-        
-        # Calculate an Institutional Confirmation Confidence Score
-        # Measures the projected directional momentum against the last known market price
-        last_price = history_array[-1]
-        projected_change = ((median_predictions[-1] - last_price) / last_price) * 100
-        
+
+        # Convert to numpy: shape [num_samples, prediction_length]
+        samples = forecast[0].numpy()  # drop batch dimension
+
+        # Derive quantile bands across the sample axis
+        low_bounds         = np.quantile(samples, 0.10, axis=0)  # 10th percentile
+        median_predictions = np.quantile(samples, 0.50, axis=0)  # median
+        high_bounds        = np.quantile(samples, 0.90, axis=0)  # 90th percentile
+
+        # Projected momentum: % change from last known price to median forecast end
+        last_price = float(history_array[-1])
+        projected_change = ((median_predictions[-1] - last_price) / last_price) * 100.0
+
         return {
             "low": low_bounds,
             "median": median_predictions,
             "high": high_bounds,
-            "projected_change_pct": projected_change
+            "projected_change_pct": float(projected_change),
         }