feat: Implement Aevo Perpetual Connector

hummingbot/connector/derivative/aevo_perpetual/__init__.py

@@ -0,0 +1,5 @@
+from .aevo_perpetual_derivative import AevoPerpetualDerivative
+
+__all__ = [
+    "AevoPerpetualDerivative",
+]

hummingbot/connector/derivative/aevo_perpetual/aevo_perpetual_api_order_book_data_source.py

@@ -0,0 +1,140 @@
+import asyncio
+from typing import Any, Dict, List, Optional
+from hummingbot.core.data_type.order_book_message import OrderBookMessage
+from hummingbot.core.data_type.order_book import OrderBook
+from hummingbot.connector.derivative.aevo_perpetual import aevo_perpetual_constants as CONSTANTS
+from hummingbot.connector.derivative.aevo_perpetual import aevo_perpetual_utils as utils
+
+from hummingbot.core.api_throttler.async_throttler import AsyncThrottler
+from hummingbot.connector.derivative.aevo_perpetual.aevo_perpetual_auth import AevoPerpetualAuth
+from hummingbot.core.data_type.order_book_tracker_data_source import OrderBookTrackerDataSource
+
+class AevoPerpetualAPIOrderBookDataSource(OrderBookTrackerDataSource):
+    def __init__(self,
+                 trading_pairs: List[str],
+                 domain: str = "aevo",
+                 api_factory: Optional[Any] = None,
+                 throttler: Optional[AsyncThrottler] = None,
+                 time_synchronizer: Optional[Any] = None):
+        super().__init__(trading_pairs)
+        self._domain = domain
+        self._throttler = throttler
+        self._api_factory = api_factory
+        self._time_synchronizer = time_synchronizer
+
+    async def get_last_traded_prices(self, trading_pairs: List[str], domain: Optional[str] = None) -> Dict[str, float]:
+        return await self._get_last_traded_prices(trading_pairs)
+
+    async def _get_last_traded_prices(self, trading_pairs: List[str]) -> Dict[str, float]:
+        res = await self._api_factory.call_rest(
+            method="GET",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.TICKER_PATH_URL}"
+        )
+        # Aevo returns list of tickers. Map 'instrument_name' to price.
+        # Example response: [{"instrument_name": "ETH-PERP", "mark_price": "2000.5", ...}, ...]
+        results = {}
+        for market in res:
+            name = market.get("instrument_name", "") # e.g., ETH-PERP
+            hb_name = utils.convert_to_hb_symbol(name)
+            if "mark_price" in market:
+                results[hb_name] = float(market["mark_price"])
+        return results
+
+    async def get_new_order_book(self, trading_pair: str) -> OrderBook:
+        # Aevo Orderbook Endpoint: /order_book?instrument_name=...
+        exchange_symbol = utils.convert_to_exchange_symbol(trading_pair)
+        params = {"instrument_name": exchange_symbol}
+        snapshot = await self._api_factory.call_rest(
+            method="GET",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.SNAPSHOT_PATH_URL}",
+            params=params
+        )
+        # Snapshot structure: {"bids": [[price, size], ...], "asks": ...}
+        # Convert to OrderBookMessage or OrderBook object
+        # Note: Hummingbot expects specific mapping, usually handled by message parser.
+        # For now, we return the raw snapshot or an OrderBook object depending on base class calc.
+        # Check base class: OrderBookTrackerDataSource usually returns OrderBookMessage from snapshot.
+        from hummingbot.core.data_type.order_book_message import OrderBookMessage, OrderBookMessageType
+
+        timestamp = snapshot.get("timestamp", self._time_synchronizer.time() * 1e9)
+        return OrderBookMessage(
+            OrderBookMessageType.SNAPSHOT,
+            {
+                "trading_pair": trading_pair,
+                "update_id": int(timestamp),
+                "bids": snapshot.get("bids", []),
+                "asks": snapshot.get("asks", [])
+            },
+            timestamp=timestamp * 1e-9
+        )
+
+
+    async def listen_for_subscriptions(self):
+        ws = None
+        while True:
+            try:
+                ws = await self._api_factory.get_ws_connection(CONSTANTS.AEVO_WS_URL)
+                await ws.connect()
+
+                # Subscribe to channels
+                for pair in self._trading_pairs:
+                    subscribe_request = {
+                        "op": "subscribe",
+                        "data": [
+                            f"{CONSTANTS.WS_TOPIC_ORDERBOOK}:{pair}",
+                            f"{CONSTANTS.WS_TOPIC_TRADES}:{pair}"
+                        ]
+                    }
+                    await ws.send_json(subscribe_request)
+
+                async for msg in ws.iter_messages():
+                    if msg.data:
+                        data = msg.json()
+                        channel = data.get("channel")
+
+                        if channel and channel.startswith(CONSTANTS.WS_TOPIC_ORDERBOOK):
+                            # Parse Order Book Snapshot/Update
+                            # Aevo sends full snapshots or updates. Assuming snapshot for simplicity or parsing both via same logic if format aligns
+                            payload = data.get("data", {})
+                            if payload.get("type") == "snapshot":
+                                order_book_message = OrderBookMessage(
+                                    OrderBookMessageType.SNAPSHOT,
+                                    {
+                                        "trading_pair": channel.split(":")[-1],
+                                        "update_id": int(payload.get("timestamp", self._time_synchronizer.time() * 1e9)),
+                                        "bids": payload.get("bids", []),
+                                        "asks": payload.get("asks", [])
+                                    },
+                                    timestamp=payload.get("timestamp", self._time_synchronizer.time() * 1e9) * 1e-9
+                                )
+                                self._message_queue.put_nowait(order_book_message)
+
+                        elif channel and channel.startswith(CONSTANTS.WS_TOPIC_TRADES):
+                            # Parse Trades
+                            payload = data.get("data", [])
+                            # Payload might be a list of trades
+                            for trade in payload:
+                                trade_msg = OrderBookMessage(
+                                    OrderBookMessageType.TRADE,
+                                    {
+                                        "trading_pair": channel.split(":")[-1],
+                                        "trade_type": 1.0 if trade.get("side", "").lower() == "buy" else 2.0,
+                                        "trade_id": trade.get("trade_id"),
+                                        "update_id": int(trade.get("timestamp", 0)),
+                                        "price": trade.get("price"),
+                                        "amount": trade.get("amount")
+                                    },
+                                    timestamp=trade.get("timestamp", self._time_synchronizer.time() * 1e9) * 1e-9
+                                )
+                                self._message_queue.put_nowait(trade_msg)
+
+            except asyncio.CancelledError:
+                raise
+            except Exception as e:
+                # Log error and reconnect
+                self.logger().error(f"WS Error: {e}", exc_info=True)
+                await asyncio.sleep(5)
+            finally:
+                if ws:
+                    await ws.disconnect()
+

