Spaces:
Sleeping
Sleeping
| import numpy as np | |
| import pandas as pd | |
| import logging | |
| import matplotlib.pyplot as plt | |
| from statsmodels.tsa.arima.model import ARIMA | |
| import yfinance as yf | |
| import gradio as gr | |
| logging.basicConfig(level=logging.INFO) | |
| def fetch_data(period='1d'): | |
| logging.info(f"Fetching data for the period {period}...") | |
| data = yf.download(tickers='ETH-USD', period=period, interval='1m') | |
| if data.empty: | |
| logging.error("No data fetched. Check the period or ticker symbol.") | |
| return None | |
| logging.info(f"Fetched {len(data)} data points for the period {period}.") | |
| return data | |
| def make_predictions(data, predict_steps, freq): | |
| if data is None or data.empty: | |
| logging.error("No data available for model training.") | |
| return None | |
| logging.info(f"Starting model training with {len(data)} data points...") | |
| # Check for NaN values in the data | |
| if data['Close'].isna().any(): | |
| logging.error("Data contains NaN values. Please clean the data before model training.") | |
| return None | |
| try: | |
| model = ARIMA(data['Close'], order=(5, 1, 0)) | |
| model_fit = model.fit() | |
| logging.info(model_fit.summary()) | |
| except Exception as e: | |
| logging.error(f"Model training failed: {e}") | |
| return None | |
| logging.info("Model training completed.") | |
| logging.info("Generating predictions...") | |
| try: | |
| forecast = model_fit.forecast(steps=predict_steps) | |
| if np.isnan(forecast).any(): | |
| logging.error("Generated predictions contain NaN values. Model might be improperly configured.") | |
| return None | |
| except Exception as e: | |
| logging.error(f"Prediction generation failed: {e}") | |
| return None | |
| future_dates = pd.date_range(start=data.index[-1], periods=predict_steps + 1, freq=freq, inclusive='right') | |
| forecast_df = pd.DataFrame(forecast, index=future_dates[1:], columns=['Prediction']) | |
| logging.info(f"Forecast Data:\n{forecast_df.head()}") | |
| logging.info("Predictions generated successfully.") | |
| return forecast_df | |
| def plot_eth(period='1d'): | |
| data = fetch_data(period) | |
| predict_steps = 5 # Modify as needed | |
| freq = 'T' # 'T' stands for minutes | |
| forecast_df = make_predictions(data, predict_steps, freq) | |
| if forecast_df is None: | |
| logging.error("Failed to generate predictions.") | |
| return None | |
| plt.figure(figsize=(10, 5)) | |
| plt.plot(data['Close'], label='Actual ETH Price') | |
| plt.plot(forecast_df['Prediction'], label='Forecasted ETH Price', linestyle='dotted', color='orange') | |
| plt.title('ETH Price Prediction') | |
| plt.xlabel('Time') | |
| plt.ylabel('Price (USD)') | |
| plt.legend() | |
| plt.grid(True) | |
| plt.tight_layout() | |
| # Save the plot to a file | |
| plot_filename = '/home/user/app/eth_price_prediction.png' | |
| plt.savefig(plot_filename) | |
| logging.info("Plotting completed.") | |
| return plot_filename | |
| def refresh_predictions(period): | |
| plot_filename = plot_eth(period) | |
| if plot_filename is None: | |
| return "Error in generating plot." | |
| return plot_filename | |
| iface = gr.Interface(fn=refresh_predictions, inputs="text", outputs="image", live=True) | |
| iface.launch() | |