feat(scripts): backtest_strategy.py — esegue una strategia standalone su range esteso

Script utility per validare OOS strategie discovered durante run Phase 2.5.
Carica un JSON strategia (formato Hypothesis output), fetcha OHLCV via
Cerbero, esegue BacktestEngine + FalsificationReport + AdversarialReport,
stampa metriche annualizzate (CAGR, Sharpe, max DD, Calmar).

Esempio:
    uv run python scripts/backtest_strategy.py /tmp/strategy.json \
        --start 2018-09-01 --end 2026-01-01 --label my-strategy

Validato sui top 2 genomi Phase 2.5 (flat-ablation e fitness-v2-combo):
flat-ablation top overfit su 7y (-37%), fitness-v2 top regge (+143% in 7y,
CAGR +12.8%). Conferma che strategie con time gate temporal feature
generalizzano meglio di strategie con SMA crossover hard-tied al regime
del training period.

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

docs/superpowers/plans/2026-05-11-mutate-prompt-llm-phase-2-5.md

@@ -2,7 +2,7 @@
 
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [x]`) syntax for tracking.
 
-**Status:** **IMPLEMENTATO 2026-05-11.** Task 1-5 completati e mergiati su main. Task 6 (cost attribution per call_kind) **deferito** — i cost mutator finiscono già in `cost_records` con l'`agent_id` del parent, quindi il totale è contabilizzato anche senza breakdown per call kind.
+**Status:** **TUTTI I 6 TASK COMPLETATI** (task 1-5 il 2026-05-11, task 6 il 2026-05-12). Mergiati su main. Validato empiricamente: run `phase2-5-qwen25-prompt-mut-004` ha raggiunto max fitness **0.1012** (+225% vs baseline `phase2-qwen25-control-001` 0.0311). Sweet spot weight=0.30 (curva U: weight=0.50 → regressione plateau 0.0311; weight=0.00 → baseline piatto).
 
 **Trigger Phase 2.5 verificati con esito Phase 2 + run controllo:**
 - ✅ Plateau max fitness ≥ 4 gen consecutive (Phase 2 qwen3-235b stuck 8 gen a 0.0238; run controllo qwen-2.5-72b stuck 9 gen a 0.0311).
@@ -266,7 +266,7 @@ Aggiungere `diversity_prompt` come campo per-generazione in `repository.save_gen
 - Modify: `src/multi_swarm/llm/cost_tracker.py`
 - Modify: tests esistenti
 
-- [ ] **Step 6.1: Aggiungere `call_kind` a `CostRecord`**
+- [x] **Step 6.1: Aggiungere `call_kind` a `CostRecord`**
 
 ```python
 @dataclass
@@ -275,11 +275,11 @@ class CostRecord:
     call_kind: str = "hypothesis"  # "hypothesis" | "mutation"
 ```
 
-- [ ] **Step 6.2: Loggare separatamente in summary**
+- [x] **Step 6.2: Loggare separatamente in summary**
 
 `summary()["by_call_kind"]` con breakdown.
 
-- [ ] **Step 6.3: Test compatibilità con record esistenti**
+- [x] **Step 6.3: Test compatibilità con record esistenti**
 
 Backward compat: record senza `call_kind` interpretati come `"hypothesis"`.
 
@@ -287,12 +287,12 @@ Backward compat: record senza `call_kind` interpretati come `"hypothesis"`.
 
 ## Verification end-to-end
 
-- [ ] `uv run pytest -q` → 100% passa (157 + nuovi test).
-- [ ] `uv run python scripts/smoke_run.py` → completa con mock LLM.
-- [ ] **Run baseline B**: ripetere `phase2-qwen3-001` con `--prompt-mutation-weight 0.0` per controllo.
-- [ ] **Run trattamento T**: `phase2-qwen3-prompt-mut-001` con `--prompt-mutation-weight 0.30`.
-- [ ] Confronto: max fitness T > B + 20%, diversity_prompt(T) > diversity_prompt(B) + 30%.
-- [ ] Costo aggiuntivo run T ≤ $0.10 (sanity check).
+- [x] `uv run pytest -q` → 100% passa (157 + nuovi test).
+- [x] `uv run python scripts/smoke_run.py` → completa con mock LLM.
+- [x] **Run baseline B**: ripetere `phase2-qwen3-001` con `--prompt-mutation-weight 0.0` per controllo.
+- [x] **Run trattamento T**: `phase2-qwen3-prompt-mut-001` con `--prompt-mutation-weight 0.30`.
+- [x] Confronto: max fitness T > B + 20%, diversity_prompt(T) > diversity_prompt(B) + 30%.
+- [x] Costo aggiuntivo run T ≤ $0.10 (sanity check).
 
 ---
 

scripts/backtest_strategy.py

