app.py

"""

Arvanu Chronos Forecaster - Time Series Prediction API

Part of the Arvanu AI Prediction Ensemble for Premium Tiers



Uses amazon/chronos-bolt-base for fast, accurate probabilistic forecasting

of market odds trajectories.

"""

import gradio as gr
import numpy as np
import torch
from chronos import ChronosBoltPipeline
import json

# Load model on startup (cached)
print("Loading Chronos-Bolt model...")
pipeline = ChronosBoltPipeline.from_pretrained(
    "amazon/chronos-bolt-base",
    device_map="cpu",  # HF free tier is CPU only
    torch_dtype=torch.float32,
)
print("Model loaded successfully!")


def forecast_odds(

    historical_prices: str,

    prediction_horizon: int = 24,

    num_samples: int = 20,

) -> dict:
    """

    Forecast market odds trajectory.

    

    Args:

        historical_prices: JSON array of historical YES prices (0-1 range)

                          e.g., "[0.52, 0.54, 0.55, 0.58, 0.56, ...]"

        prediction_horizon: Number of time steps to forecast (default: 24)

        num_samples: Number of sample trajectories for uncertainty (default: 20)

    

    Returns:

        JSON with forecast, trend analysis, and confidence metrics

    """
    try:
        # Parse input
        if isinstance(historical_prices, str):
            prices = json.loads(historical_prices)
        else:
            prices = list(historical_prices)
        
        if len(prices) < 10:
            return {"error": "Need at least 10 historical data points"}
        
        # Ensure values are in valid range
        prices = [max(0.01, min(0.99, float(p))) for p in prices]
        
        # Convert to tensor
        context = torch.tensor(prices, dtype=torch.float32).unsqueeze(0)
        
        # Generate forecasts
        with torch.no_grad():
            forecasts = pipeline.predict(
                context=context,
                prediction_length=prediction_horizon,
                num_samples=num_samples,
            )
        
        # forecasts shape: (1, num_samples, prediction_horizon)
        forecast_np = forecasts[0].numpy()
        
        # Calculate quantiles
        q10 = np.percentile(forecast_np, 10, axis=0).tolist()
        q50 = np.percentile(forecast_np, 50, axis=0).tolist()  # Median
        q90 = np.percentile(forecast_np, 90, axis=0).tolist()
        
        # Trend analysis
        current_price = prices[-1]
        forecast_end = q50[-1]
        price_change = forecast_end - current_price
        
        # Determine trend
        if price_change > 0.03:
            trend = "strongly_bullish"
            trend_strength = min(1.0, price_change * 10)
        elif price_change > 0.01:
            trend = "bullish"
            trend_strength = min(0.7, price_change * 10)
        elif price_change < -0.03:
            trend = "strongly_bearish"
            trend_strength = min(1.0, abs(price_change) * 10)
        elif price_change < -0.01:
            trend = "bearish"
            trend_strength = min(0.7, abs(price_change) * 10)
        else:
            trend = "neutral"
            trend_strength = 0.3
        
        # Calculate momentum (rate of change)
        if len(prices) >= 5:
            recent_momentum = (prices[-1] - prices[-5]) / 5
        else:
            recent_momentum = 0
        
        # Volatility from forecast spread
        avg_spread = np.mean(np.array(q90) - np.array(q10))
        volatility = float(avg_spread)
        
        # Confidence based on forecast tightness and trend clarity
        # Tighter forecasts = higher confidence
        confidence = max(0.3, min(0.95, 1.0 - (volatility * 2)))
        
        # Adjust confidence based on trend strength
        if trend in ["strongly_bullish", "strongly_bearish"]:
            confidence = min(0.95, confidence * 1.15)
        
        # Direction for ensemble (matches NLP output format)
        if trend in ["bullish", "strongly_bullish"]:
            direction = "YES"
            direction_confidence = 0.5 + (trend_strength * 0.4)
        elif trend in ["bearish", "strongly_bearish"]:
            direction = "NO"
            direction_confidence = 0.5 + (trend_strength * 0.4)
        else:
            # Neutral - slight lean based on momentum
            direction = "YES" if recent_momentum > 0 else "NO"
            direction_confidence = 0.5
        
        return {
            "success": True,
            "forecast": {
                "median": q50,
                "lower_bound": q10,
                "upper_bound": q90,
            },
            "analysis": {
                "trend": trend,
                "trend_strength": round(trend_strength, 3),
                "price_change_predicted": round(price_change, 4),
                "current_price": round(current_price, 4),
                "forecast_end_price": round(forecast_end, 4),
                "momentum": round(recent_momentum, 4),
                "volatility": round(volatility, 4),
            },
            "ensemble_output": {
                "direction": direction,
                "confidence": round(direction_confidence, 3),
                "model_confidence": round(confidence, 3),
            },
            "meta": {
                "model": "chronos-bolt-base",
                "input_length": len(prices),
                "horizon": prediction_horizon,
            }
        }
        
    except Exception as e:
        return {
            "success": False,
            "error": str(e),
        }


def forecast_api(historical_prices: str, prediction_horizon: int = 24) -> str:
    """API endpoint wrapper that returns JSON string"""
    result = forecast_odds(historical_prices, prediction_horizon)
    return json.dumps(result, indent=2)


# Create Gradio interface
with gr.Blocks(title="Arvanu Chronos Forecaster") as demo:
    gr.Markdown("""

    # 🔮 Arvanu Chronos Forecaster

    

    **Time-Series Prediction API for Market Odds**

    

    Part of the Arvanu AI Prediction Ensemble. Uses Amazon's Chronos-Bolt 

    for probabilistic forecasting of market price trajectories.

    

    ## API Usage

    

    ```python

    import requests

    

    response = requests.post(

        "https://mythman-arvanu-chronos.hf.space/api/predict",

        json={

            "data": [

                "[0.52, 0.54, 0.55, 0.58, 0.56, 0.59, 0.61, 0.60, 0.62, 0.64]",

                24  # prediction horizon

            ]

        }

    )

    result = response.json()

    ```

    """)
    
    with gr.Row():
        with gr.Column():
            prices_input = gr.Textbox(
                label="Historical Prices (JSON array)",
                placeholder='[0.52, 0.54, 0.55, 0.58, 0.56, 0.59, 0.61, 0.60, 0.62, 0.64]',
                lines=3,
            )
            horizon_input = gr.Slider(
                minimum=1,
                maximum=48,
                value=24,
                step=1,
                label="Prediction Horizon (time steps)",
            )
            submit_btn = gr.Button("Generate Forecast", variant="primary")
        
        with gr.Column():
            output = gr.JSON(label="Forecast Result")
    
    submit_btn.click(
        fn=forecast_odds,
        inputs=[prices_input, horizon_input],
        outputs=output,
    )
    
    gr.Examples(
        examples=[
            ['[0.52, 0.54, 0.55, 0.58, 0.56, 0.59, 0.61, 0.60, 0.62, 0.64, 0.63, 0.65]', 24],
            ['[0.72, 0.71, 0.69, 0.68, 0.70, 0.67, 0.65, 0.64, 0.63, 0.62, 0.60, 0.58]', 12],
            ['[0.50, 0.51, 0.50, 0.49, 0.50, 0.51, 0.50, 0.50, 0.49, 0.50, 0.51, 0.50]', 24],
        ],
        inputs=[prices_input, horizon_input],
    )

# Launch with API enabled
demo.launch()