app.py

# app.py: Fundamental Analysis Pipeline on Hugging Face Spaces
# Version: 4.0 (Dynamic User Input with Robust Safeguards)

import gradio as gr
import yfinance
import pandas as pd
import numpy as np
import numpy_financial as npf
import warnings
import traceback

# Suppress warnings for a cleaner UI
warnings.simplefilter(action='ignore', category=FutureWarning)

# --- 1. GLOBAL CONFIGURATION & CONSTANTS ---

# Peer groups are still useful for the Comps model if a user enters a supported ticker.
PEER_GROUPS = {
    'CHTR': ['CMCSA', 'T', 'VZ'], 'CRM': ['NOW', 'ADBE', 'MSFT'], 'TTD': ['GOOGL', 'MGNI', 'PUBM'],
    'LULU': ['NKE', 'DECK', 'UAA'], 'FICO': ['V', 'MA', 'AXP'], 'AAPL': ['MSFT', 'GOOGL', 'AMZN'],
    'MSFT': ['AAPL', 'GOOGL', 'AMZN', 'ORCL'], 'GOOGL': ['MSFT', 'AAPL', 'AMZN', 'META']
}
# DCF Assumptions
RISK_FREE_RATE = 0.042 # Default fallback
MARKET_RISK_PREMIUM = 0.055
PERPETUAL_GROWTH_RATE = 0.025
PROJECTION_YEARS = 5

# --- 2. PIPELINE FUNCTIONS ---

def get_stock_data(ticker_str):
    """
    Fetches data for a single stock and validates the ticker.
    Returns a tuple: (stock_data_dict, company_name_str)
    """
    print(f"LOG: [get_stock_data] Fetching data for {ticker_str}...")
    ticker_obj = yfinance.Ticker(ticker_str)
    
    # SAFEGUARD: Validate the ticker by checking the info dictionary.
    if not ticker_obj.info or 'symbol' not in ticker_obj.info:
        raise ValueError(f"Could not find a valid stock for ticker '{ticker_str}'. Please check the symbol and try again.")
        
    company_name = ticker_obj.info.get('longName', ticker_str)
    
    stock_data = {
        'profile': ticker_obj.info,
        'income_statement': ticker_obj.financials.T,
        'balance_sheet': ticker_obj.balance_sheet.T,
        'cash_flow': ticker_obj.cashflow.T
    }
    
    if stock_data['income_statement'].empty:
        raise ValueError(f"Financial statements for {ticker_str} are empty. The stock may be a fund, ETF, or have no available data.")
    
    print(f"LOG: [get_stock_data] Data fetch for {company_name} successful.")
    return stock_data, company_name

def calculate_metrics(stock_data):
    """Calculates key financial ratios from raw statement data."""
    print("LOG: [calculate_metrics] Starting calculation...")
    income_df = stock_data['income_statement']
    cash_flow_df = stock_data['cash_flow']
    metrics_df = pd.DataFrame(index=income_df.index)
    
    operating_income = pd.to_numeric(income_df.get('Operating Income'), errors='coerce')
    total_revenue = pd.to_numeric(income_df.get('Total Revenue'), errors='coerce')
    metrics_df['operatingMargin'] = operating_income / total_revenue
    
    capex = pd.to_numeric(cash_flow_df.get('Capital Expenditure'), errors='coerce').fillna(0)
    operating_cash = pd.to_numeric(cash_flow_df.get('Operating Cash Flow'), errors='coerce').fillna(0)
    metrics_df['freeCashFlow'] = operating_cash + capex
    
    print("LOG: [calculate_metrics] Calculation successful.")
    return metrics_df.sort_index()