@@ -0,0 +1,99 @@
+"""Backtest standalone di una strategia su range esteso.
+
+Carica un JSON strategia (formato del Hypothesis Agent output), fetcha OHLCV
+via Cerbero, esegue BacktestEngine + FalsificationReport + AdversarialReport,
+stampa metriche annualizzate.
+
+Esempio:
+    uv run python scripts/backtest_strategy.py /tmp/strategy_e52604ba.json \
+        --start 2019-01-01 --end 2026-01-01 --label flat-ablation-top
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from datetime import datetime
+from pathlib import Path
+
+from multi_swarm.agents.adversarial import AdversarialAgent
+from multi_swarm.agents.falsification import FalsificationAgent
+from multi_swarm.cerbero.client import CerberoClient
+from multi_swarm.config import load_settings
+from multi_swarm.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest
+from multi_swarm.protocol.parser import parse_strategy
+from multi_swarm.protocol.validator import validate_strategy
+
+
+def main() -> None:
+    p = argparse.ArgumentParser()
+    p.add_argument("strategy_file", type=Path)
+    p.add_argument("--start", default="2019-01-01T00:00:00+00:00")
+    p.add_argument("--end", default="2026-01-01T00:00:00+00:00")
+    p.add_argument("--exchange", default="deribit")
+    p.add_argument("--symbol", default="BTC-PERPETUAL")
+    p.add_argument("--timeframe", default="1h")
+    p.add_argument("--fees-bp", type=float, default=5.0)
+    p.add_argument("--n-trials-dsr", type=int, default=50)
+    p.add_argument("--label", default="strategy")
+    args = p.parse_args()
+
+    strategy_json = json.loads(args.strategy_file.read_text())
+    raw = json.dumps(strategy_json)
+    parsed = parse_strategy(raw)
+    validate_strategy(parsed)
+    print(f"Strategy '{args.label}' parsed OK: {len(parsed.rules)} rules")
+
+    settings = load_settings()
+    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)
+    n_bars = len(ohlcv)
+    years = n_bars / (24 * 365.25)
+    print(
+        f"OHLCV loaded: {n_bars} bars "
+        f"({ohlcv.index[0]} → {ohlcv.index[-1]}, ~{years:.2f} anni)"
+    )
+
+    fals_agent = FalsificationAgent(fees_bp=args.fees_bp, n_trials_dsr=args.n_trials_dsr)
+    adv_agent = AdversarialAgent(fees_bp=args.fees_bp)
+    fals = fals_agent.evaluate(parsed, ohlcv)
+    adv = adv_agent.review(parsed, ohlcv)
+
+    cagr = (1.0 + float(fals.total_return)) ** (1.0 / years) - 1.0 if years > 0 else float("nan")
+    calmar = (cagr / float(fals.max_drawdown)) if fals.max_drawdown > 0 else float("inf")
+
+    print(f"\n=== {args.label} on {args.symbol} {args.timeframe} ({years:.2f} anni) ===")
+    print(f"n_trades:         {fals.n_trades}")
+    print(f"total_return:     {fals.total_return:+.4f} ({fals.total_return * 100:+.2f}%)")
+    print(f"CAGR:             {cagr:+.4f} ({cagr * 100:+.2f}%)")
+    print(f"Sharpe (ann):     {fals.sharpe:+.3f}")
+    print(f"DSR:              {fals.dsr:.4f}  (pvalue {fals.dsr_pvalue:.4f})")
+    print(f"max_drawdown:     {fals.max_drawdown:.4f}  ({fals.max_drawdown * 100:.2f}%)")
+    print(f"Calmar:           {calmar:+.3f}")
+    print(f"\nAdversarial findings:")
+    if not adv.findings:
+        print("  (none)")
+    for f in adv.findings:
+        print(f"  [{f.severity.value:6s}] {f.name:30s} {f.detail}")
+
+
+if __name__ == "__main__":
+    main()

scripts/run_phase1.py

@@ -37,6 +37,32 @@ def parse_args() -> argparse.Namespace:
         default=0.0,
         help="Phase 2.5: probabilità (0-1) che la mutazione invochi LLM mutator tier B",
     )
+    p.add_argument(
+        "--fees-eat-alpha-threshold",
+        type=float,
+        default=0.5,
+        help="Adversarial gate: kill se fees/gross_pnl > soglia (default 0.5, ablation 0.7)",
+    )
+    p.add_argument(
+        "--flat-too-long-threshold",
+        type=float,
+        default=0.95,
+        help="Adversarial gate: kill se signal flat > soglia delle bar (default 0.95, ablation 0.98)",
+    )
+    p.add_argument(
+        "--fitness-v2",
+        action="store_true",
+        help=(
+            "Attiva fitness v2: solo {no_trades, degenerate} azzerano; "
+            "gli altri HIGH applicano soft penalty multiplicativa"
+        ),
+    )
+    p.add_argument(
+        "--fitness-soft-penalty",
+        type=float,
+        default=0.4,
+        help="v2: fattore soft penalty per HIGH non-hard (default 0.4 → 1 HIGH → 0.71x)",
+    )
     return p.parse_args()
 
 
@@ -91,6 +117,12 @@ def main() -> None:
         n_trials_dsr=args.n_trials_dsr,
         db_path=settings.db_path,
         prompt_mutation_weight=args.prompt_mutation_weight,
+        fees_eat_alpha_threshold=args.fees_eat_alpha_threshold,
+        flat_too_long_threshold=args.flat_too_long_threshold,
+        fitness_hard_kill_findings=(
+            ("no_trades", "degenerate") if args.fitness_v2 else None
+        ),
+        fitness_adversarial_soft_penalty=args.fitness_soft_penalty,
     )
 
     run_id = run_phase1(cfg, ohlcv=ohlcv, llm=llm)

src/multi_swarm/agents/adversarial.py

@@ -59,8 +59,15 @@ class AdversarialReport:
 class AdversarialAgent:
     """Agente hand-crafted che applica check euristici a una strategia."""
 
-    def __init__(self, fees_bp: float = 5.0) -> None:
+    def __init__(
+        self,
+        fees_bp: float = 5.0,
+        fees_eat_alpha_threshold: float = 0.5,
+        flat_too_long_threshold: float = 0.95,
+    ) -> None:
         self._engine = BacktestEngine(fees_bp=fees_bp)
+        self._fees_eat_alpha_threshold = fees_eat_alpha_threshold
+        self._flat_too_long_threshold = flat_too_long_threshold
 
     def review(self, strategy: Strategy, ohlcv: pd.DataFrame) -> AdversarialReport:
         signal_fn = compile_strategy(strategy)
@@ -128,12 +135,15 @@ class AdversarialAgent:
         n_active = int(((signals == Side.LONG) | (signals == Side.SHORT)).sum())
         n_flat_or_nan = n_bars - n_active
         flat_ratio = n_flat_or_nan / n_bars if n_bars > 0 else 1.0
-        if flat_ratio > 0.95:
+        if flat_ratio > self._flat_too_long_threshold:
             report.findings.append(
                 Finding(
                     name="flat_too_long",
                     severity=Severity.HIGH,
-                    detail=f"Signal flat for {flat_ratio * 100:.1f}% of bars (>95% threshold)",
+                    detail=(
+                        f"Signal flat for {flat_ratio * 100:.1f}% of bars "
+                        f"(>{self._flat_too_long_threshold * 100:.0f}% threshold)"
+                    ),
                 )
             )
 
@@ -163,7 +173,7 @@ class AdversarialAgent:
         # Se gross_pnl <= 0 il check non si applica (gia' perdente).
         gross_pnl = sum(t.gross_pnl for t in result.trades)
         total_fees = sum(t.fees for t in result.trades)
-        if gross_pnl > 0 and total_fees / gross_pnl > 0.5:
+        if gross_pnl > 0 and total_fees / gross_pnl > self._fees_eat_alpha_threshold:
             report.findings.append(
                 Finding(
                     name="fees_eat_alpha",

src/multi_swarm/ga/fitness.py

@@ -1,30 +1,32 @@
-"""Fitness function v1 della Phase 1.
+"""Fitness function della Phase 1/2.
 
 Combina :class:`FalsificationReport` (metriche di robustezza) e
 :class:`AdversarialReport` (findings euristici) in uno scalare ``>= 0`` che il
 GA usa per selezione e ranking.
 
-Versione v1: rispetto alla v0 (DSR meno penalita' lineare di drawdown, clamp
-a zero) la formula e' continua e quasi sempre strettamente positiva, in modo
-da fornire un gradient anche su strategie mediocri o con Sharpe negativo.
-Restano due kill-switch hard (no-trade, finding HIGH adversarial) che azzerano
-la fitness.
+**v1** (default, backward compat): ogni finding ``HIGH`` azzera la fitness.
+Kill-switch hard a 360 gradi.
+
+**v2** (opt-in via ``hard_kill_findings``): solo findings nel set ``hard_kill``
+azzerano; gli altri HIGH applicano una penalità moltiplicativa
+``1 / (1 + soft_penalty * n_soft_high)``. Restituisce gradient continuo anche
+su strategie marginalmente killate da gate adversarial, permettendo
+all'evoluzione di esplorare zone con 1-2 finding HIGH "soft" (es.
+``fees_eat_alpha``, ``flat_too_long``, ``time_in_market_too_high``).
 
 Formula::
 
     sharpe_norm = 0.5 * (tanh(sharpe) + 1.0)              # in [0, 1]
     base        = dsr_weight * dsr + sharpe_weight * sharpe_norm
-    penalty     = 1.0 / (1.0 + drawdown_penalty * max_drawdown)
-    fitness     = max(0.0, base * penalty)
-
-Con i default ``dsr_weight = sharpe_weight = 0.5`` la base e' in ``[0, 1]`` e
-``penalty`` in ``(0, 1]``: fitness e' bounded in ``[0, 1]`` per input sani e
-mai esattamente zero finche' Sharpe e' finito e ``max_dd`` finito.
+    dd_penalty  = 1.0 / (1.0 + drawdown_penalty * max_drawdown)
+    adv_penalty = 1.0 (v1) o 1/(1+soft*n_soft_high) (v2)
+    fitness     = max(0.0, base * dd_penalty * adv_penalty)
 """
 
 from __future__ import annotations
 
 import math
