Financial Analysis Earnings

#!/usr/bin/env bash
# Earnings Analysis — zero dependencies, runs with system Python
set -e
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
exec python3 "$SCRIPT_DIR/scripts/earnings_report.py" "$@"

#!/usr/bin/env python3
"""
One-command earnings data aggregator.

Usage:
    python earnings_report.py AAPL
    python earnings_report.py MSFT --peers AAPL GOOGL

Output: JSON with quarterly financials (SEC EDGAR), current quote, and optional peer comparison.

No pip dependencies required (SEC EDGAR uses stdlib; Yahoo Finance quote uses urllib).
"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from get_fundamentals import fetch_fundamentals
from get_quote import fetch_quote
from get_comps import fetch_comps


def run_earnings_report(symbol: str, peers: list[str] | None = None) -> dict:
    symbol = symbol.upper()

    try:
        quarterly = fetch_fundamentals(symbol, form="10-Q", periods=8)
    except Exception as e:
        quarterly = {"entity_name": symbol, "periods": [], "error": str(e)}

    try:
        annual = fetch_fundamentals(symbol, form="10-K", periods=3)
    except Exception as e:
        annual = {"periods": [], "error": str(e)}

    try:
        quote = fetch_quote([symbol])
        quote_obj = quote["quotes"][0] if quote.get("quotes") and len(quote["quotes"]) > 0 else None
    except Exception:
        quote_obj = None

    result = {
        "symbol": symbol,
        "entity_name": quarterly.get("entity_name", symbol),
        "quote": quote_obj,
        "quarterly_financials": quarterly.get("periods", []),
        "annual_financials": annual.get("periods", []),
    }

    if peers:
        all_symbols = [symbol] + [p.upper() for p in peers]
        try:
            comps = fetch_comps(all_symbols)
            result["peer_comparison"] = comps.get("companies", [])
        except Exception as e:
            result["peer_comparison"] = {"error": str(e)}

    return result


def main():
    parser = argparse.ArgumentParser(description="Earnings data aggregator (SEC EDGAR + Yahoo Finance)")
    parser.add_argument("symbol", help="Stock ticker (e.g., AAPL)")
    parser.add_argument("--peers", nargs="+", help="Peer symbols for comparison (e.g., MSFT GOOGL)")

    args = parser.parse_args()
    result = run_earnings_report(args.symbol, args.peers)
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Comparable company analysis using SEC EDGAR data.

Usage:
    python get_comps.py AAPL MSFT GOOGL META

Output: JSON with financial metrics for each company (margins, growth, multiples).
        P/E and P/S require market cap from Yahoo Finance (optional dependency).

No required dependencies — uses SEC EDGAR (stdlib only).
"""

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(__file__))
from get_fundamentals import fetch_fundamentals


def _try_get_market_cap(symbol: str):
    """Try to get market cap from Yahoo Finance. Returns None if unavailable."""
    try:
        from get_quote import fetch_quote
        result = fetch_quote([symbol])
        quotes = result.get("quotes", [])
        if quotes and "market_cap" in quotes[0]:
            return quotes[0].get("market_cap")
    except Exception:
        pass
    return None


