Financial Data Collector

# Python cache
__pycache__/
*.pyc
*.pyo

# Test outputs
*_financial_data.json

Security scan passed
Scanned at: 2026-06-13T19:44:42.093312
Tool: gitleaks + pattern-based validation
Content hash: 191228b4d56b8c3814596c77dd1ac78d6d1ac25ce4b2eb81db3d487ebef49a2b

#!/usr/bin/env python3
"""
Collect real financial data for a US publicly traded company.
Output: structured JSON consumable by downstream financial skills.

Usage:
    python collect_data.py TICKER [--years 5] [--output path/to/output.json]

Examples:
    python collect_data.py META
    python collect_data.py AAPL --years 3 --output /tmp/aapl_data.json
    python collect_data.py NVDA --years 5

Dependencies: yfinance, pandas (via uv inline or pip)
"""
# /// script
# requires-python = ">=3.11"
# dependencies = ["yfinance>=0.2.0", "pandas>=2.0.0"]
# ///

import argparse
import json
import sys
import time
from datetime import date
from pathlib import Path

import pandas as pd
import yfinance as yf

# Field name aliases — yfinance row indices vary across versions
FIELD_ALIASES = {
    "revenue": ["Total Revenue", "Revenue", "Operating Revenue"],
    "ebit": ["Operating Income", "EBIT"],
    "ebitda": ["EBITDA", "Normalized EBITDA"],
    "tax_expense": ["Tax Provision", "Income Tax Expense"],
    "net_income": ["Net Income", "Net Income Common Stockholders"],
    "capex": ["Capital Expenditure", "Capital Expenditures"],
    "operating_cash_flow": ["Operating Cash Flow", "Cash Flow From Continuing Operating Activities"],
    "depreciation_amortization": ["Depreciation And Amortization", "Depreciation Amortization Depletion"],
    "free_cash_flow": ["Free Cash Flow"],
    "change_in_nwc": ["Change In Working Capital", "Changes In Working Capital"],
    "sbc": ["Stock Based Compensation"],
    "total_debt": ["Total Debt"],
    "long_term_debt": ["Long Term Debt"],
    "short_term_debt": ["Short Long Term Debt", "Current Debt"],
    "cash": ["Cash And Cash Equivalents"],
    "short_investments": ["Other Short Term Investments", "Short Term Investments"],
    "current_assets": ["Current Assets"],
    "current_liabilities": ["Current Liabilities"],
    "total_assets": ["Total Assets"],
    "total_equity": ["Stockholders Equity", "Total Equity Gross Minority Interest"],
}


def safe_get(df: pd.DataFrame, field_key: str, col) -> float | None:
    """Safely extract a value from a DataFrame using alias chain."""
    aliases = FIELD_ALIASES.get(field_key, [field_key])
    for alias in aliases:
        if alias in df.index:
            val = df.loc[alias, col]
            if pd.notna(val):
                return float(val)
    return None


def get_year_col(df: pd.DataFrame, year: int):
    """Find the column matching a target year in a yfinance DataFrame."""
    matches = [c for c in df.columns if c.year == year]
    return matches[0] if matches else None


def collect_market_data(ticker_obj: yf.Ticker) -> dict:
    """Collect real-time market data."""
    info = ticker_obj.info
    price = info.get("currentPrice") or info.get("regularMarketPrice")
    shares_raw = info.get("sharesOutstanding")
    shares = shares_raw / 1e6 if shares_raw else None
    mcap_raw = info.get("marketCap")
    mcap = mcap_raw / 1e6 if mcap_raw else None
    beta = info.get("beta")

    return {
        "current_price": price,
        "shares_outstanding_millions": round(shares, 2) if shares else None,
        "market_cap_millions": round(mcap, 2) if mcap else None,
        "beta_5y_monthly": round(beta, 3) if beta else None,
    }


def collect_risk_free_rate() -> float | None:
    """Get 10Y Treasury yield as risk-free rate proxy."""
    try:
        tnx = yf.Ticker("^TNX")
        hist = tnx.history(period="5d")
        if hist.empty:
            return None
        yield_pct = hist["Close"].iloc[-1]
        return round(float(yield_pct) / 100, 4)  # Convert percentage to decimal
    except Exception:
        return None


