root
6ff021fbf4
Crypto opera 24/7: la cadenza settimanale lunedì-only era un retaggio TradFi senza giustificazione. La nuova cadenza è giornaliera (cron 0 14 * * *), con i gate quantitativi a decidere se entrare o saltare. Cambiamenti principali: * runtime/orchestrator.py — _CRON_ENTRY 0 14 * * * (era MON) * runtime/auto_pause.py — pause_until(days=) (era weeks=); minimo clamp 1 giorno (era 1 settimana) * core/backtest.py — MondayPick→DailyPick, monday_picks→daily_picks (1 pick per calendar-day all'ora target); Sharpe annualization su ~120 trade/anno (era 52) * config/schema.py — default cron daily; max_concurrent_positions 1→5; AutoPauseConfig.pause_weeks→pause_days, default 14 * runtime/option_chain_snapshot_cycle.py + orchestrator — cron */15 per accumulo continuo dataset di backtest empirico Strategy yamls (config_version 1.3.0 → 1.4.0, hash rigenerati): * strategy.yaml — max_concurrent 1→5, cap_aggregate coerente * strategy.aggressiva.yaml — max_concurrent 2→8, cap_aggregate 3200→6400, max_contracts_per_trade invariato a 16 * strategy.conservativa.yaml — max_concurrent 1→3 * tutti — pause_weeks→pause_days: 14 GUI (pages/7_📚_Strategia.py): * slider Trade/anno: range 20-200 (era 8-30), default 110, help riallineato sulla math 365 candidature × pass-rate 30-40% * card profili: versione letta dinamicamente da config_version invece che hard-coded "v1.2.0" * warning "entrambi perdono soldi" ora valuta i P/L effettivi (cons['annual_pl'], aggr['annual_pl']) invece del win_rate grezzo; aggiunto stato intermedio quando solo conservativo è in perdita Tests (450/450 passati): * test_auto_pause: pause_days, clamp ≥1 giorno * test_backtest: rinomina + ridisegno daily picks (assert su calendar-day dedupe e hour filter) * test_sizing_engine: other_open_positions=5 per cap default * test_config_loader: version 1.4.0 Docs (README + 9 file in docs/) — tutti i riferimenti weekly/lunedì allineati a daily/24-7, volume option_chain ricalcolato per cron */15 (~1.1 MB/giorno, ~400 MB/anno). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
158 lines
4.8 KiB
Python
158 lines
4.8 KiB
Python
"""TDD per :mod:`cerbero_bite.runtime.auto_pause` (§7-bis F)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime, timedelta
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
|
||
from cerbero_bite.config.schema import AutoPauseConfig
|
||
from cerbero_bite.runtime.auto_pause import (
|
||
evaluate_drawdown_breach,
|
||
is_paused,
|
||
pause_until,
|
||
)
|
||
from cerbero_bite.state.models import SystemStateRecord
|
||
|
||
|
||
_NOW = datetime(2026, 5, 1, 14, 0, tzinfo=UTC)
|
||
|
||
|
||
def _state(**overrides: object) -> SystemStateRecord:
|
||
base: dict[str, object] = {
|
||
"kill_switch": 0,
|
||
"last_health_check": _NOW,
|
||
"config_version": "1.0.0",
|
||
"started_at": _NOW - timedelta(hours=1),
|
||
}
|
||
base.update(overrides)
|
||
return SystemStateRecord(**base) # type: ignore[arg-type]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# is_paused
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_is_paused_returns_false_when_state_is_none() -> None:
|
||
status = is_paused(None, now=_NOW)
|
||
assert status.paused is False
|
||
|
||
|
||
def test_is_paused_returns_false_when_until_is_none() -> None:
|
||
status = is_paused(_state(), now=_NOW)
|
||
assert status.paused is False
|
||
|
||
|
||
def test_is_paused_returns_true_when_until_in_future() -> None:
|
||
status = is_paused(
|
||
_state(auto_pause_until=_NOW + timedelta(weeks=2),
|
||
auto_pause_reason="DD breach"),
|
||
now=_NOW,
|
||
)
|
||
assert status.paused is True
|
||
assert status.reason == "DD breach"
|
||
|
||
|
||
def test_is_paused_returns_false_when_until_in_past() -> None:
|
||
status = is_paused(
|
||
_state(auto_pause_until=_NOW - timedelta(seconds=1)),
|
||
now=_NOW,
|
||
)
|
||
assert status.paused is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# pause_until
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_pause_until_adds_days() -> None:
|
||
until = pause_until(_NOW, days=14)
|
||
assert until == _NOW + timedelta(days=14)
|
||
|
||
|
||
def test_pause_until_clamps_to_one_day_minimum() -> None:
|
||
# days <= 0 deve cmq dare almeno 1 giorno di pausa, altrimenti
|
||
# la cron giornaliera potrebbe scattare comunque.
|
||
assert pause_until(_NOW, days=0) == _NOW + timedelta(days=1)
|
||
assert pause_until(_NOW, days=-3) == _NOW + timedelta(days=1)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# evaluate_drawdown_breach
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _cfg(**overrides: object) -> AutoPauseConfig:
|
||
base: dict[str, object] = {
|
||
"enabled": True,
|
||
"lookback_trades": 5,
|
||
"max_drawdown_pct": Decimal("0.10"),
|
||
"pause_days": 14,
|
||
}
|
||
base.update(overrides)
|
||
return AutoPauseConfig(**base) # type: ignore[arg-type]
|
||
|
||
|
||
def test_drawdown_breach_when_enabled_and_threshold_exceeded() -> None:
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(),
|
||
recent_pnl_usd=[Decimal("-50"), Decimal("-60"), Decimal("-40"),
|
||
Decimal("-30"), Decimal("-20")], # cum −200 USD
|
||
capital_usd=Decimal("1500"),
|
||
)
|
||
# |200| / 1500 = 0.133 > 0.10
|
||
assert decision.should_pause is True
|
||
assert decision.reason is not None
|
||
assert "rolling DD" in decision.reason
|
||
|
||
|
||
def test_no_breach_when_filter_disabled() -> None:
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(enabled=False),
|
||
recent_pnl_usd=[Decimal("-200")] * 5, # massacro
|
||
capital_usd=Decimal("1500"),
|
||
)
|
||
assert decision.should_pause is False
|
||
|
||
|
||
def test_no_breach_when_lookback_insufficient() -> None:
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(lookback_trades=5),
|
||
recent_pnl_usd=[Decimal("-100")] * 3, # solo 3 trade, serve 5
|
||
capital_usd=Decimal("1500"),
|
||
)
|
||
assert decision.should_pause is False
|
||
|
||
|
||
def test_no_breach_when_cumulative_positive() -> None:
|
||
# Anche con tante perdite, se la somma è positiva non scattiamo.
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(),
|
||
recent_pnl_usd=[Decimal("-100"), Decimal("-50"),
|
||
Decimal("300"), Decimal("-20"), Decimal("-10")],
|
||
capital_usd=Decimal("1500"),
|
||
)
|
||
assert decision.should_pause is False
|
||
|
||
|
||
def test_no_breach_when_below_threshold() -> None:
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(),
|
||
recent_pnl_usd=[Decimal("-30")] * 5, # cum −150 / 1500 = 10% esatto
|
||
capital_usd=Decimal("1500"),
|
||
)
|
||
# esattamente alla soglia (>=) ⇒ pausa armata
|
||
assert decision.should_pause is True
|
||
|
||
|
||
def test_no_breach_when_capital_zero_or_negative() -> None:
|
||
decision = evaluate_drawdown_breach(
|
||
cfg=_cfg(),
|
||
recent_pnl_usd=[Decimal("-100")] * 5,
|
||
capital_usd=Decimal("0"),
|
||
)
|
||
assert decision.should_pause is False
|