hummingbot/connector/derivative/aevo_perpetual/aevo_perpetual_auth.py

@@ -0,0 +1,80 @@
+import hmac
+import hashlib
+import time
+import json
+from typing import Dict, Any, Optional
+from hummingbot.connector.time_synchronizer import TimeSynchronizer
+from hummingbot.core.web_assistant.auth import AuthBase
+from hummingbot.core.web_assistant.connections.data_types import RESTRequest, WSRequest
+
+class AevoPerpetualAuth(AuthBase):
+    def __init__(self, api_key: str, api_secret: str, time_provider: TimeSynchronizer):
+        self.api_key = api_key
+        self.api_secret = api_secret
+        self.time_provider = time_provider
+
+    async def rest_authenticate(self, request: RESTRequest) -> RESTRequest:
+        headers = {}
+        if request.headers:
+            headers.update(request.headers)
+
+        # Precision is Key! Aevo expects nanoseconds.
+        # Ensure we multiply by 1e9 and cast to int before string.
+        timestamp = str(int(self.time_provider.time() * 1e9))
+        signature = self._generate_signature(timestamp, request.method, request.url, request.data)
+
+        headers.update({
+            "AEVO-ACCESS-KEY": self.api_key,
+            "AEVO-ACCESS-SIG": signature,
+            "AEVO-ACCESS-TIMESTAMP": timestamp,
+        })
+        request.headers = headers
+        return request
+
+    async def ws_authenticate(self, request: WSRequest) -> WSRequest:
+        return request  # Websocket auth often handled differently, checking docs
+
+    def _generate_signature(self, timestamp: str, method: str, url: str, data: Optional[Dict[str, Any]]) -> str:
+        # Aevo specific signature generation (check docs for exact format)
+        # Typically: HMAC-SHA256(secret, timestamp + method + path + body)
+        # Aevo expects: timestamp + method + path + body
+        # URL parsing to extract path + query
+        path = url
+        if "https://" in url:
+             path = "/" + url.split("https://")[-1].split("/", 1)[-1]
+        elif "http://" in url:
+             path = "/" + url.split("http://")[-1].split("/", 1)[-1]
+
+        payload = f"{timestamp}{method.upper()}{path}"
+        if data:
+            payload += json.dumps(data, separators=(',', ':'))
+
+        return hmac.new(
+            self.api_secret.encode("utf-8"),
+            payload.encode("utf-8"),
+            hashlib.sha256
+        ).hexdigest()
+
+    def get_ws_auth_payload(self) -> Dict[str, Any]:
+        # WebSocket Auth (Standard Aevo Pattern)
+        timestamp = str(int(self.time_provider.time() * 1e9))
+
+        # Signing value often differs for WS. 
+        # Checking similar exchanges, often it's just timestamp or specific string.
+        # For Aevo, let's assume it signs the timestamp similar to REST but without method/url.
+        # If docs say otherwise, we adjust. 
+        # Payload for sig: timestamp
+        signature = hmac.new(
+            self.api_secret.encode("utf-8"),
+            timestamp.encode("utf-8"),
+            hashlib.sha256
+        ).hexdigest()
+
+        return {
+            "op": "auth",
+            "data": {
+                "key": self.api_key,
+                "sig": signature,
+                "timestamp": timestamp
+            }
+        }