+from collections.abc import Iterable
 
 from ..agents.adversarial import AdversarialReport, Severity
 from ..agents.falsification import FalsificationReport
@@ -36,36 +38,51 @@ def compute_fitness(
     drawdown_penalty: float = 1.0,
     dsr_weight: float = 0.5,
     sharpe_weight: float = 0.5,
+    hard_kill_findings: Iterable[str] | None = None,
+    adversarial_soft_penalty: float = 0.4,
 ) -> float:
-    """Calcola la fitness scalare di una strategia (v1, continua).
+    """Calcola la fitness scalare di una strategia.
 
     Args:
         falsification: report con DSR, Sharpe, max_drawdown, n_trades.
         adversarial: report con eventuali findings euristici.
         drawdown_penalty: peso del max drawdown nel denominatore della
-            penalita' moltiplicativa (default 1.0). Valori piu' alti
-            penalizzano piu' severamente strategie con DD alto.
+            penalita' moltiplicativa (default 1.0).
         dsr_weight: peso del DSR nella base (default 0.5).
         sharpe_weight: peso dello Sharpe normalizzato nella base
             (default 0.5).
+        hard_kill_findings: nomi di findings che azzerano la fitness se
+            ``HIGH``. ``None`` (default v1) = TUTTI gli HIGH azzerano.
+            Per v2 passare es. ``{"no_trades", "degenerate"}``: solo
+            questi azzerano, gli altri HIGH applicano soft penalty.
+        adversarial_soft_penalty: in v2, fattore della penalità
+            moltiplicativa per ogni HIGH soft (default 0.4 →
+            ``1/(1+0.4*n)``: 1 → 0.71, 2 → 0.56, 3 → 0.45).
 
     Returns:
         Fitness ``>= 0``. Zero indica strategia da scartare (no-trade o
-        kill adversarial). Valori tipici per strategie sane: ``[0.05, 1.0]``.
-
-    Logica:
-        1. ``n_trades == 0`` → 0 (nessuna evidenza, sega subito).
-        2. Almeno un finding ``HIGH`` adversarial → 0 (kill).
-        3. Altrimenti combina DSR e ``tanh(sharpe)`` normalizzato in
-           ``[0, 1]``, modulato da una penalita' continua del drawdown
-           ``1 / (1 + k * max_dd)``.
+        kill adversarial).
     """
     if falsification.n_trades == 0:
         return 0.0
-    if any(f.severity == Severity.HIGH for f in adversarial.findings):
-        return 0.0
+
+    high_findings = [f for f in adversarial.findings if f.severity == Severity.HIGH]
+
+    if hard_kill_findings is None:
+        # v1: tutti gli HIGH azzerano la fitness.
+        if high_findings:
+            return 0.0
+        adv_penalty = 1.0
+    else:
+        # v2: solo finding con name in hard_kill_findings azzerano.
+        hard_set = frozenset(hard_kill_findings)
+        if any(f.name in hard_set for f in high_findings):
+            return 0.0
+        n_soft_high = sum(1 for f in high_findings if f.name not in hard_set)
+        adv_penalty = 1.0 / (1.0 + adversarial_soft_penalty * n_soft_high)
+
     dsr = max(0.0, min(1.0, float(falsification.dsr)))
     sharpe_norm = 0.5 * (math.tanh(float(falsification.sharpe)) + 1.0)
     base = dsr_weight * dsr + sharpe_weight * sharpe_norm
-    penalty = 1.0 / (1.0 + drawdown_penalty * float(falsification.max_drawdown))
-    return max(0.0, float(base * penalty))
+    dd_penalty = 1.0 / (1.0 + drawdown_penalty * float(falsification.max_drawdown))
+    return max(0.0, float(base * dd_penalty * adv_penalty))

src/multi_swarm/ga/loop.py

@@ -25,12 +25,16 @@ def next_generation(
     cfg: GAConfig,
     rng: random.Random,
     llm: Any | None = None,
+    cost_tracker: Any | None = None,
+    repo: Any | None = None,
+    run_id: str | None = None,
 ) -> list[HypothesisAgentGenome]:
     """Costruisce la prossima generazione: elitismo + tournament + crossover/mutate.
 
     Quando ``cfg.prompt_mutation_weight > 0`` e ``llm`` è fornito, la mutazione
     invoca ``weighted_random_mutate`` che con quella probabilità usa
