feat(agents): 5 metriche geometrico-frattali + style-aware focus block

Aggiunge al MarketSummary 5 metriche regime-aware:
  - efficiency_ratio (Kaufman): discriminatore trending/ranging
  - tail_index_left/right (Hill): pesantezza code, robust vs kurtosis
  - structural_uptrend (HH/HL Dow-style): trend strutturale senza lag MA
  - compression_ratio: vol coil pre-breakout
  - spectral_entropy + dominant_cycle (gated): struttura ciclica nel FFT

Architettura "Style-aware focus, no filtering":
  - Tutte le 5 metriche restano visibili a tutti gli stili (landscape GA smooth)
  - prompts.json v2.2: ogni stile dichiara focus_metrics: list[str]
  - USER_TEMPLATE renderizza "Focus per la tua lente: ..." con i valori prioritari
  - Mutation cognitive_style preserva accesso a tutte le metriche (no discontinuita)

PromptLibrary esteso con focus field (parsato da JSON entry styles).
HypothesisAgent accetta prompt_library nel costruttore; orchestrator lo passa.

7 directive aggiornate per interpretare i 5 nuovi input attraverso la lente:
  - physicist: efficiency_ratio + dominant_cycle (modi armonici)
  - engineer: efficiency_ratio < 0.2 = no signal
  - psychologist: tail_left/right = paura/euforia ricorrente
  - ecc.

