app.py

import pandas as pd
import yfinance as yf
import numpy as np
import gradio as gr
import matplotlib.pyplot as plt
from functools import lru_cache
import asyncio
import concurrent.futures
import time
from typing import Dict, List, Optional, Any, Tuple
import logging

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('stock_analyzer')

# Cache Yahoo Finance data to avoid rate limits
@lru_cache(maxsize=100)
def get_financial_data(ticker: str) -> Optional[Dict[str, Any]]:
    """
    Fetch financial data for a given stock ticker using Yahoo Finance.
    
    Args:
        ticker: Stock symbol to fetch data for
        
    Returns:
        Dictionary of financial metrics or None if fetch failed
    """
    try:
        stock = yf.Ticker(ticker)
        info = stock.info
        
        return {
            'Ticker': ticker,
            'PE_Ratio': info.get('forwardPE'),
            'Debt_to_Equity': info.get('debtToEquity'),
            'Revenue_Growth': info.get('revenueGrowth'),
            'ROE': info.get('returnOnEquity'),
            'ROA': info.get('returnOnAssets'),
            'Gross_Margin': info.get('grossMargins'),
            'EBITDA': info.get('ebitda'),
            'Market_Cap': info.get('marketCap'),
            'Dividend_Yield': info.get('dividendYield'),
            'Profit_Margin': info.get('profitMargins'),
            'EPS_Growth': info.get('earningsGrowth'),
            'Price_to_Book': info.get('priceToBook'),
            'Current_Price': info.get('currentPrice')
        }
    except Exception as e:
        logger.error(f"Error fetching data for {ticker}: {e}")
        return None

# Fetch data concurrently for multiple tickers
async def fetch_data_concurrently(tickers: List[str]) -> List[Dict[str, Any]]:
    """
    Fetch financial data for multiple tickers concurrently.
    
    Args:
        tickers: List of stock symbols
        
    Returns:
        List of financial data dictionaries for each ticker
    """
    loop = asyncio.get_event_loop()
    with concurrent.futures.ThreadPoolExecutor() as executor:
        tasks = [
            loop.run_in_executor(
                executor,
                get_financial_data,
                ticker
            )
            for ticker in tickers
        ]
        results = await asyncio.gather(*tasks)
    return [r for r in results if r is not None]

# Robust normalization using winsorization (cap outliers at specified percentiles)
def normalize(series: pd.Series, reverse: bool = False, 
              lower_percentile: float = 0.10, upper_percentile: float = 0.90) -> pd.Series:
    """
    Normalize a series to a 0-10 scale using winsorization.
    
    Args:
        series: Data series to normalize
        reverse: If True, reverse the normalization (10 becomes low, 0 becomes high)
        lower_percentile: Lower percentile for clipping
        upper_percentile: Upper percentile for clipping
        
    Returns:
        Normalized series
    """
    if series.isna().all() or len(series.unique()) <= 1:
        return pd.Series(5, index=series.index)  # Default to middle value if no variation
        
    q_low = series.quantile(lower_percentile)
    q_high = series.quantile(upper_percentile)
    
    # Avoid division by zero
    if q_high == q_low:
        return pd.Series(5, index=series.index)
        
    clipped_series = series.clip(q_low, q_high)
    
    if reverse:
        return 10 * (1 - (clipped_series - q_low) / (q_high - q_low))
    return 10 * ((clipped_series - q_low) / (q_high - q_low))