def fetch_comps(symbols: list[str]) -> dict:
    if len(symbols) > 10:
        return {"error": "Maximum 10 symbols allowed", "companies": []}

    companies = []

    for sym in symbols:
        sym = sym.upper()
        try:
            # Get 2 years of annual data for YoY growth
            data = fetch_fundamentals(sym, form="10-K", periods=2)
            if data.get("error"):
                companies.append({"symbol": sym, "error": data["error"]})
                continue

            periods = data.get("periods", [])
            if not periods:
                companies.append({"symbol": sym, "entity_name": data.get("entity_name", sym), "error": "No annual data"})
                continue

            latest = periods[0]
            prior = periods[1] if len(periods) > 1 else None

            revenue = latest.get("revenue")
            gross_profit = latest.get("gross_profit")
            operating_income = latest.get("operating_income")
            net_income = latest.get("net_income")
            eps = latest.get("eps_diluted")

            # Margins
            gross_margin = (gross_profit / revenue * 100) if revenue and gross_profit else None
            op_margin = (operating_income / revenue * 100) if revenue and operating_income else None
            net_margin = (net_income / revenue * 100) if revenue and net_income else None

            # YoY growth
            prior_rev = prior.get("revenue") if prior else None
            prior_ni = prior.get("net_income") if prior else None
            rev_growth = ((revenue - prior_rev) / abs(prior_rev) * 100) if revenue and prior_rev and prior_rev != 0 else None
            ni_growth = ((net_income - prior_ni) / abs(prior_ni) * 100) if net_income is not None and prior_ni and prior_ni != 0 else None

            # Try to get market cap for multiples
            market_cap = _try_get_market_cap(sym)
            pe = (market_cap / net_income) if market_cap and net_income and net_income > 0 else None
            ps = (market_cap / revenue) if market_cap and revenue and revenue > 0 else None

            comp = {
                "symbol": sym,
                "entity_name": data.get("entity_name", sym),
                "fiscal_year": latest.get("fiscal_year"),
                "market_cap": market_cap,
                "revenue": revenue,
                "gross_profit": gross_profit,
                "operating_income": operating_income,
                "net_income": net_income,
                "eps_diluted": eps,
                "gross_margin_pct": round(gross_margin, 1) if gross_margin is not None else None,
                "operating_margin_pct": round(op_margin, 1) if op_margin is not None else None,
                "net_margin_pct": round(net_margin, 1) if net_margin is not None else None,
                "revenue_growth_yoy_pct": round(rev_growth, 1) if rev_growth is not None else None,
                "net_income_growth_yoy_pct": round(ni_growth, 1) if ni_growth is not None else None,
                "pe_ratio": round(pe, 1) if pe is not None else None,
                "ps_ratio": round(ps, 1) if ps is not None else None,
            }
            companies.append(comp)

        except Exception as e:
            companies.append({"symbol": sym, "error": str(e)})

    return {"companies": companies}


def main():
    parser = argparse.ArgumentParser(description="Comparable company analysis via SEC EDGAR")
    parser.add_argument("symbols", nargs="+", help="Stock symbols (e.g., AAPL MSFT GOOGL)")
    args = parser.parse_args()

    result = fetch_comps([s.upper() for s in args.symbols])
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Fetch quarterly and annual financial statements from SEC EDGAR XBRL API.

Usage:
    python get_fundamentals.py <SYMBOL> [--form 10-K|10-Q] [--periods N]

Output: JSON with an array of financial periods.

