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)