deltaforge-real-time-cross-.../deltaforge_mvp/dsl.py

127 lines
2.5 KiB
Python

from __future__ import annotations
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
import json
@dataclass
class LocalArbProblem:
id: str
venue: str
assets: List[str]
objective: Dict[str, float]
constraints: Dict[str, float]
solver_hint: Optional[str] = None
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "LocalArbProblem":
data = json.loads(s)
return LocalArbProblem(**data)
@dataclass
class SharedSignals:
version: int
signals: Dict[str, float]
privacy_tag: str = "public"
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "SharedSignals":
data = json.loads(s)
return SharedSignals(**data)
@dataclass
class PlanDelta:
delta: Dict[str, float]
timestamp: float
author: str
contract_id: str
signature: str = ""
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "PlanDelta":
data = json.loads(s)
return PlanDelta(**data)
@dataclass
class DualVariables:
multipliers: Dict[str, float]
convergence_status: str = "not-converged"
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "DualVariables":
data = json.loads(s)
return DualVariables(**data)
@dataclass
class PrivacyBudget:
budget: float
expiry: float
leakage_bound: Optional[float] = None
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "PrivacyBudget":
data = json.loads(s)
return PrivacyBudget(**data)
@dataclass
class AuditLog:
entry: str
signer: str
timestamp: float
contract_id: str
version: str = "0.1"
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "AuditLog":
data = json.loads(s)
return AuditLog(**data)
@dataclass
class PolicyBlock:
safety: Dict[str, float]
exposure_controls: Dict[str, float]
def to_json(self) -> str:
return json.dumps(asdict(self))
@staticmethod
def from_json(s: str) -> "PolicyBlock":
data = json.loads(s)
return PolicyBlock(**data)
__all__ = [
"LocalArbProblem",
"SharedSignals",
"PlanDelta",
"DualVariables",
"PrivacyBudget",
"AuditLog",
"PolicyBlock",
]