Adriano Dal Pastro
478 lines
18 KiB
Python
478 lines
18 KiB
Python
"""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 = os.getenv("PYTHAGORAS_SMOKE_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()
|