ingestion/yf_fallback.py

"""ingestion/yf_fallback.py — fill NULL EdgarData fields from yfinance + Alpha Vantage.

Used by ingest.py immediately before upsert_metrics. Only activates when at least one
critical field (revenue, eps, gross_margin, operating_margin, free_cash_flow) is None.
"""
from __future__ import annotations

import dataclasses
from datetime import datetime, timedelta
from typing import Optional

import yfinance as yf

from ingestion.edgar import EdgarData
from ingestion.alphavantage import fetch_earnings

_CRITICAL = ("revenue", "eps", "gross_margin", "operating_margin", "free_cash_flow")
_YF_WINDOW_DAYS = 90  # period-end must be within this many days before filing_date

_YF_CACHE: dict[str, dict] = {}


def fill_missing_metrics(edgar: EdgarData) -> EdgarData:
    """Fill NULL EdgarData fields from Alpha Vantage and yfinance. Returns patched copy."""
    if all(getattr(edgar, f) is not None for f in _CRITICAL):
        return edgar

    updates: dict = {}
    contexts = dict(edgar.metric_contexts)
    warnings = list(edgar.quality_warnings)
    is_annual = edgar.form_type == "10-K"

    if edgar.eps is None:
        eps, eps_context = _av_eps_with_context(edgar.ticker, edgar.filing_date, is_annual)
        if eps is not None:
            updates["eps"] = eps
            contexts["eps"] = eps_context
            warnings.append("eps:fallback_non_sec")
            print(f"INFO: yf_fallback filled eps for {edgar.ticker} {edgar.period} = {eps} [AV]")

    needs_yf = (
        edgar.revenue is None
        or edgar.gross_margin is None
        or edgar.operating_margin is None
        or edgar.free_cash_flow is None
        or (edgar.eps is None and "eps" not in updates)
    )
    if needs_yf:
        _fill_from_yf(edgar, updates, is_annual, contexts, warnings)

    if not updates:
        return edgar
    updates["metric_contexts"] = contexts
    updates["quality_warnings"] = sorted(set(warnings))
    updates["data_quality_status"] = "CHECK_REQUIRED"
    return dataclasses.replace(edgar, **updates)


def _av_eps_with_context(
    ticker: str, filing_date: str, is_annual: bool
) -> tuple[Optional[float], dict]:
    data, _ = fetch_earnings(ticker)
    if not data:
        return None, {}
    source = data.get("annualEarnings" if is_annual else "quarterlyEarnings", [])
    try:
        anchor = datetime.strptime(filing_date, "%Y-%m-%d")
    except ValueError:
        return None, {}
    target = anchor - timedelta(days=45)
    best_eps, best_delta = None, timedelta(days=_YF_WINDOW_DAYS)
    best_entry: dict = {}
    for entry in source:
        try:
            d = datetime.strptime(entry["fiscalDateEnding"], "%Y-%m-%d")
        except (ValueError, KeyError):
            continue
        delta = abs(d - target)
        if delta < best_delta:
            try:
                best_eps = float(entry["reportedEPS"])
                best_delta = delta
                best_entry = entry
            except (ValueError, TypeError):
                pass
    context = {
        "source": "alpha_vantage",
        "selection": "fallback_non_sec",
        "fiscal_date_ending": best_entry.get("fiscalDateEnding"),
        "reported_date": best_entry.get("reportedDate"),
    } if best_entry else {}
    return best_eps, context


def _av_eps(ticker: str, filing_date: str, is_annual: bool) -> Optional[float]:
    """Backward-compatible scalar wrapper."""
    return _av_eps_with_context(ticker, filing_date, is_annual)[0]


def _get_yf_data(ticker: str, is_annual: bool) -> dict:
    key = f"{ticker}:{'annual' if is_annual else 'quarterly'}"
    if key not in _YF_CACHE:
        t = yf.Ticker(ticker)
        if is_annual:
            _YF_CACHE[key] = {"income": t.financials, "cashflow": t.cashflow}
        else:
            _YF_CACHE[key] = {"income": t.quarterly_financials, "cashflow": t.quarterly_cashflow}
    return _YF_CACHE[key]