Data source: SEC EDGAR (https://data.sec.gov)
No API key required. No external dependencies.
Rate limited to ~8 req/s per SEC guidelines.

Important: 10-Q data is YTD cumulative. Derive standalone quarters by subtraction:
  Q1 standalone = Q1 YTD
  Q2 standalone = H1 YTD - Q1 YTD
  Q3 standalone = 9M YTD - H1 YTD
  Q4 standalone = Full-year (10-K) - 9M YTD
"""

import argparse
import json
import sys
import time
import urllib.request
from datetime import datetime

BASE_URL = "https://data.sec.gov"
TICKER_URL = "https://www.sec.gov/files/company_tickers.json"
USER_AGENT = "finchat-skills contact@finchat.ai"
MIN_REQUEST_INTERVAL = 0.12  # ~8 req/s

_last_request_at = 0.0
_ticker_cik_map = None


def _throttle():
    global _last_request_at
    elapsed = time.time() - _last_request_at
    if elapsed < MIN_REQUEST_INTERVAL:
        time.sleep(MIN_REQUEST_INTERVAL - elapsed)
    _last_request_at = time.time()


def _get_json(url: str) -> dict:
    _throttle()
    req = urllib.request.Request(url)
    req.add_header("User-Agent", USER_AGENT)
    req.add_header("Accept", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read())
    except (urllib.request.URLError, urllib.error.HTTPError) as e:
        raise RuntimeError(f"SEC EDGAR request failed for {url}: {e}")


def _ensure_ticker_map() -> dict[str, int]:
    global _ticker_cik_map
    if _ticker_cik_map is not None:
        return _ticker_cik_map
    raw = _get_json(TICKER_URL)
    _ticker_cik_map = {}
    for entry in raw.values():
        _ticker_cik_map[entry["ticker"].upper()] = entry["cik_str"]
    return _ticker_cik_map


def _lookup_cik(ticker: str) -> int:
    m = _ensure_ticker_map()
    cik = m.get(ticker.upper())
    if cik is None:
        raise ValueError(f"Ticker {ticker!r} not found in SEC EDGAR")
    return int(cik)


# ── XBRL parsing ──


def _first_concept(usgaap: dict, *names: str):
    """Return the concept with the most recent data among the given names."""
    best = None
    best_end = ""
    for name in names:
        concept = usgaap.get(name)
        if not concept:
            continue
        units = concept.get("units", {})
        for unit_facts in units.values():
            for f in unit_facts:
                end = f.get("end", "")
                if end > best_end:
                    best_end = end
                    best = concept
            break
    return best


def _facts_for_form(concept, form: str) -> dict[str, float]:
    """Extract period_end -> value map for a filing form type."""
    m = {}
    if not concept:
        return m
    units = concept.get("units", {})
    for unit_facts in units.values():
        for f in unit_facts:
            if f.get("form") == form:
                end = f.get("end", "")
                if end not in m:
                    m[end] = f.get("val", 0)
        break
    return m


def _anchor_periods(concept, form: str, n: int) -> list[dict]:
    """Return the most recent N filing facts for a form type."""
    if not concept:
        return []
    facts = []
    units = concept.get("units", {})
    for unit_facts in units.values():
        for f in unit_facts:
            if f.get("form") == form:
                facts.append(f)
        break
    facts.sort(key=lambda f: f.get("end", ""), reverse=True)
    seen = set()
    deduped = []
    for f in facts:
        end = f.get("end", "")
        if end not in seen:
            seen.add(end)
            deduped.append(f)
    return deduped[:n] if n > 0 else deduped


def _fiscal_quarter_from_period(start: str, end: str, form: str) -> tuple[int, bool]:
    """Derive fiscal quarter and YTD flag from filing dates.
    10-K -> quarter=4, isYTD=False
    10-Q ~3mo -> Q1, ~6mo -> Q2, ~9mo -> Q3, all isYTD=True
    """
    if form == "10-K":
        return 4, False
    if not start or not end:
        return 0, True
    try:
        t0 = datetime.strptime(start, "%Y-%m-%d")
        t1 = datetime.strptime(end, "%Y-%m-%d")
    except ValueError:
        return 0, True
    days = (t1 - t0).days
    if days < 105:
        return 1, True
    elif days < 196:
        return 2, True
    else:
        return 3, True


def _period_label(start: str, end: str, fy: int, fq: int, form: str) -> str:
    """Build human-readable label, e.g. 'FY2025 Q1 (Oct-Dec 2024)'."""
    if not start or not end:
        return f"FY{fy} Annual" if form == "10-K" else f"FY{fy} Q{fq}"
    try:
        t0 = datetime.strptime(start, "%Y-%m-%d")
        t1 = datetime.strptime(end, "%Y-%m-%d")
    except ValueError:
        return f"FY{fy} Annual" if form == "10-K" else f"FY{fy} Q{fq}"
    s = t0.strftime("%b %Y")
    e = t1.strftime("%b %Y")
    if form == "10-K":
        return f"FY{fy} Annual ({s}\u2013{e})"
    return f"FY{fy} Q{fq} ({s}\u2013{e})"


def _eps_for_period(eps: float, is_ytd: bool, quarter: int):
    """Return EPS only when directly usable as standalone figure.
    Q2/Q3 YTD EPS is NOT additive — zeroed to prevent misuse."""
    if not is_ytd:
        return eps  # 10-K full-year
    if quarter == 1:
        return eps  # Q1 YTD == Q1 standalone
    return None  # Q2/Q3: must compute from standalone net_income / shares


def _derived_gross_profit(gp, revenue, cost_of_revenue):
    if gp is not None:
        return gp
    if revenue and cost_of_revenue:
        return revenue - cost_of_revenue
    return None


def _normalize_shares_scale(shares, net_income, eps):
    """Correct for companies reporting shares in thousands or millions."""
    if not shares or shares <= 0 or not eps or not net_income:
        return shares if shares else None
    implied = abs(net_income) / abs(eps)
    ratio = implied / shares
    if 500 <= ratio < 5000:
        return shares * 1000
    elif 500_000 <= ratio < 5_000_000:
        return shares * 1_000_000
    return shares


# ── Main fetch ──


def fetch_fundamentals(symbol: str, form: str = "10-K", periods: int = 4) -> dict:
    symbol = symbol.upper()
    if periods <= 0:
        periods = 4
    elif periods > 8:
        periods = 8

    cik = _lookup_cik(symbol)
    url = f"{BASE_URL}/api/xbrl/companyfacts/CIK{cik:010d}.json"
    raw = _get_json(url)

    entity_name = raw.get("entityName", symbol)
    usgaap = raw.get("facts", {}).get("us-gaap")
    if not usgaap:
        return {"symbol": symbol, "entity_name": entity_name, "error": "No US-GAAP data available", "periods": []}

    rev_concept = _first_concept(usgaap,
        "Revenues",
        "RevenueFromContractWithCustomerExcludingAssessedTax",
        "RevenueFromContractWithCustomerIncludingAssessedTax",
        "SalesRevenueNet",
        "SalesRevenueGoodsNet",
    )

    anchor_concept = rev_concept or usgaap.get("NetIncomeLoss")
    if not anchor_concept:
        return {"symbol": symbol, "entity_name": entity_name, "error": "No financial data found", "periods": []}

    anchors = _anchor_periods(anchor_concept, form, periods)
    if not anchors:
        return {"symbol": symbol, "entity_name": entity_name, "error": f"No {form} filings found", "periods": []}

    # Build period->value lookup maps
    rev_map = _facts_for_form(rev_concept, form)
    ni_map = _facts_for_form(_first_concept(usgaap, "NetIncomeLoss", "NetIncomeLossAttributableToParent", "ProfitLoss"), form)
    gp_map = _facts_for_form(_first_concept(usgaap, "GrossProfit", "GrossProfitLoss"), form)
    cor_map = _facts_for_form(_first_concept(usgaap, "CostOfRevenue", "CostOfGoodsAndServicesSold", "CostOfGoodsSold"), form)
    oi_map = _facts_for_form(_first_concept(usgaap, "OperatingIncomeLoss"), form)
    eps_map = _facts_for_form(_first_concept(usgaap, "EarningsPerShareDiluted"), form)
    ocf_map = _facts_for_form(_first_concept(usgaap, "NetCashProvidedByUsedInOperatingActivities"), form)
    capex_map = _facts_for_form(_first_concept(usgaap,
        "PaymentsToAcquirePropertyPlantAndEquipment",
        "PaymentsForCapitalImprovements",
        "PaymentsToAcquireOtherProductiveAssets",
    ), form)
    assets_map = _facts_for_form(usgaap.get("Assets"), form)
    liab_map = _facts_for_form(usgaap.get("Liabilities"), form)
    equity_map = _facts_for_form(_first_concept(usgaap,
        "StockholdersEquity",
        "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest",
    ), form)
    ltd_map = _facts_for_form(_first_concept(usgaap, "LongTermDebt", "LongTermDebtNoncurrent"), form)
    interest_map = _facts_for_form(_first_concept(usgaap, "InterestExpense", "InterestAndDebtExpense"), form)
    tax_map = _facts_for_form(_first_concept(usgaap, "IncomeTaxExpenseBenefit"), form)
    pretax_map = _facts_for_form(_first_concept(usgaap,
        "IncomeLossFromContinuingOperationsBeforeIncomeTaxesExtraordinaryItemsNoncontrollingInterest",
        "IncomeLossFromContinuingOperationsBeforeIncomeTaxesMinorityInterestAndIncomeLossFromEquityMethodInvestments",
    ), form)
    cash_map = _facts_for_form(_first_concept(usgaap,
        "CashAndCashEquivalentsAtCarryingValue",
        "CashCashEquivalentsAndShortTermInvestments",
        "CashCashEquivalentsRestrictedCashAndRestrictedCashEquivalents",
    ), form)
    shares_map = _facts_for_form(_first_concept(usgaap,
        "CommonStockSharesOutstanding",
        "WeightedAverageNumberOfDilutedSharesOutstanding",
    ), form)

    results = []
    for af in anchors:
        end = af.get("end", "")
        start = af.get("start", "")
        fy = af.get("fy", 0)

        q, is_ytd = _fiscal_quarter_from_period(start, end, form)
        raw_eps = eps_map.get(end, 0)
        eps = _eps_for_period(raw_eps, is_ytd, q)
        net_income = ni_map.get(end, 0)
        shares = _normalize_shares_scale(shares_map.get(end, 0), net_income, raw_eps)
        revenue = rev_map.get(end, 0)
        gross_profit = _derived_gross_profit(gp_map.get(end), revenue, cor_map.get(end, 0))

        period = {
            "period": end,
            "form": form,
            "fiscal_year": fy,
            "fiscal_quarter": q,
            "is_ytd": is_ytd,
            "period_label": _period_label(start, end, fy, q, form),
            "revenue": revenue if revenue else None,
            "gross_profit": gross_profit,
            "operating_income": oi_map.get(end),
            "net_income": net_income if net_income else None,
            "eps_diluted": eps,
            "pretax_income": pretax_map.get(end),
            "interest_expense": interest_map.get(end),
            "income_tax_expense": tax_map.get(end),
            "total_assets": assets_map.get(end),
            "total_liabilities": liab_map.get(end),
            "shareholders_equity": equity_map.get(end),
            "cash_and_equivalents": cash_map.get(end),
            "long_term_debt": ltd_map.get(end),
            "shares_outstanding": shares,
            "operating_cash_flow": ocf_map.get(end),
            "capital_expenditure": capex_map.get(end),
        }
        results.append(period)

    return {
        "symbol": symbol,
        "entity_name": entity_name,
        "form": form,
        "periods": results,
    }


def main():
    parser = argparse.ArgumentParser(description="Fetch financial fundamentals from SEC EDGAR")
    parser.add_argument("symbol", help="Stock ticker symbol (e.g., AAPL)")
    parser.add_argument("--form", choices=["10-K", "10-Q"], default="10-K", help="Filing type (default: 10-K)")
    parser.add_argument("--periods", type=int, default=4, help="Number of periods to fetch (default: 4)")

    args = parser.parse_args()
    result = fetch_fundamentals(args.symbol.upper(), args.form, args.periods)
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Fetch current stock quote from Yahoo Finance.

Usage:
    python get_quote.py AAPL [MSFT GOOGL ...]

Output: JSON with current price, market cap, shares outstanding, PE, etc.

No pip dependencies — uses Yahoo Finance endpoints via urllib with cookie+crumb auth.
"""

import argparse
import json
import sys
import urllib.request
import urllib.error
import http.cookiejar

YF_CHART_URL = "https://query1.finance.yahoo.com/v8/finance/chart/{symbol}?range=1d&interval=1d"
YF_CRUMB_URL = "https://query2.finance.yahoo.com/v1/test/getcrumb"
YF_SUMMARY_URL = "https://query2.finance.yahoo.com/v10/finance/quoteSummary/{symbol}?modules=price,defaultKeyStatistics,summaryDetail&crumb={crumb}"

_opener = None
_crumb = None


def _init_session():
    """Initialize cookie jar and fetch crumb (one-time per process)."""
    global _opener, _crumb
    if _opener is not None:
        return

    cj = http.cookiejar.CookieJar()
    _opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cj))
    _opener.addheaders = [("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)")]

    # Step 1: hit a Yahoo page to get cookies
    try:
        _opener.open("https://fc.yahoo.com/", timeout=10)
    except urllib.error.HTTPError:
        pass  # 404 is expected, but cookies are set
    except Exception:
        pass

    # Step 2: fetch crumb using those cookies
    try:
        resp = _opener.open(YF_CRUMB_URL, timeout=10)
        _crumb = resp.read().decode("utf-8").strip()
    except Exception:
        _crumb = None


