`assets/bot-template/safe_eval.py`

99 lines
"""
safe_eval.py — a deliberately tiny, sandboxed expression evaluator.

Strategy rules ("price <= 0.08 and volume_usd >= 100000") are user/AI authored,
so we NEVER hand them to Python's built-in eval(). Instead we parse each rule to
an AST and walk only a whitelist of node types + variable names. Anything else
(function calls, attribute access, imports, dunder names) is refused.

This is the single most important safety boundary in the bot: a strategy file is
data, not code. Keep it that way.
"""

import ast
import operator

# Only these operators are allowed inside a rule expression.
_BIN_OPS = {
    ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul,
    ast.Div: operator.truediv, ast.Mod: operator.mod, ast.Pow: operator.pow,
}
_CMP_OPS = {
    ast.Eq: operator.eq, ast.NotEq: operator.ne, ast.Lt: operator.lt,
    ast.LtE: operator.le, ast.Gt: operator.gt, ast.GtE: operator.ge,
}
_BOOL_OPS = {ast.And: all, ast.Or: any}
_UNARY_OPS = {ast.USub: operator.neg, ast.UAdd: operator.pos, ast.Not: operator.not_}


class RuleError(ValueError):
    """A strategy rule that is malformed or references an unknown variable."""


def _eval_node(node, variables):
    if isinstance(node, ast.Expression):
        return _eval_node(node.body, variables)
    if isinstance(node, ast.BoolOp):
        fn = _BOOL_OPS[type(node.op)]
        return bool(fn(_eval_node(v, variables) for v in node.values))
    if isinstance(node, ast.BinOp):
        op = _BIN_OPS.get(type(node.op))
        if op is None:
            raise RuleError(f"operator not allowed: {type(node.op).__name__}")
        return op(_eval_node(node.left, variables), _eval_node(node.right, variables))
    if isinstance(node, ast.UnaryOp):
        op = _UNARY_OPS.get(type(node.op))
        if op is None:
            raise RuleError(f"unary operator not allowed: {type(node.op).__name__}")
        return op(_eval_node(node.operand, variables))
    if isinstance(node, ast.Compare):
        left = _eval_node(node.left, variables)
        for op_node, comparator in zip(node.ops, node.comparators):
            op = _CMP_OPS.get(type(op_node))
            if op is None:
                raise RuleError(f"comparison not allowed: {type(op_node).__name__}")
            right = _eval_node(comparator, variables)
            if not op(left, right):
                return False
            left = right
        return True
    if isinstance(node, ast.Name):
        if node.id in ("True", "False"):  # py<3.8 safety; harmless otherwise
            return node.id == "True"
        if node.id not in variables:
            raise RuleError(f"unknown variable: {node.id!r}")
        return variables[node.id]
    if isinstance(node, ast.Constant):  # numbers, bools, strings
        return node.value
    raise RuleError(f"expression element not allowed: {type(node).__name__}")


def eval_rule(expr, variables):
    """Evaluate a strategy rule string against a dict of market variables."""
    try:
        tree = ast.parse(expr, mode="eval")
    except SyntaxError as e:
        raise RuleError(f"cannot parse rule {expr!r}: {e}") from e
    return _eval_node(tree, variables)


def check_rule(expr, allowed_vars):
    """Static check: parses the rule and confirms every name is a known variable.
    Returns (ok, message). Used by the strategy validator before a bot ever runs."""
    dummy = {name: 0 for name in allowed_vars}
    try:
        eval_rule(expr, dummy)
        return True, "ok"
    except RuleError as e:
        return False, str(e)


# The variables a rule may reference. venue_adapter.py populates these per market.
MARKET_VARS = (
    "price",            # current YES price, 0..1
    "volume_usd",       # lifetime USD volume
    "liquidity_usd",    # order-book liquidity
    "spread",           # best ask - best bid, 0..1
    "days_to_resolve",  # days until market resolution
)