From fbb7753cc6a295ee78979010ba68526723e9c634 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Mon, 27 Apr 2026 10:14:06 +0200 Subject: [PATCH] Phase 1: core algorithms Implementa i sette algoritmi puri di docs/03-algorithms.md con disciplina TDD: 112 test, copertura statement+branch al 100% su core/ e config/, mypy --strict pulito, ruff pulito. Moduli: - config/schema.py: StrategyConfig Pydantic v2 con validatori di consistenza (kelly, delta, OTM, spread width, profit/stop). - core/types.py: OptionQuote e OptionLeg condivisi. - core/entry_validator.py: validate_entry (accumula motivi) e compute_bias (bull_put/bear_call/iron_condor/None). - core/liquidity_gate.py: check OI/volume/spread/depth + slippage stimato in % del credito. - core/sizing_engine.py: Quarter Kelly con cap 200/1000 EUR e bande DVOL. - core/combo_builder.py: select_strikes (DTE/OTM/delta/width/credit) e build (ComboProposal con credit/max_loss/breakeven). - core/greeks_aggregator.py: somma firmata BUY/SELL, theta in USD. - core/exit_decision.py: 6 trigger ordinati con eccezione skip-time vicino a profit (mark in (50%,70%] credito). - core/kelly_recalibration.py: full/quarter Kelly, confidence per sample size, blend medio in fascia 30-99 trade. Co-Authored-By: Claude Opus 4.7 (1M context) --- pyproject.toml | 6 +- src/cerbero_bite/config/__init__.py | 43 +++ src/cerbero_bite/config/schema.py | 298 +++++++++++++++++ src/cerbero_bite/core/__init__.py | 7 + src/cerbero_bite/core/combo_builder.py | 253 ++++++++++++++ src/cerbero_bite/core/entry_validator.py | 155 +++++++++ src/cerbero_bite/core/exit_decision.py | 158 +++++++++ src/cerbero_bite/core/greeks_aggregator.py | 63 ++++ src/cerbero_bite/core/kelly_recalibration.py | 152 +++++++++ src/cerbero_bite/core/liquidity_gate.py | 135 ++++++++ src/cerbero_bite/core/sizing_engine.py | 92 ++++++ src/cerbero_bite/core/types.py | 72 ++++ tests/unit/test_combo_builder.py | 331 +++++++++++++++++++ tests/unit/test_config_schema.py | 104 ++++++ tests/unit/test_entry_validator.py | 247 ++++++++++++++ tests/unit/test_exit_decision.py | 273 +++++++++++++++ tests/unit/test_greeks_aggregator.py | 100 ++++++ tests/unit/test_kelly_recalibration.py | 155 +++++++++ tests/unit/test_liquidity_gate.py | 265 +++++++++++++++ tests/unit/test_sizing_engine.py | 182 ++++++++++ 20 files changed, 3090 insertions(+), 1 deletion(-) create mode 100644 src/cerbero_bite/config/schema.py create mode 100644 src/cerbero_bite/core/combo_builder.py create mode 100644 src/cerbero_bite/core/entry_validator.py create mode 100644 src/cerbero_bite/core/exit_decision.py create mode 100644 src/cerbero_bite/core/greeks_aggregator.py create mode 100644 src/cerbero_bite/core/kelly_recalibration.py create mode 100644 src/cerbero_bite/core/liquidity_gate.py create mode 100644 src/cerbero_bite/core/sizing_engine.py create mode 100644 src/cerbero_bite/core/types.py create mode 100644 tests/unit/test_combo_builder.py create mode 100644 tests/unit/test_config_schema.py create mode 100644 tests/unit/test_entry_validator.py create mode 100644 tests/unit/test_exit_decision.py create mode 100644 tests/unit/test_greeks_aggregator.py create mode 100644 tests/unit/test_kelly_recalibration.py create mode 100644 tests/unit/test_liquidity_gate.py create mode 100644 tests/unit/test_sizing_engine.py diff --git a/pyproject.toml b/pyproject.toml index 70d022f..bc67b6e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,10 +86,14 @@ select = [ ignore = [ "PLR0913", # too many arguments (we accept config-heavy functions) "PLR2004", # magic value (we have many domain constants in tests) + "PLR0911", # too many return statements (rule engines have many early returns) + "RUF001", # ambiguous unicode in strings (we use math symbols × ≤ ≥) + "RUF002", # ambiguous unicode in docstrings + "RUF003", # ambiguous unicode in comments ] [tool.ruff.lint.per-file-ignores] -"tests/**" = ["PLR2004", "ARG", "S101"] +"tests/**" = ["PLR2004", "ARG", "S101", "ERA001", "B017"] [tool.ruff.format] quote-style = "double" diff --git a/src/cerbero_bite/config/__init__.py b/src/cerbero_bite/config/__init__.py index e69de29..aa1ca40 100644 --- a/src/cerbero_bite/config/__init__.py +++ b/src/cerbero_bite/config/__init__.py @@ -0,0 +1,43 @@ +"""Strategy configuration: schema, loader, validation.""" + +from cerbero_bite.config.schema import ( + AssetConfig, + DvolAdjustmentBand, + EntryConfig, + ExecutionConfig, + ExitConfig, + KellyConfig, + LiquidityConfig, + McpConfig, + MonitoringConfig, + ShortStrikeSpec, + SizingConfig, + SpreadType, + SpreadWidthSpec, + StorageConfig, + StrategyConfig, + StructureConfig, + TelegramConfig, + golden_config, +) + +__all__ = [ + "AssetConfig", + "DvolAdjustmentBand", + "EntryConfig", + "ExecutionConfig", + "ExitConfig", + "KellyConfig", + "LiquidityConfig", + "McpConfig", + "MonitoringConfig", + "ShortStrikeSpec", + "SizingConfig", + "SpreadType", + "SpreadWidthSpec", + "StorageConfig", + "StrategyConfig", + "StructureConfig", + "TelegramConfig", + "golden_config", +] diff --git a/src/cerbero_bite/config/schema.py b/src/cerbero_bite/config/schema.py new file mode 100644 index 0000000..7a409f5 --- /dev/null +++ b/src/cerbero_bite/config/schema.py @@ -0,0 +1,298 @@ +"""Pydantic schema for ``strategy.yaml``. + +The configuration is the single source of truth for every threshold used +by the :mod:`cerbero_bite.core` algorithms. The schema is intentionally +verbose: every numeric threshold from +``docs/10-config-spec.md`` is exposed as a typed field so that callers +can refer to the same name they would write in YAML. + +All monetary and price-related thresholds use :class:`decimal.Decimal` +to preserve the precision conventions documented in +``docs/03-algorithms.md §0.2`` (no ``float`` for money or option prices). + +Only the fields required by Phase 1 (core algorithms) are populated with +defaults that match the golden config; fields belonging to runtime, +storage, MCP, etc. are typed as ``dict[str, Any]`` or made optional so +that the schema can absorb the full ``strategy.yaml`` without forcing +non-core sections to be implemented before their owning module is +written. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +# --------------------------------------------------------------------------- +# Asset +# --------------------------------------------------------------------------- + + +class AssetConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + symbol: str = "ETH" + exchange: str = "deribit" + + +# --------------------------------------------------------------------------- +# Entry +# --------------------------------------------------------------------------- + + +class EntryConfig(BaseModel): + """Filters and bias thresholds documented in ``01-strategy-rules.md``.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + cron: str = "0 14 * * MON" + skip_holidays_country: str = "IT" + + # access filters (§2) + capital_min_usd: Decimal = Field(default=Decimal("720")) + dvol_min: Decimal = Field(default=Decimal("35")) + dvol_max: Decimal = Field(default=Decimal("90")) + funding_perp_abs_max_annualized: Decimal = Field(default=Decimal("0.80")) + eth_holdings_pct_max: Decimal = Field(default=Decimal("0.30")) + no_position_concurrent: bool = True + exclude_macro_severity: list[str] = Field(default_factory=lambda: ["high"]) + exclude_macro_countries: list[str] = Field(default_factory=lambda: ["US", "EU"]) + + # directional bias (§3.1) + trend_window_days: int = 30 + trend_bull_threshold_pct: Decimal = Field(default=Decimal("0.05")) + trend_bear_threshold_pct: Decimal = Field(default=Decimal("-0.05")) + funding_bull_threshold_annualized: Decimal = Field(default=Decimal("0.20")) + funding_bear_threshold_annualized: Decimal = Field(default=Decimal("-0.20")) + iron_condor_dvol_min: Decimal = Field(default=Decimal("55")) + iron_condor_adx_max: Decimal = Field(default=Decimal("20")) + iron_condor_trend_neutral_band_pct: Decimal = Field(default=Decimal("0.05")) + + +# --------------------------------------------------------------------------- +# Structure +# --------------------------------------------------------------------------- + + +class ShortStrikeSpec(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + delta_target: Decimal = Field(default=Decimal("0.12")) + delta_min: Decimal = Field(default=Decimal("0.10")) + delta_max: Decimal = Field(default=Decimal("0.15")) + distance_otm_pct_min: Decimal = Field(default=Decimal("0.15")) + distance_otm_pct_max: Decimal = Field(default=Decimal("0.25")) + + +class SpreadWidthSpec(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + target_pct_of_spot: Decimal = Field(default=Decimal("0.04")) + min_pct_of_spot: Decimal = Field(default=Decimal("0.03")) + max_pct_of_spot: Decimal = Field(default=Decimal("0.05")) + + +class StructureConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + dte_target: int = 18 + dte_min: int = 14 + dte_max: int = 21 + + short_strike: ShortStrikeSpec = Field(default_factory=ShortStrikeSpec) + spread_width: SpreadWidthSpec = Field(default_factory=SpreadWidthSpec) + credit_to_width_ratio_min: Decimal = Field(default=Decimal("0.30")) + + +# --------------------------------------------------------------------------- +# Liquidity +# --------------------------------------------------------------------------- + + +class LiquidityConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + open_interest_min: int = 100 + volume_24h_min: int = 20 + bid_ask_spread_pct_max: Decimal = Field(default=Decimal("0.15")) + book_depth_top3_min: int = 5 + slippage_pct_of_credit_max: Decimal = Field(default=Decimal("0.08")) + + +# --------------------------------------------------------------------------- +# Sizing +# --------------------------------------------------------------------------- + + +class DvolAdjustmentBand(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + dvol_under: Decimal + multiplier: Decimal + + +def _default_dvol_bands() -> list[DvolAdjustmentBand]: + return [ + DvolAdjustmentBand(dvol_under=Decimal("45"), multiplier=Decimal("1.00")), + DvolAdjustmentBand(dvol_under=Decimal("60"), multiplier=Decimal("0.85")), + DvolAdjustmentBand(dvol_under=Decimal("80"), multiplier=Decimal("0.65")), + ] + + +class SizingConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + kelly_fraction: Decimal = Field(default=Decimal("0.13")) + + cap_per_trade_eur: Decimal = Field(default=Decimal("200")) + cap_aggregate_open_eur: Decimal = Field(default=Decimal("1000")) + max_concurrent_positions: int = 1 + max_contracts_per_trade: int = 4 + + dvol_adjustment: list[DvolAdjustmentBand] = Field(default_factory=_default_dvol_bands) + dvol_no_entry_threshold: Decimal = Field(default=Decimal("80")) + + +# --------------------------------------------------------------------------- +# Exit +# --------------------------------------------------------------------------- + + +class ExitConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + profit_take_pct_of_credit: Decimal = Field(default=Decimal("0.50")) + stop_loss_mark_x_credit: Decimal = Field(default=Decimal("2.50")) + vol_stop_dvol_increase: Decimal = Field(default=Decimal("10")) + time_stop_dte_remaining: int = 7 + time_stop_skip_if_close_to_profit_pct: Decimal = Field(default=Decimal("0.70")) + delta_breach_threshold: Decimal = Field(default=Decimal("0.30")) + adverse_move_4h_pct: Decimal = Field(default=Decimal("0.05")) + + monitor_cron: str = "0 2,14 * * *" + user_confirmation_timeout_min: int = 30 + escalate_on_timeout: list[str] = Field( + default_factory=lambda: ["CLOSE_STOP", "CLOSE_VOL", "CLOSE_DELTA"] + ) + + +# --------------------------------------------------------------------------- +# Kelly recalibration +# --------------------------------------------------------------------------- + + +class KellyConfig(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + lookback_days: int = 365 + min_sample_low_confidence: int = 30 + min_sample_high_confidence: int = 100 + weight_when_medium_confidence: Decimal = Field(default=Decimal("0.50")) + + +# --------------------------------------------------------------------------- +# Sections that core/ algorithms do not consume — kept loose for now +# --------------------------------------------------------------------------- + + +class _LooseSection(BaseModel): + """Catch-all section used for parts of strategy.yaml not yet typed.""" + + model_config = ConfigDict(frozen=True, extra="allow") + + +class ExecutionConfig(_LooseSection): ... + + +class MonitoringConfig(_LooseSection): ... + + +class StorageConfig(_LooseSection): ... + + +class McpConfig(_LooseSection): ... + + +class TelegramConfig(_LooseSection): ... + + +# --------------------------------------------------------------------------- +# Root +# --------------------------------------------------------------------------- + + +class StrategyConfig(BaseModel): + """Validated representation of ``strategy.yaml``.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + config_version: str + config_hash: str + last_review: str + last_reviewer: str + + asset: AssetConfig = Field(default_factory=AssetConfig) + entry: EntryConfig = Field(default_factory=EntryConfig) + structure: StructureConfig = Field(default_factory=StructureConfig) + liquidity: LiquidityConfig = Field(default_factory=LiquidityConfig) + sizing: SizingConfig = Field(default_factory=SizingConfig) + exit: ExitConfig = Field(default_factory=ExitConfig) + kelly_recalibration: KellyConfig = Field(default_factory=KellyConfig) + + execution: ExecutionConfig = Field(default_factory=ExecutionConfig) + monitoring: MonitoringConfig = Field(default_factory=MonitoringConfig) + storage: StorageConfig = Field(default_factory=StorageConfig) + mcp: McpConfig = Field(default_factory=McpConfig) + telegram: TelegramConfig = Field(default_factory=TelegramConfig) + + @model_validator(mode="after") + def _validate_consistency(self) -> StrategyConfig: + s = self.structure.short_strike + if not (s.delta_min <= s.delta_target <= s.delta_max): + raise ValueError("delta_target outside [delta_min, delta_max]") + if s.distance_otm_pct_min >= s.distance_otm_pct_max: + raise ValueError("OTM range invalid") + + sw = self.structure.spread_width + if not (sw.min_pct_of_spot <= sw.target_pct_of_spot <= sw.max_pct_of_spot): + raise ValueError("spread_width target outside min/max") + + if not (Decimal("0") < self.sizing.kelly_fraction < Decimal("0.5")): + raise ValueError("kelly_fraction out of safe range (0, 0.5)") + + if self.exit.profit_take_pct_of_credit >= Decimal("1"): + raise ValueError("profit_take >= 100% impossible") + if self.exit.stop_loss_mark_x_credit <= Decimal("1"): + raise ValueError("stop_loss multiple <= 1× makes no sense") + + if self.entry.dvol_min >= self.entry.dvol_max: + raise ValueError("dvol_min must be < dvol_max") + + return self + + +# --------------------------------------------------------------------------- +# Convenience +# --------------------------------------------------------------------------- + + +SpreadType = Literal["bull_put", "bear_call", "iron_condor"] + + +def golden_config(**overrides: Any) -> StrategyConfig: + """Return a :class:`StrategyConfig` populated with the golden defaults. + + The caller may pass per-section overrides (``entry={...}``, + ``sizing={...}``, …) to mutate specific values without rebuilding the + whole tree. Useful in tests. + """ + base: dict[str, Any] = { + "config_version": "1.0.0-test", + "config_hash": "0" * 64, + "last_review": "2026-04-26", + "last_reviewer": "test", + } + base.update(overrides) + return StrategyConfig(**base) diff --git a/src/cerbero_bite/core/__init__.py b/src/cerbero_bite/core/__init__.py index e69de29..85fb493 100644 --- a/src/cerbero_bite/core/__init__.py +++ b/src/cerbero_bite/core/__init__.py @@ -0,0 +1,7 @@ +"""Pure rule-engine algorithms (no I/O, no LLM, no clock). + +Every module here is deterministic: same input → same output. Inputs are +typed via Pydantic models, outputs likewise. The orchestrator in +:mod:`cerbero_bite.runtime` is the only layer allowed to bring in real +data and call these functions. +""" diff --git a/src/cerbero_bite/core/combo_builder.py b/src/cerbero_bite/core/combo_builder.py new file mode 100644 index 0000000..260eb86 --- /dev/null +++ b/src/cerbero_bite/core/combo_builder.py @@ -0,0 +1,253 @@ +"""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 _select_short( + quotes: list[OptionQuote], + *, + spot: Decimal, + cfg: StrategyConfig, +) -> OptionQuote | None: + """Pick the short-leg quote with delta closest to target inside both bands.""" + sc = cfg.structure.short_strike + 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 (sc.delta_min <= abs_delta <= sc.delta_max): + continue + eligible.append(q) + if not eligible: + return None + return min(eligible, key=lambda q: abs(q.delta.copy_abs() - sc.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, +) -> 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) + 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, + ) diff --git a/src/cerbero_bite/core/entry_validator.py b/src/cerbero_bite/core/entry_validator.py new file mode 100644 index 0000000..2ccc263 --- /dev/null +++ b/src/cerbero_bite/core/entry_validator.py @@ -0,0 +1,155 @@ +"""Pure functions for the *entry* phase of the decision loop. + +Two responsibilities (both deterministic, no I/O): + +* :func:`validate_entry` — accumulate all blocking reasons that come from + the access filters in ``docs/01-strategy-rules.md §2``. +* :func:`compute_bias` — translate market trend/funding/regime into a + :data:`SpreadType` choice per ``docs/01-strategy-rules.md §3.1``. +""" + +from __future__ import annotations + +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config import SpreadType, StrategyConfig + +__all__ = [ + "EntryContext", + "EntryDecision", + "TrendContext", + "compute_bias", + "validate_entry", +] + + +class EntryContext(BaseModel): + """Snapshot of the inputs needed to decide whether to open a trade.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + capital_usd: Decimal + dvol_now: Decimal + funding_perp_annualized: Decimal + eth_holdings_pct_of_portfolio: Decimal + next_macro_event_in_days: int | None + has_open_position: bool + + +class EntryDecision(BaseModel): + """Result of :func:`validate_entry`. ``reasons`` holds *all* blocking reasons.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + accepted: bool + reasons: list[str] + + +class TrendContext(BaseModel): + """Market regime inputs for :func:`compute_bias`.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + eth_now: Decimal + eth_30d_ago: Decimal + funding_cross_annualized: Decimal + dvol_now: Decimal + adx_14: Decimal + + +def validate_entry(ctx: EntryContext, cfg: StrategyConfig) -> EntryDecision: + """Return the entry decision, collecting *every* failing condition. + + Order matches the documentation but does not short-circuit: callers + receive the full list of reasons so the report message can explain + why a trade was skipped without needing multiple passes. + """ + reasons: list[str] = [] + entry_cfg = cfg.entry + structure_cfg = cfg.structure + + if ctx.has_open_position: + reasons.append("open position already exists") + + if ctx.capital_usd < entry_cfg.capital_min_usd: + reasons.append( + f"capital below minimum ({ctx.capital_usd} < {entry_cfg.capital_min_usd})" + ) + + if ctx.dvol_now < entry_cfg.dvol_min: + reasons.append(f"dvol too low ({ctx.dvol_now} < {entry_cfg.dvol_min})") + elif ctx.dvol_now > entry_cfg.dvol_max: + reasons.append(f"dvol too high ({ctx.dvol_now} > {entry_cfg.dvol_max})") + + if ( + ctx.next_macro_event_in_days is not None + and ctx.next_macro_event_in_days <= structure_cfg.dte_target + ): + reasons.append( + f"macro event within DTE window ({ctx.next_macro_event_in_days} days)" + ) + + if abs(ctx.funding_perp_annualized) > entry_cfg.funding_perp_abs_max_annualized: + reasons.append( + f"funding rate beyond cap ({ctx.funding_perp_annualized} vs " + f"±{entry_cfg.funding_perp_abs_max_annualized})" + ) + + if ctx.eth_holdings_pct_of_portfolio > entry_cfg.eth_holdings_pct_max: + reasons.append( + f"eth holdings above cap ({ctx.eth_holdings_pct_of_portfolio} > " + f"{entry_cfg.eth_holdings_pct_max})" + ) + + return EntryDecision(accepted=not reasons, reasons=reasons) + + +def _trend_signal( + pct_change: Decimal, + bull_threshold: Decimal, + bear_threshold: Decimal, +) -> str: + """Classify a percentage change into ``bull`` / ``bear`` / ``neutral``.""" + if pct_change >= bull_threshold: + return "bull" + if pct_change <= bear_threshold: + return "bear" + return "neutral" + + +def compute_bias(ctx: TrendContext, cfg: StrategyConfig) -> SpreadType | None: + """Return the spread type to attempt, or ``None`` to skip the week.""" + entry_cfg = cfg.entry + + if ctx.eth_30d_ago <= 0: + # Invalid market history: refuse to opine on bias. + return None + trend_pct = ctx.eth_now / ctx.eth_30d_ago - Decimal("1") + trend = _trend_signal( + trend_pct, + entry_cfg.trend_bull_threshold_pct, + entry_cfg.trend_bear_threshold_pct, + ) + + funding = _trend_signal( + ctx.funding_cross_annualized, + entry_cfg.funding_bull_threshold_annualized, + entry_cfg.funding_bear_threshold_annualized, + ) + + if trend == "bull" and funding == "bull": + return "bull_put" + if trend == "bear" and funding == "bear": + return "bear_call" + if trend == "neutral" and funding == "neutral": + if ( + ctx.dvol_now >= entry_cfg.iron_condor_dvol_min + and ctx.adx_14 < entry_cfg.iron_condor_adx_max + ): + return "iron_condor" + return None + # All remaining combinations (discordant or one-neutral-one-directional) + # explicitly fall through to "no entry". + return None diff --git a/src/cerbero_bite/core/exit_decision.py b/src/cerbero_bite/core/exit_decision.py new file mode 100644 index 0000000..3e8677f --- /dev/null +++ b/src/cerbero_bite/core/exit_decision.py @@ -0,0 +1,158 @@ +"""Exit-decision rule engine (``docs/03-algorithms.md §6``). + +Pure function over a :class:`PositionSnapshot`. Triggers are evaluated +in the documented order; the first match wins. ``HOLD`` is returned +when no rule fires. The time-stop ``skip_if_close_to_profit`` exception +is interpreted as ``mark ≤ 70% × credit_received`` (re-using the same +units as the profit gate), which is the only interpretation that yields +non-trivial behaviour: rule 1 takes everything below 50% credit; the +exception covers the (50%, 70%] credit band where we wait for rule 1 +to fire next cycle. +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config import SpreadType, StrategyConfig +from cerbero_bite.core.types import OptionLeg + +__all__ = ["ExitAction", "ExitDecisionResult", "PositionSnapshot", "evaluate"] + + +ExitAction = Literal[ + "HOLD", + "CLOSE_PROFIT", + "CLOSE_STOP", + "CLOSE_VOL", + "CLOSE_TIME", + "CLOSE_DELTA", + "CLOSE_AVERSE", +] + + +class PositionSnapshot(BaseModel): + """Inputs for :func:`evaluate`. All ETH amounts are *totals* for the position.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + proposal_id: UUID + spread_type: SpreadType + legs: list[OptionLeg] + + credit_received_eth: Decimal + credit_received_usd: Decimal + spot_at_entry: Decimal + dvol_at_entry: Decimal + expiry: datetime + opened_at: datetime + + eth_price_usd_now: Decimal + spot_now: Decimal + dvol_now: Decimal + mark_combo_now_eth: Decimal + delta_short_now: Decimal + return_4h_now: Decimal + now: datetime + + +class ExitDecisionResult(BaseModel): + """Action + human-readable reason + PnL estimate at the moment of decision.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + action: ExitAction + reason: str + pnl_estimate_eth: Decimal + pnl_estimate_usd: Decimal + + +def _days_to_expiry(snapshot: PositionSnapshot) -> Decimal: + delta = snapshot.expiry - snapshot.now + return Decimal(str(delta.total_seconds())) / Decimal("86400") + + +def _adverse_move(spread_type: SpreadType, return_4h: Decimal, threshold: Decimal) -> bool: + if spread_type == "bull_put": + return return_4h <= -threshold + if spread_type == "bear_call": + return return_4h >= threshold + # iron_condor: adverse on either side + return return_4h.copy_abs() >= threshold + + +def evaluate(snapshot: PositionSnapshot, cfg: StrategyConfig) -> ExitDecisionResult: + """Return the exit action for the given position snapshot.""" + ec = cfg.exit + credit = snapshot.credit_received_eth + debit = snapshot.mark_combo_now_eth + + pnl_eth = credit - debit + pnl_usd = pnl_eth * snapshot.eth_price_usd_now + + profit_take_thresh = credit * ec.profit_take_pct_of_credit + stop_thresh = credit * ec.stop_loss_mark_x_credit + skip_time_thresh = credit * ec.time_stop_skip_if_close_to_profit_pct + days_left = _days_to_expiry(snapshot) + + def _result(action: ExitAction, reason: str) -> ExitDecisionResult: + return ExitDecisionResult( + action=action, + reason=reason, + pnl_estimate_eth=pnl_eth, + pnl_estimate_usd=pnl_usd, + ) + + # 1. Profit take + if debit <= profit_take_thresh: + return _result( + "CLOSE_PROFIT", + f"mark {debit} ≤ {ec.profit_take_pct_of_credit:.0%} of credit {credit}", + ) + + # 2. Stop loss + if debit >= stop_thresh: + return _result( + "CLOSE_STOP", + f"mark {debit} ≥ {ec.stop_loss_mark_x_credit}× credit {credit}", + ) + + # 3. Vol stop + if snapshot.dvol_now >= snapshot.dvol_at_entry + ec.vol_stop_dvol_increase: + return _result( + "CLOSE_VOL", + f"DVOL {snapshot.dvol_now} ≥ entry {snapshot.dvol_at_entry} " + f"+ {ec.vol_stop_dvol_increase}", + ) + + # 4. Time stop with "close to profit" exception + if days_left <= ec.time_stop_dte_remaining and debit > skip_time_thresh: + return _result( + "CLOSE_TIME", + f"DTE {days_left:.2f} ≤ {ec.time_stop_dte_remaining} and mark " + f"{debit} above skip threshold {skip_time_thresh}", + ) + # When DTE ≤ 7 but mark is in the (50%, 70%]-credit "close to profit" + # zone, we deliberately fall through; rule 1 will fire next cycle. + + # 5. Strike tested + if snapshot.delta_short_now.copy_abs() >= ec.delta_breach_threshold: + return _result( + "CLOSE_DELTA", + f"|delta_short| {snapshot.delta_short_now.copy_abs()} ≥ " + f"{ec.delta_breach_threshold}", + ) + + # 6. Explosive adverse move + if _adverse_move(snapshot.spread_type, snapshot.return_4h_now, ec.adverse_move_4h_pct): + return _result( + "CLOSE_AVERSE", + f"4h return {snapshot.return_4h_now} adverse for {snapshot.spread_type}", + ) + + return _result("HOLD", "all triggers within tolerance") diff --git a/src/cerbero_bite/core/greeks_aggregator.py b/src/cerbero_bite/core/greeks_aggregator.py new file mode 100644 index 0000000..fe728f4 --- /dev/null +++ b/src/cerbero_bite/core/greeks_aggregator.py @@ -0,0 +1,63 @@ +"""Aggregate greeks for a multi-leg position (``docs/03-algorithms.md §5``). + +Pure summation with the BUY/SELL sign and the leg size. Theta is +converted to USD/day; the other greeks are returned in the same units +they came in (per-contract, dimensionless or fractional). +""" + +from __future__ import annotations + +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.core.types import OptionLeg + +__all__ = ["AggregateGreeks", "aggregate"] + + +class AggregateGreeks(BaseModel): + """Net greeks for an option spread.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + delta_net: Decimal + gamma_net: Decimal + theta_net: Decimal # USD per day + vega_net: Decimal + + +def _sign(side: str) -> Decimal: + return Decimal("1") if side == "BUY" else Decimal("-1") + + +def aggregate( + *, + legs: list[OptionLeg], + eth_price_usd: Decimal, +) -> AggregateGreeks: + """Return :class:`AggregateGreeks` summed with side and size weights. + + ``theta`` is multiplied by ``eth_price_usd`` to convert from + ETH/day (Deribit native) to USD/day, which is what the report + consumes. + """ + delta = Decimal("0") + gamma = Decimal("0") + theta_eth = Decimal("0") + vega = Decimal("0") + + for leg in legs: + sign = _sign(leg.side) + weight = sign * Decimal(leg.size) + delta += weight * leg.delta + gamma += weight * leg.gamma + theta_eth += weight * leg.theta + vega += weight * leg.vega + + return AggregateGreeks( + delta_net=delta, + gamma_net=gamma, + theta_net=theta_eth * eth_price_usd, + vega_net=vega, + ) diff --git a/src/cerbero_bite/core/kelly_recalibration.py b/src/cerbero_bite/core/kelly_recalibration.py new file mode 100644 index 0000000..c104069 --- /dev/null +++ b/src/cerbero_bite/core/kelly_recalibration.py @@ -0,0 +1,152 @@ +"""Monthly Kelly recalibration (``docs/03-algorithms.md §7``). + +Pure function over a list of closed-trade records. Returns a fresh +:class:`KellyResult` summarising the empirical edge and a recommendation +that is *never* applied automatically — the orchestrator surfaces it as +a monthly report and Adriano decides whether to bump +``sizing.kelly_fraction`` in ``strategy.yaml``. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config import StrategyConfig + +__all__ = ["Confidence", "KellyResult", "TradeRecord", "recalibrate"] + + +Confidence = Literal["low", "medium", "high"] + + +class TradeRecord(BaseModel): + """One closed trade fed to the recalibration.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + proposal_id: UUID + pnl_usd: Decimal + risk_usd: Decimal + closed_at: datetime + outcome: str + + +class KellyResult(BaseModel): + """Recommendation report; the operator may or may not apply it.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + win_rate: Decimal + avg_win_pct_risk: Decimal + avg_loss_pct_risk: Decimal + full_kelly_pct: Decimal + quarter_kelly_pct: Decimal + sample_size: int + recommended_fraction: Decimal + confidence: Confidence + + +def _within_lookback( + trades: list[TradeRecord], + *, + now: datetime, + lookback_days: int, +) -> list[TradeRecord]: + cutoff = now - timedelta(days=lookback_days) + return [t for t in trades if t.closed_at >= cutoff] + + +def _confidence_for( + n: int, + *, + low_threshold: int, + high_threshold: int, +) -> Confidence: + if n < low_threshold: + return "low" + if n < high_threshold: + return "medium" + return "high" + + +def recalibrate( + *, + trades: list[TradeRecord], + now: datetime, + cfg: StrategyConfig, +) -> KellyResult: + """Return the empirical Kelly summary + recommended fraction.""" + krc = cfg.kelly_recalibration + in_window = _within_lookback(trades, now=now, lookback_days=krc.lookback_days) + n = len(in_window) + + if n == 0: + return KellyResult( + win_rate=Decimal("0"), + avg_win_pct_risk=Decimal("0"), + avg_loss_pct_risk=Decimal("0"), + full_kelly_pct=Decimal("0"), + quarter_kelly_pct=Decimal("0"), + sample_size=0, + recommended_fraction=cfg.sizing.kelly_fraction, + confidence="low", + ) + + wins = [t for t in in_window if t.pnl_usd > 0] + losses = [t for t in in_window if t.pnl_usd < 0] + win_rate = Decimal(len(wins)) / Decimal(n) + + avg_win = ( + sum((t.pnl_usd / t.risk_usd for t in wins), Decimal("0")) / Decimal(len(wins)) + if wins + else Decimal("0") + ) + avg_loss = ( + sum((-t.pnl_usd / t.risk_usd for t in losses), Decimal("0")) / Decimal(len(losses)) + if losses + else Decimal("0") + ) + + if avg_loss == 0: + # No losses — fall back to win_rate as an upper bound on full Kelly. + full_kelly = win_rate + else: + b = avg_win / avg_loss + if b == 0: + full_kelly = Decimal("0") + else: + full_kelly = (win_rate * b - (Decimal("1") - win_rate)) / b + if full_kelly < 0: + full_kelly = Decimal("0") + + quarter_kelly = full_kelly * Decimal("0.25") + + confidence = _confidence_for( + n, + low_threshold=krc.min_sample_low_confidence, + high_threshold=krc.min_sample_high_confidence, + ) + + if confidence == "low": + recommended = cfg.sizing.kelly_fraction + elif confidence == "medium": + weight = krc.weight_when_medium_confidence + recommended = weight * quarter_kelly + (Decimal("1") - weight) * cfg.sizing.kelly_fraction + else: + recommended = quarter_kelly + + return KellyResult( + win_rate=win_rate, + avg_win_pct_risk=avg_win, + avg_loss_pct_risk=avg_loss, + full_kelly_pct=full_kelly, + quarter_kelly_pct=quarter_kelly, + sample_size=n, + recommended_fraction=recommended, + confidence=confidence, + ) diff --git a/src/cerbero_bite/core/liquidity_gate.py b/src/cerbero_bite/core/liquidity_gate.py new file mode 100644 index 0000000..1904c70 --- /dev/null +++ b/src/cerbero_bite/core/liquidity_gate.py @@ -0,0 +1,135 @@ +"""Pre-trade liquidity gate (``docs/03-algorithms.md §2``). + +Validates each leg against open-interest, volume, bid-ask, depth +thresholds and computes the expected slippage as a percentage of the +credit. Returns the full list of failing reasons together with the +slippage estimate so the report can render a precise rejection. +""" + +from __future__ import annotations + +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config import StrategyConfig + +__all__ = ["InstrumentSnapshot", "LiquidityCheck", "check"] + + +class InstrumentSnapshot(BaseModel): + """Per-instrument market snapshot used by the liquidity check.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + instrument: str + bid: Decimal + ask: Decimal + mid: Decimal + open_interest: int + volume_24h: int + book_depth_top3: int + + +class LiquidityCheck(BaseModel): + """Outcome of :func:`check`: accepted flag, full reasons, slippage estimate.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + accepted: bool + reasons: list[str] + estimated_slippage_pct_of_credit: Decimal + + +def _check_leg( + snap: InstrumentSnapshot, + *, + label: str, + cfg: StrategyConfig, +) -> list[str]: + """Return the list of failing reasons for a single leg.""" + reasons: list[str] = [] + liq = cfg.liquidity + + if snap.open_interest < liq.open_interest_min: + reasons.append( + f"{label} open interest below threshold " + f"({snap.open_interest} < {liq.open_interest_min})" + ) + if snap.volume_24h < liq.volume_24h_min: + reasons.append( + f"{label} 24h volume below threshold " + f"({snap.volume_24h} < {liq.volume_24h_min})" + ) + if snap.book_depth_top3 < liq.book_depth_top3_min: + reasons.append( + f"{label} book depth top3 below threshold " + f"({snap.book_depth_top3} < {liq.book_depth_top3_min})" + ) + + if snap.mid > 0: + spread_pct = (snap.ask - snap.bid) / snap.mid + if spread_pct > liq.bid_ask_spread_pct_max: + reasons.append( + f"{label} bid-ask spread {spread_pct:.4f} above cap " + f"{liq.bid_ask_spread_pct_max}" + ) + else: + reasons.append(f"{label} mid price not positive ({snap.mid})") + + return reasons + + +def check( + *, + short_leg: InstrumentSnapshot, + long_leg: InstrumentSnapshot, + credit: Decimal, + n_contracts: int, + cfg: StrategyConfig, +) -> LiquidityCheck: + """Validate the liquidity of a 2-leg vertical spread. + + Args: + short_leg: the leg the engine intends to *sell*. + long_leg: the leg the engine intends to *buy* as protection. + credit: net mid-price credit of the combo, in ETH. + n_contracts: number of combo contracts that will be sent. + cfg: validated strategy configuration. + + Returns: + :class:`LiquidityCheck` with ``accepted`` set when every leg + passes its thresholds *and* the estimated slippage is within + the configured cap. + """ + reasons: list[str] = _check_leg(short_leg, label="short leg", cfg=cfg) + reasons.extend(_check_leg(long_leg, label="long leg", cfg=cfg)) + + if n_contracts <= 0: + reasons.append(f"non-positive number of contracts ({n_contracts})") + + if credit <= 0: + reasons.append(f"non-positive credit ({credit})") + # Cannot compute slippage_pct without a positive credit. + return LiquidityCheck( + accepted=False, + reasons=reasons, + estimated_slippage_pct_of_credit=Decimal("0"), + ) + + slippage_eth_per_contract = (short_leg.ask - short_leg.mid) + (long_leg.mid - long_leg.bid) + n_for_slippage = max(n_contracts, 0) + slippage_total_eth = slippage_eth_per_contract * Decimal(n_for_slippage) + pct = slippage_total_eth / credit + + if pct > cfg.liquidity.slippage_pct_of_credit_max: + reasons.append( + f"estimated slippage {pct:.4f} above cap " + f"{cfg.liquidity.slippage_pct_of_credit_max}" + ) + + return LiquidityCheck( + accepted=not reasons, + reasons=reasons, + estimated_slippage_pct_of_credit=pct, + ) diff --git a/src/cerbero_bite/core/sizing_engine.py b/src/cerbero_bite/core/sizing_engine.py new file mode 100644 index 0000000..4cc6239 --- /dev/null +++ b/src/cerbero_bite/core/sizing_engine.py @@ -0,0 +1,92 @@ +"""Quarter-Kelly sizing with caps (``docs/03-algorithms.md §3``). + +Pure function: given capital, max-loss-per-contract, DVOL and the +aggregate state, returns the number of contracts to send. Returns +``n_contracts == 0`` together with a human-readable ``reason_if_zero`` +when no entry is allowed; the orchestrator never opens a position when +``n_contracts == 0``. +""" + +from __future__ import annotations + +from decimal import Decimal + +from pydantic import BaseModel, ConfigDict + +from cerbero_bite.config import StrategyConfig + +__all__ = ["SizingContext", "SizingResult", "compute_contracts"] + + +class SizingContext(BaseModel): + """Snapshot of the inputs needed to size a single trade.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + capital_usd: Decimal + max_loss_per_contract_usd: Decimal + dvol_now: Decimal + open_engagement_usd: Decimal + eur_to_usd: Decimal + other_open_positions: int + + +class SizingResult(BaseModel): + """Result of :func:`compute_contracts`. ``reason_if_zero`` is set iff n=0.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + n_contracts: int + risk_dollars: Decimal + reason_if_zero: str | None + + +def _zero(reason: str) -> SizingResult: + return SizingResult(n_contracts=0, risk_dollars=Decimal("0"), reason_if_zero=reason) + + +def _dvol_multiplier(dvol_now: Decimal, cfg: StrategyConfig) -> Decimal | None: + """Return the size multiplier for the current DVOL, or ``None`` if no entry.""" + for band in cfg.sizing.dvol_adjustment: + if dvol_now < band.dvol_under: + return band.multiplier + return None + + +def compute_contracts(ctx: SizingContext, cfg: StrategyConfig) -> SizingResult: + """Return the contract count after Kelly, caps, DVOL and engagement checks.""" + if ctx.max_loss_per_contract_usd <= 0: + return _zero(f"non-positive max_loss_per_contract ({ctx.max_loss_per_contract_usd})") + + risk_target = ctx.capital_usd * cfg.sizing.kelly_fraction + cap_per_trade_usd = cfg.sizing.cap_per_trade_eur * ctx.eur_to_usd + risk_target = min(risk_target, cap_per_trade_usd) + + multiplier = _dvol_multiplier(ctx.dvol_now, cfg) + if multiplier is None: + return _zero(f"dvol {ctx.dvol_now} above no-entry threshold") + risk_target *= multiplier + + n = int(risk_target // ctx.max_loss_per_contract_usd) + n = min(n, cfg.sizing.max_contracts_per_trade) + + cap_aggregate_usd = cfg.sizing.cap_aggregate_open_eur * ctx.eur_to_usd + while n > 0 and ( + Decimal(n) * ctx.max_loss_per_contract_usd + ctx.open_engagement_usd + ) > cap_aggregate_usd: + n -= 1 + + if ctx.other_open_positions >= cfg.sizing.max_concurrent_positions: + return _zero( + f"already {ctx.other_open_positions} concurrent position(s); " + f"cap is {cfg.sizing.max_concurrent_positions}" + ) + + if n < 1: + return _zero("undersize after caps") + + return SizingResult( + n_contracts=n, + risk_dollars=Decimal(n) * ctx.max_loss_per_contract_usd, + reason_if_zero=None, + ) diff --git a/src/cerbero_bite/core/types.py b/src/cerbero_bite/core/types.py new file mode 100644 index 0000000..3afb0a5 --- /dev/null +++ b/src/cerbero_bite/core/types.py @@ -0,0 +1,72 @@ +"""Shared option-data types used across multiple algorithms. + +Kept in one place so that :mod:`liquidity_gate`, :mod:`combo_builder`, +:mod:`greeks_aggregator` and :mod:`exit_decision` agree on the same +record format. The orchestrator is responsible for assembling +:class:`OptionQuote` instances from the raw MCP responses. +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict + +__all__ = ["OptionLeg", "OptionQuote", "OrderSide", "PutOrCall"] + + +PutOrCall = Literal["P", "C"] +OrderSide = Literal["BUY", "SELL"] + + +class OptionQuote(BaseModel): + """Full market + greeks snapshot for a single option instrument. + + Used by combo selection (greeks + strike + expiry) and by the + liquidity gate (bid/ask/depth). Time information is in UTC. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + instrument: str + strike: Decimal + expiry: datetime + option_type: PutOrCall + + bid: Decimal + ask: Decimal + mid: Decimal + + delta: Decimal + gamma: Decimal + theta: Decimal + vega: Decimal + + open_interest: int + volume_24h: int + book_depth_top3: int + + +class OptionLeg(BaseModel): + """Single leg of an option spread, ready for execution. + + The signed greeks are stored leg-level (not net): aggregation is done + by :func:`cerbero_bite.core.greeks_aggregator.aggregate`. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + instrument: str + side: OrderSide + strike: Decimal + expiry: datetime + type: PutOrCall + size: int + + mid_price_eth: Decimal + delta: Decimal + gamma: Decimal + theta: Decimal + vega: Decimal diff --git a/tests/unit/test_combo_builder.py b/tests/unit/test_combo_builder.py new file mode 100644 index 0000000..01433d9 --- /dev/null +++ b/tests/unit/test_combo_builder.py @@ -0,0 +1,331 @@ +"""TDD for :mod:`cerbero_bite.core.combo_builder`. + +Spec: ``docs/03-algorithms.md §4`` and ``docs/01-strategy-rules.md §3``. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from uuid import UUID + +import pytest + +from cerbero_bite.config import StrategyConfig, golden_config +from cerbero_bite.core.combo_builder import build, select_strikes +from cerbero_bite.core.types import OptionQuote + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +@pytest.fixture +def now() -> datetime: + return datetime(2026, 4, 27, 14, 0, tzinfo=UTC) + + +def _quote( + *, + strike: str, + delta: str, + option_type: str = "P", + expiry_dte: int = 18, + mid: str | None = None, + bid: str | None = None, + ask: str | None = None, + open_interest: int = 500, + volume_24h: int = 100, + book_depth_top3: int = 30, + now_dt: datetime = datetime(2026, 4, 27, 14, 0, tzinfo=UTC), +) -> OptionQuote: + expiry = (now_dt + timedelta(days=expiry_dte)).replace(hour=8, minute=0, second=0) + if mid is None: + # crude pricing: deeper OTM → smaller premium; absolute delta proxy + mid = str((Decimal(delta).copy_abs() * Decimal("0.10")).quantize(Decimal("0.0001"))) + mid_d = Decimal(mid) + if bid is None: + bid = str((mid_d * Decimal("0.98")).quantize(Decimal("0.000001"))) + if ask is None: + ask = str((mid_d * Decimal("1.02")).quantize(Decimal("0.000001"))) + return OptionQuote( + instrument=f"ETH-{expiry.strftime('%-d%b%y').upper()}-{strike}-{option_type}", + strike=Decimal(strike), + expiry=expiry, + option_type=option_type, # type: ignore[arg-type] + bid=Decimal(bid), + ask=Decimal(ask), + mid=mid_d, + delta=Decimal(delta), + gamma=Decimal("0.001"), + theta=Decimal("-0.0005"), + vega=Decimal("0.10"), + open_interest=open_interest, + volume_24h=volume_24h, + book_depth_top3=book_depth_top3, + ) + + +# --------------------------------------------------------------------------- +# select_strikes +# --------------------------------------------------------------------------- + + +def _bull_put_chain(now_dt: datetime) -> list[OptionQuote]: + """Realistic bull-put chain around spot 3000. + + Mids are tuned so that the candidate (2475 short, 2350 long) yields + credit_usd / width_usd ≈ 36% — comfortably above the 30% gate. + """ + # Distance OTM in [15%, 25%] of 3000 → strikes in [2250, 2550]. + return [ + # Way too close (delta too high) — should be ignored. + _quote(strike="2700", delta="-0.30", mid="0.060", now_dt=now_dt), + # In OTM range, delta in [0.10, 0.15] + _quote(strike="2550", delta="-0.14", mid="0.022", now_dt=now_dt), + _quote(strike="2475", delta="-0.12", mid="0.020", now_dt=now_dt), # closest to 0.12 + _quote(strike="2400", delta="-0.10", mid="0.014", now_dt=now_dt), + # Below OTM range (too far) — delta too small even if in range + _quote(strike="2250", delta="-0.06", mid="0.006", now_dt=now_dt), + # The long strike candidates (further OTM, ~4% width below short) + _quote(strike="2370", delta="-0.09", mid="0.008", now_dt=now_dt), + _quote(strike="2350", delta="-0.08", mid="0.005", now_dt=now_dt), + ] + + +def test_select_strikes_picks_short_closest_to_delta_target( + cfg: StrategyConfig, now: datetime +) -> None: + chain = _bull_put_chain(now) + result = select_strikes( + chain=chain, + bias="bull_put", + spot=Decimal("3000"), + now=now, + cfg=cfg, + ) + assert result is not None + short, long_ = result + assert short.strike == Decimal("2475") # |delta|=0.12 closest to 0.12 + # width target = 4% × 3000 = 120 → long_strike target 2355; nearest at 2350 + assert long_.strike == Decimal("2350") + + +def test_select_strikes_returns_none_when_no_strike_in_otm_range( + cfg: StrategyConfig, now: datetime +) -> None: + # All strikes too close (< 15% OTM). + chain = [ + _quote(strike="2900", delta="-0.30", now_dt=now), + _quote(strike="2800", delta="-0.20", now_dt=now), + ] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_returns_none_when_delta_out_of_band( + cfg: StrategyConfig, now: datetime +) -> None: + # OTM range OK but delta way off [0.10, 0.15]. + chain = [ + _quote(strike="2475", delta="-0.05", now_dt=now), + _quote(strike="2400", delta="-0.04", now_dt=now), + ] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_returns_none_when_long_width_outside_range( + cfg: StrategyConfig, now: datetime +) -> None: + # Short OK; long candidates are all too close (< 3%) or too far (> 5%). + short = _quote(strike="2475", delta="-0.12", mid="0.020", now_dt=now) + too_close = _quote(strike="2470", delta="-0.115", mid="0.018", now_dt=now) # 0.2% width + too_far = _quote(strike="2200", delta="-0.05", mid="0.005", now_dt=now) # 9.17% width + chain = [short, too_close, too_far] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_returns_none_when_credit_to_width_below_min( + cfg: StrategyConfig, now: datetime +) -> None: + # Width 125 USD; credit (mid_short - mid_long) only 0.0001 ETH ≈ 0.3 USD → 0.24%. + short = _quote(strike="2475", delta="-0.12", mid="0.012", now_dt=now) + long_ = _quote(strike="2350", delta="-0.08", mid="0.0119", now_dt=now) + chain = [short, long_] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_filters_out_options_outside_dte_window( + cfg: StrategyConfig, now: datetime +) -> None: + chain = [ + _quote(strike="2475", delta="-0.12", mid="0.012", expiry_dte=5, now_dt=now), + _quote(strike="2350", delta="-0.08", mid="0.007", expiry_dte=5, now_dt=now), + _quote(strike="2475", delta="-0.12", mid="0.012", expiry_dte=40, now_dt=now), + _quote(strike="2350", delta="-0.08", mid="0.007", expiry_dte=40, now_dt=now), + ] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_picks_expiry_closest_to_dte_target( + cfg: StrategyConfig, now: datetime +) -> None: + # Two valid expiries: ~16 DTE and ~21 DTE. Target=18 → 16 closer than 21. + chain = [ + _quote(strike="2475", delta="-0.12", mid="0.020", expiry_dte=17, now_dt=now), + _quote(strike="2350", delta="-0.08", mid="0.005", expiry_dte=17, now_dt=now), + _quote(strike="2475", delta="-0.12", mid="0.025", expiry_dte=22, now_dt=now), + _quote(strike="2350", delta="-0.08", mid="0.008", expiry_dte=22, now_dt=now), + ] + result = select_strikes( + chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg + ) + assert result is not None + short, _ = result + # 17 days requested, but expiry is set at 08:00 UTC and now at 14:00, so + # the calendar-day delta is 16 days for the closer expiry, 21 for the far. + picked_dte = (short.expiry - now).days + assert picked_dte == 16 + + +def test_select_strikes_bear_call_picks_calls_above_spot( + cfg: StrategyConfig, now: datetime +) -> None: + spot = Decimal("3000") + chain = [ + # OTM call range = [3450, 3750] (15% to 25% above). + # Mids tuned so credit ≈ 36% of width (≥ 30% gate). + _quote(strike="3525", delta="0.12", mid="0.020", option_type="C", now_dt=now), + _quote(strike="3645", delta="0.09", mid="0.005", option_type="C", now_dt=now), + _quote(strike="3450", delta="0.14", mid="0.024", option_type="C", now_dt=now), + # noise: puts that should be ignored + _quote(strike="2475", delta="-0.12", mid="0.020", option_type="P", now_dt=now), + ] + result = select_strikes(chain=chain, bias="bear_call", spot=spot, now=now, cfg=cfg) + assert result is not None + short, long_ = result + assert short.option_type == "C" + assert long_.option_type == "C" + # short should be 3525 (delta 0.12, closest to target) + assert short.strike == Decimal("3525") + # width target = 120 above → 3645 + assert long_.strike == Decimal("3645") + + +def test_select_strikes_iron_condor_not_supported( + cfg: StrategyConfig, now: datetime +) -> None: + # IC needs a different orchestration; for now select_strikes returns None for IC. + chain = _bull_put_chain(now) + assert ( + select_strikes(chain=chain, bias="iron_condor", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_returns_none_when_chain_has_no_matching_type( + cfg: StrategyConfig, now: datetime +) -> None: + # Bull put requested but the chain only has calls within the DTE window. + chain = [ + _quote(strike="3525", delta="0.12", mid="0.020", option_type="C", now_dt=now), + _quote(strike="3645", delta="0.08", mid="0.005", option_type="C", now_dt=now), + ] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + +def test_select_strikes_returns_none_when_no_long_candidate_exists( + cfg: StrategyConfig, now: datetime +) -> None: + # Only the short strike exists; nothing further OTM to pair as the long leg. + chain = [_quote(strike="2475", delta="-0.12", mid="0.020", now_dt=now)] + assert ( + select_strikes(chain=chain, bias="bull_put", spot=Decimal("3000"), now=now, cfg=cfg) + is None + ) + + + + +# --------------------------------------------------------------------------- +# build +# --------------------------------------------------------------------------- + + +def test_build_returns_proposal_with_correct_legs_and_metrics( + cfg: StrategyConfig, now: datetime +) -> None: + short = _quote(strike="2475", delta="-0.12", mid="0.012", now_dt=now) + long_ = _quote(strike="2350", delta="-0.08", mid="0.007", now_dt=now) + proposal = build( + short=short, + long_=long_, + n_contracts=2, + spot=Decimal("3000"), + dvol=Decimal("50"), + cfg=cfg, + now=now, + spread_type="bull_put", + ) + # legs ordered short-first + assert len(proposal.legs) == 2 + assert proposal.legs[0].side == "SELL" + assert proposal.legs[0].strike == Decimal("2475") + assert proposal.legs[1].side == "BUY" + assert proposal.legs[1].strike == Decimal("2350") + assert proposal.legs[0].size == 2 + assert proposal.legs[1].size == 2 + + # credit per contract = 0.012 - 0.007 = 0.005 ETH; n=2 → 0.010 ETH + assert proposal.credit_target_eth == Decimal("0.010") + # credit USD = 0.010 × 3000 = 30 + assert proposal.credit_target_usd == Decimal("30") + + # width = 125 USD per contract; credit_per_contract_usd = 0.005×3000 = 15; + # max_loss per contract = 125 - 15 = 110 USD; n=2 → 220. + assert proposal.max_loss_usd == Decimal("220") + assert proposal.spot_at_proposal == Decimal("3000") + assert proposal.dvol_at_proposal == Decimal("50") + assert proposal.spread_type == "bull_put" + # breakeven (bull put) = short_strike - credit_per_contract_usd = 2475 - 15 = 2460 + assert proposal.breakeven == Decimal("2460.000") + assert isinstance(proposal.proposal_id, UUID) + + +def test_build_bear_call_breakeven_above_short_strike( + cfg: StrategyConfig, now: datetime +) -> None: + short = _quote(strike="3525", delta="0.12", mid="0.012", option_type="C", now_dt=now) + long_ = _quote(strike="3645", delta="0.08", mid="0.007", option_type="C", now_dt=now) + proposal = build( + short=short, + long_=long_, + n_contracts=1, + spot=Decimal("3000"), + dvol=Decimal("55"), + cfg=cfg, + now=now, + spread_type="bear_call", + ) + # credit per contract = 0.005 ETH × 3000 = 15 USD + # breakeven = 3525 + 15 = 3540 + assert proposal.breakeven == Decimal("3540") + assert proposal.spread_type == "bear_call" diff --git a/tests/unit/test_config_schema.py b/tests/unit/test_config_schema.py new file mode 100644 index 0000000..eb6c83d --- /dev/null +++ b/tests/unit/test_config_schema.py @@ -0,0 +1,104 @@ +"""Unit tests for :mod:`cerbero_bite.config.schema`.""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.config import ( + EntryConfig, + ExitConfig, + ShortStrikeSpec, + SizingConfig, + SpreadWidthSpec, + StrategyConfig, + StructureConfig, + golden_config, +) + + +def test_golden_config_builds_with_defaults() -> None: + cfg = golden_config() + assert cfg.config_version == "1.0.0-test" + assert cfg.sizing.kelly_fraction == Decimal("0.13") + assert cfg.exit.profit_take_pct_of_credit == Decimal("0.50") + assert cfg.entry.dvol_min == Decimal("35") + assert cfg.entry.dvol_max == Decimal("90") + + +def test_dvol_adjustment_default_bands_ordered() -> None: + cfg = golden_config() + bands = cfg.sizing.dvol_adjustment + assert [b.dvol_under for b in bands] == [Decimal("45"), Decimal("60"), Decimal("80")] + assert [b.multiplier for b in bands] == [ + Decimal("1.00"), + Decimal("0.85"), + Decimal("0.65"), + ] + + +def test_validator_rejects_kelly_out_of_safe_range() -> None: + with pytest.raises(ValueError, match="kelly_fraction"): + golden_config(sizing=SizingConfig(kelly_fraction=Decimal("0.6"))) + + +def test_validator_rejects_delta_target_outside_band() -> None: + bad = ShortStrikeSpec( + delta_target=Decimal("0.20"), + delta_min=Decimal("0.10"), + delta_max=Decimal("0.15"), + ) + with pytest.raises(ValueError, match="delta_target"): + golden_config(structure=StructureConfig(short_strike=bad)) + + +def test_validator_rejects_inverted_otm_range() -> None: + bad = ShortStrikeSpec( + distance_otm_pct_min=Decimal("0.30"), + distance_otm_pct_max=Decimal("0.20"), + ) + with pytest.raises(ValueError, match="OTM range"): + golden_config(structure=StructureConfig(short_strike=bad)) + + +def test_validator_rejects_inverted_spread_width() -> None: + bad = SpreadWidthSpec( + target_pct_of_spot=Decimal("0.06"), + min_pct_of_spot=Decimal("0.03"), + max_pct_of_spot=Decimal("0.05"), + ) + with pytest.raises(ValueError, match="spread_width"): + golden_config(structure=StructureConfig(spread_width=bad)) + + +def test_validator_rejects_profit_take_ge_100pct() -> None: + with pytest.raises(ValueError, match="profit_take"): + golden_config(exit=ExitConfig(profit_take_pct_of_credit=Decimal("1.0"))) + + +def test_validator_rejects_stop_loss_le_1x() -> None: + with pytest.raises(ValueError, match="stop_loss"): + golden_config(exit=ExitConfig(stop_loss_mark_x_credit=Decimal("1.0"))) + + +def test_validator_rejects_inverted_dvol_range() -> None: + with pytest.raises(ValueError, match="dvol_min"): + golden_config(entry=EntryConfig(dvol_min=Decimal("90"), dvol_max=Decimal("35"))) + + +def test_strategy_config_is_immutable() -> None: + cfg = golden_config() + with pytest.raises(Exception): # pydantic raises ValidationError on frozen + cfg.entry = EntryConfig() # type: ignore[misc] + + +def test_extra_fields_rejected_on_root() -> None: + with pytest.raises(ValueError): + StrategyConfig( + config_version="x", + config_hash="0" * 64, + last_review="2026-04-26", + last_reviewer="test", + unknown_field=42, # type: ignore[call-arg] + ) diff --git a/tests/unit/test_entry_validator.py b/tests/unit/test_entry_validator.py new file mode 100644 index 0000000..43f1727 --- /dev/null +++ b/tests/unit/test_entry_validator.py @@ -0,0 +1,247 @@ +"""TDD for :mod:`cerbero_bite.core.entry_validator`. + +Spec: ``docs/01-strategy-rules.md §2,§3.1`` and ``docs/03-algorithms.md §1``. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.config import StrategyConfig, golden_config +from cerbero_bite.core.entry_validator import ( + EntryContext, + TrendContext, + compute_bias, + validate_entry, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _good_ctx(**overrides: object) -> EntryContext: + base: dict[str, object] = { + "capital_usd": Decimal("1500"), + "dvol_now": Decimal("50"), + "funding_perp_annualized": Decimal("0.10"), + "eth_holdings_pct_of_portfolio": Decimal("0.10"), + "next_macro_event_in_days": None, + "has_open_position": False, + } + base.update(overrides) + return EntryContext(**base) # type: ignore[arg-type] + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +# --------------------------------------------------------------------------- +# validate_entry — happy path +# --------------------------------------------------------------------------- + + +def test_validate_entry_accepts_clean_context(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(), cfg) + assert decision.accepted is True + assert decision.reasons == [] + + +# --------------------------------------------------------------------------- +# validate_entry — single-reason rejections +# --------------------------------------------------------------------------- + + +def test_open_position_blocks_entry(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(has_open_position=True), cfg) + assert decision.accepted is False + assert any("position" in r for r in decision.reasons) + + +def test_capital_below_minimum_blocks_entry(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(capital_usd=Decimal("719.99")), cfg) + assert decision.accepted is False + assert any("capital" in r for r in decision.reasons) + + +def test_capital_at_minimum_is_accepted(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(capital_usd=Decimal("720")), cfg) + assert decision.accepted is True + + +def test_dvol_below_minimum_blocks_entry(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(dvol_now=Decimal("34.99")), cfg) + assert decision.accepted is False + assert any("dvol" in r for r in decision.reasons) + + +def test_dvol_above_maximum_blocks_entry(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(dvol_now=Decimal("90.01")), cfg) + assert decision.accepted is False + assert any("dvol" in r for r in decision.reasons) + + +def test_dvol_at_boundaries_is_accepted(cfg: StrategyConfig) -> None: + assert validate_entry(_good_ctx(dvol_now=Decimal("35")), cfg).accepted + assert validate_entry(_good_ctx(dvol_now=Decimal("90")), cfg).accepted + + +def test_macro_event_within_dte_blocks_entry(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(next_macro_event_in_days=5), cfg) + assert decision.accepted is False + assert any("macro" in r for r in decision.reasons) + + +def test_macro_event_at_dte_boundary_blocks_entry(cfg: StrategyConfig) -> None: + # next_macro_event_in_days <= dte_target → block. dte_target = 18. + decision = validate_entry(_good_ctx(next_macro_event_in_days=18), cfg) + assert decision.accepted is False + + +def test_macro_event_beyond_dte_is_ok(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(next_macro_event_in_days=19), cfg) + assert decision.accepted is True + + +def test_macro_none_is_ok(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(next_macro_event_in_days=None), cfg) + assert decision.accepted is True + + +def test_funding_above_abs_cap_blocks(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(funding_perp_annualized=Decimal("0.81")), cfg) + assert decision.accepted is False + assert any("funding" in r for r in decision.reasons) + + +def test_funding_negative_below_neg_cap_blocks(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(funding_perp_annualized=Decimal("-0.81")), cfg) + assert decision.accepted is False + + +def test_funding_at_cap_is_accepted(cfg: StrategyConfig) -> None: + assert validate_entry(_good_ctx(funding_perp_annualized=Decimal("0.80")), cfg).accepted + + +def test_eth_holdings_above_cap_blocks(cfg: StrategyConfig) -> None: + decision = validate_entry(_good_ctx(eth_holdings_pct_of_portfolio=Decimal("0.31")), cfg) + assert decision.accepted is False + assert any("holdings" in r for r in decision.reasons) + + +def test_eth_holdings_at_cap_is_accepted(cfg: StrategyConfig) -> None: + assert validate_entry( + _good_ctx(eth_holdings_pct_of_portfolio=Decimal("0.30")), cfg + ).accepted + + +# --------------------------------------------------------------------------- +# validate_entry — accumulates ALL failure reasons +# --------------------------------------------------------------------------- + + +def test_validate_entry_accumulates_all_reasons(cfg: StrategyConfig) -> None: + decision = validate_entry( + _good_ctx( + has_open_position=True, + capital_usd=Decimal("100"), + dvol_now=Decimal("10"), + funding_perp_annualized=Decimal("1.5"), + eth_holdings_pct_of_portfolio=Decimal("0.9"), + next_macro_event_in_days=2, + ), + cfg, + ) + assert decision.accepted is False + assert len(decision.reasons) >= 6 + + +# --------------------------------------------------------------------------- +# compute_bias +# --------------------------------------------------------------------------- + + +def _trend( + *, + eth_now: str = "3000", + eth_30d_ago: str = "3000", + funding_cross: str = "0", + dvol: str = "50", + adx: str = "25", +) -> TrendContext: + return TrendContext( + eth_now=Decimal(eth_now), + eth_30d_ago=Decimal(eth_30d_ago), + funding_cross_annualized=Decimal(funding_cross), + dvol_now=Decimal(dvol), + adx_14=Decimal(adx), + ) + + +def test_bias_both_bull_returns_bull_put(cfg: StrategyConfig) -> None: + # +6% trend, +25% funding + ctx = _trend(eth_now="3180", eth_30d_ago="3000", funding_cross="0.25") + assert compute_bias(ctx, cfg) == "bull_put" + + +def test_bias_both_bear_returns_bear_call(cfg: StrategyConfig) -> None: + # -6% trend, -25% funding + ctx = _trend(eth_now="2820", eth_30d_ago="3000", funding_cross="-0.25") + assert compute_bias(ctx, cfg) == "bear_call" + + +def test_bias_discordant_returns_none(cfg: StrategyConfig) -> None: + # bull trend, bear funding + ctx = _trend(eth_now="3180", eth_30d_ago="3000", funding_cross="-0.25") + assert compute_bias(ctx, cfg) is None + + +def test_bias_neutral_with_high_dvol_low_adx_returns_iron_condor( + cfg: StrategyConfig, +) -> None: + # 0% trend, 0% funding, dvol 60, adx 15 → IC + ctx = _trend( + eth_now="3000", eth_30d_ago="3000", funding_cross="0", dvol="60", adx="15" + ) + assert compute_bias(ctx, cfg) == "iron_condor" + + +def test_bias_neutral_with_low_dvol_returns_none(cfg: StrategyConfig) -> None: + ctx = _trend( + eth_now="3000", eth_30d_ago="3000", funding_cross="0", dvol="50", adx="15" + ) + assert compute_bias(ctx, cfg) is None + + +def test_bias_neutral_with_high_adx_returns_none(cfg: StrategyConfig) -> None: + ctx = _trend( + eth_now="3000", eth_30d_ago="3000", funding_cross="0", dvol="60", adx="22" + ) + assert compute_bias(ctx, cfg) is None + + +def test_bias_one_neutral_one_bull_returns_none(cfg: StrategyConfig) -> None: + # +6% trend, +10% funding (neutral) + ctx = _trend(eth_now="3180", eth_30d_ago="3000", funding_cross="0.10") + assert compute_bias(ctx, cfg) is None + + +def test_bias_at_bull_threshold_is_bull_put(cfg: StrategyConfig) -> None: + # exactly +5% trend, exactly +20% funding + ctx = _trend(eth_now="3150", eth_30d_ago="3000", funding_cross="0.20") + assert compute_bias(ctx, cfg) == "bull_put" + + +def test_bias_at_bear_threshold_is_bear_call(cfg: StrategyConfig) -> None: + ctx = _trend(eth_now="2850", eth_30d_ago="3000", funding_cross="-0.20") + assert compute_bias(ctx, cfg) == "bear_call" + + +def test_bias_zero_division_safe(cfg: StrategyConfig) -> None: + # eth_30d_ago == 0 must not crash; treat as no-bias (neutral) + ctx = _trend(eth_now="3000", eth_30d_ago="0", funding_cross="0", dvol="60", adx="15") + assert compute_bias(ctx, cfg) is None diff --git a/tests/unit/test_exit_decision.py b/tests/unit/test_exit_decision.py new file mode 100644 index 0000000..3a3b1e1 --- /dev/null +++ b/tests/unit/test_exit_decision.py @@ -0,0 +1,273 @@ +"""TDD for :mod:`cerbero_bite.core.exit_decision`. + +Spec: ``docs/01-strategy-rules.md §7`` and ``docs/03-algorithms.md §6``. +The eight mandatory cases listed in §6 are exercised here, with the +``mark=30% credit`` case re-interpreted to satisfy the time-stop +exception consistently with rule 1 (which always fires earlier when +mark ≤ 50% credit). +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from uuid import uuid4 + +import pytest + +from cerbero_bite.config import SpreadType, StrategyConfig, golden_config +from cerbero_bite.core.exit_decision import PositionSnapshot, evaluate +from cerbero_bite.core.types import OptionLeg + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +def _legs(spread_type: SpreadType = "bull_put", size: int = 1) -> list[OptionLeg]: + expiry = datetime(2026, 5, 15, 8, 0, tzinfo=UTC) + if spread_type == "bull_put": + return [ + OptionLeg( + instrument="ETH-X-2475-P", + side="SELL", + strike=Decimal("2475"), + expiry=expiry, + type="P", + size=size, + mid_price_eth=Decimal("0.020"), + delta=Decimal("-0.12"), + gamma=Decimal("0.001"), + theta=Decimal("-0.0005"), + vega=Decimal("0.10"), + ), + OptionLeg( + instrument="ETH-X-2350-P", + side="BUY", + strike=Decimal("2350"), + expiry=expiry, + type="P", + size=size, + mid_price_eth=Decimal("0.005"), + delta=Decimal("-0.08"), + gamma=Decimal("0.001"), + theta=Decimal("-0.0003"), + vega=Decimal("0.07"), + ), + ] + return [ + OptionLeg( + instrument="ETH-X-3525-C", + side="SELL", + strike=Decimal("3525"), + expiry=expiry, + type="C", + size=size, + mid_price_eth=Decimal("0.020"), + delta=Decimal("0.12"), + gamma=Decimal("0.001"), + theta=Decimal("-0.0005"), + vega=Decimal("0.10"), + ), + OptionLeg( + instrument="ETH-X-3645-C", + side="BUY", + strike=Decimal("3645"), + expiry=expiry, + type="C", + size=size, + mid_price_eth=Decimal("0.005"), + delta=Decimal("0.08"), + gamma=Decimal("0.001"), + theta=Decimal("-0.0003"), + vega=Decimal("0.07"), + ), + ] + + +def _snapshot( + *, + spread_type: SpreadType = "bull_put", + credit_received_eth: str = "0.030", + mark_combo_now_eth: str = "0.020", + dvol_at_entry: str = "50", + dvol_now: str = "50", + delta_short_now: str | None = None, + return_4h_now: str = "0", + days_to_expiry: int = 14, + spot_at_entry: str = "3000", + spot_now: str = "3000", + eth_price_usd_now: str = "3000", +) -> PositionSnapshot: + now = datetime(2026, 4, 27, 14, 0, tzinfo=UTC) + expiry = now + timedelta(days=days_to_expiry) + if delta_short_now is None: + delta_short_now = "-0.12" if spread_type == "bull_put" else "0.12" + legs = _legs(spread_type=spread_type) + return PositionSnapshot( + proposal_id=uuid4(), + spread_type=spread_type, + legs=legs, + credit_received_eth=Decimal(credit_received_eth), + credit_received_usd=Decimal(credit_received_eth) * Decimal(eth_price_usd_now), + spot_at_entry=Decimal(spot_at_entry), + dvol_at_entry=Decimal(dvol_at_entry), + expiry=expiry, + opened_at=now - timedelta(days=4), + eth_price_usd_now=Decimal(eth_price_usd_now), + spot_now=Decimal(spot_now), + dvol_now=Decimal(dvol_now), + mark_combo_now_eth=Decimal(mark_combo_now_eth), + delta_short_now=Decimal(delta_short_now), + return_4h_now=Decimal(return_4h_now), + now=now, + ) + + +# --------------------------------------------------------------------------- +# Mandatory cases from docs/03-algorithms.md §6 +# --------------------------------------------------------------------------- + + +def test_mark_at_50pct_credit_triggers_close_profit(cfg: StrategyConfig) -> None: + snap = _snapshot(credit_received_eth="0.030", mark_combo_now_eth="0.015") + res = evaluate(snap, cfg) + assert res.action == "CLOSE_PROFIT" + + +def test_mark_at_250pct_credit_triggers_close_stop(cfg: StrategyConfig) -> None: + snap = _snapshot(credit_received_eth="0.030", mark_combo_now_eth="0.075") + res = evaluate(snap, cfg) + assert res.action == "CLOSE_STOP" + + +def test_dvol_increase_above_band_triggers_close_vol(cfg: StrategyConfig) -> None: + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.020", # 67% credit, no profit/stop + dvol_at_entry="50", + dvol_now="62", # +12, above +10 trigger + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_VOL" + + +def test_six_days_left_mark_above_skip_threshold_triggers_close_time( + cfg: StrategyConfig, +) -> None: + # mark = 80% credit > 70% credit (skip threshold) → CLOSE_TIME + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.024", + days_to_expiry=6, + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_TIME" + + +def test_six_days_left_mark_in_skip_zone_holds(cfg: StrategyConfig) -> None: + # mark = 60% credit ∈ (50%, 70%] → close to profit, time stop is skipped. + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.018", + days_to_expiry=6, + ) + res = evaluate(snap, cfg) + assert res.action == "HOLD" + + +def test_short_delta_breach_triggers_close_delta(cfg: StrategyConfig) -> None: + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.020", + delta_short_now="-0.32", + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_DELTA" + + +def test_4h_adverse_move_bull_put_triggers_close_averse(cfg: StrategyConfig) -> None: + snap = _snapshot( + spread_type="bull_put", + credit_received_eth="0.030", + mark_combo_now_eth="0.020", + return_4h_now="-0.07", + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_AVERSE" + + +def test_4h_adverse_move_bear_call_triggers_close_averse(cfg: StrategyConfig) -> None: + snap = _snapshot( + spread_type="bear_call", + credit_received_eth="0.030", + mark_combo_now_eth="0.020", + return_4h_now="0.07", + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_AVERSE" + + +def test_neutral_state_returns_hold(cfg: StrategyConfig) -> None: + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.020", + ) + res = evaluate(snap, cfg) + assert res.action == "HOLD" + + +# --------------------------------------------------------------------------- +# Trigger ordering — first match wins +# --------------------------------------------------------------------------- + + +def test_profit_wins_over_vol_stop(cfg: StrategyConfig) -> None: + # mark = 30% credit AND DVOL +12 → profit takes precedence + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.009", + dvol_at_entry="50", + dvol_now="62", + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_PROFIT" + + +def test_stop_wins_over_delta_breach(cfg: StrategyConfig) -> None: + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.075", # 250% → stop + delta_short_now="-0.40", # would also be CLOSE_DELTA + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_STOP" + + +# --------------------------------------------------------------------------- +# PnL reporting +# --------------------------------------------------------------------------- + + +def test_pnl_in_eth_and_usd_reflect_credit_minus_debit(cfg: StrategyConfig) -> None: + snap = _snapshot( + credit_received_eth="0.030", + mark_combo_now_eth="0.018", + eth_price_usd_now="3100", + ) + res = evaluate(snap, cfg) + # pnl_eth = 0.030 - 0.018 = 0.012; pnl_usd = 0.012 × 3100 = 37.2 + assert res.pnl_estimate_eth == Decimal("0.012") + assert res.pnl_estimate_usd == Decimal("37.2") + + +def test_iron_condor_adverse_move_either_direction(cfg: StrategyConfig) -> None: + snap = _snapshot( + spread_type="iron_condor", + credit_received_eth="0.030", + mark_combo_now_eth="0.020", + return_4h_now="-0.06", + ) + res = evaluate(snap, cfg) + assert res.action == "CLOSE_AVERSE" diff --git a/tests/unit/test_greeks_aggregator.py b/tests/unit/test_greeks_aggregator.py new file mode 100644 index 0000000..24df8de --- /dev/null +++ b/tests/unit/test_greeks_aggregator.py @@ -0,0 +1,100 @@ +"""TDD for :mod:`cerbero_bite.core.greeks_aggregator`. + +Spec: ``docs/03-algorithms.md §5``. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from decimal import Decimal + +from cerbero_bite.core.greeks_aggregator import aggregate +from cerbero_bite.core.types import OptionLeg + + +def _leg( + *, + side: str, + size: int = 1, + delta: str, + gamma: str = "0.001", + theta: str = "-0.0005", + vega: str = "0.10", + strike: str = "2400", + option_type: str = "P", + instrument: str = "ETH-X-2400-P", +) -> OptionLeg: + return OptionLeg( + instrument=instrument, + side=side, # type: ignore[arg-type] + strike=Decimal(strike), + expiry=datetime(2026, 5, 15, 8, 0, tzinfo=UTC), + type=option_type, # type: ignore[arg-type] + size=size, + mid_price_eth=Decimal("0.012"), + delta=Decimal(delta), + gamma=Decimal(gamma), + theta=Decimal(theta), + vega=Decimal(vega), + ) + + +def test_bull_put_spread_net_delta_long_positive() -> None: + """SELL put (negative delta) + BUY further-OTM put → net delta positive.""" + legs = [ + _leg(side="SELL", delta="-0.12"), + _leg(side="BUY", delta="-0.08", strike="2300"), + ] + res = aggregate(legs=legs, eth_price_usd=Decimal("3000")) + # delta_net = -1*-0.12 + +1*-0.08 = 0.12 - 0.08 = +0.04 + assert res.delta_net == Decimal("0.04") + + +def test_bear_call_spread_net_delta_short_negative() -> None: + legs = [ + _leg(side="SELL", delta="0.12", option_type="C"), + _leg(side="BUY", delta="0.08", option_type="C", strike="3650"), + ] + res = aggregate(legs=legs, eth_price_usd=Decimal("3000")) + # delta_net = -1*0.12 + +1*0.08 = -0.04 + assert res.delta_net == Decimal("-0.04") + + +def test_size_scales_each_leg() -> None: + legs = [ + _leg(side="SELL", size=3, delta="-0.12"), + _leg(side="BUY", size=3, delta="-0.08", strike="2300"), + ] + res = aggregate(legs=legs, eth_price_usd=Decimal("3000")) + assert res.delta_net == Decimal("0.12") # 3 × 0.04 + + +def test_theta_converted_to_usd_per_day() -> None: + legs = [ + _leg(side="SELL", delta="-0.12", theta="-0.0005"), # selling → +0.0005 ETH theta + _leg(side="BUY", delta="-0.08", theta="-0.0003", strike="2300"), # +0.0003 cost + ] + res = aggregate(legs=legs, eth_price_usd=Decimal("3000")) + # theta_eth = -1*(-0.0005) + +1*(-0.0003) = 0.0005 - 0.0003 = 0.0002 + # theta_usd = 0.0002 × 3000 = 0.60 + assert res.theta_net == Decimal("0.60") + + +def test_gamma_and_vega_summed_with_sign() -> None: + legs = [ + _leg(side="SELL", delta="-0.12", gamma="0.0020", vega="0.30"), + _leg(side="BUY", delta="-0.08", gamma="0.0015", vega="0.20", strike="2300"), + ] + res = aggregate(legs=legs, eth_price_usd=Decimal("3000")) + # gamma: -1*0.0020 + 1*0.0015 = -0.0005 + # vega: -1*0.30 + 1*0.20 = -0.10 + assert res.gamma_net == Decimal("-0.0005") + assert res.vega_net == Decimal("-0.10") + + +def test_empty_legs_returns_zero_greeks() -> None: + res = aggregate(legs=[], eth_price_usd=Decimal("3000")) + assert res.delta_net == Decimal("0") + assert res.gamma_net == Decimal("0") + assert res.theta_net == Decimal("0") + assert res.vega_net == Decimal("0") diff --git a/tests/unit/test_kelly_recalibration.py b/tests/unit/test_kelly_recalibration.py new file mode 100644 index 0000000..069a247 --- /dev/null +++ b/tests/unit/test_kelly_recalibration.py @@ -0,0 +1,155 @@ +"""TDD for :mod:`cerbero_bite.core.kelly_recalibration`. + +Spec: ``docs/03-algorithms.md §7``. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from uuid import uuid4 + +import pytest + +from cerbero_bite.config import StrategyConfig, golden_config +from cerbero_bite.core.kelly_recalibration import TradeRecord, recalibrate + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +@pytest.fixture +def now() -> datetime: + return datetime(2026, 4, 27, 14, 0, tzinfo=UTC) + + +def _trade( + *, + pnl_usd: str, + risk_usd: str = "100", + days_ago: int = 30, + outcome: str = "CLOSE_PROFIT", + now_dt: datetime = datetime(2026, 4, 27, 14, 0, tzinfo=UTC), +) -> TradeRecord: + return TradeRecord( + proposal_id=uuid4(), + pnl_usd=Decimal(pnl_usd), + risk_usd=Decimal(risk_usd), + closed_at=now_dt - timedelta(days=days_ago), + outcome=outcome, + ) + + +# --------------------------------------------------------------------------- +# Confidence bands and recommended fraction +# --------------------------------------------------------------------------- + + +def test_empty_history_returns_low_confidence_with_cfg_kelly( + cfg: StrategyConfig, now: datetime +) -> None: + res = recalibrate(trades=[], now=now, cfg=cfg) + assert res.confidence == "low" + assert res.sample_size == 0 + assert res.recommended_fraction == cfg.sizing.kelly_fraction + assert res.win_rate == Decimal("0") + assert res.full_kelly_pct == Decimal("0") + assert res.quarter_kelly_pct == Decimal("0") + + +def test_low_sample_keeps_cfg_kelly_fraction(cfg: StrategyConfig, now: datetime) -> None: + trades = [_trade(pnl_usd="10", now_dt=now) for _ in range(20)] + trades += [_trade(pnl_usd="-15", now_dt=now) for _ in range(5)] + res = recalibrate(trades=trades, now=now, cfg=cfg) + assert res.sample_size == 25 + assert res.confidence == "low" + assert res.recommended_fraction == cfg.sizing.kelly_fraction + + +def test_medium_sample_blends_quarter_kelly_with_cfg( + cfg: StrategyConfig, now: datetime +) -> None: + # 60 trades, 70% wins (avg_win=10/100=0.10), 30% losses (avg_loss=15/100=0.15) + # b = 0.10 / 0.15 = 0.667 + # full_kelly = (0.7 × 0.667 - 0.3) / 0.667 = (0.467 - 0.3) / 0.667 = 0.25 + # quarter_kelly = 0.0625 + # blended = 0.5 × 0.0625 + 0.5 × 0.13 = 0.09625 + wins = [_trade(pnl_usd="10", now_dt=now) for _ in range(42)] + losses = [_trade(pnl_usd="-15", now_dt=now) for _ in range(18)] + res = recalibrate(trades=wins + losses, now=now, cfg=cfg) + assert res.sample_size == 60 + assert res.confidence == "medium" + assert res.win_rate == Decimal("0.7") + expected = Decimal("0.5") * res.quarter_kelly_pct + Decimal("0.5") * cfg.sizing.kelly_fraction + assert res.recommended_fraction == expected + + +def test_high_sample_adopts_quarter_kelly(cfg: StrategyConfig, now: datetime) -> None: + wins = [_trade(pnl_usd="10", now_dt=now) for _ in range(120)] + losses = [_trade(pnl_usd="-15", now_dt=now) for _ in range(40)] + res = recalibrate(trades=wins + losses, now=now, cfg=cfg) + assert res.sample_size == 160 + assert res.confidence == "high" + assert res.recommended_fraction == res.quarter_kelly_pct + + +# --------------------------------------------------------------------------- +# Lookback filter +# --------------------------------------------------------------------------- + + +def test_trades_older_than_lookback_are_ignored( + cfg: StrategyConfig, now: datetime +) -> None: + fresh = [_trade(pnl_usd="10", days_ago=30, now_dt=now) for _ in range(40)] + stale = [_trade(pnl_usd="-50", days_ago=400, now_dt=now) for _ in range(60)] + res = recalibrate(trades=fresh + stale, now=now, cfg=cfg) + # Only 40 trades in window; all wins → win_rate = 1.0 + assert res.sample_size == 40 + assert res.win_rate == Decimal("1") + + +# --------------------------------------------------------------------------- +# Edge cases for full_kelly arithmetic +# --------------------------------------------------------------------------- + + +def test_all_losses_clip_full_kelly_to_zero(cfg: StrategyConfig, now: datetime) -> None: + trades = [_trade(pnl_usd="-20", now_dt=now) for _ in range(40)] + res = recalibrate(trades=trades, now=now, cfg=cfg) + assert res.win_rate == Decimal("0") + assert res.full_kelly_pct == Decimal("0") + assert res.quarter_kelly_pct == Decimal("0") + + +def test_all_wins_yield_full_kelly_equal_to_win_rate( + cfg: StrategyConfig, now: datetime +) -> None: + # No losses → division by zero. Convention: full_kelly = win_rate (= 1.0). + trades = [_trade(pnl_usd="20", now_dt=now) for _ in range(40)] + res = recalibrate(trades=trades, now=now, cfg=cfg) + assert res.win_rate == Decimal("1") + assert res.avg_loss_pct_risk == Decimal("0") + assert res.full_kelly_pct == Decimal("1") + assert res.quarter_kelly_pct == Decimal("0.25") + + +def test_avg_loss_uses_absolute_value(cfg: StrategyConfig, now: datetime) -> None: + wins = [_trade(pnl_usd="10", now_dt=now) for _ in range(30)] + losses = [_trade(pnl_usd="-20", now_dt=now) for _ in range(30)] + res = recalibrate(trades=wins + losses, now=now, cfg=cfg) + # avg_win_pct_risk = 10/100 = 0.10; avg_loss_pct_risk = 20/100 = 0.20 (positive) + assert res.avg_win_pct_risk == Decimal("0.1") + assert res.avg_loss_pct_risk == Decimal("0.2") + + +def test_zero_pnl_counts_as_loss_for_winrate(cfg: StrategyConfig, now: datetime) -> None: + # Spec: "trade con pnl > 0" → wins. 0 is not a win. + trades = [ + *[_trade(pnl_usd="10", now_dt=now) for _ in range(20)], + *[_trade(pnl_usd="0", now_dt=now) for _ in range(20)], + ] + res = recalibrate(trades=trades, now=now, cfg=cfg) + assert res.win_rate == Decimal("0.5") diff --git a/tests/unit/test_liquidity_gate.py b/tests/unit/test_liquidity_gate.py new file mode 100644 index 0000000..120c418 --- /dev/null +++ b/tests/unit/test_liquidity_gate.py @@ -0,0 +1,265 @@ +"""TDD for :mod:`cerbero_bite.core.liquidity_gate`. + +Spec: ``docs/01-strategy-rules.md §4`` and ``docs/03-algorithms.md §2``. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.config import StrategyConfig, golden_config +from cerbero_bite.core.liquidity_gate import InstrumentSnapshot, check + + +def _snap( + *, + instrument: str = "ETH-13MAY26-2400-P", + bid: str = "0.090", + ask: str = "0.100", + mid: str = "0.095", + open_interest: int = 500, + volume_24h: int = 80, + book_depth_top3: int = 20, +) -> InstrumentSnapshot: + return InstrumentSnapshot( + instrument=instrument, + bid=Decimal(bid), + ask=Decimal(ask), + mid=Decimal(mid), + open_interest=open_interest, + volume_24h=volume_24h, + book_depth_top3=book_depth_top3, + ) + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +# Two-leg vertical: short at 0.095 mid, long at 0.060 mid → credit 0.035 ETH. +# Tight book: per-contract slippage 0.001 ETH → ~2.86% of credit at n=1. +def _good_pair() -> tuple[InstrumentSnapshot, InstrumentSnapshot]: + short = _snap(bid="0.0945", ask="0.0955", mid="0.0950") + long_ = _snap( + instrument="ETH-13MAY26-2300-P", + bid="0.0595", + ask="0.0605", + mid="0.0600", + ) + return short, long_ + + +def test_clean_pair_passes(cfg: StrategyConfig) -> None: + # tight book: per-contract slippage 0.001 ETH → 2.86% of 0.035 credit. + short, long_ = _good_pair() + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is True + assert res.reasons == [] + + +def test_short_leg_oi_below_threshold_fails(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + short = _snap( + instrument=short.instrument, + bid=str(short.bid), + ask=str(short.ask), + mid=str(short.mid), + open_interest=99, + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("open interest" in r for r in res.reasons) + + +def test_long_leg_volume_below_threshold_fails(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + long_ = _snap( + instrument=long_.instrument, + bid=str(long_.bid), + ask=str(long_.ask), + mid=str(long_.mid), + volume_24h=10, + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("volume" in r for r in res.reasons) + + +def test_spread_pct_above_cap_fails(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + # Make bid-ask huge: bid 0.05, ask 0.20 → spread/mid > 0.15 + short = _snap( + instrument=short.instrument, + bid="0.050", + ask="0.200", + mid="0.125", + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.060"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("bid-ask" in r for r in res.reasons) + + +def test_book_depth_below_threshold_fails(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + short = _snap( + instrument=short.instrument, + bid=str(short.bid), + ask=str(short.ask), + mid=str(short.mid), + book_depth_top3=4, + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("depth" in r for r in res.reasons) + + +def test_slippage_above_8pct_fails(cfg: StrategyConfig) -> None: + # tight book * n=10 → 0.010 / 0.035 ≈ 28% slippage + short, long_ = _good_pair() + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=10, + cfg=cfg, + ) + assert res.accepted is False + assert any("slippage" in r for r in res.reasons) + assert res.estimated_slippage_pct_of_credit > Decimal("0.08") + + +def test_slippage_well_below_cap_passes(cfg: StrategyConfig) -> None: + # tight book: short ask-mid = 0.0005, long mid-bid = 0.0005 → slip = 0.001 per + # contract, credit 0.035 → 2.85% per contract. + short = _snap(bid="0.0945", ask="0.0955", mid="0.0950") + long_ = _snap( + instrument="ETH-13MAY26-2300-P", + bid="0.0595", + ask="0.0605", + mid="0.0600", + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is True + assert res.estimated_slippage_pct_of_credit < Decimal("0.08") + + +def test_zero_credit_is_rejected(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("credit" in r for r in res.reasons) + + +def test_negative_credit_is_rejected(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("-0.001"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + + +def test_zero_or_negative_n_contracts_is_rejected(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + for n in (0, -1): + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=n, + cfg=cfg, + ) + assert res.accepted is False + assert any("contracts" in r for r in res.reasons) + + +def test_non_positive_mid_is_flagged(cfg: StrategyConfig) -> None: + short, long_ = _good_pair() + bad_short = _snap( + instrument=short.instrument, + bid="0", + ask="0", + mid="0", + open_interest=short.open_interest, + volume_24h=short.volume_24h, + book_depth_top3=short.book_depth_top3, + ) + res = check( + short_leg=bad_short, + long_leg=long_, + credit=Decimal("0.035"), + n_contracts=1, + cfg=cfg, + ) + assert res.accepted is False + assert any("mid price not positive" in r for r in res.reasons) + + +def test_failures_accumulate(cfg: StrategyConfig) -> None: + short = _snap(open_interest=10, volume_24h=2, book_depth_top3=1, bid="0.001", ask="0.500") + long_ = _snap( + instrument="ETH-13MAY26-2300-P", + open_interest=5, + volume_24h=1, + book_depth_top3=1, + bid="0.001", + ask="0.500", + ) + res = check( + short_leg=short, + long_leg=long_, + credit=Decimal("0.001"), + n_contracts=10, + cfg=cfg, + ) + assert res.accepted is False + # at least 4 categories per leg + slippage = many reasons + assert len(res.reasons) >= 5 diff --git a/tests/unit/test_sizing_engine.py b/tests/unit/test_sizing_engine.py new file mode 100644 index 0000000..f5825c4 --- /dev/null +++ b/tests/unit/test_sizing_engine.py @@ -0,0 +1,182 @@ +"""TDD for :mod:`cerbero_bite.core.sizing_engine`. + +Spec: ``docs/01-strategy-rules.md §5`` and ``docs/03-algorithms.md §3``. +The five mandatory test cases listed in §3 are reproduced here verbatim. +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from cerbero_bite.config import StrategyConfig, golden_config +from cerbero_bite.core.sizing_engine import SizingContext, compute_contracts + +# Standard fixture: EUR/USD = 1.075 (cap 200 EUR ≈ 215 USD, 1000 EUR ≈ 1075 USD) +EUR_USD = Decimal("1.075") + + +def _ctx( + *, + capital_usd: str = "1500", + max_loss_per_contract_usd: str = "93", + dvol_now: str = "40", + open_engagement_usd: str = "0", + eur_to_usd: Decimal = EUR_USD, + other_open_positions: int = 0, +) -> SizingContext: + return SizingContext( + capital_usd=Decimal(capital_usd), + max_loss_per_contract_usd=Decimal(max_loss_per_contract_usd), + dvol_now=Decimal(dvol_now), + open_engagement_usd=Decimal(open_engagement_usd), + eur_to_usd=eur_to_usd, + other_open_positions=other_open_positions, + ) + + +@pytest.fixture +def cfg() -> StrategyConfig: + return golden_config() + + +# --------------------------------------------------------------------------- +# Mandatory examples from docs/03-algorithms.md §3 +# --------------------------------------------------------------------------- + + +def test_minimum_capital_dvol_40_yields_one_contract(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(capital_usd="720", dvol_now="40"), cfg) + assert res.n_contracts == 1 + assert res.reason_if_zero is None + + +def test_capital_1500_dvol_50_yields_one_contract(cfg: StrategyConfig) -> None: + # 13% × 1500 = 195; adj 0.85 → 165.75; floor(165.75/93) = 1 + res = compute_contracts(_ctx(capital_usd="1500", dvol_now="50"), cfg) + assert res.n_contracts == 1 + + +def test_capital_1500_dvol_40_yields_two_contracts(cfg: StrategyConfig) -> None: + # 13% × 1500 = 195; cap 215 USD; adj 1.0 → 195; floor(195/93) = 2 + res = compute_contracts(_ctx(capital_usd="1500", dvol_now="40"), cfg) + assert res.n_contracts == 2 + + +def test_capital_5000_dvol_40_capped_at_two_contracts(cfg: StrategyConfig) -> None: + # 13% × 5000 = 650; cap 215 USD wins; floor(215/93) = 2 + res = compute_contracts(_ctx(capital_usd="5000", dvol_now="40"), cfg) + assert res.n_contracts == 2 + + +def test_capital_100000_dvol_40_capped_at_two_contracts(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(capital_usd="100000", dvol_now="40"), cfg) + assert res.n_contracts == 2 + + +# --------------------------------------------------------------------------- +# DVOL adjustment bands and no-entry threshold +# --------------------------------------------------------------------------- + + +def test_dvol_in_60_to_80_band_uses_0_65_multiplier(cfg: StrategyConfig) -> None: + # 13% × 5000 = 650; cap 215; adj 0.65 → 139.75; floor(139.75/93) = 1 + res = compute_contracts(_ctx(capital_usd="5000", dvol_now="65"), cfg) + assert res.n_contracts == 1 + + +def test_dvol_at_80_returns_zero(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(capital_usd="5000", dvol_now="80"), cfg) + assert res.n_contracts == 0 + assert res.reason_if_zero is not None + assert "dvol" in res.reason_if_zero.lower() + + +def test_dvol_above_no_entry_threshold_returns_zero(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(capital_usd="5000", dvol_now="85"), cfg) + assert res.n_contracts == 0 + + +# --------------------------------------------------------------------------- +# Aggregate engagement constraint +# --------------------------------------------------------------------------- + + +def test_aggregate_cap_reduces_contracts(cfg: StrategyConfig) -> None: + # cap_aggregate ≈ 1075 USD. With 1000 USD already engaged and ML=93, + # only floor((1075-1000)/93) = 0 new contracts should fit. + res = compute_contracts( + _ctx( + capital_usd="100000", + dvol_now="40", + open_engagement_usd="1000", + ), + cfg, + ) + assert res.n_contracts == 0 + assert res.reason_if_zero is not None + + +def test_aggregate_cap_partially_reduces(cfg: StrategyConfig) -> None: + # 800 already engaged → free 275; floor(275/93)=2 but also kelly cap may bind. + # 13% × 100000 = 13000; cap 215 → 215; floor(215/93)=2 contracts. + # 2*93 = 186; 186+800=986 ≤ 1075 ✓ → 2. + res = compute_contracts( + _ctx(capital_usd="100000", dvol_now="40", open_engagement_usd="800"), + cfg, + ) + assert res.n_contracts == 2 + + +def test_aggregate_cap_at_975_reduces_to_one(cfg: StrategyConfig) -> None: + # 975 engaged + 2*93 = 1161 > 1075 → drop to 1: 975 + 93 = 1068 ≤ 1075 ✓ + res = compute_contracts( + _ctx(capital_usd="100000", dvol_now="40", open_engagement_usd="975"), + cfg, + ) + assert res.n_contracts == 1 + + +# --------------------------------------------------------------------------- +# Concurrent positions guard +# --------------------------------------------------------------------------- + + +def test_concurrent_positions_at_cap_returns_zero(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(capital_usd="1500", other_open_positions=1), cfg) + assert res.n_contracts == 0 + assert res.reason_if_zero is not None + assert "position" in res.reason_if_zero.lower() + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +def test_below_one_contract_returns_zero(cfg: StrategyConfig) -> None: + # capital 600 → 13% * 600 = 78; floor(78/93)=0 → undersize + res = compute_contracts(_ctx(capital_usd="600", dvol_now="40"), cfg) + assert res.n_contracts == 0 + assert "undersize" in (res.reason_if_zero or "").lower() + + +def test_max_contracts_per_trade_cap_binds(cfg: StrategyConfig) -> None: + # very high capital + tiny ML → would compute > 4 → cap at 4. + res = compute_contracts( + _ctx( + capital_usd="100000", + max_loss_per_contract_usd="1", + dvol_now="40", + eur_to_usd=Decimal("100"), # cap 20000 USD per trade → no cap + ), + cfg, + ) + assert res.n_contracts == cfg.sizing.max_contracts_per_trade + + +def test_zero_max_loss_per_contract_returns_zero(cfg: StrategyConfig) -> None: + res = compute_contracts(_ctx(max_loss_per_contract_usd="0"), cfg) + assert res.n_contracts == 0 + assert res.reason_if_zero is not None