diff --git a/README.md b/README.md index 7353003..301b9c2 100644 --- a/README.md +++ b/README.md @@ -334,3 +334,27 @@ well-communicated process. Further reading: [`docs/methodology.md`](docs/methodology.md), [`docs/analytics_explanation.md`](docs/analytics_explanation.md), [`docs/portfolio_relevance.md`](docs/portfolio_relevance.md). + + +--- + +## Options Lab v1 + +NeuroQuant now includes an offline-first options research layer in +[`src/neuroquant/options_lab/`](src/neuroquant/options_lab/) for realistic +defined-risk option experiments. + +The first production-style slice supports observed historical option-chain +snapshots, canonical contract models, Black-Scholes-Merton sanity checks, +Greeks and IV inversion, IV-versus-realized-volatility features, deterministic +NBBO-aware execution, liquidity filters, bull-call-spread selection/backtests, +and a shadow paper ledger that compares broker paper fills with realistic and +conservative fills. + +The options engine **does not fabricate historical contracts or missing +quotes**. Historical research requires an observed chain dataset. Midpoint +fills are treated as an optimistic benchmark rather than ground truth. + +See [`docs/options_lab.md`](docs/options_lab.md) for the data contract, +execution assumptions, example usage, research protocol, and current +limitations. diff --git a/docs/options_lab.md b/docs/options_lab.md new file mode 100644 index 0000000..a812450 --- /dev/null +++ b/docs/options_lab.md @@ -0,0 +1,127 @@ +# NeuroQuant Options Lab + +NeuroQuant Options Lab extends the existing equity/ETF research engine with an +offline-first framework for **observed option-chain research** and **paper-trade +reconciliation**. + +The design goal is not to make simulated options look perfect. It is to make +the assumptions visible, conservative, reproducible, and easy to stress. + +## What v1 supports + +- Canonical calls/puts and bull-call spreads. +- Black-Scholes-Merton pricing, Greeks, and implied-volatility inversion for + sanity checks and scenario work. +- Historical chain CSV normalization using actual contract identity: + timestamp, expiration, strike, type, underlying price, bid and ask. +- Optional bid/ask sizes, volume, open interest, IV and Greeks. +- NBBO-aware fill modes: + - midpoint: optimistic benchmark only; + - realistic: configurable adverse movement from midpoint; + - conservative: buy at ask / sell at bid. +- Per-contract commissions. +- Liquidity gates for bid/ask width, zero bids, volume and open interest. +- Realized-volatility features and IV/RV filters. +- Delta/DTE/width-based bull-call-spread selection. +- Event-driven historical bull-call-spread backtests. +- Profit target, stop loss, exit-DTE and max-hold exits. +- Shadow paper ledger for comparing broker paper fills with NeuroQuant's + realistic and conservative fills. + +## Data contract + +Historical option backtests require observed chain snapshots. NeuroQuant does +not fabricate missing contracts or historical quotes. + +Minimum CSV columns: + +```text +timestamp +symbol +expiration +strike +option_type +underlying_price +bid +ask +``` + +Optional fields: + +```text +bid_size ask_size volume open_interest +iv delta gamma theta vega iv_rank +``` + +Example: + +```python +from neuroquant.options_lab import ( + load_option_chain_csv, + VerticalBacktestConfig, + backtest_bull_call_entries, +) + +chain = load_option_chain_csv("data/options/tsla_chain_history.csv") + +trades = backtest_bull_call_entries( + chain, + entry_timestamps=[ + "2026-01-15 20:00:00+00:00", + "2026-03-02 20:00:00+00:00", + ], + symbol="TSLA", + config=VerticalBacktestConfig(), +) +``` + +## Why the execution model is intentionally conservative + +Paper brokers can fill at the midpoint even when a real order might not have +received that price. The lab therefore separates three values: + +1. broker paper fill; +2. deterministic realistic fill; +3. conservative touch fill. + +A strategy is not considered robust because it works on midpoint fills. It +should be re-run under wider spreads, worse slippage, larger commissions and +stricter liquidity filters. + +## Long-volatility research hypothesis + +One initial study supported by v1 is: + +> When implied volatility is inexpensive relative to recent realized +> volatility, does a defined-risk bull call spread have positive out-of-sample +> expectancy after realistic transaction costs? + +A rigorous experiment should predefine: + +- signal used to choose entry dates; +- IV/RV threshold; +- target long and short deltas; +- DTE range; +- minimum liquidity; +- spread-width range; +- exits; +- fill model; +- commissions. + +Then evaluate chronologically with train/test or walk-forward splits. Do not +optimize and score on the same period. + +## Important v1 limitations + +- No broker connection and no live order routing. +- No live data API in the repository. +- No American early-exercise/assignment engine yet. +- No dividend-event assignment model yet. +- No portfolio margin model. +- Backtests need historical snapshots dense enough to observe both legs at + entries and exits; missing quotes produce no fabricated fill. +- BSM is a scenario/sanity-check tool, not a replacement for observed NBBO. + +These limitations are deliberate. The next production-style milestones are a +provider adapter, early-assignment model, portfolio risk aggregation, and a +paper-broker adapter behind an interface that keeps research code broker-neutral. diff --git a/src/neuroquant/options_lab/__init__.py b/src/neuroquant/options_lab/__init__.py new file mode 100644 index 0000000..3a0e013 --- /dev/null +++ b/src/neuroquant/options_lab/__init__.py @@ -0,0 +1,74 @@ +"""NeuroQuant Options Lab. + +Offline-first option research primitives for observed-chain backtests, realistic +execution modelling, volatility research, and paper-trade reconciliation. +""" + +from .backtest import ( + VerticalBacktestConfig, + backtest_bull_call_entries, + summarize_vertical_backtest, +) +from .data import chain_at_or_before, load_option_chain_csv, normalize_option_chain +from .execution import ( + ExecutionConfig, + combo_cash_flow, + execution_price, + is_liquid, + leg_cash_flow, +) +from .models import BullCallSpread, OptionContract, OptionLeg, OptionQuote, OptionType, Side +from .paper import ShadowTrade, append_shadow_trade, reconciliation_report +from .pricing import Greeks, black_scholes_greeks, black_scholes_price, implied_volatility +from .providers import OptionDataProvider, PaperBroker, ProviderCapabilities, RECOMMENDED_CAPABILITY_PROFILES +from .research import bootstrap_trade_returns, chronological_trade_split, execution_stress_summary +from .strategies import CheapVolBullishConfig, build_cheap_vol_bullish_entries +from .selector import BullCallSelectionConfig, select_bull_call_spread +from .volatility import ( + attach_realized_volatility, + iv_to_rv_ratio, + long_volatility_candidate_mask, + realized_volatility, +) + +__all__ = [ + "OptionType", + "Side", + "OptionContract", + "OptionQuote", + "OptionLeg", + "BullCallSpread", + "Greeks", + "black_scholes_price", + "black_scholes_greeks", + "implied_volatility", + "ExecutionConfig", + "execution_price", + "leg_cash_flow", + "combo_cash_flow", + "is_liquid", + "normalize_option_chain", + "load_option_chain_csv", + "chain_at_or_before", + "realized_volatility", + "attach_realized_volatility", + "iv_to_rv_ratio", + "long_volatility_candidate_mask", + "BullCallSelectionConfig", + "select_bull_call_spread", + "VerticalBacktestConfig", + "backtest_bull_call_entries", + "summarize_vertical_backtest", + "ShadowTrade", + "append_shadow_trade", + "reconciliation_report", + "OptionDataProvider", + "PaperBroker", + "ProviderCapabilities", + "RECOMMENDED_CAPABILITY_PROFILES", + "CheapVolBullishConfig", + "build_cheap_vol_bullish_entries", + "chronological_trade_split", + "bootstrap_trade_returns", + "execution_stress_summary", +] diff --git a/src/neuroquant/options_lab/backtest.py b/src/neuroquant/options_lab/backtest.py new file mode 100644 index 0000000..baa50e5 --- /dev/null +++ b/src/neuroquant/options_lab/backtest.py @@ -0,0 +1,260 @@ +"""Event-driven bull-call-spread backtester using observed option quotes.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from .data import normalize_option_chain +from .execution import ExecutionConfig, execution_price +from .models import Side +from .selector import BullCallSelectionConfig, select_bull_call_spread + + +@dataclass(frozen=True) +class VerticalBacktestConfig: + """Rules for a defined-risk long vertical backtest.""" + + selection: BullCallSelectionConfig = BullCallSelectionConfig() + execution: ExecutionConfig = ExecutionConfig() + quantity: int = 1 + max_hold_days: int = 60 + exit_dte: int = 21 + profit_target_pct_max: float | None = 0.50 + stop_loss_pct_debit: float | None = 0.60 + fill_mode: str = "realistic" + + def __post_init__(self) -> None: + if self.quantity <= 0: + raise ValueError("quantity must be positive") + if self.max_hold_days <= 0: + raise ValueError("max_hold_days must be positive") + if self.exit_dte < 0: + raise ValueError("exit_dte cannot be negative") + + +def _same_contract_rows( + chain: pd.DataFrame, + symbol: str, + expiration, + strike: float, + timestamp, +) -> pd.DataFrame: + rows = chain[ + (chain["symbol"] == symbol) + & (chain["expiration"] == expiration) + & (chain["strike"] == strike) + & (chain["option_type"] == "call") + & (chain["timestamp"] <= timestamp) + ] + if rows.empty: + return rows + return rows.sort_values("timestamp").tail(1) + + +def _spread_close_value( + chain: pd.DataFrame, + symbol: str, + expiration, + long_strike: float, + short_strike: float, + timestamp, + config: VerticalBacktestConfig, +) -> tuple[float, dict] | None: + long_rows = _same_contract_rows( + chain, symbol, expiration, long_strike, timestamp + ) + short_rows = _same_contract_rows( + chain, symbol, expiration, short_strike, timestamp + ) + if long_rows.empty or short_rows.empty: + return None + long = long_rows.iloc[0].to_dict() + short = short_rows.iloc[0].to_dict() + + # To close: sell the long and buy back the short. + long_px = execution_price(long, Side.SELL, config.execution, config.fill_mode) + short_px = execution_price(short, Side.BUY, config.execution, config.fill_mode) + credit = long_px - short_px + commission = 2 * config.execution.commission_per_contract + dollars = ( + credit * 100.0 - commission + ) * config.quantity + return float(dollars), {"long": long, "short": short} + + +def backtest_bull_call_entries( + chain: pd.DataFrame, + entry_timestamps, + symbol: str, + config: VerticalBacktestConfig | None = None, +) -> pd.DataFrame: + """Backtest independent bull-call entries on specified timestamps. + + The function uses only chain observations available at each entry and each + later exit evaluation. It makes no synthetic fill at a missing quote. + """ + + config = config or VerticalBacktestConfig() + data = normalize_option_chain(chain) + data = data[data["symbol"] == symbol.upper()].copy() + if data.empty: + raise ValueError(f"no chain data for {symbol}") + + results = [] + for raw_entry in entry_timestamps: + entry_ts = pd.Timestamp(raw_entry) + entry_ts = ( + entry_ts.tz_localize("UTC") + if entry_ts.tzinfo is None + else entry_ts.tz_convert("UTC") + ) + available = data[data["timestamp"] <= entry_ts].copy() + if available.empty: + continue + key = ["expiration", "strike", "option_type"] + available = available.loc[ + available.groupby(key)["timestamp"].idxmax() + ].reset_index(drop=True) + + try: + pick = select_bull_call_spread( + available, config.selection, config.execution + ) + except ValueError: + continue + + long = pick["long"] + short = pick["short"] + long_px = execution_price( + long, Side.BUY, config.execution, config.fill_mode + ) + short_px = execution_price( + short, Side.SELL, config.execution, config.fill_mode + ) + debit_per_share = long_px - short_px + if debit_per_share <= 0: + continue + + commissions_open = 2 * config.execution.commission_per_contract + initial_debit = ( + debit_per_share * 100.0 + commissions_open + ) * config.quantity + width = float(pick["width"]) + max_value = width * 100.0 * config.quantity + max_profit = max_value - initial_debit + breakeven = float(long["strike"]) + initial_debit / (100.0 * config.quantity) + + expiry = pd.Timestamp(pick["expiration"], tz="UTC") + last_allowed = min( + entry_ts + pd.Timedelta(days=config.max_hold_days), + expiry, + ) + checkpoints = sorted( + data[ + (data["timestamp"] > entry_ts) + & (data["timestamp"] <= last_allowed) + ]["timestamp"].unique() + ) + + exit_cash = None + exit_ts = None + exit_reason = None + for ts in checkpoints: + close = _spread_close_value( + data, + symbol.upper(), + pick["expiration"], + float(long["strike"]), + float(short["strike"]), + pd.Timestamp(ts), + config, + ) + if close is None: + continue + value, quotes = close + pnl = value - initial_debit + current_dte = int(quotes["long"]["dte"]) + + if ( + config.profit_target_pct_max is not None + and max_profit > 0 + and pnl >= max_profit * config.profit_target_pct_max + ): + exit_cash, exit_ts, exit_reason = value, pd.Timestamp(ts), "profit_target" + break + if ( + config.stop_loss_pct_debit is not None + and pnl <= -initial_debit * config.stop_loss_pct_debit + ): + exit_cash, exit_ts, exit_reason = value, pd.Timestamp(ts), "stop_loss" + break + if current_dte <= config.exit_dte: + exit_cash, exit_ts, exit_reason = value, pd.Timestamp(ts), "exit_dte" + break + + if exit_cash is None: + # Use the latest observable quote before the configured horizon. + close = _spread_close_value( + data, + symbol.upper(), + pick["expiration"], + float(long["strike"]), + float(short["strike"]), + last_allowed, + config, + ) + if close is None: + continue + exit_cash, quotes = close + exit_ts = pd.Timestamp(quotes["long"]["timestamp"]) + exit_reason = "max_hold" + + pnl = float(exit_cash - initial_debit) + results.append( + { + "symbol": symbol.upper(), + "entry_timestamp": entry_ts, + "exit_timestamp": exit_ts, + "expiration": pick["expiration"], + "long_strike": float(long["strike"]), + "short_strike": float(short["strike"]), + "quantity": config.quantity, + "entry_debit": float(initial_debit), + "breakeven": breakeven, + "max_profit": float(max_profit), + "exit_value": float(exit_cash), + "pnl": pnl, + "return_on_debit": pnl / initial_debit, + "exit_reason": exit_reason, + } + ) + + return pd.DataFrame(results) + + +def summarize_vertical_backtest(trades: pd.DataFrame) -> dict: + """Compact strategy-level diagnostics for option-trade outcomes.""" + + if trades.empty: + return { + "trade_count": 0, + "win_rate": float("nan"), + "mean_return_on_debit": float("nan"), + "median_return_on_debit": float("nan"), + "total_pnl": 0.0, + "profit_factor": float("nan"), + } + pnl = trades["pnl"].astype(float) + wins = pnl[pnl > 0].sum() + losses = -pnl[pnl < 0].sum() + return { + "trade_count": int(len(trades)), + "win_rate": float((pnl > 0).mean()), + "mean_return_on_debit": float(trades["return_on_debit"].mean()), + "median_return_on_debit": float(trades["return_on_debit"].median()), + "total_pnl": float(pnl.sum()), + "profit_factor": float(wins / losses) if losses > 0 else float("inf"), + } diff --git a/src/neuroquant/options_lab/data.py b/src/neuroquant/options_lab/data.py new file mode 100644 index 0000000..cbb5795 --- /dev/null +++ b/src/neuroquant/options_lab/data.py @@ -0,0 +1,123 @@ +"""Offline option-chain ingestion and normalization. + +The lab never invents historical contracts. Research data must come from an +observed chain snapshot (CSV/export/API materialized by the user/provider). +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd + + +REQUIRED_COLUMNS = { + "timestamp", + "symbol", + "expiration", + "strike", + "option_type", + "underlying_price", + "bid", + "ask", +} + +OPTIONAL_COLUMNS = ( + "bid_size", + "ask_size", + "volume", + "open_interest", + "iv", + "delta", + "gamma", + "theta", + "vega", + "iv_rank", +) + + +def normalize_option_chain(frame: pd.DataFrame) -> pd.DataFrame: + """Return a validated, canonical option-chain dataframe.""" + + if not isinstance(frame, pd.DataFrame) or frame.empty: + raise ValueError("option chain must be a non-empty DataFrame") + + lowered = {str(c).lower().strip(): c for c in frame.columns} + missing = [c for c in REQUIRED_COLUMNS if c not in lowered] + if missing: + raise ValueError(f"option chain missing required columns: {sorted(missing)}") + + keep = list(REQUIRED_COLUMNS) + [c for c in OPTIONAL_COLUMNS if c in lowered] + out = frame[[lowered[c] for c in keep]].copy() + out.columns = keep + + out["timestamp"] = pd.to_datetime(out["timestamp"], utc=True, errors="coerce") + out["expiration"] = pd.to_datetime(out["expiration"], errors="coerce").dt.date + out["symbol"] = out["symbol"].astype(str).str.upper().str.strip() + out["option_type"] = out["option_type"].astype(str).str.lower().str.strip() + out["option_type"] = out["option_type"].replace({"c": "call", "p": "put"}) + + numeric = [ + "strike", + "underlying_price", + "bid", + "ask", + *[c for c in OPTIONAL_COLUMNS if c in out.columns], + ] + for col in numeric: + out[col] = pd.to_numeric(out[col], errors="coerce") + + if out["timestamp"].isna().any() or out["expiration"].isna().any(): + raise ValueError("timestamp/expiration contain invalid values") + if (~out["option_type"].isin(["call", "put"])).any(): + raise ValueError("option_type must be call/put (or C/P)") + if (out["strike"] <= 0).any() or (out["underlying_price"] <= 0).any(): + raise ValueError("strike and underlying_price must be positive") + if (out["bid"] < 0).any() or (out["ask"] < 0).any(): + raise ValueError("bid/ask cannot be negative") + if (out["ask"] < out["bid"]).any(): + raise ValueError("crossed quotes found: ask below bid") + + out["dte"] = ( + pd.to_datetime(out["expiration"]) + - out["timestamp"].dt.tz_convert(None).dt.normalize() + ).dt.days + out = out[out["dte"] >= 0].copy() + + key = ["timestamp", "symbol", "expiration", "strike", "option_type"] + if out.duplicated(key).any(): + raise ValueError("duplicate contract snapshots found") + + return out.sort_values(key).reset_index(drop=True) + + +def load_option_chain_csv(path: str | Path) -> pd.DataFrame: + path = Path(path) + if not path.exists(): + raise ValueError(f"option-chain CSV not found: {path}") + return normalize_option_chain(pd.read_csv(path)) + + +def chain_at_or_before( + frame: pd.DataFrame, + timestamp, + symbol: str | None = None, +) -> pd.DataFrame: + """Latest complete-ish snapshot at or before a timestamp. + + For intraday vendor data, contracts can update at slightly different + timestamps. We therefore take the latest row per contract at/before the + requested instant rather than requiring a single exact timestamp. + """ + + data = normalize_option_chain(frame) + ts = pd.Timestamp(timestamp) + ts = ts.tz_localize("UTC") if ts.tzinfo is None else ts.tz_convert("UTC") + data = data[data["timestamp"] <= ts] + if symbol: + data = data[data["symbol"] == symbol.upper()] + if data.empty: + return data + key = ["symbol", "expiration", "strike", "option_type"] + idx = data.groupby(key)["timestamp"].idxmax() + return data.loc[idx].sort_values(["expiration", "option_type", "strike"]).reset_index(drop=True) diff --git a/src/neuroquant/options_lab/execution.py b/src/neuroquant/options_lab/execution.py new file mode 100644 index 0000000..a9ed4a7 --- /dev/null +++ b/src/neuroquant/options_lab/execution.py @@ -0,0 +1,152 @@ +"""NBBO-aware execution model for realistic paper and historical fills.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Mapping + +import math + +from .models import Side + + +@dataclass(frozen=True) +class ExecutionConfig: + """Deterministic execution assumptions. + + slippage_half_spread_fraction controls how far from midpoint a realistic + fill moves toward the adverse side. Zero is midpoint and one crosses the + full half-spread to ask for buys or bid for sells. + """ + + commission_per_contract: float = 0.65 + slippage_half_spread_fraction: float = 0.35 + min_open_interest: int = 25 + min_volume: int = 0 + max_spread_pct: float = 0.35 + reject_zero_bid: bool = True + + def __post_init__(self) -> None: + if self.commission_per_contract < 0: + raise ValueError("commission_per_contract cannot be negative") + if not 0.0 <= self.slippage_half_spread_fraction <= 1.0: + raise ValueError("slippage_half_spread_fraction must be in [0, 1]") + if self.max_spread_pct <= 0: + raise ValueError("max_spread_pct must be positive") + + +def _number( + row: Mapping[str, Any], key: str, default: float | None = None +) -> float | None: + value = row.get(key, default) + if value is None: + return default + try: + value = float(value) + except (TypeError, ValueError): + return default + if math.isnan(value): + return default + return value + + +def quote_mid(row: Mapping[str, Any]) -> float: + bid = _number(row, "bid") + ask = _number(row, "ask") + if bid is None or ask is None: + raise ValueError("quote requires bid and ask") + if bid < 0 or ask < 0 or ask < bid: + raise ValueError("invalid NBBO quote") + return (bid + ask) / 2.0 + + +def quote_spread_pct(row: Mapping[str, Any]) -> float: + bid = float(row["bid"]) + ask = float(row["ask"]) + mid = (bid + ask) / 2.0 + return (ask - bid) / mid if mid > 0 else float("inf") + + +def is_liquid(row: Mapping[str, Any], config: ExecutionConfig) -> bool: + bid = _number(row, "bid", 0.0) or 0.0 + ask = _number(row, "ask", 0.0) or 0.0 + if ask < bid or ask <= 0: + return False + if config.reject_zero_bid and bid <= 0: + return False + if quote_spread_pct(row) > config.max_spread_pct: + return False + + oi = _number(row, "open_interest") + if oi is not None and oi < config.min_open_interest: + return False + volume = _number(row, "volume") + if volume is not None and volume < config.min_volume: + return False + return True + + +def execution_price( + row: Mapping[str, Any], + side: Side | str, + config: ExecutionConfig, + mode: str = "realistic", +) -> float: + """Estimate fill from observed NBBO. + + mid is optimistic benchmarking; realistic applies configured adverse + slippage; conservative crosses to ask for buys and bid for sells. + """ + + side = Side(side) + bid = float(row["bid"]) + ask = float(row["ask"]) + if ask < bid or bid < 0: + raise ValueError("invalid NBBO quote") + mid = (bid + ask) / 2.0 + half = (ask - bid) / 2.0 + + if mode == "mid": + return float(mid) + if mode == "conservative": + return float(ask if side is Side.BUY else bid) + if mode != "realistic": + raise ValueError("mode must be mid, realistic, or conservative") + + slip = config.slippage_half_spread_fraction * half + value = mid + slip if side is Side.BUY else mid - slip + return float(min(value, ask) if side is Side.BUY else max(value, bid)) + + +def leg_cash_flow( + row: Mapping[str, Any], + side: Side | str, + quantity: int, + config: ExecutionConfig, + mode: str = "realistic", + multiplier: int = 100, +) -> float: + """Signed dollar cash flow net of commission.""" + + if quantity <= 0: + raise ValueError("quantity must be positive") + side = Side(side) + px = execution_price(row, side, config, mode) + gross = px * multiplier * quantity + signed = -gross if side is Side.BUY else gross + commission = config.commission_per_contract * quantity + return float(signed - commission) + + +def combo_cash_flow( + legs: list[tuple[Mapping[str, Any], Side | str, int]], + config: ExecutionConfig, + mode: str = "realistic", + multiplier: int = 100, +) -> float: + return float( + sum( + leg_cash_flow(row, side, qty, config, mode, multiplier) + for row, side, qty in legs + ) + ) diff --git a/src/neuroquant/options_lab/models.py b/src/neuroquant/options_lab/models.py new file mode 100644 index 0000000..43d734a --- /dev/null +++ b/src/neuroquant/options_lab/models.py @@ -0,0 +1,119 @@ +"""Core option-domain models used by NeuroQuant Options Lab.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import date, datetime +from enum import Enum + + +class OptionType(str, Enum): + CALL = "call" + PUT = "put" + + +class Side(str, Enum): + BUY = "buy" + SELL = "sell" + + +@dataclass(frozen=True) +class OptionContract: + """Canonical option identity without vendor-specific IDs.""" + + symbol: str + expiration: date + strike: float + option_type: OptionType + multiplier: int = 100 + + def __post_init__(self) -> None: + if not self.symbol: + raise ValueError("symbol must be non-empty") + if self.strike <= 0: + raise ValueError("strike must be positive") + if self.multiplier <= 0: + raise ValueError("multiplier must be positive") + + +@dataclass(frozen=True) +class OptionQuote: + """One NBBO-style option quote snapshot.""" + + contract: OptionContract + timestamp: datetime + underlying_price: float + bid: float + ask: float + bid_size: int | None = None + ask_size: int | None = None + volume: int | None = None + open_interest: int | None = None + iv: float | None = None + delta: float | None = None + gamma: float | None = None + theta: float | None = None + vega: float | None = None + + def __post_init__(self) -> None: + if self.underlying_price <= 0: + raise ValueError("underlying_price must be positive") + if self.bid < 0 or self.ask < 0: + raise ValueError("bid/ask cannot be negative") + if self.ask < self.bid: + raise ValueError("crossed quote: ask is below bid") + + @property + def mid(self) -> float: + return (self.bid + self.ask) / 2.0 + + @property + def spread(self) -> float: + return self.ask - self.bid + + @property + def spread_pct(self) -> float: + return self.spread / self.mid if self.mid > 0 else float("inf") + + +@dataclass(frozen=True) +class OptionLeg: + contract: OptionContract + side: Side + quantity: int = 1 + + def __post_init__(self) -> None: + if self.quantity <= 0: + raise ValueError("quantity must be positive") + + +@dataclass(frozen=True) +class BullCallSpread: + """Long lower-strike call plus short higher-strike call.""" + + long_call: OptionContract + short_call: OptionContract + quantity: int = 1 + + def __post_init__(self) -> None: + if self.long_call.symbol != self.short_call.symbol: + raise ValueError("spread legs must share the same underlying") + if self.long_call.expiration != self.short_call.expiration: + raise ValueError("spread legs must share the same expiration") + if self.long_call.option_type is not OptionType.CALL: + raise ValueError("long leg must be a call") + if self.short_call.option_type is not OptionType.CALL: + raise ValueError("short leg must be a call") + if self.long_call.strike >= self.short_call.strike: + raise ValueError("bull call long strike must be below short strike") + if self.quantity <= 0: + raise ValueError("quantity must be positive") + + @property + def width(self) -> float: + return self.short_call.strike - self.long_call.strike + + def expiry_value_per_share(self, spot: float) -> float: + long_intrinsic = max(spot - self.long_call.strike, 0.0) + short_intrinsic = max(spot - self.short_call.strike, 0.0) + return long_intrinsic - short_intrinsic diff --git a/src/neuroquant/options_lab/paper.py b/src/neuroquant/options_lab/paper.py new file mode 100644 index 0000000..562aacb --- /dev/null +++ b/src/neuroquant/options_lab/paper.py @@ -0,0 +1,56 @@ +"""Shadow-ledger utilities for reconciling broker paper fills with realistic fills.""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from pathlib import Path + +import pandas as pd + + +@dataclass(frozen=True) +class ShadowTrade: + trade_id: str + timestamp: str + symbol: str + strategy: str + action: str + broker_fill: float + realistic_fill: float + conservative_fill: float + quantity: int + notes: str = "" + + +COLUMNS = list(ShadowTrade.__annotations__.keys()) + + +def append_shadow_trade(path: str | Path, trade: ShadowTrade) -> pd.DataFrame: + """Append one immutable paper-vs-realistic execution record.""" + + path = Path(path) + row = pd.DataFrame([asdict(trade)]) + if path.exists(): + existing = pd.read_csv(path) + out = pd.concat([existing, row], ignore_index=True) + else: + out = row + if out["trade_id"].duplicated().any(): + raise ValueError("trade_id must be unique") + path.parent.mkdir(parents=True, exist_ok=True) + out.to_csv(path, index=False) + return out + + +def reconciliation_report(frame: pd.DataFrame) -> dict: + if frame.empty: + return {"fills": 0, "mean_broker_minus_realistic": 0.0, "mean_realistic_minus_conservative": 0.0} + return { + "fills": int(len(frame)), + "mean_broker_minus_realistic": float( + (frame["broker_fill"] - frame["realistic_fill"]).mean() + ), + "mean_realistic_minus_conservative": float( + (frame["realistic_fill"] - frame["conservative_fill"]).mean() + ), + } diff --git a/src/neuroquant/options_lab/pricing.py b/src/neuroquant/options_lab/pricing.py new file mode 100644 index 0000000..39816e8 --- /dev/null +++ b/src/neuroquant/options_lab/pricing.py @@ -0,0 +1,170 @@ +"""Dependency-free Black-Scholes-Merton pricing and Greeks. + +Historical backtests should prefer observed NBBO prices over theoretical prices +whenever market quotes are available. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from math import erf, exp, log, pi, sqrt + +from .models import OptionType + + +def _norm_cdf(x: float) -> float: + return 0.5 * (1.0 + erf(x / sqrt(2.0))) + + +def _norm_pdf(x: float) -> float: + return exp(-0.5 * x * x) / sqrt(2.0 * pi) + + +@dataclass(frozen=True) +class Greeks: + delta: float + gamma: float + theta_per_day: float + vega_per_vol_point: float + rho_per_rate_point: float + + +def _d1_d2( + spot: float, + strike: float, + t_years: float, + rate: float, + volatility: float, + dividend_yield: float, +) -> tuple[float, float]: + if min(spot, strike, t_years, volatility) <= 0: + raise ValueError("spot, strike, t_years and volatility must be positive") + root_t = sqrt(t_years) + d1 = ( + log(spot / strike) + + (rate - dividend_yield + 0.5 * volatility * volatility) * t_years + ) / (volatility * root_t) + return d1, d1 - volatility * root_t + + +def black_scholes_price( + spot: float, + strike: float, + t_years: float, + rate: float, + volatility: float, + option_type: OptionType | str = OptionType.CALL, + dividend_yield: float = 0.0, +) -> float: + """Return BSM price per share.""" + + kind = OptionType(option_type) + if t_years <= 0: + if kind is OptionType.CALL: + return max(spot - strike, 0.0) + return max(strike - spot, 0.0) + d1, d2 = _d1_d2( + spot, strike, t_years, rate, volatility, dividend_yield + ) + discounted_spot = spot * exp(-dividend_yield * t_years) + discounted_strike = strike * exp(-rate * t_years) + if kind is OptionType.CALL: + return discounted_spot * _norm_cdf(d1) - discounted_strike * _norm_cdf(d2) + return discounted_strike * _norm_cdf(-d2) - discounted_spot * _norm_cdf(-d1) + + +def black_scholes_greeks( + spot: float, + strike: float, + t_years: float, + rate: float, + volatility: float, + option_type: OptionType | str = OptionType.CALL, + dividend_yield: float = 0.0, +) -> Greeks: + """Return common Greeks. + + Theta is dollars per share per calendar day. + Vega and rho are dollars per share for a one percentage-point change. + """ + + kind = OptionType(option_type) + d1, d2 = _d1_d2( + spot, strike, t_years, rate, volatility, dividend_yield + ) + root_t = sqrt(t_years) + discount_q = exp(-dividend_yield * t_years) + discount_r = exp(-rate * t_years) + pdf = _norm_pdf(d1) + + if kind is OptionType.CALL: + delta = discount_q * _norm_cdf(d1) + theta_year = ( + -(spot * discount_q * pdf * volatility) / (2.0 * root_t) + - rate * strike * discount_r * _norm_cdf(d2) + + dividend_yield * spot * discount_q * _norm_cdf(d1) + ) + rho = strike * t_years * discount_r * _norm_cdf(d2) + else: + delta = discount_q * (_norm_cdf(d1) - 1.0) + theta_year = ( + -(spot * discount_q * pdf * volatility) / (2.0 * root_t) + + rate * strike * discount_r * _norm_cdf(-d2) + - dividend_yield * spot * discount_q * _norm_cdf(-d1) + ) + rho = -strike * t_years * discount_r * _norm_cdf(-d2) + + gamma = discount_q * pdf / (spot * volatility * root_t) + vega = spot * discount_q * pdf * root_t + + return Greeks( + delta=float(delta), + gamma=float(gamma), + theta_per_day=float(theta_year / 365.0), + vega_per_vol_point=float(vega / 100.0), + rho_per_rate_point=float(rho / 100.0), + ) + + +def implied_volatility( + market_price: float, + spot: float, + strike: float, + t_years: float, + rate: float, + option_type: OptionType | str = OptionType.CALL, + dividend_yield: float = 0.0, + lower: float = 1e-4, + upper: float = 5.0, + tolerance: float = 1e-7, + max_iterations: int = 200, +) -> float: + """Solve implied volatility with bisection.""" + + if market_price < 0: + raise ValueError("market_price cannot be negative") + if t_years <= 0: + raise ValueError("implied volatility is undefined at or after expiration") + + low_price = black_scholes_price( + spot, strike, t_years, rate, lower, option_type, dividend_yield + ) + high_price = black_scholes_price( + spot, strike, t_years, rate, upper, option_type, dividend_yield + ) + if not (low_price - tolerance <= market_price <= high_price + tolerance): + raise ValueError("market_price is outside the model solvable range") + + lo, hi = lower, upper + for _ in range(max_iterations): + mid = (lo + hi) / 2.0 + value = black_scholes_price( + spot, strike, t_years, rate, mid, option_type, dividend_yield + ) + if abs(value - market_price) <= tolerance: + return float(mid) + if value < market_price: + lo = mid + else: + hi = mid + return float((lo + hi) / 2.0) diff --git a/src/neuroquant/options_lab/providers.py b/src/neuroquant/options_lab/providers.py new file mode 100644 index 0000000..fcf32ff --- /dev/null +++ b/src/neuroquant/options_lab/providers.py @@ -0,0 +1,59 @@ +"""Provider-neutral interfaces for future market-data and paper-broker adapters. + +Concrete network adapters are intentionally not bundled in v1. Keeping provider +I/O behind protocols prevents research logic from depending on ThetaData, +Cboe, IBKR, or any single vendor. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import Protocol + +import pandas as pd + + +class OptionDataProvider(Protocol): + def chain_snapshot(self, symbol: str, timestamp: datetime) -> pd.DataFrame: + """Return observed option quotes available at timestamp.""" + + +class PaperBroker(Protocol): + def submit_vertical( + self, + symbol: str, + expiration: str, + long_strike: float, + short_strike: float, + quantity: int, + limit_debit: float | None = None, + ) -> str: + """Submit a defined-risk paper vertical and return broker order ID.""" + + +@dataclass(frozen=True) +class ProviderCapabilities: + historical_nbbo: bool + greeks: bool + open_interest: bool + intraday: bool + paper_orders: bool + + +RECOMMENDED_CAPABILITY_PROFILES = { + "historical_options_vendor": ProviderCapabilities( + historical_nbbo=True, + greeks=True, + open_interest=True, + intraday=True, + paper_orders=False, + ), + "paper_broker": ProviderCapabilities( + historical_nbbo=False, + greeks=True, + open_interest=True, + intraday=True, + paper_orders=True, + ), +} diff --git a/src/neuroquant/options_lab/research.py b/src/neuroquant/options_lab/research.py new file mode 100644 index 0000000..9343bc7 --- /dev/null +++ b/src/neuroquant/options_lab/research.py @@ -0,0 +1,88 @@ +"""Validation and robustness tools for option-trade result tables.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + + +def chronological_trade_split( + trades: pd.DataFrame, + train_fraction: float = 0.70, +) -> tuple[pd.DataFrame, pd.DataFrame]: + """Chronological train/test split; never shuffle time-series trades.""" + + if not 0.1 <= train_fraction <= 0.9: + raise ValueError("train_fraction must be between 0.1 and 0.9") + if trades.empty: + return trades.copy(), trades.copy() + data = trades.sort_values("entry_timestamp").reset_index(drop=True) + cut = max(1, min(len(data) - 1, int(len(data) * train_fraction))) + if len(data) == 1: + return data.copy(), data.iloc[0:0].copy() + return data.iloc[:cut].copy(), data.iloc[cut:].copy() + + +def bootstrap_trade_returns( + trades: pd.DataFrame, + n_simulations: int = 2000, + seed: int = 42, +) -> pd.DataFrame: + """Bootstrap trade return sequences for robustness, not forecasting.""" + + if n_simulations <= 0: + raise ValueError("n_simulations must be positive") + if "return_on_debit" not in trades.columns or trades.empty: + raise ValueError("trades requires non-empty return_on_debit") + values = trades["return_on_debit"].dropna().to_numpy(dtype=float) + if len(values) == 0: + raise ValueError("no finite trade returns") + + rng = np.random.default_rng(seed) + rows = [] + for i in range(n_simulations): + sample = rng.choice(values, size=len(values), replace=True) + equity = np.cumprod(1.0 + sample) + peaks = np.maximum.accumulate(equity) + dd = equity / peaks - 1.0 + rows.append( + { + "simulation": i, + "compound_return": float(equity[-1] - 1.0), + "max_drawdown": float(dd.min()), + "losing_path": bool(equity[-1] < 1.0), + } + ) + return pd.DataFrame(rows) + + +def execution_stress_summary( + scenario_results: dict[str, pd.DataFrame], +) -> pd.DataFrame: + """Compare already-run fill/cost scenarios in one table.""" + + rows = [] + for name, trades in scenario_results.items(): + if trades.empty: + rows.append( + { + "scenario": name, + "trade_count": 0, + "total_pnl": 0.0, + "mean_return_on_debit": float("nan"), + "win_rate": float("nan"), + } + ) + continue + rows.append( + { + "scenario": name, + "trade_count": int(len(trades)), + "total_pnl": float(trades["pnl"].sum()), + "mean_return_on_debit": float( + trades["return_on_debit"].mean() + ), + "win_rate": float((trades["pnl"] > 0).mean()), + } + ) + return pd.DataFrame(rows) diff --git a/src/neuroquant/options_lab/selector.py b/src/neuroquant/options_lab/selector.py new file mode 100644 index 0000000..fe26ac7 --- /dev/null +++ b/src/neuroquant/options_lab/selector.py @@ -0,0 +1,111 @@ +"""Contract selection for reproducible vertical-spread research.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pandas as pd + +from .execution import ExecutionConfig, is_liquid + + +@dataclass(frozen=True) +class BullCallSelectionConfig: + min_dte: int = 60 + max_dte: int = 150 + target_long_delta: float = 0.60 + target_short_delta: float = 0.30 + delta_tolerance: float = 0.20 + min_width_pct_spot: float = 0.08 + max_width_pct_spot: float = 0.40 + require_iv_below_rv: bool = False + + def __post_init__(self) -> None: + if self.min_dte < 1 or self.max_dte < self.min_dte: + raise ValueError("invalid DTE range") + if not 0 < self.target_short_delta < self.target_long_delta < 1: + raise ValueError("target deltas must satisfy 0 < short < long < 1") + if self.delta_tolerance <= 0: + raise ValueError("delta_tolerance must be positive") + + +def _candidate_calls( + chain: pd.DataFrame, + selection: BullCallSelectionConfig, + execution: ExecutionConfig, +) -> pd.DataFrame: + calls = chain[ + (chain["option_type"] == "call") + & chain["dte"].between(selection.min_dte, selection.max_dte) + ].copy() + if calls.empty: + return calls + calls = calls[ + calls.apply(lambda row: is_liquid(row, execution), axis=1) + ].copy() + if selection.require_iv_below_rv: + if not {"iv", "realized_vol"}.issubset(calls.columns): + raise ValueError("IV dict: + """Select a liquid bull call spread without peeking forward. + + Expiration is chosen by closeness to the midpoint of the DTE window. + Within that expiration, long and short legs are chosen by target delta. + """ + + selection = selection or BullCallSelectionConfig() + execution = execution or ExecutionConfig() + calls = _candidate_calls(chain, selection, execution) + if calls.empty: + raise ValueError("no liquid call candidates satisfy the filters") + if "delta" not in calls.columns or calls["delta"].isna().all(): + raise ValueError("delta is required for deterministic contract selection") + + target_dte = (selection.min_dte + selection.max_dte) / 2.0 + expiries = ( + calls.groupby("expiration", as_index=False)["dte"] + .median() + .assign(distance=lambda x: (x["dte"] - target_dte).abs()) + .sort_values(["distance", "dte"]) + ) + expiry = expiries.iloc[0]["expiration"] + sub = calls[calls["expiration"] == expiry].copy() + + sub["long_error"] = (sub["delta"] - selection.target_long_delta).abs() + long = sub.sort_values(["long_error", "strike"]).iloc[0] + if long["long_error"] > selection.delta_tolerance: + raise ValueError("no long call close enough to target delta") + + spot = float(long["underlying_price"]) + shorts = sub[sub["strike"] > float(long["strike"])].copy() + if shorts.empty: + raise ValueError("no higher-strike call available for short leg") + shorts["width_pct"] = (shorts["strike"] - float(long["strike"])) / spot + shorts = shorts[ + shorts["width_pct"].between( + selection.min_width_pct_spot, selection.max_width_pct_spot + ) + ].copy() + if shorts.empty: + raise ValueError("no short call satisfies spread-width constraints") + + shorts["short_error"] = (shorts["delta"] - selection.target_short_delta).abs() + short = shorts.sort_values(["short_error", "strike"]).iloc[0] + if short["short_error"] > selection.delta_tolerance: + raise ValueError("no short call close enough to target delta") + + return { + "long": long.to_dict(), + "short": short.to_dict(), + "expiration": expiry, + "spot": spot, + "width": float(short["strike"] - long["strike"]), + } diff --git a/src/neuroquant/options_lab/strategies.py b/src/neuroquant/options_lab/strategies.py new file mode 100644 index 0000000..8b8a6fa --- /dev/null +++ b/src/neuroquant/options_lab/strategies.py @@ -0,0 +1,99 @@ +"""Causal entry-signal recipes that feed the options backtester.""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pandas as pd + +from .volatility import realized_volatility + + +@dataclass(frozen=True) +class CheapVolBullishConfig: + """Simple explainable signal for long-volatility bullish research.""" + + fast_ma: int = 20 + slow_ma: int = 60 + rv_window: int = 20 + max_iv_to_rv: float = 1.0 + cooldown_days: int = 10 + + def __post_init__(self) -> None: + if not 1 <= self.fast_ma < self.slow_ma: + raise ValueError("require 1 <= fast_ma < slow_ma") + if self.rv_window < 2: + raise ValueError("rv_window must be at least 2") + if self.max_iv_to_rv <= 0: + raise ValueError("max_iv_to_rv must be positive") + if self.cooldown_days < 0: + raise ValueError("cooldown_days cannot be negative") + + +def build_cheap_vol_bullish_entries( + underlying: pd.DataFrame, + atm_iv: pd.DataFrame, + config: CheapVolBullishConfig | None = None, +) -> pd.DataFrame: + """Generate causal daily entry decisions. + + The rule is deliberately interpretable: + 1. prior-close fast MA > slow MA; + 2. prior observed ATM IV / prior realized volatility <= threshold; + 3. a cooldown prevents overlapping daily entries. + + Both trend and volatility inputs are shifted one bar before the decision. + """ + + config = config or CheapVolBullishConfig() + if not {"timestamp", "close"}.issubset(underlying.columns): + raise ValueError("underlying requires timestamp and close") + if not {"timestamp", "iv"}.issubset(atm_iv.columns): + raise ValueError("atm_iv requires timestamp and iv") + + u = underlying[["timestamp", "close"]].copy() + u["timestamp"] = pd.to_datetime(u["timestamp"], utc=True) + u = u.sort_values("timestamp").drop_duplicates("timestamp") + u["fast_ma"] = u["close"].rolling(config.fast_ma).mean() + u["slow_ma"] = u["close"].rolling(config.slow_ma).mean() + u["realized_vol"] = realized_volatility(u["close"], config.rv_window) + + v = atm_iv[["timestamp", "iv"]].copy() + v["timestamp"] = pd.to_datetime(v["timestamp"], utc=True) + v = v.sort_values("timestamp").drop_duplicates("timestamp") + + data = pd.merge_asof(u, v, on="timestamp", direction="backward") + for col in ["fast_ma", "slow_ma", "realized_vol", "iv"]: + data[f"lag_{col}"] = data[col].shift(1) + + data["iv_to_rv"] = data["lag_iv"] / data["lag_realized_vol"] + raw = ( + (data["lag_fast_ma"] > data["lag_slow_ma"]) + & (data["iv_to_rv"] <= config.max_iv_to_rv) + ).fillna(False) + + entries = [] + last_entry = None + for ts, signal in zip(data["timestamp"], raw): + allowed = ( + last_entry is None + or (ts - last_entry).days >= config.cooldown_days + ) + take = bool(signal and allowed) + entries.append(take) + if take: + last_entry = ts + + data["entry_signal"] = entries + return data[ + [ + "timestamp", + "close", + "lag_fast_ma", + "lag_slow_ma", + "lag_iv", + "lag_realized_vol", + "iv_to_rv", + "entry_signal", + ] + ].copy() diff --git a/src/neuroquant/options_lab/volatility.py b/src/neuroquant/options_lab/volatility.py new file mode 100644 index 0000000..22c1965 --- /dev/null +++ b/src/neuroquant/options_lab/volatility.py @@ -0,0 +1,76 @@ +"""Volatility features for option research.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + + +TRADING_DAYS = 252 + + +def realized_volatility(close: pd.Series, window: int = 20) -> pd.Series: + """Annualized close-to-close realized volatility.""" + + if window < 2: + raise ValueError("window must be at least 2") + prices = pd.Series(close, dtype=float) + if (prices <= 0).any(): + raise ValueError("close prices must be positive") + log_returns = np.log(prices).diff() + return log_returns.rolling(window).std(ddof=0) * np.sqrt(TRADING_DAYS) + + +def attach_realized_volatility( + chain: pd.DataFrame, + underlying: pd.DataFrame, + window: int = 20, + timestamp_col: str = "timestamp", +) -> pd.DataFrame: + """Backward as-of merge realized volatility onto option snapshots.""" + + if timestamp_col not in chain.columns: + raise ValueError("chain requires timestamp") + if timestamp_col not in underlying.columns or "close" not in underlying.columns: + raise ValueError("underlying requires timestamp and close") + + u = underlying[[timestamp_col, "close"]].copy() + u[timestamp_col] = pd.to_datetime(u[timestamp_col], utc=True) + u = u.sort_values(timestamp_col) + u["realized_vol"] = realized_volatility(u["close"], window) + + c = chain.copy() + c[timestamp_col] = pd.to_datetime(c[timestamp_col], utc=True) + c = c.sort_values(timestamp_col) + return pd.merge_asof( + c, + u[[timestamp_col, "realized_vol"]], + on=timestamp_col, + direction="backward", + ) + + +def iv_to_rv_ratio(iv: pd.Series, realized_vol: pd.Series) -> pd.Series: + """IV divided by realized volatility. Values below one mean IV is lower.""" + + iv = pd.Series(iv, dtype=float) + rv = pd.Series(realized_vol, dtype=float) + return iv / rv.replace(0.0, np.nan) + + +def long_volatility_candidate_mask( + frame: pd.DataFrame, + max_iv_to_rv: float = 1.0, + max_iv_rank: float | None = None, +) -> pd.Series: + """Research filter for relatively inexpensive long-volatility setups.""" + + if "iv" not in frame.columns or "realized_vol" not in frame.columns: + raise ValueError("frame requires iv and realized_vol") + ratio = iv_to_rv_ratio(frame["iv"], frame["realized_vol"]) + mask = ratio <= max_iv_to_rv + if max_iv_rank is not None: + if "iv_rank" not in frame.columns: + raise ValueError("max_iv_rank requires an iv_rank column") + mask &= frame["iv_rank"] <= max_iv_rank + return mask.fillna(False) diff --git a/tests/test_options_lab.py b/tests/test_options_lab.py new file mode 100644 index 0000000..97f5803 --- /dev/null +++ b/tests/test_options_lab.py @@ -0,0 +1,210 @@ +"""Tests for NeuroQuant Options Lab.""" + +from __future__ import annotations + +from datetime import date + +import pandas as pd +import pytest + +from neuroquant.options_lab import ( + BullCallSelectionConfig, + BullCallSpread, + ExecutionConfig, + OptionContract, + OptionType, + Side, + VerticalBacktestConfig, + backtest_bull_call_entries, + black_scholes_greeks, + black_scholes_price, + execution_price, + implied_volatility, + normalize_option_chain, + realized_volatility, + select_bull_call_spread, + summarize_vertical_backtest, +) + + +def test_bsm_call_known_value_and_iv_roundtrip(): + price = black_scholes_price( + spot=100, + strike=100, + t_years=1, + rate=0.05, + volatility=0.20, + option_type="call", + ) + assert price == pytest.approx(10.4506, rel=1e-4) + iv = implied_volatility(price, 100, 100, 1, 0.05, "call") + assert iv == pytest.approx(0.20, rel=1e-5) + + +def test_greeks_are_sensible(): + g = black_scholes_greeks(100, 100, 0.5, 0.04, 0.30, "call") + assert 0 < g.delta < 1 + assert g.gamma > 0 + assert g.vega_per_vol_point > 0 + assert g.theta_per_day < 0 + + +def test_bull_call_expiry_value_is_capped(): + long = OptionContract("XYZ", date(2027, 1, 15), 100, OptionType.CALL) + short = OptionContract("XYZ", date(2027, 1, 15), 120, OptionType.CALL) + spread = BullCallSpread(long, short) + assert spread.expiry_value_per_share(90) == 0 + assert spread.expiry_value_per_share(110) == 10 + assert spread.expiry_value_per_share(150) == 20 + + +def test_execution_realistic_is_between_mid_and_touch(): + row = {"bid": 4.0, "ask": 5.0, "open_interest": 100, "volume": 10} + cfg = ExecutionConfig(slippage_half_spread_fraction=0.4) + buy = execution_price(row, Side.BUY, cfg) + sell = execution_price(row, Side.SELL, cfg) + assert 4.5 < buy < 5.0 + assert 4.0 < sell < 4.5 + + +def test_chain_normalization_and_selector(): + ts = "2026-09-28T19:00:00Z" + expiry = "2027-01-15" + rows = [] + for strike, delta, bid, ask in [ + (90, 0.75, 15.0, 15.4), + (100, 0.61, 10.0, 10.4), + (110, 0.46, 6.0, 6.3), + (120, 0.31, 3.2, 3.4), + (130, 0.20, 1.7, 1.9), + ]: + rows.append( + { + "timestamp": ts, + "symbol": "XYZ", + "expiration": expiry, + "strike": strike, + "option_type": "C", + "underlying_price": 105.0, + "bid": bid, + "ask": ask, + "open_interest": 500, + "volume": 50, + "delta": delta, + "iv": 0.35, + } + ) + chain = normalize_option_chain(pd.DataFrame(rows)) + pick = select_bull_call_spread( + chain, + BullCallSelectionConfig( + min_dte=60, + max_dte=150, + target_long_delta=0.60, + target_short_delta=0.30, + min_width_pct_spot=0.10, + max_width_pct_spot=0.30, + ), + ExecutionConfig(max_spread_pct=0.20), + ) + assert pick["long"]["strike"] == 100 + assert pick["short"]["strike"] == 120 + + +def test_realized_volatility_is_positive_after_warmup(): + close = pd.Series([100 + i + (i % 3) for i in range(40)], dtype=float) + rv = realized_volatility(close, 10) + assert rv.iloc[:9].isna().all() + assert rv.dropna().gt(0).all() + + +def test_vertical_backtest_uses_observed_quotes_and_closes_profitably(): + entry = pd.Timestamp("2026-09-01T20:00:00Z") + later = pd.Timestamp("2026-09-21T20:00:00Z") + expiry = "2026-12-18" + + chain = pd.DataFrame( + [ + { + "timestamp": entry, + "symbol": "XYZ", + "expiration": expiry, + "strike": 100, + "option_type": "call", + "underlying_price": 105, + "bid": 9.8, + "ask": 10.2, + "open_interest": 1000, + "volume": 100, + "delta": 0.60, + "iv": 0.35, + }, + { + "timestamp": entry, + "symbol": "XYZ", + "expiration": expiry, + "strike": 120, + "option_type": "call", + "underlying_price": 105, + "bid": 3.0, + "ask": 3.2, + "open_interest": 1000, + "volume": 100, + "delta": 0.30, + "iv": 0.35, + }, + { + "timestamp": later, + "symbol": "XYZ", + "expiration": expiry, + "strike": 100, + "option_type": "call", + "underlying_price": 122, + "bid": 23.0, + "ask": 23.4, + "open_interest": 1000, + "volume": 100, + "delta": 0.82, + "iv": 0.37, + }, + { + "timestamp": later, + "symbol": "XYZ", + "expiration": expiry, + "strike": 120, + "option_type": "call", + "underlying_price": 122, + "bid": 8.0, + "ask": 8.3, + "open_interest": 1000, + "volume": 100, + "delta": 0.55, + "iv": 0.37, + }, + ] + ) + + config = VerticalBacktestConfig( + selection=BullCallSelectionConfig( + min_dte=60, + max_dte=150, + min_width_pct_spot=0.10, + max_width_pct_spot=0.30, + ), + execution=ExecutionConfig( + commission_per_contract=0.65, + slippage_half_spread_fraction=0.0, + max_spread_pct=0.20, + ), + max_hold_days=40, + profit_target_pct_max=0.40, + stop_loss_pct_debit=None, + fill_mode="mid", + ) + trades = backtest_bull_call_entries(chain, [entry], "XYZ", config) + assert len(trades) == 1 + assert trades.iloc[0]["pnl"] > 0 + assert trades.iloc[0]["exit_reason"] == "profit_target" + summary = summarize_vertical_backtest(trades) + assert summary["trade_count"] == 1 + assert summary["win_rate"] == 1.0 diff --git a/tests/test_options_research.py b/tests/test_options_research.py new file mode 100644 index 0000000..bf78813 --- /dev/null +++ b/tests/test_options_research.py @@ -0,0 +1,55 @@ +"""Research-protocol tests for Options Lab.""" + +import pandas as pd + +from neuroquant.options_lab import ( + CheapVolBullishConfig, + bootstrap_trade_returns, + build_cheap_vol_bullish_entries, + chronological_trade_split, +) + + +def test_cheap_vol_signal_is_lagged_and_has_cooldown(): + dates = pd.date_range("2026-01-01", periods=100, freq="B", tz="UTC") + underlying = pd.DataFrame( + {"timestamp": dates, "close": [100 + i * 0.5 for i in range(100)]} + ) + atm_iv = pd.DataFrame( + {"timestamp": dates, "iv": [0.05] * 100} + ) + result = build_cheap_vol_bullish_entries( + underlying, + atm_iv, + CheapVolBullishConfig( + fast_ma=5, + slow_ma=10, + rv_window=5, + max_iv_to_rv=1_000_000.0, + cooldown_days=10, + ), + ) + entries = result[result["entry_signal"]] + assert len(entries) > 1 + gaps = entries["timestamp"].diff().dropna().dt.days + assert (gaps >= 10).all() + + +def test_chronological_split_keeps_order(): + trades = pd.DataFrame( + { + "entry_timestamp": pd.to_datetime( + ["2026-03-01", "2026-01-01", "2026-02-01"], utc=True + ), + "return_on_debit": [0.3, 0.1, -0.2], + } + ) + train, test = chronological_trade_split(trades, 0.67) + assert train["entry_timestamp"].max() < test["entry_timestamp"].min() + + +def test_bootstrap_trade_returns_is_reproducible(): + trades = pd.DataFrame({"return_on_debit": [0.1, -0.05, 0.2, -0.1]}) + a = bootstrap_trade_returns(trades, n_simulations=20, seed=7) + b = bootstrap_trade_returns(trades, n_simulations=20, seed=7) + pd.testing.assert_frame_equal(a, b)