def collect_income_statement(ticker_obj: yf.Ticker, years: list[int]) -> dict:
    """Collect income statement data for specified years."""
    try:
        financials = ticker_obj.financials
    except Exception:
        return {str(y): {"_source": "yfinance error"} for y in years}

    if financials is None or financials.empty:
        return {str(y): {"_source": "yfinance returned empty"} for y in years}

    result = {}
    nan_years = []
    for year in years:
        col = get_year_col(financials, year)
        if col is None:
            result[str(year)] = {"_source": f"yfinance has no column for {year}"}
            nan_years.append(year)
            continue

        revenue = safe_get(financials, "revenue", col)
        ebit = safe_get(financials, "ebit", col)

        if revenue is None and ebit is None:
            nan_years.append(year)
            result[str(year)] = {
                "revenue": None, "ebit": None, "ebitda": None,
                "tax_expense": None, "net_income": None,
                "_source": f"yfinance returned NaN for {year} — supplement from 10-K filing",
            }
        else:
            result[str(year)] = {
                "revenue": revenue,
                "ebit": ebit,
                "ebitda": safe_get(financials, "ebitda", col),
                "tax_expense": safe_get(financials, "tax_expense", col),
                "net_income": safe_get(financials, "net_income", col),
                "_source": "yfinance",
            }

    return result, nan_years


def collect_cash_flow(ticker_obj: yf.Ticker, years: list[int]) -> dict:
    """Collect cash flow data."""
    try:
        cf = ticker_obj.cashflow
    except Exception:
        return {str(y): {"_source": "yfinance error"} for y in years}, []

    if cf is None or cf.empty:
        return {str(y): {"_source": "yfinance returned empty"} for y in years}, years

    result = {}
    nan_years = []
    for year in years:
        col = get_year_col(cf, year)
        if col is None:
            result[str(year)] = {"_source": f"yfinance has no column for {year}"}
            nan_years.append(year)
            continue

        ocf = safe_get(cf, "operating_cash_flow", col)
        capex = safe_get(cf, "capex", col)

        if ocf is None and capex is None:
            nan_years.append(year)
            result[str(year)] = {
                "operating_cash_flow": None, "capex": None,
                "depreciation_amortization": None, "free_cash_flow": None,
                "change_in_nwc": None, "sbc": None,
                "_source": f"yfinance returned NaN for {year}",
            }
        else:
            result[str(year)] = {
                "operating_cash_flow": ocf,
                "capex": capex,  # Negative = outflow (preserved)
                "depreciation_amortization": safe_get(cf, "depreciation_amortization", col),
                "free_cash_flow": safe_get(cf, "free_cash_flow", col),
                "change_in_nwc": safe_get(cf, "change_in_nwc", col),
                "sbc": safe_get(cf, "sbc", col),
                "_source": "yfinance",
            }

    return result, nan_years


def collect_balance_sheet(ticker_obj: yf.Ticker, latest_year: int) -> dict:
    """Collect balance sheet data for the latest year."""
    try:
        bs = ticker_obj.balance_sheet
    except Exception:
        return {str(latest_year): {"_source": "yfinance error"}}

    if bs is None or bs.empty:
        return {str(latest_year): {"_source": "yfinance returned empty"}}

    col = get_year_col(bs, latest_year)
    if col is None:
        return {str(latest_year): {"_source": f"yfinance has no column for {latest_year}"}}

    total_debt = safe_get(bs, "total_debt", col)
    if total_debt is None:
        lt = safe_get(bs, "long_term_debt", col) or 0
        st = safe_get(bs, "short_term_debt", col) or 0
        total_debt = lt + st if (lt or st) else None

    cash = safe_get(bs, "cash", col)
    short_inv = safe_get(bs, "short_investments", col)
    cash_equiv = (cash or 0) + (short_inv or 0) if (cash is not None or short_inv is not None) else None
    net_debt = (total_debt - cash_equiv) if (total_debt is not None and cash_equiv is not None) else None

    return {
        str(latest_year): {
            "total_debt": total_debt,
            "cash_and_equivalents": cash_equiv,
            "net_debt": round(net_debt, 2) if net_debt is not None else None,
            "current_assets": safe_get(bs, "current_assets", col),
            "current_liabilities": safe_get(bs, "current_liabilities", col),
            "total_assets": safe_get(bs, "total_assets", col),
            "total_equity": safe_get(bs, "total_equity", col),
            "_source": "yfinance",
        }
    }


