Files
Multi_Swarm_Coevolutive/src/multi_swarm/protocol/validator.py
T
Adriano 44eb6436c1 refactor(protocol): swap S-expression grammar for strict JSON Schema
Sostituisce la grammatica S-expression con uno schema JSON stretto. La
grammatica S-expression falliva il parsing nel 64% delle generazioni del
modello Qwen3-235B sul run reale; JSON e' nativo per gli LLM moderni e
si parsa con json.loads.

Cambiamenti principali:
- grammar.py: costanti rinominate LOGICAL_OPS / COMPARATOR_OPS /
  CROSSOVER_OPS / ACTION_VALUES / KIND_VALUES.
- parser.py: nuovo AST a dataclass tipizzato (OpNode, IndicatorNode,
  FeatureNode, LiteralNode, Rule, Strategy); parse_strategy ora consuma
  JSON tramite json.loads.
- validator.py: walk dispatchato per tipo (isinstance) invece di
  pattern-matching su 'kind'; arity check su operatori e indicator.
- compiler.py: traversal del nuovo AST tipizzato, dispatch per
  isinstance; logica indicator/feature/literal invariata.
- hypothesis.py: prompt SYSTEM riscritto con esempi JSON e vincoli
  espliciti su no-nesting; estrazione via fence ```json``` + fallback
  brace-balanced.
- __init__.py: re-export pubblico delle entita' del protocollo.
- Tutti i test (parser, validator, compiler, hypothesis_agent,
  falsification, adversarial, e2e, smoke_run) migrati a JSON.
- Rimossa dipendenza sexpdata da pyproject.toml + uv.lock.

Test: 135 passed (era 122; aggiunti casi parser/validator).
ruff + mypy strict clean. Smoke run end-to-end OK.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-10 21:17:26 +02:00

110 lines
3.2 KiB
Python

"""Semantic validation for the JSON-based strategy AST.
Il parser garantisce già shape sintattica (op vs kind, struttura args/params,
tipi base). Qui si controllano vincoli semantici di Phase 1:
* Arity di operatori logici / comparatori / crossover.
* Whitelist indicator + arity dei params.
* Whitelist feature.
* Niente nesting di indicator (params puramente numerici, garantito già dal
parser ma ricontrollato esplicitamente per chiarezza).
"""
from __future__ import annotations
from .grammar import (
COMPARATOR_OPS,
CROSSOVER_OPS,
KNOWN_FEATURES,
KNOWN_INDICATORS,
LOGICAL_OPS,
)
from .parser import (
FeatureNode,
IndicatorNode,
LiteralNode,
Node,
OpNode,
Strategy,
)
# Numero di parametri numerici accettati dopo il nome dell'indicatore.
# (min, max) sui soli numeri. Indicatori non sono annidabili in Phase 1.
INDICATOR_ARITY: dict[str, tuple[int, int]] = {
"sma": (1, 1), # length
"rsi": (1, 1), # length
"atr": (1, 1), # length
"realized_vol": (1, 1), # window
"macd": (0, 3), # fast, slow, signal (tutti opzionali)
}
class ValidationError(Exception):
"""Raised when an AST violates Phase 1 protocol semantics."""
def validate_strategy(strategy: Strategy) -> None:
"""Walk every rule of the strategy and assert semantic constraints."""
for rule in strategy.rules:
_validate_node(rule.condition)
def _validate_node(node: Node) -> None:
if isinstance(node, OpNode):
_validate_op(node)
return
if isinstance(node, IndicatorNode):
_validate_indicator(node)
return
if isinstance(node, FeatureNode):
if node.name not in KNOWN_FEATURES:
raise ValidationError(f"unknown feature: {node.name}")
return
if isinstance(node, LiteralNode):
# parser ha già validato il tipo numerico
return
raise ValidationError(f"unexpected node type: {type(node).__name__}")
def _validate_op(node: OpNode) -> None:
op = node.op
n = len(node.args)
if op in LOGICAL_OPS:
if op == "not":
if n != 1:
raise ValidationError(f"'not' needs 1 arg, got {n}")
else:
if n < 2:
raise ValidationError(f"'{op}' needs >=2 args, got {n}")
for a in node.args:
_validate_node(a)
return
if op in COMPARATOR_OPS:
if n != 2:
raise ValidationError(f"'{op}' needs 2 args, got {n}")
for a in node.args:
_validate_node(a)
return
if op in CROSSOVER_OPS:
if n != 2:
raise ValidationError(f"'{op}' needs 2 args, got {n}")
for a in node.args:
_validate_node(a)
return
raise ValidationError(f"unexpected op in expression: {op}")
def _validate_indicator(node: IndicatorNode) -> None:
if node.name not in KNOWN_INDICATORS:
raise ValidationError(f"unknown indicator: {node.name}")
n_params = len(node.params)
min_p, max_p = INDICATOR_ARITY[node.name]
if not (min_p <= n_params <= max_p):
raise ValidationError(
f"indicator '{node.name}' arity {n_params} out of [{min_p},{max_p}]"
)