def _fill_from_yf(
    edgar: EdgarData,
    updates: dict,
    is_annual: bool,
    contexts: Optional[dict] = None,
    warnings: Optional[list[str]] = None,
) -> None:
    import pandas as pd

    try:
        data = _get_yf_data(edgar.ticker, is_annual)
        income = data["income"]
        cashflow_df = data["cashflow"]
    except Exception:
        return

    if income is None or income.empty:
        return

    try:
        filing_dt = datetime.strptime(edgar.filing_date, "%Y-%m-%d")
    except ValueError:
        return

    col = _nearest_col(income, filing_dt)
    if col is None:
        return

    contexts = contexts if contexts is not None else {}
    warnings = warnings if warnings is not None else []

    def _get(df, column, *row_names: str) -> Optional[float]:
        for name in row_names:
            try:
                v = df.loc[name, column]
                if v is not None and not pd.isna(v):
                    return float(v)
            except (KeyError, TypeError):
                pass
        return None

    rev = edgar.revenue
    if rev is None:
        rev = _get(income, col, "Total Revenue", "TotalRevenue")
        if rev is not None:
            updates["revenue"] = rev
            contexts["revenue"] = _yf_context("income", col, "Total Revenue")
            warnings.append("revenue:fallback_non_sec")
            print(f"INFO: yf_fallback filled revenue for {edgar.ticker} {edgar.period} = {rev} [yf]")

    if edgar.gross_margin is None and rev:
        gp = _get(income, col, "Gross Profit", "GrossProfit")
        if gp is not None:
            updates["gross_margin"] = round(gp / rev, 4)
            contexts["gross_margin"] = _yf_context("income", col, "Gross Profit / Revenue")
            warnings.append("gross_margin:fallback_non_sec")
            print(f"INFO: yf_fallback filled gross_margin for {edgar.ticker} {edgar.period} [yf]")

    if edgar.operating_margin is None and rev:
        op = _get(income, col, "Operating Income", "EBIT", "OperatingIncome")
        if op is not None:
            updates["operating_margin"] = round(op / rev, 4)
            contexts["operating_margin"] = _yf_context("income", col, "Operating Income / Revenue")
            warnings.append("operating_margin:fallback_non_sec")
            print(f"INFO: yf_fallback filled operating_margin for {edgar.ticker} {edgar.period} [yf]")

    if edgar.eps is None and "eps" not in updates:
        eps = _get(income, col, "Diluted EPS", "Basic EPS", "DilutedEPS", "BasicEPS")
        if eps is not None:
            updates["eps"] = eps
            contexts["eps"] = _yf_context("income", col, "Diluted EPS")
            warnings.append("eps:fallback_non_sec")
            print(f"INFO: yf_fallback filled eps for {edgar.ticker} {edgar.period} = {eps} [yf]")

    if edgar.free_cash_flow is None:
        cf_col = _nearest_col(cashflow_df, filing_dt) if (cashflow_df is not None and not cashflow_df.empty) else None
        if cf_col is not None:
            cfo = _get(cashflow_df, cf_col, "Operating Cash Flow", "Total Cash From Operating Activities")
            capex = _get(cashflow_df, cf_col, "Capital Expenditure", "CapitalExpenditure")
            if cfo is not None and capex is not None:
                updates["free_cash_flow"] = cfo + capex  # yfinance reports capex as negative cash flow
                contexts["free_cash_flow"] = _yf_context(
                    "cashflow", cf_col, "Operating Cash Flow + Capital Expenditure"
                )
                warnings.append("free_cash_flow:fallback_non_sec")
                print(f"INFO: yf_fallback filled free_cash_flow for {edgar.ticker} {edgar.period} [yf]")


def _yf_context(statement: str, column, row_label: str) -> dict:
    try:
        period_end = column.to_pydatetime().date().isoformat()
    except AttributeError:
        period_end = str(column)
    return {
        "source": "yfinance",
        "selection": "fallback_non_sec",
        "statement": statement,
        "period_end": period_end,
        "row_label": row_label,
    }


def _nearest_col(df, filing_dt: datetime):
    """Return the df column whose date is closest to but before filing_dt, within _YF_WINDOW_DAYS."""
    best_col, best_delta = None, timedelta(days=_YF_WINDOW_DAYS)
    for col in df.columns:
        try:
            col_dt = col.to_pydatetime().replace(tzinfo=None)
        except AttributeError:
            continue
        if col_dt >= filing_dt:
            continue
        delta = filing_dt - col_dt
        if delta < best_delta:
            best_delta = delta
            best_col = col
    return best_col