import streamlit as st import pandas as pd import numpy as np import yfinance as yf from sklearn.ensemble import RandomForestRegressor from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import plotly.graph_objects as go from plotly.subplots import make_subplots import ta from datetime import datetime, timedelta import time from yahooquery import Screener st.set_page_config(page_title="Crypto Price Prediction App", layout="wide") @st.cache_data(ttl=1800) def get_available_crypto_tickers(): try: s = Screener() data = s.get_screeners('all_cryptocurrencies_us', count=250) return [item['symbol'] for item in data['all_cryptocurrencies_us']['quotes']] except Exception as e: st.warning(f"Failed to fetch available tickers: {e}") return [] @st.cache_data(ttl=1800) def fetch_crypto_data_robust(symbol, start_date, end_date): available_tickers = get_available_crypto_tickers() tickers_to_try = [symbol + "-USD"] + [t for t in available_tickers if symbol in t] for ticker in tickers_to_try: try: data = yf.download( ticker, start=start_date, end=end_date, progress=False, timeout=45 ) if not data.empty and len(data) > 20: required_columns = ['Open', 'High', 'Low', 'Close', 'Volume'] if all(col in data.columns for col in required_columns): return data, ticker except: continue st.error(f"❌ Could not fetch data for {symbol} after multiple attempts") return pd.DataFrame(), None def test_yfinance_connection(): try: test_data = yf.download("BTC-USD", period="2d", interval="60m", progress=False) return not test_data.empty except: return False @st.cache_data def add_technical_indicators(df): if df.empty: return df df = df.copy() df['SMA_20'] = df['Close'].rolling(window=20).mean() df['EMA_20'] = df['Close'].ewm(span=20).mean() delta = df['Close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss df['RSI'] = 100 - (100 / (1 + rs)) ema12 = df['Close'].ewm(span=12).mean() ema26 = df['Close'].ewm(span=26).mean() df['MACD'] = ema12 - ema26 df['MACD_Signal'] = df['MACD'].ewm(span=9).mean() sma = df['Close'].rolling(window=20).mean() std = df['Close'].rolling(window=20).std() df['Bollinger_High'] = sma + (std * 2) df['Bollinger_Low'] = sma - (std * 2) return df def prepare_features(df): if df.empty: return pd.DataFrame(), pd.Series() df = df.copy() df['Returns'] = df['Close'].pct_change() df['Target'] = df['Returns'].shift(-1) features = ['Open', 'High', 'Low', 'Close', 'Volume', 'Returns', 'SMA_20', 'EMA_20', 'RSI', 'MACD', 'MACD_Signal', 'Bollinger_High', 'Bollinger_Low'] df.dropna(inplace=True) X = df[features] y = df['Target'] return X, y def train_model(X, y): split_idx = int(len(X) * 0.8) X_train, X_test = X.iloc[:split_idx], X.iloc[split_idx:] y_train, y_test = y.iloc[:split_idx], y.iloc[split_idx:] model = RandomForestRegressor(n_estimators=100, max_depth=15, random_state=42, n_jobs=-1) model.fit(X_train, y_train) return model, X_test, y_test def create_analysis_chart(df, ticker): plot_df = df.iloc[-300:] fig = make_subplots( rows=3, cols=1, shared_xaxes=True, vertical_spacing=0.05, row_heights=[0.6, 0.2, 0.2], subplot_titles=[f'Price Chart ({ticker})', 'MACD', 'RSI'] ) fig.add_trace(go.Candlestick(x=plot_df.index, open=plot_df['Open'], high=plot_df['High'], low=plot_df['Low'], close=plot_df['Close'], name='Price'), row=1, col=1) fig.add_trace(go.Scatter(x=plot_df.index, y=plot_df['MACD'], name='MACD', line=dict(color='blue')), row=2, col=1) fig.add_trace(go.Scatter(x=plot_df.index, y=plot_df['MACD_Signal'], name='Signal Line', line=dict(color='orange')), row=2, col=1) fig.add_trace(go.Scatter(x=plot_df.index, y=plot_df['RSI'], name='RSI', line=dict(color='purple')), row=3, col=1) fig.add_hline(y=70, line_dash="dash", line_color="red", row=3, col=1) fig.add_hline(y=30, line_dash="dash", line_color="green", row=3, col=1) fig.update_layout( height=800, xaxis_rangeslider_visible=False, hovermode='x unified', yaxis=dict(title='Price', tickformat=',.0f', autorange=True) ) return fig def main(): st.title("🚀 Crypto Price Prediction App") st.markdown("**Advanced ML-powered cryptocurrency analysis**") with st.sidebar: st.header("⚙️ Configuration") if st.button("🔍 Test Data Connection"): if test_yfinance_connection(): st.success("✅ Connection successful") else: st.error("❌ Connection failed - check internet") crypto_symbol = st.selectbox("Cryptocurrency", ["BTC", "ETH", "ADA", "XRP", "DOT"], index=0) start_date = st.date_input("Start Date", datetime.now() - timedelta(days=180), max_value=datetime.now() - timedelta(days=7)) end_date = st.date_input("End Date", datetime.now(), min_value=start_date + timedelta(days=7)) if st.sidebar.button("📊 Run Full Analysis", type="primary"): with st.spinner("Analyzing cryptocurrency data..."): try: data, ticker = fetch_crypto_data_robust(crypto_symbol, start_date, end_date) if data.empty: raise ValueError("No data retrieved") data = add_technical_indicators(data) X, y = prepare_features(data) model, X_test, y_test = train_model(X, y) predictions = model.predict(X_test) mse = mean_squared_error(y_test, predictions) mae = mean_absolute_error(y_test, predictions) r2 = r2_score(y_test, predictions) current_price = float(data['Close'].iloc[-1]) prediction_arr = model.predict(X.iloc[[-1]]) prediction = float(prediction_arr[0]) if isinstance(prediction_arr, (list, np.ndarray, pd.Series)) else float(prediction_arr) st.subheader("📊 Market Analysis") col1, col2 = st.columns(2) with col1: st.plotly_chart(create_analysis_chart(data, ticker), use_container_width=True) with col2: st.metric("Current Price", f"${current_price:,.2f}") st.metric("Predicted Next Period Return", f"{prediction*100:.2f}%") st.write("**Model Performance**") st.metric("R² Score", f"{r2:.4f}") st.metric("MAE", f"{mae:.4f}") st.metric("MSE", f"{mse:.6f}") st.write("**Technical Signals**") if 'RSI' in data.columns: rsi = data['RSI'].iloc[-1] if rsi > 70: st.warning("RSI Overbought (Above 70)") elif rsi < 30: st.success("RSI Oversold (Below 30)") if prediction > 0.05: st.success("Strong Buy Signal") elif prediction < -0.05: st.error("Strong Sell Signal") st.markdown("---") st.subheader("🧠 Automated Insight") analysis_text = f""" The model predicts a {prediction*100:.2f}% return for the next period. The RSI is currently {rsi:.2f}, which is {'overbought' if rsi > 70 else 'oversold' if rsi < 30 else 'neutral'}. Model confidence is measured by an R² score of {r2:.4f}. """ st.info(analysis_text) st.warning("**Disclaimer:** This is not financial advice. Cryptocurrency markets are highly volatile.") except Exception as e: st.error(f"Analysis failed: {str(e)}") if __name__ == "__main__": main()