Spaces:
Sleeping
Sleeping
| """ | |
| 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() | |