Data

The data package provides a clean, unified interface for fetching, loading, and saving all market and historical data required by the optpricing library.

compact_snapshot

compact_snapshot(df: DataFrame) -> pd.DataFrame

Liquid-only, slimmed snapshot for compressed storage (e.g. the scheduled GitHub Action that grows the corpus): about 25 KB per ticker and day with zstd, versus ~130 KB for a full snapshot.

Keeps the rows selected by :func:liquid_option_mask and price_source, drops columns that are constant, informational or reconstructible (marketPrice is the bid-ask mid, contractSymbol the OCC code; both are restored by :func:load_market_snapshot) and stores quotes in float32 (cent precision for strikes below ~80,000) and counts in int32.

Parameters:

Name	Type	Description	Default
df	DataFrame	A full snapshot frame.	required

Returns:

Type	Description
DataFrame	The compacted frame.

Source code in src/optpricing/data/market_data_manager.py

def compact_snapshot(df: pd.DataFrame) -> pd.DataFrame:
    """
    Liquid-only, slimmed snapshot for compressed storage (e.g. the scheduled
    GitHub Action that grows the corpus): about 25 KB per ticker and day with
    zstd, versus ~130 KB for a full snapshot.

    Keeps the rows selected by :func:`liquid_option_mask` and ``price_source``,
    drops columns that are constant, informational or reconstructible
    (``marketPrice`` is the bid-ask mid, ``contractSymbol`` the OCC code; both
    are restored by :func:`load_market_snapshot`) and stores quotes in float32
    (cent precision for strikes below ~80,000) and counts in int32.

    Parameters
    ----------
    df : pd.DataFrame
        A full snapshot frame.

    Returns
    -------
    pd.DataFrame
        The compacted frame.
    """
    out = df[liquid_option_mask(df)].copy()
    out = out.drop(columns=[c for c in _COMPACT_DROP if c in out.columns])
    for col in ("bid", "ask", "impliedVolatility", "strike", "maturity"):
        if col in out.columns:
            out[col] = out[col].astype("float32")
    for col in ("volume", "openInterest"):
        if col in out.columns:
            out[col] = out[col].fillna(0).astype("int32")
    return out.reset_index(drop=True)

get_available_snapshot_dates

get_available_snapshot_dates(ticker: str, *, ascending: bool = False) -> list[str]

Lists all available snapshot dates for a given ticker.

Scans the market data directory for saved parquet files corresponding to the ticker and extracts the date from the filenames.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker to search for, e.g., 'SPY'.	required
ascending	bool	Return oldest first (what a backtest needs); by default newest first.	False

Returns:

Type	Description
list[str]	A sorted list of available dates in 'YYYY-MM-DD' format.

Source code in src/optpricing/data/market_data_manager.py

def get_available_snapshot_dates(ticker: str, *, ascending: bool = False) -> list[str]:
    """
    Lists all available snapshot dates for a given ticker.

    Scans the market data directory for saved parquet files corresponding
    to the ticker and extracts the date from the filenames.

    Parameters
    ----------
    ticker : str
        The stock ticker to search for, e.g., 'SPY'.
    ascending : bool, optional
        Return oldest first (what a backtest needs); by default newest first.

    Returns
    -------
    list[str]
        A sorted list of available dates in 'YYYY-MM-DD' format.
    """
    try:
        files = [
            f.name
            for f in MARKET_SNAPSHOT_DIR.iterdir()
            if f.name.startswith(f"{ticker}_") and f.name.endswith(".parquet")
        ]
        return sorted(
            [f.replace(f"{ticker}_", "").replace(".parquet", "") for f in files],
            reverse=not ascending,
        )

    except FileNotFoundError:
        return []

get_live_dividend_yield

get_live_dividend_yield(ticker: str) -> float

Fetches the live trailing dividend yield for a ticker using yfinance.

Handles the unit change in yfinance (fraction before 0.2.5x, percent since).

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker to search for, e.g., 'SPY'.	required

Returns:

Type	Description
float	The continuous dividend yield as a decimal, or zero.

Source code in src/optpricing/data/market_data_manager.py

def get_live_dividend_yield(ticker: str) -> float:
    """
    Fetches the live trailing dividend yield for a ticker using yfinance.

    Handles the unit change in yfinance (fraction before 0.2.5x, percent since).

    Parameters
    ----------
    ticker : str
        The stock ticker to search for, e.g., 'SPY'.

    Returns
    -------
    float
        The continuous dividend yield as a decimal, or zero.
    """
    logger.info("Fetching live dividend yield for %s...", ticker)
    try:
        info = yf.Ticker(ticker).info
        value = info.get("dividendYield")
        if value is None:
            value = info.get("trailingAnnualDividendYield")
        return _normalise_dividend_yield(value)
    except Exception as e:
        logger.error("FAILED to fetch dividend yield for %s. Error: %s", ticker, e)
        return 0.0

