diff --git a/README.md b/README.md index 21bbbd6..e39b8d7 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ Stack: Python 3.13, uv workspace, hatchling, pytest+pytest-mock+responses, opena ```bash uv sync # installa entrambi i workspace member come editable cp .env.example .env # compila CERBERO_*_TOKEN e OPENROUTER_API_KEY -uv run pytest # 238 test attesi (234 core + 4 smoke strategy_crypto) +uv run pytest # 250 test attesi (246 core + 4 smoke strategy_crypto) ``` ### Variabili .env richieste @@ -150,16 +150,27 @@ uv run mypy src/ scripts/ # Smoke run (MockLLM + OHLCV sintetico, no API calls) uv run python scripts/smoke_run.py -# Run reale Phase 1/2 (Cerbero + OpenRouter, ~$0.07 per run K=20 10gen) +# Run reale Phase 1/2 (Cerbero + OpenRouter, ~$0.15-0.25 per run K=20 10gen, +# ~$0.40-0.55 per run esteso K=40 20gen con WFA OOS) uv run python scripts/run_phase1.py \ --name run-XXX \ --exchange deribit --symbol BTC-PERPETUAL --timeframe 1h \ --start 2024-01-01T00:00:00+00:00 \ --end 2026-01-01T00:00:00+00:00 \ --population-size 20 --n-generations 10 \ - --prompt-mutation-weight 0.30 --fitness-v2 + --prompt-mutation-weight 0.30 --fitness-v2 \ + --llm-concurrency 8 # 5-8x speedup wall time (default 1) # Default --prompt-library: importlib.resources del package strategy_crypto/prompts.json +# Multi-fold validation di un run esistente (anti-overfit, 7y expanding-window) +uv run python scripts/validate_run.py \ + --run-id \ + --top-k 10 --n-folds 4 --train-ratio 0.5 \ + --start 2018-09-01T00:00:00+00:00 --end 2026-01-01T00:00:00+00:00 \ + --fitness-v2 \ + --output-json state/validation-XXX.json +# Ranking per "robust_score" = min(fitness_oos) su tutti i fold. + # Backtest standalone di una strategia JSON uv run python scripts/backtest_strategy.py \ --strategy src/strategy_crypto/strategy_crypto/strategies/btc_fb63e851.json \ @@ -175,6 +186,21 @@ uv run python -m multi_swarm_core.dashboard.nicegui_app # GA core (/, /converg uv run python -m strategy_crypto.frontend.nicegui_app # Strategy crypto (/ paper) ``` +## Performance & Validation + +**Backtest engine vettorializzato** (`backtest/engine.py`): rimosso il loop `pd.iterrows()` a favore di state machine numpy. Speedup misurati: + +| Dataset | Before (iterrows) | After (vectorized) | Speedup | +|---------|-------------------|--------------------|---------| +| 2 anni (17545 bar) | 470 ms | **28 ms** | **16.8×** | +| 7 anni (64297 bar) | 1744 ms | **154 ms** | **11.3×** | + +Equivalenza numerica garantita: 5 parity test parametrici vs. reference implementation legacy (`test_backtest_engine_vectorized.py`). + +**Parallel propose LLM** (`orchestrator/run.py`): `--llm-concurrency N` lancia N chiamate `hypothesis_agent.propose()` concorrenti per generazione tramite `ThreadPoolExecutor`. OpenRouter qwen-2.5 regge 6-10 concorrenti senza rate-limit. Default 1 = backward-compat. + +**Multi-fold validation tool** (`scripts/validate_run.py`): qualunque run completato puo' essere rivalutato post-hoc su N fold expanding-window di un dataset esteso (tipicamente 7 anni). Vital per evitare il single-hold-out overfit: il GA puo' selezionare un genome con `fitness_is` alta che collassa OOS (osservato su `phase1-extended-001`: elite IS Sharpe 1.93, OOS Sharpe -1.00). Ranking finale per `robust_score = min(fitness_oos)`. Output JSON con per-fold breakdown + aggregati mean/min/std. + ## Dashboard (split core + strategy) Due NiceGUI dashboard distinte (dark palette, palette neon): diff --git a/scripts/run_phase1.py b/scripts/run_phase1.py index 0b73166..0f1d299 100644 --- a/scripts/run_phase1.py +++ b/scripts/run_phase1.py @@ -114,6 +114,16 @@ def parse_args() -> argparse.Namespace: "Schema: {styles: {: {directive: }}}" ), ) + p.add_argument( + "--llm-concurrency", + type=int, + default=1, + help=( + "Numero di propose() LLM concorrenti per generazione (default 1 = " + "serial). 6-10 tipicamente accettati da OpenRouter qwen-2.5 senza " + "rate-limit; riduce wall time GA loop di 5-8x." + ), + ) return p.parse_args() @@ -187,6 +197,7 @@ def main() -> None: eval_oos_during_loop=args.eval_oos_during_loop, fitness_combined_alpha=args.fitness_combined_alpha, prompt_library=prompt_library, + llm_concurrency=args.llm_concurrency, ) run_id = run_phase1(cfg, ohlcv=ohlcv, llm=llm) diff --git a/scripts/validate_run.py b/scripts/validate_run.py new file mode 100644 index 0000000..4bebb01 --- /dev/null +++ b/scripts/validate_run.py @@ -0,0 +1,271 @@ +"""Multi-fold validation di un run esistente. + +Prende un ``run_id`` da ``state/runs.db``, seleziona i top-K genomi per fitness IS, +e li rivaluta su N fold expanding-window di un dataset OHLCV (tipicamente piu' +lungo del train del GA). Output: per-fold + aggregati (mean / min / std) della +fitness OOS. + +Use case: il GA puo' selezionare un "lucky-shot" overfit a uno specifico regime. +Validare i top-K su finestre temporali diverse rivela quali strategie sono +robuste vs overfitter. + +Esempio:: + + python scripts/validate_run.py \\ + --run-id e263651598894da688d95fda90a34a96 \\ + --top-k 10 --n-folds 4 \\ + --symbol BTC-PERPETUAL --timeframe 1h \\ + --start 2018-09-01 --end 2026-01-01 +""" + +from __future__ import annotations + +import argparse +import json +import statistics +from datetime import datetime +from pathlib import Path + +import pandas as pd # type: ignore[import-untyped] + +from multi_swarm_core.agents.adversarial import AdversarialAgent +from multi_swarm_core.agents.falsification import FalsificationAgent +from multi_swarm_core.agents.hypothesis import _try_parse +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.ga.fitness import compute_fitness +from multi_swarm_core.persistence.repository import Repository + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description="Multi-fold validation di top-K genomi") + p.add_argument("--run-id", required=True, help="run_id da validare") + p.add_argument("--top-k", type=int, default=10, help="quanti genomi top valutare") + p.add_argument("--n-folds", type=int, default=4, help="numero fold expanding-window") + p.add_argument( + "--train-ratio", + type=float, + default=0.5, + help="frazione iniziale per il train iniziale (folds testano la coda)", + ) + p.add_argument("--symbol", default="BTC-PERPETUAL") + p.add_argument("--timeframe", default="1h") + p.add_argument("--exchange", default="deribit", choices=["deribit", "bybit", "hyperliquid"]) + p.add_argument("--start", default="2018-09-01T00:00:00+00:00") + p.add_argument("--end", default="2026-01-01T00:00:00+00:00") + p.add_argument("--fees-bp", type=float, default=5.0) + p.add_argument("--n-trials-dsr", type=int, default=50) + p.add_argument( + "--fees-eat-alpha-threshold", type=float, default=0.5, + ) + p.add_argument( + "--flat-too-long-threshold", type=float, default=0.95, + ) + p.add_argument( + "--undertrading-threshold", type=int, default=10, + ) + p.add_argument( + "--fitness-v2", action="store_true", + help="Coerente con --fitness-v2 del run originale", + ) + p.add_argument( + "--fitness-soft-penalty", type=float, default=0.4, + ) + p.add_argument( + "--output-json", + type=Path, + default=None, + help="Path JSON dove salvare i risultati (default: stdout solo)", + ) + return p.parse_args() + + +def main() -> None: + args = parse_args() + settings = load_settings() + + # Repository: top-K genomi per fitness IS, con raw_text parsable. + repo = Repository(settings.ga_db_path) + repo.init_schema() + run = repo.get_run(args.run_id) + if run is None: + raise SystemExit(f"run_id non trovato: {args.run_id}") + print(f"Validating run: {run['name']} ({args.run_id})") + print(f" status: {run['status']}, cost: ${run['total_cost_usd']:.4f}") + + 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) + + # Dedup by genome_id (gli elite vengono salvati una sola volta ma possono apparire + # in evaluations multiple se rivalutati con eval_oos_during_loop). + seen_ids: set[str] = set() + top_genomes: list[dict] = [] + for e in parseable: + if e["genome_id"] in seen_ids: + continue + seen_ids.add(e["genome_id"]) + top_genomes.append(e) + if len(top_genomes) >= args.top_k: + break + print(f" selected top-{len(top_genomes)} genomes for validation") + + # OHLCV: carica il dataset esteso. + 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) + req = OHLCVRequest( + symbol=args.symbol, + timeframe=args.timeframe, + start=datetime.fromisoformat(args.start), + end=datetime.fromisoformat(args.end), + exchange=args.exchange, + ) + ohlcv = loader.load(req) + print(f" OHLCV: {len(ohlcv)} bars from {ohlcv.index[0]} to {ohlcv.index[-1]}") + + splits = expanding_walk_forward( + ohlcv.index, train_ratio=args.train_ratio, n_folds=args.n_folds, + ) + print(f" generated {len(splits)} folds") + for s in splits: + print(f" fold {s.fold}: test [{s.test_idx[0]} -> {s.test_idx[-1]}] ({len(s.test_idx)} bars)") + + fals_agent = FalsificationAgent(fees_bp=args.fees_bp, n_trials_dsr=args.n_trials_dsr) + adv_agent = AdversarialAgent( + fees_bp=args.fees_bp, + fees_eat_alpha_threshold=args.fees_eat_alpha_threshold, + flat_too_long_threshold=args.flat_too_long_threshold, + undertrading_threshold=args.undertrading_threshold, + ) + hard_kill = ( + ("no_trades", "degenerate", "undertrading") if args.fitness_v2 else None + ) + + # Itera per ogni genome + fold. + results: list[dict] = [] + for gi, ev in enumerate(top_genomes): + strategy, parse_err = _try_parse(ev["raw_text"] or "") + if strategy is None: + print(f" [{gi}] {ev['genome_id'][:12]} skip (parse error: {parse_err})") + continue + + per_fold: list[dict] = [] + for s in splits: + test_df = ohlcv.loc[s.test_idx] + try: + fals = fals_agent.evaluate(strategy, test_df) + adv = adv_agent.review(strategy, test_df) + fit = compute_fitness( + fals, adv, + hard_kill_findings=hard_kill, + adversarial_soft_penalty=args.fitness_soft_penalty, + ) + except Exception as e: + print(f" fold {s.fold} eval failed: {e}") + continue + per_fold.append({ + "fold": s.fold, + "fitness": float(fit), + "sharpe": float(fals.sharpe), + "dsr": float(fals.dsr), + "dsr_pvalue": float(fals.dsr_pvalue), + "return": float(fals.total_return), + "max_dd": float(fals.max_drawdown), + "n_trades": int(fals.n_trades), + "test_start": str(s.test_idx[0]), + "test_end": str(s.test_idx[-1]), + }) + + if not per_fold: + continue + + fits = [pf["fitness"] for pf in per_fold] + sharps = [pf["sharpe"] for pf in per_fold] + results.append({ + "genome_id": ev["genome_id"], + "fitness_is": float(ev["fitness"]), + "sharpe_is": float(ev["sharpe"]), + "folds": per_fold, + "fitness_oos_mean": statistics.mean(fits), + "fitness_oos_min": min(fits), + "fitness_oos_max": max(fits), + "fitness_oos_std": statistics.pstdev(fits) if len(fits) > 1 else 0.0, + "sharpe_oos_mean": statistics.mean(sharps), + "sharpe_oos_min": min(sharps), + "robust_score": min(fits), # min across folds = pessimismo + }) + + # Ranking finale: per robust_score (min fitness) decrescente. + results.sort(key=lambda r: r["robust_score"], reverse=True) + + print() + print(f"{'='*120}") + print(f"VALIDATION RESULTS ({len(results)} genomes, {len(splits)} folds)") + print(f"{'='*120}") + print( + f"{'rank':>4} {'genome':12} {'fit_is':>8} {'sh_is':>7} " + f"{'fit_mean':>9} {'fit_min':>8} {'fit_max':>8} {'fit_std':>8} " + f"{'sh_mean':>8} {'sh_min':>8} {'robust':>7}" + ) + print("-" * 120) + for rank, r in enumerate(results, 1): + print( + f"{rank:>4} {r['genome_id'][:12]:12} " + f"{r['fitness_is']:>8.4f} {r['sharpe_is']:>7.3f} " + f"{r['fitness_oos_mean']:>9.4f} {r['fitness_oos_min']:>8.4f} " + f"{r['fitness_oos_max']:>8.4f} {r['fitness_oos_std']:>8.4f} " + f"{r['sharpe_oos_mean']:>8.3f} {r['sharpe_oos_min']:>8.3f} " + f"{r['robust_score']:>7.4f}" + ) + + if results: + winner = results[0] + print() + print(f"ROBUST WINNER: {winner['genome_id']}") + print(f" fitness_is={winner['fitness_is']:.4f}, " + f"fitness_oos_min={winner['fitness_oos_min']:.4f}, " + f"fitness_oos_mean={winner['fitness_oos_mean']:.4f}") + print(f" sharpe_is={winner['sharpe_is']:.3f}, " + f"sharpe_oos_min={winner['sharpe_oos_min']:.3f}") + print(f" per-fold breakdown:") + for pf in winner["folds"]: + print( + f" fold {pf['fold']} [{pf['test_start'][:10]} .. {pf['test_end'][:10]}]: " + f"fit={pf['fitness']:.4f} sharpe={pf['sharpe']:.3f} " + f"ret={pf['return']:.3f} n_trades={pf['n_trades']}" + ) + + if args.output_json: + payload = { + "run_id": args.run_id, + "run_name": run["name"], + "n_folds": len(splits), + "top_k_requested": args.top_k, + "top_k_evaluated": len(results), + "symbol": args.symbol, + "timeframe": args.timeframe, + "start": args.start, + "end": args.end, + "ohlcv_bars": len(ohlcv), + "results": results, + } + args.output_json.write_text(json.dumps(payload, indent=2, default=str)) + print(f"\nResults saved to: {args.output_json}") + + +if __name__ == "__main__": + main() diff --git a/src/multi_swarm_core/multi_swarm_core/backtest/engine.py b/src/multi_swarm_core/multi_swarm_core/backtest/engine.py index df2e14a..6d37133 100644 --- a/src/multi_swarm_core/multi_swarm_core/backtest/engine.py +++ b/src/multi_swarm_core/multi_swarm_core/backtest/engine.py @@ -2,6 +2,7 @@ from __future__ import annotations from dataclasses import dataclass +import numpy as np import pandas as pd # type: ignore[import-untyped] from .orders import Position, Side, Trade @@ -28,74 +29,110 @@ class BacktestEngine: self.fees_bp = fees_bp def run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult: + n = len(ohlcv) + if n == 0: + empty = pd.Series([], dtype=float) + return BacktestResult(equity_curve=empty, returns=empty, trades=[]) + signals = signals.reindex(ohlcv.index).ffill().fillna(Side.FLAT) - # Esecuzione con delay 1: segnale a t-1 esegue a open di t. - shifted = [Side.FLAT, *list(signals.iloc[:-1])] - executed_side = pd.Series(shifted, index=ohlcv.index, dtype=object) + executed = pd.Series( + [Side.FLAT, *list(signals.iloc[:-1])], + index=ohlcv.index, + dtype=object, + ) + + # Codifica side in int per vectorizzazione: 0=FLAT, +1=LONG, -1=SHORT. + side_code = np.where( + executed.values == Side.LONG, 1, + np.where(executed.values == Side.SHORT, -1, 0), + ).astype(np.int8) + opens = ohlcv["open"].to_numpy(dtype=np.float64) + closes = ohlcv["close"].to_numpy(dtype=np.float64) + ts_index = ohlcv.index + + # Identifica transizioni: punto in cui side[i] != side[i-1] (con side[-1]=0). + prev = np.concatenate(([0], side_code[:-1])) + change = side_code != prev + + # Indici di entry (cambio verso side != 0). + entry_idxs = np.flatnonzero(change & (side_code != 0)) + # Indici di chiusura: per ogni entry, il prossimo indice dove side[i] != side_entry. + # Vectorized: per ogni entry_idx, cerca change & side != side_entry oltre l'entry. - position: Position | None = None - position_entry_ts: pd.Timestamp | None = None trades: list[Trade] = [] - equity = 0.0 - equity_history: list[float] = [] - returns_history: list[float] = [] - prev_equity = 0.0 + # realized_pnl[t]: PnL netto cumulato dopo le chiusure avvenute a OPEN di t. + realized_pnl = np.zeros(n, dtype=np.float64) + fees_rate = self.fees_bp / 10000.0 + size = 1.0 - for ts, row in ohlcv.iterrows(): - target_side = executed_side.loc[ts] - current_side = position.side if position else Side.FLAT + # Posizione corrente all'inizio di ogni indice t (prima di applicare il transitorio): + # used per MtM computation. open_side_at_t / open_entry_at_t. + open_side = np.zeros(n, dtype=np.int8) + open_entry = np.zeros(n, dtype=np.float64) - if target_side != current_side: - if position is not None: - assert position_entry_ts is not None - trade = Trade( - entry_ts=position_entry_ts, - exit_ts=ts, - side=position.side, - size=position.size, - entry_price=position.entry_price, - exit_price=float(row["open"]), - fees_bp=self.fees_bp, - ) - trades.append(trade) - equity += trade.net_pnl - position = None - position_entry_ts = None - if target_side in (Side.LONG, Side.SHORT): - position = Position( - side=target_side, entry_price=float(row["open"]), size=1.0 - ) - position_entry_ts = ts + for entry_idx in entry_idxs: + entry_side = int(side_code[entry_idx]) + entry_price = opens[entry_idx] + # Cerca exit: primo indice > entry_idx dove side differisce. + after = side_code[entry_idx + 1:] + rel = np.flatnonzero(after != entry_side) + if rel.size > 0: + exit_idx = entry_idx + 1 + int(rel[0]) + exit_price = opens[exit_idx] + exit_ts = ts_index[exit_idx] + gross = entry_side * (exit_price - entry_price) * size + fees = fees_rate * size * (entry_price + exit_price) + net = gross - fees + # La chiusura avviene a open[exit_idx]: dal bar exit_idx in poi il + # PnL e' realizzato (non piu' MtM). + realized_pnl[exit_idx:] += net + # Posizione aperta in [entry_idx, exit_idx-1]. + open_side[entry_idx:exit_idx] = entry_side + open_entry[entry_idx:exit_idx] = entry_price + trades.append(Trade( + entry_ts=ts_index[entry_idx], + exit_ts=exit_ts, + side=Side.LONG if entry_side == 1 else Side.SHORT, + size=size, + entry_price=entry_price, + exit_price=exit_price, + fees_bp=self.fees_bp, + )) + else: + # Ultima posizione ancora aperta: chiusura forced a close[-1]. + # Parita' col loop legacy: MtM su [entry_idx, n-1), realized totale + # SOLO al bar n-1 (legacy fa equity_history[-1] = equity). + last_close = closes[-1] + gross = entry_side * (last_close - entry_price) * size + fees = fees_rate * size * (entry_price + last_close) + net = gross - fees + if entry_idx < n - 1: + open_side[entry_idx:n - 1] = entry_side + open_entry[entry_idx:n - 1] = entry_price + realized_pnl[-1] += net + trades.append(Trade( + entry_ts=ts_index[entry_idx], + exit_ts=ts_index[-1], + side=Side.LONG if entry_side == 1 else Side.SHORT, + size=size, + entry_price=entry_price, + exit_price=last_close, + fees_bp=self.fees_bp, + )) - mark = float(row["close"]) - mtm = position.unrealized_pnl(mark) if position else 0.0 - current_equity = equity + mtm - equity_history.append(current_equity) - returns_history.append(current_equity - prev_equity) - prev_equity = current_equity - - if position is not None: - assert position_entry_ts is not None - last_ts = ohlcv.index[-1] - last_close = float(ohlcv["close"].iloc[-1]) - trade = Trade( - entry_ts=position_entry_ts, - exit_ts=last_ts, - side=position.side, - size=position.size, - entry_price=position.entry_price, - exit_price=last_close, - fees_bp=self.fees_bp, - ) - trades.append(trade) - equity += trade.net_pnl - equity_history[-1] = equity - if len(returns_history) >= 2: - returns_history[-1] = equity - equity_history[-2] + # MtM unrealized per ogni bar in cui c'e' una posizione aperta. + mtm = open_side.astype(np.float64) * (closes - open_entry) * size + equity_arr = realized_pnl + mtm + # Returns = first diff dell'equity (col loop legacy il primo bar e' equity[0]-0). + returns_arr = np.concatenate(([equity_arr[0]], np.diff(equity_arr))) return BacktestResult( - equity_curve=pd.Series(equity_history, index=ohlcv.index, name="equity"), - returns=pd.Series(returns_history, index=ohlcv.index, name="returns"), + equity_curve=pd.Series(equity_arr, index=ts_index, name="equity"), + returns=pd.Series(returns_arr, index=ts_index, name="returns"), trades=trades, ) + + +# Lo facade Position re-export e' tenuto per backward-compat con import legacy. +__all__ = ["BacktestEngine", "BacktestResult", "Position", "Side", "Signal", "Trade"] diff --git a/src/multi_swarm_core/multi_swarm_core/orchestrator/run.py b/src/multi_swarm_core/multi_swarm_core/orchestrator/run.py index 52dad54..b86acc5 100644 --- a/src/multi_swarm_core/multi_swarm_core/orchestrator/run.py +++ b/src/multi_swarm_core/multi_swarm_core/orchestrator/run.py @@ -13,6 +13,7 @@ possa leggere lo stato a run terminato (o in corso). from __future__ import annotations import random +from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path @@ -20,13 +21,13 @@ import pandas as pd # type: ignore[import-untyped] from ..agents.adversarial import AdversarialAgent from ..agents.falsification import FalsificationAgent -from ..agents.hypothesis import HypothesisAgent +from ..agents.hypothesis import HypothesisAgent, HypothesisProposal, MarketSummary from ..agents.market_summary import build_market_summary from ..ga.fitness import compute_fitness from ..ga.initial import build_initial_population from ..ga.loop import GAConfig, next_generation from ..ga.summary import generation_summary -from ..genome.hypothesis import ModelTier +from ..genome.hypothesis import HypothesisAgentGenome, ModelTier from ..genome.mutation import set_cognitive_styles from ..genome.prompt_library import PromptLibrary from ..llm.client import LLMClient @@ -73,6 +74,29 @@ class RunConfig: # i 6 builtin (PromptLibrary.default()). Tipicamente caricata da # strategy_crypto/prompts.json via PromptLibrary.from_json(). prompt_library: PromptLibrary | None = None + # Numero di propose() LLM concorrenti per generazione. 1 = sequenziale (default, + # backward compat). 6-10 tipicamente accettati da OpenRouter qwen-2.5 senza + # rate-limit. Riduce wall time GA loop di 5-8x su tier C. + llm_concurrency: int = 1 + + +def _parallel_propose( + agent: HypothesisAgent, + genomes: list[HypothesisAgentGenome], + market: MarketSummary, + n_workers: int, +) -> list[HypothesisProposal]: + """Esegue ``agent.propose()`` su una lista di genomi, opzionalmente in parallelo. + + ``n_workers <= 1`` mantiene il comportamento serial originale (ordine fisso, + determinismo data un seed). ``n_workers > 1`` usa un thread pool: l'order + dei risultati e' preservato (1:1 con ``genomes``). OpenAI/openrouter client + e' thread-safe; ``PromptLibrary`` e ``HypothesisAgent`` non hanno stato mutabile. + """ + if n_workers <= 1 or len(genomes) <= 1: + return [agent.propose(g, market) for g in genomes] + with ThreadPoolExecutor(max_workers=n_workers) as pool: + return list(pool.map(lambda g: agent.propose(g, market), genomes)) def run_phase1( @@ -142,11 +166,20 @@ def run_phase1( try: for gen in range(cfg.n_generations): + # Step 1: raccogli i genomi da valutare in questa generazione (esclude + # elite gia' presenti nella cache fitnesses) e lancia propose() in + # parallelo. La sezione DB-write resta serial sotto. + uncached = [g for g in population if g.id not in fitnesses] + proposals = _parallel_propose( + hypothesis_agent, uncached, market, cfg.llm_concurrency + ) + proposal_by_id = {g.id: p for g, p in zip(uncached, proposals, strict=True)} + for genome in population: if genome.id in fitnesses: continue # elite gia' valutata in generazione precedente repo.save_genome(run_id=run_id, generation_idx=gen, genome=genome) - proposal = hypothesis_agent.propose(genome, market) + proposal = proposal_by_id[genome.id] # Registra costo per OGNI completion (incluse retry). for completion in proposal.completions: cost_record = cost_tracker.record( @@ -220,7 +253,7 @@ def run_phase1( cfg.fitness_combined_alpha * fit + (1.0 - cfg.fitness_combined_alpha) * fit_oos_inloop ) - except Exception: # noqa: BLE001 + except Exception: pass # fallback: usa solo IS repo.save_evaluation( run_id=run_id, @@ -261,7 +294,7 @@ def run_phase1( # WFA re-eval: i top_k genomi (by fitness in-sample > 0) vengono rivalutati # sul test_ohlcv. Le metriche OOS finiscono in evaluations.fitness_oos etc. if test_ohlcv is not None and len(test_ohlcv) >= 100: - from ..agents.hypothesis import _try_parse # noqa: PLC0415 + from ..agents.hypothesis import _try_parse all_evals = repo.list_evaluations(run_id) top_evals = sorted( @@ -276,7 +309,7 @@ def run_phase1( try: fals_oos = falsification_agent.evaluate(strategy, test_ohlcv) adv_oos = adversarial_agent.review(strategy, test_ohlcv) - except Exception: # noqa: BLE001 + except Exception: continue fit_oos = compute_fitness( fals_oos, adv_oos, diff --git a/src/multi_swarm_core/tests/unit/test_backtest_engine_vectorized.py b/src/multi_swarm_core/tests/unit/test_backtest_engine_vectorized.py new file mode 100644 index 0000000..ef67b1a --- /dev/null +++ b/src/multi_swarm_core/tests/unit/test_backtest_engine_vectorized.py @@ -0,0 +1,160 @@ +"""Parity check: engine vettorializzato vs reference iterrows implementation. + +Mantiene una copia inline del loop ``iterrows`` come reference per garantire +che la vettorizzazione produca esattamente gli stessi trades, equity_curve e +returns su input pseudocasuali, indipendentemente dal regime di prezzo. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest +from multi_swarm_core.backtest.engine import BacktestEngine, BacktestResult +from multi_swarm_core.backtest.orders import Position, Side, Trade + + +def _legacy_run( + ohlcv: pd.DataFrame, signals: pd.Series, fees_bp: float = 5.0 +) -> BacktestResult: + """Reference implementation: il loop iterrows originale (pre-vectorize). + + Mantenuto qui esclusivamente come oracolo per i test di parità. + """ + signals = signals.reindex(ohlcv.index).ffill().fillna(Side.FLAT) + shifted = [Side.FLAT, *list(signals.iloc[:-1])] + executed_side = pd.Series(shifted, index=ohlcv.index, dtype=object) + + position: Position | None = None + position_entry_ts: pd.Timestamp | None = None + trades: list[Trade] = [] + equity = 0.0 + equity_history: list[float] = [] + returns_history: list[float] = [] + prev_equity = 0.0 + + for ts, row in ohlcv.iterrows(): + target_side = executed_side.loc[ts] + current_side = position.side if position else Side.FLAT + if target_side != current_side: + if position is not None: + assert position_entry_ts is not None + trade = Trade( + entry_ts=position_entry_ts, + exit_ts=ts, + side=position.side, + size=position.size, + entry_price=position.entry_price, + exit_price=float(row["open"]), + fees_bp=fees_bp, + ) + trades.append(trade) + equity += trade.net_pnl + position = None + position_entry_ts = None + if target_side in (Side.LONG, Side.SHORT): + position = Position( + side=target_side, entry_price=float(row["open"]), size=1.0 + ) + position_entry_ts = ts + mark = float(row["close"]) + mtm = position.unrealized_pnl(mark) if position else 0.0 + current_equity = equity + mtm + equity_history.append(current_equity) + returns_history.append(current_equity - prev_equity) + prev_equity = current_equity + if position is not None: + assert position_entry_ts is not None + last_ts = ohlcv.index[-1] + last_close = float(ohlcv["close"].iloc[-1]) + trade = Trade( + entry_ts=position_entry_ts, + exit_ts=last_ts, + side=position.side, + size=position.size, + entry_price=position.entry_price, + exit_price=last_close, + fees_bp=fees_bp, + ) + trades.append(trade) + equity += trade.net_pnl + equity_history[-1] = equity + if len(returns_history) >= 2: + returns_history[-1] = equity - equity_history[-2] + + return BacktestResult( + equity_curve=pd.Series(equity_history, index=ohlcv.index, name="equity"), + returns=pd.Series(returns_history, index=ohlcv.index, name="returns"), + trades=trades, + ) + + +def _random_ohlcv(n: int, seed: int) -> pd.DataFrame: + rng = np.random.default_rng(seed) + rets = rng.normal(0.0, 0.01, size=n) + close = 100.0 * np.exp(np.cumsum(rets)) + idx = pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC") + return pd.DataFrame( + { + "open": close * (1 + rng.normal(0, 0.001, n)), + "high": close * 1.005, + "low": close * 0.995, + "close": close, + "volume": rng.uniform(1.0, 100.0, n), + }, + index=idx, + ) + + +def _random_signals(n: int, seed: int, p_change: float = 0.1) -> pd.Series: + """Segnali con persistenza: ad ogni bar con prob p_change cambia stato.""" + rng = np.random.default_rng(seed + 999) + states = [Side.LONG, Side.SHORT, Side.FLAT] + out: list[Side] = [rng.choice(states)] + for _ in range(1, n): + out.append(rng.choice(states) if rng.random() < p_change else out[-1]) + idx = pd.date_range("2024-01-01", periods=n, freq="1h", tz="UTC") + return pd.Series(out, index=idx, dtype=object) + + +@pytest.mark.parametrize("seed", [0, 1, 42, 123, 999]) +def test_vectorized_equals_legacy(seed: int) -> None: + df = _random_ohlcv(500, seed) + signals = _random_signals(500, seed) + engine = BacktestEngine(fees_bp=5.0) + new = engine.run(df, signals) + ref = _legacy_run(df, signals, fees_bp=5.0) + + pd.testing.assert_series_equal( + new.equity_curve, ref.equity_curve, rtol=1e-9, atol=1e-9 + ) + pd.testing.assert_series_equal( + new.returns, ref.returns, rtol=1e-9, atol=1e-9 + ) + assert len(new.trades) == len(ref.trades) + for nt, rt in zip(new.trades, ref.trades, strict=True): + assert nt.entry_ts == rt.entry_ts + assert nt.exit_ts == rt.exit_ts + assert nt.side == rt.side + assert nt.entry_price == pytest.approx(rt.entry_price, abs=1e-12) + assert nt.exit_price == pytest.approx(rt.exit_price, abs=1e-12) + assert nt.net_pnl == pytest.approx(rt.net_pnl, abs=1e-12) + + +def test_vectorized_handles_position_still_open_at_end() -> None: + """Edge case: signal LONG fino all'ultimo bar (exit a close[-1] forced).""" + df = _random_ohlcv(100, seed=7) + signals = pd.Series([Side.LONG] * 100, index=df.index) + new = BacktestEngine(fees_bp=10.0).run(df, signals) + ref = _legacy_run(df, signals, fees_bp=10.0) + pd.testing.assert_series_equal(new.equity_curve, ref.equity_curve, atol=1e-9) + assert len(new.trades) == 1 + assert new.trades[0].side == Side.LONG + + +def test_vectorized_zero_signals_no_trades() -> None: + df = _random_ohlcv(50, seed=3) + signals = pd.Series([Side.FLAT] * 50, index=df.index) + new = BacktestEngine().run(df, signals) + assert len(new.trades) == 0 + assert (new.equity_curve == 0).all() diff --git a/src/multi_swarm_core/tests/unit/test_orchestrator_parallel_propose.py b/src/multi_swarm_core/tests/unit/test_orchestrator_parallel_propose.py new file mode 100644 index 0000000..7a4b9e4 --- /dev/null +++ b/src/multi_swarm_core/tests/unit/test_orchestrator_parallel_propose.py @@ -0,0 +1,78 @@ +"""Test che `_parallel_propose` preservi l'ordine dei risultati e funzioni +sia in modalita' sequenziale (workers=1) che concorrente (workers>1). + +Non vogliamo testare il vero `HypothesisAgent.propose()` (che fa chiamate +LLM); usiamo un dummy con una latenza simulata per validare ordine e parallelismo. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Any + +from multi_swarm_core.orchestrator.run import _parallel_propose + + +@dataclass +class _FakeGenome: + id: str + + +@dataclass +class _FakeProposal: + genome_id: str + + +class _FakeAgent: + """Agent dummy: propose() dorme 50ms e ritorna un proposal con l'id del genome.""" + + def __init__(self, delay_s: float = 0.05) -> None: + self._delay = delay_s + self.call_count = 0 + + def propose(self, genome: _FakeGenome, market: Any) -> _FakeProposal: + time.sleep(self._delay) + self.call_count += 1 + return _FakeProposal(genome_id=genome.id) + + +def test_parallel_propose_preserves_order_serial() -> None: + agent = _FakeAgent(delay_s=0.01) + genomes = [_FakeGenome(id=f"g{i}") for i in range(5)] + results = _parallel_propose(agent, genomes, market=None, n_workers=1) + assert [r.genome_id for r in results] == ["g0", "g1", "g2", "g3", "g4"] + assert agent.call_count == 5 + + +def test_parallel_propose_preserves_order_concurrent() -> None: + agent = _FakeAgent(delay_s=0.05) + genomes = [_FakeGenome(id=f"g{i}") for i in range(8)] + results = _parallel_propose(agent, genomes, market=None, n_workers=4) + assert [r.genome_id for r in results] == [f"g{i}" for i in range(8)] + assert agent.call_count == 8 + + +def test_parallel_propose_actually_parallelizes() -> None: + """Wall time con 4 worker su 4 task da 100ms deve essere ~100ms, non ~400ms.""" + agent = _FakeAgent(delay_s=0.1) + genomes = [_FakeGenome(id=f"g{i}") for i in range(4)] + t0 = time.time() + _parallel_propose(agent, genomes, market=None, n_workers=4) + elapsed = time.time() - t0 + # serial sarebbe 0.4s; con 4 worker scendiamo a ~0.1s (max 0.2 per overhead). + assert elapsed < 0.2, f"expected <200ms with 4 workers, got {elapsed * 1000:.0f}ms" + + +def test_parallel_propose_handles_single_genome() -> None: + agent = _FakeAgent() + results = _parallel_propose(agent, [_FakeGenome(id="solo")], market=None, n_workers=8) + assert len(results) == 1 + assert results[0].genome_id == "solo" + + +def test_parallel_propose_empty_input() -> None: + agent = _FakeAgent() + results = _parallel_propose(agent, [], market=None, n_workers=4) + assert results == [] + assert agent.call_count == 0