69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
"""Adapters: MVP wiring skeleton for Phase 0 MercuryMesh.
|
|
|
|
This module provides tiny, well-scoped adapters that stand in for
|
|
- ExchangeDataFeedAdapter
|
|
- VenueBrokerAdapter
|
|
and a small helper to demonstrate a Phase 0 end-to-end delta aggregation
|
|
workflow using the existing FederatedCoordinator.
|
|
|
|
These are deliberately minimal and safe stubs intended for educational
|
|
labs and integration testing. They do not perform real network IO and are
|
|
fully compatible with the in-repo primitives in core.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Dict, List
|
|
|
|
from .core import MarketStateSnapshot, PlanDelta
|
|
from .core import FederatedCoordinator
|
|
|
|
|
|
class ExchangeDataFeedAdapter:
|
|
"""Stub adapter simulating a data feed from an exchange."""
|
|
|
|
def __init__(self, name: str = "exchange") -> None:
|
|
self.name = name
|
|
|
|
def connect(self) -> bool:
|
|
# In a real implementation, establish TLS session, authenticate, etc.
|
|
return True
|
|
|
|
def fetch_signals(self) -> List[Dict[str, float]]:
|
|
# Placeholder: return no signals by default.
|
|
return []
|
|
|
|
|
|
class VenueBrokerAdapter:
|
|
"""Stub adapter simulating a venue broker connection."""
|
|
|
|
def __init__(self, venue: str = "venue") -> None:
|
|
self.venue = venue
|
|
|
|
def connect(self) -> bool:
|
|
return True
|
|
|
|
def send_delta(self, delta: PlanDelta) -> None:
|
|
# In a real system, this would transmit delta to the venue.
|
|
# Here we simply acknowledge the action for the interface.
|
|
pass
|
|
|
|
|
|
def phase0_demo(adapter_a: ExchangeDataFeedAdapter, adapter_b: ExchangeDataFeedAdapter) -> PlanDelta:
|
|
"""Demonstrate a tiny Phase 0 end-to-end delta aggregation.
|
|
|
|
- Collect signals from two adapters (could be empty in the MVP).
|
|
- Run the FederatedCoordinator to compute an aggregate delta.
|
|
- Return the resulting PlanDelta.
|
|
This is intentionally simple and side-effect free.
|
|
"""
|
|
signals: List[Dict[str, float]] = []
|
|
signals.extend(adapter_a.fetch_signals())
|
|
signals.extend(adapter_b.fetch_signals())
|
|
|
|
delta = FederatedCoordinator.aggregate_signals(signals)
|
|
return delta
|
|
|
|
# Public API surface for MVP wiring exploration
|
|
__all__ = ["ExchangeDataFeedAdapter", "VenueBrokerAdapter", "phase0_demo"]
|