68 lines
2.7 KiB
Python
68 lines
2.7 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
|
|
|
|
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) -> 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), getattr(signals, "version", 0))
|
|
if hasattr(problem, "version") or hasattr(signals, "version")
|
|
else 1,
|
|
"objects": {
|
|
"LocalArbProblem": problem.to_dict(),
|
|
"SharedSignals": {
|
|
"version": signals.version,
|
|
"price_delta_by_asset": dict(signals.price_delta_by_asset),
|
|
"cross_corr": dict(signals.cross_corr),
|
|
"liquidity_estimates": dict(signals.liquidity_estimates),
|
|
},
|
|
},
|
|
"morphisms": {
|
|
"SharedSignals": {
|
|
"version": signals.version,
|
|
# shallow representation; real implementations may include attestations
|
|
"connections": list(signals.cross_corr.keys()),
|
|
}
|
|
},
|
|
"plan_delta_ref": None, # linked PlanDelta would be populated at runtime
|
|
}
|
|
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", {}),
|
|
}
|