"""JSON-based parser per la strategia di trading (Phase 1). L'AST è una piccola gerarchia di dataclass: * :class:`Strategy` è il top-level (lista di :class:`Rule`). * :class:`Rule` accoppia una condizione (Node) ad un'azione (str). * :class:`Node` è un'unione: nodi operatore (:class:`OpNode`) e nodi leaf (:class:`IndicatorNode`, :class:`FeatureNode`, :class:`LiteralNode`). Convenzione di shape sui dict in input: * Nodi operatore: ``{"op": "", "args": [, ...]}``. * Nodi indicator: ``{"kind": "indicator", "name": "", "params": [, ...]}``. * Nodi feature: ``{"kind": "feature", "name": ""}``. * Nodi literal: ``{"kind": "literal", "value": }``. """ from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any from .grammar import ( ACTION_VALUES, ALL_OPS, ) class ParseError(Exception): """Raised when a JSON strategy cannot be parsed into a valid AST.""" # --------------------------------------------------------------------------- # Dataclass AST # --------------------------------------------------------------------------- @dataclass class OpNode: """Operator node: logical / comparator / crossover.""" op: str args: list[Node] = field(default_factory=list) @dataclass class IndicatorNode: """Leaf: indicatore tecnico calcolato sul dataframe OHLCV.""" name: str params: list[float] = field(default_factory=list) @dataclass class FeatureNode: """Leaf: colonna OHLCV (open/high/low/close/volume).""" name: str @dataclass class LiteralNode: """Leaf: costante numerica.""" value: float Node = OpNode | IndicatorNode | FeatureNode | LiteralNode @dataclass class Rule: condition: Node action: str @dataclass class Strategy: rules: list[Rule] # --------------------------------------------------------------------------- # Conversione dict -> Node # --------------------------------------------------------------------------- def _to_node(obj: Any) -> Node: if not isinstance(obj, dict): raise ParseError(f"Node must be a JSON object, got {type(obj).__name__}") has_op = "op" in obj has_kind = "kind" in obj if has_op and has_kind: raise ParseError( "Node cannot define both 'op' and 'kind' (mutually exclusive)" ) if not has_op and not has_kind: raise ParseError("Node must define either 'op' or 'kind'") if has_op: op = obj["op"] if not isinstance(op, str): raise ParseError(f"'op' must be a string, got {type(op).__name__}") if op not in ALL_OPS: raise ParseError(f"Unknown op: {op!r}") raw_args = obj.get("args") if not isinstance(raw_args, list): raise ParseError(f"Operator '{op}' missing 'args' list") args = [_to_node(a) for a in raw_args] return OpNode(op=op, args=args) # leaf node kind = obj["kind"] if not isinstance(kind, str): raise ParseError(f"'kind' must be a string, got {type(kind).__name__}") if kind == "indicator": name = obj.get("name") if not isinstance(name, str): raise ParseError("indicator node requires string 'name'") raw_params = obj.get("params", []) if not isinstance(raw_params, list): raise ParseError("indicator 'params' must be a list") params: list[float] = [] for p in raw_params: if isinstance(p, bool) or not isinstance(p, (int, float)): raise ParseError( f"indicator '{name}' params accept only numbers, got {p!r}" ) params.append(float(p)) return IndicatorNode(name=name, params=params) if kind == "feature": name = obj.get("name") if not isinstance(name, str): raise ParseError("feature node requires string 'name'") return FeatureNode(name=name) if kind == "literal": if "value" not in obj: raise ParseError("literal node requires 'value'") value = obj["value"] if isinstance(value, bool) or not isinstance(value, (int, float)): raise ParseError(f"literal value must be numeric, got {value!r}") return LiteralNode(value=float(value)) raise ParseError(f"Unknown leaf kind: {kind!r}") # --------------------------------------------------------------------------- # Top-level parser # --------------------------------------------------------------------------- def parse_strategy(src: str) -> Strategy: """Parse a JSON strategy string into a :class:`Strategy` AST. Lo schema atteso è:: { "rules": [ {"condition": , "action": ""}, ... ] } Raise :class:`ParseError` su JSON malformato o struttura inattesa. """ try: parsed = json.loads(src) except json.JSONDecodeError as e: raise ParseError(f"invalid JSON: {e}") from e if not isinstance(parsed, dict): raise ParseError("Top-level must be a JSON object with 'rules'") if "rules" not in parsed: raise ParseError("Top-level object must contain 'rules' key") raw_rules = parsed["rules"] if not isinstance(raw_rules, list): raise ParseError("'rules' must be a list") 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, dict): raise ParseError(f"Rule must be a JSON object, got {raw!r}") if "condition" not in raw or "action" not in raw: raise ParseError( f"Rule must contain 'condition' and 'action' keys: {raw!r}" ) action = raw["action"] if not isinstance(action, str): raise ParseError(f"action must be a string, got {action!r}") if action not in ACTION_VALUES: raise ParseError( f"action must be one of {sorted(ACTION_VALUES)}, got {action!r}" ) cond = _to_node(raw["condition"]) rules.append(Rule(condition=cond, action=action)) return Strategy(rules=rules)