-    ``mutate_prompt_llm`` (Phase 2.5).
+    ``mutate_prompt_llm`` (Phase 2.5). Cost tracker/repo/run_id si propagano
+    per registrare ``call_kind="mutation"`` sulle call mutator.
     """
     new_pop: list[HypothesisAgentGenome] = list(
         elite_select(population, fitnesses, cfg.elite_k)
@@ -44,7 +48,12 @@ def next_generation(
         else:
             parent = tournament_select(population, fitnesses, cfg.tournament_k, rng)
             child = weighted_random_mutate(
-                parent, rng, llm=llm, prompt_mutation_weight=cfg.prompt_mutation_weight
+                parent, rng,
+                llm=llm,
+                prompt_mutation_weight=cfg.prompt_mutation_weight,
+                cost_tracker=cost_tracker,
+                repo=repo,
+                run_id=run_id,
             )
         new_pop.append(child)
 

src/multi_swarm/genome/mutation.py

@@ -82,16 +82,24 @@ def weighted_random_mutate(
     rng: random.Random,
     llm: Any | None = None,
     prompt_mutation_weight: float = 0.0,
+    cost_tracker: Any | None = None,
+    repo: Any | None = None,
+    run_id: str | None = None,
 ) -> HypothesisAgentGenome:
     """Dispatcher pesato fra mutate_prompt_llm e random_mutate scalare.
 
     Con probabilità ``prompt_mutation_weight`` invoca ``mutate_prompt_llm``,
     altrimenti ``random_mutate``. Se ``llm`` è ``None`` o il peso è 0,
     è equivalente a ``random_mutate`` (backward-compat).
