From 68e0b009e9350cf099e7850f1fb4f66dcb80a03e Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Tue, 19 May 2026 14:16:07 +0000 Subject: [PATCH] feat(strategy_pythagoras): GA smoke-test runner (BTC+ETH dual run + invariance bonus) --- scripts/run_pythagoras_smoke.py | 477 ++++++++++++++++++++++++++++++++ 1 file changed, 477 insertions(+) create mode 100644 scripts/run_pythagoras_smoke.py diff --git a/scripts/run_pythagoras_smoke.py b/scripts/run_pythagoras_smoke.py new file mode 100644 index 0000000..0fe1990 --- /dev/null +++ b/scripts/run_pythagoras_smoke.py @@ -0,0 +1,477 @@ +"""Smoke test del GA per strategy_pythagoras. + +Esegue 2 run di Phase 1 (BTC 5m + ETH 5m), poi cross-rank i genomi +comuni applicando il bonus di asset-invariance (corr_signal sui pattern +di entry entro +/-36 barre = +/-3h su 5m TF, vedi paper Pythagoras p.43). + +Configurazione (per spec §4): +- Population 20, 5 generations +- Asset: BTC-PERPETUAL 5m + ETH-PERPETUAL 5m (Cerbero deribit) +- Train window: 2024-07-01 -> 2024-12-31 +- Test window: 2025-01-01 -> 2025-01-31 (caricato come coda dello stesso + range; non usato dal GA ma necessario per dataset continuo se in futuro + si attiva WFA) +- Stili cognitivi: 7 da strategy_pythagoras/prompts.json +- Indicatori Pythagoras: candle_pattern, pythagorean_ratio, fractal_mirror + (registrati nel compiler tramite import side-effect di strategy_pythagoras.indicators) +- Fitness post-processing cross-asset: apply_invariance_bonus +- Output: top 50 winners persisted in state/strategy_pythagoras.db + (tabella pythagoras_winners) + +Adattamento all'API reale di run_phase1 (Task 4.1 findings + verifica diretta): + +- ``run_phase1(cfg: RunConfig, ohlcv: pd.DataFrame, llm: LLMClient) -> str`` + ritorna un ``run_id``. Non c'e' un fitness hook esterno: il GA loop + invoca ``compute_fitness`` inline e persiste via + ``repo.save_evaluation``. Per il bonus invariance dobbiamo: + 1. lanciare due ``run_phase1`` indipendenti, uno per asset; + 2. caricare le evaluations via ``repo.list_evaluations(run_id)``; + 3. ricompilare la strategia (``_try_parse`` + ``compile_strategy``) sui + segnali di ciascun OHLCV per estrarre gli entry index; + 4. calcolare ``corr_signal`` sugli entry binari (Series int-indexed) + e applicare ``apply_invariance_bonus``. + +- Le serie OHLCV NON sono shippate in repo come ``src/strategy_crypto/series/``: + il default loader Cerbero le cachea in ``./series/{cache_key}.parquet`` + (cache key = sha1 di ``exchange|symbol|timeframe|start|end``). Riusiamo + quel meccanismo: caricamento via ``CerberoOHLCVLoader``, identico a + ``scripts/run_phase1.py``. + +Shape effettivo del dict ritornato da ``repo.list_evaluations(run_id)`` +(vedi ``persistence/repository.py:213`` e schema in ``schema.py``): + + { + 'run_id', 'genome_id', 'fitness', 'dsr', 'dsr_pvalue', 'sharpe', + 'max_dd', 'total_return', 'n_trades', 'parse_error', 'raw_text', + 'eval_ts', 'fitness_oos', 'sharpe_oos', 'return_oos', + 'max_dd_oos', 'n_trades_oos' + } + +Nota: ``cognitive_style`` e ``generation`` NON sono nelle evaluations; +vanno presi via ``repo.list_genomes(run_id)`` (payload_json del genoma). +``raw_text`` contiene il completion grezzo del LLM, da cui si estrae +nuovamente lo ``Strategy`` AST via ``_try_parse``. +""" +from __future__ import annotations + +import json +import logging +import os +import sqlite3 +from datetime import datetime +from pathlib import Path +from typing import Any + +import pandas as pd # type: ignore[import-untyped] + +# Side-effect import: registra candle_pattern, pythagorean_ratio, fractal_mirror +# in compiler.INDICATOR_FNS prima che il GA inizi a compilare strategie. +# (Il compiler in protocol/compiler.py importa gia' i 3 simboli dal package +# strategy_pythagoras.indicators, ma facciamo l'import esplicito qui per +# rendere la dipendenza chiara e indipendente dall'ordine di import.) +import strategy_pythagoras.indicators # noqa: F401 +from multi_swarm_core.agents.hypothesis import _try_parse +from multi_swarm_core.backtest.orders import Side +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.genome.hypothesis import ModelTier +from multi_swarm_core.genome.prompt_library import PromptLibrary +from multi_swarm_core.llm.client import LLMClient +from multi_swarm_core.orchestrator.run import RunConfig, run_phase1 +from multi_swarm_core.persistence.repository import Repository +from multi_swarm_core.protocol.compiler import compile_strategy +from strategy_pythagoras.fitness_invariance import ( + apply_invariance_bonus, + corr_signal, +) + +ROOT = Path(__file__).resolve().parents[1] +DB_PATH = Path( + os.getenv("STRATEGY_PYTHAGORAS_DB_PATH", str(ROOT / "state" / "strategy_pythagoras.db")) +) +PROMPTS_PATH = ROOT / "src" / "strategy_pythagoras" / "strategy_pythagoras" / "prompts.json" +RUN_NAME = "pythagoras-smoke-001" + +# GA configuration (smoke per spec §4) +POPULATION = 20 +GENERATIONS = 5 + +# Data window +TRAIN_START = datetime.fromisoformat("2024-07-01T00:00:00+00:00") +TRAIN_END = datetime.fromisoformat("2024-12-31T23:55:00+00:00") +# Carichiamo anche gennaio 2025 come coda (per usi futuri: WFA OOS). +# Il GA loop in questa fase usa l'intero range; e' compito di un eventuale +# wfa_train_split (non attivato qui per coerenza con spec §4 smoke). +TEST_END = datetime.fromisoformat("2025-01-31T23:55:00+00:00") + +ASSETS: list[tuple[str, str]] = [ + ("BTC-PERPETUAL", "btc"), + ("ETH-PERPETUAL", "eth"), +] +TIMEFRAME = "5m" +EXCHANGE = "deribit" +TOP_K_PERSIST = 50 + +logger = logging.getLogger(RUN_NAME) + + +def init_winners_table(con: sqlite3.Connection) -> None: + """Crea ``pythagoras_winners`` se non esiste (idempotente).""" + con.execute( + """ + CREATE TABLE IF NOT EXISTS pythagoras_winners ( + genome_id TEXT PRIMARY KEY, + cognitive_style TEXT, + fitness REAL, + sharpe_btc REAL, + sharpe_eth REAL, + invariance_score REAL, + rules_json TEXT, + generation INTEGER, + run_name TEXT + ) + """ + ) + con.commit() + + +def _load_ohlcv(loader: CerberoOHLCVLoader, symbol: str) -> pd.DataFrame: + """Carica la finestra ``TRAIN_START -> TEST_END`` per ``symbol`` su 5m.""" + req = OHLCVRequest( + symbol=symbol, + timeframe=TIMEFRAME, + start=TRAIN_START, + end=TEST_END, + exchange=EXCHANGE, + ) + ohlcv = loader.load(req) + logger.info( + "OHLCV loaded for %s: %d bars (%s -> %s)", + symbol, len(ohlcv), + ohlcv.index[0] if len(ohlcv) else "n/a", + ohlcv.index[-1] if len(ohlcv) else "n/a", + ) + return ohlcv + + +def _build_run_config( + run_name: str, symbol: str, prompt_library: PromptLibrary, db_path: Path, +) -> RunConfig: + """Costruisce il ``RunConfig`` per un singolo asset. + + Usa lo stesso GA-core DB del progetto (``settings.ga_db_path`` se override + non passato): vi vengono scritte ``runs``, ``generations``, ``genomes``, + ``evaluations`` per la run. + """ + return RunConfig( + run_name=run_name, + population_size=POPULATION, + n_generations=GENERATIONS, + elite_k=2, + tournament_k=3, + p_crossover=0.5, + seed=42, + model_tier=ModelTier.C, + symbol=symbol, + timeframe=TIMEFRAME, + fees_bp=5.0, + n_trials_dsr=50, + db_path=db_path, + prompt_library=prompt_library, + # Smoke: niente WFA, niente eval OOS in loop, niente prompt mutation LLM. + # I parametri restano sui default sicuri di RunConfig. + ) + + +def run_ga_for_asset( + asset_label: str, + symbol: str, + ohlcv: pd.DataFrame, + prompt_library: PromptLibrary, + llm: LLMClient, + ga_db_path: Path, +) -> tuple[str, Repository]: + """Lancia ``run_phase1`` per un asset. + + Ritorna ``(run_id, repo)`` per il caller, che usera' ``repo`` per + estrarre evaluations + genomes a fine del run. + """ + run_name = f"{RUN_NAME}-{asset_label}" + cfg = _build_run_config(run_name, symbol, prompt_library, ga_db_path) + logger.info("Starting GA run '%s' on %s (%d bars)", run_name, symbol, len(ohlcv)) + run_id = run_phase1(cfg, ohlcv=ohlcv, llm=llm) + logger.info("Run '%s' completed: run_id=%s", run_name, run_id) + repo = Repository(ga_db_path) + return run_id, repo + + +def _entries_series_from_eval( + eval_row: dict[str, Any], ohlcv: pd.DataFrame, +) -> pd.Series | None: + """Ricostruisce gli entries binari (Side.LONG/SHORT -> 1, altrimenti 0) + a partire dal ``raw_text`` salvato nell'eval row. + + Ritorna ``None`` se il raw_text non e' parsabile (caso parse_error). + L'index della Series ritornata e' INTERO posizionale (0..N-1) come + richiesto da ``corr_signal`` (vedi tests in + ``strategy_pythagoras/tests/test_fitness_invariance.py``). + """ + raw = eval_row.get("raw_text") + if not raw: + return None + strategy, parse_err = _try_parse(raw) + if strategy is None: + logger.debug( + "skip genome %s: parse error '%s'", + eval_row.get("genome_id"), parse_err, + ) + return None + try: + signal_fn = compile_strategy(strategy) + signals = signal_fn(ohlcv) + except Exception as exc: + logger.debug( + "skip genome %s: compile/exec error: %s", + eval_row.get("genome_id"), exc, + ) + return None + # 1 dove il signal e' LONG o SHORT (entry attiva), 0 altrove. + is_entry = signals.isin([Side.LONG, Side.SHORT]).fillna(False).astype(int) + # Riassegna integer index per il match in corr_signal (che somma delta + # interi all'index e fa il test ``ti + delta in b_set``). + return pd.Series(is_entry.values, index=range(len(is_entry)), dtype="int64") + + +def _collect_evaluations( + repo: Repository, run_id: str, ohlcv: pd.DataFrame, +) -> dict[str, dict[str, Any]]: + """Carica evaluations + genomes per ``run_id`` e li unisce per genome_id. + + Returns: dict ``{genome_id: row}`` dove ``row`` contiene i campi + dell'eval + ``cognitive_style``, ``generation``, ``strategy_json`` + (dict del genoma serializzato) e ``entries`` (pd.Series int-indexed). + """ + evals = repo.list_evaluations(run_id) + genomes = repo.list_genomes(run_id) + genome_by_id: dict[str, dict[str, Any]] = {} + for grow in genomes: + try: + payload = json.loads(grow["payload_json"]) + except (json.JSONDecodeError, TypeError): + payload = {} + genome_by_id[grow["id"]] = payload + + out: dict[str, dict[str, Any]] = {} + for ev in evals: + gid = ev["genome_id"] + payload = genome_by_id.get(gid, {}) + row = dict(ev) + row["cognitive_style"] = payload.get("cognitive_style", "") + row["generation"] = int(payload.get("generation", 0)) + # ``raw_text`` e' il completion grezzo; lo ri-parsiamo in + # _entries_series_from_eval. Salviamo la rappresentazione canonica + # ``strategy_json`` per persistenza (best-effort: se il parse fallisce + # salviamo il raw_text come fallback). + strategy, _err = _try_parse(row.get("raw_text") or "") + if strategy is not None: + # Strategy non e' direttamente JSON-serializable: serializziamo + # la struttura nominale tramite dataclasses.asdict-like fallback. + try: + row["strategy_json"] = _strategy_to_jsonable(strategy) + except Exception: + row["strategy_json"] = {"raw_text": row.get("raw_text", "")} + else: + row["strategy_json"] = {"raw_text": row.get("raw_text", "")} + row["entries"] = _entries_series_from_eval(ev, ohlcv) + out[gid] = row + return out + + +def _strategy_to_jsonable(strategy: Any) -> dict[str, Any]: + """Serializza un ``Strategy`` AST in dict JSON-friendly. + + Strategy/Rule/Node sono dataclass: usiamo ``dataclasses.asdict`` quando + possibile, con fallback a ``str(strategy)`` se la struttura contiene + membri non-serializzabili (es. enum non-Str). + """ + import dataclasses + if dataclasses.is_dataclass(strategy): + try: + return dataclasses.asdict(strategy) + except TypeError: + pass + return {"repr": repr(strategy)} + + +def compute_invariance_for_pair( + btc_evals: dict[str, dict[str, Any]], + eth_evals: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: + """Per ogni genome_id presente in entrambi i run, calcola invariance + bonus. + + Lo stesso ``genome_id`` puo' apparire in entrambi i run perche' l'id e' + deterministico (sha1 di system_prompt+feature_access+temperature+...) e + il seed del GA e' fisso: il founder set e i mutanti hanno alta probabilita' + di collisione cross-asset. Quando il genoma compare in entrambi, le + metriche ``sharpe`` IS sono comparabili e ha senso valutare l'invarianza. + """ + out: list[dict[str, Any]] = [] + common_ids = set(btc_evals) & set(eth_evals) + logger.info( + "Common genomes BTC ∩ ETH: %d (BTC: %d, ETH: %d)", + len(common_ids), len(btc_evals), len(eth_evals), + ) + for gid in common_ids: + b = btc_evals[gid] + e = eth_evals[gid] + entries_btc: pd.Series | None = b.get("entries") + entries_eth: pd.Series | None = e.get("entries") + if entries_btc is None or entries_eth is None: + inv = 0.0 + elif len(entries_btc) == 0 or len(entries_eth) == 0: + inv = 0.0 + else: + try: + # Allineiamo a lunghezza minima: i due asset possono avere + # un numero di bars leggermente diverso (gap nel feed Cerbero). + # corr_signal lavora solo sugli index 1 -> il troncamento non + # introduce bias asimmetrici. + min_len = min(len(entries_btc), len(entries_eth)) + inv = corr_signal( + entries_btc.iloc[:min_len].reset_index(drop=True), + entries_eth.iloc[:min_len].reset_index(drop=True), + ) + except Exception as exc: + logger.warning("corr_signal failed for %s: %s", gid, exc) + inv = 0.0 + sharpe_btc = float(b.get("sharpe") or 0.0) + sharpe_eth = float(e.get("sharpe") or 0.0) + mean_sharpe = 0.5 * (sharpe_btc + sharpe_eth) + boosted = apply_invariance_bonus(mean_sharpe, inv) + out.append({ + "genome_id": gid, + "cognitive_style": b.get("cognitive_style") or e.get("cognitive_style", ""), + "fitness": float(boosted), + "sharpe_btc": sharpe_btc, + "sharpe_eth": sharpe_eth, + "invariance_score": float(inv), + "rules_json": json.dumps(b.get("strategy_json") or {}, default=str), + "generation": int(b.get("generation", 0)), + "run_name": RUN_NAME, + }) + return sorted(out, key=lambda r: r["fitness"], reverse=True) + + +def persist_winners(con: sqlite3.Connection, winners: list[dict[str, Any]]) -> None: + if not winners: + logger.warning("No winners to persist") + return + con.executemany( + """ + INSERT OR REPLACE INTO pythagoras_winners + (genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, + invariance_score, rules_json, generation, run_name) + VALUES (:genome_id, :cognitive_style, :fitness, :sharpe_btc, :sharpe_eth, + :invariance_score, :rules_json, :generation, :run_name) + """, + winners, + ) + con.commit() + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", + ) + settings = load_settings() + + # Prompt library Pythagoras (NON quello di strategy_crypto). + if not PROMPTS_PATH.exists(): + raise FileNotFoundError(f"Prompts file not found: {PROMPTS_PATH}") + prompt_library = PromptLibrary.from_json(PROMPTS_PATH) + logger.info( + "PromptLibrary loaded from %s: %d styles (%s)", + PROMPTS_PATH, len(prompt_library.styles), + ", ".join(prompt_library.cognitive_styles), + ) + + # Cerbero client + OHLCV loader (riusa la cache parquet in ./series). + 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) + + # LLM client (qwen-2.5-72b tier C come da spec progetto - vedi MEMORY: + # cambiare modello senza ricalibrare = regressione dimostrata). + llm = LLMClient( + openrouter_api_key=settings.openrouter_api_key.get_secret_value(), + model_tier_s=settings.llm_model_tier_s, + model_tier_a=settings.llm_model_tier_a, + model_tier_b=settings.llm_model_tier_b, + model_tier_c=settings.llm_model_tier_c, + model_tier_d=settings.llm_model_tier_d, + openrouter_base_url=settings.openrouter_base_url, + ) + + # Setup DB winners (separato dal GA core DB). + DB_PATH.parent.mkdir(parents=True, exist_ok=True) + con = sqlite3.connect(DB_PATH) + try: + init_winners_table(con) + logger.info("Winners DB initialized at %s", DB_PATH) + + # Carica OHLCV per entrambi gli asset PRIMA dei run GA, cosi' se la + # rete o Cerbero sono giu' falliamo subito senza sprecare chiamate LLM. + ohlcv_by_asset: dict[str, pd.DataFrame] = {} + for symbol, label in ASSETS: + ohlcv_by_asset[label] = _load_ohlcv(loader, symbol) + + # Run GA per asset. Usa il GA-core DB definito in settings; ogni run + # crea un proprio run_id e set di evaluations isolato. + evals_by_asset: dict[str, dict[str, dict[str, Any]]] = {} + for symbol, label in ASSETS: + run_id, repo = run_ga_for_asset( + asset_label=label, + symbol=symbol, + ohlcv=ohlcv_by_asset[label], + prompt_library=prompt_library, + llm=llm, + ga_db_path=settings.ga_db_path, + ) + evals_by_asset[label] = _collect_evaluations( + repo, run_id, ohlcv_by_asset[label] + ) + logger.info( + "%s: %d evaluations collected", label.upper(), + len(evals_by_asset[label]), + ) + + # Cross-rank con invariance bonus. + winners = compute_invariance_for_pair( + evals_by_asset["btc"], evals_by_asset["eth"], + ) + logger.info( + "Computed invariance bonus for %d common genomes", len(winners), + ) + + top = winners[:TOP_K_PERSIST] + persist_winners(con, top) + logger.info( + "Persisted top %d winners to %s (table: pythagoras_winners)", + len(top), DB_PATH, + ) + finally: + con.close() + + +if __name__ == "__main__": + main()