get_live_option_chain

get_live_option_chain(ticker: str) -> pd.DataFrame | None

Fetches a live option chain from the configured data provider.

The data provider is determined by the live_data_provider key in the config.yaml file. Supported providers are "yfinance".

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker for which to fetch the option chain, e.g., 'SPY'.	required

Returns:

Type	Description
DataFrame | None	A DataFrame containing the formatted option chain data (schema v2), or None if the fetch fails or no data is returned.

Source code in src/optpricing/data/market_data_manager.py

def get_live_option_chain(ticker: str) -> pd.DataFrame | None:
    """
    Fetches a live option chain from the configured data provider.

    The data provider is determined by the `live_data_provider` key in the
    `config.yaml` file. Supported providers are "yfinance".

    Parameters
    ----------
    ticker : str
        The stock ticker for which to fetch the option chain, e.g., 'SPY'.

    Returns
    -------
    pd.DataFrame | None
        A DataFrame containing the formatted option chain data (schema v2),
        or None if the fetch fails or no data is returned.
    """
    provider = _config.get("live_data_provider", "yfinance").lower()
    if provider != "yfinance":
        logger.warning(
            "Unknown live_data_provider '%s'. Defaulting to yfinance.", provider
        )
    return _fetch_from_yfinance(ticker)

get_live_risk_free_rate

get_live_risk_free_rate() -> float | None

Current risk-free rate from the live ^IRX quote.

Returns:

Type	Description
float | None	Continuously compounded rate, or None if the quote is unavailable.

Source code in src/optpricing/data/historical_manager.py

def get_live_risk_free_rate() -> float | None:
    """
    Current risk-free rate from the live ``^IRX`` quote.

    Returns
    -------
    float | None
        Continuously compounded rate, or None if the quote is unavailable.
    """
    try:
        info = yf.Ticker(RISK_FREE_TICKER).fast_info
        quote = info.get("lastPrice", None)
        if quote is None:
            quote = info.get("last_price", None)
        if quote is None or not np.isfinite(quote):
            return None
        return tbill_discount_to_continuous(float(quote))
    except Exception as e:
        logger.warning("Could not fetch %s: %s", RISK_FREE_TICKER, e)
        return None

historical_returns_if_available

historical_returns_if_available(ticker: str, period: str = '10y') -> pd.Series | None

Historical log returns from the local file, or None if there is no file.

Unlike :func:load_historical_returns this never downloads anything, so it is safe inside offline workflows.

Source code in src/optpricing/data/historical_manager.py

def historical_returns_if_available(
    ticker: str,
    period: str = "10y",
) -> pd.Series | None:
    """
    Historical log returns from the local file, or None if there is no file.

    Unlike :func:`load_historical_returns` this never downloads anything, so
    it is safe inside offline workflows.
    """
    filename = HISTORICAL_DIR / f"{ticker}_{period}_returns.parquet"
    if not filename.exists():
        return None
    return pd.read_parquet(filename)["log_return"]

is_v2_snapshot

is_v2_snapshot(df: DataFrame) -> bool

True if the frame carries the schema-v2 market-parameter columns.

Source code in src/optpricing/data/market_data_manager.py

def is_v2_snapshot(df: pd.DataFrame) -> bool:
    """True if the frame carries the schema-v2 market-parameter columns."""
    return all(col in df.columns for col in V2_COLUMNS)

liquid_option_mask

liquid_option_mask(df: DataFrame, *, max_relative_spread: float = 0.3, min_maturity_days: float = 3.0, max_maturity_years: float = 1.5, moneyness: tuple[float, float] = (0.7, 1.3), min_open_interest: int = 5, min_volume: int = 1) -> pd.Series

Boolean mask of options worth keeping in a compact snapshot.

A row passes when it has a positive bid, a bid-ask spread no wider than max_relative_spread of the mid, open interest of at least min_open_interest or volume of at least min_volume, a maturity inside the window and a strike within the moneyness band.

Parameters:

Name	Type	Description	Default
df	DataFrame	A snapshot frame.	required
max_relative_spread	float	Maximum (ask - bid) / mid, by default 0.3.	0.3
min_maturity_days	float	Maturity window, by default 3 days to 1.5 years.	3.0
max_maturity_years	float	Maturity window, by default 3 days to 1.5 years.	3.0
moneyness	tuple[float, float]	Allowed strike / spot range, by default (0.7, 1.3).	(0.7, 1.3)
min_open_interest	int	Activity floor (either one suffices), by default 5 contracts / 1 trade.	5
min_volume	int	Activity floor (either one suffices), by default 5 contracts / 1 trade.	5

