import yfinance as yf import pandas as pd import matplotlib.pyplot as plt import seaborn as sns from prophet import Prophet import gradio as gr from ta.trend import SMAIndicator, EMAIndicator, MACD from ta.momentum import RSIIndicator # Function for trading strategy signals def trading_signals(df, strategy="day trading"): signals = "" df = df.copy() try: if strategy.lower() == "day trading": # Use Bollinger Bands and RSI for intraday trades df['RSI'] = RSIIndicator(df['Close'], window=14).rsi() df['EMA_9'] = EMAIndicator(df['Close'], window=9).ema_indicator() df['EMA_21'] = EMAIndicator(df['Close'], window=21).ema_indicator() df['Signal'] = (df['EMA_9'] > df['EMA_21']).astype(int) signals = f"Day Trading Signals: {'Buy' if df['Signal'].iloc[-1] else 'Sell'}" elif strategy.lower() == "swing trading": # Swing Trading: SMA crossover df['SMA_50'] = SMAIndicator(df['Close'], window=50).sma_indicator() df['SMA_200'] = SMAIndicator(df['Close'], window=200).sma_indicator() df['Signal'] = (df['SMA_50'] > df['SMA_200']).astype(int) signals = f"Swing Trading Signals: {'Bullish' if df['Signal'].iloc[-1] else 'Bearish'}" elif strategy.lower() == "momentum trading": # Momentum Trading: RSI + MACD df['RSI'] = RSIIndicator(df['Close'], window=14).rsi() macd = MACD(df['Close']) df['MACD'] = macd.macd() df['MACD_signal'] = macd.macd_signal() # Simplified momentum condition df['Momentum'] = ((df['RSI'] > 50) & (df['MACD'] > df['MACD_signal'])).astype(int) signals = f"Momentum Trading Signal: {'Strong Momentum' if df['Momentum'].iloc[-1] else 'Weak Momentum'}" elif strategy.lower() == "scalping": # Scalping: EMA crossover on 1-min data df['EMA_3'] = EMAIndicator(df['Close'], window=3).ema_indicator() df['EMA_9'] = EMAIndicator(df['Close'], window=9).ema_indicator() df['Signal'] = (df['EMA_3'] > df['EMA_9']).astype(int) signals = f"Scalping Signal: {'Buy' if df['Signal'].iloc[-1] else 'Sell'}" else: signals = "Invalid Strategy" except Exception as e: signals = f"Error in trading signals: {e}" return signals # Main function to fetch data, generate trends, and trading signals def stock_analysis_with_trading(ticker, period="6mo", interval="1d", analysis_type="closing price trend", strategy=None): try: # Fetch historical stock data with enhanced error handling stock = yf.Ticker(ticker) # Try different methods to fetch data try: df = stock.history(period=period, interval=interval) except Exception as initial_error: # Fallback to different periods or intervals try: df = stock.history(period="1y", interval="1d") except Exception as fallback_error: return f"Error fetching data: {initial_error} | Fallback Error: {fallback_error}", None if df is None or df.empty: return f"No data found for ticker: {ticker}. Check ticker symbol and network connection.", None if analysis_type.lower() == "closing price trend": # Plot Closing Price Trend plt.figure(figsize=(10, 6)) plt.plot(df.index, df['Close'], label='Closing Price') plt.title(f"Closing Price Trend for {ticker}") plt.xlabel("Date") plt.ylabel("Closing Price (USD)") plt.grid() plt.legend() plot_path = f"{ticker}_closing_price_trend.png" plt.savefig(plot_path) plt.close() return f"Closing Price Analysis for {ticker}", plot_path elif analysis_type.lower() == "forecast": # Forecast with Prophet forecast_df = df.reset_index()[['Date', 'Close']] forecast_df.columns = ['ds', 'y'] model = Prophet() model.fit(forecast_df) future = model.make_future_dataframe(periods=30) forecast = model.predict(future) plt.figure(figsize=(10, 6)) model.plot(forecast) plt.title(f"Forecasted Closing Price for {ticker}") plt.xlabel("Date") plt.ylabel("Price (USD)") plot_path = f"{ticker}_forecast_trend.png" plt.savefig(plot_path) plt.close() return "Forecast Generated Successfully", plot_path elif strategy: signals = trading_signals(df, strategy) return signals, None else: return "Invalid analysis type or strategy provided.", None except Exception as e: return f"Comprehensive Error: {e}", None # Gradio Interface def chat_with_agent(ticker, period, interval, query, strategy): response, plot_path = stock_analysis_with_trading(ticker, period, interval, query, strategy) return response, plot_path if plot_path else None # Define Gradio Interface iface = gr.Interface( fn=chat_with_agent, inputs=[ gr.Textbox(label="Stock Ticker (e.g., AAPL, MSFT)"), gr.Textbox(label="Time Period (e.g., 6mo, 1d)", value="6mo"), gr.Textbox(label="Data Interval (e.g., 1d, 1h)", value="1d"), gr.Textbox(label="Query (e.g., closing price trend, forecast)", value="closing price trend"), gr.Textbox(label="Trading Strategy (optional)", value=""), ], outputs=[ gr.Textbox(label="Analysis Result or Trading Signal"), gr.Image(label="Trend or Forecast Visualization") ], title="Stock Analysis & Trading Strategy Agent", description="Analyze stock trends, forecast prices, and generate trading signals using technical indicators." ) # Launch the interface iface.launch(share=True)