Multi_Swarm_Coevolutive/scripts/yearly_strategies.py

"""Per-year breakdown delle 4 strategie: 2 NEW (BTC 238e4812 + ETH c04dff7086)
+ 2 OLD freezate (btc_9cf506b8 hardened-001 + eth_facd6af85d5d).

Backtest anno-per-anno (2019-2025) sul tick di discovery di ciascuna strategia.
Output: trade, wins/losses, win%, return%, max DD%, Sharpe per ogni anno.
"""
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


STRATEGIES = [
    # (label, path, symbol, timeframe)
    ("BTC NEW (238e4812, paper attuale)", "btc_238e4812.json", "BTC-PERPETUAL", "1h"),
    ("BTC OLD (9cf506b8, hardened-001 prev paper)", "archive/btc_9cf506b8.json", "BTC-PERPETUAL", "1h"),
    ("ETH NEW (c04dff7086, paper attuale)", "eth_c04dff7086.json", "ETH-PERPETUAL", "5m"),
    ("ETH OLD (facd6af85d5d, prev paper)", "archive/eth_facd6af85d5d.json", "ETH-PERPETUAL", "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 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, timeframe in STRATEGIES:
        path = strategies_dir / fname
        strat = parse_strategy(path.read_text())

        # Carica intero range una volta sola
        ohlcv = loader.load(OHLCVRequest(
            symbol=symbol, timeframe=timeframe,
            start=datetime.fromisoformat("2018-09-01T00:00:00+00:00"),
            end=datetime.fromisoformat("2026-01-01T00:00:00+00:00"),
        ))

        print(f"\n{'=' * 110}")
        print(f">>> {label}")
        print(f"    symbol={symbol} timeframe={timeframe} | {len(ohlcv)} bars total")
        print(f"    {'year':<6} {'bars':>6} {'trades':>7} {'wins':>5} {'losses':>7} {'win%':>6} {'avg_w':>10} {'avg_l':>10} {'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
            aw = (sum(wins) / nw) if nw else 0.0
            al = (sum(losses) / nl) if nl 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} {len(slice_df):>6} {n:>7} {nw:>5} {nl:>7} {wr:>5.1f}% {aw:>10.1f} {al:>10.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"    {'='*5} TOTALS 7y:                 {sum_trades:>7} {sum_wins:>5} {sum_trades-sum_wins:>7} {overall_wr:>5.1f}%                       cum_ret={sum_ret:+.2%}")


if __name__ == "__main__":
    main()