PythagorasGoal/Old/scripts/games/agent_brief.py

"""
agent_brief — genera il "digest" ANONIMO che ogni agente cieco riceve.

L'agente non sa che sono BTC/ETH ne' che e' crypto: vede solo due serie X e Y
(rinominate dal motore A/B), una finestra normalizzata (base 100) e statistiche
aggregate. Da queste deve proporre una regola che "anticipi" i movimenti.

Genera anche il MENU dei blocchi (famiglie + range parametri) che l'agente puo'
comporre, in modo che l'output sia una spec backtestabile.
"""
from __future__ import annotations

import json

import numpy as np

from scripts.games.engine import load_anon


def _stats(close, high, low):
    r = np.diff(np.log(close))
    r = r[np.isfinite(r)]
    out = {
        "n_bars": int(len(close)),
        "ret_vol_pct": round(float(np.std(r) * 100), 4),
        "ret_autocorr_lag1": round(float(np.corrcoef(r[:-1], r[1:])[0, 1]), 4),
        "ret_autocorr_lag5": round(float(np.corrcoef(r[:-5], r[5:])[0, 1]), 4),
        "pct_up_bars": round(float(np.mean(r > 0) * 100), 2),
        "skew": round(float(((r - r.mean()) ** 3).mean() / (r.std() ** 3 + 1e-12)), 3),
        "kurtosis": round(float(((r - r.mean()) ** 4).mean() / (r.std() ** 4 + 1e-12)), 2),
    }
    # tendenza a rientrare dopo grandi mosse (|z|>2): segno del rendimento successivo
    z = (r - r.mean()) / (r.std() + 1e-12)
    big = np.where(np.abs(z[:-1]) > 2)[0]
    if len(big) > 20:
        nxt = r[big + 1]
        same = np.sign(r[big]) == np.sign(nxt)
        out["after_big_move_continues_pct"] = round(float(np.mean(same) * 100), 1)
    return out


def make_digest(tf: str, window: int = 60, seed: int = 0):
    data = load_anon(tf)
    n = data["n"]
    # finestra recente normalizzata (base 100) per "vedere" la forma
    s = max(0, n - window)
    dig = {"timeframe_id": {"5m": "T1", "15m": "T2", "30m": "T3", "1h": "T4",
                            "2h": "T5", "4h": "T6", "1d": "T7"}.get(tf, "T?"),
           "n_bars_total": n, "series": {}}
    for name in ("A", "B"):
        o = data[name]
        c = o["close"]
        norm = (c[s:] / c[s] * 100.0)
        dig["series"][{"A": "X", "B": "Y"}[name]] = {
            "stats": _stats(c, o["high"], o["low"]),
            "recent_window_norm": [round(float(v), 2) for v in norm],
        }
    # relazione fra le due serie
    ra = np.diff(np.log(data["A"]["close"]))
    rb = np.diff(np.log(data["B"]["close"]))
    m = min(len(ra), len(rb))
    dig["XY_return_correlation"] = round(float(np.corrcoef(ra[:m], rb[:m])[0, 1]), 4)
    lr = np.log(data["A"]["close"][:m + 1] / data["B"]["close"][:m + 1])
    dig["XY_logratio_ret_autocorr"] = round(
        float(np.corrcoef(np.diff(lr)[:-1], np.diff(lr)[1:])[0, 1]), 4)
    return dig


MENU = {
    "obiettivo": ("Proponi UNA regola che anticipi i movimenti futuri per un PnL "
                  "netto positivo dopo costi (0.10% andata+ritorno per trade). "
                  "Servono >=10 operazioni al mese. Non sai cosa siano X e Y."),
    "famiglie": {
        "zscore": "fade/segui lo z-score del prezzo su 'lookback' barre (entry_thr in sigma)",
        "breakout": "rottura del canale max/min su 'lookback' barre (reversion=fade la rottura)",
        "ma_cross": "incrocio EMA veloce(lookback)/lenta(lookback*slow_mult)",
        "rsi": "RSI(lookback); entry_thr scala le bande attorno a 50",
        "momentum": "rendimento su 'lookback' barre vs soglia entry_thr (%)",
        "pairs": "market-neutral sullo z del log-rapporto X/Y (long una/short l'altra)",
    },
    "direzione": ["reversion (vai contro la mossa)", "trend (segui la mossa)"],
    "serie": ["X", "Y (solo per single-family)", "pairs usa entrambe"],
    "exit": "tp_atr / sl_atr (in unita' ATR), max_bars (durata massima)",
    "range": {
        "lookback": "5-120", "entry_thr": "1.0-3.5", "tp_atr": "0.5-4.0",
        "sl_atr": "1.0-5.0", "max_bars": "6-120", "slow_mult": "2-6",
        "exit_thr (pairs)": "0.2-1.0",
    },
    "output_schema": {
        "family": "una di [zscore,breakout,ma_cross,rsi,momentum,pairs]",
        "series": "X|Y|AB(pairs)", "direction": "reversion|trend",
        "params": "dict coi parametri scelti", "hypothesis": "1-2 frasi: cosa hai notato",
    },
}


if __name__ == "__main__":
    import sys
    tf = sys.argv[1] if len(sys.argv) > 1 else "1h"
    print(json.dumps(make_digest(tf), indent=2)[:2000])