Upload 10 files

app/core/config.py

@@ -0,0 +1,21 @@
+from pydantic_settings import BaseSettings
+from typing import List
+import os
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "Amazon Chronos Forecasting API"
+    HOST: str = "0.0.0.0"
+    PORT: int = 8000
+    
+    # Model Configuration
+    MODEL_ID: str = "amazon/chronos-t5-tiny"
+    MODEL_PATH: str = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../chronos-t5-tiny"))
+    MODEL_SOURCE: str = "remote" # 'local' or 'remote'
+    
+    # Lepaskan CORS agar bisa diakses dari mana saja (Vercel)
+    CORS_ORIGINS: List[str] = ["*"]
+
+    class Config:
+        env_file = ".env"
+
+settings = Settings()

app/main.py

@@ -0,0 +1,26 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+from app.routes import forecast
+from app.core.config import settings
+
+app = FastAPI(title=settings.PROJECT_NAME)
+
+# Set up CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=settings.CORS_ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy", "service": settings.PROJECT_NAME}
+
+# Include routes
+app.include_router(forecast.router, prefix="/api", tags=["Forecast"])
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("app.main:app", host=settings.HOST, port=settings.PORT, reload=True)

app/routes/forecast.py

@@ -0,0 +1,32 @@
+from fastapi import APIRouter, HTTPException
+from app.schemas.forecast import ForecastRequest, ForecastResponse
+from app.services.chronos_service import chronos_service
+from app.core.config import settings
+
+router = APIRouter()
+
+@router.post("/forecast", response_model=ForecastResponse)
+async def get_forecast(request: ForecastRequest):
+    try:
+        # Perform prediction
+        result = chronos_service.predict(request.series, request.prediction_length)
+        
+        # Calculate some simple insight
+        last_val = request.series[-1]
+        first_forecast = result["forecast"][0]
+        change = ((first_forecast - last_val) / last_val) * 100
+        trend = "naik" if change > 0 else "turun"
+        insight = f"Tren diprediksi akan {trend} sebesar {abs(change):.1f}% pada langkah pertama."
+
+        return ForecastResponse(
+            context_name=request.context_name,
+            input_length=len(request.series),
+            prediction_length=request.prediction_length,
+            forecast=result["forecast"],
+            lower_bound=result.get("lower_bound"),
+            upper_bound=result.get("upper_bound"),
+            model_name=settings.MODEL_ID,
+            insight=insight
+        )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Forecasting failed: {str(e)}")

app/schemas/forecast.py

@@ -0,0 +1,24 @@
+from pydantic import BaseModel, Field, field_validator
+from typing import List, Optional
+
+class ForecastRequest(BaseModel):
+    series: List[float] = Field(..., min_items=5, description="Historical time series data")
+    prediction_length: int = Field(..., ge=1, le=30, description="Number of steps to forecast")
+    context_name: Optional[str] = Field("Untitled Series", description="Name of the context for the series")
+
+    @field_validator('series')
+    @classmethod
+    def validate_series(cls, v: List[float]) -> List[float]:
+        if not v:
+            raise ValueError("Series cannot be empty")
+        return v
+
+class ForecastResponse(BaseModel):
+    context_name: str
+    input_length: int
+    prediction_length: int
+    forecast: List[float]
+    lower_bound: Optional[List[float]] = None
+    upper_bound: Optional[List[float]] = None
+    model_name: str
+    insight: Optional[str] = None

app/services/chronos_service.py

@@ -0,0 +1,87 @@
+import torch
+from chronos import ChronosPipeline
+from app.core.config import settings
+import numpy as np
+import os
+
+class ChronosService:
+    def __init__(self):
+        self.pipeline = None
+        self.model_name = settings.MODEL_ID
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        self.load_model()
+
+    def load_model(self):
+        try:
+            print(f"Loading model {self.model_name} from {settings.MODEL_SOURCE}...")
+            # Ensure path is absolute for safety
+            model_path = settings.MODEL_PATH
+            if not os.path.exists(model_path) and settings.MODEL_SOURCE == "local":
+                 print(f"Warning: Local model path {model_path} not found. Falling back to remote.")
+                 source = settings.MODEL_ID
+            else:
+                 source = model_path if settings.MODEL_SOURCE == "local" else settings.MODEL_ID
+
+            self.pipeline = ChronosPipeline.from_pretrained(
+                source,
+                device_map=self.device,
+                torch_dtype=torch.bfloat16 if self.device == "cuda" else torch.float32,
+            )
+            print("Model loaded successfully.")
+        except Exception as e:
+            print(f"Error loading model: {e}")
+            self.pipeline = None
+            print("ChronosService will operate in MOCK MODE.")
+
+    def predict(self, series: list, prediction_length: int):
+        if self.pipeline is None:
+            return self._mock_predict(series, prediction_length)
+
+        try:
+            # Chronos expects a 1D or 2D tensor
+            context = torch.tensor(series, dtype=torch.float32)
+            
+            # The model predicts multiple samples
+            forecast = self.pipeline.predict(
+                context,
+                prediction_length,
+                num_samples=20,
+            ) # shape [num_samples, prediction_length] or [1, num_samples, prediction_length]
+            
+            # Forecast is usually [batch, samples, horizon]
+            # Since we pass 1 series, it's [samples, horizon] or [1, samples, horizon]
+            samples = forecast[0] if len(forecast.shape) == 3 else forecast
+            
+            # Calculate median, lower and upper bounds (10th and 90th percentiles)
+            low = np.quantile(samples.numpy(), 0.1, axis=0)
+            median = np.quantile(samples.numpy(), 0.5, axis=0)
+            high = np.quantile(samples.numpy(), 0.9, axis=0)
+            
+            return {
+                "forecast": median.tolist(),
+                "lower_bound": low.tolist(),
+                "upper_bound": high.tolist()
+            }
+        except Exception as e:
+            print(f"Prediction error: {e}")
+            return self._mock_predict(series, prediction_length)
+
+    def _mock_predict(self, series: list, prediction_length: int):
+        last_val = series[-1]
+        # Simple linear trend + some noise for mock
+        forecast = []
+        current = last_val
+        for i in range(1, prediction_length + 1):
+            current = current * (1 + 0.05 * (np.random.random() - 0.2)) # Slight upward trend
+            forecast.append(float(current))
+            
+        low = [v * 0.9 for v in forecast]
+        high = [v * 1.1 for v in forecast]
+        return {
+            "forecast": forecast,
+            "lower_bound": low,
+            "upper_bound": high
+        }
+
+# Singleton instance
+chronos_service = ChronosService()