Update app.py

app.py

@@ -1,186 +1,149 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """
-app.py
-Crypto & Stock Prediction Dashboard
-Using Chronos AI (ZeroGPU) or Technical Analysis
+app.py - Enhanced Crypto & Stock Predictor with intuitive UI/UX
 """
 
-import numpy as np
-import pandas as pd
+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 Retrieval
+# Data Fetching
 # ------------------------
-def get_historical_data(symbol: str, timeframe: str, period: str = "1y") -> pd.DataFrame:
-    """
-    Fetch historical OHLCV data and compute returns, RSI, MACD.
-    """
-    df = yf.download(symbol, period=period, interval=timeframe)
-    df.dropna(inplace=True)
-    df['Returns'] = df['Close'].pct_change()
-    # Technical indicators
-    df['RSI'] = 100 - (100 / (1 + df['Returns'].rolling(14).mean() / df['Returns'].rolling(14).std()))
-    df['MACD'] = df['Close'].ewm(span=12, adjust=False).mean() - df['Close'].ewm(span=26, adjust=False).mean()
-    df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
-    # Dummy market info columns
-    df['Market_Cap'] = np.nan
-    df['Sector'] = "N/A"
-    df['Industry'] = "N/A"
-    df['Dividend_Yield'] = np.nan
-    df['Annualized_Vol'] = df['Returns'].rolling(252).std() * np.sqrt(252)
-    df['Max_Drawdown'] = (df['Close'].cummax() - df['Close']) / df['Close'].cummax()
-    return df
+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)}")
 
 # ------------------------
-# Prediction Logic
+# Indicators
 # ------------------------
-def make_prediction(symbol: str, timeframe: str, prediction_days: int, strategy: str):
-    try:
-        df = get_historical_data(symbol, timeframe)
-        returns = df['Returns'].dropna()
-        
-        if strategy == "chronos":
-            try:
-                # Use ZeroGPU / CPU-compatible Chronos (mocked)
-                mean_pred = returns[-prediction_days:].values
-                std_pred = returns[-prediction_days:].std() * np.ones_like(mean_pred)
-                
-                actual_len = len(mean_pred)
-                if actual_len < prediction_days:
-                    last_pred = mean_pred[-1]
-                    last_std = std_pred[-1]
-                    extension = last_pred * (1 + np.random.normal(0, last_std, prediction_days - actual_len))
-                    mean_pred = np.concatenate([mean_pred, extension])
-                    std_pred = np.concatenate([std_pred, np.full(prediction_days - actual_len, last_std)])
-
-                # Denormalize predictions
-                mean_pred = mean_pred * returns.std() + returns.mean()
-                
-                # Convert returns to price predictions
-                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:])
-                
-                # Confidence intervals
-                upper_bound = price_predictions * (1 + 1.96 * std_pred)
-                lower_bound = price_predictions * (1 - 1.96 * std_pred)
-
-            except Exception:
-                strategy = "technical"
-
-        if strategy == "technical":
-            last_price = df['Close'].iloc[-1]
-            avg_return = df['Returns'].mean()
-            price_predictions = [last_price * (1 + avg_return) ** i for i in range(1, prediction_days + 1)]
-            price_predictions = np.array(price_predictions)
-            upper_bound = price_predictions * 1.1
-            lower_bound = price_predictions * 0.9
-
-        # Prediction dates
-        last_date = df.index[-1]
-        freq = "D" if timeframe == "1d" else ("H" if timeframe == "1h" else "15T")
-        future_dates = pd.date_range(start=last_date + pd.Timedelta(1, freq=freq), periods=prediction_days, freq=freq)
-
-        predictions = {
-            "dates": future_dates,
-            "mean": price_predictions,
-            "upper": upper_bound,
-            "lower": lower_bound,
-            "strategy": strategy
-        }
-
-        fig = create_prediction_plot(df, predictions, symbol, timeframe)
-        return predictions, fig
+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
 
-    except Exception as e:
-        raise Exception(f"Prediction failed: {str(e)}")
+# ------------------------
+# 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}
 
 # ------------------------
-# Plotting
+# Plotly Figure
 # ------------------------
-def create_prediction_plot(df: pd.DataFrame, predictions: dict, symbol: str, timeframe: str) -> go.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.1,
+        vertical_spacing=0.15,
         row_width=[0.3, 0.7]
     )
