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()