src/option_implied_lab/pipeline_v2.py

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,
    )