Returns:

Type	Description
Series	Boolean mask aligned with df.

Source code in src/optpricing/data/market_data_manager.py

def liquid_option_mask(
    df: pd.DataFrame,
    *,
    max_relative_spread: float = 0.3,
    min_maturity_days: float = 3.0,
    max_maturity_years: float = 1.5,
    moneyness: tuple[float, float] = (0.7, 1.3),
    min_open_interest: int = 5,
    min_volume: int = 1,
) -> pd.Series:
    """
    Boolean mask of options worth keeping in a compact snapshot.

    A row passes when it has a positive bid, a bid-ask spread no wider than
    ``max_relative_spread`` of the mid, open interest of at least
    ``min_open_interest`` or volume of at least ``min_volume``, a maturity
    inside the window and a strike within the moneyness band.

    Parameters
    ----------
    df : pd.DataFrame
        A snapshot frame.
    max_relative_spread : float, optional
        Maximum ``(ask - bid) / mid``, by default 0.3.
    min_maturity_days, max_maturity_years : float, optional
        Maturity window, by default 3 days to 1.5 years.
    moneyness : tuple[float, float], optional
        Allowed ``strike / spot`` range, by default (0.7, 1.3).
    min_open_interest, min_volume : int, optional
        Activity floor (either one suffices), by default 5 contracts / 1 trade.

    Returns
    -------
    pd.Series
        Boolean mask aligned with ``df``.
    """
    spot = df["spot_price"]
    mid = df["marketPrice"]
    spread = (df["ask"] - df["bid"]) / mid.where(mid > 0, np.nan)
    zeros = pd.Series(0.0, index=df.index)
    open_interest = df["openInterest"] if "openInterest" in df.columns else zeros
    volume = df["volume"] if "volume" in df.columns else zeros
    active = (open_interest.fillna(0) >= min_open_interest) | (
        volume.fillna(0) >= min_volume
    )
    mask = (
        (df["bid"] > 0)
        & (spread <= max_relative_spread)
        & active
        & (df["maturity"] >= min_maturity_days / 365.25)
        & (df["maturity"] <= max_maturity_years)
        & (df["strike"] / spot >= moneyness[0])
        & (df["strike"] / spot <= moneyness[1])
    )
    return mask.fillna(False)

load_historical_prices

load_historical_prices(ticker: str, period: str = '10y') -> pd.DataFrame

Loads the historical frame for a ticker, downloading it if missing.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker, e.g., 'SPY'.	required
period	str	The history period of the file, by default "10y".	'10y'

Returns:

Type	Description
DataFrame	Frame indexed by date with log_return and, for v2 files, close and dividends.

Raises:

Type	Description
FileNotFoundError	If the data file cannot be found and also fails to be downloaded.

Source code in src/optpricing/data/historical_manager.py

def load_historical_prices(
    ticker: str,
    period: str = "10y",
) -> pd.DataFrame:
    """
    Loads the historical frame for a ticker, downloading it if missing.

    Parameters
    ----------
    ticker : str
        The stock ticker, e.g., 'SPY'.
    period : str, optional
        The history period of the file, by default "10y".

    Returns
    -------
    pd.DataFrame
        Frame indexed by date with ``log_return`` and, for v2 files, ``close``
        and ``dividends``.

    Raises
    ------
    FileNotFoundError
        If the data file cannot be found and also fails to be downloaded.
    """
    filename = HISTORICAL_DIR / f"{ticker}_{period}_returns.parquet"
    if not filename.exists():
        logger.info("No historical data for %s; fetching and saving now.", ticker)
        save_historical_returns([ticker], period)
    if not filename.exists():
        raise FileNotFoundError(f"Could not find or save historical data for {ticker}.")
    return pd.read_parquet(filename)

load_historical_returns

load_historical_returns(ticker: str, period: str = '10y') -> pd.Series

Loads historical log returns, fetching and saving them if not found.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker for which to load returns, e.g., 'SPY'.	required
period	str	The time period for which to fetch data, e.g., "10y", "5y", "1mo". Defaults to "10y".	'10y'

Returns:

Type	Description
Series	A pandas Series containing the historical log returns.

Raises:

Type	Description
FileNotFoundError	If the data file cannot be found and also fails to be downloaded.

Source code in src/optpricing/data/historical_manager.py

