PythagorasGoal/scripts/analysis/eval_signal.py

"""EVALUATOR STANDARD per i segnali della ricerca multi-agente (Fase frattale, v2.0.0).

Ogni agente scrive SOLO una funzione `signal(df, asset, tf) -> np.ndarray` (posizione per barra
in [-1,1], decisa entro close[i]) in un file. Questo evaluator la valuta in modo UNIFORME e ONESTO
sull'harness research_lab, e — cruciale — esegue un GUARD ANTI-LOOK-AHEAD automatico: ricalcola il
segnale su prefissi del df e verifica che pos[i] non dipenda da barre future (leak>0 = sospetto).

    uv run python scripts/analysis/eval_signal.py <signal_file.py> <BTC|ETH> <5m|15m|1h> [--holdout]

Stampa una riga "RESULT_JSON:{...}" con tutte le metriche (gli agenti riportano quei campi esatti).
"""
from __future__ import annotations
import sys
import json
import importlib.util
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))
import numpy as np
import pandas as pd
from scripts.analysis.research_lab import backtest, buy_hold, mc_pvalue, load_tf, ts, _net_series, VAL_START, HOLDOUT_START


def load_signal(path):
    spec = importlib.util.spec_from_file_location("usig", path)
    m = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(m)
    if not hasattr(m, "signal"):
        raise AttributeError("il file non definisce signal(df, asset, tf)")
    return m.signal


def causality_guard(signal, df, asset, tf, k=12):
    """Ricalcola il segnale su prefissi df[:i+1] e confronta pos[i] col run completo.
    Se differiscono -> il segnale usa dati FUTURI (look-ahead). Ritorna #violazioni (0 = pulito)."""
    full = np.asarray(signal(df, asset, tf), float)
    n = len(df)
    if len(full) != n:
        return -1
    rng = np.random.default_rng(0)
    idx = rng.integers(int(n * 0.6), n - 1, size=k)
    bad = 0
    for i in idx:
        try:
            p = np.asarray(signal(df.iloc[:i + 1].copy(), asset, tf), float)
        except Exception:
            bad += 1; continue
        if len(p) != i + 1 or not np.isclose(np.nan_to_num(p[i]), np.nan_to_num(full[i]), atol=1e-6):
            bad += 1
    return bad


def main():
    args = sys.argv[1:]
    holdout = "--holdout" in args
    args = [a for a in args if a != "--holdout"]
    sigfile, asset, tf = args[0], args[1].upper(), args[2]
    res = {"asset": asset, "tf": tf, "sigfile": sigfile}
    try:
        signal = load_signal(sigfile)
        df = load_tf(asset, tf)
        pos = np.asarray(signal(df, asset, tf), float)
        res["n"] = int(len(df))
        res["len_ok"] = bool(len(pos) == len(df))
        if not res["len_ok"]:
            res["error"] = f"len(pos)={len(pos)} != len(df)={len(df)}"
            print("RESULT_JSON:" + json.dumps(res)); return
        res["finite"] = bool(np.isfinite(np.nan_to_num(pos, nan=0.0)).all())
        res["leak"] = int(causality_guard(signal, df, asset, tf))
        full = backtest(df, pos, tf)
        oos = backtest(df, pos, tf, lo=VAL_START, hi=HOLDOUT_START)
        bh = buy_hold(df, tf)
        _, p, _, _ = mc_pvalue(df, pos, tf, n=250)
        res.update(
            implemented=True,
            full_sharpe=round(full.sharpe, 3), full_ret=round(full.ret, 3), full_dd=round(full.maxdd, 3),
            oos_sharpe=round(oos.sharpe, 3), bh_sharpe=round(bh.sharpe, 3),
            gross_sharpe=round(backtest(df, pos, tf, fee_rt=0.0).sharpe, 3),
            fee02_sharpe=round(backtest(df, pos, tf, fee_rt=0.002).sharpe, 3),
            turnover=round(full.ntrades, 1), exposure=round(full.exposure, 3),
            null_p=round(p, 4),
            beats_bh=bool(full.sharpe > bh.sharpe and oos.sharpe > 0),
        )
        # breadth per-anno (pre-hold-out): % anni positivi, anni rossi consecutivi
        net, _, _, _ = _net_series(df, pos)
        s = pd.Series(net, index=ts(df))
        s = s[s.index < pd.Timestamp(HOLDOUT_START, tz="UTC")]
        yr = {int(y): float((1 + g).prod() - 1) for y, g in s.groupby(s.index.year)}
        vals = list(yr.values())
        max_consec_red = 0; cur = 0
        for v in vals:
            cur = cur + 1 if v < 0 else 0
            max_consec_red = max(max_consec_red, cur)
        res["per_year_preho"] = {y: round(v, 3) for y, v in yr.items()}
        res["pct_years_pos"] = round(sum(v > 0 for v in vals) / len(vals), 2) if vals else 0.0
        res["max_consec_red_years"] = int(max_consec_red)
        if holdout:
            ho = backtest(df, pos, tf, lo=HOLDOUT_START)
            res["holdout_sharpe"] = round(ho.sharpe, 3)
            res["holdout_ret"] = round(ho.ret, 3)
            res["holdout_dd"] = round(ho.maxdd, 3)
    except Exception as e:
        res["implemented"] = False
        res["error"] = f"{type(e).__name__}: {str(e)[:200]}"
    print("RESULT_JSON:" + json.dumps(res))


if __name__ == "__main__":
    main()