# Calculate scores with customizable weights
def calculate_scores(df: pd.DataFrame, growth_weight: float, 
                     value_weight: float, risk_weight: float) -> pd.DataFrame:
    """
    Calculate stock scores based on various financial metrics.
    
    Args:
        df: DataFrame containing financial metrics
        growth_weight: Weight for growth metrics
        value_weight: Weight for value metrics
        risk_weight: Weight for risk metrics
        
    Returns:
        DataFrame with added score columns
    """
    # Make a copy to avoid modifying the original
    scored_df = df.copy()
    
    # Growth Metrics (higher is better)
    scored_df['Revenue_Growth_Score'] = normalize(df['Revenue_Growth'])
    scored_df['EPS_Growth_Score'] = normalize(df['EPS_Growth'])
    scored_df['ROE_Score'] = normalize(df['ROE'])
    scored_df['ROA_Score'] = normalize(df['ROA'])
    
    # Calculate Growth Score with nan handling
    growth_cols = ['Revenue_Growth_Score', 'EPS_Growth_Score', 'ROE_Score', 'ROA_Score']
    scored_df['Growth_Score'] = scored_df[growth_cols].mean(axis=1)

    # Value Metrics (lower is better)
    scored_df['PE_Ratio_Score'] = normalize(df['PE_Ratio'], reverse=True)
    scored_df['Price_to_Book_Score'] = normalize(df['Price_to_Book'], reverse=True)
    scored_df['Dividend_Yield_Score'] = normalize(df['Dividend_Yield'])  # Higher yield is better
    
    # Calculate Value Score
    value_cols = ['PE_Ratio_Score', 'Price_to_Book_Score', 'Dividend_Yield_Score']
    scored_df['Value_Score'] = scored_df[value_cols].mean(axis=1)

    # Risk Metrics (higher values indicate lower risk)
    scored_df['Debt_to_Equity_No_Risk_Score'] = normalize(df['Debt_to_Equity'], reverse=True)  # Lower debt = lower risk
    scored_df['Profit_Margin_No_Risk_Score'] = normalize(df['Profit_Margin'])  # Higher margin = lower risk
    scored_df['Market_Cap_No_Risk_Score'] = normalize(df['Market_Cap'])  # Larger companies = lower risk
    
    # Calculate No_Risk_Score
    no_risk_cols = ['Debt_to_Equity_No_Risk_Score', 'Profit_Margin_No_Risk_Score', 'Market_Cap_No_Risk_Score']
    scored_df['No_Risk_Score'] = scored_df[no_risk_cols].mean(axis=1)

    # Normalize weights to ensure they sum to 1.0
    total = growth_weight + value_weight + risk_weight
    if total == 0:
        growth_weight = value_weight = risk_weight = 1/3
    else:
        growth_weight /= total
        value_weight /= total
        risk_weight /= total

    # Total Score (Weighted Average)
    scored_df['Total_Score'] = (
        growth_weight * scored_df['Growth_Score'] + 
        value_weight * scored_df['Value_Score'] + 
        risk_weight * scored_df['No_Risk_Score']
    )

    return scored_df

# Generate bar chart for scores with custom styling
def plot_bar_chart(df: pd.DataFrame) -> plt.Figure:
    """
    Generate a bar chart showing Growth, Value, and No_Risk scores for each ticker.
    
    Args:
        df: DataFrame containing score data
        
    Returns:
        Matplotlib figure
    """
    # Set a modern style
    plt.style.use('seaborn-v0_8-whitegrid')
    
    fig, ax = plt.subplots(figsize=(12, 7))
    
    # Custom colors
    colors = ['#4CAF50', '#2196F3', '#FF9800']
    
    df.set_index('Ticker')[['Growth_Score', 'Value_Score', 'No_Risk_Score']].plot(
        kind='bar', 
        stacked=False,  # Changed to unstacked for better comparison
        color=colors,
        width=0.7,
        alpha=0.8,
        ax=ax
    )
    
    # Add total score as a line
    total_scores = df.set_index('Ticker')['Total_Score']
    ax2 = ax.twinx()
    ax2.plot(range(len(total_scores)), total_scores, 'ro-', linewidth=2.5, markersize=8, label='Total Score')
    ax2.set_ylim(0, 10.5)
    ax2.set_ylabel('Total Score', fontsize=12, color='r')
    
    # Enhance appearance
    ax.set_title("Stock Analysis Scores", fontsize=16, fontweight='bold', pad=20)
    ax.set_ylabel("Component Scores (0-10)", fontsize=12)
    ax.set_xlabel("", fontsize=12)
    ax.tick_params(axis='x', rotation=45)
    ax.set_ylim(0, 10.5)
    ax.grid(axis='y', linestyle='--', alpha=0.7)
    
    # Add legend
    lines, labels = ax.get_legend_handles_labels()
    lines2, labels2 = ax2.get_legend_handles_labels()
    ax.legend(lines + lines2, labels + labels2, loc='upper center', bbox_to_anchor=(0.5, -0.15), 
              ncol=4, frameon=True, fontsize=10)
    
    plt.tight_layout()
    return fig

