yfinance-ui / app.py
airabbitX's picture
Create app.py
bccac95 verified
import gradio as gr
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import datetime as dt
import json
from io import StringIO
# Helper functions for data processing
def format_large_number(num):
"""Format large numbers to K, M, B, T"""
if num is None or pd.isna(num):
return "N/A"
if isinstance(num, str):
return num
if abs(num) >= 1_000_000_000_000:
return f"{num / 1_000_000_000_000:.2f}T"
elif abs(num) >= 1_000_000_000:
return f"{num / 1_000_000_000:.2f}B"
elif abs(num) >= 1_000_000:
return f"{num / 1_000_000:.2f}M"
elif abs(num) >= 1_000:
return f"{num / 1_000:.2f}K"
else:
return f"{num:.2f}"
def get_ticker_info(ticker_symbol):
"""Get basic information about a ticker"""
try:
ticker = yf.Ticker(ticker_symbol)
info = ticker.info
# Create a more readable format
important_info = {
"Name": info.get("shortName", "N/A"),
"Sector": info.get("sector", "N/A"),
"Industry": info.get("industry", "N/A"),
"Country": info.get("country", "N/A"),
"Market Cap": format_large_number(info.get("marketCap", "N/A")),
"Current Price": info.get("currentPrice", info.get("regularMarketPrice", "N/A")),
"52 Week High": info.get("fiftyTwoWeekHigh", "N/A"),
"52 Week Low": info.get("fiftyTwoWeekLow", "N/A"),
"Website": info.get("website", "N/A"),
"Business Summary": info.get("longBusinessSummary", "N/A")
}
# Convert to formatted string
info_str = ""
for key, value in important_info.items():
info_str += f"**{key}**: {value}\n\n"
return info_str
except Exception as e:
return f"Error retrieving ticker info: {str(e)}"
def get_historical_data(ticker_symbol, period, interval):
"""Get historical price data and create a plotly chart"""
try:
ticker = yf.Ticker(ticker_symbol)
history = ticker.history(period=period, interval=interval)
if history.empty:
return "No historical data available for this ticker", None
# Create Plotly figure
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=history.index,
open=history['Open'],
high=history['High'],
low=history['Low'],
close=history['Close'],
name='Price'
))
# Add volume as bar chart
fig.add_trace(go.Bar(
x=history.index,
y=history['Volume'],
name='Volume',
yaxis='y2',
marker_color='rgba(0, 100, 80, 0.4)'
))
# Layout with secondary y-axis
fig.update_layout(
title=f'{ticker_symbol} Price History',
yaxis_title='Price',
yaxis2=dict(
title='Volume',
overlaying='y',
side='right',
showgrid=False
),
xaxis_rangeslider_visible=False,
height=500
)
return f"Successfully retrieved historical data for {ticker_symbol}", fig
except Exception as e:
return f"Error retrieving historical data: {str(e)}", None
def get_financial_data(ticker_symbol, statement_type, period_type):
"""Get financial statements data"""
try:
ticker = yf.Ticker(ticker_symbol)
if statement_type == "Income Statement":
if period_type == "Annual":
data = ticker.income_stmt
else: # Quarterly
data = ticker.quarterly_income_stmt
elif statement_type == "Balance Sheet":
if period_type == "Annual":
data = ticker.balance_sheet
else: # Quarterly
data = ticker.quarterly_balance_sheet
elif statement_type == "Cash Flow":
if period_type == "Annual":
data = ticker.cashflow
else: # Quarterly
data = ticker.quarterly_cashflow
if data is None or data.empty:
return f"No {statement_type} data available for {ticker_symbol}"
# Format the DataFrame for display
data = data.fillna("N/A")
# Format date columns to be more readable
data.columns = [col.strftime('%Y-%m-%d') if hasattr(col, 'strftime') else str(col) for col in data.columns]
# HTML representation will be more readable in the UI
return data.to_html(classes="table table-striped")
except Exception as e:
return f"Error retrieving financial data: {str(e)}"
def get_company_news(ticker_symbol):
"""Get latest news for the company"""
try:
ticker = yf.Ticker(ticker_symbol)
news = ticker.news
if not news:
return "No recent news available for this ticker"
# Format news items
formatted_news = ""
for i, item in enumerate(news[:5]): # Show top 5 news items
# Extract from nested content structure if present
news_item = item.get('content', item)
# Get title
title = news_item.get('title', 'No title')
# Get publisher
publisher = "Unknown publisher"
if 'provider' in news_item and isinstance(news_item['provider'], dict):
publisher = news_item['provider'].get('displayName', 'Unknown publisher')
# Get link
link = "#"
if 'clickThroughUrl' in news_item and isinstance(news_item['clickThroughUrl'], dict):
link = news_item['clickThroughUrl'].get('url', '#')
elif 'canonicalUrl' in news_item and isinstance(news_item['canonicalUrl'], dict):
link = news_item['canonicalUrl'].get('url', '#')
# Get date
publish_date = 'Unknown date'
if 'pubDate' in news_item:
publish_date = news_item['pubDate']
formatted_news += f"### {i+1}. {title}\n\n"
formatted_news += f"**Source**: {publisher} | **Date**: {publish_date}\n\n"
formatted_news += f"**Link**: [Read full article]({link})\n\n"
# Add description if available
if 'description' in news_item:
description = news_item['description']
# Limit description length and strip HTML tags
if len(description) > 200:
description = description[:200] + "..."
formatted_news += f"{description}\n\n"
formatted_news += "---\n\n"
return formatted_news
except Exception as e:
return f"Error retrieving news: {str(e)}"
def get_analyst_recommendations(ticker_symbol):
"""Get analyst recommendations"""
try:
ticker = yf.Ticker(ticker_symbol)
recommendations = ticker.recommendations
if recommendations is None or recommendations.empty:
return "No analyst recommendations available for this ticker"
# Create a figure for visualization
fig = plt.figure(figsize=(10, 6))
# Count occurrences of each recommendation
rec_counts = recommendations['To Grade'].value_counts()
# Create a pie chart
plt.pie(rec_counts, labels=rec_counts.index, autopct='%1.1f%%',
shadow=True, startangle=90, colors=['#ff9999','#66b3ff','#99ff99','#ffcc99','#c2c2f0'])
plt.axis('equal') # Equal aspect ratio ensures that pie is drawn as a circle
plt.title(f'Analyst Recommendations for {ticker_symbol}')
return f"Found {len(recommendations)} analyst recommendations for {ticker_symbol}", fig
except Exception as e:
return f"Error retrieving analyst recommendations: {str(e)}", None
def get_options_data(ticker_symbol, expiration_date=None):
"""Get options chain data for the ticker"""
try:
ticker = yf.Ticker(ticker_symbol)
# Get available expiration dates
expirations = ticker.options
if not expirations:
return "No options data available for this ticker", None
# If no expiration date is provided or the provided one is invalid, use the first available
if expiration_date is None or expiration_date not in expirations:
expiration_date = expirations[0]
# Get options chain for the selected expiration date
options = ticker.option_chain(expiration_date)
calls = options.calls
puts = options.puts
# Prepare data for visualization
strike_prices = sorted(list(set(calls['strike'].tolist() + puts['strike'].tolist())))
call_volumes = []
put_volumes = []
for strike in strike_prices:
call_vol = calls[calls['strike'] == strike]['volume'].sum()
put_vol = puts[puts['strike'] == strike]['volume'].sum()
call_volumes.append(call_vol)
put_volumes.append(put_vol)
# Create figure for visualization
fig = plt.figure(figsize=(12, 6))
# Plot the data
plt.bar(np.array(strike_prices) - 0.2, call_volumes, width=0.4, label='Calls', color='green', alpha=0.6)
plt.bar(np.array(strike_prices) + 0.2, put_volumes, width=0.4, label='Puts', color='red', alpha=0.6)
plt.xlabel('Strike Price')
plt.ylabel('Volume')
plt.title(f'Options Volume for {ticker_symbol} (Expiry: {expiration_date})')
plt.legend()
plt.grid(True, alpha=0.3)
# Format for readability
current_price = ticker.info.get('regularMarketPrice', ticker.info.get('currentPrice', None))
if current_price:
plt.axvline(x=current_price, color='blue', linestyle='--', label=f'Current Price: {current_price}')
plt.legend()
# Create summary table data
summary = f"""
### Options Summary for {ticker_symbol} (Expiry: {expiration_date})
**Available Expiration Dates:** {', '.join(expirations)}
#### Calls Summary:
- Count: {len(calls)}
- Total Volume: {calls['volume'].sum():,}
- Average Implied Volatility: {calls['impliedVolatility'].mean():.2%}
#### Puts Summary:
- Count: {len(puts)}
- Total Volume: {puts['volume'].sum():,}
- Average Implied Volatility: {puts['impliedVolatility'].mean():.2%}
"""
return summary, fig
except Exception as e:
return f"Error retrieving options data: {str(e)}", None
def get_institutional_holders(ticker_symbol):
"""Get institutional holders of the stock"""
try:
ticker = yf.Ticker(ticker_symbol)
holders = ticker.institutional_holders
if holders is None or holders.empty:
return "No institutional holders data available for this ticker", None
# Create figure for visualization
fig = plt.figure(figsize=(12, 6))
# Sort by percentage held
holders = holders.sort_values(by='% Out', ascending=False)
# Take top 10 holders for visualization
top_holders = holders.head(10)
# Plot the data
plt.barh(top_holders['Holder'], top_holders['% Out'] * 100)
plt.xlabel('Percentage Held (%)')
plt.ylabel('Institution')
plt.title(f'Top Institutional Holders of {ticker_symbol}')
plt.grid(True, alpha=0.3)
# Format x-axis as percentage
plt.gca().xaxis.set_major_formatter(plt.FuncFormatter(lambda x, _: f'{x:.1f}%'))
# Format the DataFrame for display
holders_html = holders.to_html(classes="table table-striped")
return holders_html, fig
except Exception as e:
return f"Error retrieving institutional holders: {str(e)}", None
def get_sector_industry_info(ticker_symbol):
"""Get sector and industry information for the ticker"""
try:
ticker = yf.Ticker(ticker_symbol)
info = ticker.info
sector_key = info.get('sectorKey')
industry_key = info.get('industryKey')
if not sector_key or not industry_key:
return "Sector or industry information not available for this ticker"
try:
# Get sector information
sector = yf.Sector(sector_key)
sector_info = f"""
### Sector Information
**Name:** {sector.name}
**Key:** {sector.key}
**Symbol:** {sector.symbol}
#### Overview
{sector.overview}
#### Top Companies in {sector.name} Sector
"""
for company in sector.top_companies[:5]: # Show top 5 companies
sector_info += f"- {company.get('name', 'N/A')} ({company.get('symbol', 'N/A')})\n"
# Get industry information
industry = yf.Industry(industry_key)
industry_info = f"""
### Industry Information
**Name:** {industry.name}
**Key:** {industry.key}
**Sector:** {industry.sector_name}
#### Top Performing Companies in {industry.name}
"""
for company in industry.top_performing_companies[:5]: # Show top 5 companies
industry_info += f"- {company.get('name', 'N/A')} ({company.get('symbol', 'N/A')})\n"
return sector_info + industry_info
except Exception as e:
return f"Error retrieving sector/industry details: {str(e)}"
except Exception as e:
return f"Error retrieving sector/industry information: {str(e)}"
def search_stocks(query, max_results=10):
"""Search for stocks using the YF Search API"""
try:
# First try with the standard approach
search_results = yf.Search(query, max_results=max_results)
quotes = search_results.quotes
if not quotes:
return "No search results found"
# Format the results
formatted_results = "### Search Results\n\n"
for quote in quotes:
symbol = quote.get('symbol', 'N/A')
name = quote.get('shortname', quote.get('longname', 'N/A'))
exchange = quote.get('exchange', 'N/A')
quote_type = quote.get('quoteType', 'N/A').capitalize()
formatted_results += f"**{symbol}** - {name}\n"
formatted_results += f"Exchange: {exchange} | Type: {quote_type}\n\n"
return formatted_results
except AttributeError as e:
if "has no attribute 'update'" in str(e):
# Alternative: Use the Ticker directly for basic information
try:
# If search fails, try to get info directly for the symbol
if len(query.strip()) <= 5: # Likely a symbol
ticker = yf.Ticker(query.strip())
info = ticker.info
formatted_results = "### Direct Ticker Results\n\n"
formatted_results += f"**{query.strip()}** - {info.get('shortName', 'N/A')}\n"
formatted_results += f"Exchange: {info.get('exchange', 'N/A')} | "
formatted_results += f"Type: {info.get('quoteType', 'N/A').capitalize()}\n\n"
return formatted_results
else:
return f"Search functionality unavailable due to version compatibility issue. If you know the exact ticker symbol, try entering it in the Single Ticker Analysis tab."
except:
return f"Search functionality unavailable due to version compatibility issue. If you know the exact ticker symbol, try entering it in the Single Ticker Analysis tab."
else:
return f"Error searching stocks: {str(e)}"
except Exception as e:
return f"Error searching stocks: {str(e)}"
def get_multi_ticker_comparison(ticker_symbols, period="1y"):
"""Compare multiple tickers in a single chart"""
try:
if not ticker_symbols:
return "Please enter at least one ticker symbol", None
# Split input string into list of ticker symbols
tickers = [t.strip() for t in ticker_symbols.split() if t.strip()]
if not tickers:
return "Please enter at least one ticker symbol", None
# Download data for all tickers
data = yf.download(tickers, period=period, group_by='ticker')
if data.empty:
return "No data available for the provided tickers", None
# For a single ticker, the structure is different
if len(tickers) == 1:
ticker = tickers[0]
price_data = data['Close']
price_data.name = ticker
price_data = pd.DataFrame(price_data)
else:
# Extract closing prices for each ticker
price_data = pd.DataFrame()
for ticker in tickers:
try:
if (ticker, 'Close') in data.columns:
price_data[ticker] = data[ticker]['Close']
except:
continue
if price_data.empty:
return "Could not retrieve closing price data for the provided tickers", None
# Normalize the data to start at 100 for fair comparison
normalized_data = price_data.copy()
for col in normalized_data.columns:
normalized_data[col] = normalized_data[col] / normalized_data[col].iloc[0] * 100
# Create figure for visualization
fig = plt.figure(figsize=(12, 6))
for col in normalized_data.columns:
plt.plot(normalized_data.index, normalized_data[col], label=col)
plt.xlabel('Date')
plt.ylabel('Normalized Price (Base = 100)')
plt.title(f'Comparative Performance ({period})')
plt.legend()
plt.grid(True, alpha=0.3)
# Calculate performance metrics
performance = {}
for ticker in price_data.columns:
start_price = price_data[ticker].iloc[0]
end_price = price_data[ticker].iloc[-1]
pct_change = (end_price - start_price) / start_price * 100
performance[ticker] = pct_change
# Create a summary of the performance
summary = "### Performance Summary\n\n"
for ticker, pct in sorted(performance.items(), key=lambda x: x[1], reverse=True):
summary += f"**{ticker}**: {pct:.2f}%\n\n"
return summary, fig
except Exception as e:
return f"Error comparing tickers: {str(e)}", None
def get_market_status():
"""Get current market status and summary"""
try:
# Get US market status
us_market = yf.Market("US")
status = us_market.status
if not status:
return "Unable to retrieve market status"
# Format the response
market_info = "### Market Status\n\n"
market_state = status.get('marketState', 'Unknown')
trading_status = "Open" if market_state == "REGULAR" else "Closed"
market_info += f"**US Market Status:** {trading_status} ({market_state})\n\n"
# Get summary for different markets
markets = ["US", "EUROPE", "ASIA", "CRYPTOCURRENCIES"]
for market_id in markets:
try:
market = yf.Market(market_id)
summary = market.summary
if summary is None:
market_info += f"### {market_id} Market Summary\n\nNo data available\n\n---\n\n"
continue
market_info += f"### {market_id} Market Summary\n\n"
# Make sure we handle the summary data correctly, regardless of its type
summary_items = []
if isinstance(summary, list):
summary_items = summary[:5] # Get first 5 items
elif hasattr(summary, '__getitem__'):
try:
summary_items = summary[:5] # Try to get first 5 items
except:
# If slicing fails, try to convert to list first
try:
summary_items = list(summary)[:5]
except:
summary_items = []
# Display market indices
if not summary_items:
market_info += "No summary data available\n\n"
else:
for item in summary_items:
if not isinstance(item, dict):
continue
symbol = item.get('symbol', 'N/A')
name = item.get('shortName', item.get('longName', 'N/A'))
price = item.get('regularMarketPrice', 'N/A')
change = item.get('regularMarketChangePercent', 0)
# Format change with color indicator
change_text = f"{change:.2f}%" if isinstance(change, (int, float)) else change
if isinstance(change, (int, float)):
if change > 0:
change_text = f"🟢 +{change_text}"
elif change < 0:
change_text = f"🔴 {change_text}"
market_info += f"**{name} ({symbol}):** {price} ({change_text})\n\n"
market_info += "---\n\n"
except Exception as e:
market_info += f"### {market_id} Market Summary\n\nError retrieving {market_id} market summary: {str(e)}\n\n---\n\n"
return market_info
except Exception as e:
return f"Error retrieving market status: {str(e)}"
# Gradio UI components
with gr.Blocks(title="YFinance Explorer") as app:
gr.Markdown("# YFinance Explorer\nA comprehensive tool to test all features of the yfinance library")
with gr.Tab("Single Ticker Analysis"):
with gr.Row():
ticker_input = gr.Textbox(label="Enter Ticker Symbol", placeholder="e.g. AAPL, MSFT, GOOG", value="AAPL")
ticker_submit = gr.Button("Analyze")
with gr.Tabs():
with gr.Tab("Overview"):
ticker_info_output = gr.Markdown()
with gr.Tab("Price History"):
with gr.Row():
period_dropdown = gr.Dropdown(
choices=["1d", "5d", "1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"],
value="1y",
label="Period"
)
interval_dropdown = gr.Dropdown(
choices=["1m", "2m", "5m", "15m", "30m", "60m", "90m", "1h", "1d", "5d", "1wk", "1mo", "3mo"],
value="1d",
label="Interval"
)
history_status = gr.Markdown()
history_plot = gr.Plot()
with gr.Tab("Financials"):
with gr.Row():
statement_dropdown = gr.Dropdown(
choices=["Income Statement", "Balance Sheet", "Cash Flow"],
value="Income Statement",
label="Financial Statement"
)
period_type_dropdown = gr.Dropdown(
choices=["Annual", "Quarterly"],
value="Annual",
label="Period Type"
)
financial_data_output = gr.HTML()
with gr.Tab("News"):
news_output = gr.Markdown()
with gr.Tab("Multi-Ticker Comparison"):
with gr.Row():
multi_ticker_input = gr.Textbox(label="Enter Ticker Symbols (space separated)", placeholder="e.g. AAPL MSFT GOOG", value="AAPL MSFT GOOG")
comparison_period = gr.Dropdown(
choices=["1mo", "3mo", "6mo", "1y", "2y", "5y", "10y", "ytd", "max"],
value="1y",
label="Comparison Period"
)
compare_button = gr.Button("Compare")
comparison_status = gr.Markdown()
comparison_plot = gr.Plot()
with gr.Tab("Market Status"):
market_status_button = gr.Button("Get Market Status")
market_status_output = gr.Markdown()
with gr.Tab("Stock Search"):
with gr.Row():
search_input = gr.Textbox(label="Search Term", placeholder="Enter company name or ticker")
max_results_slider = gr.Slider(minimum=5, maximum=30, value=10, step=5, label="Max Results")
search_button = gr.Button("Search")
search_results = gr.Markdown()
# Event handlers
ticker_submit.click(
fn=get_ticker_info,
inputs=[ticker_input],
outputs=[ticker_info_output]
)
ticker_submit.click(
fn=get_historical_data,
inputs=[ticker_input, period_dropdown, interval_dropdown],
outputs=[history_status, history_plot]
)
ticker_submit.click(
fn=get_financial_data,
inputs=[ticker_input, statement_dropdown, period_type_dropdown],
outputs=[financial_data_output]
)
ticker_submit.click(
fn=get_company_news,
inputs=[ticker_input],
outputs=[news_output]
)
compare_button.click(
fn=get_multi_ticker_comparison,
inputs=[multi_ticker_input, comparison_period],
outputs=[comparison_status, comparison_plot]
)
market_status_button.click(
fn=get_market_status,
inputs=[],
outputs=[market_status_output]
)
search_button.click(
fn=search_stocks,
inputs=[search_input, max_results_slider],
outputs=[search_results]
)
# Update statement and interval options based on selections
def update_interval_choices(period):
if period in ["1d", "5d"]:
return gr.Dropdown.update(choices=["1m", "2m", "5m", "15m", "30m", "60m", "90m", "1h"], value="1m")
else:
return gr.Dropdown.update(choices=["1d", "5d", "1wk", "1mo", "3mo"], value="1d")
period_dropdown.change(
fn=update_interval_choices,
inputs=[period_dropdown],
outputs=[interval_dropdown]
)
if __name__ == "__main__":
app.launch()