Files
Cerbero-Bite/tests/unit/test_combo_builder.py
root 1c6baaee83 feat(strategy): F+D+A miglioramenti — auto-pause, vol-harvest, delta dinamico
Implementa tre miglioramenti dalla roadmap di "📚 Strategia" + scaffolding del quarto.
Tutti retro-compatibili: i defaults della golden config disabilitano le nuove funzioni
così il comportamento attuale resta invariato finché l'operatore non le accende
esplicitamente in `strategy.yaml`. Il profilo `strategy.aggressiva.yaml` opta-in
agli incrementi più impattanti.

**F — Auto-pause su drawdown rolling (§7-bis)**

Circuit breaker sopra il kill-switch tecnico. Quando le ultime N posizioni
chiuse hanno cumulato perdite oltre `max_drawdown_pct × capitale_attuale`,
l'engine si auto-mette in pausa per `pause_weeks` settimane. Difende dai
regime change non rilevati dai filtri quant — se i filtri stanno fallendo
sistematicamente, fermarsi è meglio che continuare a sanguinare.

- `AutoPauseConfig` + `cfg.auto_pause` (top-level, default disabled).
- Migrazione SQL `0004_auto_pause.sql`: `system_state.auto_pause_until`
  e `auto_pause_reason` (NULL = engine attivo).
- Nuovo modulo puro `runtime/auto_pause.py` con `is_paused()` (gate I/O-free)
  e `evaluate_drawdown_breach()` (decide se armare).
- `entry_cycle` consulta `is_paused` subito dopo il kill-switch e arma
  la pausa dopo aver calcolato il capitale; nuovo status `_STATUS_AUTO_PAUSED`.
- Repository: `set_auto_pause`, `recent_closed_position_pnls_usd`.
- 12 test unitari: gate filter on/off, lookback insufficiente, soglia
  esatta, capitale non valido, transizioni paused → not-paused.

**D — Vol-collapse harvest (§7-bis)**

Exit opportunistica: quando DVOL è scesa di tot punti rispetto all'entry
e siamo in profit, esce subito. Edge IV-RV catturato, non c'è motivo di
tenere fino al profit-take. Nuovo `ExitAction = "CLOSE_VOL_HARVEST"`,
gate `exit.vol_harvest_dvol_decrease` (default 0 = off). 5 test unitari.

**A — Delta target dinamico per regime DVOL (§3.2)**

Strike short adattivo alla volatilità: a DVOL bassa il margine OTM è
generoso ⇒ posso prendere più premio (delta 0.15); a DVOL alta voglio
più safety distance (delta 0.10). Nuovo `DeltaByDvolBand` (step
function); quando `delta_by_dvol` è popolato, `_select_short` legge
la prima banda ascending con `dvol_now ≤ dvol_under`. Default vuoto =
comportamento invariato. `select_strikes` accetta nuovo kwarg
`dvol_now`, propagato da `entry_cycle`. 4 test unitari.

**C — Scaffolding profit-take graduale (§7.1bis)**

Schema in place ma runtime non ancora wirato. Aggiunge `PartialProfitLevel`
e `exit.profit_take_partial_levels` (default vuoto). Nuovo
`ExitAction = "CLOSE_PROFIT_PARTIAL"` nella Literal. La pipeline di
chiusure parziali nel runtime (entry_cycle / repository / clients)
richiede refactor del position model — lasciato come TODO per un PR
dedicato. La schema è pronta a recepire la config futura senza altri
breaking change.

**Profili aggiornati**

- `strategy.yaml` (golden, 1.2.0): tutto disabilitato by default.
- `strategy.conservativa.yaml` (1.2.0-cons): identico al golden.
- `strategy.aggressiva.yaml` (1.2.0-aggr): A+D+F enabled
  (delta_by_dvol 0.15/0.12/0.10, vol_harvest a 15 pt vol,
  auto_pause @ 15% DD su 5 trade, 2 settimane pausa).

Bump versioni 1.1.0 → 1.2.0, hash ricalcolati, test pinning aggiornato.

Suite: 426 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 20:07:25 +00:00

