diff --git a/deltaforge_mvp/energi_bridge.py b/deltaforge_mvp/energi_bridge.py new file mode 100644 index 0000000..f0a4787 --- /dev/null +++ b/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"]