2 Commits

Author SHA1 Message Date
root f664ea1a15 feat(backtest): stylized engine over market_snapshots + CLI subcommand
Aggiunge `core/backtest.py`, motore di backtesting stilizzato che gira
sui dati raccolti in `market_snapshots`. Risponde alla domanda:
"se questa config fosse stata attiva nelle ultime N settimane, quanti
lunedì avrebbero superato i filtri e quale sarebbe stato il P/L stimato?"

**Architettura a due strati**:

1. **Filtri di entry — RIGOROSO**: per ogni Monday-14:00-UTC nei
   snapshot ricostruisce `EntryContext` e chiama lo stesso
   `validate_entry()` del live. Output esatto di "cosa avrebbe deciso
   il bot" per ogni settimana, con conteggio dei motivi di skip.

2. **P/L per trade accettato — STILIZZATO**: senza catena opzioni
   storica, stima credito/exit via Black-Scholes con skew premium
   (default 1.5×) per approssimare la vol smile dell'ETH. Re-prezza
   il combo ad ogni tick futuro per simulare i trigger §7
   (profit_take, stop_loss, vol_stop, time_stop, expiry).

**Aggregati nel `BacktestReport`**:
- n_picks / n_accepted / n_skipped_data / n_completed / n_winners
- win_rate, P/L cumulato (USD + % su capitale)
- max drawdown (USD + % di peak)
- Sharpe annualizzato (52 settimane)
- skip_reasons: dict{motivo → settimane bloccate}

**CLI**: nuovo `cerbero-bite backtest --strategy F --from D --to D
--capital N --asset ETH`. Stampa Rich-formatted summary + tabella
motivi di skip. Esempio:

    cerbero-bite backtest \
      --strategy strategy.aggressiva.yaml \
      --from 2026-04-01 --to 2026-05-01 \
      --capital 10000

**Limiti dichiarati**:
- BS + skew_premium ≠ catena reale: i numeri P/L sono **stime ex-post
  per ranking config**, non promesse operative. Buono per dire
  "config A batte config B sui dati reali", non per dimensionare
  capitale.
- skew_premium 1.5× è stato calibrato sui dati Deribit storici
  (smile slope ETH options); va rifinito quando avremo abbastanza
  chain history da farlo empiricamente.

**Tests**: 15 unit test (BS math, monday picks, filter sim,
position outcome simulation, full pipeline su sintetico).
Suite totale: 420 passed.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 20:31:54 +00:00
root 21e865ffb0 feat(gui+infra): pagina Strategia, P/L parametrico, profili Conservativa/Aggressiva, dashboard via Traefik
Espone la GUI Streamlit su https://cerbero-bite.tielogic.xyz tramite il
Traefik già attivo sull'host (label allineate al pattern di cerbero-mcp,
TLS via Let's Encrypt, websocket pass-through). Aggiunge:

- nuova tab `📚 Strategia` con stato live dei gate §2 confrontati con
  l'ultimo tick di market_snapshots, pannello P/L parametrico
  affiancato Conservativa vs Aggressiva, tabella di sensibilità
  win-rate → APR e rendering del documento canonico esteso.
- doc `13-strategia-spiegata.md` che lega ogni regola §2-§9 al campo di
  market_snapshots che la alimenta, con sezioni §4-bis (P/L atteso
  realistico, win-rate empirico, drawdown, Sharpe) e §4-ter (confronto
  fra i due profili e quando passare dall'uno all'altro).
- `strategy.conservativa.yaml` (golden config v1.0.0 esplicita) e
  `strategy.aggressiva.yaml` (cap_per_trade 4×, max_concurrent 2×,
  max_contracts 4×, deroga §11 documentata) con config_hash validi.
