66 lines
2.6 KiB
Python
66 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
"""EnergiBridge-inspired interoperability bridge for CrossVenueArbX primitives.
|
|
|
|
This module provides a tiny, production-light translation layer that maps
|
|
CrossVenueArbX primitives (LocalArbProblem, SharedSignals, PlanDelta, etc.)
|
|
into a vendor-agnostic, CatOpt-like intermediate representation (IR).
|
|
|
|
Rationale:
|
|
- Enables interoperability with downstream systems that expect a canonical IR
|
|
- Keeps changes small and isolated to a single bridge module
|
|
- Supports deterministic replay semantics by producing stable, versioned IR blocks
|
|
"""
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
from .core import LocalArbProblem, SharedSignals, PlanDelta
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Bridge utility to convert CrossVenueArbX primitives to a CatOpt-like IR."""
|
|
|
|
@staticmethod
|
|
def to_catopt(
|
|
problem: LocalArbProblem,
|
|
signals: SharedSignals,
|
|
plan_delta: Optional[PlanDelta] = None,
|
|
) -> Dict[str, Any]:
|
|
"""Convert a LocalArbProblem and its associated SharedSignals into a
|
|
canonical, vendor-agnostic representation.
|
|
|
|
The output is a simple nested dict structure with clear segregation of
|
|
objects and morphisms, suitable for transport or logging in cross-venue
|
|
systems. This lightweight representation is intentionally simple to keep
|
|
the MVP lean while still providing a predictable schema.
|
|
"""
|
|
catopt: Dict[str, Any] = {
|
|
"version": max(getattr(problem, "version", 0), signals.version, 1),
|
|
"contract_id": problem.fingerprint(),
|
|
"objects": {
|
|
"LocalArbProblem": {**problem.to_dict(), "fingerprint": problem.fingerprint()},
|
|
"SharedSignals": {**signals.to_dict(), "fingerprint": signals.fingerprint()},
|
|
},
|
|
"morphisms": {
|
|
"SharedSignals": {
|
|
"version": signals.version,
|
|
"connections": sorted((repr(key) for key in signals.cross_corr.keys())),
|
|
}
|
|
},
|
|
"plan_delta_ref": plan_delta.to_dict() if plan_delta is not None else None,
|
|
}
|
|
return catopt
|
|
|
|
@staticmethod
|
|
def from_catopt(catopt: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Inverse of to_catopt for debugging/demo purposes.
|
|
|
|
This is intentionally simple and returns the raw object-like payload for
|
|
downstream processing or logging.
|
|
"""
|
|
return {
|
|
"objects": catopt.get("objects", {}),
|
|
"morphisms": catopt.get("morphisms", {}),
|
|
"plan_delta_ref": catopt.get("plan_delta_ref"),
|
|
}
|