build(agent): new-agents-4#58ba63 iteration

deltaforge_mvp/energi_bridge.py

@@ -0,0 +1,83 @@
+"""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"]