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) <noreply@anthropic.com>
This commit is contained in:
2026-04-27 10:14:06 +02:00
parent 881bc8a1bf
commit fbb7753cc6
20 changed files with 3090 additions and 1 deletions
+155
View File
@@ -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")