def _yf_get(url: str, use_session: bool = False) -> dict | None:
    """Fetch JSON from Yahoo Finance."""
    try:
        if use_session and _opener:
            resp = _opener.open(url, timeout=10)
        else:
            req = urllib.request.Request(url)
            req.add_header("User-Agent", "Mozilla/5.0")
            resp = urllib.request.urlopen(req, timeout=10)
        return json.loads(resp.read())
    except (urllib.error.URLError, urllib.error.HTTPError, json.JSONDecodeError):
        return None


def _raw_val(obj):
    """Extract raw value from Yahoo Finance formatted object like {'raw': 1.23, 'fmt': '1.23'}."""
    if isinstance(obj, dict):
        return obj.get("raw")
    return obj


def fetch_quote(symbols: list[str]) -> dict:
    _init_session()
    quotes = []

    for sym in symbols:
        sym = sym.upper()
        try:
            # Chart endpoint for price (reliable, no auth needed)
            data = _yf_get(YF_CHART_URL.format(symbol=sym))
            if not data:
                quotes.append({"symbol": sym, "error": "Failed to fetch quote"})
                continue

            result_list = data.get("chart", {}).get("result")
            if not result_list:
                quotes.append({"symbol": sym, "error": "No chart data"})
                continue

            meta = result_list[0].get("meta", {})

            quote = {
                "symbol": sym,
                "name": meta.get("longName") or meta.get("shortName") or sym,
                "current_price": meta.get("regularMarketPrice"),
                "previous_close": meta.get("previousClose") or meta.get("chartPreviousClose"),
                "market_cap": None,
                "shares_outstanding": None,
                "enterprise_value": None,
                "volume": meta.get("regularMarketVolume"),
                "fifty_two_week_high": meta.get("fiftyTwoWeekHigh"),
                "fifty_two_week_low": meta.get("fiftyTwoWeekLow"),
                "pe_trailing": None,
                "pe_forward": None,
                "eps_trailing": None,
                "beta": None,
                "dividend_yield": None,
                "currency": meta.get("currency"),
                "exchange": meta.get("exchangeName"),
            }

            # Try quoteSummary with crumb for richer data
            if _crumb:
                summary = _yf_get(
                    YF_SUMMARY_URL.format(symbol=sym, crumb=urllib.request.quote(_crumb)),
                    use_session=True,
                )
                if summary:
                    modules = summary.get("quoteSummary", {}).get("result")
                    if modules:
                        mod = modules[0]
                        price = mod.get("price", {})
                        stats = mod.get("defaultKeyStatistics", {})
                        detail = mod.get("summaryDetail", {})

                        quote["name"] = _raw_val(price.get("longName")) or price.get("shortName") or quote["name"]
                        quote["market_cap"] = _raw_val(price.get("marketCap"))
                        quote["shares_outstanding"] = _raw_val(stats.get("sharesOutstanding")) or _raw_val(price.get("sharesOutstanding"))
                        quote["enterprise_value"] = _raw_val(stats.get("enterpriseValue"))
                        quote["pe_trailing"] = _raw_val(detail.get("trailingPE"))
                        quote["pe_forward"] = _raw_val(stats.get("forwardPE"))
                        quote["eps_trailing"] = _raw_val(stats.get("trailingEps"))
                        quote["beta"] = _raw_val(stats.get("beta"))
                        quote["dividend_yield"] = _raw_val(detail.get("dividendYield"))

            quotes.append(quote)

        except Exception as e:
            quotes.append({"symbol": sym, "error": str(e)})

    return {"quotes": quotes}


def main():
    parser = argparse.ArgumentParser(description="Fetch stock quotes")
    parser.add_argument("symbols", nargs="+", help="Stock symbols")
    args = parser.parse_args()

    result = fetch_quote([s.upper() for s in args.symbols])
    print(json.dumps(result, indent=2, default=str))


if __name__ == "__main__":
    main()