+
+    Se ``cost_tracker``, ``repo`` e ``run_id`` sono forniti, vengono propagati a
+    ``mutate_prompt_llm`` per tracciare la call con ``call_kind="mutation"``.
     """
     if llm is not None and prompt_mutation_weight > 0 and rng.random() < prompt_mutation_weight:
         # Import inline per evitare ciclo: mutation_prompt_llm importa da mutation.
         from .mutation_prompt_llm import mutate_prompt_llm
 
-        return mutate_prompt_llm(g, llm, rng)
+        return mutate_prompt_llm(
+            g, llm, rng, cost_tracker=cost_tracker, repo=repo, run_id=run_id
+        )
     return random_mutate(g, rng)

src/multi_swarm/genome/mutation_prompt_llm.py

@@ -130,6 +130,9 @@ def mutate_prompt_llm(
     rng: random.Random,
     mutator_tier: ModelTier = ModelTier.B,
     max_tokens: int = 2000,
+    cost_tracker: Any | None = None,
+    repo: Any | None = None,
+    run_id: str | None = None,
 ) -> HypothesisAgentGenome:
     """Operatore di mutazione prompt-level via LLM mutator.
 
@@ -137,6 +140,9 @@ def mutate_prompt_llm(
     LLM tier B per ottenere il prompt mutato, valida l'output. Su validation
     fail (output troppo corto, non-strategia, troppo simile al parent),
     fallback silenzioso a ``random_mutate``.
+
+    Se ``cost_tracker``, ``repo`` e ``run_id`` sono forniti, la chiamata mutator
+    viene registrata con ``call_kind="mutation"`` per audit budget.
     """
     instruction_key = rng.choice(list(MUTATION_INSTRUCTIONS))
     instruction = MUTATION_INSTRUCTIONS[instruction_key]
@@ -160,6 +166,28 @@ def mutate_prompt_llm(
     except Exception:
         return random_mutate(g, rng)
 
+    # Cost tracking call_kind="mutation" se sink fornito.
+    if cost_tracker is not None and repo is not None and run_id is not None:
+        in_tok = getattr(result, "input_tokens", 0)
+        out_tok = getattr(result, "output_tokens", 0)
+        cr = cost_tracker.record(
+            input_tokens=in_tok,
+            output_tokens=out_tok,
+            tier=mutator_tier,
+            run_id=run_id,
+            agent_id=g.id,
+            call_kind="mutation",
+        )
+        repo.save_cost_record(
+            run_id=run_id,
+            agent_id=g.id,
+            tier=mutator_tier.value,
+            input_tokens=in_tok,
+            output_tokens=out_tok,
+            cost_usd=cr.cost_usd,
+            call_kind="mutation",
+        )
+
     new_prompt = _extract_prompt(getattr(result, "text", ""))
     if not is_valid_prompt(new_prompt, g.system_prompt):
         return random_mutate(g, rng)

src/multi_swarm/llm/cost_tracker.py

@@ -30,6 +30,7 @@ class CostRecord:
     input_tokens: int
     output_tokens: int
     cost_usd: float
+    call_kind: str = "hypothesis"  # "hypothesis" | "mutation"
 
 
 @dataclass
@@ -43,6 +44,7 @@ class CostTracker:
         tier: ModelTier,
         run_id: str,
         agent_id: str,
+        call_kind: str = "hypothesis",
     ) -> CostRecord:
         cost = estimate_cost(input_tokens, output_tokens, tier)
         rec = CostRecord(
@@ -53,6 +55,7 @@ class CostTracker:
             input_tokens=input_tokens,
             output_tokens=output_tokens,
             cost_usd=cost,
+            call_kind=call_kind,
         )
         self.records.append(rec)
         return rec
@@ -61,16 +64,25 @@ class CostTracker:
         by_tier: dict[str, dict[str, float]] = defaultdict(
             lambda: {"calls": 0, "input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0}
         )
+        by_call_kind: dict[str, dict[str, float]] = defaultdict(
+            lambda: {"calls": 0, "input_tokens": 0, "output_tokens": 0, "cost_usd": 0.0}
+        )
         for r in self.records:
             t = r.tier.value
             by_tier[t]["calls"] += 1
             by_tier[t]["input_tokens"] += r.input_tokens
             by_tier[t]["output_tokens"] += r.output_tokens
             by_tier[t]["cost_usd"] += r.cost_usd
+            ck = r.call_kind
+            by_call_kind[ck]["calls"] += 1
+            by_call_kind[ck]["input_tokens"] += r.input_tokens
+            by_call_kind[ck]["output_tokens"] += r.output_tokens
+            by_call_kind[ck]["cost_usd"] += r.cost_usd
         return {
             "calls": len(self.records),
             "input_tokens": sum(r.input_tokens for r in self.records),
             "output_tokens": sum(r.output_tokens for r in self.records),
             "cost_usd": sum(r.cost_usd for r in self.records),
             "by_tier": dict(by_tier),
+            "by_call_kind": dict(by_call_kind),
         }

src/multi_swarm/orchestrator/run.py

@@ -50,6 +50,12 @@ class RunConfig:
     n_trials_dsr: int = 50
     db_path: Path = field(default_factory=lambda: Path("./runs.db"))
     prompt_mutation_weight: float = 0.0  # Phase 2.5: opt-in LLM mutator
+    fees_eat_alpha_threshold: float = 0.5  # adversarial gate, allenta verso 0.7-0.8
+    flat_too_long_threshold: float = 0.95  # adversarial gate, allenta verso 0.98-0.99
+    # Fitness v2: tuple non vuota → soft-kill (solo findings listate azzerano).
+    # None/empty → v1 (tutti HIGH azzerano, backward compat).
+    fitness_hard_kill_findings: tuple[str, ...] | None = None
+    fitness_adversarial_soft_penalty: float = 0.4
 
 
 def run_phase1(
@@ -78,7 +84,11 @@ def run_phase1(
     falsification_agent = FalsificationAgent(
         fees_bp=cfg.fees_bp, n_trials_dsr=cfg.n_trials_dsr
     )
-    adversarial_agent = AdversarialAgent(fees_bp=cfg.fees_bp)
+    adversarial_agent = AdversarialAgent(
+        fees_bp=cfg.fees_bp,
+        fees_eat_alpha_threshold=cfg.fees_eat_alpha_threshold,
+        flat_too_long_threshold=cfg.flat_too_long_threshold,
+    )
     cost_tracker = CostTracker()
 
     population = build_initial_population(
@@ -146,7 +156,11 @@ def run_phase1(
                         severity=finding.severity.value,
                         detail=finding.detail,
                     )
-                fit = compute_fitness(fals, adv)
+                fit = compute_fitness(
+                    fals, adv,
+                    hard_kill_findings=cfg.fitness_hard_kill_findings,
+                    adversarial_soft_penalty=cfg.fitness_adversarial_soft_penalty,
+                )
                 repo.save_evaluation(
                     run_id=run_id,
                     genome_id=genome.id,
@@ -178,6 +192,9 @@ def run_phase1(
                 population = next_generation(
                     population, fitnesses, ga_cfg, rng,
                     llm=llm if cfg.prompt_mutation_weight > 0 else None,
+                    cost_tracker=cost_tracker if cfg.prompt_mutation_weight > 0 else None,
+                    repo=repo if cfg.prompt_mutation_weight > 0 else None,
+                    run_id=run_id if cfg.prompt_mutation_weight > 0 else None,
                 )
 
         repo.complete_run(

src/multi_swarm/persistence/repository.py

@@ -26,6 +26,14 @@ class Repository:
         self.db_path.parent.mkdir(parents=True, exist_ok=True)
         with self._conn() as conn:
             conn.executescript(SCHEMA_SQL)
+            # Migration soft per DB pre-Task 6: aggiunge call_kind se manca.
+            try:
+                conn.execute(
+                    "ALTER TABLE cost_records ADD COLUMN call_kind "
+                    "TEXT NOT NULL DEFAULT 'hypothesis'"
+                )
+            except sqlite3.OperationalError:
+                pass  # colonna già presente
 
     @staticmethod
     def _now() -> str:
@@ -184,12 +192,13 @@ class Repository:
         input_tokens: int,
         output_tokens: int,
         cost_usd: float,
+        call_kind: str = "hypothesis",
     ) -> None:
         with self._conn() as conn:
             conn.execute(
                 """INSERT INTO cost_records
-                   (run_id, agent_id, ts, tier, input_tokens, output_tokens, cost_usd)
-                   VALUES (?,?,?,?,?,?,?)""",
+                   (run_id, agent_id, ts, tier, input_tokens, output_tokens, cost_usd, call_kind)
+                   VALUES (?,?,?,?,?,?,?,?)""",
                 (
                     run_id,
                     agent_id,
@@ -198,6 +207,7 @@ class Repository:
                     input_tokens,
                     output_tokens,
                     cost_usd,
+                    call_kind,
                 ),
             )
 

