95 lines
2.0 KiB
Python
95 lines
2.0 KiB
Python
from __future__ import annotations
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Any, Optional, List
|
|
import time
|
|
|
|
@dataclass
|
|
class LocalEvent:
|
|
event_id: str
|
|
event_type: str # e.g., "MDTick", "Signal", etc.
|
|
asset: str
|
|
venue: str
|
|
timestamp: float
|
|
payload: Dict[str, Any]
|
|
metadata: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
@dataclass
|
|
class PlanDelta:
|
|
delta_id: str
|
|
timestamp: float
|
|
author: str
|
|
contract_id: str
|
|
delta_payload: Dict[str, Any]
|
|
signature: str
|
|
|
|
@dataclass
|
|
class OrderEvent:
|
|
order_id: str
|
|
timestamp: float
|
|
side: str
|
|
quantity: int
|
|
price: float
|
|
provenance: str = ""
|
|
delta_id: Optional[str] = None
|
|
|
|
@dataclass
|
|
class FillEvent:
|
|
fill_id: str
|
|
timestamp: float
|
|
order_id: str
|
|
quantity: int
|
|
price: float
|
|
venue: str = ""
|
|
delta_id: Optional[str] = None
|
|
|
|
@dataclass
|
|
class RiskCheck:
|
|
check_id: str
|
|
timestamp: float
|
|
result: bool
|
|
budget_used: float
|
|
details: Dict[str, Any] = field(default_factory=dict)
|
|
|
|
@dataclass
|
|
class AuditLog:
|
|
log_id: str
|
|
timestamp: float
|
|
action: str
|
|
payload: Dict[str, Any]
|
|
signature: str
|
|
|
|
@dataclass
|
|
class Metadata:
|
|
version: str
|
|
nonce: str
|
|
source_adapter: str
|
|
|
|
@dataclass
|
|
class TraceGraph:
|
|
nodes: List[object] = field(default_factory=list)
|
|
edges: List[tuple] = field(default_factory=list) # (src_id, dst_id, tag)
|
|
|
|
def add_node(self, node: object) -> object:
|
|
self.nodes.append(node)
|
|
return node
|
|
|
|
def add_edge(self, src_id: str, dst_id: str, tag: str) -> None:
|
|
self.edges.append((src_id, dst_id, tag))
|
|
|
|
def merkle_root(self) -> str:
|
|
import hashlib
|
|
items = []
|
|
for n in self.nodes:
|
|
items.append(str(hash(n)).encode())
|
|
if not items:
|
|
return ""
|
|
level = [hashlib.sha256(i).hexdigest() for i in items]
|
|
while len(level) > 1:
|
|
next_level = []
|
|
for i in range(0, len(level), 2):
|
|
left = level[i]
|
|
right = level[i + 1] if i + 1 < len(level) else left
|
|
next_level.append(hashlib.sha256((left + right).encode()).hexdigest())
|
|
level = next_level
|
|
return level[0]
|