# Generate radar plot for scores with improved styling
def plot_radar_chart(df: pd.DataFrame, tickers: List[str]) -> plt.Figure:
    """
    Generate a radar chart comparing selected tickers.
    
    Args:
        df: DataFrame containing score data
        tickers: List of tickers to include in the radar chart
        
    Returns:
        Matplotlib figure
    """
    # Filter for requested tickers only
    plot_df = df[df['Ticker'].isin(tickers)]
    
    if plot_df.empty:
        # Fallback if none of the requested tickers are found
        plot_df = df.head(min(3, len(df)))
        tickers = plot_df['Ticker'].tolist()
    
    # Categories and angles
    categories = ['Growth', 'Value', 'No_Risk', 'Total']
    N = len(categories)
    angles = [n / float(N) * 2 * np.pi for n in range(N)]
    angles += angles[:1]  # Close the loop
    
    # Set up plot
    fig = plt.figure(figsize=(10, 8))
    ax = fig.add_subplot(111, polar=True)
    
    # Custom colors for each ticker
    colors = plt.cm.viridis(np.linspace(0, 1, len(tickers)))
    
    # Plot each ticker
    for i, ticker in enumerate(tickers):
        ticker_data = plot_df[plot_df['Ticker'] == ticker]
        if ticker_data.empty:
            continue
            
        values = ticker_data[['Growth_Score', 'Value_Score', 'No_Risk_Score', 'Total_Score']].values.flatten().tolist()
        values += values[:1]  # Close the loop
        
        ax.plot(angles, values, linewidth=2, linestyle='solid', color=colors[i], label=ticker)
        ax.fill(angles, values, color=colors[i], alpha=0.1)
    
    # Set chart properties
    ax.set_xticks(angles[:-1])
    ax.set_xticklabels(categories, size=12)
    ax.set_yticks(np.arange(2, 12, 2))
    ax.set_yticklabels(np.arange(2, 12, 2), size=10)
    ax.set_ylim(0, 10)
    
    # Add title and legend
    plt.title("Stock Comparison Radar Chart", size=16, fontweight='bold', pad=20)
    plt.legend(loc='upper right', bbox_to_anchor=(0.1, 0.1), frameon=True)
    
    return fig

# Generate a detailed metrics table
def create_metrics_table(df: pd.DataFrame) -> pd.DataFrame:
    """
    Create a detailed metrics table for display.
    
    Args:
        df: DataFrame with stock data
        
    Returns:
        DataFrame with formatted metrics
    """
    metrics_df = df[['Ticker', 'Current_Price', 'PE_Ratio', 'Price_to_Book', 
                     'Debt_to_Equity', 'ROE', 'ROA', 'Revenue_Growth', 
                     'EPS_Growth', 'Profit_Margin', 'Dividend_Yield']].copy()
    
    # Format percentages
    for col in ['ROE', 'ROA', 'Revenue_Growth', 'EPS_Growth', 'Profit_Margin', 'Dividend_Yield']:
        metrics_df[col] = metrics_df[col].apply(lambda x: f"{x*100:.2f}%" if pd.notnull(x) else "N/A")
    
    # Format ratios
    for col in ['PE_Ratio', 'Price_to_Book', 'Debt_to_Equity']:
        metrics_df[col] = metrics_df[col].apply(lambda x: f"{x:.2f}" if pd.notnull(x) else "N/A")
        
    # Format price
    metrics_df['Current_Price'] = metrics_df['Current_Price'].apply(lambda x: f"${x:.2f}" if pd.notnull(x) else "N/A")
    
    return metrics_df

# Main analysis function for Gradio app
async def analyze_tickers(
    tickers: str, 
    growth_weight: float, 
    value_weight: float, 
    risk_weight: float,
    top_n: int = 5
) -> Tuple[pd.DataFrame, pd.DataFrame, plt.Figure, plt.Figure]:
    """
    Analyze stock tickers and generate visualizations.
    
    Args:
        tickers: Comma-separated list of stock tickers
        growth_weight: Weight for growth metrics
        value_weight: Weight for value metrics
        risk_weight: Weight for risk metrics
        top_n: Number of top stocks to highlight
        
    Returns:
        Tuple containing scores DataFrame, metrics DataFrame, bar chart, and radar chart
    """
    start_time = time.time()
    
    # Parse and clean ticker list
    ticker_list = [t.strip().upper() for t in tickers.split(",") if t.strip()]
    
    if not ticker_list:
        return pd.DataFrame(), pd.DataFrame(), plt.figure(), plt.figure()
    
    # Fetch data asynchronously
    data = await fetch_data_concurrently(ticker_list)
    
    if not data:
        logger.warning("No valid data retrieved for any tickers")
        return pd.DataFrame(), pd.DataFrame(), plt.figure(), plt.figure()
    
    # Create DataFrame
    df = pd.DataFrame(data)
    
    # Handle missing values - replace with median for numerical columns
    numerical_cols = df.select_dtypes(include=[np.number]).columns
    df[numerical_cols] = df[numerical_cols].fillna(df[numerical_cols].median())
    
    # Calculate scores
    df = calculate_scores(df, growth_weight, value_weight, risk_weight)
    
    # Sort by Total Score
    df = df.sort_values(by='Total_Score', ascending=False).reset_index(drop=True)
    
    # Create metrics table
    metrics_table = create_metrics_table(df)
    
    # Generate bar chart for all tickers
    bar_chart = plot_bar_chart(df)
    
    # Generate radar chart for top N tickers
    top_tickers = df.head(min(top_n, len(df)))['Ticker'].tolist()
    radar_chart = plot_radar_chart(df, top_tickers)
    
    # Prepare scores table for display
    scores_table = df[['Ticker', 'Total_Score', 'Growth_Score', 'Value_Score', 'No_Risk_Score']].copy()
    scores_table = scores_table.round(2)
    
    logger.info(f"Analysis completed in {time.time() - start_time:.2f} seconds")
    
    return scores_table, metrics_table, bar_chart, radar_chart