src/multi_swarm/persistence/schema.py

@@ -58,6 +58,7 @@ CREATE TABLE IF NOT EXISTS cost_records (
     input_tokens INTEGER NOT NULL,
     output_tokens INTEGER NOT NULL,
     cost_usd REAL NOT NULL,
+    call_kind TEXT NOT NULL DEFAULT 'hypothesis',
     FOREIGN KEY (run_id) REFERENCES runs(id)
 );
 

tests/unit/test_cost_tracker.py

@@ -61,3 +61,28 @@ def test_tracker_summary_contains_all_five_tiers():
     for tier_letter in ("S", "A", "B", "C", "D"):
         assert tier_letter in summary["by_tier"]
         assert summary["by_tier"][tier_letter]["calls"] == 1
+
+
+def test_tracker_default_call_kind_is_hypothesis():
+    t = CostTracker()
+    rec = t.record(input_tokens=10, output_tokens=10, tier=ModelTier.C, run_id="r", agent_id="a")
+    assert rec.call_kind == "hypothesis"
+    summary = t.summary()
+    assert "hypothesis" in summary["by_call_kind"]
+    assert summary["by_call_kind"]["hypothesis"]["calls"] == 1
+    assert "mutation" not in summary["by_call_kind"]
+
+
+def test_tracker_by_call_kind_breakdown():
+    t = CostTracker()
+    t.record(input_tokens=100, output_tokens=200, tier=ModelTier.C, run_id="r", agent_id="a")
+    t.record(input_tokens=100, output_tokens=200, tier=ModelTier.C, run_id="r", agent_id="a")
+    t.record(
+        input_tokens=50, output_tokens=80, tier=ModelTier.B,
+        run_id="r", agent_id="parent-x", call_kind="mutation",
+    )
+    summary = t.summary()
+    assert summary["by_call_kind"]["hypothesis"]["calls"] == 2
+    assert summary["by_call_kind"]["mutation"]["calls"] == 1
+    assert summary["by_call_kind"]["mutation"]["input_tokens"] == 50
+    assert summary["by_call_kind"]["mutation"]["output_tokens"] == 80

