Multi_Swarm_Coevolutive/scripts/multi_tick_strategies.py

"""2 winner cross-tick: BTC 238e4812 + ETH c04dff7086 su 5m / 15m / 1h.

Per ogni combinazione strategy × timeframe: backtest year-by-year (2019-2025)
con metriche per-anno e totale 7y.
"""
from __future__ import annotations

from datetime import datetime
from pathlib import Path

from multi_swarm_core.backtest.engine import BacktestEngine
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.metrics.basic import max_drawdown, sharpe_ratio, total_return
from multi_swarm_core.protocol.compiler import compile_strategy
from multi_swarm_core.protocol.parser import parse_strategy


WINNERS = [
    # (label, path, symbol)
    ("BTC NEW (238e4812, native=1h)", "btc_238e4812.json", "BTC-PERPETUAL"),
    ("ETH NEW (c04dff7086, native=5m)", "eth_c04dff7086.json", "ETH-PERPETUAL"),
]
TIMEFRAMES = ["5m", "15m", "1h"]

YEARS = [
    ("2019", "2019-01-01T00:00:00+00:00", "2020-01-01T00:00:00+00:00"),
    ("2020", "2020-01-01T00:00:00+00:00", "2021-01-01T00:00:00+00:00"),
    ("2021", "2021-01-01T00:00:00+00:00", "2022-01-01T00:00:00+00:00"),
    ("2022", "2022-01-01T00:00:00+00:00", "2023-01-01T00:00:00+00:00"),
    ("2023", "2023-01-01T00:00:00+00:00", "2024-01-01T00:00:00+00:00"),
    ("2024", "2024-01-01T00:00:00+00:00", "2025-01-01T00:00:00+00:00"),
    ("2025", "2025-01-01T00:00:00+00:00", "2026-01-01T00:00:00+00:00"),
]


def evaluate(strat, ohlcv, engine, label, tf) -> None:
    print(f"\n  >>> tick={tf} | {len(ohlcv)} bars")
    print(f"  {'year':<6} {'trades':>7} {'wins':>5} {'losses':>7} {'win%':>6} {'ret':>8} {'maxDD':>7} {'sharpe':>7}")
    sum_ret = 0.0
    sum_trades = 0
    sum_wins = 0
    for year_label, start, end in YEARS:
        mask = (ohlcv.index >= datetime.fromisoformat(start)) & (ohlcv.index < datetime.fromisoformat(end))
        slice_df = ohlcv[mask]
        if len(slice_df) == 0:
            continue
        try:
            signal_fn = compile_strategy(strat)
            signals = signal_fn(slice_df)
            bt = engine.run(slice_df, signals)
        except Exception as e:
            print(f"  {year_label:<6} ERROR: {e}")
            continue
        trades = bt.trades
        n = len(trades)
        wins = [t.net_pnl for t in trades if t.net_pnl > 0]
        losses = [t.net_pnl for t in trades if t.net_pnl <= 0]
        nw, nl = len(wins), len(losses)
        wr = (nw / n * 100) if n else 0.0
        if n > 0:
            notional = float(slice_df["close"].iloc[0])
            eq = (bt.equity_curve / notional) + 1.0
            ret = total_return(eq)
            dd = max_drawdown(eq)
            sr = sharpe_ratio(bt.returns, periods_per_year=8760)
        else:
            ret = dd = sr = 0.0
        print(f"  {year_label:<6} {n:>7} {nw:>5} {nl:>7} {wr:>5.1f}% {ret:>7.2%} {dd:>6.2%} {sr:>7.3f}")
        sum_ret += ret
        sum_trades += n
        sum_wins += nw
    overall_wr = (sum_wins / sum_trades * 100) if sum_trades else 0.0
    print(f"  ===== 7y TOT: {sum_trades:>7} {sum_wins:>5} {sum_trades-sum_wins:>7} {overall_wr:>5.1f}%  cum_ret={sum_ret:+.2%}")


def main() -> None:
    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)
    engine = BacktestEngine(fees_bp=5.0)
    strategies_dir = Path("/app/strategies")

    for label, fname, symbol in WINNERS:
        path = strategies_dir / fname
        strat = parse_strategy(path.read_text())
        print(f"\n{'='*100}")
        print(f">>> {label} — symbol={symbol}")
        for tf in TIMEFRAMES:
            try:
                ohlcv = loader.load(OHLCVRequest(
                    symbol=symbol, timeframe=tf,
                    start=datetime.fromisoformat("2018-09-01T00:00:00+00:00"),
                    end=datetime.fromisoformat("2026-01-01T00:00:00+00:00"),
                ))
                evaluate(strat, ohlcv, engine, label, tf)
            except Exception as e:
                print(f"\n  >>> tick={tf} FAILED TO LOAD: {e}")


if __name__ == "__main__":
    main()