def load_historical_returns(
    ticker: str,
    period: str = "10y",
) -> pd.Series:
    """
    Loads historical log returns, fetching and saving them if not found.

    Parameters
    ----------
    ticker : str
        The stock ticker for which to load returns, e.g., 'SPY'.
    period : str, optional
        The time period for which to fetch data, e.g., "10y", "5y", "1mo".
        Defaults to "10y".

    Returns
    -------
    pd.Series
        A pandas Series containing the historical log returns.

    Raises
    ------
    FileNotFoundError
        If the data file cannot be found and also fails to be downloaded.
    """
    return load_historical_prices(ticker, period)["log_return"]

load_market_snapshot

load_market_snapshot(ticker: str, snapshot_date: str) -> pd.DataFrame | None

Loads a previously saved market data snapshot for a specific date.

Version-1 files are upgraded to schema v2 in memory (see :func:upgrade_snapshot); the file on disk is not modified.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker of the desired snapshot, e.g., 'SPY'.	required
snapshot_date	str	The date of the snapshot in 'YYYY-MM-DD' format.	required

Returns:

Type	Description
DataFrame | None	A DataFrame containing the snapshot data, or None if the file is not found.

Source code in src/optpricing/data/market_data_manager.py

def load_market_snapshot(ticker: str, snapshot_date: str) -> pd.DataFrame | None:
    """
    Loads a previously saved market data snapshot for a specific date.

    Version-1 files are upgraded to schema v2 in memory (see
    :func:`upgrade_snapshot`); the file on disk is not modified.

    Parameters
    ----------
    ticker : str
        The stock ticker of the desired snapshot, e.g., 'SPY'.
    snapshot_date : str
        The date of the snapshot in 'YYYY-MM-DD' format.

    Returns
    -------
    pd.DataFrame | None
        A DataFrame containing the snapshot data, or None if the file
        is not found.
    """
    filename = MARKET_SNAPSHOT_DIR / f"{ticker}_{snapshot_date}.parquet"
    if not filename.exists():
        logger.error("Snapshot file not found: %s", filename)
        return None

    logger.info("Loading data from %s...", filename)
    df = pd.read_parquet(filename)
    if "marketPrice" not in df.columns or "contractSymbol" not in df.columns:
        df = restore_compact_columns(df, ticker)
    if not is_v2_snapshot(df):
        df = upgrade_snapshot(df, ticker, snapshot_date)
    return df

load_risk_free_rate

load_risk_free_rate(as_of: date | str | Timestamp, *, max_staleness_days: int = 10) -> float | None

Risk-free rate on (or shortly before) a date from the stored history.

Parameters:

Name	Type	Description	Default
as_of	date | str | Timestamp	Valuation date.	required
max_staleness_days	int	Oldest acceptable observation, by default 10 days before as_of.	10

Returns:

Type	Description
float | None	Continuously compounded rate, or None when no history is stored or the nearest observation is too old.

Source code in src/optpricing/data/historical_manager.py

def load_risk_free_rate(
    as_of: date | str | pd.Timestamp,
    *,
    max_staleness_days: int = 10,
) -> float | None:
    """
    Risk-free rate on (or shortly before) a date from the stored history.

    Parameters
    ----------
    as_of : date | str | pd.Timestamp
        Valuation date.
    max_staleness_days : int, optional
        Oldest acceptable observation, by default 10 days before ``as_of``.

    Returns
    -------
    float | None
        Continuously compounded rate, or None when no history is stored or
        the nearest observation is too old.
    """
    filename = HISTORICAL_DIR / RATE_FILE
    if not filename.exists():
        return None
    frame = pd.read_parquet(filename)
    idx = pd.DatetimeIndex(frame.index)
    if idx.tz is not None:
        idx = idx.tz_localize(None)
    as_of_ts = pd.Timestamp(as_of).normalize()
    mask = (idx <= as_of_ts) & (idx >= as_of_ts - pd.Timedelta(days=max_staleness_days))
    if not mask.any():
        return None
    return float(frame["rate"].to_numpy()[np.flatnonzero(mask)[-1]])

option_contract_symbols

option_contract_symbols(ticker: str, n_contracts: int, *, chain: DataFrame | None = None, as_of: date | None = None) -> list[str]

Most active option contracts of a ticker, for streaming alongside the spot.

Parameters:

Name	Type	Description	Default
ticker	str	Underlying ticker.	required
n_contracts	int	How many contracts to return (ranked by open interest, then volume).	required
chain	DataFrame | None	An option chain to rank; the live chain is fetched when omitted.	None
as_of	date | None	Only contracts expiring after this date are considered (default today).	None

Returns:

Type	Description
list[str]	Contract symbols such as SPY250919C00600000.

Source code in src/optpricing/data/stream.py

