"""YFinanceClient — a free drop-in for ai-hedge-fund's DataClient protocol. hedge_fund/data/protocol.py names `YFinanceClient` in its own docstring as the example of a swappable provider: no inheritance, just match the method names. This file is that example, actually written. Honest about what it cannot do: * `get_financial_metrics` has **no filing_date**. yfinance exposes statements by *report period*, never by the date the filing became public. This client therefore filters on `report_period <= end_date`, which is LOOKAHEAD: a quarter that ended 2026-06-30 looks visible on 2026-07-01 even though the 8-K did not land until 2026-07-30. The leak is measured in `compare_fd_vs_yf.py`. * `get_earnings_history` maps `Ticker.get_earnings_dates()`. The announcement date is used as `filing_date`, and `source_type` is hard-coded "8-K" because yfinance does not say which filing it came from. PEAD's 8-K-beats-10-Q dedupe therefore has nothing to dedupe. * `get_news` / `get_insider_trades` return yfinance's shapes mapped onto the FD models; several FD fields have no yfinance counterpart and stay None. Usage:: from yf_client import YFinanceClient from hedge_fund.data import CachedDataClient fd = CachedDataClient(YFinanceClient(), cache_dir="~/.hedge-fund/cache/yf") """ from __future__ import annotations import math import pandas as pd import yfinance as yf from hedge_fund.data.models import ( CompanyFacts, CompanyNews, Earnings, EarningsRecord, FinancialMetrics, InsiderTrade, Price, ) def _f(value): """None for NaN / missing, float otherwise.""" if value is None: return None try: f = float(value) except (TypeError, ValueError): return None return None if (math.isnan(f) or math.isinf(f)) else f class YFinanceClient: """Structural DataClient backed by yfinance. No API key, no quota.""" def __init__(self) -> None: self._tickers: dict[str, yf.Ticker] = {} def _t(self, ticker: str) -> yf.Ticker: if ticker not in self._tickers: self._tickers[ticker] = yf.Ticker(ticker) return self._tickers[ticker] # -- prices --------------------------------------------------------- def get_prices(self, ticker, start_date, end_date, interval="day", interval_multiplier=1, **_) -> list[Price]: if interval != "day" or interval_multiplier != 1: raise ValueError("this client only serves daily bars") # yfinance's `end` is exclusive; FD's end_date is inclusive. end_excl = (pd.Timestamp(end_date) + pd.Timedelta(days=1)).date().isoformat() frame = self._t(ticker).history( start=start_date, end=end_excl, interval="1d", auto_adjust=False, ) out: list[Price] = [] for stamp, row in frame.iterrows(): out.append(Price( open=_f(row["Open"]) or 0.0, close=_f(row["Close"]) or 0.0, high=_f(row["High"]) or 0.0, low=_f(row["Low"]) or 0.0, volume=int(_f(row["Volume"]) or 0), time=stamp.date().isoformat(), )) return out # -- earnings ------------------------------------------------------- def get_earnings_history(self, ticker, limit: int = 12) -> list[EarningsRecord]: frame = self._t(ticker).get_earnings_dates(limit=limit) if frame is None or frame.empty: return [] out: list[EarningsRecord] = [] for stamp, row in frame.iterrows(): surprise_pct = _f(row.get("Surprise(%)")) reported = _f(row.get("Reported EPS")) if surprise_pct is None or reported is None: continue # a scheduled, not-yet-reported quarter announced = stamp.date() out.append(EarningsRecord( ticker=ticker, # yfinance gives the ANNOUNCEMENT date and nothing else, so it # stands in for filing_date. report_period is reconstructed as # the quarter end before it — an approximation, not a fact. filing_date=announced.isoformat(), report_period=(pd.Timestamp(announced) - pd.offsets.QuarterEnd()) .date().isoformat(), period="quarterly", source_type="8-K", # yfinance does not say; PEAD needs a value quarterly={ "eps_actual": reported, "eps_estimate": _f(row.get("EPS Estimate")), "eps_surprise": "BEAT" if surprise_pct >= 0 else "MISS", "eps_surprise_percent": surprise_pct, }, )) return out def get_earnings(self, ticker) -> Earnings | None: history = self.get_earnings_history(ticker, limit=1) if not history: return None newest = history[0] return Earnings(ticker=ticker, report_period=newest.report_period, period="quarterly", quarterly=newest.quarterly) # -- fundamentals --------------------------------------------------- def get_financial_metrics(self, ticker, end_date, period="ttm", limit: int = 10) -> list[FinancialMetrics]: """Quarterly statement-derived metrics, newest first. NOT point-in-time. yfinance has no filing_date, so this filters on report_period — every row is visible ~3-6 weeks before it was. """ t = self._t(ticker) income, balance = t.quarterly_income_stmt, t.quarterly_balance_sheet if income is None or income.empty: return [] info = t.info or {} shares = _f(info.get("sharesOutstanding")) rows: list[FinancialMetrics] = [] for stamp in income.columns: report_period = pd.Timestamp(stamp).date().isoformat() if report_period > end_date: continue col = income[stamp] bal = balance[stamp] if (balance is not None and stamp in balance.columns) else None revenue = _f(col.get("Total Revenue")) gross = _f(col.get("Gross Profit")) operating = _f(col.get("Operating Income")) net = _f(col.get("Net Income")) equity = _f(bal.get("Stockholders Equity")) if bal is not None else None debt = _f(bal.get("Total Debt")) if bal is not None else None rows.append(FinancialMetrics( ticker=ticker, report_period=report_period, period=period, filing_date=None, # <- the whole problem, in one field market_cap=_f(info.get("marketCap")), # latest only: lookahead gross_margin=(gross / revenue) if gross and revenue else None, operating_margin=(operating / revenue) if operating and revenue else None, net_margin=(net / revenue) if net and revenue else None, return_on_equity=(net / equity) if net and equity else None, debt_to_equity=(debt / equity) if debt and equity else None, current_ratio=_f(info.get("currentRatio")), earnings_per_share=(net / shares) if net and shares else None, book_value_per_share=(equity / shares) if equity and shares else None, )) if len(rows) >= limit: break return rows def get_company_facts(self, ticker) -> CompanyFacts | None: info = self._t(ticker).info or {} if not info.get("shortName"): return None return CompanyFacts( ticker=ticker, name=info.get("shortName"), sector=info.get("sector"), industry=info.get("industry"), market_cap=_f(info.get("marketCap")), # latest only: lookahead ) def get_market_cap(self, ticker, end_date) -> float | None: facts = self.get_company_facts(ticker) return facts.market_cap if facts else None # -- news / insiders ------------------------------------------------ def get_news(self, ticker, end_date, start_date=None, limit: int = 1000) -> list[CompanyNews]: out: list[CompanyNews] = [] for item in (self._t(ticker).news or [])[:limit]: content = item.get("content", item) published = (content.get("pubDate") or "")[:10] if published and published > end_date: continue if start_date and published and published < start_date: continue out.append(CompanyNews( ticker=ticker, title=content.get("title") or "", author=(content.get("provider") or {}).get("displayName") or "", source=(content.get("provider") or {}).get("displayName") or "", date=published or end_date, url=(content.get("canonicalUrl") or {}).get("url") or "", )) return out def get_insider_trades(self, ticker, end_date, start_date=None, limit: int = 1000) -> list[InsiderTrade]: frame = self._t(ticker).insider_transactions if frame is None or frame.empty: return [] out: list[InsiderTrade] = [] for _, row in frame.iterrows(): filed = pd.Timestamp(row["Start Date"]).date().isoformat() if filed > end_date: continue if start_date and filed < start_date: continue out.append(InsiderTrade( ticker=ticker, name=str(row.get("Insider") or ""), title=str(row.get("Position") or ""), filing_date=filed, transaction_date=filed, transaction_shares=_f(row.get("Shares")), transaction_value=_f(row.get("Value")), )) if len(out) >= limit: break return out if __name__ == "__main__": # smoke check: the protocol is actually satisfied from hedge_fund.data.protocol import DataClient client = YFinanceClient() assert isinstance(client, DataClient), "YFinanceClient does not match DataClient" bars = client.get_prices("AAPL", "2026-09-14", "2026-09-18") assert bars and all(b.close > 0 for b in bars), bars events = client.get_earnings_history("AAPL", limit=8) assert events and events[0].quarterly.eps_surprise in ("BEAT", "MISS"), events print(f"ok: {len(bars)} bars, {len(events)} earnings events, " f"latest {events[0].filing_date} {events[0].quarterly.eps_surprise}")