# Custom CSS for better appearance
custom_css = """
.gradio-container {
    font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
}

.container {
    max-width: 1200px;
    margin: auto;
}

button#analyze-btn {
    background-color: #003366;  /* Dark blue color */
    color: white;
    border: none;
}


"""

# Gradio interface
def create_gradio_interface():
    """Create and configure the Gradio interface"""

    
    with gr.Blocks(theme=gr.themes.Monochrome(),css=custom_css) as iface:
        gr.Markdown("# Fundamental Financial Analysis")
        gr.Markdown("""
        Enter comma-separated stock tickers and adjust the weights to analyze stocks based on 
        growth potential, value metrics, and risk factors.
        """)
        
        with gr.Row():
            tickers_input = gr.Textbox(
                label="Stock Tickers (comma-separated)", 
                placeholder="AAPL, MSFT, GOOG, AMZN, TSLA",
                lines=1
            )
            analyze_btn = gr.Button("Analyze Stocks", variant="primary")
        
        with gr.Row():
            with gr.Column():
                growth_weight = gr.Slider(
                    minimum=0, maximum=1, step=0.05, 
                    label="Growth Weight", value=0.4
                )
            with gr.Column():
                value_weight = gr.Slider(
                    minimum=0, maximum=1, step=0.05, 
                    label="Value Weight", value=0.4
                )
            with gr.Column():
                risk_weight = gr.Slider(
                    minimum=0, maximum=1, step=0.05, 
                    label="Risk Weight", value=0.2
                )
        
        with gr.Tabs():
            with gr.TabItem("Scores & Charts"):
                with gr.Row():
                    with gr.Column():
                        scores_output = gr.Dataframe(label="Stock Scores")
                    with gr.Column():
                        metrics_output = gr.Dataframe(label="Financial Metrics")
                
                with gr.Row():
                    with gr.Column():
                        bar_chart_output = gr.Plot(label="Component Scores Chart")
                    with gr.Column():
                        radar_chart_output = gr.Plot(label="Top Stocks Comparison")
            
            with gr.TabItem("Help & Information"):
                gr.Markdown("""
                ## How to Use This Tool
                
                1. Enter stock tickers separated by commas (e.g., "AAPL, MSFT, GOOG")
                2. Adjust weights based on your investment strategy:
                   - **Growth Weight**: Emphasizes revenue growth, EPS growth, ROE, and ROA
                   - **Value Weight**: Focuses on PE ratio, price-to-book, and dividend yield
                   - **Risk Weight**: Considers debt-to-equity ratio, profit margins, and market cap
                3. Click "Analyze Stocks" to see results
                
                ## About the Scores
                
                All metrics are normalized on a scale of 0-10, with higher being better:
                - **Growth Score**: Higher values indicate stronger growth potential
                - **Value Score**: Higher values indicate the stock may be undervalued
                - **No_Risk_Score**: Higher values suggest lower relative risk
                - **Total Score**: Weighted average of the three component scores
                
                ## Data Source
                
                Financial data is provided by Yahoo Finance via the yfinance package.
                """)
        
        # Set up the async functionality with a wrapper function
        def analyze_wrapper(*args):
            return asyncio.run(analyze_tickers(*args))
        
        analyze_btn.click(
            analyze_wrapper,
            inputs=[tickers_input, growth_weight, value_weight, risk_weight],
            outputs=[scores_output, metrics_output, bar_chart_output, radar_chart_output]
        )
    
    return iface

# Entry point
if __name__ == "__main__":
    logger.info("Starting Stock Analyzer app")
    iface = create_gradio_interface()
    iface.launch() #share=False