#!/usr/bin/env python3 """ Step 2: Financial Data Extraction & Screening ============================================== Fetches market data and quarterly financial statements for Toast (TOST) and its competitive set to establish a baseline for performance analysis. Tickers: - TOST: Toast Inc - SQ: Block Inc (Square) - VYX: NCR Voyix Corp (recent spin-off, may have limited history) - PAR: PAR Technology Corp - FI: Fiserv Inc Author: K-Dense Coding Agent Date: 2026-02-03 """ import os import sys import warnings import time from datetime import datetime, timedelta from pathlib import Path import pandas as pd import numpy as np import yfinance as yf import matplotlib.pyplot as plt # Suppress warnings for cleaner output warnings.filterwarnings('ignore') # Set seeds for reproducibility np.random.seed(42) # Configure matplotlib for non-interactive use plt.switch_backend('Agg') plt.rcParams['font.family'] = 'sans-serif' plt.rcParams['font.size'] = 10 plt.rcParams['axes.linewidth'] = 0.5 # Session paths SESSION_DIR = Path('/app/sandbox/session_20260203_085615_5f10d3d2357f') RESULTS_DIR = SESSION_DIR / 'results' FIGURES_DIR = SESSION_DIR / 'figures' # Ensure directories exist RESULTS_DIR.mkdir(exist_ok=True) FIGURES_DIR.mkdir(exist_ok=True) # Define ticker universe # Note: Block Inc (formerly Square) retained the SQ ticker after rebranding # Fiserv ticker is FISV, not FI TICKERS = { 'TOST': 'Toast Inc', 'SQ': 'Block Inc (formerly Square)', 'VYX': 'NCR Voyix Corp', 'PAR': 'PAR Technology Corp', 'FISV': 'Fiserv Inc' } def log_progress(message: str) -> None: """Log progress with timestamp.""" timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"[{timestamp}] {message}") def fetch_stock_data(tickers: list, years: int = 3) -> pd.DataFrame: """ Fetch historical daily stock price data for the given tickers. Parameters ---------- tickers : list List of ticker symbols years : int Number of years of historical data to fetch Returns ------- pd.DataFrame Daily price data with columns: Date, Ticker, Open, High, Low, Close, Volume, Adj Close """ log_progress(f"Fetching {years}-year daily price data for {len(tickers)} tickers...") end_date = datetime.now() start_date = end_date - timedelta(days=years * 365) all_data = [] for i, ticker in enumerate(tickers): try: log_progress(f" [{i+1}/{len(tickers)}] Downloading {ticker}...") stock = yf.Ticker(ticker) hist = stock.history(start=start_date, end=end_date, interval='1d') if hist.empty: log_progress(f" WARNING: No data returned for {ticker}") continue # Reset index to make Date a column hist = hist.reset_index() hist['Ticker'] = ticker # Select and rename columns cols_map = { 'Date': 'Date', 'Open': 'Open', 'High': 'High', 'Low': 'Low', 'Close': 'Close', 'Volume': 'Volume' } # Handle different column names (some versions use different casing) available_cols = hist.columns.tolist() selected_cols = ['Date', 'Ticker'] for target_col, source_col in cols_map.items(): if target_col in available_cols: selected_cols.append(target_col) elif source_col in available_cols: hist = hist.rename(columns={source_col: target_col}) selected_cols.append(target_col) # Ensure we have the basics if 'Close' in hist.columns: df = hist[['Date', 'Ticker', 'Open', 'High', 'Low', 'Close', 'Volume']].copy() df['Date'] = pd.to_datetime(df['Date']).dt.tz_localize(None) all_data.append(df) log_progress(f" Retrieved {len(df)} daily records ({df['Date'].min().date()} to {df['Date'].max().date()})") else: log_progress(f" WARNING: Missing 'Close' price for {ticker}") # Rate limiting time.sleep(0.5) except Exception as e: log_progress(f" ERROR fetching {ticker}: {str(e)}") continue if not all_data: log_progress("ERROR: No stock data retrieved for any ticker!") return pd.DataFrame() combined = pd.concat(all_data, ignore_index=True) log_progress(f"Total daily price records: {len(combined)}") return combined def fetch_financial_statements(tickers: list) -> pd.DataFrame: """ Fetch quarterly financial statements for the given tickers. Extracts: - Total Revenue - Gross Profit - Operating Income - Net Income - Cash & Equivalents (from balance sheet) - Total Debt (from balance sheet) Returns ------- pd.DataFrame Quarterly financial data in long format """ log_progress(f"Fetching quarterly financial statements for {len(tickers)} tickers...") all_financials = [] for i, ticker in enumerate(tickers): try: log_progress(f" [{i+1}/{len(tickers)}] Fetching financials for {ticker}...") stock = yf.Ticker(ticker) # === Income Statement (Quarterly) === income_q = stock.quarterly_income_stmt if income_q is None or income_q.empty: log_progress(f" WARNING: No quarterly income statement for {ticker}") income_data = pd.DataFrame() else: # Transpose: dates become rows income_data = income_q.T.copy() income_data.index.name = 'Date' income_data = income_data.reset_index() log_progress(f" Income statement: {len(income_data)} quarters") # === Balance Sheet (Quarterly) === balance_q = stock.quarterly_balance_sheet if balance_q is None or balance_q.empty: log_progress(f" WARNING: No quarterly balance sheet for {ticker}") balance_data = pd.DataFrame() else: balance_data = balance_q.T.copy() balance_data.index.name = 'Date' balance_data = balance_data.reset_index() log_progress(f" Balance sheet: {len(balance_data)} quarters") # === Cash Flow (Quarterly) === cf_q = stock.quarterly_cashflow if cf_q is None or cf_q.empty: log_progress(f" WARNING: No quarterly cash flow for {ticker}") else: log_progress(f" Cash flow: {len(cf_q.T)} quarters") # === Merge and Extract Key Metrics === if income_data.empty: log_progress(f" SKIPPING {ticker}: No income data available") continue # Create base dataframe from income statement df = pd.DataFrame() df['Date'] = pd.to_datetime(income_data['Date']).dt.tz_localize(None) df['Ticker'] = ticker # Extract income statement items with flexible column names income_cols = income_data.columns.tolist() # Revenue mapping (try multiple possible column names) revenue_candidates = ['Total Revenue', 'TotalRevenue', 'Revenue', 'Operating Revenue'] for col in revenue_candidates: if col in income_cols: df['Total Revenue'] = income_data[col].values break else: df['Total Revenue'] = np.nan # Gross Profit gp_candidates = ['Gross Profit', 'GrossProfit'] for col in gp_candidates: if col in income_cols: df['Gross Profit'] = income_data[col].values break else: df['Gross Profit'] = np.nan # Operating Income oi_candidates = ['Operating Income', 'OperatingIncome', 'EBIT'] for col in oi_candidates: if col in income_cols: df['Operating Income'] = income_data[col].values break else: df['Operating Income'] = np.nan # Net Income ni_candidates = ['Net Income', 'NetIncome', 'Net Income Common Stockholders'] for col in ni_candidates: if col in income_cols: df['Net Income'] = income_data[col].values break else: df['Net Income'] = np.nan # === Merge Balance Sheet Items === if not balance_data.empty: balance_cols = balance_data.columns.tolist() balance_data['Date'] = pd.to_datetime(balance_data['Date']).dt.tz_localize(None) # Cash & Equivalents cash_candidates = ['Cash And Cash Equivalents', 'CashAndCashEquivalents', 'Cash Cash Equivalents And Short Term Investments', 'Cash', 'Cash And Short Term Investments'] for col in cash_candidates: if col in balance_cols: balance_data['Cash'] = balance_data[col] break else: balance_data['Cash'] = np.nan # Total Debt debt_candidates = ['Total Debt', 'TotalDebt', 'Long Term Debt', 'Long Term Debt And Capital Lease Obligation'] for col in debt_candidates: if col in balance_cols: balance_data['Total Debt'] = balance_data[col] break else: # Try to calculate: Short Term + Long Term Debt st_debt = None lt_debt = None for col in ['Current Debt', 'CurrentDebt', 'Short Term Debt']: if col in balance_cols: st_debt = balance_data[col] break for col in ['Long Term Debt', 'LongTermDebt']: if col in balance_cols: lt_debt = balance_data[col] break if st_debt is not None and lt_debt is not None: balance_data['Total Debt'] = st_debt.fillna(0) + lt_debt.fillna(0) elif lt_debt is not None: balance_data['Total Debt'] = lt_debt else: balance_data['Total Debt'] = np.nan # Merge balance sheet data balance_subset = balance_data[['Date', 'Cash', 'Total Debt']].copy() df = df.merge(balance_subset, on='Date', how='left') else: df['Cash'] = np.nan df['Total Debt'] = np.nan # Calculate Net Debt = Total Debt - Cash df['Net Debt'] = df['Total Debt'].fillna(0) - df['Cash'].fillna(0) all_financials.append(df) # Rate limiting time.sleep(0.5) except Exception as e: log_progress(f" ERROR fetching financials for {ticker}: {str(e)}") import traceback traceback.print_exc() continue if not all_financials: log_progress("ERROR: No financial data retrieved for any ticker!") return pd.DataFrame() combined = pd.concat(all_financials, ignore_index=True) # Sort by ticker and date combined = combined.sort_values(['Ticker', 'Date'], ascending=[True, False]) log_progress(f"Total quarterly financial records: {len(combined)}") return combined def fetch_company_summary(tickers: list) -> pd.DataFrame: """ Fetch current valuation metrics for each ticker. Metrics: - Market Cap - Enterprise Value - Trailing PE - Forward PE - Price to Sales - Price to Book """ log_progress(f"Fetching company summary metrics for {len(tickers)} tickers...") summary_data = [] for i, ticker in enumerate(tickers): try: log_progress(f" [{i+1}/{len(tickers)}] Fetching summary for {ticker}...") stock = yf.Ticker(ticker) info = stock.info if not info: log_progress(f" WARNING: No info available for {ticker}") continue record = { 'Ticker': ticker, 'Company Name': TICKERS.get(ticker, ticker), 'Market Cap': info.get('marketCap'), 'Enterprise Value': info.get('enterpriseValue'), 'Trailing PE': info.get('trailingPE'), 'Forward PE': info.get('forwardPE'), 'Price to Sales': info.get('priceToSalesTrailing12Months'), 'Price to Book': info.get('priceToBook'), 'EV/Revenue': info.get('enterpriseToRevenue'), 'EV/EBITDA': info.get('enterpriseToEbitda'), 'Beta': info.get('beta'), '52 Week High': info.get('fiftyTwoWeekHigh'), '52 Week Low': info.get('fiftyTwoWeekLow'), 'Current Price': info.get('currentPrice') or info.get('previousClose'), 'Revenue Growth': info.get('revenueGrowth'), 'Gross Margins': info.get('grossMargins'), 'Operating Margins': info.get('operatingMargins'), 'Profit Margins': info.get('profitMargins'), 'Fetch Date': datetime.now().strftime('%Y-%m-%d') } summary_data.append(record) log_progress(f" Retrieved: Market Cap = ${info.get('marketCap', 0)/1e9:.1f}B") # Rate limiting time.sleep(0.3) except Exception as e: log_progress(f" ERROR fetching summary for {ticker}: {str(e)}") continue if not summary_data: log_progress("ERROR: No summary data retrieved!") return pd.DataFrame() df = pd.DataFrame(summary_data) log_progress(f"Summary data retrieved for {len(df)} companies") return df def create_revenue_validation_plot(financials_df: pd.DataFrame, output_path: Path) -> None: """ Create a validation plot showing quarterly revenue trends for all tickers. """ log_progress("Creating revenue trends validation plot...") if financials_df.empty: log_progress("WARNING: No financial data to plot") return # Filter for valid revenue data plot_data = financials_df[['Date', 'Ticker', 'Total Revenue']].dropna(subset=['Total Revenue']) if plot_data.empty: log_progress("WARNING: No revenue data available for plotting") return # Create figure fig, ax = plt.subplots(figsize=(12, 6)) # Define colors for each ticker colors = { 'TOST': '#FF6B6B', # Red 'SQ': '#4ECDC4', # Teal (Block Inc) 'VYX': '#45B7D1', # Blue 'PAR': '#96CEB4', # Green 'FISV': '#FFEAA7' # Yellow (Fiserv) } # Plot each ticker tickers_plotted = [] for ticker in plot_data['Ticker'].unique(): ticker_data = plot_data[plot_data['Ticker'] == ticker].sort_values('Date') if len(ticker_data) < 2: log_progress(f" Skipping {ticker}: insufficient data points ({len(ticker_data)})") continue # Convert revenue to billions for readability revenue_billions = ticker_data['Total Revenue'] / 1e9 color = colors.get(ticker, '#999999') ax.plot(ticker_data['Date'], revenue_billions, marker='o', markersize=5, linewidth=2, label=f'{ticker} ({TICKERS.get(ticker, ticker)})', color=color) tickers_plotted.append(ticker) log_progress(f" Plotted {ticker}: {len(ticker_data)} quarters") # Formatting ax.set_xlabel('Quarter', fontsize=11, fontweight='bold') ax.set_ylabel('Total Revenue (Billions USD)', fontsize=11, fontweight='bold') ax.set_title('Quarterly Revenue Trends: Toast (TOST) vs Competitive Set\n(Raw Data Validation)', fontsize=13, fontweight='bold') # Legend ax.legend(loc='upper left', fontsize=9, framealpha=0.9) # Grid ax.grid(True, alpha=0.3, linestyle='--') # Rotate x-axis labels plt.xticks(rotation=45, ha='right') # Add data freshness note today = datetime.now().strftime('%Y-%m-%d') ax.annotate(f'Data as of: {today}', xy=(0.99, 0.01), xycoords='axes fraction', fontsize=8, ha='right', va='bottom', style='italic', color='gray') # Tight layout plt.tight_layout() # Save plt.savefig(output_path, dpi=300, bbox_inches='tight', facecolor='white') plt.close() log_progress(f"Revenue trends plot saved to: {output_path}") log_progress(f" Tickers plotted: {', '.join(tickers_plotted)}") def main(): """Main execution function.""" log_progress("=" * 60) log_progress("STEP 2: Financial Data Extraction & Screening") log_progress("=" * 60) tickers = list(TICKERS.keys()) log_progress(f"Target tickers: {tickers}") # === 1. Fetch Daily Stock Price Data === log_progress("\n[1/4] Fetching daily stock price data...") market_data = fetch_stock_data(tickers, years=3) if not market_data.empty: market_output = RESULTS_DIR / 'market_data_daily.csv' market_data.to_csv(market_output, index=False) log_progress(f"Saved market data: {market_output}") log_progress(f" Records: {len(market_data)}") log_progress(f" Date range: {market_data['Date'].min().date()} to {market_data['Date'].max().date()}") else: log_progress("WARNING: No market data saved (empty DataFrame)") # === 2. Fetch Quarterly Financial Statements === log_progress("\n[2/4] Fetching quarterly financial statements...") financials_data = fetch_financial_statements(tickers) if not financials_data.empty: financials_output = RESULTS_DIR / 'financial_data_quarterly.csv' financials_data.to_csv(financials_output, index=False) log_progress(f"Saved financial data: {financials_output}") log_progress(f" Records: {len(financials_data)}") # Validate data coverage for ticker in tickers: ticker_data = financials_data[financials_data['Ticker'] == ticker] quarters = len(ticker_data) log_progress(f" {ticker}: {quarters} quarters") else: log_progress("WARNING: No financial data saved (empty DataFrame)") # === 3. Fetch Company Summary/Valuation Metrics === log_progress("\n[3/4] Fetching company summary metrics...") summary_data = fetch_company_summary(tickers) if not summary_data.empty: summary_output = RESULTS_DIR / 'company_summary.csv' summary_data.to_csv(summary_output, index=False) log_progress(f"Saved company summary: {summary_output}") # Display key metrics log_progress("\n Valuation Summary:") for _, row in summary_data.iterrows(): ticker = row['Ticker'] mcap = row.get('Market Cap', 0) pe = row.get('Trailing PE', 'N/A') mcap_str = f"${mcap/1e9:.1f}B" if pd.notna(mcap) else "N/A" pe_str = f"{pe:.1f}" if pd.notna(pe) and pe != 'N/A' else "N/A" log_progress(f" {ticker}: Market Cap = {mcap_str}, Trailing P/E = {pe_str}") else: log_progress("WARNING: No summary data saved (empty DataFrame)") # === 4. Generate Validation Plot === log_progress("\n[4/4] Generating revenue trends validation plot...") plot_output = FIGURES_DIR / 'revenue_trends_raw.png' create_revenue_validation_plot(financials_data, plot_output) # === Summary === log_progress("\n" + "=" * 60) log_progress("STEP 2 COMPLETE: Financial Data Extraction") log_progress("=" * 60) log_progress("\nOutput Files:") log_progress(f" - {RESULTS_DIR / 'market_data_daily.csv'}") log_progress(f" - {RESULTS_DIR / 'financial_data_quarterly.csv'}") log_progress(f" - {RESULTS_DIR / 'company_summary.csv'}") log_progress(f" - {FIGURES_DIR / 'revenue_trends_raw.png'}") # Validate success criteria log_progress("\nValidation Checks:") checks_passed = 0 total_checks = 4 # Check 1: Data for all tickers if not market_data.empty: tickers_with_data = market_data['Ticker'].unique() log_progress(f" [CHECK] Tickers with market data: {len(tickers_with_data)}/5 - {list(tickers_with_data)}") if len(tickers_with_data) >= 4: checks_passed += 1 # Check 2: At least 8 quarters of financial data if not financials_data.empty: for ticker in tickers: quarters = len(financials_data[financials_data['Ticker'] == ticker]) if quarters >= 8: log_progress(f" [CHECK] {ticker}: {quarters} quarters >= 8 - PASS") else: log_progress(f" [CHECK] {ticker}: {quarters} quarters < 8 - PARTIAL (new company or spin-off)") # Count as pass if most tickers have 8+ quarters tickers_with_8q = sum(1 for t in tickers if len(financials_data[financials_data['Ticker'] == t]) >= 8) if tickers_with_8q >= 3: checks_passed += 1 # Check 3: Revenue and Gross Profit columns exist if not financials_data.empty: has_revenue = 'Total Revenue' in financials_data.columns and financials_data['Total Revenue'].notna().any() has_gp = 'Gross Profit' in financials_data.columns and financials_data['Gross Profit'].notna().any() log_progress(f" [CHECK] Revenue column: {'PASS' if has_revenue else 'FAIL'}") log_progress(f" [CHECK] Gross Profit column: {'PASS' if has_gp else 'FAIL'}") if has_revenue and has_gp: checks_passed += 1 # Check 4: Summary data exists if not summary_data.empty and len(summary_data) >= 3: log_progress(f" [CHECK] Company summary data: {len(summary_data)} companies - PASS") checks_passed += 1 log_progress(f"\nOverall: {checks_passed}/{total_checks} validation checks passed") return checks_passed >= 3 # Success if at least 3 of 4 checks pass if __name__ == '__main__': success = main() sys.exit(0 if success else 1)