from __future__ import annotations from datetime import datetime import pandas as pd from .arbitrage import ( assign_candidate_confidence, scan_arbitrage_candidates, summarize_arbitrage, ) from .data import fetch_option_snapshot from .features import compute_features from .no_arb import run_all_checks, run_checks_by_expiry, select_best_quality_expiry from .oipd_adapter import OIPDConfig, fit_oipd_distribution, prepare_oipd_chain from .strategy import generate_candidates, score_candidates, select_best from .types import DensityEstimate, OptionSnapshot def _available_expiries(snapshot: OptionSnapshot) -> pd.DataFrame: expiries = ( snapshot.options[snapshot.options["option_type"] == "call"]["expiry"] .dropna() .drop_duplicates() .sort_values() ) now = pd.Timestamp(snapshot.snapshot_time).tz_localize(None) df = pd.DataFrame({"expiry": pd.to_datetime(expiries)}) df["dte_days"] = (df["expiry"] - now).dt.days return df.reset_index(drop=True) def _filter_chain_for_v2( options: pd.DataFrame, spot: float, expiry: datetime, moneyness_band: float, min_open_interest: int, min_volume: int, ) -> pd.DataFrame: out = options[pd.to_datetime(options["expiry"]) == pd.to_datetime(expiry)].copy() out = out[ (out["strike"] >= (1.0 - moneyness_band) * spot) & (out["strike"] <= (1.0 + moneyness_band) * spot) ] if "openInterest" in out.columns: out = out[out["openInterest"].fillna(0) >= min_open_interest] if "volume" in out.columns: out = out[out["volume"].fillna(0) >= min_volume] return out def analyze_snapshot_v2( snapshot: OptionSnapshot, expiry: datetime | None = None, moneyness_band: float = 0.2, min_open_interest: int = 0, min_volume: int = 0, risk_lambda: float = 0.5, oipd_config: OIPDConfig | None = None, arb_min_edge: float = 0.0, arb_min_edge_per_width: float = 0.0, arb_min_leg_open_interest: int = 0, ) -> dict[str, object]: cfg = oipd_config or OIPDConfig() diagnostics = run_all_checks(snapshot) per_expiry_diag = run_checks_by_expiry(snapshot) available_expiries = _available_expiries(snapshot) diag_df = pd.DataFrame( [ { "check": d.name, "passed": d.passed, "violations": d.violations, "comparisons": d.comparisons, "violation_rate": d.violation_rate, "details": d.details, } for d in diagnostics ] ) per_expiry_summary = ( per_expiry_diag.groupby("expiry", as_index=False) .agg( failed_checks=("passed", lambda s: int((~s).sum())), total_checks=("passed", "count"), mean_violation_rate=("violation_rate", "mean"), ) .sort_values("mean_violation_rate") .reset_index(drop=True) ) if expiry is None: expiry = select_best_quality_expiry(snapshot) filtered = _filter_chain_for_v2( snapshot.options, spot=snapshot.spot, expiry=expiry, moneyness_band=moneyness_band, min_open_interest=min_open_interest, min_volume=min_volume, ) if filtered.empty or len(filtered) < 10: filtered = snapshot.options chain = prepare_oipd_chain(filtered, expiry=expiry) dist = fit_oipd_distribution( chain=chain, spot=snapshot.spot, valuation_time=snapshot.snapshot_time, config=cfg, ) density_df = ( dist.density[["strike", "density"]] .copy() .sort_values("strike") .reset_index(drop=True) ) density = DensityEstimate( strikes=density_df["strike"], density=density_df["density"], expiry=pd.to_datetime(expiry).to_pydatetime(), ) features = compute_features(snapshot, density) candidates = generate_candidates(snapshot.spot, density, snapshot.options) scored = score_candidates( candidates, density=density, spot0=snapshot.spot, risk_lambda=risk_lambda ) best = select_best(scored) scored_df = pd.DataFrame( [ { "strategy": s.candidate.name, "expected_payoff": s.expected_payoff, "downside_q05": s.downside_q05, "objective": s.objective, } for s in scored ] ) arbitrage_candidates = scan_arbitrage_candidates( snapshot.options, expiry=pd.to_datetime(expiry), min_edge=arb_min_edge, min_edge_per_width=arb_min_edge_per_width, min_leg_open_interest=arb_min_leg_open_interest, spot=snapshot.spot, ) exp_quality = per_expiry_summary[ per_expiry_summary["expiry"] == pd.to_datetime(expiry) ] mean_violation_rate = ( float(exp_quality.iloc[0]["mean_violation_rate"]) if not exp_quality.empty else 1.0 ) failed_checks = ( int(exp_quality.iloc[0]["failed_checks"]) if not exp_quality.empty else 3 ) arbitrage_candidates = assign_candidate_confidence( arbitrage_candidates, mean_violation_rate=mean_violation_rate, failed_checks=failed_checks, ) arbitrage_summary = summarize_arbitrage(arbitrage_candidates) return { "engine": "oipd_v2", "ticker": snapshot.ticker, "spot": snapshot.spot, "snapshot_time": snapshot.snapshot_time, "selected_expiry": pd.to_datetime(expiry).to_pydatetime(), "diagnostics": diag_df, "diagnostics_by_expiry": per_expiry_diag, "diagnostics_by_expiry_summary": per_expiry_summary, "available_expiries": available_expiries, "density": density_df, "features": features, "scored": scored_df, "best_strategy": { "name": best.candidate.name, "params": best.candidate.params, "expected_payoff": best.expected_payoff, "downside_q05": best.downside_q05, "objective": best.objective, }, "arbitrage_candidates": arbitrage_candidates, "arbitrage_summary": arbitrage_summary, "engine_meta": dist.metadata, } def analyze_ticker_v2( ticker: str, max_expiries: int = 8, expiry: datetime | None = None, expiry_mode: str = "auto", moneyness_band: float = 0.2, min_open_interest: int = 0, min_volume: int = 0, risk_lambda: float = 0.5, oipd_config: OIPDConfig | None = None, ) -> dict[str, object]: snapshot = fetch_option_snapshot(ticker=ticker, max_expiries=max_expiries) selected_expiry = expiry if expiry_mode == "auto": selected_expiry = expiry elif expiry_mode == "manual": selected_expiry = expiry else: raise ValueError(f"Unknown expiry_mode: {expiry_mode}") return analyze_snapshot_v2( snapshot=snapshot, expiry=selected_expiry, moneyness_band=moneyness_band, min_open_interest=min_open_interest, min_volume=min_volume, risk_lambda=risk_lambda, oipd_config=oipd_config, )