def calculate_wacc(stock_data):
    """Calculates the Weighted Average Cost of Capital (WACC)."""
    print("LOG: [calculate_wacc] Starting WACC calculation...")
    try:
        profile = stock_data['profile']
        balance_df_latest = stock_data['balance_sheet'].iloc[-1]
        income_df_latest = stock_data['income_statement'].iloc[-1]

        market_cap = np.nan_to_num(profile.get('marketCap', 0))
        total_debt = np.nan_to_num(balance_df_latest.get('Total Debt', 0))
        beta = np.nan_to_num(profile.get('beta', 1.0))
        interest_expense = np.nan_to_num(income_df_latest.get('Interest Expense', 0))
        income_before_tax = np.nan_to_num(income_df_latest.get('Pretax Income'))
        tax_provision = np.nan_to_num(income_df_latest.get('Tax Provision', 0))
        
        cost_of_equity = RISK_FREE_RATE + beta * MARKET_RISK_PREMIUM
        cost_of_debt = abs(interest_expense) / total_debt if total_debt > 0 else 0
        tax_rate = tax_provision / income_before_tax if income_before_tax != 0 else 0.21
        tax_rate = max(0, min(tax_rate, 0.40))
        
        equity_value = market_cap
        debt_value = total_debt
        total_value = equity_value + debt_value
        
        if total_value == 0: return None
        
        wacc = (equity_value / total_value * cost_of_equity) + \
               (debt_value / total_value * cost_of_debt * (1 - tax_rate))
        print(f"LOG: [calculate_wacc] WACC calculation successful: {wacc:.3%}")
        return wacc
    except Exception as e:
        print(f"LOG: [calculate_wacc] WACC calculation failed: {e}")
        return None

def run_dcf_valuation(stock_data, financial_metrics, wacc):
    """Runs the DCF valuation model."""
    print("LOG: [run_dcf_valuation] Starting DCF valuation...")
    try:
        if wacc is None: return None
            
        last_fcf = financial_metrics['freeCashFlow'].dropna().iloc[-1] if not financial_metrics['freeCashFlow'].dropna().empty else None
        if last_fcf is None or last_fcf <= 0: return None

        revenue_series = stock_data['income_statement']['Total Revenue'].dropna()
        revenue_cagr = (revenue_series.iloc[-1] / revenue_series.iloc[0]) ** (1/len(revenue_series)) - 1 if len(revenue_series) > 1 and revenue_series.iloc[0] > 0 else 0.05
        initial_growth_rate = max(0.01, min(revenue_cagr, 0.20))
        projected_fcf = []
        fcf = last_fcf
        for i in range(1, PROJECTION_YEARS + 1):
            growth_rate = initial_growth_rate + i * (PERPETUAL_GROWTH_RATE - initial_growth_rate) / PROJECTION_YEARS
            fcf *= (1 + growth_rate)
            projected_fcf.append(fcf)
        terminal_value = (projected_fcf[-1] * (1 + PERPETUAL_GROWTH_RATE)) / (wacc - PERPETUAL_GROWTH_RATE)
        if terminal_value < 0: return None
        cash_flows = projected_fcf
        cash_flows[-1] += terminal_value
        enterprise_value = npf.npv(wacc, [0] + cash_flows)
        latest_bs = stock_data['balance_sheet'].iloc[-1]
        net_debt = np.nan_to_num(latest_bs.get('Total Debt', 0)) - np.nan_to_num(latest_bs.get('Cash And Cash Equivalents', 0))
        shares_outstanding = stock_data['profile']['sharesOutstanding']
        equity_value = enterprise_value - net_debt
        
        print("LOG: [run_dcf_valuation] DCF valuation successful.")
        return equity_value / shares_outstanding
    except Exception as e:
        print(f"LOG: [run_dcf_valuation] DCF valuation failed: {e}")
        return None

def run_comps_valuation(stock_data):
    """Runs the Comparable Company Analysis."""
    print("LOG: [run_comps_valuation] Starting Comps valuation...")
    try:
        ticker = stock_data['profile']['symbol']
        peer_list = PEER_GROUPS.get(ticker, [])
        if not peer_list: return None
        
        peer_ratios = [yfinance.Ticker(p).info.get('enterpriseToEbitda') for p in peer_list]
        median_ev_ebitda = pd.Series([r for r in peer_ratios if r]).median()
        target_ebitda = stock_data['profile'].get('ebitda')
        shares_outstanding = stock_data['profile'].get('sharesOutstanding')

        if not all([median_ev_ebitda, target_ebitda, shares_outstanding]) or target_ebitda <= 0:
            return None
        
        implied_ev = target_ebitda * median_ev_ebitda
        latest_bs = stock_data['balance_sheet'].iloc[-1]
        net_debt = np.nan_to_num(latest_bs.get('Total Debt', 0)) - np.nan_to_num(latest_bs.get('Cash And Cash Equivalents', 0))
        implied_equity_value = implied_ev - net_debt
        
        print("LOG: [run_comps_valuation] Comps valuation successful.")
        return implied_equity_value / shares_outstanding
    except Exception as e:
        print(f"LOG: [run_comps_valuation] Comps failed with an exception: {e}")
        return None

