Multi_Swarm_Coevolutive/scripts/analyze_btc_winners.py

"""Analisi per-trade dei top-K candidate del run BTC.

Per ciascun genome top-K, ri-esegue il backtest su ogni fold WFA e raccoglie:
- n_trades, n_wins, n_losses, win_rate
- max_drawdown
- return, sharpe
- list trade pnl summary

Output stampato a stdout, non scrive su disco.
"""
from __future__ import annotations

import argparse
from datetime import datetime

import pandas as pd  # type: ignore[import-untyped]

from multi_swarm_core.agents.hypothesis import _try_parse
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.data.splits import expanding_walk_forward
from multi_swarm_core.metrics.basic import max_drawdown, sharpe_ratio, total_return
from multi_swarm_core.persistence.repository import Repository
from multi_swarm_core.protocol.compiler import compile_strategy


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--run-id", required=True)
    ap.add_argument("--top-k", type=int, default=10)
    ap.add_argument("--n-folds", type=int, default=4)
    ap.add_argument("--train-ratio", type=float, default=0.5)
    ap.add_argument("--symbol", default="BTC-PERPETUAL")
    ap.add_argument("--timeframe", default="1h")
    ap.add_argument("--start", default="2018-09-01T00:00:00+00:00")
    ap.add_argument("--end", default="2026-01-01T00:00:00+00:00")
    ap.add_argument("--fees-bp", type=float, default=5.0)
    args = ap.parse_args()

    settings = load_settings()
    repo = Repository(settings.ga_db_path)
    repo.init_schema()
    all_evals = repo.list_evaluations(args.run_id)
    parseable = [
        e for e in all_evals
        if e.get("raw_text") and not e.get("parse_error") and e["fitness"] > 0
    ]
    parseable.sort(key=lambda e: e["fitness"], reverse=True)
    seen: set[str] = set()
    top: list[dict] = []
    for e in parseable:
        if e["genome_id"] in seen:
            continue
        seen.add(e["genome_id"])
        top.append(e)
        if len(top) >= args.top_k:
            break

    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)
    ohlcv = loader.load(OHLCVRequest(
        symbol=args.symbol,
        timeframe=args.timeframe,
        start=datetime.fromisoformat(args.start),
        end=datetime.fromisoformat(args.end),
    ))
    splits = expanding_walk_forward(ohlcv.index, train_ratio=args.train_ratio, n_folds=args.n_folds)
    engine = BacktestEngine(fees_bp=args.fees_bp)

    print(f"\n{'=' * 110}")
    print(f"PER-TRADE ANALYSIS — top-{len(top)} genomes × {len(splits)} folds")
    print(f"{'=' * 110}")

    for ev in top:
        strat, err = _try_parse(ev["raw_text"] or "")
        if strat is None:
            print(f"\n>>> {ev['genome_id'][:16]} — parse error: {err}")
            continue

        print(f"\n>>> {ev['genome_id']} (fit_IS={ev['fitness']:.4f}, sharpe_IS={ev['sharpe']:.3f})")
        print(f"{'fold':<5} {'period':<26} {'trades':>7} {'wins':>5} {'losses':>7} {'win%':>6} {'avg_w':>9} {'avg_l':>9} {'ret':>7} {'maxDD':>7} {'sharpe':>7}")
        for s in splits:
            test_df = ohlcv.loc[s.test_idx]
            try:
                signal_fn = compile_strategy(strat)
                signals = signal_fn(test_df)
                bt = engine.run(test_df, signals)
            except Exception as e:
                print(f"  fold {s.fold}: error {e}")
                continue
            trades = bt.trades
            n_trades = 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]
            n_wins = len(wins)
            n_losses = len(losses)
            win_rate = (n_wins / n_trades * 100) if n_trades else 0.0
            avg_w = (sum(wins) / n_wins) if n_wins else 0.0
            avg_l = (sum(losses) / n_losses) if n_losses else 0.0
            # Normalize equity for DD/return
            if n_trades > 0:
                notional = float(test_df["close"].iloc[0])
                equity_pos = (bt.equity_curve / notional) + 1.0
                ret_pct = total_return(equity_pos)
                dd = max_drawdown(equity_pos)
                sr = sharpe_ratio(bt.returns, periods_per_year=8760)
            else:
                ret_pct = dd = sr = 0.0
            period = f"{str(s.test_idx[0])[:10]}..{str(s.test_idx[-1])[:10]}"
            print(f"{s.fold:<5} {period:<26} {n_trades:>7} {n_wins:>5} {n_losses:>7} {win_rate:>5.1f}% {avg_w:>9.1f} {avg_l:>9.1f} {ret_pct:>6.2%} {dd:>6.2%} {sr:>7.3f}")


if __name__ == "__main__":
    main()