Spaces:
Runtime error
Runtime error
File size: 2,298 Bytes
d5b7ee9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Adapter registry — discovers and instantiates trading adapters."""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from trading_cli.execution.adapters.base import TradingAdapter
if TYPE_CHECKING:
pass
logger = logging.getLogger(__name__)
# Global registry of available adapters
_ADAPTERS: dict[str, type[TradingAdapter]] = {}
def register_adapter(adapter_class: type[TradingAdapter]) -> type[TradingAdapter]:
"""Decorator to register an adapter class.
Usage:
@register_adapter
class AlpacaAdapter(TradingAdapter):
...
"""
# Instantiate temporarily to get adapter_id
# We assume adapter_id is a class property or can be called without args
try:
instance = adapter_class.__new__(adapter_class)
adapter_id = adapter_class.adapter_id.fget(instance) if hasattr(adapter_class.adapter_id, 'fget') else getattr(adapter_class, 'adapter_id', None)
if adapter_id:
_ADAPTERS[adapter_id] = adapter_class
logger.debug("Registered adapter: %s", adapter_id)
except Exception:
# Fallback: use class name lowercase
adapter_id = adapter_class.__name__.lower().replace("adapter", "")
_ADAPTERS[adapter_id] = adapter_class
logger.debug("Registered adapter (fallback): %s", adapter_id)
return adapter_class
def get_adapter(adapter_id: str) -> type[TradingAdapter] | None:
"""Get adapter class by ID."""
return _ADAPTERS.get(adapter_id)
def list_adapters() -> list[str]:
"""List all registered adapter IDs."""
return list(_ADAPTERS.keys())
def create_adapter(adapter_id: str, config: dict) -> TradingAdapter:
"""Create an adapter instance from config.
Args:
adapter_id: Adapter identifier ('alpaca', 'binance', 'kraken', 'demo').
config: Configuration dict with API keys and settings.
Returns:
TradingAdapter instance.
Raises:
ValueError: If adapter_id is not registered.
"""
adapter_class = get_adapter(adapter_id)
if adapter_class is None:
available = list_adapters()
raise ValueError(
f"Unknown adapter: '{adapter_id}'. "
f"Available adapters: {available}"
)
return adapter_class(config)
|