def collect_analyst_estimates(ticker_obj: yf.Ticker) -> dict:
    """Collect analyst consensus estimates."""
    result = {"revenue_next_fy": None, "revenue_fy_after": None, "eps_next_fy": None, "_source": "missing"}

    try:
        rev_est = ticker_obj.revenue_estimate
        if rev_est is not None and not rev_est.empty:
            if "0y" in rev_est.index:
                result["revenue_next_fy"] = float(rev_est.loc["0y", "avg"]) / 1e6 if pd.notna(rev_est.loc["0y", "avg"]) else None
            if "+1y" in rev_est.index:
                result["revenue_fy_after"] = float(rev_est.loc["+1y", "avg"]) / 1e6 if pd.notna(rev_est.loc["+1y", "avg"]) else None
            result["_source"] = "yfinance revenue_estimate"
    except Exception:
        pass

    try:
        eps = ticker_obj.eps_trend
        if eps is not None and not eps.empty:
            if "0y" in eps.index:
                val = eps.loc["0y", "current"] if "current" in eps.columns else eps.iloc[eps.index.get_loc("0y"), 0]
                result["eps_next_fy"] = float(val) if pd.notna(val) else None
    except Exception:
        pass

    return result


def main():
    parser = argparse.ArgumentParser(description="Collect financial data for a US public company")
    parser.add_argument("ticker", help="Stock ticker symbol (e.g., META, AAPL)")
    parser.add_argument("--years", type=int, default=5, help="Number of historical years (default: 5)")
    parser.add_argument("--output", help="Output JSON file path (default: {TICKER}_financial_data.json)")
    args = parser.parse_args()

    ticker_symbol = args.ticker.upper()
    current_year = date.today().year
    # Exclude current year if annual report not yet filed (typically before Q4 earnings)
    latest_full_year = current_year - 1
    target_years = list(range(latest_full_year - args.years + 1, latest_full_year + 1))

    output_path = args.output or f"{ticker_symbol}_financial_data.json"

    print(f"Collecting data for {ticker_symbol} ({target_years[0]}-{target_years[-1]})...")
    start_time = time.time()

    ticker_obj = yf.Ticker(ticker_symbol)

    # Verify ticker is valid
    info = ticker_obj.info
    if not info or info.get("regularMarketPrice") is None:
        print(f"ERROR: {ticker_symbol} not found or no market data available", file=sys.stderr)
        sys.exit(1)

    company_name = info.get("longName") or info.get("shortName") or ticker_symbol

    # Collect all data
    print("  Market data...", end=" ", flush=True)
    market_data = collect_market_data(ticker_obj)
    print("OK")

    print("  Risk-free rate...", end=" ", flush=True)
    rfr = collect_risk_free_rate()
    print(f"{rfr:.4f}" if rfr else "MISSING")

    print("  Income statement...", end=" ", flush=True)
    income_data, is_nan_years = collect_income_statement(ticker_obj, target_years)
    print(f"OK (NaN years: {is_nan_years or 'none'})")

    print("  Cash flow...", end=" ", flush=True)
    cf_data, cf_nan_years = collect_cash_flow(ticker_obj, target_years)
    print(f"OK (NaN years: {cf_nan_years or 'none'})")

    print("  Balance sheet...", end=" ", flush=True)
    bs_data = collect_balance_sheet(ticker_obj, latest_full_year)
    print("OK")

    print("  Analyst estimates...", end=" ", flush=True)
    estimates = collect_analyst_estimates(ticker_obj)
    print("OK" if estimates["_source"] != "missing" else "PARTIAL/MISSING")

    elapsed = round(time.time() - start_time, 1)
    all_nan_years = sorted(set(is_nan_years + cf_nan_years))

    # Assemble output
    output = {
        "ticker": ticker_symbol,
        "company_name": company_name,
        "data_date": date.today().isoformat(),
        "currency": "USD",
        "unit": "millions_usd",
        "data_sources": {
            "market_data": "yfinance (live)",
            "historical_financials": "yfinance annual financials",
            "risk_free_rate": "yfinance ^TNX (10Y Treasury)",
        },
        "market_data": market_data,
        "income_statement": income_data,
        "cash_flow": cf_data,
        "balance_sheet": bs_data,
        "wacc_inputs": {
            "risk_free_rate": rfr,
            "beta": market_data.get("beta_5y_monthly"),
            "credit_rating": None,
            "_source": "yfinance + ^TNX",
        },
        "analyst_estimates": estimates,
        "metadata": {
            "_capex_convention": "negative = cash outflow (yfinance convention)",
            "_fcf_note": "yfinance FCF = OperatingCF + CapEx. Does NOT deduct SBC.",
            "_nan_years": all_nan_years,
            "_collection_duration_seconds": elapsed,
            "_target_years": target_years,
        },
    }

    # Write output
    Path(output_path).write_text(json.dumps(output, indent=2, ensure_ascii=False))
    print(f"\nSaved to {output_path} ({elapsed}s)")

    # Summary
    if all_nan_years:
        print(f"\n⚠️  WARNING: Years {all_nan_years} have missing data (NaN from yfinance).")
        print("   Supplement from 10-K filings or SEC EDGAR before using in models.")

    # Quick sanity check
    if market_data["current_price"] and market_data["shares_outstanding_millions"]:
        computed_mcap = market_data["current_price"] * market_data["shares_outstanding_millions"]
        reported_mcap = market_data["market_cap_millions"]
        if reported_mcap and abs(computed_mcap - reported_mcap) / reported_mcap > 0.05:
            print(f"⚠️  Market cap mismatch: Price×Shares={computed_mcap:.0f}M vs Reported={reported_mcap:.0f}M")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Validate financial data JSON output from collect_data.py.