hummingbot/connector/derivative/aevo_perpetual/aevo_perpetual_constants.py

@@ -0,0 +1,29 @@
+from hummingbot.core.api_throttler.data_types import RateLimit
+
+AEVO_BASE_URL = "https://api.aevo.xyz"
+AEVO_WS_URL = "wss://ws.aevo.xyz"
+
+# Public Endpoints
+SNAPSHOT_PATH_URL = "/order_book"
+TICKER_PATH_URL = "/markets"
+INSTRUMENT_PATH_URL = "/markets"
+TRADES_PATH_URL = "/trades"
+
+# Private Endpoints
+ORDER_PATH_URL = "/orders"
+ACCOUNT_PATH_URL = "/account"
+POSITIONS_PATH_URL = "/positions"
+
+# Websocket Topics
+WS_TOPIC_ORDERBOOK = "orderbook"
+WS_TOPIC_TRADES = "trades"
+WS_TOPIC_TICKER = "ticker"
+
+# Rate Limits
+RATE_LIMITS = [
+    RateLimit(limit_id=SNAPSHOT_PATH_URL, limit=100, time_interval=10),
+    RateLimit(limit_id=TICKER_PATH_URL, limit=100, time_interval=10),
+    RateLimit(limit_id=TRADES_PATH_URL, limit=100, time_interval=10),
+    RateLimit(limit_id=ORDER_PATH_URL, limit=50, time_interval=10),
+    RateLimit(limit_id=ACCOUNT_PATH_URL, limit=50, time_interval=10),
+]

hummingbot/connector/derivative/aevo_perpetual/aevo_perpetual_derivative.py

