"""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