50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
"""Simple policy DSL and evaluator for BeXProof.
|
|
|
|
This module provides a lightweight policy container and a tiny evaluator.
|
|
Policies are represented as JSON-like strings for simplicity in this MVP.
|
|
In production, replace with a proper DSL parser and validator.
|
|
"""
|
|
from __future__ import annotations
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Any
|
|
|
|
|
|
@dataclass
|
|
class Policy:
|
|
version: int
|
|
rules: Dict[str, Any]
|
|
|
|
|
|
def load_policy(policy_text: str) -> Policy:
|
|
# Accept either JSON or Python-like dict string (with single quotes)
|
|
data = None
|
|
try:
|
|
data = json.loads(policy_text)
|
|
except Exception:
|
|
# try Python-style dict string
|
|
try:
|
|
data = json.loads(policy_text.replace("'", '"'))
|
|
except Exception:
|
|
raise ValueError("Policy text is not valid JSON or Python-like dict string")
|
|
if not isinstance(data, dict) or "version" not in data or "rules" not in data:
|
|
raise ValueError("Policy must contain 'version' and 'rules' keys")
|
|
return Policy(version=int(data["version"]), rules=data["rules"])
|
|
|
|
|
|
def evaluate_policy(log: Dict[str, Any], policy: Policy) -> bool:
|
|
# Minimal evaluation: all top-level rules keys are checked if present in log
|
|
# This is a small MVP; in production, rules would be richer and more formal.
|
|
for key, thresh in policy.rules.items():
|
|
if key == "price_improvement_min":
|
|
if log.get("price_improvement", 0) < float(thresh):
|
|
return False
|
|
elif key == "latency_budget_ms":
|
|
if log.get("latency_ms", float("inf")) > int(thresh):
|
|
return False
|
|
elif key == "slippage_max":
|
|
if log.get("slippage", float("inf")) > float(thresh):
|
|
return False
|
|
# additional rules can be added here
|
|
return True
|