From c38311e5fac722f9e1c0f01a976537d488e6b4d8 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Mon, 11 May 2026 23:49:46 +0200 Subject: [PATCH] feat(phase-2.5): mutate_prompt_llm operator + weighted dispatcher + GA wiring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implementazione completa Phase 2.5 (LLM prompt mutator) seguendo il piano in docs/superpowers/plans/2026-05-11-mutate-prompt-llm-phase-2-5.md. Nuovo modulo src/multi_swarm/genome/mutation_prompt_llm.py: - 6 mutation instructions (tighten_threshold, swap_comparator, add_condition, remove_condition, change_timeframe, add_temporal_gate) - mutate_prompt_llm(g, llm, rng, mutator_tier=ModelTier.B): clona genome con tier B per la call mutator, costruisce system+user prompt con istruzione scelta random, estrae prompt da tag ..., valida - is_valid_prompt(): 3 check (lunghezza >= 50, keyword tecnica, diff > 5% Levenshtein-like via difflib.SequenceMatcher) - Fallback random_mutate su qualsiasi validation fail O LLM exception Esteso src/multi_swarm/genome/mutation.py con weighted_random_mutate dispatcher: con probabilità prompt_mutation_weight invoca mutate_prompt_llm, altrimenti random_mutate. Backward compat: llm=None oppure weight=0 → solo scalare. Integrazione GA loop + RunConfig: - GAConfig.prompt_mutation_weight: float = 0.0 (default off) - next_generation(llm=...) propagato all'invocazione mutator - RunConfig.prompt_mutation_weight con stesso default - run_phase1: passa llm a next_generation solo se weight > 0 - scripts/run_phase1.py: flag CLI --prompt-mutation-weight Tests (+18, 175 totale): - tests/unit/test_mutation_prompt_llm.py (12): extract_prompt, is_valid_prompt 3 check, operator success + fallback su 3 modi (invalid/identical/exception), tier B per LLM call, istruzione scelta dal pool - tests/unit/test_mutation_dispatcher.py (4): weight 0/1/None + distribuzione 30/70 su 1000 estrazioni con tolleranza ±5% - tests/integration/test_ga_loop_with_prompt_mutator.py (2): loop con weight=1.0 produce prompt evoluti; backward compat weight=0 invariato Co-Authored-By: Claude Opus 4.7 (1M context) --- scripts/run_phase1.py | 7 + src/multi_swarm/ga/loop.py | 16 +- src/multi_swarm/genome/mutation.py | 20 ++ src/multi_swarm/genome/mutation_prompt_llm.py | 167 ++++++++++++++++ src/multi_swarm/orchestrator/run.py | 7 +- .../test_ga_loop_with_prompt_mutator.py | 104 ++++++++++ tests/unit/test_mutation_dispatcher.py | 102 ++++++++++ tests/unit/test_mutation_prompt_llm.py | 178 ++++++++++++++++++ 8 files changed, 597 insertions(+), 4 deletions(-) create mode 100644 src/multi_swarm/genome/mutation_prompt_llm.py create mode 100644 tests/integration/test_ga_loop_with_prompt_mutator.py create mode 100644 tests/unit/test_mutation_dispatcher.py create mode 100644 tests/unit/test_mutation_prompt_llm.py diff --git a/scripts/run_phase1.py b/scripts/run_phase1.py index daabea1..b709c82 100644 --- a/scripts/run_phase1.py +++ b/scripts/run_phase1.py @@ -31,6 +31,12 @@ def parse_args() -> argparse.Namespace: p.add_argument("--end", default="2026-01-01T00:00:00+00:00") p.add_argument("--fees-bp", type=float, default=5.0) p.add_argument("--n-trials-dsr", type=int, default=50) + p.add_argument( + "--prompt-mutation-weight", + type=float, + default=0.0, + help="Phase 2.5: probabilità (0-1) che la mutazione invochi LLM mutator tier B", + ) return p.parse_args() @@ -84,6 +90,7 @@ def main() -> None: fees_bp=args.fees_bp, n_trials_dsr=args.n_trials_dsr, db_path=settings.db_path, + prompt_mutation_weight=args.prompt_mutation_weight, ) run_id = run_phase1(cfg, ohlcv=ohlcv, llm=llm) diff --git a/src/multi_swarm/ga/loop.py b/src/multi_swarm/ga/loop.py index cfbbbd9..9397783 100644 --- a/src/multi_swarm/ga/loop.py +++ b/src/multi_swarm/ga/loop.py @@ -2,10 +2,11 @@ from __future__ import annotations import random from dataclasses import dataclass +from typing import Any from ..genome.crossover import uniform_crossover from ..genome.hypothesis import HypothesisAgentGenome -from ..genome.mutation import random_mutate +from ..genome.mutation import weighted_random_mutate from .selection import elite_select, tournament_select @@ -15,6 +16,7 @@ class GAConfig: elite_k: int tournament_k: int p_crossover: float + prompt_mutation_weight: float = 0.0 # Phase 2.5: opt-in LLM mutator def next_generation( @@ -22,8 +24,14 @@ def next_generation( fitnesses: dict[str, float], cfg: GAConfig, rng: random.Random, + llm: Any | None = None, ) -> list[HypothesisAgentGenome]: - """Costruisce la prossima generazione: elitismo + tournament + crossover/mutate.""" + """Costruisce la prossima generazione: elitismo + tournament + crossover/mutate. + + Quando ``cfg.prompt_mutation_weight > 0`` e ``llm`` è fornito, la mutazione + invoca ``weighted_random_mutate`` che con quella probabilità usa + ``mutate_prompt_llm`` (Phase 2.5). + """ new_pop: list[HypothesisAgentGenome] = list( elite_select(population, fitnesses, cfg.elite_k) ) @@ -35,7 +43,9 @@ def next_generation( child = uniform_crossover(p1, p2, rng) else: parent = tournament_select(population, fitnesses, cfg.tournament_k, rng) - child = random_mutate(parent, rng) + child = weighted_random_mutate( + parent, rng, llm=llm, prompt_mutation_weight=cfg.prompt_mutation_weight + ) new_pop.append(child) return new_pop[: cfg.population_size] diff --git a/src/multi_swarm/genome/mutation.py b/src/multi_swarm/genome/mutation.py index 4699f94..cef606b 100644 --- a/src/multi_swarm/genome/mutation.py +++ b/src/multi_swarm/genome/mutation.py @@ -75,3 +75,23 @@ MUTATION_OPS = ( def random_mutate(g: HypothesisAgentGenome, rng: random.Random) -> HypothesisAgentGenome: op = rng.choice(MUTATION_OPS) return op(g, rng) + + +def weighted_random_mutate( + g: HypothesisAgentGenome, + rng: random.Random, + llm: Any | None = None, + prompt_mutation_weight: float = 0.0, +) -> HypothesisAgentGenome: + """Dispatcher pesato fra mutate_prompt_llm e random_mutate scalare. + + Con probabilità ``prompt_mutation_weight`` invoca ``mutate_prompt_llm``, + altrimenti ``random_mutate``. Se ``llm`` è ``None`` o il peso è 0, + è equivalente a ``random_mutate`` (backward-compat). + """ + if llm is not None and prompt_mutation_weight > 0 and rng.random() < prompt_mutation_weight: + # Import inline per evitare ciclo: mutation_prompt_llm importa da mutation. + from .mutation_prompt_llm import mutate_prompt_llm + + return mutate_prompt_llm(g, llm, rng) + return random_mutate(g, rng) diff --git a/src/multi_swarm/genome/mutation_prompt_llm.py b/src/multi_swarm/genome/mutation_prompt_llm.py new file mode 100644 index 0000000..20213fc --- /dev/null +++ b/src/multi_swarm/genome/mutation_prompt_llm.py @@ -0,0 +1,167 @@ +"""Phase 2.5 operator: ``mutate_prompt_llm``. + +Quinto operatore di mutazione che riscrive il ``system_prompt`` di un genoma +usando un LLM tier B come "mutator". Genera diversità prompt-level dove gli +altri quattro operatori toccano solo i quattro parametri scalari. + +Fallback sicuro: se la mutazione LLM produce output invalido o troppo simile +al parent, l'operatore degrada silenziosamente a ``random_mutate``. +""" + +from __future__ import annotations + +import random +import re +from dataclasses import replace +from difflib import SequenceMatcher +from typing import Any, Protocol + +from .hypothesis import HypothesisAgentGenome, ModelTier +from .mutation import _clone_with, random_mutate + +# Sei tipi di mutazione "atomiche", scelti uniformemente. +MUTATION_INSTRUCTIONS: dict[str, str] = { + "tighten_threshold": ( + "Rendi UNA soglia numerica nella strategia più restrittiva del 10-20%. " + "Esempio: se RSI > 70 diventa RSI > 78. Lascia tutto il resto identico." + ), + "swap_comparator": ( + "Inverti UN comparator (gt -> lt, gte -> lte o viceversa) in una sola " + "condizione. Mantieni lo stesso intent generale della strategia." + ), + "add_condition": ( + "Aggiungi UNA condizione AND a una rule esistente per renderla più " + "selettiva. La condizione deve usare una feature/indicator coerente con " + "il resto della strategia." + ), + "remove_condition": ( + "Rimuovi UNA condizione ridondante o ovvia da una rule, semplificando la " + "logica senza alterarne l'intent principale." + ), + "change_timeframe": ( + "Modifica UNA finestra rolling/lookback di +/- 20-40% (es. SMA(50) -> " + "SMA(70)). Solo un parametro temporale." + ), + "add_temporal_gate": ( + "Aggiungi UN gate temporale alla strategia usando una delle feature " + "'hour', 'dow', 'is_weekend', 'minute_of_hour' per filtrare il " + "trading a specifici momenti." + ), +} + +# Keyword tecniche minime per validare che il prompt sia ancora "una strategia". +_VALID_KEYWORDS = ( + "rsi", "sma", "ema", "atr", "momentum", "breakout", "mean", "reversion", + "macd", "vwap", "bb", "bollinger", "stoch", "trend", "signal", "buy", + "sell", "long", "short", "entry", "exit", "stop", "rule", "condition", + "if", "when", "and", "or", "gt", "lt", ">", "<", "ge", "le", + "hour", "dow", "weekend", "indicator", "feature", +) + +_MIN_PROMPT_LENGTH = 50 +_MIN_DIFF_RATIO = 0.05 # Levenshtein-like: prompt deve essere almeno 5% diverso + +_MUTATOR_SYSTEM_PROMPT = ( + "Sei un mutator evolutivo per prompt di strategie di trading algoritmico. " + "Ricevi un PROMPT originale e una ISTRUZIONE di mutazione atomica. " + "Produci una versione modificata del prompt che applica SOLO quella " + "mutazione, preservando intent e struttura generale. " + "Output: solo il nuovo prompt fra tag .... " + "Nessun preambolo, nessuna spiegazione." +) + +_PROMPT_RE = re.compile(r"\s*(.*?)\s*", re.DOTALL | re.IGNORECASE) + + +class _LLMClientLike(Protocol): + """Subset minimo dell'API LLMClient che usa l'operatore. + + Permette di mockare l'LLM nei test senza importare la classe concreta. + """ + + def complete( + self, + genome: HypothesisAgentGenome, + system: str, + user: str, + max_tokens: int = ..., + ) -> Any: ... + + +def _extract_prompt(text: str) -> str: + """Estrae il prompt mutato dal completion text. + + Cerca tag ``...``. Se assenti, ritorna il testo strip. + """ + m = _PROMPT_RE.search(text) + if m: + return m.group(1).strip() + return text.strip() + + +def _string_diff_ratio(a: str, b: str) -> float: + """Ritorna ``1 - similarity`` (0.0 = identici, 1.0 = completamente diversi).""" + if not a and not b: + return 0.0 + return 1.0 - SequenceMatcher(None, a, b).ratio() + + +def is_valid_prompt(new_prompt: str, parent_prompt: str) -> bool: + """Validation gate per il prompt LLM-mutato. + + Tre check: + 1. Lunghezza minima 50 caratteri. + 2. Contiene almeno una keyword tecnica (rsi, sma, signal, ecc). + 3. Diversità Levenshtein-like > 5% rispetto al parent. + """ + if len(new_prompt) < _MIN_PROMPT_LENGTH: + return False + lowered = new_prompt.lower() + if not any(kw in lowered for kw in _VALID_KEYWORDS): + return False + if _string_diff_ratio(new_prompt, parent_prompt) < _MIN_DIFF_RATIO: + return False + return True + + +def mutate_prompt_llm( + g: HypothesisAgentGenome, + llm: _LLMClientLike, + rng: random.Random, + mutator_tier: ModelTier = ModelTier.B, + max_tokens: int = 2000, +) -> HypothesisAgentGenome: + """Operatore di mutazione prompt-level via LLM mutator. + + Sceglie una mutation-instruction casuale fra sei tipi, fa una chiamata + LLM tier B per ottenere il prompt mutato, valida l'output. Su validation + fail (output troppo corto, non-strategia, troppo simile al parent), + fallback silenzioso a ``random_mutate``. + """ + instruction_key = rng.choice(list(MUTATION_INSTRUCTIONS)) + instruction = MUTATION_INSTRUCTIONS[instruction_key] + + user_prompt = ( + f"PROMPT ORIGINALE:\n{g.system_prompt}\n\n" + f"ISTRUZIONE DI MUTAZIONE ({instruction_key}):\n{instruction}\n\n" + f"Genera la versione modificata fra tag ...." + ) + + # Mutator usa un tier diverso (B) — clone temporaneo del genoma con tier override. + mutator_genome = replace(g, model_tier=mutator_tier) + + try: + result = llm.complete( + mutator_genome, + system=_MUTATOR_SYSTEM_PROMPT, + user=user_prompt, + max_tokens=max_tokens, + ) + except Exception: + return random_mutate(g, rng) + + new_prompt = _extract_prompt(getattr(result, "text", "")) + if not is_valid_prompt(new_prompt, g.system_prompt): + return random_mutate(g, rng) + + return _clone_with(g, system_prompt=new_prompt) diff --git a/src/multi_swarm/orchestrator/run.py b/src/multi_swarm/orchestrator/run.py index ee23e67..a915c94 100644 --- a/src/multi_swarm/orchestrator/run.py +++ b/src/multi_swarm/orchestrator/run.py @@ -49,6 +49,7 @@ class RunConfig: fees_bp: float = 5.0 n_trials_dsr: int = 50 db_path: Path = field(default_factory=lambda: Path("./runs.db")) + prompt_mutation_weight: float = 0.0 # Phase 2.5: opt-in LLM mutator def run_phase1( @@ -90,6 +91,7 @@ def run_phase1( elite_k=cfg.elite_k, tournament_k=cfg.tournament_k, p_crossover=cfg.p_crossover, + prompt_mutation_weight=cfg.prompt_mutation_weight, ) try: @@ -173,7 +175,10 @@ def run_phase1( ) if gen < cfg.n_generations - 1: - population = next_generation(population, fitnesses, ga_cfg, rng) + population = next_generation( + population, fitnesses, ga_cfg, rng, + llm=llm if cfg.prompt_mutation_weight > 0 else None, + ) repo.complete_run( run_id, total_cost=repo.total_cost(run_id), status="completed" diff --git a/tests/integration/test_ga_loop_with_prompt_mutator.py b/tests/integration/test_ga_loop_with_prompt_mutator.py new file mode 100644 index 0000000..d1fc077 --- /dev/null +++ b/tests/integration/test_ga_loop_with_prompt_mutator.py @@ -0,0 +1,104 @@ +"""Integration test Phase 2.5: GA loop con LLM mutator attivo. + +Verifica che ``next_generation`` con ``prompt_mutation_weight > 0`` e ``llm`` +fornito produca figli con system_prompt mutato dall'LLM (e non solo scalari). +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass + +from multi_swarm.ga.loop import GAConfig, next_generation +from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier + +_PROMPT_TEMPLATES = ( + "Strategia mean-reversion 1h. Entry long RSI(14) < 30 e close > SMA(50). Stop 2%.", + "Strategia momentum breakout. Entry long close > SMA(20) e ATR(14) crescente.", + "Strategia trend-following 4h. Long SMA(20) > SMA(50). Short opposito.", +) + + +def _make_pop(n: int) -> list[HypothesisAgentGenome]: + return [ + HypothesisAgentGenome( + system_prompt=_PROMPT_TEMPLATES[i % len(_PROMPT_TEMPLATES)], + feature_access=["close", "high"], + temperature=0.9 + 0.01 * i, + top_p=0.95, + model_tier=ModelTier.C, + lookback_window=200, + cognitive_style="physicist", + ) + for i in range(n) + ] + + +@dataclass +class _Result: + text: str + + +class _MutatorLLM: + """Mock che produce un prompt diverso (e valido) a ogni call.""" + + def __init__(self) -> None: + self.calls = 0 + + def complete(self, genome, system, user, max_tokens: int = 2000) -> _Result: + self.calls += 1 + # Prompt sempre diverso per garantire validation pass. + return _Result( + text=( + f"Strategia evolved #{self.calls}. Entry long quando " + f"RSI(14) < {25 + self.calls % 10} e close > SMA({40 + self.calls}). " + f"Exit short quando momentum decade. Trade rule {self.calls}." + ) + ) + + +def test_loop_with_prompt_mutator_produces_prompt_diversity() -> None: + """Con weight 1.0 + crossover 0 (solo mutation), tutti i child non-elite + devono avere system_prompt diverso dai parent (LLM-mutated).""" + rng = random.Random(0) + pop = _make_pop(5) + fitnesses = {g.id: 0.0 for g in pop} + cfg = GAConfig( + population_size=5, + elite_k=1, + tournament_k=2, + p_crossover=0.0, # nessun crossover → tutta la non-elite è mutation + prompt_mutation_weight=1.0, + ) + llm = _MutatorLLM() + + new_pop = next_generation(pop, fitnesses, cfg, rng, llm=llm) + + assert len(new_pop) == 5 + # 4 non-elite figli, tutti con prompt evoluti. + parent_prompts = {g.system_prompt for g in pop} + evolved = [g for g in new_pop[1:] if g.system_prompt not in parent_prompts] + assert len(evolved) >= 3, f"Solo {len(evolved)} figli con prompt mutato" + assert llm.calls >= 4 + + +def test_loop_backward_compat_no_llm_no_prompt_mutation() -> None: + """Default weight=0.0 + llm=None → comportamento identico a Phase 2.""" + rng = random.Random(0) + pop = _make_pop(5) + fitnesses = {g.id: float(i) for i, g in enumerate(pop)} + cfg = GAConfig( + population_size=5, + elite_k=1, + tournament_k=2, + p_crossover=0.0, + prompt_mutation_weight=0.0, + ) + + new_pop = next_generation(pop, fitnesses, cfg, rng, llm=None) + + assert len(new_pop) == 5 + # Nessun child con prompt diverso dai parent: solo mutazioni scalari. + parent_prompts = {g.system_prompt for g in pop} + for child in new_pop: + assert child.system_prompt in parent_prompts diff --git a/tests/unit/test_mutation_dispatcher.py b/tests/unit/test_mutation_dispatcher.py new file mode 100644 index 0000000..8adb6cf --- /dev/null +++ b/tests/unit/test_mutation_dispatcher.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import random +from collections import Counter +from dataclasses import dataclass + +from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier +from multi_swarm.genome.mutation import weighted_random_mutate + +_PROMPT = ( + "Strategia mean-reversion 1h BTC. Entry long quando RSI(14) < 30 e " + "close > SMA(50). Exit short quando RSI(14) > 70." +) + + +def _make_genome() -> HypothesisAgentGenome: + return HypothesisAgentGenome( + system_prompt=_PROMPT, + feature_access=["close"], + temperature=0.9, + top_p=0.95, + model_tier=ModelTier.C, + lookback_window=200, + cognitive_style="physicist", + ) + + +@dataclass +class _R: + text: str + + +class _AlwaysMutateLLM: + """Mock LLM che ritorna sempre un prompt mutato valido.""" + + def __init__(self) -> None: + self.calls = 0 + + def complete(self, genome, system, user, max_tokens: int = 2000) -> _R: + self.calls += 1 + return _R( + text=( + "Strategia momentum 1h BTC. Entry long quando close > " + f"SMA(70) e ATR(14) crescente. Exit con stop loss 3% (call #{self.calls})." + ) + ) + + +def test_weighted_dispatcher_zero_weight_never_calls_llm() -> None: + llm = _AlwaysMutateLLM() + rng = random.Random(0) + parent = _make_genome() + + for _ in range(50): + weighted_random_mutate(parent, rng, llm=llm, prompt_mutation_weight=0.0) + + assert llm.calls == 0 + + +def test_weighted_dispatcher_full_weight_always_calls_llm() -> None: + llm = _AlwaysMutateLLM() + rng = random.Random(0) + parent = _make_genome() + + for _ in range(20): + child = weighted_random_mutate( + parent, rng, llm=llm, prompt_mutation_weight=1.0 + ) + assert child.system_prompt != parent.system_prompt + + assert llm.calls == 20 + + +def test_weighted_dispatcher_none_llm_falls_back_to_scalar() -> None: + """Senza llm passato (backward compat) → solo mutazione scalare.""" + rng = random.Random(0) + parent = _make_genome() + + for _ in range(50): + child = weighted_random_mutate(parent, rng, llm=None, prompt_mutation_weight=0.5) + assert child.system_prompt == parent.system_prompt + + +def test_weighted_dispatcher_distribution_30_70() -> None: + """Su 1000 estrazioni con weight=0.3 il prompt mutator deve essere chiamato ~300 volte.""" + llm = _AlwaysMutateLLM() + rng = random.Random(123) + parent = _make_genome() + + counter: Counter[str] = Counter() + for _ in range(1000): + child = weighted_random_mutate( + parent, rng, llm=llm, prompt_mutation_weight=0.3 + ) + if child.system_prompt != parent.system_prompt: + counter["prompt"] += 1 + else: + counter["scalar"] += 1 + + # 30% ± 5% tolerance + assert 250 <= counter["prompt"] <= 350, f"prompt mutations: {counter['prompt']}" + assert 650 <= counter["scalar"] <= 750, f"scalar mutations: {counter['scalar']}" diff --git a/tests/unit/test_mutation_prompt_llm.py b/tests/unit/test_mutation_prompt_llm.py new file mode 100644 index 0000000..bbb5784 --- /dev/null +++ b/tests/unit/test_mutation_prompt_llm.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import random +from dataclasses import dataclass + +from multi_swarm.genome.hypothesis import HypothesisAgentGenome, ModelTier +from multi_swarm.genome.mutation_prompt_llm import ( + MUTATION_INSTRUCTIONS, + _extract_prompt, + is_valid_prompt, + mutate_prompt_llm, +) + +_BASE_PROMPT = ( + "Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 30 e " + "close > SMA(50). Exit short quando RSI(14) > 70. Stop loss 2%." +) + + +def _make_genome(prompt: str = _BASE_PROMPT) -> HypothesisAgentGenome: + return HypothesisAgentGenome( + system_prompt=prompt, + feature_access=["close", "high", "low"], + temperature=0.9, + top_p=0.95, + model_tier=ModelTier.C, + lookback_window=200, + cognitive_style="physicist", + ) + + +@dataclass +class _FakeResult: + text: str + + +class _FakeLLM: + """Mock LLMClient: ritorna una risposta configurata in input.""" + + def __init__(self, response_text: str = "", raise_exc: bool = False) -> None: + self.response_text = response_text + self.raise_exc = raise_exc + self.last_call: dict[str, object] | None = None + + def complete( + self, + genome: HypothesisAgentGenome, + system: str, + user: str, + max_tokens: int = 2000, + ) -> _FakeResult: + self.last_call = { + "genome_tier": genome.model_tier, + "system": system, + "user": user, + "max_tokens": max_tokens, + } + if self.raise_exc: + raise RuntimeError("simulated LLM failure") + return _FakeResult(text=self.response_text) + + +def test_extract_prompt_from_tag() -> None: + raw = "preambolo blah\nStrategia RSI > 75 short, SMA(60) trend.\nblabla" + assert _extract_prompt(raw) == "Strategia RSI > 75 short, SMA(60) trend." + + +def test_extract_prompt_no_tag_returns_stripped_text() -> None: + raw = " Strategia momentum breakout su 1h con ATR(14) " + assert _extract_prompt(raw) == "Strategia momentum breakout su 1h con ATR(14)" + + +def test_is_valid_prompt_accepts_proper_strategy() -> None: + new = ( + "Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 25 e " + "close > SMA(50). Exit short quando RSI(14) > 75." + ) + assert is_valid_prompt(new, _BASE_PROMPT) is True + + +def test_is_valid_prompt_rejects_too_short() -> None: + assert is_valid_prompt("short", _BASE_PROMPT) is False + + +def test_is_valid_prompt_rejects_no_strategy_keywords() -> None: + bad = "Questo è un testo a caso che parla del meteo di domani e della pioggia." + assert is_valid_prompt(bad, _BASE_PROMPT) is False + + +def test_is_valid_prompt_rejects_identical_prompt() -> None: + assert is_valid_prompt(_BASE_PROMPT, _BASE_PROMPT) is False + + +def test_mutate_prompt_llm_produces_mutated_child() -> None: + mutated = ( + "Strategia mean-reversion 1h su BTC. Entry long quando RSI(14) < 25 e " + "close > SMA(70). Exit short quando RSI(14) > 78." + ) + llm = _FakeLLM(response_text=f"{mutated}") + parent = _make_genome() + rng = random.Random(0) + + child = mutate_prompt_llm(parent, llm, rng) + + assert child.system_prompt == mutated + assert child.id != parent.id + assert child.parent_ids == [*parent.parent_ids, parent.id] + assert child.generation == parent.generation + 1 + assert child.model_tier == ModelTier.C # tier C preservato sul child + + +def test_mutate_prompt_llm_uses_mutator_tier_b_for_llm_call() -> None: + mutated = ( + "Strategia momentum breakout 1h. Entry long quando close > SMA(60) e " + "ATR(14) crescente. Exit con stop loss 3%." + ) + llm = _FakeLLM(response_text=f"{mutated}") + parent = _make_genome() + rng = random.Random(0) + + mutate_prompt_llm(parent, llm, rng) + + assert llm.last_call is not None + assert llm.last_call["genome_tier"] == ModelTier.B + + +def test_mutate_prompt_llm_falls_back_on_invalid_output() -> None: + """Output troppo corto -> fallback random_mutate (cambia uno scalare).""" + llm = _FakeLLM(response_text="nope") + parent = _make_genome() + rng = random.Random(42) + + child = mutate_prompt_llm(parent, llm, rng) + + # random_mutate preserva system_prompt, cambia uno dei 4 scalari/style. + assert child.system_prompt == parent.system_prompt + assert child.parent_ids == [*parent.parent_ids, parent.id] + assert child.generation == parent.generation + 1 + + +def test_mutate_prompt_llm_falls_back_on_identical_output() -> None: + llm = _FakeLLM(response_text=f"{_BASE_PROMPT}") + parent = _make_genome() + rng = random.Random(42) + + child = mutate_prompt_llm(parent, llm, rng) + + assert child.system_prompt == parent.system_prompt + assert child.generation == parent.generation + 1 + + +def test_mutate_prompt_llm_falls_back_on_llm_exception() -> None: + llm = _FakeLLM(raise_exc=True) + parent = _make_genome() + rng = random.Random(7) + + child = mutate_prompt_llm(parent, llm, rng) + + # Fallback random_mutate sempre produce un child valido. + assert child.system_prompt == parent.system_prompt + assert child.generation == parent.generation + 1 + + +def test_mutate_prompt_llm_picks_one_of_six_instructions() -> None: + """Verifica che il system message dell'LLM includa una delle 6 istruzioni.""" + mutated = ( + "Strategia RSI 1h. Entry long quando RSI(14) < 28 e close > SMA(50). " + "Exit short quando RSI(14) > 72." + ) + llm = _FakeLLM(response_text=f"{mutated}") + parent = _make_genome() + + mutate_prompt_llm(parent, llm, random.Random(0)) + + assert llm.last_call is not None + user_text = str(llm.last_call["user"]) + matched_keys = [k for k in MUTATION_INSTRUCTIONS if k in user_text] + assert len(matched_keys) >= 1, f"User prompt non contiene istruzione: {user_text[:200]}"