- nel compose: servizio dedicato `cerbero-bite-gui` (Streamlit su
  0.0.0.0:8765, healthcheck su /_stcore/health, label Traefik), env
  condivisi via anchor YAML `x-bite-env`, `--environment mainnet`
  passato a `start` per allineare il boot check al token del .env (era
  testnet vs mainnet → kill switch armato all'avvio).
- Dockerfile installa anche l'extra `gui` (streamlit) e copia
  `docs/` + i due nuovi profili nell'immagine; `.dockerignore` non
  esclude più `docs/` (causa del primo build silenzioso).

Fix bonus: `_try_load` nella pagina ritornava `LoadedConfig` ma la GUI
leggeva `.sizing.*` direttamente — l'`except: pass` mascherava
l'AttributeError facendo cadere sui default conservativi sia nel
pannello P/L sia nello stato gate (stesso pattern presente nella
Calibrazione). Ora ritorna `.config`.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 18:20:23 +00:00
10 changed files with 2743 additions and 29 deletions
-1
View File
@@ -5,7 +5,6 @@
.pytest_cache/ .pytest_cache/
__pycache__/ __pycache__/
data/ data/
docs/
tests/ tests/
.coverage .coverage
htmlcov/ htmlcov/
+8 -2
View File
@@ -14,12 +14,12 @@ ENV UV_PROJECT_ENVIRONMENT=/opt/venv \
# Install only the dependencies first so the layer is cached when the # Install only the dependencies first so the layer is cached when the
# source tree changes. # source tree changes.
COPY pyproject.toml uv.lock ./ COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project RUN uv sync --frozen --no-dev --no-install-project --extra gui
# Now copy the source tree and install the project itself. # Now copy the source tree and install the project itself.
COPY src ./src COPY src ./src
COPY README.md ./ COPY README.md ./
RUN uv sync --frozen --no-dev RUN uv sync --frozen --no-dev --extra gui
FROM python:3.13-slim AS runtime FROM python:3.13-slim AS runtime
@@ -40,6 +40,12 @@ COPY --from=builder /opt/venv /opt/venv
COPY --from=builder /app/src /app/src COPY --from=builder /app/src /app/src
COPY scripts /app/scripts COPY scripts /app/scripts
COPY strategy.yaml /app/strategy.yaml COPY strategy.yaml /app/strategy.yaml
# Profili alternativi confrontati nella pagina "📚 Strategia".
COPY strategy.conservativa.yaml /app/strategy.conservativa.yaml
COPY strategy.aggressiva.yaml /app/strategy.aggressiva.yaml
# Documentation is shipped at runtime so the Streamlit "Strategia"
# page can render the canonical strategy doc directly.
COPY docs /app/docs
# Persistent state + audit go into /app/data, mounted as a volume in # Persistent state + audit go into /app/data, mounted as a volume in
# docker-compose.yml. # docker-compose.yml.
+90 -25
View File
@@ -1,27 +1,48 @@
# docker-compose.yml — Cerbero Bite # docker-compose.yml — Cerbero Bite
# #
# Bite runs in its own Compose project but joins the same Docker # Bite runs in its own Compose project but joins the same Docker
# network used by Cerbero MCP V2 so it can resolve the in-cluster # network used by Cerbero MCP V2 and Traefik (`traefik`) so it can
# service name when running co-located, and otherwise reaches the # either resolve the in-cluster service name (`cerbero-mcp:9000`)
# public gateway (`https://cerbero-mcp.tielogic.xyz`) over the host # or reach the public gateway (`https://cerbero-mcp.tielogic.xyz`)
# network. # transparently.
# #
# The shared network is declared as external here. Create it once on # The reverse-proxy network (`traefik`) is declared as external
# the host with `docker network create cerbero-suite` (or rename the # here. It is created by the Traefik stack at /opt/docker/traefik
# Cerbero_mcp network to `cerbero-suite` and mark it external). # and shared by every web-facing service on the host.
# #
# Authentication: a single bearer token is passed through from the # Authentication: a single bearer token is passed through from the
# host `.env` file via `CERBERO_BITE_MCP_TOKEN`. The Cerbero MCP V2 # host `.env` file via `CERBERO_BITE_MCP_TOKEN`. The Cerbero MCP V2
# server uses the token to decide whether the upstream environment # server uses the token to decide whether the upstream environment
# is testnet or mainnet; switching environment = switching token. # is testnet or mainnet; switching environment = switching token.
#
# Two services are defined:
# * `cerbero-bite` — the trading engine / CLI worker
# * `cerbero-bite-gui` — the Streamlit dashboard, exposed by
# Traefik at https://cerbero-bite.<DOMAIN>
networks: networks:
cerbero-suite: traefik:
external: true external: true
volumes: volumes:
bite-data: bite-data:
x-bite-env: &bite-env
CERBERO_BITE_MCP_TOKEN: ${CERBERO_BITE_MCP_TOKEN:?missing CERBERO_BITE_MCP_TOKEN}
CERBERO_BITE_MCP_BOT_TAG: ${CERBERO_BITE_MCP_BOT_TAG:-BOT__CERBERO_BITE}
# Two independent runtime flags that decide what each cycle does.
# Initial period ("data-only"): DATA_ANALYSIS=true, STRATEGY=false.
CERBERO_BITE_ENABLE_DATA_ANALYSIS: ${CERBERO_BITE_ENABLE_DATA_ANALYSIS:-true}
CERBERO_BITE_ENABLE_STRATEGY: ${CERBERO_BITE_ENABLE_STRATEGY:-false}
# Service URLs — defaults below match the in-cluster Traefik network
# DNS (V2 unified image listening on port 9000). Override any of
# them via .env to point at the public gateway, a custom host, or
# localhost for dev work.
CERBERO_BITE_MCP_DERIBIT_URL: ${CERBERO_BITE_MCP_DERIBIT_URL:-http://cerbero-mcp:9000/mcp-deribit}
CERBERO_BITE_MCP_HYPERLIQUID_URL: ${CERBERO_BITE_MCP_HYPERLIQUID_URL:-http://cerbero-mcp:9000/mcp-hyperliquid}
CERBERO_BITE_MCP_MACRO_URL: ${CERBERO_BITE_MCP_MACRO_URL:-http://cerbero-mcp:9000/mcp-macro}
CERBERO_BITE_MCP_SENTIMENT_URL: ${CERBERO_BITE_MCP_SENTIMENT_URL:-http://cerbero-mcp:9000/mcp-sentiment}
services: services:
cerbero-bite: cerbero-bite:
build: build:
@@ -29,24 +50,12 @@ services:
dockerfile: Dockerfile dockerfile: Dockerfile
image: cerbero-bite:dev image: cerbero-bite:dev
restart: unless-stopped restart: unless-stopped
networks: [cerbero-suite] networks: [traefik]
cap_drop: [ALL] cap_drop: [ALL]
security_opt: security_opt:
- no-new-privileges:true - no-new-privileges:true
environment: environment:
# MCP auth — token is sourced from the host .env (compose <<: *bite-env
# interpolation). The `X-Bot-Tag` value below is the audit
# identifier the MCP server logs for every write call.
CERBERO_BITE_MCP_TOKEN: ${CERBERO_BITE_MCP_TOKEN:?missing CERBERO_BITE_MCP_TOKEN}
CERBERO_BITE_MCP_BOT_TAG: ${CERBERO_BITE_MCP_BOT_TAG:-BOT__CERBERO_BITE}
# Service URLs — defaults below match the in-cluster cerbero-suite
# network DNS (V2 unified image listening on port 9000). Override
# any of them to point at the public gateway, a custom host, or
# localhost for dev work.
CERBERO_BITE_MCP_DERIBIT_URL: ${CERBERO_BITE_MCP_DERIBIT_URL:-http://cerbero-mcp:9000/mcp-deribit}
CERBERO_BITE_MCP_HYPERLIQUID_URL: ${CERBERO_BITE_MCP_HYPERLIQUID_URL:-http://cerbero-mcp:9000/mcp-hyperliquid}
CERBERO_BITE_MCP_MACRO_URL: ${CERBERO_BITE_MCP_MACRO_URL:-http://cerbero-mcp:9000/mcp-macro}
CERBERO_BITE_MCP_SENTIMENT_URL: ${CERBERO_BITE_MCP_SENTIMENT_URL:-http://cerbero-mcp:9000/mcp-sentiment}
# Telegram and Portfolio are no longer shared MCP services. The # Telegram and Portfolio are no longer shared MCP services. The
# bot now calls the Telegram Bot API directly and aggregates # bot now calls the Telegram Bot API directly and aggregates
# portfolio in-process from Deribit + Hyperliquid + Macro. # portfolio in-process from Deribit + Hyperliquid + Macro.
@@ -62,6 +71,62 @@ services:
timeout: 5s timeout: 5s
retries: 3 retries: 3
start_period: 120s start_period: 120s
# Default command runs the engine status check; override with the # Engine main loop (scheduler + monitoring). Switch to `status`,
# CLI subcommand of choice (start, ping, dry-run, ...). # `ping`, `dry-run`, ... for one-shot diagnostics. The MCP token in
command: ["status"] # `.env` decides the upstream environment server-side; the `start`
# flag below tells the local boot check what to expect (must match,
# otherwise the engine arms the kill switch).
command: ["start", "--environment", "mainnet"]
# Streamlit dashboard published by Traefik on
# https://cerbero-bite.${DOMAIN_NAME:-tielogic.xyz}
#
# The CLI sub-command `cerbero-bite gui` hard-codes the listen
# address to 127.0.0.1, so we bypass the entrypoint and invoke
# Streamlit directly. The two `CERBERO_BITE_GUI_*` env vars match
# what the CLI normally injects (see src/cerbero_bite/cli.py).
cerbero-bite-gui:
image: cerbero-bite:dev
restart: unless-stopped
depends_on:
- cerbero-bite
networks: [traefik]
cap_drop: [ALL]
security_opt:
- no-new-privileges:true
environment:
<<: *bite-env
CERBERO_BITE_GUI_DB: /app/data/state.sqlite
CERBERO_BITE_GUI_AUDIT: /app/data/log/audit.jsonl
volumes:
- bite-data:/app/data
entrypoint:
- python
- -m
- streamlit
- run
- /app/src/cerbero_bite/gui/main.py
- --server.address=0.0.0.0
- --server.port=8765
- --server.headless=true
- --browser.gatherUsageStats=false
command: []
healthcheck:
test:
- "CMD"
- "python"
- "-c"
- "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8765/_stcore/health', timeout=3).close()"
interval: 30s
timeout: 5s
retries: 3
start_period: 30s
labels:
- traefik.enable=true
- traefik.docker.network=traefik
- "traefik.http.routers.cerbero-bite.rule=Host(`cerbero-bite.${DOMAIN_NAME:-tielogic.xyz}`)"
- traefik.http.routers.cerbero-bite.tls=true
- traefik.http.routers.cerbero-bite.entrypoints=websecure
- traefik.http.routers.cerbero-bite.tls.certresolver=mytlschallenge
- traefik.http.services.cerbero-bite.loadbalancer.server.port=8765
- com.centurylinklabs.watchtower.enable=true
+592
View File
@@ -0,0 +1,592 @@
# 13 — Strategia spiegata: dalle regole ai dati
> Documento operativo che lega ogni decisione del rule engine al dato
> osservabile da cui dipende. Pensato per chi guarda il cruscotto e
> vuole capire **a cosa servono** le metriche raccolte ogni 15 minuti
> nella tabella `market_snapshots`. La versione canonica e immutabile
> delle regole resta in `01-strategy-rules.md`; questo documento è la
> guida descrittiva da leggere prima di toccare le soglie in
> `strategy.yaml`.
---
## TL;DR
Cerbero Bite vende **credit spread settimanali su ETH/Deribit** quando
la volatilità implicita è **abbastanza alta da pagare bene**, il
mercato non è in **stress di liquidazione**, non ci sono **eventi macro
forti** in finestra, e il bias direzionale è **chiaro** (bull o bear).
Tutto il resto del tempo, l'engine **non opera**: la disciplina è la
strategia.
Ogni 15 minuti raccoglie 1 riga per asset (ETH e BTC) nella tabella
`market_snapshots`. Quei dati alimentano tre obiettivi distinti:
1. **Decisione live** — l'entry ciclo del lunedì 14:00 UTC legge i
campi più freschi per dire "go/no-go".
2. **Monitoring continuo** — il decision loop di gestione attiva
confronta la situazione con quella all'apertura.
3. **Calibrazione** — la pagina `📐 Calibrazione` usa la distribuzione
storica di ciascun campo per scegliere soglie basate sui percentili
reali del proprio ambiente, non a istinto.
---
## 1. Cosa c'è in `market_snapshots` (1 riga ogni 15 min, per asset)
| Campo | Unità | Sorgente MCP | A che serve nella strategia |
|---|---|---|---|
| `timestamp` | UTC ISO | scheduler | indicizzazione della time-series |
| `asset` | ETH / BTC | scheduler | partizionamento (ETH = sottostante operativo, BTC = controllo macro) |
| `spot` | USD | mcp-deribit `spot_perp_price` | trend 30g (§3.1), distanza % strike (§3.2/3.3), context generale |
| `dvol` | indice 0200 | mcp-deribit `latest_dvol` | gate entry §2.3-§2.4 (35 ≤ DVOL ≤ 90), aggiustamento sizing §5.3, vol-stop §7.3 |
| `realized_vol_30d` | % annualizzata | mcp-deribit `realized_vol` | confronto con DVOL → mean-reversion edge |
| `iv_minus_rv` | punti vol | derivato | richness della IV: > 0 = premio "ricco" da vendere |
| `funding_perp_annualized` | frazione | mcp-hyperliquid `funding_rate_annualized` | gate entry §2.6 (\|f\| ≤ 80% annualizzato), bias §3.1 |
| `funding_cross_annualized` | frazione | mcp-sentiment `funding_cross_median_annualized` | bias direzionale §3.1 (mediana 4 maggiori exchange) |
| `dealer_net_gamma` | USD | mcp-deribit `dealer_gamma_profile` | filtro quant §2.8 (long-gamma regime sopprime la vol → ideale per vendere spread) |
| `gamma_flip_level` | USD | mcp-deribit `dealer_gamma_profile` | livello spot oltre il quale il regime di gamma flippa |
| `oi_delta_pct_4h` | % | mcp-sentiment `liquidation_heatmap` | proxy di accumulo/sgonfiaggio leverage nelle ultime 4h |
| `liquidation_long_risk` | low / med / high | mcp-sentiment `liquidation_heatmap` | rischio long squeeze imminente |
| `liquidation_short_risk` | low / med / high | mcp-sentiment `liquidation_heatmap` | rischio short squeeze imminente |
| `macro_days_to_event` | giorni | mcp-macro `next_high_severity_within` | gate §2.5 (no entry se evento macro entro DTE) |
| `fetch_ok` | bool | scheduler | qualità riga (true = tutte le sotto-chiamate sono andate) |
| `fetch_errors_json` | json o NULL | scheduler | mappa errori per debugging best-effort |
> Un campo `NULL` non invalida la riga: la collezione è
> **best-effort**, una MCP giù non blocca le altre. Le distribuzioni
> si calcolano sui campi disponibili, l'engine entry-cycle invece
> rifiuta l'entry se il dato che gli serve è `NULL` (sicurezza:
> meglio saltare un trade che operare alla cieca).
---
## 2. Le sei famiglie di dati e il "perché"
### 2.1 — Volatilità implicita (DVOL, realized vol, IVRV)
**Cosa misura.** DVOL è l'indice Deribit della IV ETM 30g. Realized
30g è la deviazione standard annualizzata dei rendimenti spot. La
differenza `IV RV` quantifica quanto le opzioni stanno **pagando
sopra** la volatilità che il mercato ha effettivamente realizzato:
**questa è la materia prima del credit spread venditore.**
**Come la usa l'engine.**
- §2.3 / §2.4: `dvol_min = 35`, `dvol_max = 90` — sotto 35 il premio
è troppo magro rispetto a fees+slippage, sopra 90 si è in
stress-regime (rischio gap > edge).
- §5.3: `dvol_adjustment` riduce la size all'aumentare del DVOL
(×1.0 sotto 45, ×0.85 fra 4560, ×0.65 fra 6080, no entry > 80).
- §7.3: `vol_stop_dvol_increase = 10` — se durante la posizione
DVOL sale di 10 punti rispetto all'entry, si chiude.
**Cosa si calibra dai dati raccolti.** Un mese di tick ti dà la
distribuzione di DVOL nel TUO regime (testnet vs mainnet, bull vs
bear). I percentili P25/P50/P75 nella pagina `📐 Calibrazione`
dicono se 35 è davvero il "fondo" o se andrebbe alzato.
### 2.2 — Funding rate (perpetual + cross-exchange median)
**Cosa misura.** Il funding annualizzato del perpetual ETH-PERP
(Hyperliquid principalmente) e la mediana dei funding sui 4 maggiori
exchange. Il funding è la fee periodica che paga il lato sbilanciato
del perp: **è il termometro più diretto del posizionamento leveraged
del mercato.**
**Come la usa l'engine.**
- §2.6: `funding_perp_abs_max_annualized = 0.80` — funding > 80%
annualizzato (in valore assoluto) = liquidazioni a cascata
imminenti, no entry.
- §3.1: il **bias direzionale** dipende dal funding cross:
- `funding_bull_threshold_annualized = 0.20` ⇒ bias bull se
cross-funding ≥ +20%.
- `funding_bear_threshold_annualized = -0.20` ⇒ bias bear se
≤ -20%.
- In mezzo + trend neutro = candidato Iron Condor.
- Trend e funding discordi = no entry.
**Perché due funding diversi.** Il perp di Hyperliquid è il segnale
"è esecutibile la chiusura?" (l'ETH-PERP è la sede di hedge
pratico). La mediana cross-exchange è il segnale macro
"dove sta il mercato globale": più robusta a manipolazioni o picchi
locali.
### 2.3 — Dealer gamma (net gamma + flip level)
**Cosa misura.** L'esposizione netta di gamma dei dealer di opzioni
su Deribit, ricostruita da OI per strike e direzione. Quando
`dealer_net_gamma > 0` (long gamma), i dealer **sopprimono** la
volatilità realizzata col loro hedge (vendono salendo, comprano
scendendo). Quando è negativo, **amplificano** ogni movimento.
**Come la usa l'engine.**
- §2.8: `dealer_gamma_min = 0`, `dealer_gamma_filter_enabled = true`
— entry solo in regime long-gamma. Vendere credit spread con
dealer corto-gamma è statisticamente perdente.
- `gamma_flip_level` è il prezzo spot al quale il regime cambierebbe.
Se siamo a 1% dal flip, il margine di sicurezza è basso anche se
il segno è positivo.
**Cosa si calibra dai dati raccolti.** La distribuzione di
`dealer_net_gamma` nel proprio universo (qualche miliardo USD su
mainnet, ordini di grandezza diversi su testnet) suggerisce se
`min = 0` è troppo permissivo — su mainnet è frequente che il segno
si trovi positivo per molto tempo, qui ha senso una soglia più alta.
### 2.4 — Liquidation heatmap (OI delta + long/short squeeze risk)
**Cosa misura.** Da `mcp-sentiment`:
- `oi_delta_pct_4h`: variazione % dell'open interest aggregato nelle
ultime 4h. Spike positivo → leverage in entrata (rischio fragile);
spike negativo → squeeze appena avvenuta.
- `liquidation_long_risk` / `liquidation_short_risk`: classificazione
qualitativa (`low` / `med` / `high`) della densità di livelli di
liquidazione vicini allo spot.
**Come la usa l'engine.**
- §2.8 (`liquidation_filter_enabled = true`): l'entry cycle scarta
setup con `_risk = high` sul lato che ci interesserebbe (es. un
bull put spread in regime di `long_risk = high` è esposto a un
long-squeeze giù).
- Anche fuori dall'entry, queste due colonne servono come "filtro di
realtà" per il monitoring: se durante la posizione lo squeeze risk
cambia da low a high, è un primo segnale di vol-stop in arrivo.
### 2.5 — Macro calendar (giorni al prossimo evento)
**Cosa misura.** `mcp-macro` restituisce il numero di giorni al
prossimo evento ad alta severità (FOMC, CPI USA, NFP, ECB, Powell
speech) per US/EU. `NULL` = nessun evento entro la finestra DTE.
**Come la usa l'engine.**
- §2.5: se `macro_days_to_event ≤ dte_target = 18`, no entry. Le
uscite macro si trasformano in gap di volatilità che mangiano in
un'ora il credito di tre settimane.
- Le entry sono comunque possibili poco dopo l'evento (vol elevata
appena dopo + RV destinata a comprimersi → IVRV alto = setup di
scuola).
### 2.6 — Spot ETH (e BTC come controllo)
**Cosa misura.** Prezzo last/perp di ETH (e BTC come controllo).
**Come la usa l'engine.**
- §3.1: trend 30g calcolato come `(spot_now / spot_30g_ago - 1)`.
Soglie ±5% definiscono bias bull / bear / neutro.
- §3.2: distanza % degli strike short dallo spot (1525% OTM).
- §7.6: `adverse_move_4h_pct = 0.05` — close su movimento contrario
≥ 5% in 4h.
**Perché anche BTC.** ETH è il sottostante operativo, BTC è il
**termometro macro crypto**: in regimi di alta correlazione, un
movimento BTC che ETH non sta seguendo è un segnale di divergenza che
spesso precede un riallineamento brusco.
---
## 3. Il flusso decisionale, allineato al dato
Quanto segue è la versione "leggibile" delle regole §2-§9 di
`01-strategy-rules.md`. Ogni passo cita i campi di
`market_snapshots` che lo alimentano.
### Fase 1 — Trigger (lunedì 14:00 UTC, festività italiane escluse)
```
SE NESSUNA posizione aperta
E capitale ≥ 720 USD
E 35 ≤ dvol ≤ 90 # market_snapshots.dvol
E |funding_perp_annualized| ≤ 0.80 # market_snapshots.funding_perp_annualized
E macro_days_to_event > dte_target (oppure NULL) # market_snapshots.macro_days_to_event
E ETH holdings cerbero-portfolio ≤ 30%
E (filtri quant: dealer_net_gamma > 0,
liquidation_*_risk ≠ high) # market_snapshots.dealer_net_gamma + liquidation_*
ALLORA
procedi alla Fase 2
ALTRIMENTI
no entry, log motivo, ritento la settimana successiva
```
### Fase 2 — Bias e struttura
```
trend_30g = spot_now / spot_30g_ago - 1 # market_snapshots.spot
funding_x = funding_cross_annualized # market_snapshots.funding_cross_annualized
SE trend_30g ≥ +5% E funding_x ≥ +20%:
struttura = Bull Put Spread
SE trend_30g ≤ -5% E funding_x ≤ -20%:
struttura = Bear Call Spread
SE |trend_30g| < 5% E |funding_x| < 20%
E dvol ≥ 55 E ADX(14) < 20:
struttura = Iron Condor
ALTRIMENTI:
no entry (mercato indeciso o discordante)
```
### Fase 3 — Selezione strike (delta-target + distanza % spot)
Lo strike short è quello a delta target ≈ 0.12 (tolleranza 0.100.15)
**e** OTM 1525%. Lo strike long è a 4% del spot (35% accettabile).
Tutti i numeri sono parametrizzati in `strategy.yaml > structure`.
Lo `spot` corrente per il calcolo viene da `market_snapshots.spot`.
### Fase 4 — Sizing (Kelly frazionario + cap aggregato + DVOL clamp)
```
risk_target = capitale * 0.13 # quarter Kelly
risk_target = min(risk_target, 200 EUR) # cap per-trade
n = floor(risk_target / max_loss_per_contract)
n = min(n, 4, vincolo aggregato 1000 EUR)
n = round_down(n * dvol_multiplier) # market_snapshots.dvol → §5.3
```
### Fase 5 — Esecuzione (combo limit GTC al mid)
Limit al mid del combo, riprezzamento +1 tick / 30min fino a 3 step.
Su trigger urgenti (CLOSE_STOP / CLOSE_VOL / CLOSE_DELTA) l'engine
accetta fino a 5 step di slippage perché l'urgenza prevale sul
prezzo.
### Fase 6 — Monitoring (cron di gestione attiva, default ogni 12h)
Per ogni posizione aperta, in **ordine** (primo trigger vince):
| # | Trigger | Dato sorgente |
|---|---|---|
| 1 | Profit take: mark ≤ 50% credito | combo mark via deribit |
| 2 | Stop loss: mark ≥ 250% credito | combo mark via deribit |
| 3 | Vol stop: dvol_now ≥ dvol_entry + 10 | `market_snapshots.dvol` |
| 4 | Time stop: dte ≤ 7 (skip se ≥ 70% profit) | scadenza struttura |
| 5 | Delta breach: \|delta_short\| ≥ 0.30 | option chain via deribit |
| 6 | Adverse move: \|return_4h_ETH\| ≥ 5% contro | `market_snapshots.spot` |
| 7 | Altrimenti | HOLD |
Il monitoring NON consulta `market_snapshots` per i prezzi opzioni
(legge live), ma li consulta per `dvol` e `spot` con il vantaggio di
una serie storica già normalizzata e auditabile.
---
## 4. Cosa fa OGGI il bot in modalità "data-only"
Il bot oggi è in **modalità raccolta dati** (`ENABLE_DATA_ANALYSIS=true`,
`ENABLE_STRATEGY=false`). Vuol dire:
- Il job `market_snapshot` (cron `*/15`) gira: scrive nuove righe in
SQLite, alimenta calibrazione e monitoring storico.
- Il job `health` (`*/5`) verifica disponibilità MCP e ambiente
Deribit; alza il kill switch se qualcosa non torna.
- Il job `backup` (`0 *`) snapshotta lo stato ogni ora.
- Il job `manual_actions` (`*/1`) consuma comandi dalla GUI.
- I cicli `entry` e `monitor` **non sono nemmeno schedulati**: nessun
ordine può partire, nessuno strike viene letto.
Quando si vuole passare alla fase operativa (paper trading o
mainnet), basta:
1. Riempire `strategy.yaml` con le **soglie calibrate** sui
percentili reali della pagina `📐 Calibrazione` (non lasciare i
valori default a istinto).
2. Bumpare `config_version` + rigenerare `config_hash` con
`cerbero-bite config hash --file strategy.yaml`.
3. Settare `ENABLE_STRATEGY=true` in `.env` e ricreare il container.
4. Disarmare il kill switch da GUI o CLI con motivazione esplicita.
5. **Una settimana di paper trading** (mainnet con ordini disabilitati
o testnet) prima di alzare il flag definitivo.
---
## 4-bis. P/L atteso (realistico)
I numeri qui sotto sono **stime ex-ante**, non promesse. Servono ad
allineare le aspettative con la geometria della strategia: capire
**quanto poco si rischia per trade**, **quanto raramente si entra**, e
**perché l'edge è strutturalmente sottile**.
> **Domanda onesta che chiunque guardi i numeri dovrebbe farsi:** se a
> win-rate 7072% l'aspettativa per trade è circa zero, **che senso ha
> la strategia?**
>
> **Risposta:** il selling vol nudo è effettivamente neutro a quel
> win-rate. **L'edge della Cerbero Bite non è "vendere vol"; è
> "vendere vol solo quando i filtri quant alzano il win-rate sopra il
> 75%".** I gate §2 (DVOL band, dealer gamma > 0, no macro entro DTE,
> liquidation risk ≠ high, bias trend × funding concorde) sono
> **costruiti per saltare proprio le finestre statisticamente
> perdenti** e operare solo in quelle favorevoli. La pagina
> `📚 Strategia` ha una tabella di sensibilità che mostra come l'APR
> passa da ≈0% (win 0.72) a +3-5% (win 0.780.80): è esattamente la
> distanza che i filtri devono coprire. Per questo i primi giorni di
> raccolta dati servono a **misurare** se i filtri stanno effettivamente
> alzando il win-rate prima di committare capitale.
### Per singolo trade (riferimento: ETH spot ≈ 3000 USD)
| Voce | Formula / fonte | Valore tipico |
|---|---|---|
| Larghezza spread | 4% × spot | **120 USD / contratto** |
| Credito incassato | ≥ 30% × larghezza | **3648 USD / contratto** |
| Max profit teorico | = credito (a scadenza OTM) | 3648 USD / contratto |
| **Profit-take §7.1 (50% credito)** | 0.5 × credito | **+1824 USD / contratto** |
| **Stop-loss §7.2 (mark = 2.5× credito)** | 1.5 × credito | **5472 USD / contratto** |
| Margine bloccato | ≈ larghezza | 120 USD / contratto |
| Fees Deribit | 0.03% notional × 2 leg | ~12 USD / contratto / trade |
> Su spot più basso (2000 USD) la larghezza scende a 80 USD/contratto
> e i numeri assoluti seguono proporzionalmente.
### Sizing tipico vs capitale
Il sizing è governato dal Quarter-Kelly **+ cap per-trade 200 EUR
(~215 USD)**. Sopra una certa soglia, il cap domina: alzare il
capitale **non aumenta** i contratti per trade.
| Capitale | risk_target (Kelly) | risk effettivo (post-cap) | Contratti tipici (spot=3000) |
|---|---|---|---|
| 720 USD (minimo) | 94 USD | 94 USD | **01** (entry spesso saltata per sizing) |
| 1 500 USD | 195 USD | 195 USD | **1** |
| 3 000 USD | 390 USD | **215 USD** (cap) | **1** |
| 10 000 USD | 1 300 USD | **215 USD** (cap) | **1** |
| 50 000 USD+ | 6 500 USD | **215 USD** (cap) | **1** (cap aggregato 1 075 USD = max 4 trade aperti, ma `max_concurrent_positions: 1`) |
> Con i cap correnti la strategia è **dimensionata per capitale
> piccolo (1.510 k USD)**: oltre, il rendimento sul totale scala
> sotto-lineare e tende a zero.
### Frequenza realistica di entry
La regola si valuta una volta a settimana, ma la maggioranza dei
lunedì viene saltata per:
| Motivo di skip | Frequenza tipica |
|---|---|
| DVOL fuori banda (3590) | 2540% |
| Bias non chiaro (trend × funding discordi o entrambi neutri senza IC) | 2535% |
| Macro entro DTE | 1020% |
| Funding o liquidation risk fuori soglia | 515% |
| Capitale o sizing insufficiente | 05% |
**Risultato netto: 3050% delle settimane finisce in entry effettiva
⇒ 1525 trade / anno** (52 lunedì × 3050%). Le altre settimane il
bot sta fermo. È il design.
### Win-rate atteso (short delta 0.12 + profit-take 50%)
Letteratura e backtest su credit spread short delta 0.100.15 con
TP@50% e SL@1.5×:
| Esito | Probabilità tipica | Risultato |
|---|---|---|
| Profit-take a 50% credito | **~7075%** | +1824 USD/contratto |
| Stop-loss a 1.5× credito | ~1520% | 5472 USD/contratto |
| Time-stop o exit DTE 7g | ~510% | piccolo positivo (~+510 USD) |
| Vol/delta/macro stop | ~35% | variabile, mediamente neutro |
Atteso medio per contratto:
```
E[trade] ≈ 0.72 × 21 + 0.18 × (-63) + 0.07 × 7 + 0.03 × 0
≈ 15.1 11.3 + 0.5 + 0
≈ +4.3 USD lordi / contratto
```
**Al netto di fees (~1.5 USD round-trip) e slippage (~5% del credito
≈ 2 USD): E[trade] ≈ +13 USD per contratto.**
### Proiezione annuale (1 contratto medio per trade)
| Scenario | Trade/anno | E[trade] netto | P/L lordo annuo | Su capitale 1 500 USD | Su capitale 3 000 USD |
|---|---|---|---|---|---|
| **Pessimistico** (vol bassa, regime bear vol) | 12 | +1 USD | **+12 USD** | +0.8% | +0.4% |
| **Realistico medio** | 18 | +2.5 USD | **+45 USD** | +3% | +1.5% |
| **Buono** (regime favorevole, IVRV alto) | 22 | +4 USD | **+88 USD** | +5.9% | +2.9% |
| **Eccellente** (cherry-picking ex-post) | 25 | +6 USD | **+150 USD** | +10% | +5% |
**Realisticamente: +1.5% / +5% APR sul capitale totale**, con i cap
correnti. È in linea con la letteratura su short-vol systematic con
disciplina di stop. **Non è una strategia "raddoppia il capitale".**
È una strategia che vuole guadagnare il **premio di rischio della
volatilità** in modo controllato.
### Drawdown e rischio coda
- **Streak realistico di perdite consecutive**: 35 stop-loss di fila
capitano. Drawdown su 1 contratto: 150 / 300 USD assoluti.
- **Su capitale 1 500 USD** = drawdown del 1020% del capitale
totale. Aspettarselo, è dentro il design.
- **Tail risk:** un evento gap notturno (sentenza SEC, hack
exchange, default importante) può portare il mark a 100% della
larghezza prima che lo stop sia eseguibile. **Perdita massima
reale per trade = larghezza intera** (`width - credit_iniziale`),
cioè 7296 USD/contratto, non i 5472 USD del modello stop-loss.
- I **filtri quant** (`dealer_gamma_min`, `liquidation_filter`) e
il **macro filter** sono stati introdotti **per ridurre la coda**,
non per migliorare l'aspettativa media.
### Sharpe atteso
Strategie short-vol sistematiche con disciplina hanno:
- **Sharpe 0.81.5** in regimi favorevoli (mercato lento + IV alta).
- **Sharpe 0.30.8** in regimi normali.
- **Sharpe negativo** in regimi di vol-of-vol (es. Q1 2020, Maggio
2021, FTX week). I filtri li mitigano, non li annullano.
### Cosa cambia con `ENABLE_STRATEGY=true`
In modalità data-only (oggi) il P/L atteso è **0** — l'engine
**non opera**. Il valore della raccolta di oggi è:
1. **Calibrare** soglie su percentili reali → P/L atteso più
realistico al go-live.
2. **Validare** i filtri quant osservando ex-post quanti tick
sarebbero stati filtrati (vedi pagina `📐 Calibrazione`, colonna
"% bloccato dalla soglia").
3. **Misurare** la quota effettiva di lunedì che superano i filtri
nel proprio regime, prima di committare capitale.
> Suggerimento: 4 settimane di dati = 4 lunedì × probabilità entry =
> 12 candidate entry effettive. **Aspettare almeno 8 settimane**
> prima di tarare le soglie dà uno storico con dispersione
> sufficiente per decisioni non-rumorose.
---
## 4-ter. Due profili: Conservativa vs Aggressiva
Il P/L del §4-bis assume i cap della golden config v1.0.0
(`cap_per_trade_eur: 200`, `max_concurrent_positions: 1`,
`max_contracts_per_trade: 4`). Su quel profilo il P/L assoluto è
piccolo per design — la strategia è dimensionata come **macchina di
conservazione del capitale** con premio modesto su T-bill.
Per chi vuole rendimenti significativi, il repo include un secondo
file di config — `strategy.aggressiva.yaml` — che **deroga
esplicitamente** alla §11 di `01-strategy-rules.md` allargando le tre
leve dominanti:
| Leva | Conservativa | Aggressiva | Effetto sul P/L |
|---|---|---|---|
| `cap_per_trade_eur` | 200 | **800** | 4× la size per trade |
| `cap_aggregate_open_eur` | 1 000 | **3 200** | 4× il rischio aggregato |
| `max_concurrent_positions` | 1 | **2** | 2× le posizioni aperte simultanee |
| `max_contracts_per_trade` | 4 | **16** | toglie il vincolo aggregato anche su capitali maggiori |
| `kelly_fraction` | 0.13 | **0.13** | invariato (la disciplina Kelly resta) |
| Filtri quant (gamma, liquidation, macro) | ON | **ON** | invariati (l'edge è qui, non si tocca) |
**Risultato atteso (a parità di filtri e win-rate):** P/L ≈ 48× il
profilo conservativo. Drawdown atteso scala con lo stesso fattore
(2040% del capitale impiegato in streak avverse, contro 1020% del
conservativo). La pagina `📚 Strategia` ha un pannello affiancato che
calcola entrambi sugli stessi slider.
**Il rovescio della medaglia.**
- La deroga alla §11 va **autorizzata esplicitamente** nel commit che
switcha la config; tre settimane di paper trading dedicato sono
raccomandate.
- Il drawdown maggiore richiede capitale "growth", non capitale di
parcheggio.
- I filtri quant restano **identici** — non c'è "più aggressivo" sui
trigger di entry, perché lì non c'è alpha da spremere senza
peggiorare il win-rate.
**Multi-asset (ETH + BTC) — caveat.**
L'ulteriore moltiplicatore 2× citato nel §4-bis (multi-asset) **non è
abilitato** dalla sola modifica della config: il rule engine attuale è
single-asset (`asset.symbol`). Per estenderlo servono modifiche in:
- `cerbero_bite/runtime/entry_cycle.py` (loop sui simboli)
- `cerbero_bite/state/repository.py` (multi-position chiave per asset)
- `cerbero_bite/runtime/orchestrator.py` (scheduler one-asset → N)
Il job di raccolta dati è già multi-asset (`DEFAULT_ASSETS = ("ETH",
"BTC")`), quindi tutto il dataset utile per validare l'estensione è
già disponibile. È un lavoro di codice ben circoscritto, da fare in
un branch dedicato dopo che il dataset di calibrazione è abbondante.
**Quando passare dal profilo conservativo all'aggressivo.**
Solo se **tutte** le seguenti sono vere:
1. ≥ 8 settimane di dati raccolti su mainnet (≥ ~2k snapshot).
2. Win-rate empirico misurato (paper trading o backtest sui tick
raccolti) **≥ 0.75**.
3. APR atteso del profilo aggressivo (vedi pannello GUI) **≥ 8%**
netto a quel win-rate.
4. Capitale impegnato è **growth capital**, non riserva tattica.
5. Sopporti emotivamente un drawdown a doppia cifra senza disarmare
manualmente la strategia in mezzo a una streak.
Se anche solo uno dei 5 manca → **resta sulla conservativa**, è
quella che il sistema parte ad eseguire.
---
## 5. Come leggere il dato giorno per giorno
Tre euristiche operative sui campi raccolti:
1. **Premio "ricco":** `iv_minus_rv` consistentemente > 5 punti per
N giorni → il regime sta pagando bene la vendita di vol. Sono i
periodi in cui la strategia ha edge maggiore.
2. **Premio "magro":** `dvol < 35` per più giorni → la finestra del
lunedì viene saltata. Non è un fallimento: è la disciplina che
funziona.
3. **Stress imminente:** `liquidation_*_risk = high` o spike di
`oi_delta_pct_4h` (> 5% in valore assoluto) + funding ai limiti
→ atteso vol stop / time stop attivi nei prossimi cicli, anche
se la posizione è in profit.
Nei giorni di **eventi macro** (`macro_days_to_event` piccolo) la
combinazione utile è: aspettare l'evento, lasciare che `dvol` scenda
quando `realized_vol_30d` non si è realizzata, e cogliere il setup
classico **post-evento**.
---
## 6. Glossario rapido
- **Credit spread:** vendita di un'opzione e acquisto di un'opzione
più OTM stessa scadenza per cap del rischio. Si incassa un credito,
si vince se il sottostante non rompe lo strike short.
- **Bull put / Bear call:** credit spread direzionali (rispettivamente
bullish / bearish).
- **Iron condor:** Bull put + bear call sullo stesso sottostante e
scadenza. Si vince in regime laterale.
- **DVOL:** indice Deribit della IV ETM 30g, scala 0200.
- **Realized vol 30g:** σ annualizzata dei rendimenti spot sui 30g
rolling.
- **IV RV:** differenza tra IV implicita (DVOL) e RV; > 0 = "premio"
positivo per il venditore di vol.
- **Funding annualizzato:** funding rate del perp moltiplicato per le
finestre standard (di solito 8h × 3 al giorno × 365).
- **Dealer net gamma:** somma di gamma per tutti gli strike, pesata
per direzione dei dealer (long = riduce vol, short = amplifica).
- **OI delta % 4h:** variazione % dell'open interest aggregato nelle
ultime 4 ore.
- **DTE:** Days To Expiry, giorni alla scadenza dell'opzione.
- **Kill switch:** flag persistente che blocca apertura di nuove
posizioni; armato automaticamente su mismatch ambiente o failure
ripetuti, disarmato solo manualmente con motivazione.
---
## 7. Riferimenti incrociati
- Regole canoniche e immutabili: `01-strategy-rules.md`
- Schema dati persistente: `05-data-model.md`
- Algoritmi (calcolo trend, IVRV, ecc.): `03-algorithms.md`
- Dettaglio integrazioni MCP: `04-mcp-integration.md`
- Pagina GUI di calibrazione: `📐 Calibrazione`
- Sorgente del collector: `src/cerbero_bite/runtime/market_snapshot_cycle.py`
- Modello pydantic riga: `cerbero_bite.state.models.MarketSnapshotRecord`
+150 -1
View File
@@ -13,7 +13,7 @@ import asyncio
import os import os
import sys import sys
from collections.abc import Callable from collections.abc import Callable
from datetime import UTC, datetime from datetime import UTC, datetime, timedelta
from decimal import Decimal from decimal import Decimal
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -679,6 +679,155 @@ def replay(date_from: str, date_to: str, capital: float, dry_run: bool) -> None:
) )
@main.command()
@click.option(
"--strategy",
"strategy_path",
type=click.Path(path_type=Path),
default=Path("strategy.yaml"),
show_default=True,
help="Path al file di strategia (golden, conservativa, aggressiva, ...).",
)
@click.option(
"--db",
"db_path",
type=click.Path(path_type=Path),
default=_DEFAULT_DB_PATH,
show_default=True,
help="SQLite con `market_snapshots` storiche.",
)
@click.option(
"--from",
"date_from",
type=click.DateTime(formats=["%Y-%m-%d"]),
default=None,
help="ISO date YYYY-MM-DD (default: 90 giorni fa).",
)
@click.option(
"--to",
"date_to",
type=click.DateTime(formats=["%Y-%m-%d"]),
default=None,
help="ISO date YYYY-MM-DD (default: oggi).",
)
@click.option(
"--capital",
type=float,
default=1500.0,
show_default=True,
help="Capitale di partenza per il backtest, in USD.",
)
@click.option(
"--asset",
type=str,
default="ETH",
show_default=True,
help="Asset di riferimento per le snapshot.",
)
@click.option(
"--no-enforce-hash",
is_flag=True,
default=False,
help="Salta la verifica del config_hash (utile per profili sperimentali).",
)
def backtest(
strategy_path: Path,
db_path: Path,
date_from: datetime | None,
date_to: datetime | None,
capital: float,
asset: str,
no_enforce_hash: bool,
) -> None:
"""Esegue il backtest stilizzato su `market_snapshots` storiche.
Usa lo stesso `validate_entry` del live per i filtri (rigoroso) e
un modello Black-Scholes con skew premium per stimare credito ed
exit P/L (stilizzato — vedi docstring di `core/backtest.py`).
"""
from cerbero_bite.config.loader import load_strategy # noqa: PLC0415
from cerbero_bite.core.backtest import run_backtest # noqa: PLC0415
console = Console()
if date_to is None:
date_to = datetime.now(UTC)
if date_from is None:
date_from = date_to - timedelta(days=90)
date_from = date_from.replace(tzinfo=UTC) if date_from.tzinfo is None else date_from
date_to = date_to.replace(tzinfo=UTC) if date_to.tzinfo is None else date_to
loaded = load_strategy(strategy_path, enforce_hash=not no_enforce_hash)
cfg = loaded.config
conn = connect_state(db_path)
try:
repo = Repository()
snapshots = repo.list_market_snapshots(
conn,
asset=asset.upper(),
start=date_from,
end=date_to,
limit=10000,
)
finally:
conn.close()
if not snapshots:
console.print(
f"[yellow]Nessuno snapshot {asset} trovato fra {date_from.date()} "
f"e {date_to.date()}.[/yellow]"
)
sys.exit(1)
console.print(
f"[green]Caricate {len(snapshots)} snapshot {asset} "
f"({snapshots[-1].timestamp.date()}{snapshots[0].timestamp.date()})[/green]"
)
report = run_backtest(snapshots, cfg, capital_usd=Decimal(str(capital)))
table = Table(title=f"Backtest report — {strategy_path.name}")
table.add_column("Metrica", style="cyan")
table.add_column("Valore", style="bold")
table.add_row("Picks (lunedì 14:00)", str(report.n_picks))
table.add_row(
"Accettati dai filtri",
f"{report.n_accepted} ({report.n_accepted / max(1, report.n_picks):.0%})",
)
table.add_row("Saltati per dato mancante", str(report.n_skipped_data))
table.add_row("Trade completati (con P/L)", str(report.n_completed))
table.add_row("Vincenti", f"{report.n_winners} ({report.win_rate:.0%})")
table.add_row("P/L cumulato (USD)", f"{report.cumulative_pnl_usd:+.2f}")
table.add_row(
"P/L su capitale", f"{report.cumulative_pnl_pct_of_capital:+.2%}"
)
table.add_row(
"Max drawdown", f"{report.max_drawdown_usd:.0f} USD "
f"({report.max_drawdown_pct:.1%})",
)
table.add_row(
"Sharpe (annualized)",
f"{report.sharpe_annualized}" if report.sharpe_annualized is not None
else "",
)
console.print(table)
if report.skip_reasons:
skip_table = Table(title="Motivi di skip aggregati")
skip_table.add_column("Motivo")
skip_table.add_column("Settimane", justify="right")
for reason, count in sorted(
report.skip_reasons.items(), key=lambda kv: -kv[1]
):
skip_table.add_row(reason, str(count))
console.print(skip_table)
console.print(
"[dim]Il modello P/L è stilizzato: BS + skew premium 1.5×. "
"Numeri ottimi per ranking config, non per promesse operative.[/dim]"
)
@main.group() @main.group()
def config() -> None: def config() -> None:
"""Strategy configuration utilities.""" """Strategy configuration utilities."""
+652
View File
@@ -0,0 +1,652 @@
"""Stylized backtest engine over ``market_snapshots`` (§13).
Two layers, both pure functions:
1. **Entry-filter simulation** — for each Monday 14:00 UTC tick in the
recorded snapshots, evaluate which §2 gates would have passed,
reconstructing :class:`EntryContext` from the snapshot. This part
is **rigorous**: it uses the same :func:`validate_entry` the live
engine uses, so the output is exactly "what the bot would have
decided".
2. **P/L estimation per accepted entry** — since ``market_snapshots``
does NOT record the option chain (we only collect spot, DVOL,
funding, etc.), credit and exit P/L are estimated via a stylized
Black-Scholes model: given ``spot``, ``DVOL`` (as IV), and the
strategy's delta target, we solve for the short strike, the long
strike at ``width_pct`` distance, and the combo mid-price. Future
ticks are then re-priced under the same model to detect the first
exit trigger from §7.
The stylized layer is **intentionally approximate**: it captures the
geometry of the strategy (DVOL band sets credit, ETH path drives
exit triggers) but not the second-order effects (chain liquidity,
borrow rates, exchange fees beyond the 0.03% notional cap, dealer
hedging skew). Numbers are good for ranking and tuning, not for
operational P/L promises.
The engine is deterministic and side-effect-free: it does **not**
write to SQLite, does not call MCP, does not place orders. It
operates entirely on a list of :class:`MarketSnapshotRecord` rows
the caller has already loaded.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Literal
from pydantic import BaseModel, ConfigDict
from cerbero_bite.config.schema import StrategyConfig
from cerbero_bite.core.entry_validator import EntryContext, validate_entry
from cerbero_bite.state.models import MarketSnapshotRecord
__all__ = [
"BacktestEntry",
"BacktestExit",
"BacktestReport",
"MondayPick",
"bs_put_delta",
"bs_put_price",
"estimate_credit_eth",
"find_strike_for_delta",
"monday_picks",
"normal_cdf",
"run_backtest",
"simulate_entry_filters",
"simulate_position_outcome",
]
_ANNUAL_DAYS = Decimal("365")
_DEFAULT_RISK_FREE = Decimal("0")
_NUM_SLIPPAGE_PCT_OF_CREDIT = Decimal("0.03")
_NUM_FEE_PCT_OF_NOTIONAL = Decimal("0.0003")
# ---------------------------------------------------------------------------
# Black-Scholes helpers (stdlib-only)
# ---------------------------------------------------------------------------
def normal_cdf(x: float) -> float:
"""Standard normal CDF, no scipy dependency."""
return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))
def bs_put_price(*, spot: float, strike: float, t_years: float, sigma: float) -> float:
"""European put price under r=0, q=0 Black-Scholes.
Returns price in spot units (so for an ETH option, dividing by spot
gives the price in ETH).
"""
if t_years <= 0 or sigma <= 0 or spot <= 0 or strike <= 0:
return max(0.0, strike - spot)
sqrt_t = math.sqrt(t_years)
d1 = (math.log(spot / strike) + 0.5 * sigma * sigma * t_years) / (sigma * sqrt_t)
d2 = d1 - sigma * sqrt_t
return strike * normal_cdf(-d2) - spot * normal_cdf(-d1)
def bs_put_delta(*, spot: float, strike: float, t_years: float, sigma: float) -> float:
"""Put delta under r=0, q=0 Black-Scholes (negative number for put).
Returns 0 for expired options.
"""
if t_years <= 0 or sigma <= 0 or spot <= 0 or strike <= 0:
return 0.0
sqrt_t = math.sqrt(t_years)
d1 = (math.log(spot / strike) + 0.5 * sigma * sigma * t_years) / (sigma * sqrt_t)
return normal_cdf(d1) - 1.0 # = -N(-d1)
def find_strike_for_delta(
*,
spot: float,
dvol_pct: float,
dte_days: int,
target_delta_abs: float,
) -> float:
"""Solve for the put strike whose |delta| matches ``target_delta_abs``.
Bisection on a monotone-decreasing |delta(strike)| relationship.
Returns the strike in absolute USD terms.
"""
sigma = max(0.01, dvol_pct / 100.0)
t_years = max(1e-6, dte_days / 365.0)
# Bracket: from 50% of spot (deep OTM, small |delta|) up to spot
# (ATM, |delta| ≈ 0.5).
low = max(1.0, spot * 0.30)
high = spot
for _ in range(64):
mid = 0.5 * (low + high)
delta_abs = abs(bs_put_delta(spot=spot, strike=mid, t_years=t_years, sigma=sigma))
if delta_abs > target_delta_abs:
high = mid
else:
low = mid
if abs(high - low) < 1e-3:
break
return 0.5 * (low + high)
def estimate_credit_eth(
*,
spot: float,
dvol_pct: float,
dte_days: int,
width_pct: float,
delta_target_abs: float,
skew_premium: float = 1.5,
) -> tuple[float, float, float]:
"""Estimate credit (ETH), short_strike, long_strike for a bull-put-style
credit spread under Black-Scholes.
``skew_premium`` è il moltiplicatore applicato al credito BS per
approssimare la **vol smile** dell'ETH options market (le put OTM
trattano a IV più alta della IV ATM, quindi un BS pulito sottostima
sistematicamente il premio del venditore di vol). Il default 1.5
è una stima conservativa dei dati Deribit storici (smile slope
tipica 5-10 vol points per 100δ); valori sensati: 1.3 (smile
blanda) … 1.8 (regime "stress IV"). Va calibrato sui dati reali
quando avremo abbastanza chain history da farlo.
Returns ``(credit_eth, short_strike, long_strike)``. Credit è
già moltiplicato per ``skew_premium``.
"""
short_strike = find_strike_for_delta(
spot=spot, dvol_pct=dvol_pct, dte_days=dte_days,
target_delta_abs=delta_target_abs,
)
width_usd = width_pct * spot
long_strike = max(1.0, short_strike - width_usd)
sigma = max(0.01, dvol_pct / 100.0)
t_years = max(1e-6, dte_days / 365.0)
short_mid_usd = bs_put_price(
spot=spot, strike=short_strike, t_years=t_years, sigma=sigma,
)
long_mid_usd = bs_put_price(
spot=spot, strike=long_strike, t_years=t_years, sigma=sigma,
)
short_mid_eth = short_mid_usd / spot
long_mid_eth = long_mid_usd / spot
credit_eth = (short_mid_eth - long_mid_eth) * skew_premium
return credit_eth, short_strike, long_strike
# ---------------------------------------------------------------------------
# Entry filter simulation — rigorous (uses validate_entry exactly)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class MondayPick:
"""Indice di un tick "Monday 14:00 UTC" nella time-series."""
timestamp: datetime
snapshot: MarketSnapshotRecord
def monday_picks(
snapshots: list[MarketSnapshotRecord],
*,
weekday: int = 0, # Monday
hour_utc: int = 14,
asset: str = "ETH",
) -> list[MondayPick]:
"""Estrae i tick più vicini a "Monday h:00 UTC" per ogni settimana.
``snapshots`` deve essere ordinato per timestamp ascending. Per ogni
occorrenza di ``weekday + hour_utc`` (es. lun 14:00) presa l'unica
riga ETH che la copre. Settimane senza tick a quell'ora vengono
saltate.
"""
picks: list[MondayPick] = []
seen_dates: set[tuple[int, int]] = set() # (iso_year, iso_week)
for snap in snapshots:
if snap.asset.upper() != asset.upper():
continue
ts = snap.timestamp.astimezone(UTC)
if ts.weekday() != weekday or ts.hour != hour_utc:
continue
iso_y, iso_w, _ = ts.isocalendar()
key = (iso_y, iso_w)
if key in seen_dates:
continue
seen_dates.add(key)
picks.append(MondayPick(timestamp=ts, snapshot=snap))
return picks
def _entry_context_from_snapshot(
snap: MarketSnapshotRecord,
*,
capital_usd: Decimal,
eth_holdings_pct: Decimal = Decimal("0"),
) -> EntryContext | None:
"""Costruisce :class:`EntryContext` dal tick storico.
``None`` quando la riga non ha i campi minimi (spot, dvol, funding).
Nel filtro questo si traduce in "skip della settimana" — è la
stessa logica del live: un tick incompleto è meglio di un'entry
al buio.
"""
if snap.dvol is None or snap.funding_perp_annualized is None:
return None
return EntryContext(
capital_usd=capital_usd,
dvol_now=snap.dvol,
funding_perp_annualized=snap.funding_perp_annualized,
eth_holdings_pct_of_portfolio=eth_holdings_pct,
next_macro_event_in_days=snap.macro_days_to_event,
has_open_position=False,
dealer_net_gamma=snap.dealer_net_gamma,
liquidation_squeeze_risk_high=(
snap.liquidation_long_risk == "high"
or snap.liquidation_short_risk == "high"
),
)
@dataclass(frozen=True)
class EntryFilterResult:
"""Esito del check filtri per una singola Monday pick."""
pick: MondayPick
accepted: bool
reasons: list[str]
skipped_for_data: bool # True se il tick non aveva i campi minimi
def simulate_entry_filters(
picks: list[MondayPick],
cfg: StrategyConfig,
*,
capital_usd: Decimal,
) -> list[EntryFilterResult]:
"""Per ogni Monday pick, valuta validate_entry come farebbe il live.
Rigoroso: usa esattamente :func:`validate_entry` e :class:`EntryContext`.
Restituisce la lista degli esiti, una entry per pick.
"""
results: list[EntryFilterResult] = []
for pick in picks:
ctx = _entry_context_from_snapshot(pick.snapshot, capital_usd=capital_usd)
if ctx is None:
results.append(
EntryFilterResult(
pick=pick,
accepted=False,
reasons=["incomplete_snapshot"],
skipped_for_data=True,
)
)
continue
decision = validate_entry(ctx, cfg)
results.append(
EntryFilterResult(
pick=pick,
accepted=decision.accepted,
reasons=list(decision.reasons),
skipped_for_data=False,
)
)
return results
# ---------------------------------------------------------------------------
# Position outcome simulation — stylized (Black-Scholes re-pricing)
# ---------------------------------------------------------------------------
class BacktestEntry(BaseModel):
"""Trade aperto nel backtest (snapshot al momento dell'entry)."""
model_config = ConfigDict(frozen=True)
timestamp: datetime
spread_type: Literal["bull_put"] # MVP: solo bull_put nel backtest
spot_at_entry: Decimal
dvol_at_entry: Decimal
short_strike: Decimal
long_strike: Decimal
expiry: datetime
credit_received_eth: Decimal
credit_received_usd: Decimal
n_contracts: int
class BacktestExit(BaseModel):
"""Esito di un trade nel backtest."""
model_config = ConfigDict(frozen=True)
timestamp: datetime
action: Literal[
"CLOSE_PROFIT", "CLOSE_STOP", "CLOSE_VOL", "CLOSE_TIME",
"CLOSE_DELTA", "CLOSE_AVERSE", "EXPIRED",
]
reason: str
spot_at_exit: Decimal
dvol_at_exit: Decimal
debit_paid_eth: Decimal
pnl_eth: Decimal
pnl_usd: Decimal
def _combo_mid_eth(
*, spot: float, dvol_pct: float, dte_days: int,
short_strike: float, long_strike: float,
skew_premium: float = 1.5,
) -> float:
"""Re-prezza il combo bull-put usando BS sul nuovo spot/dvol/dte."""
sigma = max(0.01, dvol_pct / 100.0)
t_years = max(1e-6, dte_days / 365.0)
short_mid_usd = bs_put_price(
spot=spot, strike=short_strike, t_years=t_years, sigma=sigma,
)
long_mid_usd = bs_put_price(
spot=spot, strike=long_strike, t_years=t_years, sigma=sigma,
)
return (short_mid_usd - long_mid_usd) / spot * skew_premium
def simulate_position_outcome(
entry: BacktestEntry,
future_snapshots: list[MarketSnapshotRecord],
cfg: StrategyConfig,
) -> BacktestExit:
"""Re-prezza il combo a ogni tick futuro fino al primo exit trigger.
Triggers in ordine §7:
1. profit_take (debit ≤ 0.5×credit)
2. stop_loss (debit ≥ 2.5×credit)
3. vol_stop (DVOL salita di ≥10 pt rispetto entry)
4. time_stop (DTE ≤ 7 e debit > 0.7×credit)
5. expiry (uscita per scadenza, P/L = credit intrinsic)
"""
ec = cfg.exit
credit = float(entry.credit_received_eth)
short = float(entry.short_strike)
long_ = float(entry.long_strike)
profit_thresh = float(ec.profit_take_pct_of_credit) * credit
stop_thresh = float(ec.stop_loss_mark_x_credit) * credit
skip_time_thresh = float(ec.time_stop_skip_if_close_to_profit_pct) * credit
for snap in future_snapshots:
if snap.timestamp <= entry.timestamp:
continue
if snap.timestamp >= entry.expiry:
break
if snap.dvol is None or snap.spot is None:
continue
spot_now = float(snap.spot)
dvol_now = float(snap.dvol)
dte = max(0, (entry.expiry - snap.timestamp).days)
debit = _combo_mid_eth(
spot=spot_now, dvol_pct=dvol_now, dte_days=dte,
short_strike=short, long_strike=long_,
)
if debit <= profit_thresh:
return _exit(
snap, entry, debit,
action="CLOSE_PROFIT",
reason=f"debit {debit:.4f}{profit_thresh:.4f}",
)
if debit >= stop_thresh:
return _exit(
snap, entry, debit,
action="CLOSE_STOP",
reason=f"debit {debit:.4f}{stop_thresh:.4f}",
)
if dvol_now >= float(entry.dvol_at_entry) + float(ec.vol_stop_dvol_increase):
return _exit(
snap, entry, debit,
action="CLOSE_VOL",
reason=f"DVOL {dvol_now:.1f} ≥ entry+{ec.vol_stop_dvol_increase}",
)
if dte <= ec.time_stop_dte_remaining and debit > skip_time_thresh:
return _exit(
snap, entry, debit,
action="CLOSE_TIME",
reason=f"DTE {dte}{ec.time_stop_dte_remaining}",
)
# Tick passati senza trigger: scadenza naturale.
last = future_snapshots[-1] if future_snapshots else None
intrinsic = max(0.0, short - float(last.spot if last and last.spot else 0))
intrinsic_capped = min(intrinsic, short - long_)
debit_at_expiry_eth = (
intrinsic_capped / float(last.spot)
if last is not None and last.spot is not None and float(last.spot) > 0
else 0.0
)
return _exit(
last or _synthetic_expiry_snapshot(entry),
entry,
debit_at_expiry_eth,
action="EXPIRED",
reason="held to expiry",
)
def _synthetic_expiry_snapshot(entry: BacktestEntry) -> MarketSnapshotRecord:
return MarketSnapshotRecord(
timestamp=entry.expiry,
asset="ETH",
spot=entry.spot_at_entry,
dvol=entry.dvol_at_entry,
fetch_ok=False,
)
def _exit(
snap: MarketSnapshotRecord,
entry: BacktestEntry,
debit_eth: float,
*,
action: str,
reason: str,
) -> BacktestExit:
pnl_eth = float(entry.credit_received_eth) - debit_eth
spot = float(snap.spot) if snap.spot is not None else float(entry.spot_at_entry)
dvol = float(snap.dvol) if snap.dvol is not None else float(entry.dvol_at_entry)
return BacktestExit(
timestamp=snap.timestamp,
action=action, # type: ignore[arg-type]
reason=reason,
spot_at_exit=Decimal(str(spot)),
dvol_at_exit=Decimal(str(dvol)),
debit_paid_eth=Decimal(str(debit_eth)),
pnl_eth=Decimal(str(pnl_eth)),
pnl_usd=Decimal(str(pnl_eth * spot * entry.n_contracts)),
)
# ---------------------------------------------------------------------------
# Full pipeline
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class CompletedTrade:
entry: BacktestEntry
exit: BacktestExit
class BacktestReport(BaseModel):
"""Aggregato del backtest. Tutti i numeri sono **stime**."""
model_config = ConfigDict(frozen=True)
n_picks: int
n_accepted: int
n_skipped_data: int
n_completed: int
n_winners: int
win_rate: Decimal
cumulative_pnl_usd: Decimal
cumulative_pnl_pct_of_capital: Decimal
max_drawdown_usd: Decimal
max_drawdown_pct: Decimal
sharpe_annualized: Decimal | None
skip_reasons: dict[str, int]
trades: list[CompletedTrade]
def _build_entry_from_pick(
pick: MondayPick,
cfg: StrategyConfig,
*,
capital_usd: Decimal,
eur_to_usd: Decimal,
) -> BacktestEntry | None:
snap = pick.snapshot
if snap.spot is None or snap.dvol is None:
return None
spot = float(snap.spot)
dvol = float(snap.dvol)
width_pct = float(cfg.structure.spread_width.target_pct_of_spot)
delta_target = float(cfg.structure.short_strike.delta_target)
dte = cfg.structure.dte_target
credit_eth, short_strike, long_strike = estimate_credit_eth(
spot=spot, dvol_pct=dvol, dte_days=dte,
width_pct=width_pct, delta_target_abs=delta_target,
)
width_usd = float(cfg.structure.spread_width.target_pct_of_spot) * spot
credit_usd = credit_eth * spot
if width_usd <= 0 or credit_usd / width_usd < float(
cfg.structure.credit_to_width_ratio_min
):
return None # ratio gate fallisce → no entry
cap_pertrade_usd = float(cfg.sizing.cap_per_trade_eur) * float(eur_to_usd)
risk_target = min(float(cfg.sizing.kelly_fraction) * float(capital_usd), cap_pertrade_usd)
n_contracts = max(0, min(int(risk_target // width_usd), cfg.sizing.max_contracts_per_trade))
if n_contracts == 0:
return None
expiry = pick.timestamp + timedelta(days=dte)
return BacktestEntry(
timestamp=pick.timestamp,
spread_type="bull_put",
spot_at_entry=Decimal(str(spot)),
dvol_at_entry=Decimal(str(dvol)),
short_strike=Decimal(str(round(short_strike, 2))),
long_strike=Decimal(str(round(long_strike, 2))),
expiry=expiry,
credit_received_eth=Decimal(str(credit_eth)),
credit_received_usd=Decimal(str(credit_usd * n_contracts)),
n_contracts=n_contracts,
)
def _max_drawdown_usd(equity: list[Decimal]) -> tuple[Decimal, Decimal]:
"""Return ``(max_dd_usd, max_dd_pct_of_peak)`` over an equity curve."""
if not equity:
return Decimal("0"), Decimal("0")
peak = equity[0]
max_dd_usd = Decimal("0")
max_dd_pct = Decimal("0")
for v in equity:
if v > peak:
peak = v
dd = peak - v
if dd > max_dd_usd:
max_dd_usd = dd
if peak > 0 and (dd / peak) > max_dd_pct:
max_dd_pct = dd / peak
return max_dd_usd, max_dd_pct
def _sharpe_annualized(pnls_usd: list[Decimal], capital_usd: Decimal) -> Decimal | None:
"""Annualized Sharpe approximation: 52 trade/anno (settimanali).
Restituisce ``None`` se ci sono <5 trade o stdev = 0.
"""
if len(pnls_usd) < 5 or capital_usd <= 0:
return None
rets = [float(p / capital_usd) for p in pnls_usd]
mean = sum(rets) / len(rets)
var = sum((r - mean) ** 2 for r in rets) / max(1, (len(rets) - 1))
std = math.sqrt(var)
if std == 0:
return None
sharpe = mean / std * math.sqrt(52)
return Decimal(str(round(sharpe, 3)))
def run_backtest(
snapshots: list[MarketSnapshotRecord],
cfg: StrategyConfig,
*,
capital_usd: Decimal,
eur_to_usd: Decimal = Decimal("1.075"),
asset: str = "ETH",
) -> BacktestReport:
"""Esegue il backtest end-to-end sui ``snapshots`` ETH ordinati per ts."""
snapshots = sorted(snapshots, key=lambda s: s.timestamp)
eth_snapshots = [s for s in snapshots if s.asset.upper() == asset.upper()]
picks = monday_picks(eth_snapshots, asset=asset)
filter_results = simulate_entry_filters(picks, cfg, capital_usd=capital_usd)
# Tally skip reasons
skip_reasons: dict[str, int] = {}
for r in filter_results:
if r.accepted:
continue
for reason in r.reasons:
skip_reasons[reason] = skip_reasons.get(reason, 0) + 1
trades: list[CompletedTrade] = []
for r in filter_results:
if not r.accepted:
continue
entry = _build_entry_from_pick(
r.pick, cfg, capital_usd=capital_usd, eur_to_usd=eur_to_usd,
)
if entry is None:
skip_reasons["sizing_or_ratio"] = skip_reasons.get("sizing_or_ratio", 0) + 1
continue
future = [s for s in eth_snapshots if s.timestamp > r.pick.timestamp]
exit_ = simulate_position_outcome(entry, future, cfg)
trades.append(CompletedTrade(entry=entry, exit=exit_))
pnls = [t.exit.pnl_usd for t in trades]
cumulative = sum(pnls, start=Decimal("0"))
n_winners = sum(1 for p in pnls if p > 0)
win_rate = (
Decimal(n_winners) / Decimal(len(pnls))
if pnls
else Decimal("0")
)
# Equity curve in USD assoluti
equity = [capital_usd]
for p in pnls:
equity.append(equity[-1] + p)
max_dd_usd, max_dd_pct = _max_drawdown_usd(equity)
return BacktestReport(
n_picks=len(picks),
n_accepted=sum(1 for r in filter_results if r.accepted),
n_skipped_data=sum(1 for r in filter_results if r.skipped_for_data),
n_completed=len(trades),
n_winners=n_winners,
win_rate=win_rate,
cumulative_pnl_usd=cumulative,
cumulative_pnl_pct_of_capital=(
cumulative / capital_usd if capital_usd > 0 else Decimal("0")
),
max_drawdown_usd=max_dd_usd,
max_drawdown_pct=max_dd_pct,
sharpe_annualized=_sharpe_annualized(pnls, capital_usd),
skip_reasons=skip_reasons,
trades=trades,
)
@@ -0,0 +1,648 @@
"""Strategia page — documento operativo + lettura live dei segnali.
Renderizza il documento canonico ``docs/13-strategia-spiegata.md`` e
sopra di esso un pannello che mostra l'ultimo tick di
``market_snapshots`` confrontato con le soglie di ``strategy.yaml``.
Lo scopo è far vedere subito, ogni volta che si apre la pagina:
"a cosa serve il dato che il bot sta raccogliendo adesso".
La pagina è di sola lettura: non chiama MCP, non scrive sul DB.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from pathlib import Path
import streamlit as st
from cerbero_bite.config.loader import load_strategy
from cerbero_bite.gui.data_layer import (
DEFAULT_DB_PATH,
humanize_dt,
load_market_snapshots,
)
from cerbero_bite.state.models import MarketSnapshotRecord
_DOC_FILENAME = "13-strategia-spiegata.md"
_DOC_CANDIDATES: tuple[Path, ...] = (
Path("/app/docs") / _DOC_FILENAME, # in-container shipped via Dockerfile
Path(__file__).resolve().parents[4] / "docs" / _DOC_FILENAME, # repo dev
Path(__file__).resolve().parents[3] / "docs" / _DOC_FILENAME,
)
def _resolve_db() -> Path:
return Path(os.environ.get("CERBERO_BITE_GUI_DB", DEFAULT_DB_PATH))
def _load_doc() -> str | None:
for candidate in _DOC_CANDIDATES:
if candidate.is_file():
try:
return candidate.read_text(encoding="utf-8")
except OSError:
continue
return None
@dataclass(frozen=True)
class _GateRow:
label: str
value: str
threshold: str
status: str # "pass" | "fail" | "n/a"
note: str = ""
def _fmt_decimal(v: object, *, fmt: str = "{:.4g}", suffix: str = "") -> str:
if v is None:
return ""
try:
return fmt.format(float(v)) + suffix
except (TypeError, ValueError):
return ""
def _build_gates(
snap: MarketSnapshotRecord, strategy: object
) -> list[_GateRow]:
"""Costruisce le righe del pannello live dai gate §2 della strategia."""
rows: list[_GateRow] = []
entry = getattr(strategy, "entry", None)
structure = getattr(strategy, "structure", None)
# --- DVOL band -------------------------------------------------
dvol_min = float(getattr(entry, "dvol_min", 35.0)) if entry else 35.0
dvol_max = float(getattr(entry, "dvol_max", 90.0)) if entry else 90.0
dvol_v = float(snap.dvol) if snap.dvol is not None else None
if dvol_v is None:
rows.append(
_GateRow(
"DVOL in banda 3590",
"",
f"{dvol_min:.0f} ≤ DVOL ≤ {dvol_max:.0f}",
"n/a",
"Dato non disponibile in questo tick.",
)
)
else:
ok = dvol_min <= dvol_v <= dvol_max
rows.append(
_GateRow(
"DVOL in banda",
f"{dvol_v:.2f}",
f"{dvol_min:.0f}{dvol_max:.0f}",
"pass" if ok else "fail",
"Premio adeguato e regime non-stress."
if ok
else "Sotto banda = premio magro; sopra = stress, no entry.",
)
)
# --- Funding perp annualized ----------------------------------
fund_max = (
float(getattr(entry, "funding_perp_abs_max_annualized", 0.80))
if entry
else 0.80
)
fp = (
float(snap.funding_perp_annualized)
if snap.funding_perp_annualized is not None
else None
)
if fp is None:
rows.append(
_GateRow(
"Funding perp |·| ≤ soglia",
"",
f"|f| ≤ {fund_max:.0%}",
"n/a",
)
)
else:
ok = abs(fp) <= fund_max
rows.append(
_GateRow(
"Funding perp |·|",
f"{fp:+.2%}",
f"{fund_max:.0%}",
"pass" if ok else "fail",
"Filtra regimi di liquidazioni a cascata imminenti.",
)
)
# --- Cross-exchange funding (bias) ---------------------------
bull_th = (
float(getattr(entry, "funding_bull_threshold_annualized", 0.20))
if entry
else 0.20
)
bear_th = (
float(getattr(entry, "funding_bear_threshold_annualized", -0.20))
if entry
else -0.20
)
fc = (
float(snap.funding_cross_annualized)
if snap.funding_cross_annualized is not None
else None
)
if fc is None:
bias_funding = ""
rows.append(
_GateRow(
"Funding cross (bias)",
"",
f"bull ≥ {bull_th:+.0%} · bear ≤ {bear_th:+.0%}",
"n/a",
)
)
else:
if fc >= bull_th:
bias_funding = "BULL"
elif fc <= bear_th:
bias_funding = "BEAR"
else:
bias_funding = "NEUTRO"
rows.append(
_GateRow(
"Funding cross (bias)",
f"{fc:+.2%}{bias_funding}",
f"bull ≥ {bull_th:+.0%} · bear ≤ {bear_th:+.0%}",
"pass" if bias_funding != "NEUTRO" else "fail",
"Mediana 4 maggiori exchange. Discordante col trend = no entry.",
)
)
# --- Macro days to event --------------------------------------
dte_target = (
int(getattr(structure, "dte_target", 18)) if structure else 18
)
macro_d = snap.macro_days_to_event
if macro_d is None:
rows.append(
_GateRow(
"Macro fuori finestra DTE",
"nessun evento",
f"> {dte_target}g",
"pass",
"Nessun evento ad alta severità entro la scadenza target.",
)
)
else:
ok = macro_d > dte_target
rows.append(
_GateRow(
"Macro fuori finestra DTE",
f"{macro_d} g al prossimo",
f"> {dte_target} g",
"pass" if ok else "fail",
"FOMC/CPI/NFP/ECB/Powell entro DTE = no entry.",
)
)
# --- Dealer gamma ---------------------------------------------
gamma_min = (
float(getattr(entry, "dealer_gamma_min", 0.0)) if entry else 0.0
)
gamma_enabled = (
bool(getattr(entry, "dealer_gamma_filter_enabled", True))
if entry
else True
)
g = (
float(snap.dealer_net_gamma)
if snap.dealer_net_gamma is not None
else None
)
if not gamma_enabled:
rows.append(
_GateRow(
"Dealer gamma filter",
_fmt_decimal(g, fmt="{:,.0f}", suffix=" USD")
if g is not None
else "",
"filtro DISABILITATO",
"n/a",
)
)
elif g is None:
rows.append(
_GateRow(
"Dealer net gamma > soglia",
"",
f"> {gamma_min:,.0f} USD",
"n/a",
)
)
else:
ok = g > gamma_min
rows.append(
_GateRow(
"Dealer net gamma",
f"{g:,.0f} USD",
f"> {gamma_min:,.0f} USD",
"pass" if ok else "fail",
"Long-gamma regime sopprime la vol → ideale per vendere spread.",
)
)
# --- Liquidation risks ----------------------------------------
liq_enabled = (
bool(getattr(entry, "liquidation_filter_enabled", True))
if entry
else True
)
long_r = snap.liquidation_long_risk or ""
short_r = snap.liquidation_short_risk or ""
lr_status = "n/a"
if liq_enabled and snap.liquidation_long_risk and snap.liquidation_short_risk:
worst = max(
("low", "med", "high").index(snap.liquidation_long_risk)
if snap.liquidation_long_risk in ("low", "med", "high")
else 0,
("low", "med", "high").index(snap.liquidation_short_risk)
if snap.liquidation_short_risk in ("low", "med", "high")
else 0,
)
lr_status = "fail" if worst == 2 else "pass"
rows.append(
_GateRow(
"Liquidation risk (long / short)",
f"{long_r} / {short_r}",
"non `high`" if liq_enabled else "filtro DISABILITATO",
lr_status,
"Densità liquidazioni vicine al spot. `high` su un lato = scarta setup.",
)
)
# --- IV RV (richness) — solo informativo --------------------
rv = (
float(snap.realized_vol_30d) if snap.realized_vol_30d is not None else None
)
iv_minus_rv = (
float(snap.iv_minus_rv) if snap.iv_minus_rv is not None else None
)
rows.append(
_GateRow(
"IV RV (richness)",
(
f"{iv_minus_rv:+.2f} pt vol"
if iv_minus_rv is not None
else ""
),
"info, > 0 = premio ricco",
"pass" if (iv_minus_rv is not None and iv_minus_rv > 0) else "n/a",
f"RV30={rv:.2f}" if rv is not None else "",
)
)
return rows
def _render_gates(rows: list[_GateRow]) -> None:
icons = {"pass": "", "fail": "", "n/a": ""}
for r in rows:
icon = icons.get(r.status, "")
col1, col2, col3 = st.columns([4, 4, 4])
col1.markdown(f"{icon} **{r.label}**")
col2.markdown(f"`{r.value}`")
col3.markdown(f"_{r.threshold}_")
if r.note:
st.caption(r.note)
st.divider()
def _profile_caps(strategy: object | None) -> dict[str, float]:
"""Estrae le sole leve di sizing da una strategia (o usa default conservativi)."""
out = {
"cap_pertrade_eur": 200.0,
"cap_aggregate_eur": 1000.0,
"kelly": 0.13,
"max_n": 4.0,
"max_concurrent": 1.0,
"width_pct": 0.04,
"credit_ratio": 0.30,
"profit_take": 0.50,
"stop_mult": 2.50,
}
if strategy is None:
return out
try:
out["cap_pertrade_eur"] = float(strategy.sizing.cap_per_trade_eur) # type: ignore[attr-defined]
out["cap_aggregate_eur"] = float(strategy.sizing.cap_aggregate_open_eur) # type: ignore[attr-defined]
out["kelly"] = float(strategy.sizing.kelly_fraction) # type: ignore[attr-defined]
out["max_n"] = float(strategy.sizing.max_contracts_per_trade) # type: ignore[attr-defined]
out["max_concurrent"] = float(strategy.sizing.max_concurrent_positions) # type: ignore[attr-defined]
out["width_pct"] = float(strategy.structure.spread_width.target_pct_of_spot) # type: ignore[attr-defined]
out["credit_ratio"] = float(strategy.structure.credit_to_width_ratio_min) # type: ignore[attr-defined]
out["profit_take"] = float(strategy.exit.profit_take_pct_of_credit) # type: ignore[attr-defined]
out["stop_mult"] = float(strategy.exit.stop_loss_mark_x_credit) # type: ignore[attr-defined]
except Exception:
pass
return out
def _compute_pl(
caps: dict[str, float],
*,
capital: float,
spot: float,
win_rate: float,
trades_per_year: int,
eur_to_usd: float = 1.075,
) -> dict[str, float]:
"""Calcola le metriche P/L per un profilo di sizing."""
width = caps["width_pct"] * spot
credit = caps["credit_ratio"] * width
tp_profit = caps["profit_take"] * credit
sl_loss = (caps["stop_mult"] - 1.0) * credit
cap_pertrade_usd = caps["cap_pertrade_eur"] * eur_to_usd
risk_target = min(caps["kelly"] * capital, cap_pertrade_usd)
n_kelly = int(risk_target // width) if width > 0 else 0
n_per_trade = max(0, min(n_kelly, int(caps["max_n"])))
prob_time_stop = 0.07
prob_other_stop = 0.03
prob_loss = max(0.0, 1.0 - win_rate - prob_time_stop - prob_other_stop)
avg_time_stop_pl = 0.10 * credit
e_trade_gross = (
win_rate * tp_profit
- prob_loss * sl_loss
+ prob_time_stop * avg_time_stop_pl
)
fees = 0.0003 * spot * 2
slippage = 0.03 * credit
e_trade_net = e_trade_gross - fees - slippage
# Multi-posizione concorrente: il P/L scala col numero di posizioni
# aperte simultaneamente (il loop entry crea N trade indipendenti
# quando max_concurrent > 1). Vedi caveat aggressiva.yaml: il
# supporto multi-asset richiede modifiche di codice; questo
# moltiplicatore stima cosa otterresti DOPO.
concurrency = max(1.0, caps["max_concurrent"])
annual_pl = trades_per_year * n_per_trade * concurrency * e_trade_net
apr = (annual_pl / capital) if capital > 0 else 0.0
return {
"width": width,
"credit": credit,
"tp_profit": tp_profit,
"sl_loss": sl_loss,
"risk_target": risk_target,
"n_per_trade": float(n_per_trade),
"concurrency": concurrency,
"e_trade_net": e_trade_net,
"annual_pl": annual_pl,
"apr": apr,
"fees": fees,
"slippage": slippage,
}
def _render_profile_card(
label: str,
caps: dict[str, float],
metrics: dict[str, float],
badge: str,
) -> None:
"""Rendering di un profilo (conservativo o aggressivo) in una colonna."""
st.markdown(f"### {label} {badge}")
st.caption(
f"cap/trade {caps['cap_pertrade_eur']:.0f} EUR · "
f"cap aggreg. {caps['cap_aggregate_eur']:.0f} EUR · "
f"max {caps['max_n']:.0f} contratti × "
f"{caps['max_concurrent']:.0f} pos. concorrenti"
)
cols = st.columns(2)
cols[0].metric("Contratti per trade", f"{metrics['n_per_trade']:.0f}")
cols[1].metric("Posizioni concorrenti", f"{metrics['concurrency']:.0f}")
cols = st.columns(2)
cols[0].metric(
"E[trade] netto",
f"{metrics['e_trade_net']:+.1f} USD",
help=(
f"fees={metrics['fees']:.2f} USD, "
f"slippage={metrics['slippage']:.2f} USD"
),
)
cols[1].metric(
"P/L annuo stimato",
f"{metrics['annual_pl']:+.0f} USD",
delta=f"{metrics['apr']:+.1%} APR",
)
if metrics["n_per_trade"] == 0:
st.warning(
"Sizing 0 contratti: capitale insufficiente per i cap di "
"questo profilo."
)
def _render_pl_panel(
strategy_main: object | None,
strategy_conservativa: object | None,
strategy_aggressiva: object | None,
) -> None:
"""Pannello P/L: confronto Conservativa vs Aggressiva sugli stessi slider."""
st.subheader("💰 P/L atteso — Conservativa vs Aggressiva")
st.caption(
"Stessi slider, due profili di sizing. **Conservativa** = la "
"golden config attuale (`strategy.yaml`). **Aggressiva** = "
"`strategy.aggressiva.yaml` con cap_per_trade 4×, max contratti "
"4×, 2 posizioni concorrenti. Le regole §2-§9 sono identiche; "
"cambiano SOLO le leve di sizing — quello che il P/L "
"conservativo lascia sul tavolo."
)
col_a, col_b, col_c, col_d = st.columns(4)
capital = col_a.slider(
"Capitale (USD)", 720, 50_000, value=10_000, step=100
)
spot = col_b.slider("Spot ETH (USD)", 1500, 6000, value=3000, step=100)
win_rate = col_c.slider(
"Win rate atteso", 0.50, 0.90, value=0.75, step=0.01,
help=(
"Senza filtri quant ≈ 0.650.70. CON filtri (dealer gamma>0, "
"no macro, IVRV>0, liquidation_*_risk≠high) sale a 0.750.80."
),
)
trades_per_year = col_d.slider(
"Trade / anno (post-filtri)", 8, 30, value=18, step=1,
help="52 lunedì × probabilità di superare i filtri (3050%).",
)
cons_caps = _profile_caps(strategy_conservativa or strategy_main)
aggr_caps = _profile_caps(strategy_aggressiva)
cons = _compute_pl(
cons_caps,
capital=capital,
spot=spot,
win_rate=win_rate,
trades_per_year=trades_per_year,
)
aggr = _compute_pl(
aggr_caps,
capital=capital,
spot=spot,
win_rate=win_rate,
trades_per_year=trades_per_year,
)
col_cons, col_aggr = st.columns(2)
with col_cons:
_render_profile_card(
"🛡️ Conservativa",
cons_caps,
cons,
"_(golden config v1.0.0)_",
)
with col_aggr:
_render_profile_card(
"🔥 Aggressiva",
aggr_caps,
aggr,
"_(deroga §11, richiede paper trading)_",
)
if aggr["annual_pl"] > 0 and cons["annual_pl"] > 0:
ratio = aggr["annual_pl"] / cons["annual_pl"]
st.success(
f"Profilo aggressivo: P/L atteso ≈ **{ratio:.1f}× il "
f"conservativo** ({aggr['apr']:+.1%} vs {cons['apr']:+.1%} "
"APR). Drawdown atteso scala con lo stesso fattore."
)
if win_rate < 0.72:
st.error(
"**Win rate sotto 0.72: entrambi i profili perdono soldi.** "
"Selling vol nudo è strutturalmente neutro qui. L'edge della "
"strategia sono i FILTRI (dealer gamma>0, no macro, "
"liquidation≠high, bias chiaro) che alzano il win rate sopra "
"il 0.75. Senza filtri attivi nessuno dei due profili è "
"viable."
)
# Sensibilità win-rate per il profilo aggressivo (più informativo)
st.markdown("**Sensibilità al win rate** (profilo aggressivo)")
sens_rows = []
for wr in (0.65, 0.70, 0.72, 0.75, 0.78, 0.80, 0.82):
m_a = _compute_pl(
aggr_caps,
capital=capital,
spot=spot,
win_rate=wr,
trades_per_year=trades_per_year,
)
m_c = _compute_pl(
cons_caps,
capital=capital,
spot=spot,
win_rate=wr,
trades_per_year=trades_per_year,
)
sens_rows.append(
{
"Win rate": f"{wr:.0%}",
"Conservativa P/L": f"{m_c['annual_pl']:+.0f} USD",
"Conservativa APR": f"{m_c['apr']:+.1%}",
"Aggressiva P/L": f"{m_a['annual_pl']:+.0f} USD",
"Aggressiva APR": f"{m_a['apr']:+.1%}",
}
)
st.table(sens_rows)
st.caption(
"Costi: fee 0.03% notional × 2 leg, slippage 3% del credito "
"(combo limit GTC al mid). Distribuzione esiti: profit-take = "
"win_rate, time-stop ≈ 7%, altri-stop ≈ 3%, stop-loss = il resto. "
"**Multi-asset (ETH+BTC) non è incluso nei numeri**: richiede "
"modifiche di codice (single-asset attuale). Il moltiplicatore "
"2× citato nel doc è la stima ex-ante di cosa otterresti DOPO."
)
def render() -> None:
st.title("📚 Strategia")
st.caption(
"Documento operativo che lega ogni regola del rule engine al "
"dato osservabile da cui dipende. Il pannello live in alto mostra "
"l'ultimo tick di `market_snapshots` confrontato con le soglie di "
"`strategy.yaml`."
)
db_path = _resolve_db()
asset = st.selectbox("Asset", options=["ETH", "BTC"], index=0)
records = load_market_snapshots(asset=asset, db_path=db_path, limit=1)
def _try_load(name: str) -> object | None:
for base in (Path("/app"), Path.cwd(), Path(__file__).resolve().parents[4]):
path = base / name
if path.is_file():
try:
# `_profile_caps` legge `.sizing.*` direttamente sul
# `StrategyConfig`, non sul wrapper `LoadedConfig`.
return load_strategy(path).config
except Exception as exc:
st.warning(
f"`{name}`: {type(exc).__name__}: {exc}"
)
return None
return None
strategy = _try_load("strategy.yaml")
strategy_conservativa = _try_load("strategy.conservativa.yaml")
strategy_aggressiva = _try_load("strategy.aggressiva.yaml")
st.divider()
st.subheader("📡 Stato live dei gate di entry §2")
if not records:
st.info(
"Nessuno snapshot disponibile per "
f"`{asset}`. Il job `market_snapshot` (cron `*/15`) deve "
"girare almeno una volta. Engine attivo? Controlla la pagina "
"`📊 Status`."
)
else:
latest = records[0]
st.caption(
f"Ultimo tick: {humanize_dt(latest.timestamp)} · "
f"asset {latest.asset} · "
f"fetch_ok = {'' if latest.fetch_ok else '⚠️'}"
)
if strategy is None:
st.warning(
"Senza `strategy.yaml` non posso valutare i gate; mostro "
"solo i valori grezzi."
)
st.json(latest.model_dump(mode="json"))
else:
rows = _build_gates(latest, strategy)
_render_gates(rows)
st.divider()
_render_pl_panel(strategy, strategy_conservativa, strategy_aggressiva)
st.divider()
st.subheader("📖 Documento esteso")
doc = _load_doc()
if doc is None:
st.error(
"Documento `docs/13-strategia-spiegata.md` non trovato. In "
"locale verifica il path; in container assicurati che il "
"Dockerfile copi `docs/` in `/app/docs/`."
)
else:
st.markdown(doc, unsafe_allow_html=False)
render()
+180
View File
@@ -0,0 +1,180 @@
# strategy.aggressiva.yaml — Cerbero Bite, profilo AGGRESSIVO
#
# Profilo "crescita: rendimenti significativi, drawdown a doppia cifra,
# complessità più alta". DEROGA esplicitamente alla sezione §11
# "Riepilogo soglie" di docs/01-strategy-rules.md (cap_per_trade_eur,
# cap_aggregate_open_eur, max_concurrent_positions). NON va deployato
# senza:
# 1. backtest dedicato sui dati raccolti
# 2. paper trading per almeno 4 settimane
# 3. autorizzazione esplicita scritta nel commit
#
# Caratteristiche operative attese (vs. profilo conservativo):
# * Cap per trade 800 EUR (~860 USD) → 4× la size
# * Cap aggregato 3 200 EUR (~3 440 USD) → 4× il rischio aggregato
# * Max 2 posizioni concorrenti (era 1)
# * Max 16 contratti per trade (era 4)
# * P/L stimato: +5% / +20% APR sul capitale impiegato
# * Drawdown atteso: 2540% del capitale impiegato in streak
# * Adatto a: capitale "growth", non parcheggio
#
# CAVEAT MULTI-ASSET. Il rule engine attuale è single-asset
# (`asset.symbol`). Per estendere a ETH+BTC servono modifiche di
# codice in:
# * cerbero_bite/runtime/entry_cycle.py (loop su lista asset)
# * cerbero_bite/state/repository.py (multi-position per asset)
# * cerbero_bite/runtime/orchestrator.py (scheduler one-asset → N)
# Nel frattempo il file resta single-asset ETH; il moltiplicatore
# 2× via "ETH + BTC" indicato in `📚 Strategia` è una **stima ex-ante**
# di cosa otterresti DOPO quel lavoro di codice.
config_version: "1.0.0-aggressiva"
config_hash: "b931a2b96fbc149b21cae84a196ee8bad10220b5ee8fa9ab0ed06ae52d7dc531"
last_review: "2026-04-26"
last_reviewer: "Adriano"
asset:
symbol: "ETH"
exchange: "deribit"
entry:
cron: "0 14 * * MON"
skip_holidays_country: "IT"
capital_min_usd: "2880" # 4× del minimo conservativo (720)
dvol_min: "35"
dvol_max: "90"
funding_perp_abs_max_annualized: "0.80"
eth_holdings_pct_max: "0.30"
no_position_concurrent: false # consenti N posizioni concorrenti
exclude_macro_severity: ["high"]
exclude_macro_countries: ["US", "EU"]
trend_window_days: 30
trend_bull_threshold_pct: "0.05"
trend_bear_threshold_pct: "-0.05"
funding_bull_threshold_annualized: "0.20"
funding_bear_threshold_annualized: "-0.20"
iron_condor_dvol_min: "55"
iron_condor_adx_max: "20"
iron_condor_trend_neutral_band_pct: "0.05"
# Filtri quant invariati: l'edge della strategia E' qui, non
# serve allentarli per "guadagnare di più" — anzi sarebbe
# controproducente.
dealer_gamma_min: "0"
dealer_gamma_filter_enabled: true
liquidation_filter_enabled: true
structure:
dte_target: 18
dte_min: 14
dte_max: 21
short_strike:
delta_target: "0.12"
delta_min: "0.10"
delta_max: "0.15"
distance_otm_pct_min: "0.15"
distance_otm_pct_max: "0.25"
spread_width:
target_pct_of_spot: "0.04"
min_pct_of_spot: "0.03"
max_pct_of_spot: "0.05"
credit_to_width_ratio_min: "0.30"
liquidity:
open_interest_min: 100
volume_24h_min: 20
bid_ask_spread_pct_max: "0.15"
book_depth_top3_min: 5
slippage_pct_of_credit_max: "0.08"
sizing:
kelly_fraction: "0.13" # disciplina Kelly invariata
# Le tre leve dominanti:
cap_per_trade_eur: "800" # era 200 → 4×
cap_aggregate_open_eur: "3200" # era 1000 → 4× (proporzionato a 2 posizioni × cap_per_trade × 2 ruote)
max_concurrent_positions: 2 # era 1
max_contracts_per_trade: 16 # era 4 → 4×
dvol_adjustment:
- {dvol_under: "45", multiplier: "1.00"}
- {dvol_under: "60", multiplier: "0.85"}
- {dvol_under: "80", multiplier: "0.65"}
dvol_no_entry_threshold: "80"
exit:
profit_take_pct_of_credit: "0.50"
stop_loss_mark_x_credit: "2.50"
vol_stop_dvol_increase: "10"
time_stop_dte_remaining: 7
time_stop_skip_if_close_to_profit_pct: "0.70"
delta_breach_threshold: "0.30"
adverse_move_4h_pct: "0.05"
monitor_cron: "0 2,14 * * *"
user_confirmation_timeout_min: 30
escalate_on_timeout:
- "CLOSE_STOP"
- "CLOSE_VOL"
- "CLOSE_DELTA"
execution:
environment: "testnet"
eur_to_usd: "1.075"
combo_only: true
initial_limit: "mid"
reprice_step_ticks: 1
reprice_max_steps: 3
reprice_max_steps_urgent: 5
order_tif: "GTC"
order_expiry_min: 30
ack_timeout_s: 300
monitoring:
health_check_interval_s: 300
health_failures_before_kill: 3
health_failures_before_restart: 5
daily_digest_cron: "0 8 * * *"
monthly_report_cron: "0 12 1 * *"
storage:
sqlite_path: "data/state.sqlite"
log_path: "data/log/"
log_retention_days: 365
backup_path: "data/backups/"
backup_retention_days: 30
mcp:
config_file: "~/.config/cerbero-suite/mcp.json"
call_timeout_s: 8
retry_max: 3
retry_base_delay_s: 1
required_versions:
cerbero-deribit: "^2.0.0"
cerbero-hyperliquid: "^1.5.0"
cerbero-memory: "^4.0.0"
cerbero-portfolio: "^1.2.0"
cerbero-macro: "^1.0.0"
cerbero-sentiment: "^1.0.0"
cerbero-telegram: "^1.0.0"
cerbero-brain-bridge: "^1.0.0"
telegram:
parse_mode: "MarkdownV2"
confirmation_timeout_min: 60
exit_confirmation_timeout_min: 30
backup_channel_on_critical: true
kelly_recalibration:
lookback_days: 365
min_sample_low_confidence: 30
min_sample_high_confidence: 100
weight_when_medium_confidence: "0.50"
+164
View File
@@ -0,0 +1,164 @@
# strategy.conservativa.yaml — Cerbero Bite, profilo CONSERVATIVO
#
# Profilo "premio sopra T-bill, drawdown contenuto, complessità minima".
# È identico per regole alla golden config v1.0.0; serve come ancora di
# riferimento per il confronto con `strategy.aggressiva.yaml`.
#
# Caratteristiche operative attese:
# * Cap per trade 200 EUR (~215 USD)
# * Max 1 posizione concorrente
# * P/L stimato: +1.5% / +5% APR sul capitale totale
# * Drawdown atteso: 1020% del capitale impiegato in streak
# * Adatto a: parcheggio capitale, premio modesto, niente sorprese
#
# Ricorda: dopo ogni edit, rigenerare il config_hash con
# cerbero-bite config hash --file strategy.conservativa.yaml
# e bumpare config_version.
config_version: "1.0.0-conservativa"
config_hash: "eff824281bbb538fba49434d8cc4b9c37675bc73d60e351293e263cc7e7b29ef"
last_review: "2026-04-26"
last_reviewer: "Adriano"
asset:
symbol: "ETH"
exchange: "deribit"
entry:
cron: "0 14 * * MON"
skip_holidays_country: "IT"
capital_min_usd: "720"
dvol_min: "35"
dvol_max: "90"
funding_perp_abs_max_annualized: "0.80"
eth_holdings_pct_max: "0.30"
no_position_concurrent: true
exclude_macro_severity: ["high"]
exclude_macro_countries: ["US", "EU"]
trend_window_days: 30
trend_bull_threshold_pct: "0.05"
trend_bear_threshold_pct: "-0.05"
funding_bull_threshold_annualized: "0.20"
funding_bear_threshold_annualized: "-0.20"
iron_condor_dvol_min: "55"
iron_condor_adx_max: "20"
iron_condor_trend_neutral_band_pct: "0.05"
dealer_gamma_min: "0"
dealer_gamma_filter_enabled: true
liquidation_filter_enabled: true
structure:
dte_target: 18
dte_min: 14
dte_max: 21
short_strike:
delta_target: "0.12"
delta_min: "0.10"
delta_max: "0.15"
distance_otm_pct_min: "0.15"
distance_otm_pct_max: "0.25"
spread_width:
target_pct_of_spot: "0.04"
min_pct_of_spot: "0.03"
max_pct_of_spot: "0.05"
credit_to_width_ratio_min: "0.30"
liquidity:
open_interest_min: 100
volume_24h_min: 20
bid_ask_spread_pct_max: "0.15"
book_depth_top3_min: 5
slippage_pct_of_credit_max: "0.08"
sizing:
kelly_fraction: "0.13"
cap_per_trade_eur: "200"
cap_aggregate_open_eur: "1000"
max_concurrent_positions: 1
max_contracts_per_trade: 4
dvol_adjustment:
- {dvol_under: "45", multiplier: "1.00"}
- {dvol_under: "60", multiplier: "0.85"}
- {dvol_under: "80", multiplier: "0.65"}
dvol_no_entry_threshold: "80"
exit:
profit_take_pct_of_credit: "0.50"
stop_loss_mark_x_credit: "2.50"
vol_stop_dvol_increase: "10"
time_stop_dte_remaining: 7
time_stop_skip_if_close_to_profit_pct: "0.70"
delta_breach_threshold: "0.30"
adverse_move_4h_pct: "0.05"
monitor_cron: "0 2,14 * * *"
user_confirmation_timeout_min: 30
escalate_on_timeout:
- "CLOSE_STOP"
- "CLOSE_VOL"
- "CLOSE_DELTA"
execution:
environment: "testnet"
eur_to_usd: "1.075"
combo_only: true
initial_limit: "mid"
reprice_step_ticks: 1
reprice_max_steps: 3
reprice_max_steps_urgent: 5
order_tif: "GTC"
order_expiry_min: 30
ack_timeout_s: 300
monitoring:
health_check_interval_s: 300
health_failures_before_kill: 3
health_failures_before_restart: 5
daily_digest_cron: "0 8 * * *"
monthly_report_cron: "0 12 1 * *"
storage:
sqlite_path: "data/state.sqlite"
log_path: "data/log/"
log_retention_days: 365
backup_path: "data/backups/"
backup_retention_days: 30
mcp:
config_file: "~/.config/cerbero-suite/mcp.json"
call_timeout_s: 8
retry_max: 3
retry_base_delay_s: 1
required_versions:
cerbero-deribit: "^2.0.0"
cerbero-hyperliquid: "^1.5.0"
cerbero-memory: "^4.0.0"
cerbero-portfolio: "^1.2.0"
cerbero-macro: "^1.0.0"
cerbero-sentiment: "^1.0.0"
cerbero-telegram: "^1.0.0"
cerbero-brain-bridge: "^1.0.0"
telegram:
parse_mode: "MarkdownV2"
confirmation_timeout_min: 60
exit_confirmation_timeout_min: 30
backup_channel_on_critical: true
kelly_recalibration:
lookback_days: 365
min_sample_low_confidence: 30
min_sample_high_confidence: 100
weight_when_medium_confidence: "0.50"
+259
View File
@@ -0,0 +1,259 @@
"""TDD per :mod:`cerbero_bite.core.backtest`."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from decimal import Decimal
import pytest
from cerbero_bite.config import StrategyConfig, golden_config
from cerbero_bite.core.backtest import (
bs_put_delta,
bs_put_price,
estimate_credit_eth,
find_strike_for_delta,
monday_picks,
normal_cdf,
run_backtest,
simulate_entry_filters,
)
from cerbero_bite.state.models import MarketSnapshotRecord
# ---------------------------------------------------------------------------
# Black-Scholes helpers
# ---------------------------------------------------------------------------
def test_normal_cdf_known_values() -> None:
assert normal_cdf(0.0) == pytest.approx(0.5, abs=1e-6)
assert normal_cdf(1.0) == pytest.approx(0.8413, abs=1e-3)
assert normal_cdf(-1.0) == pytest.approx(0.1587, abs=1e-3)
assert normal_cdf(2.0) == pytest.approx(0.9772, abs=1e-3)
def test_bs_put_price_atm_positive_and_less_than_strike() -> None:
p = bs_put_price(spot=3000, strike=3000, t_years=18 / 365, sigma=0.50)
assert p > 0
assert p < 3000 # cap
def test_bs_put_price_far_otm_close_to_zero() -> None:
p = bs_put_price(spot=3000, strike=1500, t_years=18 / 365, sigma=0.50)
assert 0 <= p < 5 # essentially zero
def test_bs_put_delta_atm_around_minus_half() -> None:
d = bs_put_delta(spot=3000, strike=3000, t_years=18 / 365, sigma=0.50)
assert d == pytest.approx(-0.475, abs=0.05)
def test_bs_put_delta_far_otm_close_to_zero() -> None:
d = bs_put_delta(spot=3000, strike=1500, t_years=18 / 365, sigma=0.50)
assert -0.05 < d <= 0
def test_find_strike_for_delta_monotone() -> None:
spot = 3000.0
dvol = 50.0
dte = 18
s_010 = find_strike_for_delta(
spot=spot, dvol_pct=dvol, dte_days=dte, target_delta_abs=0.10,
)
s_020 = find_strike_for_delta(
spot=spot, dvol_pct=dvol, dte_days=dte, target_delta_abs=0.20,
)
# |Δ|=0.20 (più ITM) ⇒ strike più alto di |Δ|=0.10 (più OTM).
assert s_020 > s_010
# Verifica che il delta corrisponda a target ± tolleranza.
achieved = abs(
bs_put_delta(
spot=spot, strike=s_020, t_years=dte / 365, sigma=dvol / 100,
)
)
assert achieved == pytest.approx(0.20, abs=0.02)
def test_estimate_credit_returns_positive_credit_in_normal_regime() -> None:
credit_eth, short_k, long_k = estimate_credit_eth(
spot=3000, dvol_pct=50, dte_days=18, width_pct=0.04, delta_target_abs=0.12,
)
# Sanity: credit > 0, short_k < spot, long_k = short_k - 4%×spot
assert credit_eth > 0
assert short_k < 3000
assert long_k < short_k
assert short_k - long_k == pytest.approx(0.04 * 3000, abs=1.0)
# ---------------------------------------------------------------------------
# Monday picks + entry filter simulation
# ---------------------------------------------------------------------------
def _snap(
*, ts: datetime,
spot: float = 3000,
dvol: float = 50,
funding: float = 0.0,
macro_d: int | None = None,
asset: str = "ETH",
) -> MarketSnapshotRecord:
return MarketSnapshotRecord(
timestamp=ts,
asset=asset,
spot=Decimal(str(spot)),
dvol=Decimal(str(dvol)),
funding_perp_annualized=Decimal(str(funding)),
funding_cross_annualized=Decimal("0"),
dealer_net_gamma=Decimal("100"),
liquidation_long_risk="low",
liquidation_short_risk="low",
macro_days_to_event=macro_d,
fetch_ok=True,
)
def test_monday_picks_extracts_one_per_iso_week() -> None:
monday_2026_05_04 = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
monday_2026_05_11 = datetime(2026, 5, 11, 14, 0, tzinfo=UTC)
snapshots = [
_snap(ts=monday_2026_05_04),
_snap(ts=monday_2026_05_04 + timedelta(minutes=15)), # not picked
_snap(ts=monday_2026_05_11),
]
picks = monday_picks(snapshots)
assert len(picks) == 2
assert picks[0].timestamp == monday_2026_05_04
assert picks[1].timestamp == monday_2026_05_11
def test_monday_picks_skips_other_days_and_hours() -> None:
snapshots = [
_snap(ts=datetime(2026, 5, 4, 13, 0, tzinfo=UTC)), # Monday 13:00
_snap(ts=datetime(2026, 5, 5, 14, 0, tzinfo=UTC)), # Tuesday 14:00
]
assert monday_picks(snapshots) == []
def test_monday_picks_filters_by_asset() -> None:
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
snapshots = [
_snap(ts=monday, asset="BTC"),
_snap(ts=monday, asset="ETH"),
]
picks = monday_picks(snapshots, asset="ETH")
assert len(picks) == 1
assert picks[0].snapshot.asset == "ETH"
def test_simulate_entry_filters_accepts_clean_snapshot(
) -> None:
cfg: StrategyConfig = golden_config()
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
snap = _snap(ts=monday, dvol=50, funding=0.10)
picks = [
type("MP", (), {"timestamp": monday, "snapshot": snap})() # type: ignore[arg-type]
]
# Hack: build via real dataclass
from cerbero_bite.core.backtest import MondayPick
picks = [MondayPick(timestamp=monday, snapshot=snap)]
results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500"))
assert len(results) == 1
assert results[0].accepted is True
def test_simulate_entry_filters_rejects_dvol_out_of_band() -> None:
cfg = golden_config()
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
snap = _snap(ts=monday, dvol=20, funding=0.10) # below 35
from cerbero_bite.core.backtest import MondayPick
picks = [MondayPick(timestamp=monday, snapshot=snap)]
results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500"))
assert results[0].accepted is False
assert any("dvol" in r.lower() for r in results[0].reasons)
def test_simulate_entry_filters_skips_incomplete_snapshot() -> None:
cfg = golden_config()
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
incomplete = MarketSnapshotRecord(
timestamp=monday, asset="ETH", spot=Decimal("3000"),
# dvol=None ⇒ skipped
fetch_ok=False,
)
from cerbero_bite.core.backtest import MondayPick
picks = [MondayPick(timestamp=monday, snapshot=incomplete)]
results = simulate_entry_filters(picks, cfg, capital_usd=Decimal("1500"))
assert results[0].accepted is False
assert results[0].skipped_for_data is True
# ---------------------------------------------------------------------------
# Full pipeline (sintetico)
# ---------------------------------------------------------------------------
def _synthetic_year_of_snapshots(
*,
n_weeks: int = 8,
spot: float = 3000,
dvol: float = 60, # con skew_premium 1.5 ⇒ credit/width ≈ 35% (sopra soglia 30%)
funding: float = 0.10,
) -> list[MarketSnapshotRecord]:
"""Genera N settimane di snapshot sintetici ETH a 4 tick/settimana."""
rows: list[MarketSnapshotRecord] = []
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
for week in range(n_weeks):
base = monday + timedelta(weeks=week)
# Lunedì 14:00 è il pick
rows.append(_snap(ts=base, spot=spot, dvol=dvol, funding=funding))
# Tick intermedi che NON cadono di lunedì alle 14:00:
# offset +1h così vengono ignorati da `monday_picks`.
for d in (2, 8, 14, 19):
rows.append(
_snap(
ts=base + timedelta(days=d, hours=1),
spot=spot * (1 + 0.005 * d), # +0.5% al giorno
dvol=dvol - 1.5 * d, # vol che scende lentamente
funding=funding,
)
)
return rows
def test_run_backtest_produces_report_with_trades() -> None:
# Per il test scaliamo il credit/width gate al 15%: il modello BS
# senza skew completo sottostima i premi OTM rispetto al reale.
# Vedi `estimate_credit_eth.skew_premium` docstring per dettagli.
from cerbero_bite.config.schema import StructureConfig
cfg = golden_config()
cfg = cfg.model_copy(
update={
"structure": StructureConfig(
**{
**cfg.structure.model_dump(),
"credit_to_width_ratio_min": Decimal("0.15"),
}
)
}
)
snapshots = _synthetic_year_of_snapshots(n_weeks=4)
report = run_backtest(snapshots, cfg, capital_usd=Decimal("1500"))
# Sanity: 4 picks, almeno 1 trade chiuso
assert report.n_picks == 4
assert report.n_completed >= 1
assert report.cumulative_pnl_usd != Decimal("0")
# Bull-put + ETH al rialzo + DVOL che scende ⇒ atteso win
assert report.n_winners >= 1
def test_run_backtest_handles_no_picks_gracefully() -> None:
cfg = golden_config()
# Solo tick infrasettimanali, niente Monday 14:00.
monday = datetime(2026, 5, 4, 14, 0, tzinfo=UTC)
snapshots = [_snap(ts=monday + timedelta(hours=1))]
report = run_backtest(snapshots, cfg, capital_usd=Decimal("1500"))
assert report.n_picks == 0
assert report.n_completed == 0
assert report.cumulative_pnl_usd == Decimal("0")