84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
"""EnergiBridge: tiny canonical bridge for cross-domain interoperability.
|
|
|
|
This module provides a minimal, production-friendly scaffolding that
|
|
captures the essence of the EnergiBridge concept described in the MVP
|
|
documentation. It is intentionally lightweight and does not change the
|
|
existing runtime paths of the demo/tests. The goal is to offer a clean,
|
|
extensible place for future adapters to map canonical DeltaForge primitives
|
|
to vendor-agnostic representations and vice-versa.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, List, Optional
|
|
|
|
from .core import Asset, MarketSignal, StrategyDelta, PlanDelta
|
|
|
|
|
|
@dataclass
|
|
class LocalProblem:
|
|
"""Canonical, per-site optimization task (a light stand-in for a local problem).
|
|
|
|
This is intentionally small: it captures the asset, a compact objective tag,
|
|
and the current market signal snapshot from a venue.
|
|
"""
|
|
asset: Asset
|
|
objective: str
|
|
signals: List[MarketSignal]
|
|
deadline: Optional[float] = None
|
|
|
|
|
|
@dataclass
|
|
class ContractDescriptor:
|
|
"""Descriptor for an adapter contract in the Graph-of-Contracts registry."""
|
|
adapter_name: str
|
|
version: str
|
|
capabilities: List[str]
|
|
|
|
|
|
class GraphOfContracts:
|
|
"""Minimal in-memory registry for adapter contracts."""
|
|
|
|
def __init__(self) -> None:
|
|
self._registry: Dict[str, ContractDescriptor] = {}
|
|
|
|
def register(self, adapter_name: str, version: str, capabilities: List[str]) -> None:
|
|
self._registry[adapter_name] = ContractDescriptor(
|
|
adapter_name=adapter_name, version=version, capabilities=capabilities
|
|
)
|
|
|
|
def get(self, adapter_name: str) -> Optional[ContractDescriptor]:
|
|
return self._registry.get(adapter_name)
|
|
|
|
def list_all(self) -> List[ContractDescriptor]:
|
|
return list(self._registry.values())
|
|
|
|
|
|
class EnergiBridge:
|
|
"""Bridges canonical IR to adapter-specific signals in a minimal fashion.
|
|
|
|
The bridge exposes a small API surface suitable for MVPs:
|
|
- map_signals_to_local_problems: convert MarketSignal streams into LocalProblem
|
|
- plan_to_local_problems: convert a PlanDelta into envelope-ready LocalProblems
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.contracts = GraphOfContracts()
|
|
|
|
def map_signals_to_local_problems(self, signals: List[MarketSignal], objective: str = "delta-hedge") -> List[LocalProblem]:
|
|
"""Convert a list of market signals into LocalProblem instances."""
|
|
problems: List[LocalProblem] = []
|
|
for s in signals:
|
|
problems.append(LocalProblem(asset=s.asset, objective=objective, signals=[s]))
|
|
return problems
|
|
|
|
def plan_to_local_problems(self, plan: PlanDelta) -> List[LocalProblem]:
|
|
"""Convert a PlanDelta into a set of LocalProblems based on contained deltas."""
|
|
problems: List[LocalProblem] = []
|
|
for d in plan.deltas:
|
|
problems.append(LocalProblem(asset=d.asset, objective="delta-hedge", signals=[]))
|
|
return problems
|
|
|
|
|
|
__all__ = ["EnergiBridge", "LocalProblem", "GraphOfContracts", "ContractDescriptor"]
|