# --- 3. MAIN CONTROLLER FUNCTION ---

def analyze_stock(ticker):
    """
    The main function that runs the entire pipeline for a given ticker and returns results.
    Returns two outputs for Gradio: a Markdown string for the title, and a DataFrame for the table.
    """
    if not ticker or not ticker.strip():
        return "", pd.DataFrame({"Error": ["Please enter a ticker symbol."]})
    
    ticker = ticker.strip().upper() # Sanitize user input
    print(f"\n--- [START] Full analysis for {ticker}... ---")
    
    try:
        stock_data, company_name = get_stock_data(ticker)
        
        financial_metrics = calculate_metrics(stock_data)
        wacc = calculate_wacc(stock_data)
        dcf_value = run_dcf_valuation(stock_data, financial_metrics, wacc)
        comps_value = run_comps_valuation(stock_data)
        current_price = yfinance.Ticker(ticker).history(period='1d')['Close'].iloc[-1]
        
        valid_valuations = [v for v in [dcf_value, comps_value] if v is not None and pd.notna(v) and v > 0]
        average_value = sum(valid_valuations) / len(valid_valuations) if valid_valuations else 0
        
        upside = (average_value / current_price) - 1 if current_price > 0 else 0
        decision = "Buy" if upside > 0.25 else "Hold/Not Buy"
        if average_value == 0: decision = "Analysis Failed"
        
        op_margin_series = financial_metrics['operatingMargin'].dropna()
        op_margin = op_margin_series.iloc[-1] if not op_margin_series.empty else np.nan
        beta = stock_data['profile'].get('beta')

        summary_data = {
            "Ticker": ticker, "Current Price": f"${current_price:.2f}",
            "DCF Value": f"${dcf_value:.2f}" if dcf_value else "N/A",
            "Comps Value": f"${comps_value:.2f}" if comps_value else "N/A",
            "Avg. Intrinsic Value": f"${average_value:.2f}",
            "Upside Potential": f"{upside:.2%}" if average_value > 0 else "N/A",
            "Decision": decision,
            "Operating Margin": f"{op_margin:.2%}" if pd.notna(op_margin) else "N/A",
            "Beta": f"{beta:.2f}" if pd.notna(beta) else "N/A"
        }
        print(f"--- [SUCCESS] Full analysis for {ticker} complete. ---\n")
        
        # Return both the title and the results DataFrame
        return f"## Analysis for: {company_name} ({ticker})", pd.DataFrame([summary_data]).set_index("Ticker")

    except Exception as e:
        print(f"--- [CRITICAL ERROR] Full analysis for {ticker} failed. ---")
        traceback.print_exc()
        print("--- End of Traceback ---")
        # Return an empty title and an error DataFrame
        return "", pd.DataFrame({"Error": [f"{e}"]})

# --- 4. GRADIO INTERFACE ---

with gr.Blocks(theme=gr.themes.Soft()) as iface:
    gr.Markdown("# Fundamental Analysis Stock Valuator")
    gr.Markdown("Enter any US stock ticker (e.g., AAPL, MSFT, GOOGL) and click 'Analyze' to run a DCF and Comps valuation.")
    
    with gr.Row():
        ticker_input = gr.Textbox(
            value="AAPL",
            label="Enter a Stock Ticker"
        )
        analyze_button = gr.Button("Analyze Stock")
    
    # New component to display the company name
    company_name_md = gr.Markdown()
    
    output_df = gr.DataFrame(label="Valuation Summary")
    
    analyze_button.click(
        fn=analyze_stock,
        inputs=ticker_input,
        outputs=[company_name_md, output_df] # The function now updates two components
    )

if __name__ == "__main__":
    iface.launch()