101 lines
3.4 KiB
Python
101 lines
3.4 KiB
Python
"""
|
|
Adapter Blueprint for MercuryMesh MVP
|
|
This module provides a minimal, well-documented skeleton that teams can copy
|
|
when implementing new venue adapters. It is intentionally simple and self-contained
|
|
so it can be dropped into new repos or integrated into the existing codebase with
|
|
minimal friction.
|
|
|
|
The blueprint focuses on three responsibilities:
|
|
- Connect to a venue feed (simulated or real)
|
|
- Extract raw venue data and normalize it into a common signal format
|
|
- Emit a Signal contract compatible object that downstream components can consume
|
|
|
|
Note: This is a blueprint, not a fully-featured adapter. It lacks networking,
|
|
security, and production-grade error handling which should be added in production use.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Any, Optional
|
|
import time
|
|
|
|
|
|
@dataclass
|
|
class LocalMarketContext:
|
|
venue_id: str
|
|
symbol: str
|
|
timeframe: str
|
|
|
|
|
|
@dataclass
|
|
class MarketSignal:
|
|
venue_id: str
|
|
symbol: str
|
|
timestamp: float
|
|
features: Dict[str, float]
|
|
|
|
|
|
class AdapterBlueprint:
|
|
"""Minimal adapter blueprint.
|
|
|
|
- venue_id: identifier for the venue this adapter connects to
|
|
- registry: optional contract/GoC registry payload (not required for the blueprint)
|
|
"""
|
|
|
|
def __init__(self, venue_id: str, registry: Optional[Dict[str, Any]] = None) -> None:
|
|
self.venue_id = venue_id
|
|
self.registry = registry or {}
|
|
self.connected = False
|
|
|
|
def connect(self) -> bool:
|
|
"""Simulate a connection to the venue feed.
|
|
|
|
In a real adapter this would establish a network connection, authenticate, etc.
|
|
For the blueprint, we simply mark as connected and return True.
|
|
"""
|
|
# In a real implementation, add try/except and reconnection logic
|
|
self.connected = True
|
|
return self.connected
|
|
|
|
def fetch_raw(self) -> Any:
|
|
"""Fetch a raw payload from the venue feed (simulated).
|
|
|
|
Returns a simple dict that a concrete subclass would parse.
|
|
"""
|
|
if not self.connected:
|
|
raise RuntimeError("Adapter not connected to venue feed")
|
|
# Simulated raw payload; in practice this would be a socket/HTTP feed, etc.
|
|
now = time.time()
|
|
return {
|
|
"venue": self.venue_id,
|
|
"symbol": "XYZ",
|
|
"ts": now,
|
|
"raw_features": {
|
|
"liquidity_proxy": 0.9,
|
|
"order_flow_intensity": 0.6,
|
|
"volatility_proxy": 1.1,
|
|
},
|
|
}
|
|
|
|
def extract_signal(self, raw: Any) -> MarketSignal:
|
|
"""Normalize and map raw payload to a MarketSignal contract."""
|
|
venue = str(raw.get("venue", self.venue_id))
|
|
symbol = str(raw.get("symbol", "UNKNOWN"))
|
|
timestamp = float(raw.get("ts", time.time()))
|
|
features = dict(raw.get("raw_features", {}))
|
|
return MarketSignal(venue_id=venue, symbol=symbol, timestamp=timestamp, features=features)
|
|
|
|
def emit(self, signal: MarketSignal) -> MarketSignal:
|
|
"""Hook for downstream consumers. In a real adapter this might publish to a bus."""
|
|
# For blueprint purposes, simply return the signal to show it was produced
|
|
return signal
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Simple demonstration when run as a script
|
|
adapter = AdapterBlueprint(venue_id="venue_example")
|
|
adapter.connect()
|
|
raw = adapter.fetch_raw()
|
|
sig = adapter.extract_signal(raw)
|
|
print("Blueprint adapter produced signal:", sig)
|