def option_contract_symbols(
    ticker: str,
    n_contracts: int,
    *,
    chain: pd.DataFrame | None = None,
    as_of: date | None = None,
) -> list[str]:
    """
    Most active option contracts of a ticker, for streaming alongside the spot.

    Parameters
    ----------
    ticker : str
        Underlying ticker.
    n_contracts : int
        How many contracts to return (ranked by open interest, then volume).
    chain : pd.DataFrame | None, optional
        An option chain to rank; the live chain is fetched when omitted.
    as_of : date | None, optional
        Only contracts expiring after this date are considered (default today).

    Returns
    -------
    list[str]
        Contract symbols such as ``SPY250919C00600000``.
    """
    if n_contracts <= 0:
        return []
    if chain is None:
        from .market_data_manager import get_live_option_chain

        chain = get_live_option_chain(ticker)
    if chain is None or chain.empty or "contractSymbol" not in chain.columns:
        return []
    as_of = as_of or date.today()
    live = chain[pd.to_datetime(chain["expiry"]).dt.date > as_of]
    ranked = live.sort_values(
        ["openInterest", "volume"], ascending=False, na_position="last"
    )
    return ranked["contractSymbol"].head(n_contracts).astype(str).tolist()

record_live_quotes

record_live_quotes(symbols: list[str], duration: float, output: Path | str | None = None, *, websocket_factory: Callable[[], Any] | None = None, heartbeat: float = 15.0, flush_every: int = 1000) -> tuple[Path | None, int]

Streams quotes for symbols for duration seconds and saves them.

Parameters:

Name	Type	Description	Default
symbols	list[str]	Yahoo symbols (tickers, indices such as ^SPX or option contract symbols).	required
duration	float	Recording length in seconds (Ctrl-C stops early and still saves).	required
output	Path | str | None	Parquet destination; defaults to :func:default_recording_path.	None
websocket_factory	Callable[[], Any] | None	Returns an object with subscribe, listen(handler) and close (defaults to yfinance.WebSocket(verbose=False)); injectable for tests.	None
heartbeat	float	Seconds between re-subscriptions keeping the stream alive, by default 15.	15.0
flush_every	int	Messages between intermediate writes, by default 1000.	1000

Returns:

Type	Description
tuple[Path | None, int]	The written file (None when no message arrived) and the message count.

Source code in src/optpricing/data/stream.py

def record_live_quotes(
    symbols: list[str],
    duration: float,
    output: Path | str | None = None,
    *,
    websocket_factory: Callable[[], Any] | None = None,
    heartbeat: float = 15.0,
    flush_every: int = 1000,
) -> tuple[Path | None, int]:
    """
    Streams quotes for ``symbols`` for ``duration`` seconds and saves them.

    Parameters
    ----------
    symbols : list[str]
        Yahoo symbols (tickers, indices such as ``^SPX`` or option contract
        symbols).
    duration : float
        Recording length in seconds (Ctrl-C stops early and still saves).
    output : Path | str | None, optional
        Parquet destination; defaults to :func:`default_recording_path`.
    websocket_factory : Callable[[], Any] | None, optional
        Returns an object with ``subscribe``, ``listen(handler)`` and ``close``
        (defaults to ``yfinance.WebSocket(verbose=False)``); injectable for tests.
    heartbeat : float, optional
        Seconds between re-subscriptions keeping the stream alive, by default 15.
    flush_every : int, optional
        Messages between intermediate writes, by default 1000.

    Returns
    -------
    tuple[Path | None, int]
        The written file (None when no message arrived) and the message count.
    """
    if not symbols:
        raise ValueError("At least one symbol is required.")
    symbols = [s.upper() for s in symbols]
    path = Path(output) if output is not None else default_recording_path(symbols)
    recorder = QuoteRecorder(path, flush_every=flush_every)

    if websocket_factory is None:
        import warnings

        import yfinance as yf

        # yfinance 1.x still uses the non-context-manager `websockets` API.
        warnings.filterwarnings(
            "ignore", category=DeprecationWarning, module=r"yfinance\.live"
        )

        def websocket_factory():
            return yf.WebSocket(verbose=False)

    ws = websocket_factory()
    ws.subscribe(symbols)
    listener = threading.Thread(target=ws.listen, args=(recorder,), daemon=True)
    listener.start()
    logger.info("Streaming %s for %.0f s -> %s", symbols, duration, path)

    deadline = time.monotonic() + float(duration)
    next_heartbeat = time.monotonic() + heartbeat
    try:
        while time.monotonic() < deadline and listener.is_alive():
            time.sleep(min(0.25, max(deadline - time.monotonic(), 0.0)))
            if heartbeat > 0 and time.monotonic() >= next_heartbeat:
                try:
                    ws.subscribe(symbols)
                except Exception as e:  # pragma: no cover - network specific
                    logger.warning("Heartbeat re-subscribe failed: %s", e)
                next_heartbeat = time.monotonic() + heartbeat
    except KeyboardInterrupt:
        logger.info("Interrupted; saving what was received.")
    finally:
        # Closing the socket makes the blocked `recv` in the listener thread
        # raise ConnectionClosedOK, which yfinance logs with a traceback; that
        # is the expected shutdown path, so silence its logger meanwhile.
        yf_logger = logging.getLogger("yfinance")
        previous_level = yf_logger.level
        yf_logger.setLevel(logging.CRITICAL)
        try:
            ws.close()
        except Exception:  # pragma: no cover
            pass
        listener.join(timeout=5.0)
        yf_logger.setLevel(previous_level)

    written = recorder.flush()
    n = recorder.n_messages
    logger.info("Recorded %d messages%s", n, f" to {written}" if written else "")
    return written, n

