From 19035812c36544149fd6d8358a287c319a3e3990 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Sat, 9 May 2026 19:30:54 +0200 Subject: [PATCH] feat(protocol): grammatica S-expression (15 verbi) + parser Aggiunge il modulo `multi_swarm.protocol` con la grammatica del DSL di strategia (15 verbi: 4 azioni, 3 logici, 3 comparatori, 4 dati, `when` e `strategy`) e un parser che produce un AST tipizzato (Strategy/Rule/ Node). Lessing delegato a sexpdata, validazione del set di verbi e forma `(when )` gestita dal parser. Sollevata ParseError su top-level malformato, strategia vuota, verbi sconosciuti o azioni non terminali. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/multi_swarm/protocol/__init__.py | 0 src/multi_swarm/protocol/grammar.py | 26 ++++++++ src/multi_swarm/protocol/parser.py | 96 ++++++++++++++++++++++++++++ tests/unit/test_protocol_parser.py | 47 ++++++++++++++ 4 files changed, 169 insertions(+) create mode 100644 src/multi_swarm/protocol/__init__.py create mode 100644 src/multi_swarm/protocol/grammar.py create mode 100644 src/multi_swarm/protocol/parser.py create mode 100644 tests/unit/test_protocol_parser.py diff --git a/src/multi_swarm/protocol/__init__.py b/src/multi_swarm/protocol/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/multi_swarm/protocol/grammar.py b/src/multi_swarm/protocol/grammar.py new file mode 100644 index 0000000..c3d7422 --- /dev/null +++ b/src/multi_swarm/protocol/grammar.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +VERBS: frozenset[str] = frozenset( + { + "entry-long", + "entry-short", + "exit", + "flat", + "when", + "and", + "or", + "not", + "gt", + "lt", + "eq", + "feature", + "indicator", + "crossover", + "crossunder", + } +) + +ACTION_VERBS: frozenset[str] = frozenset({"entry-long", "entry-short", "exit", "flat"}) +LOGICAL_VERBS: frozenset[str] = frozenset({"and", "or", "not"}) +COMPARATOR_VERBS: frozenset[str] = frozenset({"gt", "lt", "eq"}) +DATA_VERBS: frozenset[str] = frozenset({"feature", "indicator", "crossover", "crossunder"}) diff --git a/src/multi_swarm/protocol/parser.py b/src/multi_swarm/protocol/parser.py new file mode 100644 index 0000000..0d8cb59 --- /dev/null +++ b/src/multi_swarm/protocol/parser.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +import sexpdata # type: ignore[import-untyped] + +from .grammar import ACTION_VERBS, VERBS + + +class ParseError(Exception): + """Raised when an S-expression strategy cannot be parsed.""" + + +@dataclass +class Node: + kind: str + args: list[Any] = field(default_factory=list) + + +@dataclass +class Rule: + kind: str # always "when" + condition: Node + action: Node + + +@dataclass +class Strategy: + kind: str # always "strategy" + rules: list[Rule] + + +def _to_node(token: Any) -> Node | float | int | str: + """Convert a sexpdata token tree into a Node (or scalar leaf).""" + if isinstance(token, sexpdata.Symbol): + name = str(token.value()) + # Bare symbols inside expressions (e.g. `rsi` in (indicator rsi 14)) + # are kept as Node-with-no-args so callers can introspect uniformly. + return Node(kind=name, args=[]) + if isinstance(token, list): + if not token: + raise ParseError("Empty s-expression") + head = token[0] + if not isinstance(head, sexpdata.Symbol): + raise ParseError(f"Non-symbol head: {head!r}") + name = str(head.value()) + if name not in VERBS: + raise ParseError(f"Unknown verb: {name}") + return Node(kind=name, args=[_to_node(arg) for arg in token[1:]]) + # numeric / string literals pass through unchanged + return token # type: ignore[no-any-return] + + +def parse_strategy(src: str) -> Strategy: + """Parse an S-expression strategy string into a Strategy AST. + + The grammar is documented in :mod:`multi_swarm.protocol.grammar` and is + intentionally tiny (15 verbs). We delegate raw S-expr lexing to + :mod:`sexpdata`, then validate the verb set ourselves. + """ + try: + parsed = sexpdata.loads(src) + except Exception as e: # sexpdata raises various exception types + raise ParseError(f"sexp parse error: {e}") from e + + if not isinstance(parsed, list) or not parsed: + raise ParseError("Top-level must be (strategy ...)") + head = parsed[0] + if not isinstance(head, sexpdata.Symbol) or str(head.value()) != "strategy": + raise ParseError("Top-level must start with 'strategy'") + + raw_rules = parsed[1:] + if not raw_rules: + raise ParseError("Strategy must contain at least one rule") + + rules: list[Rule] = [] + for raw in raw_rules: + if not isinstance(raw, list) or len(raw) != 3: + raise ParseError(f"Rule must be (when ): {raw!r}") + head_r = raw[0] + if not isinstance(head_r, sexpdata.Symbol) or str(head_r.value()) != "when": + raise ParseError(f"Rule must start with 'when': {raw!r}") + cond = _to_node(raw[1]) + action = _to_node(raw[2]) + if not isinstance(cond, Node): + raise ParseError(f"Condition must be a node: {cond!r}") + if not isinstance(action, Node): + raise ParseError(f"Action must be a node: {action!r}") + if action.kind not in ACTION_VERBS: + raise ParseError( + f"Action must be one of {sorted(ACTION_VERBS)}, got {action.kind!r}" + ) + rules.append(Rule(kind="when", condition=cond, action=action)) + + return Strategy(kind="strategy", rules=rules) diff --git a/tests/unit/test_protocol_parser.py b/tests/unit/test_protocol_parser.py new file mode 100644 index 0000000..2313b91 --- /dev/null +++ b/tests/unit/test_protocol_parser.py @@ -0,0 +1,47 @@ +import pytest + +from multi_swarm.protocol.grammar import VERBS +from multi_swarm.protocol.parser import ParseError, parse_strategy + + +def test_grammar_has_15_verbs(): + assert len(VERBS) == 15 + + +def test_parse_simple_strategy(): + src = "(strategy (when (gt (indicator rsi 14) 70.0) (entry-short)))" + ast = parse_strategy(src) + assert ast.kind == "strategy" + assert len(ast.rules) == 1 + rule = ast.rules[0] + assert rule.kind == "when" + assert rule.condition.kind == "gt" + assert rule.action.kind == "entry-short" + + +def test_parse_multiple_rules(): + src = """ + (strategy + (when (gt (indicator rsi 14) 70.0) (entry-short)) + (when (lt (indicator rsi 14) 30.0) (entry-long))) + """ + ast = parse_strategy(src) + assert len(ast.rules) == 2 + + +def test_parse_unknown_verb_raises(): + src = "(strategy (when (frobnicate 1 2) (entry-long)))" + with pytest.raises(ParseError): + parse_strategy(src) + + +def test_parse_malformed_raises(): + src = "(strategy (when" + with pytest.raises(ParseError): + parse_strategy(src) + + +def test_parse_empty_strategy_raises(): + src = "(strategy)" + with pytest.raises(ParseError): + parse_strategy(src)