-    # Historical price
-    fig.add_trace(go.Scatter(x=df.index, y=df['Close'], name='Historical Price', line=dict(color='#1f77b4')), row=1, col=1)
-    # Predicted mean
-    fig.add_trace(go.Scatter(x=predictions['dates'], y=predictions['mean'], name='Predicted Mean', line=dict(color='#ff7f0e')), row=1, col=1)
-    # Confidence interval
+
+    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=predictions['dates'].tolist() + predictions['dates'].tolist()[::-1],
-        y=predictions['upper'].tolist() + predictions['lower'].tolist()[::-1],
-        fill='toself',
-        fillcolor='rgba(255,127,14,0.2)',
-        line=dict(color='rgba(255,255,255,0)'),
-        name='Confidence Interval'
+        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)
-    # Technical indicators
-    fig.add_trace(go.Scatter(x=df.index, y=df['RSI'], name='RSI', line=dict(color='#2ca02c')), row=2, col=1)
-    fig.add_trace(go.Scatter(x=df.index, y=df['MACD'], name='MACD', line=dict(color='#d62728')), row=2, col=1)
-    fig.add_trace(go.Scatter(x=df.index, y=df['MACD_Signal'], name='MACD Signal', line=dict(color='#9467bd')), row=2, col=1)
 
-    fig.update_layout(height=800, title_text=f"{symbol} Price Prediction ({timeframe}) - Using {predictions['strategy'].upper()} Strategy", showlegend=True)
+    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
 
 # ------------------------
-# Trading Signals (dummy)
-# ------------------------
-def calculate_trading_signals(df: pd.DataFrame) -> dict:
-    signals = {}
-    signals['Buy'] = df.index[df['RSI'] < 30].tolist()
-    signals['Sell'] = df.index[df['RSI'] > 70].tolist()
-    return signals
-
-# ------------------------
-# Gradio Interface
+# Gradio Dashboard (Improved UX)
 # ------------------------
 def create_dashboard():
-    with gr.Blocks(title="Crypto & Stock Predictor") as demo:
-        gr.Markdown("# 📈 Crypto & Stock Prediction Dashboard")
-        gr.Markdown("Predict future prices using Chronos AI or technical analysis")
+    with gr.Blocks(title="📈 Crypto & Stock Predictor") as demo:
         with gr.Row():
             with gr.Column(scale=1):
-                symbol = gr.Textbox(label="Symbol", value="BTC-USD")
-                timeframe = gr.Dropdown(label="Timeframe", choices=["1d", "1h", "15m"], value="1d")
-                prediction_days = gr.Slider(label="Prediction Days", minimum=1, maximum=30, value=7, step=1)
-                strategy = gr.Dropdown(label="Strategy", choices=["chronos", "technical"], value="chronos")
+                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):
-                plot = gr.Plot(label="Prediction Results")
-                info = gr.JSON(label="Market Data")
-                signals = gr.JSON(label="Trading Signals")
-        
-        @btn_predict.click(inputs=[symbol, timeframe, prediction_days, strategy], outputs=[plot, info, signals])
-        def update_dashboard(symbol, timeframe, prediction_days, strategy):
+                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")
+
+        @btn_predict.click(inputs=[symbol, timeframe, prediction_days], outputs=[plot, signals_table, market_info])
+        def update_dashboard(symbol, timeframe, prediction_days):
             df = get_historical_data(symbol, timeframe)
-            predictions, fig = make_prediction(symbol, timeframe, prediction_days, strategy)
-            trading_signals = calculate_trading_signals(df)
-            market_info = {
+            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}",
-                "Market Cap": "N/A",
-                "Sector": df['Sector'].iloc[-1],
-                "Industry": df['Industry'].iloc[-1],
-                "Dividend Yield": "N/A",
-                "Volatility (Annualized)": f"{df['Annualized_Vol'].iloc[-1]*100:.2f}%",
-                "Max Drawdown": f"{df['Max_Drawdown'].iloc[-1]*100:.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, market_info, trading_signals
+
+            return fig, signals_data, info
 
     return demo
 
 if __name__ == "__main__":
     demo = create_dashboard()
-    demo.launch(server_name="0.0.0.0", server_port=7860, share=False)
+    demo.launch(server_name="0.0.0.0", server_port=7860)