save_historical_returns

save_historical_returns(tickers: list[str], period: str = '10y')

Fetches and saves historical prices, dividends and log returns for tickers.

For each ticker the adjusted close history is downloaded from yfinance, daily log returns are computed and the frame (close, dividends, log_return) is written to the historical data directory.

Parameters:

Name	Type	Description	Default
tickers	list[str]	A list of stock tickers to process, e.g., ['SPY', 'AAPL'].	required
period	str	The time period for which to fetch data, e.g., "10y", "5y", "1mo". Defaults to "10y".	'10y'

Source code in src/optpricing/data/historical_manager.py

def save_historical_returns(
    tickers: list[str],
    period: str = "10y",
):
    """
    Fetches and saves historical prices, dividends and log returns for tickers.

    For each ticker the adjusted close history is downloaded from yfinance,
    daily log returns are computed and the frame (``close``, ``dividends``,
    ``log_return``) is written to the historical data directory.

    Parameters
    ----------
    tickers : list[str]
        A list of stock tickers to process, e.g., ['SPY', 'AAPL'].
    period : str, optional
        The time period for which to fetch data, e.g., "10y", "5y", "1mo".
        Defaults to "10y".
    """
    logger.info("Saving %s historical returns for %s", period, tickers)
    for ticker in tickers:
        try:
            data = yf.Ticker(ticker).history(period=period, auto_adjust=True)
            if data.empty:
                logger.warning("No historical data found for %s. Skipping.", ticker)
                continue

            frame = pd.DataFrame(index=data.index)
            frame["close"] = data["Close"].astype(float)
            dividends = data["Dividends"] if "Dividends" in data else 0.0
            frame["dividends"] = pd.Series(dividends, index=data.index).fillna(0.0)
            frame["log_return"] = np.log(frame["close"] / frame["close"].shift(1))
            frame = frame.dropna(subset=["log_return"])

            filename = HISTORICAL_DIR / f"{ticker}_{period}_returns.parquet"
            filename.parent.mkdir(parents=True, exist_ok=True)
            frame.to_parquet(filename)
            logger.info("Saved %d rows to %s", len(frame), filename)
        except Exception as e:
            logger.error("FAILED to save data for %s: %s", ticker, e)

save_market_snapshot

save_market_snapshot(tickers: list[str], *, compact: bool = False, only_fresh: bool = False) -> list[Path]

Saves a snapshot of the current market option chain for given tickers.

For each ticker, it fetches the live option chain using get_live_option_chain and saves it to a parquet file named with the ticker and the New York session date (see :func:session_date).

Parameters:

Name	Type	Description	Default
tickers	list[str]	A list of stock tickers to process, e.g., ['SPY', 'AAPL'].	required
compact	bool	Store only liquid options with zstd compression (see :func:compact_snapshot), by default False.	False
only_fresh	bool	Skip tickers whose chain shows no trade today (holidays, stale feeds), by default False.	False

Returns:

Type	Description
list[Path]	Paths of the files written.

Source code in src/optpricing/data/market_data_manager.py

