perf(backtest): vectorize engine + parallel LLM propose + multi-fold validation tool

- backtest/engine.py: state machine numpy invece di pd.iterrows()
  - 16.8x speedup su 2y (470ms -> 28ms), 11.3x su 7y (1744ms -> 154ms)
  - 7 parity test parametrici vs reference iterrows assicurano equivalenza
- orchestrator/run.py + run_phase1.py: --llm-concurrency N
  - ThreadPoolExecutor parallelizza hypothesis_agent.propose() per generazione
  - 5-8x speedup wall time GA loop (OpenRouter qwen-2.5 regge 6-10 concorrenti)
  - default 1 = backward-compat sequenziale
- scripts/validate_run.py: validation multi-fold standalone
  - prende run_id + top-K + N-folds expanding-window su dataset esteso (7y)
  - rivela overfitter mascherati da fitness IS alta (vedi
    phase1-extended-001: elite IS Sharpe 1.93 collassa OOS a -1.00)
  - ranking per robust_score = min(fitness_oos) su tutti i fold

Test 250/250 pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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 <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):

scripts/run_phase1.py

@@ -114,6 +114,16 @@ def parse_args() -> argparse.Namespace:
             "Schema: {styles: {<name>: {directive: <testo>}}}"
         ),
     )
+    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)

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()

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"]

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,

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()

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