Test: +19 unit (metriche + focus rendering), +smoke MarketSummary.
Tot: 216 test pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-05-15 21:04:06 +00:00
parent 0fd31d52ec
commit b6f48e46fc
9 changed files with 630 additions and 14 deletions
@@ -2,10 +2,12 @@ from __future__ import annotations
import re import re
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any
import openai import openai
from ..genome.hypothesis import HypothesisAgentGenome 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
from ..protocol.parser import ParseError, Strategy, parse_strategy from ..protocol.parser import ParseError, Strategy, parse_strategy
from ..protocol.validator import ValidationError, validate_strategy from ..protocol.validator import ValidationError, validate_strategy
@@ -27,6 +29,13 @@ class MarketSummary:
vol_percentile: float = 50.0 # 0-100 percentile della vol corrente vol_percentile: float = 50.0 # 0-100 percentile della vol corrente
seasonality_hour: float = 0.0 # 0-1 varianza spiegata da hour seasonality_hour: float = 0.0 # 0-1 varianza spiegata da hour
seasonality_dow: float = 0.0 # 0-1 varianza spiegata da dow seasonality_dow: float = 0.0 # 0-1 varianza spiegata da dow
efficiency_ratio: float = 0.0 # Kaufman, 0-1
tail_index_left: float = 5.0 # Hill left tail
tail_index_right: float = 5.0 # Hill right tail
structural_uptrend: float = 0.5 # HH/HL score 0-1
compression: float = 1.0 # range recent / range ref
spectral_entropy: float = 1.0 # 0-1, Shannon FFT normalizzata
dominant_cycle: float | None = None # periodo barre, None se spectrum piatto
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -193,6 +202,14 @@ Regime recente (ultime 500 barre):
vol_pct: {vol_percentile:.0f}° percentile storico vol_pct: {vol_percentile:.0f}° percentile storico
stagionalita: hour={seasonality_hour:.2f}, dow={seasonality_dow:.2f} (0-1, varianza spiegata) stagionalita: hour={seasonality_hour:.2f}, dow={seasonality_dow:.2f} (0-1, varianza spiegata)
Geometria & frattali:
efficiency_ratio: {efficiency_ratio:.3f} (Kaufman, 0=noise, 1=trend efficiente)
tail_index: left={tail_index_left:.2f}, right={tail_index_right:.2f} (Hill; <2 fat tail, >5 light)
structural_uptrend: {structural_uptrend:.2f} (HH/HL score, 0.5=range)
compression: {compression:.2f} (range recent / ref; <1 compressione, >1 espansione)
spectral_entropy: {spectral_entropy:.2f} (0=struttura, 1=rumore bianco)
dominant_cycle: {dominant_cycle_str} (None se spettro piatto)
Feature accessibili dal tuo genoma: {feature_access}. Feature accessibili dal tuo genoma: {feature_access}.
Lookback massimo che puoi usare nel ragionamento: {lookback_window} barre. Lookback massimo che puoi usare nel ragionamento: {lookback_window} barre.
@@ -200,6 +217,51 @@ Genera una strategia che cerchi anomalie sfruttabili in questo regime.
""" """
def _render_focus_block(keys: list[str], market: MarketSummary) -> str:
"""Renderizza 'Focus per la tua lente' come riga del USER_TEMPLATE.
Mappa nomi simbolici a valori della MarketSummary. Skippa silenziosamente
chiavi sconosciute (fault-tolerant per evoluzione futura).
"""
field_map: dict[str, Any] = {
# Statistiche base
"mean": market.return_mean,
"std": market.return_std,
"skew": market.skew,
"kurt": market.kurtosis,
"vol_regime": market.volatility_regime,
# Regime recente
"autocorr_recent": market.autocorr_lag1_recent,
"autocorr_baseline": market.autocorr_lag1_baseline,
"hurst": market.hurst_recent,
"vol_pct": market.vol_percentile,
"seasonality_hour": market.seasonality_hour,
"seasonality_dow": market.seasonality_dow,
# Geometria/frattali
"efficiency_ratio": market.efficiency_ratio,
"tail_left": market.tail_index_left,
"tail_right": market.tail_index_right,
"structural_uptrend": market.structural_uptrend,
"compression": market.compression,
"spectral_entropy": market.spectral_entropy,
"dominant_cycle": market.dominant_cycle,
}
parts: list[str] = []
for k in keys:
if k not in field_map:
continue
v = field_map[k]
if v is None:
parts.append(f"{k}=N/A")
elif isinstance(v, float):
parts.append(f"{k}={v:.3f}")
else:
parts.append(f"{k}={v}")
if not parts:
return ""
return "\nFocus per la tua lente: " + ", ".join(parts) + "\n"
_RETRY_TEMPLATE = """\ _RETRY_TEMPLATE = """\
{original_user} {original_user}
@@ -282,11 +344,17 @@ def _try_parse(text: str) -> tuple[Strategy | None, str | None]:
class HypothesisAgent: class HypothesisAgent:
def __init__(self, llm: LLMClient, max_retries: int = 1): def __init__(
self,
llm: LLMClient,
max_retries: int = 1,
prompt_library: PromptLibrary | None = None,
):
if max_retries < 0: if max_retries < 0:
raise ValueError("max_retries must be >= 0") raise ValueError("max_retries must be >= 0")
self._llm = llm self._llm = llm
self._max_retries = max_retries self._max_retries = max_retries
self._prompt_library = prompt_library or PromptLibrary.default()
def propose( def propose(
self, self,
@@ -297,6 +365,13 @@ class HypothesisAgent:
cognitive_style=genome.cognitive_style, cognitive_style=genome.cognitive_style,
system_prompt=genome.system_prompt, system_prompt=genome.system_prompt,
) )
dominant_cycle_str = (
f"{market.dominant_cycle:.0f} barre"
if market.dominant_cycle is not None
else "N/A (spettro piatto)"
)
focus_keys = self._prompt_library.focus_metrics_for(genome.cognitive_style)
focus_block = _render_focus_block(focus_keys, market) if focus_keys else ""
original_user = USER_TEMPLATE.format( original_user = USER_TEMPLATE.format(
symbol=market.symbol, symbol=market.symbol,
timeframe=market.timeframe, timeframe=market.timeframe,
@@ -312,9 +387,16 @@ class HypothesisAgent:
vol_percentile=market.vol_percentile, vol_percentile=market.vol_percentile,
seasonality_hour=market.seasonality_hour, seasonality_hour=market.seasonality_hour,
seasonality_dow=market.seasonality_dow, seasonality_dow=market.seasonality_dow,
efficiency_ratio=market.efficiency_ratio,
tail_index_left=market.tail_index_left,
tail_index_right=market.tail_index_right,
structural_uptrend=market.structural_uptrend,
compression=market.compression,
spectral_entropy=market.spectral_entropy,
dominant_cycle_str=dominant_cycle_str,
feature_access=", ".join(genome.feature_access), feature_access=", ".join(genome.feature_access),
lookback_window=genome.lookback_window, lookback_window=genome.lookback_window,
) ) + focus_block
completions: list[CompletionResult] = [] completions: list[CompletionResult] = []
errors: list[str] = [] errors: list[str] = []
@@ -5,8 +5,13 @@ from scipy import stats # type: ignore[import-untyped]
from ..metrics.basic import ( from ..metrics.basic import (
autocorr_lag1, autocorr_lag1,
compression_ratio,
efficiency_ratio_kaufman,
hurst_exponent, hurst_exponent,
seasonality_strength, seasonality_strength,
spectral_entropy_and_cycle,
structural_uptrend_score,
tail_index_hill,
vol_percentile_historical, vol_percentile_historical,
) )
from .hypothesis import MarketSummary from .hypothesis import MarketSummary
@@ -40,6 +45,13 @@ def build_market_summary(
season_h = seasonality_strength(returns, by="hour") season_h = seasonality_strength(returns, by="hour")
season_d = seasonality_strength(returns, by="dow") season_d = seasonality_strength(returns, by="dow")
eff_ratio = efficiency_ratio_kaufman(ohlcv["close"], length=100)
tail_l = tail_index_hill(returns, side="left")
tail_r = tail_index_hill(returns, side="right")
hh_hl = structural_uptrend_score(ohlcv["close"], window=5)
compress = compression_ratio(ohlcv["close"], recent_window=50, ref_window=200)
spec_entropy, dom_cycle = spectral_entropy_and_cycle(returns, length=256)
return MarketSummary( return MarketSummary(
symbol=symbol, symbol=symbol,
timeframe=timeframe, timeframe=timeframe,
@@ -55,4 +67,11 @@ def build_market_summary(
vol_percentile=vol_pct, vol_percentile=vol_pct,
seasonality_hour=season_h, seasonality_hour=season_h,
seasonality_dow=season_d, seasonality_dow=season_d,
efficiency_ratio=eff_ratio,
tail_index_left=tail_l,
tail_index_right=tail_r,
structural_uptrend=hh_hl,
compression=compress,
spectral_entropy=spec_entropy,
dominant_cycle=dom_cycle,
) )
@@ -0,0 +1,140 @@
"""Libreria di stili cognitivi + direttive system_prompt per il GA.
Carica da un file JSON esterno (tipicamente shippato dal singolo strategy
member, es. ``strategy_crypto/prompts.json``) le coppie ``style -> directive``
usate da:
- :func:`multi_swarm_core.ga.initial.build_initial_population` per il
bootstrap della popolazione (style assegnato round-robin, directive
come system_prompt iniziale).
- :func:`multi_swarm_core.genome.mutation.mutate_cognitive_style` per
pescare i candidati di mutazione (range = key del JSON).
Schema JSON atteso::
{
"styles": {
"<name>": {"directive": "<testo system_prompt>"},
...
}
}
I 6 stili default (physicist, biologist, historian, meteorologist,
ecologist, engineer) sono comunque disponibili via :meth:`PromptLibrary.default`
per backcompat con test/script senza file esterno.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
_DEFAULT_STYLES: dict[str, str] = {
"physicist": (
"Cerca leggi conservative, simmetrie, regimi di scala. "
"Pensa in termini di flussi e potenziali."
),
"biologist": (
"Cerca pattern adattivi, nicchie ecologiche, "
"predator-prey dynamics tra partecipanti del mercato."
),
"historian": (
"Cerca pattern ricorrenti su scale temporali multiple, "
"analogie con regimi storici, mean reversion strutturali."
),
"meteorologist": (
"Cerca regimi di volatilità che si autoalimentano, "
"transizioni di stato come fronti, persistenza locale."
),
"ecologist": (
"Cerca interazioni multi-asset, correlazioni cluster, "
"segnali di stress sistemico nelle dinamiche di flusso."
),
"engineer": (
"Cerca segnali con rapporto S/N favorevole, filtri causali, "
"robustezza a perturbazioni di calibrazione."
),
}
class PromptLibraryError(ValueError):
"""Sollevata su JSON malformato o stili invalid."""
@dataclass(frozen=True)
class PromptLibrary:
"""Set immutabile di stili cognitivi + direttive system_prompt."""
styles: dict[str, str]
focus: dict[str, list[str]] # style -> lista metriche prioritarie
def __post_init__(self) -> None:
if not self.styles:
raise PromptLibraryError("PromptLibrary deve avere almeno uno stile")
for name, directive in self.styles.items():
if not isinstance(name, str) or not name.strip():
raise PromptLibraryError(f"nome stile invalido: {name!r}")
if not isinstance(directive, str) or not directive.strip():
raise PromptLibraryError(
f"directive vuota o invalida per stile {name!r}"
)
@classmethod
def default(cls) -> PromptLibrary:
"""Libreria builtin con i 6 stili originali (fallback senza file)."""
return cls(styles=dict(_DEFAULT_STYLES), focus={})
@classmethod
def from_json(cls, path: Path | str) -> PromptLibrary:
"""Carica da file JSON con schema ``{"styles": {<name>: {"directive": "...", "focus_metrics": [...]}}}``."""
p = Path(path)
try:
data = json.loads(p.read_text(encoding="utf-8"))
except FileNotFoundError as e:
raise PromptLibraryError(f"file non trovato: {p}") from e
except json.JSONDecodeError as e:
raise PromptLibraryError(f"JSON malformato in {p}: {e}") from e
if not isinstance(data, dict):
raise PromptLibraryError(f"root JSON deve essere un object, non {type(data)}")
styles_raw = data.get("styles")
if not isinstance(styles_raw, dict):
raise PromptLibraryError("manca chiave 'styles' (object) nel JSON")
styles: dict[str, str] = {}
focus: dict[str, list[str]] = {}
for name, entry in styles_raw.items():
if not isinstance(entry, dict):
raise PromptLibraryError(
f"entry per stile {name!r} deve essere object, non {type(entry)}"
)
directive = entry.get("directive")
if not isinstance(directive, str):
raise PromptLibraryError(
f"manca 'directive' (string) per stile {name!r}"
)
styles[name] = directive
fm = entry.get("focus_metrics", [])
if not isinstance(fm, list):
raise PromptLibraryError(
f"focus_metrics di {name!r} deve essere lista, non {type(fm)}"
)
focus[name] = [str(x) for x in fm]
return cls(styles=styles, focus=focus)
@property
def cognitive_styles(self) -> tuple[str, ...]:
"""Tupla immutabile dei nomi degli stili, nell'ordine di insertion del JSON."""
return tuple(self.styles)
def directive(self, style: str) -> str:
"""Ritorna la directive di ``style`` o solleva KeyError."""
return self.styles[style]
def style_at(self, index: int) -> str:
"""Round-robin: ``cognitive_styles[index % len()]``."""
return self.cognitive_styles[index % len(self.cognitive_styles)]
def focus_metrics_for(self, style: str) -> list[str]:
"""Ritorna lista delle metriche prioritarie per ``style``. Empty list se non specificate."""
return self.focus.get(style, [])
@@ -141,3 +141,135 @@ def total_return(equity: pd.Series) -> float:
if equity.iloc[0] == 0: if equity.iloc[0] == 0:
return float(equity.iloc[-1]) return float(equity.iloc[-1])
return float(equity.iloc[-1] / equity.iloc[0] - 1.0) return float(equity.iloc[-1] / equity.iloc[0] - 1.0)
def efficiency_ratio_kaufman(prices: pd.Series, length: int = 100) -> float:
"""Kaufman Efficiency Ratio: |net move| / sum(|step|) su rolling length.
Output 0-1: 0 = puro noise (movimento dissipativo), 1 = puro trend efficiente.
Discriminatore trending/ranging robusto.
"""
if len(prices) < length + 1:
return 0.0
recent = prices.iloc[-length:]
net_move = abs(recent.iloc[-1] - recent.iloc[0])
total_move = recent.diff().abs().sum()
if total_move == 0 or pd.isna(total_move):
return 0.0
return float(net_move / total_move)
def tail_index_hill(returns: pd.Series, side: str, top_frac: float = 0.05) -> float:
"""Hill estimator del tail index (pendenza coda) per side in {'left', 'right'}.
Output: indice di pesantezza coda. Valori piu bassi = coda piu pesante.
<2 = varianza infinita (Cauchy-like), 3-4 = tipico crypto, >5 = quasi-Gaussiana.
Robusto al singolo outlier (vs kurtosis).
"""
r = returns.dropna()
n = len(r)
if n < 50:
return 5.0 # default quasi-Gaussiano
k = max(int(n * top_frac), 5)
if side == "left":
tail = (-r).nlargest(k)
elif side == "right":
tail = r.nlargest(k)
else:
raise ValueError(f"side deve essere 'left' o 'right', non {side!r}")
# Filtra valori non-positivi (Hill richiede tail positiva)
tail = tail[tail > 0]
if len(tail) < 5:
return 5.0
log_tail = np.log(tail.values)
threshold = log_tail[-1]
excess = log_tail[:-1] - threshold
if excess.sum() <= 0:
return 5.0
hill = (len(excess)) / excess.sum()
return float(np.clip(hill, 1.0, 10.0))
def structural_uptrend_score(prices: pd.Series, window: int = 5) -> float:
"""Frazione di periodi in struttura HH/HL (Dow-style uptrend).
Identifica swing high/low usando rolling max/min con finestra ``window``.
Conta sequenze higher-high + higher-low (uptrend) vs lower-high + lower-low (downtrend).
Output: 0 (puro downtrend) - 0.5 (range/mixed) - 1 (puro uptrend).
"""
if len(prices) < 4 * window:
return 0.5
swing_high = prices.rolling(2 * window + 1, center=True).max() == prices
swing_low = prices.rolling(2 * window + 1, center=True).min() == prices
highs = prices[swing_high].dropna()
lows = prices[swing_low].dropna()
if len(highs) < 3 or len(lows) < 3:
return 0.5
hh = (highs.diff() > 0).sum()
lh = (highs.diff() < 0).sum()
hl = (lows.diff() > 0).sum()
ll = (lows.diff() < 0).sum()
up_signals = hh + hl
down_signals = lh + ll
total = up_signals + down_signals
if total == 0:
return 0.5
return float(up_signals / total)
def compression_ratio(prices: pd.Series, recent_window: int = 50, ref_window: int = 200) -> float:
"""Range(recent) / Range(ref). <1 = compressione vol, >1 = espansione.
Range = high - low della finestra. Cattura il "coil" pre-breakout.
"""
if len(prices) < ref_window:
return 1.0
recent = prices.iloc[-recent_window:]
ref = prices.iloc[-ref_window:]
recent_range = float(recent.max() - recent.min())
ref_range = float(ref.max() - ref.min())
if ref_range <= 0:
return 1.0
return recent_range / ref_range
def spectral_entropy_and_cycle(
returns: pd.Series, length: int = 256
) -> tuple[float, float | None]:
"""FFT su returns -> (entropy normalizzata, dominant_cycle gated).
entropy: 0-1, Shannon entropy normalizzata dello spettro di potenza.
0 = una sola frequenza domina, 1 = spettro piatto (rumore bianco).
dominant_cycle: periodo (barre) della frequenza dominante,
MA solo se entropy < 0.6 (struttura presente). Altrimenti None.
"""
r = returns.dropna()
if len(r) < length:
return 1.0, None
series = r.iloc[-length:].values
# Detrend
series = series - series.mean()
fft = np.fft.rfft(series)
power = np.abs(fft) ** 2
if power.sum() <= 0:
return 1.0, None
p = power / power.sum()
# Entropy normalizzata (Shannon / log(N))
nonzero = p[p > 0]
entropy = float(-(nonzero * np.log(nonzero)).sum() / np.log(len(nonzero)))
entropy = max(0.0, min(1.0, entropy))
if entropy >= 0.6:
return entropy, None
# Skip DC component (index 0)
if len(power) <= 1:
return entropy, None
peak_idx = int(np.argmax(power[1:])) + 1
if peak_idx == 0:
return entropy, None
cycle = float(length / peak_idx)
return entropy, cycle
@@ -27,6 +27,8 @@ from ..ga.initial import build_initial_population
from ..ga.loop import GAConfig, next_generation from ..ga.loop import GAConfig, next_generation
from ..ga.summary import generation_summary from ..ga.summary import generation_summary
from ..genome.hypothesis import ModelTier from ..genome.hypothesis import ModelTier
from ..genome.mutation import set_cognitive_styles
from ..genome.prompt_library import PromptLibrary
from ..llm.client import LLMClient from ..llm.client import LLMClient
from ..llm.cost_tracker import CostTracker from ..llm.cost_tracker import CostTracker
from ..persistence.repository import Repository from ..persistence.repository import Repository
@@ -67,6 +69,10 @@ class RunConfig:
# 2x costo backtest engine. # 2x costo backtest engine.
eval_oos_during_loop: bool = False eval_oos_during_loop: bool = False
fitness_combined_alpha: float = 0.5 # peso IS (1-alpha = OOS) fitness_combined_alpha: float = 0.5 # peso IS (1-alpha = OOS)
# Libreria di stili cognitivi + system_prompt iniziali. Se None usa
# i 6 builtin (PromptLibrary.default()). Tipicamente caricata da
# strategy_crypto/prompts.json via PromptLibrary.from_json().
prompt_library: PromptLibrary | None = None
def run_phase1( def run_phase1(
@@ -100,7 +106,7 @@ def run_phase1(
market = build_market_summary(train_ohlcv, symbol=cfg.symbol, timeframe=cfg.timeframe) market = build_market_summary(train_ohlcv, symbol=cfg.symbol, timeframe=cfg.timeframe)
hypothesis_agent = HypothesisAgent(llm=llm) hypothesis_agent = HypothesisAgent(llm=llm, prompt_library=prompt_library)
falsification_agent = FalsificationAgent( falsification_agent = FalsificationAgent(
fees_bp=cfg.fees_bp, n_trials_dsr=cfg.n_trials_dsr fees_bp=cfg.fees_bp, n_trials_dsr=cfg.n_trials_dsr
) )
@@ -112,8 +118,16 @@ def run_phase1(
) )
cost_tracker = CostTracker() cost_tracker = CostTracker()
# Propaga la libreria di stili al modulo mutation (cosi' mutate_cognitive_style
# pesca dai candidati coerenti col JSON della strategia in corso).
prompt_library = cfg.prompt_library or PromptLibrary.default()
set_cognitive_styles(prompt_library.cognitive_styles)
population = build_initial_population( population = build_initial_population(
k=cfg.population_size, model_tier=cfg.model_tier, rng=rng k=cfg.population_size,
model_tier=cfg.model_tier,
rng=rng,
prompt_library=prompt_library,
) )
fitnesses: dict[str, float] = {} fitnesses: dict[str, float] = {}
@@ -2,6 +2,7 @@ import json
from multi_swarm_core.agents.hypothesis import HypothesisAgent, MarketSummary from multi_swarm_core.agents.hypothesis import HypothesisAgent, MarketSummary
from multi_swarm_core.genome.hypothesis import HypothesisAgentGenome, ModelTier from multi_swarm_core.genome.hypothesis import HypothesisAgentGenome, ModelTier
from multi_swarm_core.genome.prompt_library import PromptLibrary
from multi_swarm_core.llm.client import CompletionResult, EmptyCompletionError from multi_swarm_core.llm.client import CompletionResult, EmptyCompletionError
@@ -248,3 +249,53 @@ def test_hypothesis_agent_returns_failed_proposal_on_only_empty_completions(mock
assert "empty_completion" in proposal.parse_error assert "empty_completion" in proposal.parse_error
# 3 tentativi tutti falliti. # 3 tentativi tutti falliti.
assert fake_llm.complete.call_count == 3 assert fake_llm.complete.call_count == 3
def test_propose_renders_focus_block_for_style(mocker): # type: ignore[no-untyped-def]
"""Con PromptLibrary che ha focus_metrics, il LLM mock riceve 'Focus per la tua lente' nel user message."""
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",
)
# PromptLibrary con focus_metrics per physicist
lib = PromptLibrary(
styles={"physicist": "Pensa come un fisico."},
focus={"physicist": ["efficiency_ratio", "spectral_entropy", "hurst"]},
)
agent = HypothesisAgent(llm=fake_llm, prompt_library=lib)
proposal = agent.propose(make_genome(), make_summary())
assert proposal.strategy is not None
# Verifica che il user message contenga il focus block
call_kwargs = fake_llm.complete.call_args.kwargs
user_msg = call_kwargs["user"]
assert "Focus per la tua lente:" in user_msg
assert "efficiency_ratio=" in user_msg
assert "spectral_entropy=" in user_msg
assert "hurst=" in user_msg
def test_propose_no_focus_block_when_style_not_in_library(mocker): # type: ignore[no-untyped-def]
"""Stile senza focus_metrics → nessun focus block nel user message."""
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",
)
# PromptLibrary senza focus_metrics per physicist
lib = PromptLibrary(
styles={"physicist": "Pensa come un fisico."},
focus={},
)
agent = HypothesisAgent(llm=fake_llm, prompt_library=lib)
proposal = agent.propose(make_genome(), make_summary())
assert proposal.strategy is not None
call_kwargs = fake_llm.complete.call_args.kwargs
user_msg = call_kwargs["user"]
assert "Focus per la tua lente:" not in user_msg
@@ -63,3 +63,51 @@ def test_build_summary_new_fields_populated() -> None:
assert isinstance(s.seasonality_dow, float) assert isinstance(s.seasonality_dow, float)
assert 0.0 <= s.seasonality_hour <= 1.0 assert 0.0 <= s.seasonality_hour <= 1.0
assert 0.0 <= s.seasonality_dow <= 1.0 assert 0.0 <= s.seasonality_dow <= 1.0
def test_build_summary_geometric_fields_populated() -> None:
"""I 7 nuovi campi geometrico-frattali devono essere non-default e nei range attesi."""
idx = pd.date_range("2024-01-01", periods=600, freq="1h", tz="UTC")
np.random.seed(5)
close = 100 + np.cumsum(np.random.normal(0, 1, 600))
df = pd.DataFrame(
{"open": close, "high": close + 0.5, "low": close - 0.5, "close": close, "volume": 1.0},
index=idx,
)
s = build_market_summary(df, symbol="BTC/USDT", timeframe="1h")
# efficiency_ratio: float in [0, 1]
assert isinstance(s.efficiency_ratio, float)
assert 0.0 <= s.efficiency_ratio <= 1.0
# tail_index_left/right: float in [1, 10]
assert isinstance(s.tail_index_left, float)
assert isinstance(s.tail_index_right, float)
assert 1.0 <= s.tail_index_left <= 10.0
assert 1.0 <= s.tail_index_right <= 10.0
# structural_uptrend: float in [0, 1]
assert isinstance(s.structural_uptrend, float)
assert 0.0 <= s.structural_uptrend <= 1.0
# compression: float > 0
assert isinstance(s.compression, float)
assert s.compression > 0.0
# spectral_entropy: float in [0, 1]
assert isinstance(s.spectral_entropy, float)
assert 0.0 <= s.spectral_entropy <= 1.0
# dominant_cycle: None or positive float
assert s.dominant_cycle is None or (isinstance(s.dominant_cycle, float) and s.dominant_cycle > 0)
# Verifica che i campi siano stati popolati (non tutti ai valori default)
defaults_only = (
s.efficiency_ratio == 0.0
and s.tail_index_left == 5.0
and s.tail_index_right == 5.0
and s.structural_uptrend == 0.5
and s.compression == 1.0
and s.spectral_entropy == 1.0
)
assert not defaults_only, "Tutti i campi geometrici sono rimasti ai valori default: calcolo non avvenuto"
@@ -4,10 +4,15 @@ import pytest
from multi_swarm_core.metrics.basic import ( from multi_swarm_core.metrics.basic import (
autocorr_lag1, autocorr_lag1,
compression_ratio,
efficiency_ratio_kaufman,
hurst_exponent, hurst_exponent,
max_drawdown, max_drawdown,
seasonality_strength, seasonality_strength,
sharpe_ratio, sharpe_ratio,
spectral_entropy_and_cycle,
structural_uptrend_score,
tail_index_hill,
total_return, total_return,
vol_percentile_historical, vol_percentile_historical,
) )
@@ -166,3 +171,120 @@ def test_seasonality_strength_invalid_by():
r = pd.Series(np.random.normal(0, 0.01, 100), index=idx) r = pd.Series(np.random.normal(0, 0.01, 100), index=idx)
with pytest.raises(ValueError, match="'minute'"): with pytest.raises(ValueError, match="'minute'"):
seasonality_strength(r, by="minute") seasonality_strength(r, by="minute")
# --- Geometric/fractal metrics tests ---
def test_efficiency_ratio_pure_trend():
"""Slope perfetto (prezzi linearmente crescenti) -> efficiency_ratio vicino a 1.0."""
prices = pd.Series(np.linspace(100.0, 200.0, 200))
er = efficiency_ratio_kaufman(prices, length=100)
assert er == pytest.approx(1.0, abs=1e-6)
def test_efficiency_ratio_random_walk():
"""Random walk iid -> efficiency_ratio basso (attorno a 0.1-0.4)."""
np.random.seed(42)
prices = pd.Series(100.0 + np.cumsum(np.random.normal(0, 1, 300)))
er = efficiency_ratio_kaufman(prices, length=100)
assert 0.0 <= er <= 0.6
def test_efficiency_ratio_short_series():
"""Serie troppo corta ritorna 0.0."""
prices = pd.Series([100.0, 101.0, 102.0])
assert efficiency_ratio_kaufman(prices, length=100) == 0.0
def test_tail_index_hill_left_right_separate():
"""tail_index_hill left e right ritornano float in [1, 10]."""
np.random.seed(7)
r = pd.Series(np.random.normal(0, 0.01, 500))
left = tail_index_hill(r, side="left")
right = tail_index_hill(r, side="right")
assert 1.0 <= left <= 10.0
assert 1.0 <= right <= 10.0
def test_tail_index_hill_invalid_side():
"""side non valido solleva ValueError."""
r = pd.Series(np.random.normal(0, 0.01, 200))
with pytest.raises(ValueError, match="'both'"):
tail_index_hill(r, side="both")
def test_tail_index_hill_short_series():
"""Serie < 50 ritorna default 5.0."""
r = pd.Series([0.01, -0.01, 0.02])
assert tail_index_hill(r, side="left") == pytest.approx(5.0)
assert tail_index_hill(r, side="right") == pytest.approx(5.0)
def test_structural_uptrend_pure_uptrend():
"""Prezzi costantemente crescenti -> structural_uptrend vicino a 1.0."""
prices = pd.Series(np.linspace(100.0, 200.0, 100))
score = structural_uptrend_score(prices, window=5)
assert score >= 0.5
def test_structural_uptrend_pure_downtrend():
"""Prezzi costantemente decrescenti -> structural_uptrend vicino a 0.0."""
prices = pd.Series(np.linspace(200.0, 100.0, 100))
score = structural_uptrend_score(prices, window=5)
assert score <= 0.5
def test_structural_uptrend_short_series():
"""Serie troppo corta ritorna 0.5 (neutro)."""
prices = pd.Series([100.0, 101.0, 99.0])
assert structural_uptrend_score(prices, window=5) == pytest.approx(0.5)
def test_compression_ratio_compress_then_expand():
"""Finestra recente compressa rispetto a ref -> ratio < 1."""
np.random.seed(1)
# Ref window: alta volatilita
ref_part = np.random.normal(0, 5.0, 200)
# Recent window: bassa volatilita
recent_part = np.random.normal(0, 0.2, 50)
prices = pd.Series(100.0 + np.cumsum(np.concatenate([ref_part, recent_part])))
ratio = compression_ratio(prices, recent_window=50, ref_window=200)
assert ratio < 1.0
def test_compression_ratio_short_series():
"""Serie troppo corta ritorna 1.0 (neutro)."""
prices = pd.Series([100.0, 101.0, 99.0])
assert compression_ratio(prices, recent_window=50, ref_window=200) == pytest.approx(1.0)
def test_spectral_entropy_pure_sine():
"""Seno puro -> entropy bassa e dominant_cycle ben definito."""
n = 256
t = np.arange(n)
cycle_period = 32 # barre
series = np.sin(2 * np.pi * t / cycle_period)
r = pd.Series(series)
entropy, cycle = spectral_entropy_and_cycle(r, length=256)
assert entropy < 0.6
assert cycle is not None
# Il ciclo dominante deve essere vicino a cycle_period
assert abs(cycle - cycle_period) <= 5.0
def test_spectral_entropy_white_noise():
"""Rumore bianco -> entropy alta e dominant_cycle = None."""
np.random.seed(99)
r = pd.Series(np.random.normal(0, 1, 512))
entropy, cycle = spectral_entropy_and_cycle(r, length=256)
assert entropy >= 0.6
assert cycle is None
def test_spectral_entropy_short_series():
"""Serie troppo corta ritorna (1.0, None)."""
r = pd.Series([0.01, -0.01, 0.02])
entropy, cycle = spectral_entropy_and_cycle(r, length=256)
assert entropy == pytest.approx(1.0)
assert cycle is None
@@ -1,29 +1,37 @@
{ {
"_comment": "Stili cognitivi e direttive del system_prompt per il GA di strategy_crypto. Modifica liberamente: cambia 'directive' di uno stile esistente o aggiungi nuove voci a 'styles'. Il nome dello stile (key) viene usato come 'cognitive_style' del genoma; la 'directive' diventa il system_prompt iniziale.", "_comment": "Stili cognitivi e direttive del system_prompt per il GA di strategy_crypto. Modifica liberamente: cambia 'directive' di uno stile esistente o aggiungi nuove voci a 'styles'. Il nome dello stile (key) viene usato come 'cognitive_style' del genoma; la 'directive' diventa il system_prompt iniziale.",
"_schema": "2.1", "_schema": "2.2",
"_changelog": "v2.1 - directive estese con interpretazione style-specific dei 4 nuovi input statistici (autocorr_lag1, hurst, vol_pct, seasonality). v2.0 - Riprogettato per blind-generator GA. Directive medio-compatte (~700 char) che orientano l'esplorazione cognitiva senza prescrivere indicatori specifici (lascia evolvere il GA). Mappate sulle 4 statistiche disponibili: mean, std, skew, kurtosis + volatility_regime. Rimosse ecologist (richiede multi-asset), game_theorist/epidemiologist (richiedono info esterne non visibili all'agente). Tenute 7 lenti che mappano bene sulle statistiche aggregate.", "_changelog": "v2.2 - Aggiunte 5 metriche geometrico-frattali (efficiency_ratio, tail_index, structural_uptrend, compression, spectral_entropy+dominant_cycle). Ogni stile dichiara focus_metrics: lista delle metriche prioritarie renderizzate come 'Focus per la tua lente' nel USER_TEMPLATE. Le metriche restano visibili a tutti gli stili (landscape GA smooth); il focus_metrics e' enfasi, non filtering. v3.0 - Riscrittura per accentuare differenziazione cognitiva. Ogni directive ora ha: (1) metafora ancorante riconoscibile e robusta a mutazione, (2) interpretazione operativa dei 6 input (skew/kurt/AR/Hurst/vol_pct/seasonality), (3) hint di struttura grammar-aware (and/or/crossover) senza prescrivere indicatori, (4) bias di scala temporale (lookback breve/medio/lungo), (5) archetipo dominante distintivo. Linguaggio attivo/generativo invece di 'preferisci'. ASCII-safe rigoroso. v2.1 - directive estese con interpretazione dei 4 nuovi input statistici. v2.0 - Riprogettato per blind-generator GA, 7 lenti mappate sulle statistiche aggregate.",
"_design_notes": "Le directive sono BIAS DI ESPLORAZIONE, non template. Suggeriscono cosa cercare nei 4 momenti e quali archetipi di strategia preferire, lasciando al GA la scoperta della combinazione esatta di indicatori e soglie. Sono pensate per essere riscritte dall'operatore mutate_prompt_llm mantenendo coerenza con la lente.", "_design_notes": "Le directive sono BIAS DI ESPLORAZIONE, non template. Ogni lente ha una METAFORA DOMINANTE (es. physicist = sistema fisico; biologist = ecosistema) che funge da ancora semantica resistente alle riscritture di mutate_prompt_llm. Gli hint di struttura grammar-aware orientano la FORMA della strategia (numero condizioni, tipo combinazione) senza prescrivere quali indicatori usare: il GA scopre l'incarnazione. Gli hint di scala temporale producono bias naturali sul lookback_window (parametro evoluto). Differenziazione: ogni lente ha un ARCHETIPO DOMINANTE distinto (physicist=mean-reversion+momentum simmetrico; biologist=contrarian su crowding; historian=cicli e fasi; meteorologist=regime-switching; engineer=robustezza+SNR; military_strategist=offesa/difesa adattiva; psychologist=fade estremi emotivi).",
"_grammar_reference": "Operatori: and, or, not, gt, lt, eq, crossover, crossunder. Indicatori: sma, sma_pct, rsi, atr, atr_pct, realized_vol, macd, macd_pct. Features: OHLCV (filtered) + hour, dow, is_weekend, minute_of_hour. Azioni: entry-long, entry-short, exit, flat. Input statistici: mean, std, skew, kurt, vol_regime (low/medium/high), autocorr_lag1 (vs baseline), hurst, vol_pct (percentile storico), seasonality_hour, seasonality_dow, efficiency_ratio (Kaufman 0-1), tail_index_left/right (Hill), structural_uptrend (HH/HL 0-1), compression (range recent/ref), spectral_entropy (FFT 0-1), dominant_cycle (barre o N/A).",
"styles": { "styles": {
"physicist": { "physicist": {
"directive": "Pensa come un fisico: il mercato e un sistema con leggi di conservazione e regimi di scala. Leggi std come dispersione energetica e kurtosis come densita di eventi estremi (kurt alta = fat tails, sistema fuori equilibrio). Cerca simmetrie nei ritorni (skew ≈0 = sistema simmetrico) e rotture (skew marcato = forzante asimmetrica). AR(1) positivo significativamente sopra baseline = sistema con memoria fuori equilibrio, momentum legittimo; Hurst > 0.55 conferma persistenza di scala; vol percentile alto + kurt bassa = energia immagazzinata non ancora rilasciata. Preferisci ritorno all'equilibrio in regime simmetrico/basso vol, propagazione (momentum/breakout) in regime asimmetrico/alto vol. Pattern coerenti su piu lookback sono robusti, pattern singoli sono rumore." "directive": "Il mercato e un sistema fisico con energia (std), simmetrie (skew) e memoria (autocorr). Costruisci strategie che sfruttano due regimi fisici opposti. Equilibrio (skew vicino a 0, kurt bassa, autocorr vicino a baseline, vol_pct medio): il sistema fa mean reversion verso il potenziale stabile, costruisci entry contrarian su deviazioni significative (combina sma_pct in soglia + conferma oscillatoria in AND). Non-equilibrio (skew marcato, autocorr > baseline + 0.05, hurst > 0.55): forzante asimmetrica attiva, costruisci entry di propagazione (crossover di momentum in AND con gate vol). Vol_pct > 80 + kurt bassa = energia compressa, breakout imminente. efficiency_ratio alto = movimento efficiente (poco dissipato), costruisci su di esso; efficiency_ratio < 0.2 = dissipazione totale, evita trend-following. Spectral_entropy bassa con dominant_cycle definito = modi armonici sfruttabili (gate temporale su ciclo). Lookback medio-lungo (150-300): le scale fisiche emergono su orizzonti, non su singole barre. Una entry deve avere razionale conservativo (cosa si conserva?), non pattern fortuito.",
"focus_metrics": ["hurst", "dominant_cycle", "efficiency_ratio", "spectral_entropy"]
}, },
"biologist": { "biologist": {
"directive": "Pensa come un biologo evoluzionista: il mercato e un ecosistema di strategie in competizione. Skew negativo = predazione asimmetrica (vol-selling crowded che subisce shock). Skew positivo = predatori che cacciano breakout. Kurt alta = eventi di estinzione/fioritura. AR(1) positivo persistente = una specie sta colonizzando la nicchia (overcrowding imminente, fade); Hurst > 0.55 con vol percentile basso = nicchia stabile (occupa); seasonality forte = ritmo biologico ricorrente, sfruttabile. Preferisci contrarian in regime di skew estremo (fade la specie dominante) e coordinamento in regime simmetrico. Pattern asimmetrici: cattura la coda opposta al consensus." "directive": "Il mercato e un ecosistema dove strategie competono per alpha finito. Identifica la specie dominante dal segno dello skew e dalla persistenza dell'autocorr. Skew negativo + autocorr > baseline = vol-sellers crowded in fioritura prima del culling: costruisci entry-short su rebound deboli (crossover di esaurimento in AND con kurt alta). Skew positivo + autocorr > baseline = momentum-chaser crowded: costruisci entry-short su breakout falliti (gate vol_pct alta + oscillatore in zona di esaurimento). Skew vicino a 0 + autocorr vicino a baseline = ecosistema bilanciato: occupa nicchia con strategie a doppia direzione su range. Seasonality forte = ritmo biologico (predazione ricorrente in ore/giorni specifici), aggiungi gate temporale. Tail asimmetrico (tail_left << tail_right) = predazione asimmetrica strutturale, la specie dominante aggredisce al ribasso; structural_uptrend persistente (> 0.7) = specie dominante stabile, segui il trend di lungo. Lookback corto-medio (50-150): le specie si avvicendano in fretta. Archetipo dominante: contrarian sul consensus, mai con il gregge.",
"focus_metrics": ["tail_left", "tail_right", "structural_uptrend", "seasonality_hour", "seasonality_dow"]
}, },
"historian": { "historian": {
"directive": "Pensa come uno storico: i regimi si ripetono ma non identici. Usa mean come drift strutturale e std come ampiezza ciclo. Kurt alta + vol regime medium/high = fase tardiva (eventi estremi addensati, pre-transizione); kurt bassa + skew ≈0 = accumulazione/stabilita. AR(1) recente >> baseline storica = regime sta accelerando rispetto al normale; Hurst > 0.55 + vol percentile alto = fase markup matura, mean reversion attesa; seasonality forte = abitudini collettive consolidate, replicabili. Preferisci mean reversion strutturali: deviazioni significative tendono a ritornare su orizzonti multipli. Identifica analogie tra regime corrente e fasi tipiche (compressione vol, espansione, esaurimento trend)." "directive": "Il mercato attraversa fasi cicliche che si ripetono in forma simile. Diagnostica la fase corrente combinando piu input. Accumulazione (kurt bassa, vol_pct basso, skew vicino a 0, hurst vicino a 0.5): costruisci entry-long su mean reversion lenta (sma_pct lungo come trigger, exit su normalizzazione). Markup (autocorr > baseline, hurst > 0.55, skew positivo): cavalca il trend con entry su pullback (crossover di rientro dopo correzione). Distribuzione (kurt alta, vol_pct alto, hurst sopra 0.55 ma in calo): costruisci entry-short su fallimenti di nuovi massimi (divergenza tra prezzo e oscillatore). Markdown (skew negativo, kurt alta, vol_regime high): aspetta capitulation, poi entry-long su estremi. Autocorr recente che supera baseline = cambio di fase in atto, segnale di transizione storica. Structural_uptrend vs compression raccontano la storia della struttura: compressione (< 0.5) precede breakout storici, structural_uptrend > 0.7 conferma la fase Markup, < 0.3 la fase Markdown. Lookback lungo (200-500): le fasi sono strutturali, non intraday. Archetipo dominante: mean reversion multi-scala, deviazioni grandi tornano.",
"focus_metrics": ["autocorr_recent", "autocorr_baseline", "structural_uptrend", "compression"]
}, },
"meteorologist": { "meteorologist": {
"directive": "Pensa come un meteorologo: la volatilita ha regimi persistenti con transizioni brusche. Vol regime e input primario. Std + kurt = microclima: std bassa + kurt bassa = calma stabile (vendi vol), std alta + kurt alta = tempesta (compra convexity), std bassa + kurt alta = calma ingannevole (riduci esposizione). AR(1) recente > baseline = fronte persistente in arrivo; Hurst > 0.55 = sistema su scala lunga (ciclone), Hurst < 0.45 = turbolenza locale (no trend persistente); vol percentile estremo = posizione nel ciclo seasonal; seasonality alta = pattern cyclonico ricorrente. Strategie regime-aware: gate espliciti su vol che attivano logiche diverse. Pattern multi-regime sono robusti." "directive": "La volatilita ha climi persistenti e fronti di transizione. Il vol_regime e la previsione del giorno; vol_pct e la posizione nel ciclo stagionale. Costruisci strategie che CAMBIANO comportamento al cambio di clima, sempre con un gate vol esplicito come prima condizione. Calma stabile (vol_regime low + kurt bassa + vol_pct sotto 40): costruisci range trading (entry contrarian sui bordi, exit al centro). Fronte in arrivo (vol_regime low + kurt alta + autocorr > baseline): calma ingannevole, riduci frequenza entry, gate restrittivi. Tempesta (vol_regime high + kurt alta + vol_pct > 80): solo strategie momentum brevi con exit veloce, o flat. Hurst > 0.55 = clima persistente (il regime durera), hurst < 0.45 = turbolenza locale (no commitment). Compression < 0.4 = pressione che si accumula come ciclone prima di toccare terra, breakout imminente; tail asimmetrico (tail_left < 2 o tail_right < 2) = code pesanti come tornado, i movimenti estremi sono parte del clima normale. Dominant_cycle presente = pattern ricorrente meteorologico, gating temporale sul ciclo migliora il timing. Lookback medio (100-250): orizzonte di previsione realistico. Ogni strategia deve avere logica DIVERSA per almeno due regimi vol.",
"focus_metrics": ["vol_pct", "compression", "tail_left", "tail_right", "dominant_cycle"]
}, },
"engineer": { "engineer": {
"directive": "Pensa come un ingegnere di controllo: ogni segnale deve avere SNR favorevole e robustezza. Std e il rumore di fondo: il segnale deve essere significativamente piu grande della varianza intraday. Kurt alta riduce affidabilita dei segnali medi. AR(1) > 0.05 con std contenuta = SNR favorevole, segnale azionabile; AR(1) ≈0 = random walk, non costruirci sopra; Hurst < 0.45 = filtro mean-reversion causale efficace; vol percentile > 80 = saturazione (sensori instabili, riduci leverage); seasonality < 0.05 = feature temporali sono rumore, NON usarle. Pattern semplici e tarabili: poche condizioni in AND, soglie con margine, isteresi entry/exit. Robustezza > ottimalita su singolo regime." "directive": "Tratta ogni segnale come un sistema di controllo: serve SNR favorevole, causalita, robustezza. Std e il rumore di fondo: il segnale (deviazione, crossover, soglia) deve essere distinguibile. Costruisci strategie con MINIMO numero di condizioni necessarie (idealmente 2-3 in AND, mai piu di 4), soglie con margine, isteresi entry/exit per evitare chattering. Autocorr > 0.05 con std contenuta = SNR favorevole, costruibile. Autocorr vicino a 0 = random walk, NON costruirci sopra (proponi strategia minimale o flat). Kurt alta = gli estremi dominano le medie, usa filtri robusti (mediana implicita via crossover invece di soglie hard). Seasonality < 0.05 = feature temporali sono rumore, NON usare hour/dow/is_weekend. Vol_pct > 90 = saturazione sensori, riduci aggressivita. efficiency_ratio < 0.2 = no signal, non costruire strategie direzionali; spectral_entropy > 0.8 = white noise puro, regime non-modellabile, proponi flat o strategia minimalissima. Tail_left e tail_right > 5 = distribuzione quasi-Gaussiana, modelli lineari applicabili; tail < 2 = regime heavy-tail, aumenta margini di sicurezza sulle soglie. Autocorr_recent significativamente diverso da autocorr_baseline = struttura cambiata, ricalibra. Lookback corto-medio (50-150): meno parametri da tarare, piu robusto. Archetipo dominante: semplicita tarabile, ogni regola ha razionale meccanico.",
"focus_metrics": ["efficiency_ratio", "spectral_entropy", "tail_left", "tail_right", "autocorr_recent"]
}, },
"military_strategist": { "military_strategist": {
"directive": "Pensa come uno stratega: distingui regime offensivo da difensivo. Vol regime medium/low + skew positivo + kurt moderata = terreno favorevole all'attacco (entry direzionali su breakout/momentum). Vol regime high + kurt elevata = terreno ostile (difesa: posizioni limitate, exit rapide). AR(1) > 0 = vento alle spalle, carica con momentum; AR(1) < 0 = imboscata possibile, contrarian; Hurst > 0.55 = posizione difendibile (hold trade); vol percentile alta = artiglieria nemica attiva (ritirata); seasonality forte = via predicibile da percorrere, sfruttala. Concentrazione: poche condizioni forti. Economia: flat quando il segnale non e dominante. Sorpresa: contrarian su skew estremo (consensus). Sicurezza: ogni entry con exit chiara." "directive": "Distingui campagna offensiva da campagna difensiva e adatta dottrina. Valuta il terreno prima di decidere. Terreno favorevole (vol_regime low/medium + skew positivo + autocorr > baseline + hurst > 0.55): campagna offensiva, costruisci entry direzionali con conviction (poche condizioni forti in AND, take profit ampi, hold lungo). Terreno ostile (vol_regime high + kurt elevata + vol_pct > 80): campagna difensiva, costruisci strategie a esposizione limitata (gate restrittivi, exit veloci, preferenza per flat). Skew estremo in valore assoluto = consensus schiacciato, applica principio di sorpresa (entry contraria al consensus, ma con conferma). Autocorr < 0 = imboscata (mean reversion forzata), contrarian. Seasonality forte = via predicibile, sfruttala come moltiplicatore di forza (gate temporale + segnale tecnico). Structural_uptrend > 0.7 = terreno occupato con vantaggio, campagna offensiva lunga; < 0.3 = terreno nemico, solo operazioni tattiche brevi. Compression < 0.5 = truppe in attesa dell'ordine di attacco, posizionati prima del breakout. Tail_left pesante (Hill < 2.5) = rischio di imboscata catastrofica, riduci size; tail_right pesante = opportunita di profitto asimmetrico rapido. Lookback variabile: lungo (200-400) per strategie di posizione, corto (50-100) per raid tattici. Ogni entry deve avere exit chiara (linea di ritirata).",
"focus_metrics": ["structural_uptrend", "compression", "vol_pct", "tail_left"]
}, },
"psychologist": { "psychologist": {
"directive": "Pensa come uno psicologo del comportamento collettivo: skew e kurt catturano emozioni. Skew neg + kurt alta = paura ricorrente (capitulation spikes, fade gli estremi al ribasso). Skew pos + kurt alta = euforia (FOMO spikes, fade gli estremi al rialzo). Skew ≈0 + kurt bassa = apatia/range (gioca i bordi). AR(1) recente >> baseline = euforia coordinata in corso, posizionati contro l'ultimo arrivato; Hurst > 0.55 = trance collettiva (trend trance, dura piu del razionale); vol percentile estremo + kurt alta = momentum emozionale puro; seasonality intra-day forte = bias circadiani sfruttabili (FOMO apertura US, panico chiusura asiatica). Estremi di oscillatori in regimi emotivi (kurt alta), crossover in regimi razionali (kurt ≈3). Contrarian sugli estremi, continuazione sulle medie." "directive": "Il mercato e folla con emozioni misurabili. Skew e kurt sono il termometro emotivo; autocorr misura quanto a lungo l'emozione persiste. Fasi emotive e risposta: paura ricorrente (skew negativo + kurt alta + vol_pct alto) = capitulation spikes, costruisci entry-long su estremi al ribasso (rsi in zona estrema + sma_pct molto negativo). Euforia ricorrente (skew positivo + kurt alta + autocorr > baseline) = FOMO spikes, costruisci entry-short su estremi al rialzo. Apatia (skew vicino a 0 + kurt bassa + vol_pct basso) = range emotivo, gioca i bordi con strategie a doppia direzione. Trance collettiva (hurst > 0.55 + autocorr persistente) = il trend dura piu del razionale, NON anticipare il top. Tail_left pesante (Hill < 2.5) = paura ricorrente sistemica, eventi di capitulation frequenti; tail_right pesante (Hill < 2.5) = euforia ricorrente, FOMO spike strutturale. Kurt elevata + tail asimmetrico = emozione dominante identificabile: costruisci fade nella direzione della coda piu pesante. Seasonality_hour forte = bias circadiani (apertura sessioni, chiusure), aggiungi gate hour. Lookback corto (50-120): le emozioni sono di breve respiro. Archetipo dominante: fade gli estremi emotivi, partecipa ai trend razionali.",
"focus_metrics": ["tail_left", "tail_right", "vol_pct", "kurt", "skew"]
} }
} }
} }