def save_market_snapshot(
    tickers: list[str],
    *,
    compact: bool = False,
    only_fresh: bool = False,
) -> list[Path]:
    """
    Saves a snapshot of the current market option chain for given tickers.

    For each ticker, it fetches the live option chain using
    `get_live_option_chain` and saves it to a parquet file named with the
    ticker and the New York session date (see :func:`session_date`).

    Parameters
    ----------
    tickers : list[str]
        A list of stock tickers to process, e.g., ['SPY', 'AAPL'].
    compact : bool, optional
        Store only liquid options with zstd compression (see
        :func:`compact_snapshot`), by default False.
    only_fresh : bool, optional
        Skip tickers whose chain shows no trade today (holidays, stale feeds),
        by default False.

    Returns
    -------
    list[Path]
        Paths of the files written.
    """
    session = session_date()
    session_str = session.strftime("%Y-%m-%d")
    logger.info("Saving market data snapshot for %s", session_str)
    written: list[Path] = []
    for ticker in tickers:
        chain_df = get_live_option_chain(ticker)

        if chain_df is None or chain_df.empty:
            logger.warning("No valid option data found for %s. Skipping.", ticker)
            continue
        if only_fresh and not is_fresh_snapshot(chain_df, session):
            logger.warning(
                "Chain for %s is not live (no trades today or quotes missing). "
                "Skipping.",
                ticker,
            )
            continue
        if compact:
            chain_df = compact_snapshot(chain_df)
            if chain_df.empty:
                logger.warning("No liquid options for %s. Skipping.", ticker)
                continue

        filename = MARKET_SNAPSHOT_DIR / f"{ticker}_{session_str}.parquet"
        filename.parent.mkdir(parents=True, exist_ok=True)
        if compact:
            chain_df.to_parquet(filename, compression="zstd", index=False)
        else:
            chain_df.to_parquet(filename)
        logger.info("Saved %d options to %s", len(chain_df), filename)
        written.append(filename)
    return written

save_risk_free_history

save_risk_free_history(period: str = '10y') -> None

Downloads the ^IRX (13-week T-bill) history and stores it as a continuously compounded rate series in the historical data directory.

Parameters:

Name	Type	Description	Default
period	str	History period, by default "10y".	'10y'

Source code in src/optpricing/data/historical_manager.py

def save_risk_free_history(period: str = "10y") -> None:
    """
    Downloads the ``^IRX`` (13-week T-bill) history and stores it as a
    continuously compounded rate series in the historical data directory.

    Parameters
    ----------
    period : str, optional
        History period, by default "10y".
    """
    data = yf.Ticker(RISK_FREE_TICKER).history(period=period)
    if data.empty:
        logger.warning("No history returned for %s.", RISK_FREE_TICKER)
        return
    rates = data["Close"].dropna().map(tbill_discount_to_continuous)
    frame = rates.to_frame(name="rate")
    frame.index = pd.DatetimeIndex(frame.index).tz_localize(None).normalize()
    filename = HISTORICAL_DIR / RATE_FILE
    filename.parent.mkdir(parents=True, exist_ok=True)
    frame.to_parquet(filename)
    logger.info("Saved %d risk-free observations to %s", len(frame), filename)

snapshot_market_params

snapshot_market_params(df: DataFrame) -> tuple[Stock, Rate]

Builds the Stock and Rate atoms described by a snapshot.

Uses the stored v2 columns; for v1 frames (or frames without them, e.g. synthetic test data) the rate defaults to DEFAULT_RISK_FREE_RATE and the dividend yield is implied from the chain, all offline.

Parameters:

Name	Type	Description	Default
df	DataFrame	A market snapshot with at least spot_price, strike, maturity, marketPrice and optionType.	required

Returns:

Type	Description
tuple[Stock, Rate]	The underlying (spot and continuous dividend yield) and the flat rate.

Source code in src/optpricing/data/market_data_manager.py

def snapshot_market_params(df: pd.DataFrame) -> tuple[Stock, Rate]:
    """
    Builds the ``Stock`` and ``Rate`` atoms described by a snapshot.

    Uses the stored v2 columns; for v1 frames (or frames without them, e.g.
    synthetic test data) the rate defaults to ``DEFAULT_RISK_FREE_RATE`` and the
    dividend yield is implied from the chain, all offline.

    Parameters
    ----------
    df : pd.DataFrame
        A market snapshot with at least ``spot_price``, ``strike``, ``maturity``,
        ``marketPrice`` and ``optionType``.

    Returns
    -------
    tuple[Stock, Rate]
        The underlying (spot and continuous dividend yield) and the flat rate.
    """
    spot = float(df["spot_price"].iloc[0])
    if is_v2_snapshot(df):
        return (
            Stock(spot=spot, dividend=float(df["dividend_yield"].iloc[0])),
            Rate(rate=float(df["risk_free_rate"].iloc[0])),
        )
    from optpricing.calibration.fit_market_params import implied_dividend_yield

    rate = DEFAULT_RISK_FREE_RATE
    calls = df[df["optionType"] == "call"]
    puts = df[df["optionType"] == "put"]
    q = (
        implied_dividend_yield(calls, puts, spot, rate)
        if len(calls) and len(puts)
        else 0
    )
    return Stock(spot=spot, dividend=float(q)), Rate(rate=rate)

trailing_dividend_yield

trailing_dividend_yield(ticker: str, as_of: date | str | Timestamp, *, period: str = '10y', spot: float | None = None) -> float | None

