LATEST / app.py
Safwanahmad619's picture
Create app.py
1a8b90a verified
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)