@@ -0,0 +1,130 @@
+from typing import Tuple, Any, Dict, Optional, List
+from hummingbot.connector.derivative.derivative_base import DerivativeBase
+from hummingbot.connector.derivative.aevo_perpetual import aevo_perpetual_constants as CONSTANTS
+from hummingbot.connector.derivative.aevo_perpetual.aevo_perpetual_api_order_book_data_source import AevoPerpetualAPIOrderBookDataSource
+from hummingbot.connector.derivative.aevo_perpetual.aevo_perpetual_user_stream_data_source import AevoPerpetualUserStreamDataSource
+from hummingbot.connector.derivative.aevo_perpetual import aevo_perpetual_utils as utils
+from hummingbot.connector.derivative.aevo_perpetual.aevo_perpetual_auth import AevoPerpetualAuth
+from hummingbot.core.web_assistant.web_assistants_factory import WebAssistantsFactory
+from hummingbot.core.web_assistant.connections.data_types import RESTMethod
+from hummingbot.core.data_type.common import OrderType
+import asyncio
+
+
+class AevoPerpetualDerivative(DerivativeBase):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._domain = "aevo"
+        self._auth: AevoPerpetualAuth = None
+        self._web_assistants_factory: WebAssistantsFactory = None
+
+    @property
+    def name(self) -> str:
+        return "aevo_perpetual"
+
+    @property
+    def supported_order_types(self) -> List[OrderType]:
+        return [OrderType.LIMIT, OrderType.MARKET]
+
+    def _create_web_assistants_factory(self) -> WebAssistantsFactory:
+        return WebAssistantsFactory(
+            throttler=self._throttler,
+            auth=self._auth
+        )
+
+    async def _make_trading_rules_request(self) -> Any:
+        return await self._api_factory.call_rest(
+            method="GET",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.INSTRUMENT_PATH_URL}"
+        )
+
+    async def _make_trading_pairs_request(self) -> Any:
+        return await self._api_factory.call_rest(
+            method="GET",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.INSTRUMENT_PATH_URL}"
+        )
+
+    @property
+    def authenticator(self):
+        return self._auth
+
+    @property
+    def rate_limits_rules(self):
+        return CONSTANTS.RATE_LIMITS
+
+    async def start_network(self):
+        await self._stop_network()
+        self._stop_network_task = asyncio.create_task(self._start_network())
+
+    def _create_order_book_data_source(self):
+        return AevoPerpetualAPIOrderBookDataSource(
+            trading_pairs=self._trading_pairs,
+            domain=self._domain,
+            api_factory=self._web_assistants_factory,
+            throttler=self._throttler,
+            time_synchronizer=self._time_synchronizer)
+
+    def _create_user_stream_data_source(self):
+        return AevoPerpetualUserStreamDataSource(
+            auth=self._auth,
+            trading_pairs=self._trading_pairs,
+            api_factory=self._web_assistants_factory,
+            domain=self._domain)
+
+    async def _start_network(self):
+        self._order_book_tracker.start()
+        self._user_stream_tracker.start()
+        self._status_polling_task = asyncio.create_task(self._status_polling_loop())
+
+    async def _stop_network(self):
+        self._order_book_tracker.stop()
+        self._user_stream_tracker.stop()
+        if self._status_polling_task is not None:
+            self._status_polling_task.cancel()
+
+
+    async def _place_order(self,
+                           order_id: str,
+                           trading_pair: str,
+                           amount: float,
+                           trade_type: str,
+                           order_type: str,
+                           price: float,
+                           **kwargs) -> Tuple[str, float]:
+        exchange_symbol = utils.convert_to_exchange_symbol(trading_pair)
+        params = {
+            "instrument_name": exchange_symbol,
+            "is_buy": trade_type.upper() == "BUY",
+            "limit_price": str(price),
+            "quantity": str(amount),
+            "post_only": kwargs.get("post_only", False),
+            "reduce_only": kwargs.get("reduce_only", False),
+            "time_in_force": kwargs.get("time_in_force", "GTC"),
+            # "client_order_id": order_id 
+            # NOTE: client_order_id support in Aevo docs is sparse.
+            # Keeping it commented out until verified with live keys. 
+        }
+
+        # Determine endpoint based on order type if needed, or just standard /orders
+        response = await self._api_factory.call_rest(
+            method="POST",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.ORDER_PATH_URL}",
+            data=params,
+            is_auth_required=True
+        )
+
+        # Parse response to get exchange order ID and timestamp
+        exchange_order_id = str(response.get("order_id", order_id))
+        transact_time = float(response.get("timestamp", self._time_synchronizer.time() * 1e9)) * 1e-9
+
+        return exchange_order_id, transact_time
+
+    async def _cancel_order(self, order_id: str, trading_pair: str, timestamp: float) -> Any:
+        # Aevo Cancel: DELETE /orders/{order_id}
+        response = await self._api_factory.call_rest(
+            method="DELETE",
+            url=f"{CONSTANTS.AEVO_BASE_URL}{CONSTANTS.ORDER_PATH_URL}/{order_id}",
+            is_auth_required=True
+        )
+        return response
+