# Install required packages # !pip install gradio yfinance beautifulsoup4 requests pandas numpy transformers xgboost scikit-learn python-dotenv # !pip install python-decouple import gradio as gr import yfinance as yf import requests from bs4 import BeautifulSoup import pandas as pd import numpy as np from transformers import AutoModelForSequenceClassification, AutoTokenizer import xgboost as xgb from datetime import datetime, timedelta import json import warnings import os from dotenv import load_dotenv from decouple import config import logging warnings.filterwarnings('ignore') load_dotenv() # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # WhatsApp API Configuration WHATSAPP_API_BASE_URL = os.getenv('WHATSAPP_API_BASE_URL') WHATSAPP_API_KEY = os.getenv('WHATSAPP_API_KEY') WHATSAPP_INSTANCE_NAME = os.getenv('WHATSAPP_INSTANCE_NAME') class ConfigManager: """ Centralized configuration management """ @staticmethod def get_api_config(): """ Retrieve API configurations securely """ return { 'amfi_base_url': config('AMFI_API_URL', default='https://api.mfapi.in/mf'), } class WhatsAppManager: def __init__(self, base_url, api_key): self.base_url = base_url self.api_key = api_key def send_message(self, instance, phone, message): if not self.base_url or not self.api_key or not instance: logging.error("WhatsApp API base URL, key or instance not configured.") return "WhatsApp API base URL, key or instance not configured." headers = { 'Content-Type': 'application/json', 'Authorization': self.api_key } payload = json.dumps({ "phone": phone, "message": message }) try: response = requests.post(f"{self.base_url}/message/sendText/{instance}", headers=headers, data=payload) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logging.error(f"Error sending WhatsApp message: {str(e)}") return f"Error sending message: {str(e)}" class AMFIApi: """ Mutual Fund API Handler with real-time data fetching """ @staticmethod def get_all_mutual_funds(): """ Retrieve comprehensive mutual funds list from AMFI API """ config = ConfigManager.get_api_config() try: response = requests.get(config['amfi_base_url']) if response.status_code == 200: return response.json() else: logging.error("Failed to fetch mutual fund data.") return "Error fetching mutual funds." except Exception as e: logging.error(f"API request error: {str(e)}") return f"Error fetching mutual funds: {str(e)}" @staticmethod def analyze_mutual_fund(scheme_code): """ Fetch real-time mutual fund NAV and analyze returns. """ try: config = ConfigManager.get_api_config() amfi_url = f"{config['amfi_base_url']}/{scheme_code}" response = requests.get(amfi_url) if response.status_code != 200: logging.error("Failed to fetch NAV data.") return None, None, "Failed to fetch live NAV data." fund_data = response.json() if 'data' not in fund_data: logging.error("Invalid fund data received.") return None, None, "Invalid fund data received." nav_data = pd.DataFrame(fund_data['data']) # Data type conversions nav_data['date'] = pd.to_datetime(nav_data['date'], format='%d-%m-%Y') nav_data['nav'] = pd.to_numeric(nav_data['nav'], errors='coerce') # Sort by date nav_data = nav_data.sort_values('date') # Calculate returns latest_nav = nav_data.iloc[-1]['nav'] first_nav = nav_data.iloc[0]['nav'] returns = { 'scheme_name': fund_data.get('meta', {}).get('scheme_name', 'Unknown'), 'current_nav': latest_nav, 'initial_nav': first_nav, 'total_return': ((latest_nav - first_nav) / first_nav) * 100 } return returns, nav_data[['date', 'nav']].rename(columns={'nav': 'NAV'}), None except Exception as e: logging.error(f"Analysis error: {str(e)}") return None, None, f"Analysis error: {str(e)}" def get_stock_data(symbol, period='3y'): try: logging.info(f"Fetching stock data for symbol: {symbol}") stock = yf.Ticker(symbol) logging.info(f"Ticker object created successfully for symbol: {symbol}") hist = stock.history(period=period) if hist.empty: logging.error(f"No stock data available for symbol: {symbol} after fetching history.") return f"No stock data available for symbol: {symbol}" logging.info(f"Successfully fetched stock data for symbol: {symbol}") return hist except Exception as e: logging.error(f"Error fetching stock data for symbol: {symbol}, error: {str(e)}") return f"Error fetching stock data: {str(e)}" def calculate_rsi(data, periods=14): delta = data.diff() gain = (delta.where(delta > 0, 0)).rolling(window=periods).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=periods).mean() rs = gain / loss return 100 - (100 / (1 + rs)) def predict_stock(symbol): df = get_stock_data(symbol) if isinstance(df, str): return df logging.info(f"Dataframe before feature creation for {symbol}: \n{df.head()}") df['SMA_20'] = df['Close'].rolling(window=20).mean() df['SMA_50'] = df['Close'].rolling(window=50).mean() df['RSI'] = calculate_rsi(df['Close']) features = ['SMA_20', 'SMA_50', 'RSI', 'Volume'] X = df[features].dropna() y = df['Close'].shift(-1).dropna() logging.info(f"Dataframe after feature creation: \nX:\n{X.head()}\ny:\n{y.head()}") #Align X and Y X = X.iloc[:len(y)] split = int(len(X) * 0.8) X_train, X_test = X[:split], X[split:] y_train, y_test = y[:split], y[split:] logging.info(f"Data split details: \nTrain Data size: {len(X_train)}\nTest Data Size: {len(X_test)}") if len(X_train) == 0 or len(y_train) == 0: logging.error(f"Insufficient training data for prediction for symbol: {symbol}") return "Insufficient data for prediction." model = xgb.XGBRegressor(objective='reg:squarederror', n_estimators=100) model.fit(X_train, y_train) if not X_test.empty: last_data = X_test.iloc[-1:] prediction = model.predict(last_data)[0] return prediction else: logging.warning(f"No test data available for prediction for symbol: {symbol}") return "No test data available for prediction." def analyze_sentiment(text): tokenizer = AutoTokenizer.from_pretrained("ProsusAI/finbert") model = AutoModelForSequenceClassification.from_pretrained("ProsusAI/finbert") inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True) outputs = model(**inputs) predictions = outputs.logits.softmax(dim=-1) labels = ['negative', 'neutral', 'positive'] return {label: float(pred) for label, pred in zip(labels, predictions[0])} def setup_notifications(phone, stock, mf, sentiment): if not WHATSAPP_API_BASE_URL or not WHATSAPP_API_KEY or not WHATSAPP_INSTANCE_NAME: return "WhatsApp API credentials or instance missing." whatsapp_manager = WhatsAppManager(WHATSAPP_API_BASE_URL, WHATSAPP_API_KEY) try: result = whatsapp_manager.send_message( WHATSAPP_INSTANCE_NAME, phone, "🎉 Welcome to AI Finance Manager!\nYour WhatsApp notifications have been set up successfully." ) alerts = [] if stock: alerts.append("Stock") if mf: alerts.append("Mutual Fund") if sentiment: alerts.append("Sentiment") return f"WhatsApp notifications set up for: {', '.join(alerts)} - {result}" except Exception as e: logging.error(f"Error setting up notifications: {str(e)}") return f"Error setting up notifications: {str(e)}" # Chatbot Function def chatbot_response(user_input): user_input = user_input.lower() if "stock" in user_input: parts = user_input.split() if len(parts) > 1: symbol = parts[-1].upper() prediction = predict_stock(symbol) if isinstance(prediction,str): return prediction else: return f"The predicted next-day closing price for {symbol} is {prediction:.2f}" else: return "Please provide a stock symbol." elif "mutual fund" in user_input: parts = user_input.split() if len(parts) > 2 and parts[1] == "code": scheme_code = parts[-1] mf_returns, mf_nav_history, error = AMFIApi.analyze_mutual_fund(scheme_code) if error: return error else: return f"Mutual Fund Analysis:\nName: {mf_returns.get('scheme_name', 'Unknown')}\nCurrent NAV: {mf_returns.get('current_nav', 'N/A'):.2f}\nTotal Return: {mf_returns.get('total_return', 'N/A'):.2f}%" else: return "Please enter the mutual fund scheme code for analysis (e.g. 'analyze mutual fund code 123456')." elif "sentiment" in user_input: return "Enter the financial news text for sentiment analysis." elif user_input.startswith("analyze sentiment"): text = user_input[len("analyze sentiment"):].strip() if text: sentiment_result = analyze_sentiment(text) if sentiment_result: return f"Sentiment Analysis: {sentiment_result}" else: return "No text provided for sentiment analysis." else: return "Please provide text for sentiment analysis." return "I can help with Stock Analysis, Mutual Funds, and Sentiment Analysis. Please ask your query." # Create Gradio Interface def create_gradio_interface(): with gr.Blocks() as app: gr.Markdown("# AI Finance & Stock Manager with Chat and WhatsApp Alerts") with gr.Tab("Chat"): chat_input = gr.Textbox(label="Ask about Stocks, Mutual Funds, or Sentiment Analysis") chat_output = gr.Textbox(label="AI Response", interactive=False) chat_btn = gr.Button("Ask AI") with gr.Tab("Stock Analysis"): stock_input = gr.Textbox(label="Enter Stock Symbol (e.g., AAPL)") stock_btn = gr.Button("Analyze Stock") stock_output = gr.DataFrame() prediction_output = gr.Number(label="Predicted Next Day Close Price") with gr.Tab("Mutual Fund Analysis"): mf_code = gr.Textbox(label="Enter Scheme Code") mf_analyze_btn = gr.Button("Analyze Fund") # Analysis Outputs mf_returns = gr.JSON(label="Fund Returns") mf_nav_history = gr.DataFrame(label="NAV History") mf_analysis_error = gr.Textbox(label="Error Messages", visible=False) with gr.Tab("WhatsApp Notifications"): phone_input = gr.Textbox(label="WhatsApp Number (with country code)") enable_stock_alerts = gr.Checkbox(label="Stock Alerts") enable_mf_alerts = gr.Checkbox(label="Mutual Fund Alerts") enable_sentiment_alerts = gr.Checkbox(label="Sentiment Alerts") notification_status = gr.Textbox(label="Notification Status", interactive=False) setup_btn = gr.Button("Setup WhatsApp Notifications") with gr.Tab("Sentiment Analysis"): text_input = gr.Textbox(label="Enter financial news or text") sentiment_btn = gr.Button("Analyze Sentiment") sentiment_output = gr.Label() # Event Handlers chat_btn.click( fn=chatbot_response, inputs=chat_input, outputs=chat_output ) stock_btn.click( fn=lambda x: (get_stock_data(x), predict_stock(x)), inputs=stock_input, outputs=[stock_output, prediction_output] ) mf_analyze_btn.click( fn=AMFIApi.analyze_mutual_fund, inputs=mf_code, outputs=[mf_returns,mf_nav_history,mf_analysis_error] ) sentiment_btn.click( fn=analyze_sentiment, inputs=text_input, outputs=sentiment_output ) setup_btn.click( fn=setup_notifications, inputs=[phone_input, enable_stock_alerts, enable_mf_alerts, enable_sentiment_alerts], outputs=notification_status ) return app # Launch the app if __name__ == "__main__": app = create_gradio_interface() app.launch(share=True, debug=True)