Adriano
f4faef6fd1
Integra due nuovi filtri dal pacchetto quant indicators rilasciato in Cerbero_mcp (commit a13e3fe). 335 test pass, mypy strict pulito, ruff clean. Filtri (§2.8 — nuovo): - dealer-gamma: blocca entry quando total_net_dealer_gamma < dealer_gamma_min (default 0). Long-gamma regime favorisce credit spread (vol-suppressing dealer flow); short-gamma flow lo amplifica ed è da evitare. - liquidation-heatmap: blocca entry quando il segnale euristico di cerbero-sentiment riporta long o short squeeze risk = "high" (cluster di liquidations imminenti entro 24h). Entrambi sono best-effort: se il tool MCP fallisce o restituisce dati anomali l'entry_cycle popola EntryContext con None e validate_entry salta il gate per non bloccare entry su problemi infrastrutturali. Wrapper: - DeribitClient.dealer_gamma_profile_eth → DealerGammaSnapshot. - SentimentClient.liquidation_heatmap → LiquidationHeatmap con property has_high_squeeze_risk. Schema: - EntryConfig.dealer_gamma_min, dealer_gamma_filter_enabled, liquidation_filter_enabled. - EntryContext.dealer_net_gamma, liquidation_squeeze_risk_high opzionali. - strategy.yaml: nuovi campi documentati con commento + hash ricalcolato (4c2be4c5...). Documentazione: - docs/04-mcp-integration.md riscritto al modello attuale (HTTP REST, no mcp SDK, no memory/brain-bridge, place_combo_order documentato, environment_info al boot). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""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 EntryConfig, 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_dealer_short_gamma_blocks_entry(cfg: StrategyConfig) -> None:
|
|
decision = validate_entry(_good_ctx(dealer_net_gamma=Decimal("-5")), cfg)
|
|
assert decision.accepted is False
|
|
assert any("dealer short-gamma" in r for r in decision.reasons)
|
|
|
|
|
|
def test_dealer_long_gamma_passes(cfg: StrategyConfig) -> None:
|
|
decision = validate_entry(_good_ctx(dealer_net_gamma=Decimal("100")), cfg)
|
|
assert decision.accepted is True
|
|
|
|
|
|
def test_dealer_gamma_none_skips_filter(cfg: StrategyConfig) -> None:
|
|
decision = validate_entry(_good_ctx(dealer_net_gamma=None), cfg)
|
|
assert decision.accepted is True
|
|
|
|
|
|
def test_liquidation_squeeze_high_blocks_entry(cfg: StrategyConfig) -> None:
|
|
decision = validate_entry(
|
|
_good_ctx(liquidation_squeeze_risk_high=True), cfg
|
|
)
|
|
assert decision.accepted is False
|
|
assert any("liquidation squeeze" in r for r in decision.reasons)
|
|
|
|
|
|
def test_liquidation_squeeze_filter_disabled_in_config(
|
|
cfg: StrategyConfig,
|
|
) -> None:
|
|
permissive = golden_config(
|
|
entry=EntryConfig(
|
|
**{**cfg.entry.model_dump(), "liquidation_filter_enabled": False}
|
|
)
|
|
)
|
|
decision = validate_entry(
|
|
_good_ctx(liquidation_squeeze_risk_high=True), permissive
|
|
)
|
|
assert decision.accepted is True
|
|
|
|
|
|
def test_dealer_gamma_filter_disabled_in_config(cfg: StrategyConfig) -> None:
|
|
permissive = golden_config(
|
|
entry=EntryConfig(
|
|
**{**cfg.entry.model_dump(), "dealer_gamma_filter_enabled": False}
|
|
)
|
|
)
|
|
decision = validate_entry(
|
|
_good_ctx(dealer_net_gamma=Decimal("-1000")), permissive
|
|
)
|
|
assert decision.accepted is True
|
|
|
|
|
|
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
|