Multi_Swarm_Coevolutive/scripts/replay_strategies_window.py

"""Replay diagnostico: per ciascuna strategia conta quanti bar avrebbero
soddisfatto le condizioni di ciascuna regola sull'ultimo `--days` di storico.

Ouput tabellare per branch: total_bars, fires, fire_rate, primo/ultimo fire.
Esegue anche un backtest grezzo (entry-on-signal, exit-on-flat) per stimare
n_trades e total_return realistici nel periodo.

Esempio:
    docker compose exec multi-swarm-paper \
        python /app/scripts/replay_strategies_window.py --days 30
"""

from __future__ import annotations

import argparse
import json
from datetime import UTC, datetime, timedelta
from pathlib import Path

import pandas as pd

from multi_swarm_core.cerbero.client import CerberoClient
from multi_swarm_core.config import load_settings
from multi_swarm_core.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest
from multi_swarm_core.protocol.compiler import _eval_node, compile_strategy
from multi_swarm_core.protocol.parser import parse_strategy

PROJECT_ROOT = Path(__file__).resolve().parent.parent


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser()
    p.add_argument("--days", type=int, default=30)
    p.add_argument("--strategies-dir", default=str(PROJECT_ROOT / "strategies"))
    return p.parse_args()


def fetch_window(loader: CerberoOHLCVLoader, symbol: str, days: int) -> pd.DataFrame:
    end = datetime.now(UTC).replace(minute=0, second=0, microsecond=0)
    start = end - timedelta(days=days)
    req = OHLCVRequest(
        symbol=symbol, timeframe="1h", start=start, end=end, exchange="deribit"
    )
    return loader._fetch(req)  # noqa: SLF001 — bypass cache


def per_branch_fires(strategy_path: Path, ohlcv: pd.DataFrame) -> list[dict]:
    raw = strategy_path.read_text()
    parsed = parse_strategy(raw)
    out = []
    for idx, rule in enumerate(parsed.rules):
        cond_series = _eval_node(rule.condition, ohlcv).fillna(False).astype(bool)
        n = int(cond_series.sum())
        first = ohlcv.index[cond_series.argmax()] if n > 0 else None
        # last fire: argmax on reversed
        last = ohlcv.index[len(cond_series) - 1 - cond_series[::-1].argmax()] if n > 0 else None
        out.append({
            "branch_idx": idx,
            "action": rule.action,
            "fires": n,
            "fire_rate_pct": round(100.0 * n / len(ohlcv), 2),
            "first_fire": first,
            "last_fire": last,
        })
    return out


def quick_pnl(strategy_path: Path, ohlcv: pd.DataFrame, fees_bp: float = 5.0) -> dict:
    """Approx: at each bar evaluate compiled signal series (long/short/flat),
    apply position to next-bar return, charge fees on changes. No leverage."""
    raw = strategy_path.read_text()
    parsed = parse_strategy(raw)
    sig_fn = compile_strategy(parsed)
    signals = sig_fn(ohlcv)  # series of "long"/"short"/"flat"
    # map to position: long=+1, short=-1, flat=0
    pos = signals.map({"long": 1, "short": -1, "flat": 0}).fillna(0).astype(int)
    rets = ohlcv["close"].pct_change().fillna(0.0)
    # next-bar execution: position decided at bar t applies to return t+1 -> shift
    pnl = pos.shift(1).fillna(0) * rets
    # fees on position changes
    changes = pos.diff().abs().fillna(0).astype(int)
    fee_per_change = fees_bp / 10_000.0
    pnl_after_fees = pnl - changes * fee_per_change
    cum = (1 + pnl_after_fees).prod() - 1
    n_trades = int((changes > 0).sum())
    time_in_market = float((pos != 0).mean())
    return {
        "n_trades": n_trades,
        "total_return_pct": round(100.0 * float(cum), 3),
        "time_in_market_pct": round(100.0 * time_in_market, 2),
    }


def main() -> None:
    args = parse_args()
    settings = load_settings()
    token = (
        settings.cerbero_mainnet_token.get_secret_value()
        if settings.cerbero_mainnet_token
        else settings.cerbero_testnet_token.get_secret_value()
    )
    cerbero = CerberoClient(
        base_url=settings.cerbero_base_url,
        token=token,
        bot_tag=settings.cerbero_bot_tag,
    )
    loader = CerberoOHLCVLoader(client=cerbero, cache_dir=settings.series_dir)

    strategies_dir = Path(args.strategies_dir)
    pairs = [
        ("BTC-PERPETUAL", sorted(strategies_dir.glob("btc_*.json"))[0]),
        ("ETH-PERPETUAL", sorted(strategies_dir.glob("eth_*.json"))[0]),
    ]

    for symbol, strat_path in pairs:
        print(f"\n=== {symbol}  strategy={strat_path.name}  window={args.days}d ===")
        ohlcv = fetch_window(loader, symbol, args.days)
        print(f"bars: {len(ohlcv)}  range: {ohlcv.index[0]} -> {ohlcv.index[-1]}")
        print("\n-- per branch --")
        for row in per_branch_fires(strat_path, ohlcv):
            print(json.dumps(row, default=str))
        print("\n-- quick pnl (next-bar exec, fees=5bp) --")
        print(json.dumps(quick_pnl(strat_path, ohlcv), default=str))


if __name__ == "__main__":
    main()