app.py

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