475 lines
16 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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"
# ---------------------------------------------------------------------------
# §3.2 (A): dynamic delta target by DVOL regime
# ---------------------------------------------------------------------------
def _cfg_with_delta_bands(cfg: StrategyConfig) -> StrategyConfig:
"""Profilo con step-function delta su DVOL.
Vol bassa (≤50) → delta 0.15 (più premio), vol media (≤70) →
0.12 (default), vol alta (≤90) → 0.10 (più safety distance).
"""
from cerbero_bite.config.schema import (
DeltaByDvolBand,
ShortStrikeSpec,
StructureConfig,
)
bands = [
DeltaByDvolBand(
dvol_under=Decimal("50"),
delta_target=Decimal("0.15"),
delta_min=Decimal("0.13"),
delta_max=Decimal("0.17"),
),
DeltaByDvolBand(
dvol_under=Decimal("70"),
delta_target=Decimal("0.12"),
delta_min=Decimal("0.10"),
delta_max=Decimal("0.15"),
),
DeltaByDvolBand(
dvol_under=Decimal("90"),
delta_target=Decimal("0.10"),
delta_min=Decimal("0.08"),
delta_max=Decimal("0.12"),
),
]
new_short = ShortStrikeSpec(
**{**cfg.structure.short_strike.model_dump(), "delta_by_dvol": bands}
)
return cfg.model_copy(
update={
"structure": StructureConfig(
**{**cfg.structure.model_dump(exclude={"short_strike"}),
"short_strike": new_short}
)
}
)
def _bull_put_chain_wide(now_dt: datetime) -> list[OptionQuote]:
"""Chain con shorts e longs per delta 0.10, 0.12, 0.15.
I mid sono tarati per superare il credit/width ≥ 30% per ogni
accoppiamento short→long testato (vedi commento §3.4).
"""
return [
# Shorts a delta 0.10 / 0.12 / 0.15 in OTM range [15-25%].
_quote(strike="2535", delta="-0.15", mid="0.026", now_dt=now_dt),
_quote(strike="2475", delta="-0.12", mid="0.020", now_dt=now_dt),
_quote(strike="2400", delta="-0.10", mid="0.015", now_dt=now_dt),
# Long candidati ~4% sotto ciascuno short.
_quote(strike="2415", delta="-0.10", mid="0.012", now_dt=now_dt),
_quote(strike="2355", delta="-0.08", mid="0.006", now_dt=now_dt),
_quote(strike="2280", delta="-0.06", mid="0.002", now_dt=now_dt),
]
def test_dynamic_delta_low_dvol_picks_higher_delta(
cfg: StrategyConfig, now: datetime
) -> None:
"""DVOL=40 → banda con delta_target=0.15."""
cfg_dyn = _cfg_with_delta_bands(cfg)
chain = _bull_put_chain_wide(now)
res = select_strikes(
chain=chain,
bias="bull_put",
spot=Decimal("3000"),
now=now,
cfg=cfg_dyn,
dvol_now=Decimal("40"),
)
assert res is not None
short, _ = res
assert short.delta == Decimal("-0.15")
def test_dynamic_delta_mid_dvol_picks_default_delta(
cfg: StrategyConfig, now: datetime
) -> None:
"""DVOL=60 → banda con delta_target=0.12."""
cfg_dyn = _cfg_with_delta_bands(cfg)
chain = _bull_put_chain_wide(now)
res = select_strikes(
chain=chain,
bias="bull_put",
spot=Decimal("3000"),
now=now,
cfg=cfg_dyn,
dvol_now=Decimal("60"),
)
assert res is not None
short, _ = res
assert short.delta == Decimal("-0.12")
def test_dynamic_delta_high_dvol_picks_lower_delta(
cfg: StrategyConfig, now: datetime
) -> None:
"""DVOL=85 → banda con delta_target=0.10 (più safety distance)."""
cfg_dyn = _cfg_with_delta_bands(cfg)
chain = _bull_put_chain_wide(now)
res = select_strikes(
chain=chain,
bias="bull_put",
spot=Decimal("3000"),
now=now,
cfg=cfg_dyn,
dvol_now=Decimal("85"),
)
assert res is not None
short, _ = res
assert short.delta == Decimal("-0.10")
def test_dynamic_delta_disabled_default_uses_static_delta(
cfg: StrategyConfig, now: datetime
) -> None:
"""delta_by_dvol vuoto (default) → comportamento invariato."""
chain = _bull_put_chain_wide(now)
res = select_strikes(
chain=chain,
bias="bull_put",
spot=Decimal("3000"),
now=now,
cfg=cfg, # golden config: delta_by_dvol=[]
dvol_now=Decimal("40"),
)
assert res is not None
short, _ = res
# Delta target statico = 0.12, quindi torna lo strike a -0.12.
assert short.delta == Decimal("-0.12")