Spaces:
Runtime error
Runtime error
| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| """ | |
| app.py - Enhanced Crypto & Stock Predictor with intuitive UI/UX | |
| """ | |
| import ccxt | |
| import yfinance as yf | |
| import pandas as pd | |
| import numpy as np | |
| import plotly.graph_objects as go | |
| from plotly.subplots import make_subplots | |
| import gradio as gr | |
| from ta.momentum import RSIIndicator | |
| from ta.trend import MACD | |
| # ------------------------ | |
| # Data Fetching | |
| # ------------------------ | |
| def get_historical_data(symbol: str, timeframe: str = "1d", limit: int = 200) -> pd.DataFrame: | |
| """Fetch historical OHLCV data using CCXT (crypto) or yfinance (stocks).""" | |
| try: | |
| if "/" in symbol: | |
| exchange = ccxt.binance() | |
| ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit) | |
| df = pd.DataFrame(ohlcv, columns=["timestamp", "Open", "High", "Low", "Close", "Volume"]) | |
| df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') | |
| df.set_index('timestamp', inplace=True) | |
| else: | |
| df = yf.download(symbol, period="1y", interval=timeframe) | |
| df['Returns'] = df['Close'].pct_change().fillna(0) | |
| return df | |
| except Exception as e: | |
| raise Exception(f"Data fetch failed for {symbol}: {str(e)}") | |
| # ------------------------ | |
| # Indicators | |
| # ------------------------ | |
| def add_indicators(df: pd.DataFrame) -> pd.DataFrame: | |
| df['SMA20'] = df['Close'].rolling(20).mean() | |
| df['SMA50'] = df['Close'].rolling(50).mean() | |
| df['Signal'] = np.where(df['SMA20'] > df['SMA50'], 'Buy', 'Sell') | |
| df['RSI'] = RSIIndicator(df['Close'], window=14).rsi() | |
| macd = MACD(df['Close'], window_slow=26, window_fast=12, window_sign=9) | |
| df['MACD'] = macd.macd() | |
| df['MACD_Signal'] = macd.macd_signal() | |
| return df | |
| # ------------------------ | |
| # Chronos AI Prediction | |
| # ------------------------ | |
| def make_chronos_prediction(df: pd.DataFrame, prediction_days: int = 7) -> dict: | |
| """Simulated Chronos AI predictions with confidence intervals.""" | |
| returns = df['Returns'].values | |
| mean_pred = np.random.normal(np.mean(returns), np.std(returns), prediction_days) | |
| std_pred = np.full(prediction_days, np.std(returns)) | |
| mean_pred = mean_pred * df['Close'].std() + df['Close'].mean() | |
| last_price = df['Close'].iloc[-1] | |
| price_predictions = [last_price] | |
| for ret in mean_pred: | |
| price_predictions.append(price_predictions[-1] * (1 + ret)) | |
| price_predictions = np.array(price_predictions[1:]) | |
| upper_bound = price_predictions * (1 + 1.96 * std_pred) | |
| lower_bound = price_predictions * (1 - 1.96 * std_pred) | |
| return {"mean": price_predictions, "upper": upper_bound, "lower": lower_bound} | |
| # ------------------------ | |
| # Plotly Figure | |
| # ------------------------ | |
| def create_prediction_plot(df: pd.DataFrame, predictions: dict, prediction_days: int) -> go.Figure: | |
| fig = make_subplots( | |
| rows=2, cols=1, | |
| subplot_titles=('Price Prediction', 'Technical Indicators'), | |
| vertical_spacing=0.15, | |
| row_width=[0.3, 0.7] | |
| ) | |
| fig.add_trace(go.Scatter(x=df.index, y=df['Close'], name='Historical Price', line=dict(color='blue')), row=1, col=1) | |
| future_dates = pd.date_range(start=df.index[-1] + pd.Timedelta(1, unit='D'), periods=prediction_days) | |
| fig.add_trace(go.Scatter(x=future_dates, y=predictions['mean'], name='Predicted Price', line=dict(color='orange', width=3)), row=1, col=1) | |
| fig.add_trace(go.Scatter( | |
| x=list(future_dates) + list(future_dates[::-1]), | |
| y=list(predictions['upper']) + list(predictions['lower'][::-1]), | |
| fill='toself', fillcolor='rgba(255,127,14,0.2)', | |
| line=dict(color='rgba(255,255,255,0)'), name='Confidence Interval' | |
| ), row=1, col=1) | |
| fig.add_trace(go.Scatter(x=df.index, y=df['RSI'], name='RSI', line=dict(color='green')), row=2, col=1) | |
| fig.add_trace(go.Scatter(x=df.index, y=df['MACD'], name='MACD', line=dict(color='red')), row=2, col=1) | |
| fig.add_trace(go.Scatter(x=df.index, y=df['MACD_Signal'], name='MACD Signal', line=dict(color='purple')), row=2, col=1) | |
| fig.update_layout( | |
| title="📊 Price & Technical Predictions", | |
| height=800, | |
| showlegend=True, | |
| template="plotly_dark" | |
| ) | |
| fig.update_yaxes(title_text="Price ($)", row=1, col=1) | |
| fig.update_yaxes(title_text="Indicator Value", row=2, col=1) | |
| return fig | |
| # ------------------------ | |
| # Gradio Dashboard (Improved UX) | |
| # ------------------------ | |
| def create_dashboard(): | |
| with gr.Blocks(title="📈 Crypto & Stock Predictor") as demo: | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| gr.Markdown("## Input Parameters") | |
| symbol = gr.Textbox(label="Symbol", value="BTC/USDT", placeholder="e.g., BTC/USDT or AAPL", info="Enter crypto pair or stock symbol") | |
| timeframe = gr.Dropdown(label="Timeframe", choices=["1d", "1h", "15m"], value="1d", info="Select candle timeframe") | |
| prediction_days = gr.Slider(label="Prediction Days", minimum=1, maximum=30, value=7, step=1, info="How many days to predict") | |
| btn_predict = gr.Button("Predict", variant="primary") | |
| with gr.Column(scale=2): | |
| with gr.Tab("Predictions"): | |
| plot = gr.Plot(label="Prediction Plot") | |
| with gr.Tab("Signals"): | |
| signals_table = gr.Dataframe(label="SMA Signals", headers=["SMA20","SMA50","Signal"], interactive=False) | |
| with gr.Tab("Market Info"): | |
| market_info = gr.JSON(label="Market Info") | |
| def update_dashboard(symbol, timeframe, prediction_days): | |
| df = get_historical_data(symbol, timeframe) | |
| df = add_indicators(df) | |
| predictions = make_chronos_prediction(df, prediction_days) | |
| fig = create_prediction_plot(df, predictions, prediction_days) | |
| signals_data = df[['SMA20','SMA50','Signal']].tail(prediction_days) | |
| info = { | |
| "Current Price": f"${df['Close'].iloc[-1]:.2f}", | |
| "RSI": f"{df['RSI'].iloc[-1]:.2f}", | |
| "MACD": f"{df['MACD'].iloc[-1]:.2f}", | |
| "MACD Signal": f"{df['MACD_Signal'].iloc[-1]:.2f}" | |
| } | |
| return fig, signals_data, info | |
| return demo | |
| if __name__ == "__main__": | |
| demo = create_dashboard() | |
| demo.launch(server_name="0.0.0.0", server_port=7860) |