import gradio as gr import yfinance as yf import pandas as pd import plotly.graph_objects as go from transformers import pipeline from datetime import datetime, timedelta import requests from bs4 import BeautifulSoup import feedparser # ------------------- Constants ------------------- KSE_100 = [ "HBL", "UBL", "MCB", "BAHL", "ABL", "LUCK", "EFERT", "FCCL", "DGKC", "MLCF", "OGDC", "PPL", "POL", "PSO", "SNGP", "ENGRO", "HUBC", "KAPCO", "NESTLE", "EFOODS", "PSX", "TRG", "SYS", "NML", "ILP", "ATRL", "NRL", "HASCOL", "SHEL", "BAFL" ] # Add all KSE-100 tickers # ------------------- Hugging Face Models ------------------- sentiment_analyzer = pipeline("text-classification", model="ProsusAI/finbert") news_summarizer = pipeline("summarization", model="facebook/bart-large-cnn") # ------------------- Technical Analysis ------------------- def calculate_rsi(data, window=14): delta = data['Close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=window).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean() rs = gain / loss return 100 - (100 / (1 + rs)) # ------------------- Data Fetching ------------------- def get_stock_data(ticker): try: stock = yf.Ticker(f"{ticker}.KA") data = stock.history(period="1y") if data.empty: return None data['RSI'] = calculate_rsi(data) return data except: return None # ------------------- Analysis Engine ------------------- def analyze_ticker(ticker): data = get_stock_data(ticker) if data is None: return None current_price = data['Close'].iloc[-1] rsi = data['RSI'].iloc[-1] # Simple Recommendation Logic if rsi < 30: status = "STRONG BUY" color = "green" elif rsi > 70: status = "STRONG SELL" color = "red" else: status = "HOLD" color = "orange" return { "ticker": ticker, "price": round(current_price, 2), "rsi": round(rsi, 2), "status": status, "color": color } # ------------------- Generate Recommendations ------------------- def get_recommendations(): recommendations = [] for ticker in KSE_100: analysis = analyze_ticker(ticker) if analysis: recommendations.append(analysis) df = pd.DataFrame(recommendations) df = df.sort_values(by='rsi') return df # ------------------- Interface Components ------------------- def create_stock_analysis(ticker): data = get_stock_data(ticker) if data is None: return "Data not available", None, None # Create Plot fig = go.Figure(data=[go.Candlestick( x=data.index, open=data['Open'], high=data['High'], low=data['Low'], close=data['Close'] )]) fig.update_layout(title=f"{ticker} Price Chart") # Analysis analysis = analyze_ticker(ticker) status_md = f"## {analysis['status']} \n" \ f"**Price**: {analysis['price']} \n" \ f"**RSI**: {analysis['rsi']}" return status_md, fig.to_html(), get_news(ticker) def get_news(ticker): try: url = f"https://www.google.com/search?q={ticker}+stock+pakistan&tbm=nws" response = requests.get(url) soup = BeautifulSoup(response.text, 'html.parser') articles = soup.find_all('div', class_='BNeawe vvjwJb AP7Wnd')[:3] return "\n\n".join([a.text for a in articles]) except: return "News unavailable" # ------------------- Gradio Interface ------------------- with gr.Blocks(title="PSX Trading Dashboard", theme=gr.themes.Soft()) as app: with gr.Row(): # Left Sidebar - KSE-100 List with gr.Column(scale=1, min_width=200): gr.Markdown("## KSE-100 Constituents") kse_list = gr.DataFrame( value=pd.DataFrame(KSE_100, columns=["Ticker"]), interactive=False, height=600 ) # Main Content with gr.Column(scale=3): gr.Markdown("# PSX Trading Dashboard") with gr.Row(): ticker_input = gr.Textbox(label="Enter Ticker", placeholder="HBL") analyze_btn = gr.Button("Analyze") status_output = gr.Markdown() chart_output = gr.HTML() news_output = gr.Textbox(label="Latest News", interactive=False) # Right Sidebar - Recommendations with gr.Column(scale=1, min_width=200): gr.Markdown("## Live Recommendations") recommendations = gr.DataFrame( headers=["Ticker", "Price", "RSI", "Status"], datatype=["str", "number", "number", "str"], interactive=False, height=600 ) # Event Handlers analyze_btn.click( fn=create_stock_analysis, inputs=ticker_input, outputs=[status_output, chart_output, news_output] ) app.load( fn=get_recommendations, outputs=recommendations, every=300 # Refresh every 5 minutes ) # ------------------- Run App ------------------- if __name__ == "__main__": app.launch(server_port=7860, share=True)