Trailing twelve-month cash dividend yield from the historical file.

Parameters:

Name	Type	Description	Default
ticker	str	The stock ticker.	required
as_of	date | str | Timestamp	Valuation date; dividends with ex-dates in the preceding 365 days count.	required
period	str	History period of the file, by default "10y".	'10y'
spot	float | None	Price to divide by; defaults to the last close on or before as_of.	None

Returns:

Type	Description
float | None	Continuously compounded yield log(1 + D/S), or None if the file is a v1 file without dividends or has no data before as_of.

Source code in src/optpricing/data/historical_manager.py

def trailing_dividend_yield(
    ticker: str,
    as_of: date | str | pd.Timestamp,
    *,
    period: str = "10y",
    spot: float | None = None,
) -> float | None:
    """
    Trailing twelve-month cash dividend yield from the historical file.

    Parameters
    ----------
    ticker : str
        The stock ticker.
    as_of : date | str | pd.Timestamp
        Valuation date; dividends with ex-dates in the preceding 365 days count.
    period : str, optional
        History period of the file, by default "10y".
    spot : float | None, optional
        Price to divide by; defaults to the last close on or before ``as_of``.

    Returns
    -------
    float | None
        Continuously compounded yield ``log(1 + D/S)``, or None if the file is
        a v1 file without dividends or has no data before ``as_of``.
    """
    filename = HISTORICAL_DIR / f"{ticker}_{period}_returns.parquet"
    if not filename.exists():
        return None
    frame = pd.read_parquet(filename)
    if "dividends" not in frame.columns:
        return None
    idx = pd.DatetimeIndex(frame.index)
    as_of_ts = pd.Timestamp(as_of)
    if idx.tz is not None:
        as_of_ts = as_of_ts.tz_localize(idx.tz) if as_of_ts.tzinfo is None else as_of_ts
    window = frame[(idx > as_of_ts - pd.Timedelta(days=365)) & (idx <= as_of_ts)]
    if window.empty:
        return None
    price = float(window["close"].iloc[-1]) if spot is None else float(spot)
    if price <= 0:
        return None
    return float(math.log1p(float(window["dividends"].sum()) / price))

upgrade_snapshot

upgrade_snapshot(df: DataFrame, ticker: str, snapshot_date: str, *, rate: float | None = None) -> pd.DataFrame

Upgrades a v1 snapshot to schema v2 without network access.

The risk-free rate is taken from the stored ^IRX history for the snapshot date (rate overrides it); when neither is available a flat DEFAULT_RISK_FREE_RATE is used. The dividend yield is implied from put-call parity given that rate.

Parameters:

Name	Type	Description	Default
df	DataFrame	The v1 snapshot frame.	required
ticker	str	Ticker symbol of the snapshot.	required
snapshot_date	str	YYYY-MM-DD date of the snapshot.	required
rate	float | None	Risk-free rate to use instead of the stored history.	None

Returns:

Type	Description
DataFrame	The frame with the v2 columns added (already-v2 frames are returned unchanged).

Source code in src/optpricing/data/market_data_manager.py

def upgrade_snapshot(
    df: pd.DataFrame,
    ticker: str,
    snapshot_date: str,
    *,
    rate: float | None = None,
) -> pd.DataFrame:
    """
    Upgrades a v1 snapshot to schema v2 without network access.

    The risk-free rate is taken from the stored ``^IRX`` history for the
    snapshot date (``rate`` overrides it); when neither is available a flat
    ``DEFAULT_RISK_FREE_RATE`` is used. The dividend yield is implied from
    put-call parity given that rate.

    Parameters
    ----------
    df : pd.DataFrame
        The v1 snapshot frame.
    ticker : str
        Ticker symbol of the snapshot.
    snapshot_date : str
        ``YYYY-MM-DD`` date of the snapshot.
    rate : float | None, optional
        Risk-free rate to use instead of the stored history.

    Returns
    -------
    pd.DataFrame
        The frame with the v2 columns added (already-v2 frames are returned
        unchanged).
    """
    if is_v2_snapshot(df):
        return df
    if rate is None:
        rate = load_risk_free_rate(snapshot_date)
    if rate is None:
        logger.warning(
            "No risk-free history for %s; using %.2f. Run `optpricing data "
            "download --rates` to store the ^IRX history.",
            snapshot_date,
            DEFAULT_RISK_FREE_RATE,
        )
        rate = DEFAULT_RISK_FREE_RATE
    yf_yield = trailing_dividend_yield(ticker, snapshot_date)
    return _attach_market_params(
        df,
        ticker=ticker,
        snapshot_date=snapshot_date,
        rate=rate,
        yf_dividend_yield=np.nan if yf_yield is None else yf_yield,
    )