""" 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()