Checks completeness, consistency, and sanity of collected data.

Usage:
    python validate_data.py path/to/output.json

Returns JSON validation report to stdout.
"""
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///

import json
import sys
from pathlib import Path


def validate(data: dict) -> dict:
    """Validate financial data JSON. Returns validation report."""
    errors = []
    warnings = []

    # 1. Required top-level fields
    for field in ["ticker", "company_name", "data_date", "market_data",
                  "income_statement", "cash_flow", "balance_sheet", "wacc_inputs"]:
        if field not in data:
            errors.append(f"Missing required field: {field}")

    if errors:
        return {"status": "error", "errors": errors, "warnings": warnings}

    md = data["market_data"]

    # 2. Market data sanity
    if md.get("current_price") is not None:
        if md["current_price"] <= 0:
            errors.append(f"Invalid stock price: {md['current_price']}")
        if md["current_price"] > 10000:
            warnings.append(f"Unusually high stock price: ${md['current_price']}")

    if md.get("shares_outstanding_millions") is not None:
        if md["shares_outstanding_millions"] <= 0:
            errors.append(f"Invalid shares outstanding: {md['shares_outstanding_millions']}")

    if md.get("beta_5y_monthly") is not None:
        beta = md["beta_5y_monthly"]
        if beta < 0.1 or beta > 5.0:
            warnings.append(f"Unusual beta: {beta} (expected 0.3-3.0)")

    # 3. Market cap cross-check
    if md.get("current_price") and md.get("shares_outstanding_millions") and md.get("market_cap_millions"):
        computed = md["current_price"] * md["shares_outstanding_millions"]
        reported = md["market_cap_millions"]
        pct_diff = abs(computed - reported) / reported
        if pct_diff > 0.05:
            # yfinance sharesOutstanding is basic; marketCap may use diluted. Known discrepancy.
            warnings.append(f"Market cap mismatch ({pct_diff:.1%}): Price×Shares(basic)={computed:.0f}M vs Reported={reported:.0f}M. Likely basic vs diluted shares.")

    # 4. Income statement completeness
    is_data = data.get("income_statement", {})
    years_with_data = 0
    for year, vals in is_data.items():
        if isinstance(vals, dict) and vals.get("revenue") is not None:
            years_with_data += 1
            # Revenue should be positive
            if vals["revenue"] <= 0:
                warnings.append(f"Non-positive revenue in {year}: {vals['revenue']}")
            # EBIT margin sanity
            if vals.get("ebit") is not None and vals["revenue"] > 0:
                margin = vals["ebit"] / vals["revenue"]
                if margin < -1.0 or margin > 0.8:
                    warnings.append(f"Unusual EBIT margin in {year}: {margin:.1%}")

    if years_with_data == 0:
        errors.append("No income statement data available for any year")
    elif years_with_data < 3:
        warnings.append(f"Only {years_with_data} years of income statement data (recommend ≥3)")

    # 5. Cash flow: CapEx sign convention
    cf_data = data.get("cash_flow", {})
    for year, vals in cf_data.items():
        if isinstance(vals, dict) and vals.get("capex") is not None:
            if vals["capex"] > 0:
                warnings.append(f"CapEx is positive in {year} ({vals['capex']}). Expected negative (outflow).")

    # 6. Balance sheet: Net debt consistency
    bs_data = data.get("balance_sheet", {})
    for year, vals in bs_data.items():
        if isinstance(vals, dict):
            td = vals.get("total_debt")
            ce = vals.get("cash_and_equivalents")
            nd = vals.get("net_debt")
            if td is not None and ce is not None and nd is not None:
                expected_nd = td - ce
                if abs(expected_nd - nd) > 1.0:  # Allow $1M rounding
                    errors.append(f"Net debt inconsistency in {year}: total_debt({td}) - cash({ce}) = {expected_nd} ≠ {nd}")

    # 7. WACC inputs
    wacc = data.get("wacc_inputs", {})
    rfr = wacc.get("risk_free_rate")
    if rfr is not None:
        if rfr < 0 or rfr > 0.15:
            warnings.append(f"Unusual risk-free rate: {rfr:.2%} (expected 1-8%)")
    else:
        warnings.append("Risk-free rate is missing")

    # 8. NaN years tracking
    meta = data.get("metadata", {})
    nan_years = meta.get("_nan_years", [])
    if nan_years:
        warnings.append(f"NaN years detected: {nan_years}. Supplement from 10-K before using in models.")

    # 9. Data source attribution
    for section in ["income_statement", "cash_flow", "balance_sheet"]:
        section_data = data.get(section, {})
        for year, vals in section_data.items():
            if isinstance(vals, dict) and "_source" not in vals:
                warnings.append(f"Missing _source attribution in {section}.{year}")

    status = "error" if errors else ("warning" if warnings else "success")
    return {
        "status": status,
        "ticker": data.get("ticker"),
        "years_with_data": years_with_data,
        "errors": errors,
        "warnings": warnings,
        "error_count": len(errors),
        "warning_count": len(warnings),
    }


def main():
    if len(sys.argv) < 2:
        print("Usage: python validate_data.py <json_file>", file=sys.stderr)
        sys.exit(1)

    json_path = sys.argv[1]
    if not Path(json_path).exists():
        print(json.dumps({"status": "error", "errors": [f"File not found: {json_path}"]}))
        sys.exit(1)

    data = json.loads(Path(json_path).read_text())
    report = validate(data)

    print(json.dumps(report, indent=2))

    if report["status"] == "error":
        sys.exit(1)
    elif report["status"] == "warning":
        sys.exit(0)  # Warnings are OK, just informational
    else:
        sys.exit(0)


if __name__ == "__main__":
    main()