"""Strike selection and combo construction (``docs/03-algorithms.md §4``). Two responsibilities: * :func:`select_strikes` — given a full option chain and a directional bias, return the (short, long) option quotes that satisfy the documented selection rules, or ``None`` when no candidate exists. * :func:`build` — assemble a :class:`ComboProposal` ready to be sent to Cerbero core, with credit, max-loss and breakeven precomputed. Iron condor selection is intentionally out of scope here; it is built by the orchestrator as two independent vertical spreads. """ from __future__ import annotations from datetime import datetime from decimal import Decimal from uuid import UUID, uuid4 from pydantic import BaseModel, ConfigDict, Field from cerbero_bite.config import SpreadType, StrategyConfig from cerbero_bite.core.types import OptionLeg, OptionQuote, PutOrCall __all__ = ["ComboProposal", "build", "select_strikes"] class ComboProposal(BaseModel): """Trade proposal ready for Telegram pre-trade and Cerbero-core dispatch.""" model_config = ConfigDict(frozen=True, extra="forbid") proposal_id: UUID = Field(default_factory=uuid4) spread_type: SpreadType legs: list[OptionLeg] credit_target_eth: Decimal credit_target_usd: Decimal max_loss_eth: Decimal max_loss_usd: Decimal breakeven: Decimal spot_at_proposal: Decimal dvol_at_proposal: Decimal expiry: datetime # --------------------------------------------------------------------------- # select_strikes # --------------------------------------------------------------------------- def _option_type_for_bias(bias: SpreadType) -> PutOrCall | None: if bias == "bull_put": return "P" if bias == "bear_call": return "C" return None # iron_condor handled at orchestrator level def _dte_days(now: datetime, expiry: datetime) -> int: """Calendar days between *now* and *expiry*, floored to int (≥ 0).""" delta = expiry - now return delta.days def _pick_expiry( chain: list[OptionQuote], *, now: datetime, cfg: StrategyConfig, ) -> datetime | None: """Return the expiry whose DTE is in range and closest to ``dte_target``.""" sc = cfg.structure candidates: dict[datetime, int] = {} for q in chain: dte = _dte_days(now, q.expiry) if sc.dte_min <= dte <= sc.dte_max: candidates.setdefault(q.expiry, dte) if not candidates: return None return min(candidates, key=lambda exp: abs(candidates[exp] - sc.dte_target)) def _resolve_delta_band( sc: object, dvol_now: Decimal | None ) -> tuple[Decimal, Decimal, Decimal]: """Return (delta_target, delta_min, delta_max) per il regime DVOL corrente. Quando ``sc.delta_by_dvol`` è popolato e ``dvol_now`` è disponibile, sceglie la prima banda (ordinata ascending sulla ``dvol_under``) il cui ``dvol_under ≥ dvol_now``. Altrimenti torna ai valori statici di ``sc``. """ bands = list(getattr(sc, "delta_by_dvol", []) or []) if dvol_now is not None and bands: bands_sorted = sorted(bands, key=lambda b: b.dvol_under) for band in bands_sorted: if dvol_now <= band.dvol_under: return band.delta_target, band.delta_min, band.delta_max last = bands_sorted[-1] return last.delta_target, last.delta_min, last.delta_max return sc.delta_target, sc.delta_min, sc.delta_max def _select_short( quotes: list[OptionQuote], *, spot: Decimal, cfg: StrategyConfig, dvol_now: Decimal | None = None, ) -> OptionQuote | None: """Pick the short-leg quote with delta closest to target inside both bands.""" sc = cfg.structure.short_strike delta_target, delta_min, delta_max = _resolve_delta_band(sc, dvol_now) eligible: list[OptionQuote] = [] for q in quotes: dist = (q.strike - spot).copy_abs() / spot if not (sc.distance_otm_pct_min <= dist <= sc.distance_otm_pct_max): continue abs_delta = q.delta.copy_abs() if not (delta_min <= abs_delta <= delta_max): continue eligible.append(q) if not eligible: return None return min(eligible, key=lambda q: abs(q.delta.copy_abs() - delta_target)) def _select_long( quotes: list[OptionQuote], *, short: OptionQuote, spot: Decimal, bias: SpreadType, cfg: StrategyConfig, ) -> OptionQuote | None: """Pick the long-leg quote whose distance from short matches the target width.""" sw = cfg.structure.spread_width width_target = spot * sw.target_pct_of_spot width_min = spot * sw.min_pct_of_spot width_max = spot * sw.max_pct_of_spot if bias == "bull_put": target_strike = short.strike - width_target candidates = [q for q in quotes if q.strike < short.strike] else: # bear_call target_strike = short.strike + width_target candidates = [q for q in quotes if q.strike > short.strike] if not candidates: return None nearest = min(candidates, key=lambda q: (q.strike - target_strike).copy_abs()) width = (short.strike - nearest.strike).copy_abs() if not (width_min <= width <= width_max): return None return nearest def select_strikes( *, chain: list[OptionQuote], bias: SpreadType, spot: Decimal, now: datetime, cfg: StrategyConfig, dvol_now: Decimal | None = None, ) -> tuple[OptionQuote, OptionQuote] | None: """Return the (short, long) quotes for the requested vertical, or ``None``. Iron condor is *not* built here: callers should request the two legs (bull_put + bear_call) separately when they need an IC. """ opt_type = _option_type_for_bias(bias) if opt_type is None: return None expiry = _pick_expiry(chain, now=now, cfg=cfg) if expiry is None: return None typed = [q for q in chain if q.expiry == expiry and q.option_type == opt_type] if not typed: return None short = _select_short(typed, spot=spot, cfg=cfg, dvol_now=dvol_now) if short is None: return None long_candidates = [q for q in typed if q.instrument != short.instrument] long_ = _select_long(long_candidates, short=short, spot=spot, bias=bias, cfg=cfg) if long_ is None: return None width_usd = (short.strike - long_.strike).copy_abs() credit_eth = short.mid - long_.mid # credit ≤ 0 → ratio non-positive < min → falls through to None below. credit_usd = credit_eth * spot if (credit_usd / width_usd) < cfg.structure.credit_to_width_ratio_min: return None return short, long_ # --------------------------------------------------------------------------- # build # --------------------------------------------------------------------------- def _make_leg( quote: OptionQuote, *, side: str, n_contracts: int, ) -> OptionLeg: return OptionLeg( instrument=quote.instrument, side=side, # type: ignore[arg-type] strike=quote.strike, expiry=quote.expiry, type=quote.option_type, size=n_contracts, mid_price_eth=quote.mid, delta=quote.delta, gamma=quote.gamma, theta=quote.theta, vega=quote.vega, ) def build( *, short: OptionQuote, long_: OptionQuote, n_contracts: int, spot: Decimal, dvol: Decimal, cfg: StrategyConfig, # noqa: ARG001 — kept for symmetry with select_strikes now: datetime, # noqa: ARG001 — opening time captured by orchestrator spread_type: SpreadType, ) -> ComboProposal: """Assemble a :class:`ComboProposal` from the two selected quotes.""" width_per_contract_usd = (short.strike - long_.strike).copy_abs() credit_per_contract_eth = short.mid - long_.mid credit_per_contract_usd = credit_per_contract_eth * spot max_loss_per_contract_usd = width_per_contract_usd - credit_per_contract_usd n_dec = Decimal(n_contracts) credit_target_eth = credit_per_contract_eth * n_dec credit_target_usd = credit_per_contract_usd * n_dec max_loss_usd = max_loss_per_contract_usd * n_dec max_loss_eth = max_loss_usd / spot if spot > 0 else Decimal("0") if spread_type == "bull_put": breakeven = short.strike - credit_per_contract_usd else: # bear_call breakeven = short.strike + credit_per_contract_usd legs = [ _make_leg(short, side="SELL", n_contracts=n_contracts), _make_leg(long_, side="BUY", n_contracts=n_contracts), ] return ComboProposal( spread_type=spread_type, legs=legs, credit_target_eth=credit_target_eth, credit_target_usd=credit_target_usd, max_loss_eth=max_loss_eth, max_loss_usd=max_loss_usd, breakeven=breakeven, spot_at_proposal=spot, dvol_at_proposal=dvol, expiry=short.expiry, )