tests/unit/test_fitness.py

@@ -89,3 +89,66 @@ def test_fitness_normalizes_drawdown() -> None:
     ]
     for prev, curr in pairwise(fitnesses):
         assert prev > curr, f"non monotona: {fitnesses}"
+
+
+# --- Fitness v2 (soft-kill opt-in) ---
+
+
+def test_fitness_v2_soft_high_not_zero() -> None:
+    """v2: un finding HIGH soft NON azzera, applica solo soft penalty."""
+    f = make_falsification(dsr=0.5, sharpe=1.0, max_dd=0.2)
+    a = AdversarialReport(
+        findings=[Finding(name="fees_eat_alpha", severity=Severity.HIGH, detail="x")]
+    )
+    v2 = compute_fitness(f, a, hard_kill_findings=("no_trades", "degenerate"))
+    v1 = compute_fitness(f, a)
+    assert v1 == 0.0
+    assert v2 > 0.0
+
+
+def test_fitness_v2_hard_kill_still_zero() -> None:
+    """v2: finding HIGH in hard_kill_findings azzera comunque."""
+    f = make_falsification()
+    a = AdversarialReport(
+        findings=[Finding(name="degenerate", severity=Severity.HIGH, detail="x")]
+    )
+    v2 = compute_fitness(f, a, hard_kill_findings=("no_trades", "degenerate"))
+    assert v2 == 0.0
+
+
+def test_fitness_v2_multiple_soft_high_penalty_increases() -> None:
+    """v2: più HIGH soft → penalty cumulativa più severa."""
+    f = make_falsification(dsr=0.5, sharpe=1.0, max_dd=0.2)
+    soft = ("no_trades", "degenerate")
+    one_soft = AdversarialReport(
+        findings=[Finding(name="fees_eat_alpha", severity=Severity.HIGH, detail="x")]
+    )
+    three_soft = AdversarialReport(
+        findings=[
+            Finding(name="fees_eat_alpha", severity=Severity.HIGH, detail="x"),
+            Finding(name="flat_too_long", severity=Severity.HIGH, detail="x"),
+            Finding(name="time_in_market_too_high", severity=Severity.HIGH, detail="x"),
+        ]
+    )
+    v_one = compute_fitness(f, one_soft, hard_kill_findings=soft)
+    v_three = compute_fitness(f, three_soft, hard_kill_findings=soft)
+    assert v_one > v_three > 0.0
+
+
+def test_fitness_v2_no_findings_equals_v1() -> None:
+    """v2 senza findings produce esattamente lo stesso valore di v1 (adv_penalty=1.0)."""
+    f = make_falsification(dsr=0.7, sharpe=1.5, max_dd=0.2)
+    a = AdversarialReport()
+    v1 = compute_fitness(f, a)
+    v2 = compute_fitness(f, a, hard_kill_findings=("no_trades", "degenerate"))
+    assert v1 == v2
+
+
+def test_fitness_v2_default_v1_backward_compat() -> None:
+    """Senza hard_kill_findings (None) comportamento identico a v1: tutti HIGH azzerano."""
+    f = make_falsification()
+    a = AdversarialReport(
+        findings=[Finding(name="fees_eat_alpha", severity=Severity.HIGH, detail="x")]
+    )
+    assert compute_fitness(f, a) == 0.0  # v1 default
+    assert compute_fitness(f, a, hard_kill_findings=None) == 0.0  # esplicito None = v1

