`scripts/validate_strategy.py`

83 lines
#!/usr/bin/env python3
"""
validate_strategy.py — validate a strategy.json BEFORE a bot ever runs it.

    scripts/validate_strategy.py path/to/strategy.json

Checks required keys, sane value ranges, and — critically — that every entry/exit
rule parses under the sandboxed evaluator and references only known market
variables. A strategy that fails here would silently misfire at runtime; catch it
now. Exit code 0 = valid, 1 = invalid.
"""

import ast
import json
import sys

# Kept in sync with assets/bot-template/safe_eval.py MARKET_VARS.
MARKET_VARS = ("price", "volume_usd", "liquidity_usd", "spread", "days_to_resolve")
_OK_NODES = (ast.Expression, ast.BoolOp, ast.BinOp, ast.UnaryOp, ast.Compare,
             ast.Name, ast.Constant, ast.And, ast.Or, ast.Not, ast.USub, ast.UAdd,
             ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Mod, ast.Pow,
             ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE, ast.Load)


def check_rule(expr):
    try:
        tree = ast.parse(expr, mode="eval")
    except SyntaxError as e:
        return f"cannot parse: {e}"
    for node in ast.walk(tree):
        if not isinstance(node, _OK_NODES):
            return f"disallowed element {type(node).__name__} in {expr!r}"
        if isinstance(node, ast.Name) and node.id not in MARKET_VARS + ("True", "False"):
            return f"unknown variable {node.id!r} in {expr!r} (allowed: {', '.join(MARKET_VARS)})"
    return None


def validate(strat):
    errs = []
    for key in ("name", "venue", "entry", "exit", "sizing", "risk"):
        if key not in strat:
            errs.append(f"missing required key: {key}")
    if strat.get("venue") not in ("polymarket", "kalshi"):
        errs.append(f"venue must be polymarket|kalshi, got {strat.get('venue')!r}")
    if strat.get("mode", "paper") not in ("paper", "live"):
        errs.append("mode must be paper|live")
    for section in ("entry", "exit"):
        for i, rule in enumerate(strat.get(section, [])):
            if "when" not in rule:
                errs.append(f"{section}[{i}] missing 'when'")
                continue
            problem = check_rule(rule["when"])
            if problem:
                errs.append(f"{section}[{i}]: {problem}")
    sizing = strat.get("sizing", {})
    if sizing.get("amount", 0) <= 0:
        errs.append("sizing.amount must be > 0")
    if sizing.get("max_position_usd", 0) < sizing.get("amount", 0):
        errs.append("sizing.max_position_usd must be >= sizing.amount")
    risk = strat.get("risk", {})
    if "max_total_exposure_usd" not in risk:
        errs.append("risk.max_total_exposure_usd is required (exposure cap)")
    return errs


def main():
    if len(sys.argv) != 2:
        sys.exit("usage: validate_strategy.py <strategy.json>")
    with open(sys.argv[1]) as f:
        strat = json.load(f)
    errs = validate(strat)
    if errs:
        print("❌ INVALID strategy:")
        for e in errs:
            print("   - " + e)
        sys.exit(1)
    print(f"✅ valid strategy: {strat['name']} ({strat['venue']}, mode={strat.get('mode','paper')})")
    print(f"   entry rules: {len(strat.get('entry', []))}  exit rules: {len(strat.get('exit', []))}")


if __name__ == "__main__":
    main()