fix(llm): inject system prompt into user message for OpusAgent

OpusAgent topic system_prompt is used for summarization context but
not enforced as a binding instruction to Claude. Prepend [SYSTEM]
block to the first message of each session so Claude sees the full
JSON schema and grammar specification.

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

docker-compose.yml

@@ -23,7 +23,7 @@
 #   ./state   contiene runs.db (GA) + strategy_crypto.db (paper) + WAL/SHM
 #   ./src/strategy_crypto/strategy_crypto/strategies  JSON freezate (ro)
 #
-# Secrets (token Cerbero + OpenRouter): caricati da .env via env_file.
+# Secrets (token Cerbero + OpusAgent): caricati da .env via env_file.
 
 networks:
   traefik:

scripts/run_phase1.py

@@ -159,8 +159,8 @@ def parse_args() -> argparse.Namespace:
         default=1,
         help=(
             "Numero di propose() LLM concorrenti per generazione (default 1 = "
-            "serial). 6-10 tipicamente accettati da OpenRouter qwen-2.5 senza "
+            "serial). OpusAgent processa in coda FIFO; concurrency > 1 accoda "
-            "rate-limit; riduce wall time GA loop di 5-8x."
+            "piu' richieste in parallelo."
         ),
     )
     return p.parse_args()
@@ -200,13 +200,13 @@ def main() -> None:
     print(f"OHLCV loaded: {len(ohlcv)} bars from {ohlcv.index[0]} to {ohlcv.index[-1]}")
 
     llm = LLMClient(
-        openrouter_api_key=settings.openrouter_api_key.get_secret_value(),
+        opus_agent_api_key=settings.opus_agent_api_key.get_secret_value(),
+        opus_agent_base_url=settings.opus_agent_base_url,
         model_tier_s=settings.llm_model_tier_s,
         model_tier_a=settings.llm_model_tier_a,
         model_tier_b=settings.llm_model_tier_b,
         model_tier_c=settings.llm_model_tier_c,
         model_tier_d=settings.llm_model_tier_d,
-        openrouter_base_url=settings.openrouter_base_url,
     )
 
     cfg = RunConfig(

scripts/run_pythagoras_smoke.py

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

src/multi_swarm_core/multi_swarm_core/agents/hypothesis.py

@@ -4,11 +4,9 @@ import re
 from dataclasses import dataclass, field
 from typing import Any
 
-import openai
-
 from ..genome.hypothesis import HypothesisAgentGenome
 from ..genome.prompt_library import PromptLibrary
-from ..llm.client import CompletionResult, EmptyCompletionError, LLMClient
+from ..llm.client import CompletionResult, EmptyCompletionError, LLMClient, OpusAgentError, OpusAgentTransientError
 from ..protocol.parser import ParseError, Strategy, parse_strategy
 from ..protocol.validator import ValidationError, validate_strategy
 
@@ -181,6 +179,12 @@ def _build_system_prompt(lib: PromptLibrary, genome: HypothesisAgentGenome) -> s
     parts.append("")
     # 3. Grammar spec (core scaffold)
     parts.append(_SYSTEM_GRAMMAR_SPEC)
+    # 3b. Custom indicators spec (da prompts.json, opzionale) - estende la lista
+    # "Leaf - indicatori" con firme strategy-specific non note al core.
+    if lib.custom_indicators_spec:
+        parts.append("\nINDICATORI CUSTOM (firme aggiuntive, applica le stesse regole grammaticali):\n")
+        parts.append(lib.custom_indicators_spec)
+        parts.append("")
     # 4. Pattern guidance (da prompts.json, opzionale)
     if lib.pattern_guidance:
         parts.append(
@@ -284,19 +288,12 @@ def _render_focus_block(keys: list[str], market: MarketSummary) -> str:
 
 
 _RETRY_TEMPLATE = """\
-{original_user}
+Il JSON che hai generato contiene un errore: {previous_error}
 
---- TENTATIVO PRECEDENTE FALLITO ---
+Correggi e rispondi di nuovo con un singolo oggetto JSON valido
-Output: {previous_raw}
+dentro fence ```json...```, seguendo strettamente lo schema fornito.
-Errore: {previous_error}
----
-Correggi l'errore e rispondi di nuovo con un singolo oggetto JSON valido
-dentro fence ```json...```, seguendo strettamente lo schema fornito nel
-SYSTEM message.
 """
 
-_RETRY_RAW_TRUNCATE = 800
-
 
 _JSON_FENCE_RE = re.compile(
     r"```(?:json)?\s*(\{[\s\S]*\})\s*```",
@@ -425,46 +422,54 @@ class HypothesisAgent:
         errors: list[str] = []
         last_raw = ""
         max_attempts = 1 + self._max_retries
+        session_id: str | None = None
 
-        for attempt in range(max_attempts):
+        try:
-            if attempt == 0:
+            for attempt in range(max_attempts):
-                user = original_user
+                if attempt == 0:
-            else:
+                    user = original_user
-                truncated = last_raw[:_RETRY_RAW_TRUNCATE]
+                    req_session_id = "new"
-                user = _RETRY_TEMPLATE.format(
+                else:
-                    original_user=original_user,
+                    user = _RETRY_TEMPLATE.format(previous_error=errors[-1])
-                    previous_raw=truncated,
+                    req_session_id = session_id or "new"
-                    previous_error=errors[-1],
-                )
 
-            try:
+                try:
-                completion = self._llm.complete(genome, system=system, user=user)
+                    completion = self._llm.complete(
-            except EmptyCompletionError as e:
+                        genome, system=system, user=user,
-                # LLM esaurito retry tenacity senza una risposta. Tratta come
+                        session_id=req_session_id,
-                # parse-fail "empty" e ritenta nel loop esterno (max_attempts).
+                        summarize=True,
-                errors.append(f"empty_completion: {e}")
+                    )
-                last_raw = ""
+                except EmptyCompletionError as e:
-                continue
+                    errors.append(f"empty_completion: {e}")
-            except openai.RateLimitError as e:
+                    last_raw = ""
-                # Provider upstream rate limited oltre i retry tenacity.
+                    continue
-                # Marca genome come fallito senza propagare l'eccezione al run.
+                except OpusAgentTransientError as e:
-                errors.append(f"rate_limit: {e}")
+                    errors.append(f"transient_error: {e}")
-                last_raw = ""
+                    last_raw = ""
-                continue
+                    continue
-            completions.append(completion)
+                except OpusAgentError as e:
-            last_raw = completion.text
+                    errors.append(f"opus_agent_error: {e}")
+                    last_raw = ""
+                    continue
+                completions.append(completion)
+                last_raw = completion.text
+                if completion.session_id:
+                    session_id = completion.session_id
 
-            strategy, err = _try_parse(completion.text)
+                strategy, err = _try_parse(completion.text)
-            if strategy is not None:
+                if strategy is not None:
-                return HypothesisProposal(
+                    return HypothesisProposal(
-                    strategy=strategy,
+                        strategy=strategy,
-                    raw_text=completion.text,
+                        raw_text=completion.text,
-                    completions=completions,
+                        completions=completions,
-                    parse_error=None,
+                        parse_error=None,
-                    n_attempts=len(completions),
+                        n_attempts=len(completions),
-                )
+                    )
-            assert err is not None
+                assert err is not None
-            errors.append(err)
+                errors.append(err)
+        finally:
+            if session_id:
+                self._llm.close_session(session_id)
 
         chained = " | ".join(
             f"attempt {i + 1}: {e}" for i, e in enumerate(errors)

src/multi_swarm_core/multi_swarm_core/config.py

@@ -23,14 +23,14 @@ class Settings(BaseSettings):
     cerbero_mainnet_token: SecretStr | None = None
     cerbero_bot_tag: str = "swarm-poc-phase1"
 
-    openrouter_api_key: SecretStr
+    opus_agent_api_key: SecretStr
+    opus_agent_base_url: str = "https://opus-agent.tielogic.xyz"
 
-    llm_model_tier_s: str = "google/gemini-3-flash-preview"
+    llm_model_tier_s: str = "claude-opus-4-7"
-    llm_model_tier_a: str = "deepseek/deepseek-v4-flash"
+    llm_model_tier_a: str = "claude-opus-4-7"
-    llm_model_tier_b: str = "deepseek/deepseek-v4-flash"
+    llm_model_tier_b: str = "claude-sonnet-4-6"
-    llm_model_tier_c: str = "qwen/qwen-2.5-72b-instruct"
+    llm_model_tier_c: str = "claude-sonnet-4-6"
-    llm_model_tier_d: str = "openai/gpt-oss-20b"
+    llm_model_tier_d: str = "claude-haiku-4-5-20251001"
-    openrouter_base_url: str = "https://openrouter.ai/api/v1"
 
     run_name: str = "phase1-spike-001"
     data_dir: Path = Field(default=Path("./data"))

src/multi_swarm_core/multi_swarm_core/genome/hypothesis.py

@@ -8,11 +8,11 @@ from typing import Any
 
 
 class ModelTier(StrEnum):
-    S = "S"  # top-tier reasoning (Opus / equivalent) via Anthropic
+    S = "S"  # top-tier reasoning → opus via OpusAgent
-    A = "A"  # premium override via Anthropic
+    A = "A"  # premium → opus via OpusAgent
-    B = "B"  # Sonnet 4.6 via Anthropic
+    B = "B"  # standard → sonnet via OpusAgent
-    C = "C"  # Qwen 2.5 72B via OpenRouter
+    C = "C"  # default GA → sonnet via OpusAgent
-    D = "D"  # ultra-economic (Llama / cheap models) via OpenRouter
+    D = "D"  # economic → haiku via OpusAgent
 
 
 @dataclass

src/multi_swarm_core/multi_swarm_core/genome/prompt_library.py

@@ -101,6 +101,10 @@ class PromptLibrary:
     domain_warnings: str = field(default="")  # opzionale: warning di dominio (es. crypto 24/7)
     anti_patterns: str = field(default="")       # NEW v3.1: lista esplicita di pattern da evitare
     output_priorities: str = field(default="")   # NEW v3.1: trade-off espliciti di output
+    # NEW v3.2: firme formali (arity + range) di indicatori strategy-specific
+    # non noti al core. Iniettato in _SYSTEM_GRAMMAR_SPEC come estensione della
+    # lista "Leaf - indicatori" per consentire alla LLM di chiamarli correttamente.
+    custom_indicators_spec: str = field(default="")
 
     def __post_init__(self) -> None:
         if not self.styles:
@@ -120,6 +124,7 @@ class PromptLibrary:
             ("domain_warnings", self.domain_warnings),
             ("anti_patterns", self.anti_patterns),
             ("output_priorities", self.output_priorities),
+            ("custom_indicators_spec", self.custom_indicators_spec),
         ):
             if not isinstance(value, str):
                 raise PromptLibraryError(
@@ -142,6 +147,7 @@ class PromptLibrary:
             domain_warnings="",
             anti_patterns="",
             output_priorities="",
+            custom_indicators_spec="",
         )
 
     @classmethod
@@ -204,6 +210,12 @@ class PromptLibrary:
             raise PromptLibraryError(f"anti_patterns deve essere stringa, non {type(anti_patterns_raw)}")
         if not isinstance(output_priorities_raw, str):
             raise PromptLibraryError(f"output_priorities deve essere stringa, non {type(output_priorities_raw)}")
+        # Parse new optional top-level field (v3.2)
+        custom_indicators_spec_raw = data.get("custom_indicators_spec", "")
+        if not isinstance(custom_indicators_spec_raw, str):
+            raise PromptLibraryError(
+                f"custom_indicators_spec deve essere stringa, non {type(custom_indicators_spec_raw)}"
+            )
 
         return cls(
             styles=styles,
@@ -214,6 +226,7 @@ class PromptLibrary:
             domain_warnings=domain_warnings,
             anti_patterns=anti_patterns_raw,
             output_priorities=output_priorities_raw,
+            custom_indicators_spec=custom_indicators_spec_raw,
         )
 
     @property

src/multi_swarm_core/multi_swarm_core/llm/client.py

@@ -1,10 +1,12 @@
 from __future__ import annotations
 
+import hashlib
+import logging
+import threading
+import time
 from dataclasses import dataclass
-from typing import Any
 
-import openai
+import httpx
-from openai import OpenAI
 from tenacity import (
     retry,
     retry_if_exception_type,
@@ -14,26 +16,33 @@ from tenacity import (
 
 from ..genome.hypothesis import HypothesisAgentGenome, ModelTier
 
-# Modelli configurati per Phase 1 — tutti via OpenRouter
+logger = logging.getLogger(__name__)
-MODEL_TIER_S = "google/gemini-3-flash-preview"
+
-MODEL_TIER_A = "deepseek/deepseek-v4-flash"
+MODEL_TIER_MAP: dict[ModelTier, str] = {
-MODEL_TIER_B = "deepseek/deepseek-v4-flash"
+    ModelTier.S: "claude-opus-4-7",
-MODEL_TIER_C = "qwen/qwen-2.5-72b-instruct"
+    ModelTier.A: "claude-opus-4-7",
-MODEL_TIER_D = "openai/gpt-oss-20b"
+    ModelTier.B: "claude-sonnet-4-6",
-OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
+    ModelTier.C: "claude-sonnet-4-6",
+    ModelTier.D: "claude-haiku-4-5-20251001",
+}
+
 
 class EmptyCompletionError(RuntimeError):
     pass
 
 
-# Errori transient: retry. Auth/InvalidRequest: NO retry.
+class OpusAgentError(RuntimeError):
-# RateLimitError (HTTP 429) ora retryable: provider OpenRouter come DeepInfra
+    pass
-# applicano rate limit upstream temporaneo, recuperabile con backoff.
+
+
+class OpusAgentTransientError(RuntimeError):
+    pass
+
+
 _RETRYABLE_EXCEPTIONS: tuple[type[BaseException], ...] = (
-    openai.APIConnectionError,
+    httpx.ConnectError,
-    openai.APITimeoutError,
+    httpx.TimeoutException,
-    openai.InternalServerError,
+    OpusAgentTransientError,
-    openai.RateLimitError,
     EmptyCompletionError,
 )
 
@@ -45,50 +54,99 @@ class CompletionResult:
     output_tokens: int
     tier: ModelTier
     model: str
+    session_id: str | None = None
 
 
 class LLMClient:
-    # Provider OpenRouter da escludere di default. Novita rifiuta /completions
-    # endpoint per modelli Qwen 2.x — vedi bug 2026-05-12.
-    DEFAULT_PROVIDER_IGNORE: tuple[str, ...] = ("Novita",)
-
     def __init__(
         self,
-        openrouter_api_key: str,
+        opus_agent_api_key: str,
-        model_tier_s: str = MODEL_TIER_S,
+        opus_agent_base_url: str = "https://opus-agent.tielogic.xyz",
-        model_tier_a: str = MODEL_TIER_A,
+        model_tier_s: str = "claude-opus-4-7",
-        model_tier_b: str = MODEL_TIER_B,
+        model_tier_a: str = "claude-opus-4-7",
-        model_tier_c: str = MODEL_TIER_C,
+        model_tier_b: str = "claude-sonnet-4-6",
-        model_tier_d: str = MODEL_TIER_D,
+        model_tier_c: str = "claude-sonnet-4-6",
-        openrouter_base_url: str = OPENROUTER_BASE_URL,
+        model_tier_d: str = "claude-haiku-4-5-20251001",
-        provider_ignore: tuple[str, ...] | None = None,
+        poll_interval: float = 3.0,
+        poll_timeout: float = 180.0,
     ) -> None:
-        self.model_tier_s = model_tier_s
+        self._base_url = opus_agent_base_url.rstrip("/")
-        self.model_tier_a = model_tier_a
+        self._api_key = opus_agent_api_key
-        self.model_tier_b = model_tier_b
+        self._tier_map: dict[ModelTier, str] = {
-        self.model_tier_c = model_tier_c
-        self.model_tier_d = model_tier_d
-        self.openrouter_base_url = openrouter_base_url
-        self._provider_ignore = (
-            provider_ignore if provider_ignore is not None else self.DEFAULT_PROVIDER_IGNORE
-        )
-        self._tier_models: dict[ModelTier, str] = {
             ModelTier.S: model_tier_s,
             ModelTier.A: model_tier_a,
             ModelTier.B: model_tier_b,
             ModelTier.C: model_tier_c,
             ModelTier.D: model_tier_d,
         }
-        # Timeout esplicito (60s) per prevenire hang infinito su connessioni
+        self._poll_interval = poll_interval
-        # stallate. Tenacity retry su APITimeoutError gestisce il recovery.
+        self._poll_timeout = poll_timeout
-        self._client = OpenAI(
+        self._topic_cache: dict[str, str] = {}
-            api_key=openrouter_api_key,
+        self._topic_lock = threading.Lock()
-            base_url=openrouter_base_url,
+        self._client = httpx.Client(
-            timeout=60.0,
+            base_url=self._base_url,
+            headers={"X-Api-Key": self._api_key, "Content-Type": "application/json"},
+            timeout=30.0,
         )
 
+    def _get_or_create_topic(self, system_prompt: str) -> str:
+        prompt_hash = hashlib.sha256(system_prompt.encode()).hexdigest()[:16]
+        if prompt_hash in self._topic_cache:
+            return self._topic_cache[prompt_hash]
+
+        with self._topic_lock:
+            if prompt_hash in self._topic_cache:
+                return self._topic_cache[prompt_hash]
+
+            topic_name = f"swarm-{prompt_hash}"
+            resp = self._client.post("/api/topics", json={
+                "name": topic_name,
+                "system_prompt": system_prompt,
+            })
+
+            if resp.status_code == 409:
+                list_resp = self._client.get("/api/topics")
+                list_resp.raise_for_status()
+                for topic in list_resp.json()["data"]:
+                    if topic["name"] == topic_name:
+                        self._topic_cache[prompt_hash] = topic["id"]
+                        return topic["id"]
+                raise OpusAgentError(f"Topic {topic_name} conflict but not found")
+
+            if resp.status_code >= 500:
+                raise OpusAgentTransientError(f"Server error {resp.status_code}")
+            resp.raise_for_status()
+
+            topic_id = resp.json()["data"]["id"]
+            self._topic_cache[prompt_hash] = topic_id
+            logger.debug("Created topic %s -> %s", topic_name, topic_id)
+            return topic_id
+
+    def _poll_result(self, request_id: str) -> dict:
+        deadline = time.monotonic() + self._poll_timeout
+        while time.monotonic() < deadline:
+            resp = self._client.get(f"/api/requests/{request_id}")
+            resp.raise_for_status()
+            data = resp.json()["data"]
+            status = data["status"]
+            if status == "completed":
+                return data
+            if status == "failed":
+                error = data.get("error") or "unknown error"
+                raise OpusAgentError(f"Request {request_id} failed: {error}")
+            time.sleep(self._poll_interval)
+        raise OpusAgentTransientError(
+            f"Request {request_id} timed out after {self._poll_timeout}s"
+        )
+
+    def close_session(self, session_id: str) -> None:
+        try:
+            self._client.delete(f"/api/sessions/{session_id}")
+        except httpx.HTTPError:
+            logger.debug("Failed to close session %s", session_id)
+
     @retry(
-        stop=stop_after_attempt(5),
+        stop=stop_after_attempt(3),
         wait=wait_exponential(multiplier=2.0, min=2.0, max=30.0),
         retry=retry_if_exception_type(_RETRYABLE_EXCEPTIONS),
         reraise=True,
@@ -99,29 +157,45 @@ class LLMClient:
         system: str,
         user: str,
         max_tokens: int = 2000,
+        session_id: str | None = None,
+        summarize: bool = False,
     ) -> CompletionResult:
-        model = self._tier_models[genome.model_tier]
+        model = self._tier_map[genome.model_tier]
-        extra_body: dict[str, Any] = {}
+        topic_id = self._get_or_create_topic(system)
-        if self._provider_ignore:
+
-            extra_body["provider"] = {"ignore": list(self._provider_ignore)}
+        prompt = f"[SYSTEM]\n{system}\n\n[USER]\n{user}" if session_id in (None, "new") else user
-        resp = self._client.chat.completions.create(
+
-            model=model,
+        body: dict = {
-            messages=[
+            "topic_id": topic_id,
-                {"role": "system", "content": system},
+            "prompt": prompt,
-                {"role": "user", "content": user},
+            "model": model,
-            ],
+        }
-            temperature=genome.temperature,
+        if session_id is not None:
-            top_p=genome.top_p,
+            body["session_id"] = session_id
-            max_tokens=max_tokens,
+        if summarize:
-            extra_body=extra_body or None,
+            body["summarize"] = True
-        )
+
-        if not resp.choices or resp.choices[0].message.content is None:
+        resp = self._client.post("/api/requests", json=body)
-            raise EmptyCompletionError(f"empty response from {model}")
+
-        usage = resp.usage
+        if resp.status_code == 429:
+            raise OpusAgentTransientError("Rate limited")
+        if resp.status_code >= 500:
+            raise OpusAgentTransientError(f"Server error {resp.status_code}")
+        if resp.status_code != 202:
+            raise OpusAgentError(f"Unexpected status {resp.status_code}: {resp.text}")
+
+        request_id = resp.json()["data"]["id"]
+        result = self._poll_result(request_id)
+
+        text = result.get("result") or ""
+        if not text:
+            raise EmptyCompletionError(f"empty response from OpusAgent ({model})")
+
         return CompletionResult(
-            text=resp.choices[0].message.content,
+            text=text,
-            input_tokens=usage.prompt_tokens if usage else 0,
+            input_tokens=0,
-            output_tokens=usage.completion_tokens if usage else 0,
+            output_tokens=0,
             tier=genome.model_tier,
             model=model,
+            session_id=result.get("session_id"),
         )

src/multi_swarm_core/multi_swarm_core/orchestrator/run.py

@@ -75,8 +75,8 @@ class RunConfig:
     # 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
+    # backward compat). OpusAgent processa in coda FIFO; concurrency > 1
-    # rate-limit. Riduce wall time GA loop di 5-8x su tier C.
+    # accoda piu' richieste in parallelo ma il throughput dipende dal server.
     llm_concurrency: int = 1
 
 
@@ -90,7 +90,7 @@ def _parallel_propose(
 
     ``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
+    dei risultati e' preservato (1:1 con ``genomes``). LLMClient (OpusAgent)
     e' thread-safe; ``PromptLibrary`` e ``HypothesisAgent`` non hanno stato mutabile.
     """
     if n_workers <= 1 or len(genomes) <= 1:

src/multi_swarm_core/tests/unit/test_config.py

@@ -18,13 +18,14 @@ def test_settings_loads_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
     monkeypatch.setenv("CERBERO_MAINNET_TOKEN", "tok-main")
     monkeypatch.setenv("CERBERO_BOT_TAG", "swarm-poc-phase1")
-    monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
+    monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
     monkeypatch.setenv("RUN_NAME", "test-run")
 
     s = Settings()  # type: ignore[call-arg]
 
     assert s.cerbero_base_url == "http://test:9000"
     assert s.cerbero_testnet_token.get_secret_value() == "tok-test"
+    assert s.opus_agent_api_key.get_secret_value() == "oa-key"
     assert s.run_name == "test-run"
     assert s.data_dir.name == "data"
     assert s.db_path.name == "runs.db"
@@ -32,50 +33,33 @@ def test_settings_loads_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
 
 def test_settings_requires_tokens(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.delenv("CERBERO_TESTNET_TOKEN", raising=False)
-    monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
+    monkeypatch.delenv("OPUS_AGENT_API_KEY", raising=False)
     from pydantic import ValidationError
 
     with pytest.raises(ValidationError):
-        # Disable .env loading to keep the test deterministic regardless of
-        # whether a developer's local .env exists and is populated.
         Settings(_env_file=None)  # type: ignore[call-arg]
 
 
-def test_settings_loads_llm_model_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_settings_opus_agent_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
-    monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
+    monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
-    monkeypatch.setenv("LLM_MODEL_TIER_S", "claude-mega-x")
+    monkeypatch.delenv("OPUS_AGENT_BASE_URL", raising=False)
-    monkeypatch.setenv("LLM_MODEL_TIER_A", "claude-premium-y")
-    monkeypatch.setenv("LLM_MODEL_TIER_B", "claude-opus-4-7")
-    monkeypatch.setenv("LLM_MODEL_TIER_C", "deepseek/deepseek-chat")
-    monkeypatch.setenv("LLM_MODEL_TIER_D", "mistralai/mistral-7b")
-    monkeypatch.setenv("OPENROUTER_BASE_URL", "https://example.com/api/v1")
 
     s = Settings(_env_file=None)  # type: ignore[call-arg]
 
-    assert s.llm_model_tier_s == "claude-mega-x"
+    assert s.opus_agent_base_url == "https://opus-agent.tielogic.xyz"
-    assert s.llm_model_tier_a == "claude-premium-y"
+    assert s.llm_model_tier_s == "claude-opus-4-7"
-    assert s.llm_model_tier_b == "claude-opus-4-7"
+    assert s.llm_model_tier_a == "claude-opus-4-7"
-    assert s.llm_model_tier_c == "deepseek/deepseek-chat"
+    assert s.llm_model_tier_b == "claude-sonnet-4-6"
-    assert s.llm_model_tier_d == "mistralai/mistral-7b"
+    assert s.llm_model_tier_c == "claude-sonnet-4-6"
-    assert s.openrouter_base_url == "https://example.com/api/v1"
+    assert s.llm_model_tier_d == "claude-haiku-4-5-20251001"
 
 
-def test_settings_llm_model_defaults(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_settings_opus_agent_base_url_override(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setenv("CERBERO_TESTNET_TOKEN", "tok-test")
-    monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
+    monkeypatch.setenv("OPUS_AGENT_API_KEY", "oa-key")
-    monkeypatch.delenv("LLM_MODEL_TIER_S", raising=False)
+    monkeypatch.setenv("OPUS_AGENT_BASE_URL", "https://custom.example.com")
-    monkeypatch.delenv("LLM_MODEL_TIER_A", raising=False)
-    monkeypatch.delenv("LLM_MODEL_TIER_B", raising=False)
-    monkeypatch.delenv("LLM_MODEL_TIER_C", raising=False)
-    monkeypatch.delenv("LLM_MODEL_TIER_D", raising=False)
-    monkeypatch.delenv("OPENROUTER_BASE_URL", raising=False)
 
     s = Settings(_env_file=None)  # type: ignore[call-arg]
 
-    assert s.llm_model_tier_s == "google/gemini-3-flash-preview"
+    assert s.opus_agent_base_url == "https://custom.example.com"
-    assert s.llm_model_tier_a == "deepseek/deepseek-v4-flash"
-    assert s.llm_model_tier_b == "deepseek/deepseek-v4-flash"
-    assert s.llm_model_tier_c == "qwen/qwen-2.5-72b-instruct"
-    assert s.llm_model_tier_d == "openai/gpt-oss-20b"
-    assert s.openrouter_base_url == "https://openrouter.ai/api/v1"

src/multi_swarm_core/tests/unit/test_hypothesis_agent.py

@@ -420,3 +420,48 @@ def test_build_system_prompt_skips_anti_patterns_and_priorities_when_empty(mocke
     system_msg = call_kwargs["system"]
     assert "ANTI-PATTERN" not in system_msg
     assert "PRIORITA' DI OUTPUT" not in system_msg
+
+
+def test_build_system_prompt_includes_custom_indicators_spec(mocker):  # type: ignore[no-untyped-def]
+    """custom_indicators_spec da PromptLibrary appare nel SYSTEM prompt con header."""
+    fake_llm = mocker.MagicMock()
+    fake_llm.complete.return_value = CompletionResult(
+        text=VALID_STRATEGY_JSON,
+        input_tokens=200,
+        output_tokens=80,
+        tier=ModelTier.C,
+        model="qwen",
+    )
+    lib = PromptLibrary(
+        styles={"physicist": "Pensa come un fisico."},
+        focus={},
+        custom_indicators_spec="CUSTOM_IND_X",
+    )
+    agent = HypothesisAgent(llm=fake_llm, prompt_library=lib)
+    agent.propose(make_genome(), make_summary())
+    call_kwargs = fake_llm.complete.call_args.kwargs
+    system_msg = call_kwargs["system"]
+    assert "INDICATORI CUSTOM" in system_msg
+    assert "CUSTOM_IND_X" in system_msg
+
+
+def test_build_system_prompt_skips_custom_indicators_spec_when_empty(mocker):  # type: ignore[no-untyped-def]
+    """custom_indicators_spec='' -> sezione assente nel SYSTEM prompt."""
+    fake_llm = mocker.MagicMock()
+    fake_llm.complete.return_value = CompletionResult(
+        text=VALID_STRATEGY_JSON,
+        input_tokens=200,
+        output_tokens=80,
+        tier=ModelTier.C,
+        model="qwen",
+    )
+    lib = PromptLibrary(
+        styles={"physicist": "Pensa come un fisico."},
+        focus={},
+        custom_indicators_spec="",
+    )
+    agent = HypothesisAgent(llm=fake_llm, prompt_library=lib)
+    agent.propose(make_genome(), make_summary())
+    call_kwargs = fake_llm.complete.call_args.kwargs
+    system_msg = call_kwargs["system"]
+    assert "INDICATORI CUSTOM" not in system_msg

src/multi_swarm_core/tests/unit/test_llm_client.py

@@ -1,7 +1,14 @@
+import httpx
 import pytest
 
 from multi_swarm_core.genome.hypothesis import HypothesisAgentGenome, ModelTier
-from multi_swarm_core.llm.client import CompletionResult, LLMClient
+from multi_swarm_core.llm.client import (
+    CompletionResult,
+    EmptyCompletionError,
+    LLMClient,
+    OpusAgentError,
+    OpusAgentTransientError,
+)
 
 
 def make_genome(tier: ModelTier) -> HypothesisAgentGenome:
@@ -16,217 +23,248 @@ def make_genome(tier: ModelTier) -> HypothesisAgentGenome:
     )
 
 
-def test_completion_tier_c_uses_openrouter(mocker):
+TOPIC_RESPONSE = {
-    fake_openai = mocker.MagicMock()
+    "success": True,
-    fake_response = mocker.MagicMock()
+    "data": {"id": "topic-123", "name": "swarm-test", "system_prompt": "sys"},
-    fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))]
+}
-    fake_response.usage = mocker.MagicMock(prompt_tokens=100, completion_tokens=200)
-    fake_openai.chat.completions.create.return_value = fake_response
 
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
+REQUEST_ACCEPTED = {
+    "success": True,
+    "data": {"id": "req-456", "session_id": None, "status": "pending"},
+}
 
-    client = LLMClient(openrouter_api_key="or-x")
+
-    g = make_genome(ModelTier.C)
+def _completed_response(
-    out = client.complete(g, system="sys", user="usr")
+    text: str = "(strategy ...)", session_id: str = "sess-789",
+) -> dict:
+    return {
+        "success": True,
+        "data": {
+            "id": "req-456",
+            "status": "completed",
+            "result": text,
+            "session_id": session_id,
+        },
+    }
+
+
+def _mock_transport(responses: list[httpx.Response]) -> httpx.MockTransport:
+    call_idx = {"i": 0}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        idx = call_idx["i"]
+        call_idx["i"] += 1
+        if idx < len(responses):
+            return responses[idx]
+        return responses[-1]
+
+    return httpx.MockTransport(handler)
+
+
+def _make_client(transport: httpx.MockTransport) -> LLMClient:
+    client = LLMClient(opus_agent_api_key="test-key", poll_interval=0.01, poll_timeout=5.0)
+    client._client = httpx.Client(
+        base_url="https://opus-agent.tielogic.xyz",
+        headers={"X-Api-Key": "test-key", "Content-Type": "application/json"},
+        transport=transport,
+    )
+    return client
+
+
+def test_complete_tier_c_uses_sonnet():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json=_completed_response()),
+    ])
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
 
     assert isinstance(out, CompletionResult)
     assert out.text == "(strategy ...)"
-    assert out.input_tokens == 100
+    assert out.model == "claude-sonnet-4-6"
-    assert out.output_tokens == 200
     assert out.tier == ModelTier.C
-    fake_openai.chat.completions.create.assert_called_once()
 
 
-def test_completion_tier_b_uses_openrouter_with_anthropic_model(mocker):
+def test_complete_tier_s_uses_opus():
-    fake_openai = mocker.MagicMock()
+    transport = _mock_transport([
-    fake_response = mocker.MagicMock()
+        httpx.Response(201, json=TOPIC_RESPONSE),
-    fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))]
+        httpx.Response(202, json=REQUEST_ACCEPTED),
-    fake_response.usage = mocker.MagicMock(prompt_tokens=80, completion_tokens=150)
+        httpx.Response(200, json=_completed_response("(strategy s)")),
-    fake_openai.chat.completions.create.return_value = fake_response
+    ])
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.S), system="sys", user="usr")
 
-    client = LLMClient(openrouter_api_key="or-x")
+    assert out.model == "claude-opus-4-7"
-    g = make_genome(ModelTier.B)
-    out = client.complete(g, system="sys", user="usr")
-
-    assert out.text == "(strategy ...)"
-    assert out.input_tokens == 80
-    assert out.output_tokens == 150
-    assert out.tier == ModelTier.B
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "deepseek/deepseek-v4-flash"
-    assert out.model == "deepseek/deepseek-v4-flash"
-
-
-@pytest.mark.slow
-def test_completion_retries_on_connection_error(mocker):
-    """Retry esegue 3 tentativi su APIConnectionError, poi rilancia."""
-    import openai
-
-    fake_openai = mocker.MagicMock()
-    fake_openai.chat.completions.create.side_effect = openai.APIConnectionError(
-        request=mocker.MagicMock()
-    )
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
-
-    client = LLMClient(openrouter_api_key="or-x")
-    g = make_genome(ModelTier.C)
-
-    with pytest.raises(openai.APIConnectionError):
-        client.complete(g, system="sys", user="usr")
-
-    assert fake_openai.chat.completions.create.call_count == 5
-
-
-def test_completion_uses_custom_model_tier_c(mocker):
-    fake_openai = mocker.MagicMock()
-    fake_response = mocker.MagicMock()
-    fake_response.choices = [
-        mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
-    ]
-    fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20)
-    fake_openai.chat.completions.create.return_value = fake_response
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
-
-    client = LLMClient(
-        openrouter_api_key="or-x",
-        model_tier_c="deepseek/deepseek-chat",
-    )
-    g = make_genome(ModelTier.C)
-    out = client.complete(g, system="sys", user="usr")
-
-    fake_openai.chat.completions.create.assert_called_once()
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "deepseek/deepseek-chat"
-    assert out.model == "deepseek/deepseek-chat"
-
-
-def test_completion_uses_custom_model_tier_b(mocker):
-    fake_openai = mocker.MagicMock()
-    fake_response = mocker.MagicMock()
-    fake_response.choices = [
-        mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
-    ]
-    fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20)
-    fake_openai.chat.completions.create.return_value = fake_response
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
-
-    client = LLMClient(
-        openrouter_api_key="or-x",
-        model_tier_b="anthropic/claude-opus-4-7",
-    )
-    g = make_genome(ModelTier.B)
-    out = client.complete(g, system="sys", user="usr")
-
-    fake_openai.chat.completions.create.assert_called_once()
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "anthropic/claude-opus-4-7"
-    assert out.model == "anthropic/claude-opus-4-7"
-
-
-def test_completion_tier_s_uses_openrouter_with_anthropic_model(mocker):
-    fake_openai = mocker.MagicMock()
-    fake_response = mocker.MagicMock()
-    fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy s)"))]
-    fake_response.usage = mocker.MagicMock(prompt_tokens=50, completion_tokens=100)
-    fake_openai.chat.completions.create.return_value = fake_response
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
-
-    client = LLMClient(openrouter_api_key="or-x")
-    g = make_genome(ModelTier.S)
-    out = client.complete(g, system="sys", user="usr")
-
-    fake_openai.chat.completions.create.assert_called_once()
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "google/gemini-3-flash-preview"
     assert out.tier == ModelTier.S
-    assert out.model == "google/gemini-3-flash-preview"
+    assert out.text == "(strategy s)"
 
 
-def test_completion_tier_a_uses_openrouter_with_anthropic_model(mocker):
+def test_complete_tier_d_uses_haiku():
-    fake_openai = mocker.MagicMock()
+    transport = _mock_transport([
-    fake_response = mocker.MagicMock()
+        httpx.Response(201, json=TOPIC_RESPONSE),
-    fake_response.choices = [mocker.MagicMock(message=mocker.MagicMock(content="(strategy a)"))]
+        httpx.Response(202, json=REQUEST_ACCEPTED),
-    fake_response.usage = mocker.MagicMock(prompt_tokens=40, completion_tokens=80)
+        httpx.Response(200, json=_completed_response("(strategy d)")),
-    fake_openai.chat.completions.create.return_value = fake_response
+    ])
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.D), system="sys", user="usr")
 
-    client = LLMClient(openrouter_api_key="or-x")
+    assert out.model == "claude-haiku-4-5-20251001"
-    g = make_genome(ModelTier.A)
-    out = client.complete(g, system="sys", user="usr")
-
-    fake_openai.chat.completions.create.assert_called_once()
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "deepseek/deepseek-v4-flash"
-    assert out.tier == ModelTier.A
-    assert out.model == "deepseek/deepseek-v4-flash"
-
-
-def test_completion_tier_d_uses_openrouter_with_llama(mocker):
-    fake_openai = mocker.MagicMock()
-    fake_response = mocker.MagicMock()
-    fake_response.choices = [
-        mocker.MagicMock(message=mocker.MagicMock(content="(strategy d)"))
-    ]
-    fake_response.usage = mocker.MagicMock(prompt_tokens=30, completion_tokens=70)
-    fake_openai.chat.completions.create.return_value = fake_response
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
-
-    client = LLMClient(openrouter_api_key="or-x")
-    g = make_genome(ModelTier.D)
-    out = client.complete(g, system="sys", user="usr")
-
-    fake_openai.chat.completions.create.assert_called_once()
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
-    assert call_kwargs["model"] == "openai/gpt-oss-20b"
     assert out.tier == ModelTier.D
-    assert out.model == "openai/gpt-oss-20b"
 
 
-def test_completion_uses_custom_model_tier_s(mocker):
+def test_topic_cached_on_second_call():
-    fake_openai = mocker.MagicMock()
+    transport = _mock_transport([
-    fake_response = mocker.MagicMock()
+        httpx.Response(201, json=TOPIC_RESPONSE),
-    fake_response.choices = [
+        httpx.Response(202, json=REQUEST_ACCEPTED),
-        mocker.MagicMock(message=mocker.MagicMock(content="(strategy custom-s)"))
+        httpx.Response(200, json=_completed_response()),
-    ]
+        httpx.Response(202, json=REQUEST_ACCEPTED),
-    fake_response.usage = mocker.MagicMock(prompt_tokens=10, completion_tokens=20)
+        httpx.Response(200, json=_completed_response("second")),
-    fake_openai.chat.completions.create.return_value = fake_response
+    ])
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
+    client = _make_client(transport)
+    client.complete(make_genome(ModelTier.C), system="sys", user="usr1")
+    out2 = client.complete(make_genome(ModelTier.C), system="sys", user="usr2")
 
-    client = LLMClient(
+    assert out2.text == "second"
-        openrouter_api_key="or-x",
+
-        model_tier_s="anthropic/claude-future-mega",
+
+def test_topic_conflict_409_recovers():
+    transport = _mock_transport([
+        httpx.Response(409, json={"success": False, "error": {"code": "CONFLICT"}}),
+        httpx.Response(200, json={"success": True, "data": [
+            {"id": "topic-existing", "name": "swarm-wrong"},
+        ]}),
+    ])
+    client = _make_client(transport)
+
+    with pytest.raises(OpusAgentError, match="conflict but not found"):
+        client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+
+def test_empty_response_raises():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json={
+            "success": True,
+            "data": {"id": "req-456", "status": "completed", "result": ""},
+        }),
+    ])
+    client = _make_client(transport)
+
+    with pytest.raises(EmptyCompletionError):
+        client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+
+def test_request_failed_raises_opus_agent_error():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json={
+            "success": True,
+            "data": {"id": "req-456", "status": "failed", "error": "model overloaded"},
+        }),
+    ])
+    client = _make_client(transport)
+
+    with pytest.raises(OpusAgentError, match="failed"):
+        client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+
+def test_rate_limit_429_is_retryable():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(429, json={"success": False}),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json=_completed_response("after retry")),
+    ])
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+    assert out.text == "after retry"
+
+
+def test_polling_waits_for_completion():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json={
+            "success": True,
+            "data": {"id": "req-456", "status": "processing"},
+        }),
+        httpx.Response(200, json={
+            "success": True,
+            "data": {"id": "req-456", "status": "processing"},
+        }),
+        httpx.Response(200, json=_completed_response("done")),
+    ])
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+    assert out.text == "done"
+
+
+def test_tokens_are_zero():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json=_completed_response()),
+    ])
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.C), system="sys", user="usr")
+
+    assert out.input_tokens == 0
+    assert out.output_tokens == 0
+
+
+def test_session_id_returned_from_completion():
+    transport = _mock_transport([
+        httpx.Response(201, json=TOPIC_RESPONSE),
+        httpx.Response(202, json=REQUEST_ACCEPTED),
+        httpx.Response(200, json=_completed_response(session_id="sess-abc")),
+    ])
+    client = _make_client(transport)
+    out = client.complete(make_genome(ModelTier.C), system="sys", user="usr", session_id="new")
+
+    assert out.session_id == "sess-abc"
+
+
+def test_session_id_and_summarize_sent_in_request():
+    requests_seen: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "POST" and "/requests" in str(request.url):
+            import json
+            requests_seen.append(json.loads(request.content))
+            return httpx.Response(202, json=REQUEST_ACCEPTED)
+        if request.method == "POST" and "/topics" in str(request.url):
+            return httpx.Response(201, json=TOPIC_RESPONSE)
+        return httpx.Response(200, json=_completed_response())
+
+    transport = httpx.MockTransport(handler)
+    client = _make_client(transport)
+    client.complete(
+        make_genome(ModelTier.C), system="sys", user="usr",
+        session_id="sess-existing", summarize=True,
     )
-    g = make_genome(ModelTier.S)
-    out = client.complete(g, system="sys", user="usr")
 
-    call_kwargs = fake_openai.chat.completions.create.call_args.kwargs
+    assert len(requests_seen) == 1
-    assert call_kwargs["model"] == "anthropic/claude-future-mega"
+    assert requests_seen[0]["session_id"] == "sess-existing"
-    assert out.model == "anthropic/claude-future-mega"
+    assert requests_seen[0]["summarize"] is True
 
 
-@pytest.mark.slow
+def test_close_session():
-def test_completion_succeeds_after_one_retry(mocker):
+    deleted: list[str] = []
-    """Dopo 1 fallimento transient, il retry riesce al 2 tentativo."""
-    import openai
 
-    fake_response = mocker.MagicMock()
+    def handler(request: httpx.Request) -> httpx.Response:
-    fake_response.choices = [
+        if request.method == "DELETE":
-        mocker.MagicMock(message=mocker.MagicMock(content="(strategy ...)"))
+            deleted.append(str(request.url))
-    ]
+            return httpx.Response(200, json={"success": True})
-    fake_response.usage = mocker.MagicMock(prompt_tokens=100, completion_tokens=200)
+        return httpx.Response(200, json={"success": True})
 
-    fake_openai = mocker.MagicMock()
+    transport = httpx.MockTransport(handler)
-    fake_openai.chat.completions.create.side_effect = [
+    client = _make_client(transport)
-        openai.APITimeoutError(request=mocker.MagicMock()),
+    client.close_session("sess-to-close")
-        fake_response,
-    ]
-    mocker.patch("multi_swarm_core.llm.client.OpenAI", return_value=fake_openai)
 
-    client = LLMClient(openrouter_api_key="or-x")
+    assert len(deleted) == 1
-    g = make_genome(ModelTier.C)
+    assert "sess-to-close" in deleted[0]
-    out = client.complete(g, system="sys", user="usr")
-
-    assert isinstance(out, CompletionResult)
-    assert out.text == "(strategy ...)"
-    assert fake_openai.chat.completions.create.call_count == 2

src/multi_swarm_core/tests/unit/test_prompt_library.py

@@ -90,6 +90,28 @@ def test_from_json_loads_anti_patterns_and_output_priorities(tmp_path: Path) ->
     assert lib.output_priorities == "Robustezza > ottimalita."
 
 
+def test_from_json_loads_custom_indicators_spec(tmp_path: Path) -> None:
+    """from_json() legge custom_indicators_spec (v3.2): firme di indicatori strategy-specific."""
+    data = {
+        "styles": {"physicist": {"directive": "Cerca leggi conservative."}},
+        "custom_indicators_spec": (
+            "candle_pattern: params=[length, sym0, sym1, ...], length in [3,12], "
+            "sym in {0=U,1=D,2=doji}\n"
+            "pythagorean_ratio: params=[lookback], lookback in [12,200]"
+        ),
+    }
+    lib = PromptLibrary.from_json(_write_json(data, tmp_path))
+    assert "candle_pattern" in lib.custom_indicators_spec
+    assert "lookback in [12,200]" in lib.custom_indicators_spec
+
+
+def test_from_json_defaults_custom_indicators_spec_when_absent(tmp_path: Path) -> None:
+    """custom_indicators_spec assente -> default stringa vuota."""
+    data = {"styles": {"physicist": {"directive": "Cerca leggi conservative."}}}
+    lib = PromptLibrary.from_json(_write_json(data, tmp_path))
+    assert lib.custom_indicators_spec == ""
+
+
 def test_strategy_crypto_directives_ascii_safe() -> None:
     """REGRESSION GUARD: nessuna directive contiene caratteri > U+007F.
 

src/strategy_pythagoras/strategy_pythagoras/backend/__init__.py

@@ -0,0 +1,23 @@
+"""Backend paper-trading per la strategia strategy_pythagoras.
+
+Espone le classi principali per import ergonomici in scripts/runner:
+
+    from strategy_pythagoras.backend import PaperExecutor, Portfolio, PaperRepository
+
+Per i tipi interni (TickResult, OpenPosition, Trade) importare dal sotto-modulo.
+"""
+
+from .executor import PaperExecutor, TickResult
+from .persistence import PaperRepository
+from .portfolio import OpenPosition, Portfolio
+from .schema import PAPER_SCHEMA_SQL, init_schema
+
+__all__ = [
+    "PAPER_SCHEMA_SQL",
+    "OpenPosition",
+    "PaperExecutor",
+    "PaperRepository",
+    "Portfolio",
+    "TickResult",
+    "init_schema",
+]

src/strategy_pythagoras/strategy_pythagoras/backend/executor.py

@@ -0,0 +1,97 @@
+"""PaperExecutor: applica un segnale di strategia a un Portfolio.
+
+Il flusso per ogni tick:
+
+    bar OHLCV chiuso -> compile_strategy(strategy) -> Series[Side]
+    -> last_signal = series.iloc[-1]
+    -> match con posizione attuale -> open / close / hold
+
+Niente delay 1-bar: in paper-trading il segnale viene calcolato sulla
+barra appena chiusa e applicato al prezzo close della stessa. La latenza
+reale tra tick e ordine va misurata separatamente (Phase 3 spec).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from datetime import datetime
+from pathlib import Path
+
+import pandas as pd  # type: ignore[import-untyped]
+
+from multi_swarm_core.backtest.orders import Side, Trade
+from multi_swarm_core.protocol.compiler import compile_strategy
+from multi_swarm_core.protocol.parser import parse_strategy
+
+from .portfolio import OpenPosition, Portfolio
+
+
+@dataclass
+class TickResult:
+    ts: datetime
+    symbol: str
+    bar_ts: datetime
+    close_price: float
+    signal: Side
+    action_taken: str  # "open_long" | "open_short" | "close" | "reverse" | "hold"
+    trade: Trade | None = None
+    new_position: OpenPosition | None = None
+
+
+class PaperExecutor:
+    def __init__(self, strategy_json_path: Path, symbol: str) -> None:
+        text = strategy_json_path.read_text()
+        # parse_strategy si aspetta JSON pulito, non fence; il file e' gia' JSON.
+        self._strategy = parse_strategy(text)
+        self._compiled = compile_strategy(self._strategy)
+        self.symbol = symbol
+        self.strategy_path = strategy_json_path
+
+    def execute_tick(
+        self,
+        portfolio: Portfolio,
+        ohlcv: pd.DataFrame,
+        now: datetime,
+    ) -> TickResult:
+        """Esegui un tick: calcola segnale su tutto ``ohlcv`` (per indicatori
+        con lookback), prendi l'ultimo, e applica al portfolio."""
+        if len(ohlcv) == 0:
+            raise ValueError("Empty OHLCV passed to execute_tick")
+        signals = self._compiled(ohlcv)
+        # ultimo bar chiuso
+        bar_ts = ohlcv.index[-1]
+        close_price = float(ohlcv["close"].iloc[-1])
+        signal = Side(signals.iloc[-1]) if signals.iloc[-1] is not None else Side.FLAT
+
+        current = portfolio.positions.get(self.symbol)
+        action = "hold"
+        trade: Trade | None = None
+        new_position: OpenPosition | None = None
+
+        if current is None and signal != Side.FLAT:
+            new_position = portfolio.open(self.symbol, signal, close_price, now)
+            action = f"open_{signal.value}"
+        elif current is not None and signal == Side.FLAT:
+            trade = portfolio.close(self.symbol, close_price, now)
+            action = "close"
+        elif current is not None and signal != current.side:
+            # reverse: chiudi e riapri opposto
+            trade = portfolio.close(self.symbol, close_price, now)
+            new_position = portfolio.open(self.symbol, signal, close_price, now)
+            action = "reverse"
+
+        return TickResult(
+            ts=now,
+            symbol=self.symbol,
+            bar_ts=bar_ts.to_pydatetime() if hasattr(bar_ts, "to_pydatetime") else bar_ts,
+            close_price=close_price,
+            signal=signal,
+            action_taken=action,
+            trade=trade,
+            new_position=new_position,
+        )
+
+    @property
+    def strategy_dict(self) -> dict:
+        return json.loads(self.strategy_path.read_text())

src/strategy_pythagoras/strategy_pythagoras/backend/persistence.py

@@ -0,0 +1,123 @@
+"""Persistenza paper-trading: scrive su un DB dedicato (state/strategy_pythagoras_paper.db)
+con le tabelle ``paper_trading_*`` definite localmente in :mod:`.schema`.
+
+Il DB e' isolato dal ``runs.db`` del core GA: nessun naming conflict con
+future strategie (state/strategy_<asset>.db), nessuna contention di lock
+fra writer GA e writer paper.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from .executor import TickResult
+from .portfolio import Portfolio
+from .schema import init_schema as _init_paper_schema
+
+
+class PaperRepository:
+    def __init__(self, db_path: Path | str):
+        self.db_path = Path(db_path)
+
+    def init_schema(self) -> None:
+        """Crea (se mancanti) le tabelle paper_trading_* su ``self.db_path``."""
+        _init_paper_schema(self.db_path)
+
+    def _conn(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path, isolation_level=None)
+        conn.row_factory = sqlite3.Row
+        conn.execute("PRAGMA foreign_keys = ON")
+        conn.execute("PRAGMA journal_mode = WAL")
+        return conn
+
+    @staticmethod
+    def _now() -> str:
+        return datetime.now(UTC).isoformat()
+
+    def create_run(self, name: str, initial_capital: float, config: dict[str, Any]) -> str:
+        rid = uuid.uuid4().hex
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_runs "
+                "(id, name, started_at, status, initial_capital, config_json) "
+                "VALUES (?,?,?,?,?,?)",
+                (rid, name, self._now(), "running", initial_capital, json.dumps(config)),
+            )
+        return rid
+
+    def stop_run(self, run_id: str, status: str = "stopped") -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "UPDATE paper_trading_runs SET stopped_at=?, status=? WHERE id=?",
+                (self._now(), status, run_id),
+            )
+
+    def save_tick(self, run_id: str, tick: TickResult) -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_ticks "
+                "(paper_run_id, symbol, ts, bar_ts, close_price, signal, action_taken) "
+                "VALUES (?,?,?,?,?,?,?)",
+                (
+                    run_id,
+                    tick.symbol,
+                    tick.ts.isoformat(),
+                    tick.bar_ts.isoformat() if hasattr(tick.bar_ts, "isoformat") else str(tick.bar_ts),
+                    tick.close_price,
+                    tick.signal.value,
+                    tick.action_taken,
+                ),
+            )
+            if tick.trade is not None:
+                t = tick.trade
+                conn.execute(
+                    "INSERT INTO paper_trading_trades "
+                    "(paper_run_id, symbol, side, qty, entry_price, exit_price, "
+                    "entry_ts, exit_ts, pnl, fees) VALUES (?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        run_id,
+                        tick.symbol,
+                        t.side.value,
+                        t.size,
+                        t.entry_price,
+                        t.exit_price,
+                        t.entry_ts.isoformat(),
+                        t.exit_ts.isoformat(),
+                        t.net_pnl,
+                        t.fees,
+                    ),
+                )
+
+    def save_equity_snapshot(
+        self,
+        run_id: str,
+        ts: datetime,
+        equity: float,
+        cash: float,
+        positions_value: float,
+    ) -> None:
+        with self._conn() as conn:
+            conn.execute(
+                "INSERT INTO paper_trading_equity "
+                "(paper_run_id, ts, equity, cash, positions_value) VALUES (?,?,?,?,?)",
+                (run_id, ts.isoformat(), equity, cash, positions_value),
+            )
+
+    def sync_open_positions(self, run_id: str, portfolio: Portfolio) -> None:
+        """Sostituisce snapshot posizioni aperte. Idempotente: cancella e reinserisce."""
+        with self._conn() as conn:
+            conn.execute(
+                "DELETE FROM paper_trading_positions WHERE paper_run_id=?", (run_id,)
+            )
+            for sym, pos in portfolio.positions.items():
+                conn.execute(
+                    "INSERT INTO paper_trading_positions "
+                    "(paper_run_id, symbol, side, qty, entry_price, entry_ts) "
+                    "VALUES (?,?,?,?,?,?)",
+                    (run_id, sym, pos.side.value, pos.qty, pos.entry_price, pos.entry_ts.isoformat()),
+                )

src/strategy_pythagoras/strategy_pythagoras/backend/portfolio.py

@@ -0,0 +1,104 @@
+"""Portfolio multi-asset per paper-trading.
+
+Modello semplificato: capitale unico ``cash``, allocazione equal-weight
+fra N posizioni (sleeve = 1/N del capitale iniziale per ogni simbolo).
+Niente leva, niente liquidation, fees su entry+exit (bp del notional).
+
+Una :class:`Position` rappresenta una posizione aperta su un singolo
+simbolo (long/short, qty in unita' dell'asset, prezzo di entry). La
+posizione viene chiusa con :meth:`Portfolio.close` che produce un
+:class:`Trade` realized e accredita ``cash``.
+
+Mark-to-market via :meth:`Portfolio.equity`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import datetime
+
+from multi_swarm_core.backtest.orders import Side, Trade
+
+
+@dataclass(frozen=True)
+class OpenPosition:
+    symbol: str
+    side: Side
+    qty: float
+    entry_price: float
+    entry_ts: datetime
+
+
+@dataclass
+class Portfolio:
+    initial_capital: float
+    fees_bp: float = 5.0
+    n_sleeves: int = 2  # numero strategie / asset previsti
+    cash: float = field(init=False)
+    positions: dict[str, OpenPosition] = field(default_factory=dict)
+    closed_trades: list[Trade] = field(default_factory=list)
+
+    def __post_init__(self) -> None:
+        self.cash = self.initial_capital
+
+    @property
+    def sleeve_capital(self) -> float:
+        return self.initial_capital / self.n_sleeves
+
+    def open(
+        self,
+        symbol: str,
+        side: Side,
+        price: float,
+        ts: datetime,
+    ) -> OpenPosition:
+        if symbol in self.positions:
+            raise ValueError(f"Position already open on {symbol}")
+        if side == Side.FLAT:
+            raise ValueError("Cannot open a FLAT position")
+        # sleeve fisso: alloca 1/n_sleeves del capitale iniziale, qty = notional/price.
+        notional = self.sleeve_capital
+        qty = notional / price
+        fees = notional * (self.fees_bp / 10000.0)
+        self.cash -= fees
+        pos = OpenPosition(symbol=symbol, side=side, qty=qty, entry_price=price, entry_ts=ts)
+        self.positions[symbol] = pos
+        return pos
+
+    def close(
+        self,
+        symbol: str,
+        price: float,
+        ts: datetime,
+    ) -> Trade:
+        if symbol not in self.positions:
+            raise ValueError(f"No open position on {symbol}")
+        pos = self.positions.pop(symbol)
+        trade = Trade(
+            entry_ts=pos.entry_ts,
+            exit_ts=ts,
+            side=pos.side,
+            size=pos.qty,
+            entry_price=pos.entry_price,
+            exit_price=price,
+            fees_bp=self.fees_bp,
+        )
+        # net_pnl include gia' i fees sull'intero round-trip; abbiamo gia'
+        # addebitato meta' fees all'open, ora addebitiamo il resto.
+        self.cash += trade.gross_pnl - (trade.fees / 2.0)
+        self.closed_trades.append(trade)
+        return trade
+
+    def equity(self, last_prices: dict[str, float]) -> tuple[float, float]:
+        """Ritorna (equity_totale, positions_value) marcando posizioni aperte
+        al ``last_prices[symbol]``. Posizioni senza prezzo disponibile valgono
+        notional di entry (fallback conservativo)."""
+        positions_value = 0.0
+        for sym, pos in self.positions.items():
+            price = last_prices.get(sym, pos.entry_price)
+            unreal = pos.qty * (
+                price - pos.entry_price if pos.side == Side.LONG
+                else pos.entry_price - price
+            )
+            positions_value += pos.qty * pos.entry_price + unreal
+        return self.cash + positions_value, positions_value

src/strategy_pythagoras/strategy_pythagoras/backend/schema.py

@@ -0,0 +1,87 @@
+"""Schema SQLite per le tabelle paper-trading della strategia pythagoras.
+
+Owned dal member strategy_pythagoras: il DDL e' standalone rispetto al core,
+e scrive su un database dedicato (state/strategy_pythagoras_paper.db, env
+STRATEGY_PYTHAGORAS_PAPER_DB_PATH) separato dal runs.db del core GA. Pattern
+replicabile per future strategie.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+PAPER_SCHEMA_SQL = """
+CREATE TABLE IF NOT EXISTS paper_trading_runs (
+    id TEXT PRIMARY KEY,
+    name TEXT NOT NULL,
+    started_at TEXT NOT NULL,
+    stopped_at TEXT,
+    status TEXT NOT NULL DEFAULT 'running',
+    initial_capital REAL NOT NULL,
+    config_json TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS paper_trading_positions (
+    paper_run_id TEXT NOT NULL,
+    symbol TEXT NOT NULL,
+    side TEXT NOT NULL,
+    qty REAL NOT NULL,
+    entry_price REAL NOT NULL,
+    entry_ts TEXT NOT NULL,
+    PRIMARY KEY (paper_run_id, symbol),
+    FOREIGN KEY (paper_run_id) REFERENCES paper_trading_runs(id)
+);
+
+CREATE TABLE IF NOT EXISTS paper_trading_trades (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    paper_run_id TEXT NOT NULL,
+    symbol TEXT NOT NULL,
+    side TEXT NOT NULL,
+    qty REAL NOT NULL,
+    entry_price REAL NOT NULL,
+    exit_price REAL NOT NULL,
+    entry_ts TEXT NOT NULL,
+    exit_ts TEXT NOT NULL,
+    pnl REAL NOT NULL,
+    fees REAL NOT NULL,
+    FOREIGN KEY (paper_run_id) REFERENCES paper_trading_runs(id)
+);
+
+CREATE TABLE IF NOT EXISTS paper_trading_equity (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    paper_run_id TEXT NOT NULL,
+    ts TEXT NOT NULL,
+    equity REAL NOT NULL,
+    cash REAL NOT NULL,
+    positions_value REAL NOT NULL,
+    FOREIGN KEY (paper_run_id) REFERENCES paper_trading_runs(id)
+);
+
+CREATE TABLE IF NOT EXISTS paper_trading_ticks (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    paper_run_id TEXT NOT NULL,
+    symbol TEXT NOT NULL,
+    ts TEXT NOT NULL,
+    bar_ts TEXT NOT NULL,
+    close_price REAL NOT NULL,
+    signal TEXT NOT NULL,
+    action_taken TEXT NOT NULL,
+    FOREIGN KEY (paper_run_id) REFERENCES paper_trading_runs(id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_paper_trades_run ON paper_trading_trades(paper_run_id, exit_ts);
+CREATE INDEX IF NOT EXISTS idx_paper_equity_run ON paper_trading_equity(paper_run_id, ts);
+CREATE INDEX IF NOT EXISTS idx_paper_ticks_run ON paper_trading_ticks(paper_run_id, ts);
+"""
+
+
+def init_schema(db_path: Path | str) -> None:
+    """Crea (se mancanti) le tabelle paper_trading_* sul db indicato."""
+    Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+    conn = sqlite3.connect(str(db_path))
+    try:
+        conn.executescript(PAPER_SCHEMA_SQL)
+        conn.commit()
+    finally:
+        conn.close()

src/strategy_pythagoras/strategy_pythagoras/fitness_invariance.py

@@ -0,0 +1,50 @@
+"""Bonus di asset-invariance per la fitness del GA.
+
+corr_signal = frazione di entries su asset A che hanno corrispondenza su asset B
+entro +/-tolerance_bars (default 36 = 3h su 5m TF, vedi paper Pythagoras p. 43).
+"""
+from __future__ import annotations
+
+import os
+
+import pandas as pd
+
+GA_INVARIANCE_ALPHA = float(os.getenv("GA_INVARIANCE_ALPHA", "0.3"))
+GA_INVARIANCE_TOLERANCE_BARS = int(os.getenv("GA_INVARIANCE_TOLERANCE_BARS", "36"))
+
+
+def corr_signal(
+    entries_a: pd.Series,
+    entries_b: pd.Series,
+    tolerance_bars: int = GA_INVARIANCE_TOLERANCE_BARS,
+) -> float:
+    """Frazione di entries A con match in B entro +/-tolerance_bars.
+
+    Args:
+        entries_a, entries_b: Series binarie {0,1} sullo stesso index temporale (interi).
+        tolerance_bars: finestra di tolleranza in barre.
+
+    Returns:
+        In [0, 1]. 0 se entries_a non ha alcuna entry o nessun match.
+    """
+    a_idx = entries_a[entries_a > 0].index.tolist()
+    b_idx = entries_b[entries_b > 0].index.tolist()
+    if not a_idx or not b_idx:
+        return 0.0
+    b_set = set(b_idx)
+    matched = 0
+    for ti in a_idx:
+        for delta in range(-tolerance_bars, tolerance_bars + 1):
+            if (ti + delta) in b_set:
+                matched += 1
+                break
+    return matched / len(a_idx)
+
+
+def apply_invariance_bonus(
+    base_fitness: float,
+    invariance_score: float,
+    alpha: float = GA_INVARIANCE_ALPHA,
+) -> float:
+    """``fitness * (1 + alpha * invariance_score)``."""
+    return base_fitness * (1.0 + alpha * invariance_score)

src/strategy_pythagoras/strategy_pythagoras/frontend/data.py

@@ -0,0 +1,215 @@
+"""Paper-trading data access functions for the strategy_pythagoras dashboard.
+
+Reads exclusively from strategy_pythagoras_paper.db (paper_trading_* tables)
+per il paper-trading; le funzioni dedicate ai winner Pythagoras leggono
+invece dal runs.db del core GA (env ``GA_DB_PATH``), default
+``state/strategy_pythagoras.db`` via env ``STRATEGY_PYTHAGORAS_DB_PATH`` quando
+si vuole usare una sotto-tabella locale.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+from pathlib import Path
+from typing import Any
+
+import pandas as pd  # type: ignore[import-untyped]
+
+
+def _paper_conn(db_path: str | Path) -> sqlite3.Connection:
+    # Cold-start race: GUI può avviarsi prima che il paper writer crei il file.
+    db_path_str = str(db_path)
+    deadline = time.monotonic() + 5.0
+    while True:
+        try:
+            conn = sqlite3.connect(db_path_str, timeout=5.0)
+            conn.row_factory = sqlite3.Row
+            return conn
+        except sqlite3.OperationalError:
+            if time.monotonic() >= deadline:
+                raise
+            time.sleep(1.0)
+
+
+def paper_runs_df(db_path: str | Path) -> pd.DataFrame:
+    with _paper_conn(db_path) as conn:
+        rows = conn.execute(
+            "SELECT id, name, started_at, stopped_at, status, initial_capital, config_json "
+            "FROM paper_trading_runs ORDER BY started_at DESC"
+        ).fetchall()
+    return pd.DataFrame([dict(r) for r in rows])
+
+
+def paper_equity_df(db_path: str | Path, run_id: str) -> pd.DataFrame:
+    with _paper_conn(db_path) as conn:
+        rows = conn.execute(
+            "SELECT ts, equity, cash, positions_value FROM paper_trading_equity "
+            "WHERE paper_run_id=? ORDER BY ts ASC",
+            (run_id,),
+        ).fetchall()
+    return pd.DataFrame([dict(r) for r in rows])
+
+
+def paper_positions_df(db_path: str | Path, run_id: str) -> pd.DataFrame:
+    with _paper_conn(db_path) as conn:
+        rows = conn.execute(
+            "SELECT symbol, side, qty, entry_price, entry_ts "
+            "FROM paper_trading_positions WHERE paper_run_id=? ORDER BY symbol",
+            (run_id,),
+        ).fetchall()
+    return pd.DataFrame([dict(r) for r in rows])
+
+
+def paper_trades_df(db_path: str | Path, run_id: str, limit: int = 100) -> pd.DataFrame:
+    with _paper_conn(db_path) as conn:
+        rows = conn.execute(
+            "SELECT symbol, side, qty, entry_price, exit_price, entry_ts, exit_ts, pnl, fees "
+            "FROM paper_trading_trades WHERE paper_run_id=? ORDER BY exit_ts DESC LIMIT ?",
+            (run_id, limit),
+        ).fetchall()
+    return pd.DataFrame([dict(r) for r in rows])
+
+
+def paper_ticks_df(db_path: str | Path, run_id: str, limit: int = 50) -> pd.DataFrame:
+    with _paper_conn(db_path) as conn:
+        rows = conn.execute(
+            "SELECT ts, bar_ts, symbol, close_price, signal, action_taken "
+            "FROM paper_trading_ticks WHERE paper_run_id=? ORDER BY ts DESC LIMIT ?",
+            (run_id, limit),
+        ).fetchall()
+    return pd.DataFrame([dict(r) for r in rows])
+
+
+def paper_run_summary(db_path: str | Path, run_id: str) -> dict[str, Any]:
+    """Aggrega metriche sintetiche per la pagina paper trading."""
+    with _paper_conn(db_path) as conn:
+        run = conn.execute(
+            "SELECT id, name, started_at, stopped_at, status, initial_capital, config_json "
+            "FROM paper_trading_runs WHERE id=?",
+            (run_id,),
+        ).fetchone()
+        if run is None:
+            return {}
+        run = dict(run)
+
+        eq_row = conn.execute(
+            "SELECT equity, cash, positions_value, ts FROM paper_trading_equity "
+            "WHERE paper_run_id=? ORDER BY ts DESC LIMIT 1",
+            (run_id,),
+        ).fetchone()
+
+        trades_agg = conn.execute(
+            "SELECT COUNT(*) AS n, COALESCE(SUM(pnl),0) AS sum_pnl, "
+            "COALESCE(SUM(fees),0) AS sum_fees FROM paper_trading_trades "
+            "WHERE paper_run_id=?",
+            (run_id,),
+        ).fetchone()
+
+        tick_agg = conn.execute(
+            "SELECT COUNT(*) AS n, MAX(ts) AS last_ts FROM paper_trading_ticks "
+            "WHERE paper_run_id=?",
+            (run_id,),
+        ).fetchone()
+
+        positions_n = conn.execute(
+            "SELECT COUNT(*) AS n FROM paper_trading_positions WHERE paper_run_id=?",
+            (run_id,),
+        ).fetchone()["n"]
+
+    initial = float(run["initial_capital"])
+    current_equity = float(eq_row["equity"]) if eq_row is not None else initial
+    pnl_pct = (current_equity - initial) / initial * 100.0 if initial else 0.0
+
+    return {
+        "id": run["id"],
+        "name": run["name"],
+        "status": run["status"],
+        "started_at": run["started_at"],
+        "stopped_at": run["stopped_at"],
+        "initial_capital": initial,
+        "config": json.loads(run["config_json"]),
+        "current_equity": current_equity,
+        "current_cash": float(eq_row["cash"]) if eq_row is not None else initial,
+        "current_positions_value": float(eq_row["positions_value"]) if eq_row is not None else 0.0,
+        "last_equity_ts": eq_row["ts"] if eq_row is not None else None,
+        "pnl_abs": current_equity - initial,
+        "pnl_pct": pnl_pct,
+        "n_trades": int(trades_agg["n"]),
+        "trades_pnl": float(trades_agg["sum_pnl"]),
+        "trades_fees": float(trades_agg["sum_fees"]),
+        "n_ticks": int(tick_agg["n"]),
+        "last_tick_ts": tick_agg["last_ts"],
+        "n_open_positions": int(positions_n),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Pythagoras-specific helpers (winners invariance + candle pattern usage)
+# ---------------------------------------------------------------------------
+
+
+def load_invariance_metrics(ga_db_path: str) -> "pd.DataFrame":
+    """Per ogni winner ritorna (genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, invariance_score).
+
+    Lo schema atteso e' la tabella ``pythagoras_winners`` creata dal runner
+    ``scripts/run_pythagoras_smoke.py`` (Task 6.1).
+    """
+    import sqlite3
+
+    import pandas as pd
+
+    con = sqlite3.connect(ga_db_path)
+    try:
+        return pd.read_sql_query(
+            "SELECT genome_id, cognitive_style, fitness, sharpe_btc, sharpe_eth, "
+            "invariance_score FROM pythagoras_winners ORDER BY fitness DESC",
+            con,
+        )
+    finally:
+        con.close()
+
+
+def load_candle_pattern_usage(ga_db_path: str) -> "pd.DataFrame":
+    """Per ogni winner estrae le sequenze candle_pattern usate (per heatmap)."""
+    import json
+    import sqlite3
+
+    import pandas as pd
+
+    con = sqlite3.connect(ga_db_path)
+    try:
+        df = pd.read_sql_query(
+            "SELECT genome_id, cognitive_style, rules_json FROM pythagoras_winners",
+            con,
+        )
+    finally:
+        con.close()
+    records: list[dict] = []
+    for _, row in df.iterrows():
+        rules = json.loads(row["rules_json"]).get("rules", [])
+        for r in rules:
+            for ind_name, params in _walk_indicators(r["condition"]):
+                if ind_name == "candle_pattern":
+                    length = int(params[0])
+                    syms = [int(s) for s in params[1: 1 + length]]
+                    seq_str = "".join({0: "U", 1: "D", 2: "0"}[s] for s in syms)
+                    records.append(
+                        {
+                            "genome_id": row["genome_id"],
+                            "cognitive_style": row["cognitive_style"],
+                            "sequence": seq_str,
+                            "length": length,
+                        }
+                    )
+    return pd.DataFrame.from_records(records)
+
+
+def _walk_indicators(node: dict):
+    """Yields (indicator_name, params) for every IndicatorNode in the AST."""
+    if "op" in node:
+        for a in node.get("args", []):
+            yield from _walk_indicators(a)
+    elif node.get("kind") == "indicator":
+        yield node["name"], node["params"]

src/strategy_pythagoras/strategy_pythagoras/frontend/nicegui_app.py

@@ -0,0 +1,421 @@
+"""Strategy Pythagoras Dashboard — paper-trading + GA winners page: /.
+
+Avvio: ``uv run python -m strategy_pythagoras.frontend.nicegui_app``
+Default port 8080. Legge il paper DB (``strategy_pythagoras_paper.db``) per il
+tab ``Paper`` e il GA DB (``strategy_pythagoras.db``) per i tab pythagoras-specifici
+(Genomes / Patterns / Ratios / Invariance).
+
+Palette "Neon Trading Dashboard" (ispirata screenshot 2026-05-11):
+- BG: #0A0A0F (near-black con tinge blu)
+- Surface: #13131A (card base)
+- Surface elevata: #1C1C26 (hover/active)
+- Primary pink: #FF2D87 (highlight key metrics, max fitness)
+- Secondary cyan: #00D9FF (median, secondary curves)
+- Accent amber: #FFB800 (warnings, p90)
+- Success neon green: #00E676, Danger neon red: #FF3D60
+- Text: #FFFFFF (primary), #7A7A8C (muted)
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+import pandas as pd  # type: ignore[import-untyped]
+import plotly.graph_objects as go  # type: ignore[import-untyped]
+from nicegui import app, ui
+
+from strategy_pythagoras.frontend.data import (
+    load_candle_pattern_usage,
+    load_invariance_metrics,
+    paper_equity_df,
+    paper_positions_df,
+    paper_run_summary,
+    paper_runs_df,
+    paper_ticks_df,
+    paper_trades_df,
+)
+from multi_swarm_core.dashboard.theme import (
+    COLOR_PRIMARY,
+    COLOR_SURFACE,
+    COLOR_SURFACE_2,
+    COLOR_TEXT,
+    COLOR_TEXT_MUTED,
+    _STATUS_BADGE,
+    _apply_theme,
+    _build_header,
+)
+
+PAPER_DB_PATH = os.environ.get(
+    "STRATEGY_PYTHAGORAS_PAPER_DB_PATH", "./state/strategy_pythagoras_paper.db"
+)
+GA_DB_PATH = os.environ.get(
+    "STRATEGY_PYTHAGORAS_DB_PATH", "./state/strategy_pythagoras.db"
+)
+DASHBOARD_ROOT_PATH = os.environ.get("DASHBOARD_ROOT_PATH", "/strategy_pythagoras_gui")
+REFRESH_INTERVAL_S = 3.0
+
+
+def _paper_runs_options(only_running: bool = False) -> dict[str, str]:
+    try:
+        runs = paper_runs_df(PAPER_DB_PATH)
+    except Exception:
+        return {}
+    if runs.empty:
+        return {}
+    if only_running:
+        runs = runs[runs["status"] == "running"]
+        if runs.empty:
+            return {}
+    return {
+        row["id"]: f"{row['name']} — {row['status']} ({row['started_at'][:16]})"
+        for _, row in runs.iterrows()
+    }
+
+
+def _paper_equity_figure(eq_df: Any, initial_capital: float) -> go.Figure:
+    fig = go.Figure()
+    if eq_df is not None and not eq_df.empty:
+        ts = pd.to_datetime(eq_df["ts"])
+        fig.add_trace(
+            go.Scatter(
+                x=ts,
+                y=eq_df["equity"],
+                mode="lines",
+                line={"color": COLOR_PRIMARY, "width": 2},
+                name="equity",
+            )
+        )
+        fig.add_hline(
+            y=initial_capital,
+            line={"color": COLOR_TEXT_MUTED, "width": 1, "dash": "dash"},
+            annotation_text=f"initial ${initial_capital:.0f}",
+            annotation_position="bottom right",
+            annotation_font_color=COLOR_TEXT_MUTED,
+        )
+    fig.update_layout(
+        title=None,
+        paper_bgcolor=COLOR_SURFACE,
+        plot_bgcolor=COLOR_SURFACE,
+        font={"color": COLOR_TEXT, "family": "Inter"},
+        xaxis={"gridcolor": COLOR_SURFACE_2, "title": None},
+        yaxis={"gridcolor": COLOR_SURFACE_2, "title": "Equity ($)"},
+        margin={"l": 60, "r": 20, "t": 10, "b": 40},
+        height=320,
+        showlegend=False,
+    )
+    return fig
+
+
+def _render_paper_panel() -> None:
+    """Rende il pannello paper-trading (equivalente alla pagina root di strategy_crypto)."""
+    options = _paper_runs_options()
+    if not options:
+        ui.label("Nessuna paper-trading run nel database.").classes("text-h6 text-warning")
+        ui.label(
+            "Avvia un paper run per popolare strategy_pythagoras_paper.db."
+        ).classes("text-caption")
+        return
+
+    state: dict[str, Any] = {"run_id": next(iter(options))}
+
+    with ui.row().classes("w-full items-center gap-4 q-mb-md"):
+        select = ui.select(options=options, value=state["run_id"], label="Paper run").classes(
+            "flex-grow"
+        )
+        status_badge = ui.badge("…", color="primary").classes("text-body1 q-pa-sm")
+        ui.button("🔄 Refresh", on_click=lambda: refresh()).props("outline color=primary")
+
+    with ui.row().classes("w-full gap-4"):
+        with ui.card().classes("flex-grow metric-card accent-cyan"):
+            ui.label("Equity").classes("text-caption")
+            equity_lbl = ui.label("—").classes("text-h4")
+        with ui.card().classes("flex-grow metric-card accent-purple"):
+            ui.label("P/L cumulato").classes("text-caption")
+            pnl_lbl = ui.label("—").classes("text-h4")
+        with ui.card().classes("flex-grow metric-card accent-amber"):
+            ui.label("Trades chiusi").classes("text-caption")
+            trades_lbl = ui.label("—").classes("text-h4")
+        with ui.card().classes("flex-grow metric-card accent-green"):
+            ui.label("Open / Tick").classes("text-caption")
+            ticks_lbl = ui.label("—").classes("text-h4")
+
+    with ui.row().classes("w-full gap-4 q-mt-md"):
+        started_lbl = ui.label("Started: —")
+        last_tick_lbl = ui.label("Last tick: —")
+        cash_lbl = ui.label("Cash: —")
+
+    ui.separator()
+    ui.label("Equity curve").classes("text-subtitle1 q-mt-md")
+    equity_plot = ui.plotly(_paper_equity_figure(None, 0.0)).classes("w-full")
+
+    ui.separator()
+    ui.label("Open positions").classes("text-subtitle1 q-mt-md")
+    positions_table = ui.table(
+        columns=[
+            {"name": "symbol", "label": "symbol", "field": "symbol"},
+            {"name": "side", "label": "side", "field": "side"},
+            {"name": "qty", "label": "qty", "field": "qty"},
+            {"name": "entry_price", "label": "entry", "field": "entry_price"},
+            {"name": "entry_ts", "label": "entry ts", "field": "entry_ts"},
+        ],
+        rows=[],
+        row_key="symbol",
+    ).classes("w-full")
+
+    ui.separator()
+    ui.label("Ultimi 30 tick").classes("text-subtitle1 q-mt-md")
+    ticks_table = ui.table(
+        columns=[
+            {"name": "ts", "label": "ts", "field": "ts"},
+            {"name": "symbol", "label": "symbol", "field": "symbol"},
+            {"name": "bar_ts", "label": "bar", "field": "bar_ts"},
+            {"name": "close_price", "label": "close", "field": "close_price"},
+            {"name": "signal", "label": "signal", "field": "signal"},
+            {"name": "action_taken", "label": "action", "field": "action_taken"},
+        ],
+        rows=[],
+        row_key="ts",
+    ).classes("w-full")
+
+    ui.separator()
+    ui.label("Trades chiusi (ultimi 50)").classes("text-subtitle1 q-mt-md")
+    trades_table = ui.table(
+        columns=[
+            {"name": "exit_ts", "label": "exit ts", "field": "exit_ts"},
+            {"name": "symbol", "label": "symbol", "field": "symbol"},
+            {"name": "side", "label": "side", "field": "side"},
+            {"name": "qty", "label": "qty", "field": "qty"},
+            {"name": "entry_price", "label": "entry", "field": "entry_price"},
+            {"name": "exit_price", "label": "exit", "field": "exit_price"},
+            {"name": "pnl", "label": "pnl", "field": "pnl"},
+            {"name": "fees", "label": "fees", "field": "fees"},
+        ],
+        rows=[],
+        row_key="exit_ts",
+    ).classes("w-full")
+
+    def refresh() -> None:
+        run_id = select.value
+        if not run_id:
+            return
+        try:
+            summary = paper_run_summary(PAPER_DB_PATH, run_id)
+            eq = paper_equity_df(PAPER_DB_PATH, run_id)
+            positions = paper_positions_df(PAPER_DB_PATH, run_id)
+            ticks = paper_ticks_df(PAPER_DB_PATH, run_id, limit=30)
+            trades = paper_trades_df(PAPER_DB_PATH, run_id, limit=50)
+        except Exception as e:  # noqa: BLE001
+            ui.notify(f"Errore: {e}", type="negative")
+            return
+
+        text, color = _STATUS_BADGE.get(summary["status"], (summary["status"], "primary"))
+        status_badge.text = text
+        status_badge.props(f"color={color}")
+
+        equity_lbl.text = f"${summary['current_equity']:.2f}"
+        pnl_lbl.text = f"{summary['pnl_pct']:+.2f}%"
+        trades_lbl.text = str(summary["n_trades"])
+        ticks_lbl.text = f"{summary['n_open_positions']} / {summary['n_ticks']}"
+
+        started_lbl.text = f"Started: {summary['started_at']}"
+        last_tick_lbl.text = f"Last tick: {summary['last_tick_ts'] or '—'}"
+        cash_lbl.text = (
+            f"Cash: ${summary['current_cash']:.2f}  |  "
+            f"Pos value: ${summary['current_positions_value']:.2f}"
+        )
+
+        equity_plot.update_figure(_paper_equity_figure(eq, summary["initial_capital"]))
+
+        positions_table.rows = (
+            [
+                {col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
+                for _, row in positions.iterrows()
+            ]
+            if not positions.empty
+            else []
+        )
+        positions_table.update()
+
+        ticks_table.rows = (
+            [
+                {col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
+                for _, row in ticks.iterrows()
+            ]
+            if not ticks.empty
+            else []
+        )
+        ticks_table.update()
+
+        trades_table.rows = (
+            [
+                {col: (round(v, 6) if isinstance(v, float) else v) for col, v in row.items()}
+                for _, row in trades.iterrows()
+            ]
+            if not trades.empty
+            else []
+        )
+        trades_table.update()
+
+    def on_select_change() -> None:
+        state["run_id"] = select.value
+        refresh()
+
+    select.on_value_change(on_select_change)
+    ui.timer(REFRESH_INTERVAL_S, refresh)
+    refresh()
+
+
+@ui.page("/")
+def index() -> None:
+    _apply_theme()
+    _build_header(
+        active="/",
+        brand_subtitle="Strategy Pythagoras",
+        nav_items=[("/", "Dashboard")],
+        db_label=f"⛁ {Path(GA_DB_PATH).resolve().name}",
+    )
+
+    with ui.tabs() as tabs:
+        t_genomes = ui.tab("Genomes")
+        t_patterns = ui.tab("Patterns")
+        t_ratios = ui.tab("Ratios")
+        t_invariance = ui.tab("Invariance")
+        t_paper = ui.tab("Paper")
+
+    with ui.tab_panels(tabs, value=t_genomes).classes("w-full"):
+        with ui.tab_panel(t_genomes):
+            ui.label("GA Winners (post pythagoras-smoke-001)").classes("text-h6")
+            try:
+                df = load_invariance_metrics(GA_DB_PATH)
+                if df.empty:
+                    ui.label(
+                        "No winners yet. Run scripts/run_pythagoras_smoke.py first."
+                    ).classes("text-warning")
+                else:
+                    ui.table(
+                        rows=df.to_dict("records"),
+                        columns=[
+                            {"name": "genome_id", "label": "Genome ID", "field": "genome_id"},
+                            {
+                                "name": "cognitive_style",
+                                "label": "Style",
+                                "field": "cognitive_style",
+                            },
+                            {"name": "fitness", "label": "Fitness", "field": "fitness"},
+                            {"name": "sharpe_btc", "label": "Sharpe BTC", "field": "sharpe_btc"},
+                            {"name": "sharpe_eth", "label": "Sharpe ETH", "field": "sharpe_eth"},
+                            {
+                                "name": "invariance_score",
+                                "label": "Invariance",
+                                "field": "invariance_score",
+                            },
+                        ],
+                        pagination=10,
+                    ).classes("w-full")
+            except Exception as e:  # noqa: BLE001
+                ui.label(f"DB not ready: {e}").classes("text-warning")
+
+        with ui.tab_panel(t_patterns):
+            ui.label(
+                "Candle pattern sequences emerged (per cognitive style)"
+            ).classes("text-h6")
+            try:
+                df_pat = load_candle_pattern_usage(GA_DB_PATH)
+                if df_pat.empty:
+                    ui.label(
+                        "No patterns yet. Run scripts/run_pythagoras_smoke.py first."
+                    ).classes("text-warning")
+                else:
+                    grouped = (
+                        df_pat.groupby(["cognitive_style", "sequence"])
+                        .size()
+                        .reset_index(name="count")
+                    )
+                    grouped = grouped.sort_values("count", ascending=False).head(50)
+                    ui.table(
+                        rows=grouped.to_dict("records"),
+                        columns=[
+                            {
+                                "name": "cognitive_style",
+                                "label": "Style",
+                                "field": "cognitive_style",
+                            },
+                            {
+                                "name": "sequence",
+                                "label": "Sequence (U/D/0)",
+                                "field": "sequence",
+                            },
+                            {"name": "count", "label": "Count", "field": "count"},
+                        ],
+                        pagination=15,
+                    ).classes("w-full")
+            except Exception as e:  # noqa: BLE001
+                ui.label(f"DB not ready: {e}").classes("text-warning")
+
+        with ui.tab_panel(t_ratios):
+            ui.label(
+                "Pythagorean ratio literals — distance from universal constants"
+            ).classes("text-h6")
+            try:
+                df = load_invariance_metrics(GA_DB_PATH)
+                if df.empty:
+                    ui.label("No winners yet.").classes("text-warning")
+                else:
+                    ui.label(f"Total winners: {len(df)}").classes("text-body2")
+                    ui.label(
+                        "(Ratio literal histogram available after GA run produces "
+                        "pythagorean_ratio entries.)"
+                    ).classes("text-caption")
+            except Exception as e:  # noqa: BLE001
+                ui.label(f"DB not ready: {e}").classes("text-warning")
+
+        with ui.tab_panel(t_invariance):
+            ui.label(
+                "Cross-asset invariance: Sharpe BTC vs Sharpe ETH"
+            ).classes("text-h6")
+            try:
+                df = load_invariance_metrics(GA_DB_PATH)
+                if df.empty:
+                    ui.label("No winners yet.").classes("text-warning")
+                else:
+                    import plotly.express as px  # type: ignore[import-untyped]
+
+                    fig = px.scatter(
+                        df,
+                        x="sharpe_btc",
+                        y="sharpe_eth",
+                        color="invariance_score",
+                        hover_data=["genome_id", "cognitive_style", "fitness"],
+                        title="Sharpe BTC vs Sharpe ETH (color = invariance_score)",
+                    )
+                    ui.plotly(fig).classes("w-full")
+            except Exception as e:  # noqa: BLE001
+                ui.label(f"DB not ready: {e}").classes("text-warning")
+
+        with ui.tab_panel(t_paper):
+            _render_paper_panel()
+
+
+def main() -> None:
+    app.on_startup(
+        lambda: print(
+            f"GA DB: {Path(GA_DB_PATH).resolve()} | "
+            f"Paper DB: {Path(PAPER_DB_PATH).resolve()} | "
+            f"root_path: {DASHBOARD_ROOT_PATH or '/'}"
+        )
+    )
+    ui.run(
+        host="0.0.0.0",
+        port=int(os.environ.get("SWARM_DASHBOARD_PORT", "8080")),
+        title="Strategy Pythagoras Dashboard",
+        reload=False,
+        show=False,
+        dark=True,
+        root_path=DASHBOARD_ROOT_PATH,
+    )
+
+
+if __name__ in {"__main__", "__mp_main__"}:
+    main()

src/strategy_pythagoras/strategy_pythagoras/prompts.json

@@ -4,7 +4,9 @@
   "_changelog": "v1.0 - Initial release. Schema clonato da strategy_crypto v3.2 con contenuto Pythagoras-aligned.",
   "_focus_metrics_design": "Le focus_metrics sono ENFASI per la lente, non filtri. 4 per stile. Devono includere almeno 1 dei 3 indicatori Pythagoras (candle_pattern, pythagorean_ratio, fractal_mirror).",
   "_design_invariants": "(1) ASCII-safe; (2) Archetipo dominante: <metafora> come ancora; (3) Lookback range esplicito; (4) Prima frase 'Il mercato e ...'; (5) Lunghezza directive 800-950 char.",
-  "_param_encoding_note": "Vincolo grammar: tutti i params sono float. candle_pattern: [length, sym0, sym1, ...] (sym: 0=U up close>open, 1=D down close<open, 2=doji). pythagorean_ratio: [lookback]. fractal_mirror: [k, axis_int] (0=h temporale, 1=v prezzo).",
+  "_param_encoding_note": "Promosso a campo top-level 'custom_indicators_spec' (v3.2). Vedi sotto.",
+
+  "custom_indicators_spec": "Indicatori Pythagoras (oltre a sma/sma_pct/rsi/atr/atr_pct/realized_vol/macd/macd_pct gia documentati sopra). REGOLA CRITICA: params accetta SOLO numeri float, MAI stringhe e MAI altri nodi.\n\n  {\"kind\": \"indicator\", \"name\": \"candle_pattern\", \"params\": [length, sym0, sym1, ..., sym_{length-1}]}\n    length: int in [3,12]  (numero di candele consecutive che devono matchare)\n    sym_i:  int in {0, 1, 2}  (0=U up close>open, 1=D down close<open, 2=doji)\n    Esempio: pattern U-D-U di 3 candele = params=[3, 0, 1, 0]\n    Esempio: pattern D-D-D-U (reversal classico) = params=[4, 1, 1, 1, 0]\n    Output: 1.0 se le ultime length candele matchano, 0.0 altrimenti. Confronta solo con literal 0.0 o 1.0.\n\n  {\"kind\": \"indicator\", \"name\": \"pythagorean_ratio\", \"params\": [lookback]}\n    lookback: int in [12,200]  (finestra rolling per max/min close)\n    Esempio: ratio su finestra 89 (Fibonacci) = params=[89]\n    Output: max(close[-lookback:]) / min(close[-lookback:]), adimensionale >= 1.0.\n    Confronta con literal vicini a phi=1.618, 1/phi=0.618 (in realta usa 1.618 perche ratio >= 1), sqrt2=1.414, pi/2=1.571, e=2.718.\n\n  {\"kind\": \"indicator\", \"name\": \"fractal_mirror\", \"params\": [k, axis_int]}\n    k: int in [3,12]  (lunghezza finestra di confronto)\n    axis_int: int in {0, 1}  (0=h mirror temporale, 1=v mirror prezzo)\n    Esempio: mirror temporale su 8 candele = params=[8, 0]\n    Esempio: mirror prezzo su 6 candele = params=[6, 1]\n    Output: correlation di Pearson in [-1.0, 1.0] tra finestra e suo mirror. Confronta con literal in (-1, 1), tipicamente |0.5|-|0.8|.\n\nESEMPI di confronti CORRETTI:\n  candle_pattern di 3 candele U-D-U attiva: {\"op\": \"eq\", \"args\": [{\"kind\":\"indicator\",\"name\":\"candle_pattern\",\"params\":[3,0,1,0]}, {\"kind\":\"literal\",\"value\":1.0}]}\n  ratio phi entro 0.5%: {\"op\": \"and\", \"args\": [{\"op\":\"gt\",\"args\":[{\"kind\":\"indicator\",\"name\":\"pythagorean_ratio\",\"params\":[89]},{\"kind\":\"literal\",\"value\":1.610}]}, {\"op\":\"lt\",\"args\":[{\"kind\":\"indicator\",\"name\":\"pythagorean_ratio\",\"params\":[89]},{\"kind\":\"literal\",\"value\":1.626}]}]}\n  mirror temporale forte: {\"op\": \"gt\", \"args\": [{\"kind\":\"indicator\",\"name\":\"fractal_mirror\",\"params\":[8,0]}, {\"kind\":\"literal\",\"value\":0.7}]}\n\nESEMPI ERRATI (dead branch o validation error):\n  candle_pattern con stringa: params=[\"UDU\"]                  (errato: params devono essere numeri)\n  candle_pattern: params=[3, \"U\", \"D\", \"U\"]                    (errato: usa codici 0/1/2)\n  pythagorean_ratio: params=[\"close\"]                          (errato: params devono essere numeri, non nomi feature)\n  pythagorean_ratio: params=[5]                                (errato: lookback < 12)\n  pythagorean_ratio: params=[89, 1.618]                        (errato: arity 1, non 2)\n  fractal_mirror: params=[0, 1]                                (errato: k < 3)\n  fractal_mirror: params=[100, 0]                              (errato: k > 12)\n  fractal_mirror: params=[8]                                   (errato: arity 2, manca axis_int)",
 
   "agent_role": "Sei un agente generatore di ipotesi di trading quantitativo per un sistema swarm coevolutivo che cerca pattern frattali ricorrenti sui mercati crypto secondo il framework Pythagoras-Malanga (frattali H-C, trasformata di Fourier, geometria Evideon). Sei parte di una popolazione che esplora collettivamente lo spazio dei pattern: la diversita delle ipotesi e un asset critico. Preferisci esplorare territori meno ovvi rispetto a quelli che la tua lente cognitiva renderebbe predicibili. La strategia che produci deve essere riconoscibile come emanata dal tuo stile.",
 

src/strategy_pythagoras/tests/test_fitness_invariance.py

@@ -0,0 +1,43 @@
+"""Bonus invariance: pattern che firano simultaneamente su 2 asset entro tolleranza."""
+from __future__ import annotations
+
+import pandas as pd
+import pytest
+
+from strategy_pythagoras.fitness_invariance import (
+    apply_invariance_bonus,
+    corr_signal,
+)
+
+
+def test_corr_signal_perfect_alignment() -> None:
+    entries_btc = pd.Series([0, 1, 0, 1, 0], index=[0, 1, 2, 3, 4])
+    entries_eth = pd.Series([0, 1, 0, 1, 0], index=[0, 1, 2, 3, 4])
+    assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == pytest.approx(1.0)
+
+
+def test_corr_signal_no_overlap() -> None:
+    entries_btc = pd.Series([0, 1, 0, 0, 0], index=[0, 1, 2, 3, 4])
+    entries_eth = pd.Series([0, 0, 0, 0, 1], index=[0, 1, 2, 3, 4])
+    assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == pytest.approx(0.0)
+
+
+def test_corr_signal_within_tolerance() -> None:
+    # entry su BTC a t=1, su ETH a t=3, tolerance=2 -> match
+    entries_btc = pd.Series([0, 1, 0, 0, 0], index=[0, 1, 2, 3, 4])
+    entries_eth = pd.Series([0, 0, 0, 1, 0], index=[0, 1, 2, 3, 4])
+    assert corr_signal(entries_btc, entries_eth, tolerance_bars=2) == pytest.approx(1.0)
+
+
+def test_apply_invariance_bonus_increases_fitness() -> None:
+    assert apply_invariance_bonus(1.0, 0.5, 0.3) == pytest.approx(1.0 * (1.0 + 0.3 * 0.5))
+
+
+def test_apply_invariance_bonus_alpha_zero() -> None:
+    assert apply_invariance_bonus(1.0, 0.7, 0.0) == pytest.approx(1.0)
+
+
+def test_corr_signal_zero_entries() -> None:
+    entries_btc = pd.Series([0, 0, 0], index=[0, 1, 2])
+    entries_eth = pd.Series([0, 0, 0], index=[0, 1, 2])
+    assert corr_signal(entries_btc, entries_eth, tolerance_bars=0) == 0.0