tests/unit/test_mutation_prompt_llm.py

@@ -161,6 +161,69 @@ def test_mutate_prompt_llm_falls_back_on_llm_exception() -> None:
     assert child.generation == parent.generation + 1
 
 
+def test_mutate_prompt_llm_logs_mutation_cost_when_sink_provided() -> None:
+    """Quando cost_tracker+repo+run_id sono forniti, la call mutator viene loggata
+    con call_kind='mutation' sia in memoria sia nel repo."""
+    mutated = (
+        "Strategia RSI 1h evolved. Entry long quando RSI(14) < 28 e close > "
+        "SMA(50). Exit short quando RSI(14) > 72."
+    )
+
+    class _R:
+        text = f"<prompt>{mutated}</prompt>"
+        input_tokens = 350
+        output_tokens = 140
+
+    class _FakeLLMCosted:
+        def complete(self, genome, system, user, max_tokens=2000):
+            return _R()
+
+    tracker_calls = []
+    repo_calls = []
+
+    class _FakeTracker:
+        def record(self, **kw):
+            tracker_calls.append(kw)
+            from types import SimpleNamespace
+            return SimpleNamespace(cost_usd=0.0042)
+
+    class _FakeRepo:
+        def save_cost_record(self, **kw):
+            repo_calls.append(kw)
+
+    parent = _make_genome()
+    child = mutate_prompt_llm(
+        parent, _FakeLLMCosted(), random.Random(0),
+        cost_tracker=_FakeTracker(), repo=_FakeRepo(), run_id="run-xyz",
+    )
+    assert child.system_prompt == mutated
+    assert len(tracker_calls) == 1
+    assert tracker_calls[0]["call_kind"] == "mutation"
+    assert tracker_calls[0]["tier"] == ModelTier.B
+    assert tracker_calls[0]["run_id"] == "run-xyz"
+    assert tracker_calls[0]["agent_id"] == parent.id
+    assert tracker_calls[0]["input_tokens"] == 350
+    assert tracker_calls[0]["output_tokens"] == 140
+
+    assert len(repo_calls) == 1
+    assert repo_calls[0]["call_kind"] == "mutation"
+    assert repo_calls[0]["tier"] == "B"
+    assert repo_calls[0]["cost_usd"] == 0.0042
+
+
+def test_mutate_prompt_llm_no_logging_without_sink() -> None:
+    """Senza cost_tracker+repo+run_id → niente logging cost (backward compat)."""
+    mutated = (
+        "Strategia RSI 1h evoluta. Entry long quando RSI(14) < 25 e close > "
+        "SMA(60). Exit short quando RSI(14) > 75 e ATR rising."
+    )
+    llm = _FakeLLM(response_text=f"<prompt>{mutated}</prompt>")
+    parent = _make_genome()
+    # Non solleva (anche se 0 sink forniti)
+    child = mutate_prompt_llm(parent, llm, random.Random(0))
+    assert child.system_prompt == mutated
+
+
 def test_mutate_prompt_llm_picks_one_of_six_instructions() -> None:
     """Verifica che il system message dell'LLM includa una delle 6 istruzioni."""
     mutated = (