request instrument and handle ticker data

examples/live/bybit/bybit_request_custom_endpoint.py

@@ -22,18 +22,20 @@
 from nautilus_trader.adapters.bybit.config import BybitExecClientConfig
 from nautilus_trader.adapters.bybit.factories import BybitLiveDataClientFactory
 from nautilus_trader.adapters.bybit.factories import BybitLiveExecClientFactory
-from nautilus_trader.common import Environment
+from nautilus_trader.adapters.bybit.schemas.market.ticker import BybitTickerData
 from nautilus_trader.common.clock import TimeEvent
+
+from nautilus_trader.common import Environment
 from nautilus_trader.config import InstrumentProviderConfig
 from nautilus_trader.config import LiveExecEngineConfig
 from nautilus_trader.config import LoggingConfig
 from nautilus_trader.config import StrategyConfig
 from nautilus_trader.config import TradingNodeConfig
-from nautilus_trader.core.message import Request
-from nautilus_trader.core.nautilus_pyo3 import InstrumentId
+from nautilus_trader.core.data import Data
 from nautilus_trader.core.uuid import UUID4
-from nautilus_trader.data.messages import DataResponse
 from nautilus_trader.live.node import TradingNode
+from nautilus_trader.model.data import DataType
+from nautilus_trader.model.identifiers import ClientId, InstrumentId
 from nautilus_trader.trading import Strategy
 
 
@@ -49,6 +51,7 @@ class RequestDemoStrategyConfig(StrategyConfig, frozen=True):
     interval: int
 
 
+
 class RequestDemoStrategy(Strategy):
     """
     Strategy showcases how to request data from a custom bybit endpoint. Some adapter
@@ -68,35 +71,28 @@ def __init__(self, config: RequestDemoStrategyConfig):
         self.instrument_id = InstrumentId.from_str(config.instrument_id)
         self.linear_request_ticker_uuid = UUID4()
 
-    def start(self):
+    def on_start(self):
         seconds_delta = timedelta(seconds=self.interval)
-        # self.clock.set_timer(
-        #     name="fetch_ticker",
-        #     interval=seconds_delta,
-        #     callback=self.send_tickers_request,
-        # )
+        self.clock.set_timer(
+            name="fetch_ticker",
+            interval=seconds_delta,
+            callback=self.send_tickers_request,
+        )
 
     def send_tickers_request(self, time_event: TimeEvent):
-        request = Request(
-            request_id=self.linear_request_ticker_uuid,
-            ts_init=self.clock.timestamp_ns(),
-            callback=self.on_data,
-            metadata={"symbol": self.instrument_id.symbol},
+        data_type = DataType(
+            BybitTickerData,
+            metadata={"symbol":self.instrument_id.symbol}
         )
-        self.msgbus.request(endpoint="bybit.data.tickers", request=request)
+        self.request_data(data_type, ClientId("BYBIT"))
 
-    def on_data(self, data: DataResponse):
-        ## check generic data response by uuid
-        if data.correlation_id == self.linear_request_ticker_uuid:
-            tickers = data.data
-            for ticker in tickers:
-                self.log.info(f"{ticker}")
+    def on_historical_data(self, data: Data):
+        if isinstance(data,BybitTickerData):
+            self.log.info(f"{data}")
 
 
-# bybit_api_key = os.getenv("BYBIT_API_KEY", None)
-# bybit_api_key = "puoVYU45dIfelFgOon"
-# bybit_api_secret = os.getenv("BYBIT_API_SECRET", None)
-# bybit_api_secret = "b1qY5GDzPR9RgcQvbnHhIT5W2iWmqTJJSRvT"
+api_key = os.getenv("BYBIT_API_KEY")
+api_secret = os.getenv("BYBIT_API_SECRET")
 
 config_node = TradingNodeConfig(
     trader_id="TESTER-001",
@@ -108,17 +104,17 @@ def on_data(self, data: DataResponse):
     ),
     data_clients={
         "BYBIT": BybitDataClientConfig(
-            api_key=None,
-            api_secret=None,
+            api_key=api_key,
+            api_secret=api_secret,
             instrument_types=[BybitInstrumentType.LINEAR],
             instrument_provider=InstrumentProviderConfig(load_all=True),
             testnet=True,
         ),
     },
     exec_clients={
         "BYBIT": BybitExecClientConfig(
-            api_key=None,
-            api_secret=None,
+            api_key=api_key,
+            api_secret=api_secret,
             instrument_types=[BybitInstrumentType.LINEAR],
             instrument_provider=InstrumentProviderConfig(load_all=True),
             testnet=True,

nautilus_trader/adapters/bybit/common/enums.py

@@ -16,6 +16,7 @@
 from enum import Enum
 from enum import unique
 
+from nautilus_trader.core.nautilus_pyo3 import PositionSide
 from nautilus_trader.model.data import BarType
 from nautilus_trader.model.enums import BarAggregation
 from nautilus_trader.model.enums import OrderSide
@@ -38,6 +39,18 @@ class BybitPositionIdx(Enum):
     SELL_HEDGE = 2
 
 
+@unique
+class BybitPositionSide(Enum):
+    BUY = 'Buy'
+    SELL = 'Sell'
+
+    def parse_to_position_side(self) -> PositionSide:
+        if self == BybitPositionSide.BUY:
+            return PositionSide.LONG
+        elif self == BybitPositionSide.SELL:
+            return PositionSide.SHORT
+
+
 @unique
 class BybitKlineInterval(Enum):
     MINUTE_1 = "1"

nautilus_trader/adapters/bybit/data.py

@@ -24,6 +24,7 @@
 from nautilus_trader.adapters.bybit.config import BybitDataClientConfig
 from nautilus_trader.adapters.bybit.http.client import BybitHttpClient
 from nautilus_trader.adapters.bybit.http.market import BybitMarketHttpAPI
+from nautilus_trader.adapters.bybit.schemas.market.ticker import BybitTickerData
 from nautilus_trader.adapters.bybit.schemas.symbol import BybitSymbol
 from nautilus_trader.adapters.bybit.schemas.ws import BybitWsMessageGeneral
 from nautilus_trader.adapters.bybit.schemas.ws import decoder_ws_ticker
@@ -48,8 +49,9 @@
 from nautilus_trader.model.data import GenericData
 from nautilus_trader.model.data import TradeTick
 from nautilus_trader.model.enums import PriceType
-from nautilus_trader.model.identifiers import ClientId
+from nautilus_trader.model.identifiers import ClientId, Venue
 from nautilus_trader.model.identifiers import InstrumentId
+from nautilus_trader.model.instruments import Instrument
 
 
 class BybitDataClient(LiveMarketDataClient):
@@ -247,6 +249,38 @@ def _get_cached_instrument_id(self, symbol: str) -> InstrumentId:
         nautilus_instrument_id: InstrumentId = bybit_symbol.parse_as_nautilus()
         return nautilus_instrument_id
 
+    async def _request_instrument(self, instrument_id: InstrumentId, correlation_id: UUID4) -> None:
+        instrument: Instrument | None = self._instrument_provider.find(instrument_id)
+        if instrument is None:
+            self._log.error(f"Cannot find instrument for {instrument_id}.")
+            return
+        data_type = DataType(
+            type=Instrument,
+            metadata={"instrument_id": instrument_id},
+        )
+        self._handle_data_response(
+            data_type=data_type,
+            data=instrument,
+            correlation_id=correlation_id,
+        )
+    async def _request_instruments(self, venue: Venue, correlation_id: UUID4) -> None:
+        all_instruments = self._instrument_provider.get_all()
+        target_instruments = []
+        for instrument in all_instruments.values():
+            if instrument.venue == venue:
+                target_instruments.append(instrument)
+        data_type = DataType(
+            type=Instrument,
+            metadata={"venue": venue},
+        )
+        self._handle_data_response(
+            data_type=data_type,
+            data=target_instruments,
+            correlation_id=correlation_id,
+        )
+
+
+
     async def _request_bars(
         self,
         bar_type: BarType,
@@ -306,3 +340,41 @@ async def _disconnect(self) -> None:
             self._update_instruments_task = None
         for instrument_type, ws_client in self._ws_clients.items():
             await ws_client.disconnect()
+
+    async def _handle_ticker_data_request(self,symbol: Symbol, correlation_id: UUID4) -> None:
+        bybit_symbol = BybitSymbol(symbol.value)
+        bybit_tickers = await self._http_market.fetch_tickers(
+            instrument_type=bybit_symbol.instrument_type,
+            symbol=bybit_symbol.raw_symbol,
+        )
+        data_type = DataType(
+            type=BybitTickerData,
+            metadata={"symbol":symbol},
+        )
+        result = []
+        for ticker in bybit_tickers:
+            ticker_data: BybitTickerData = BybitTickerData(
+                symbol=ticker.symbol,
+                bid1Price=ticker.bid1Price,
+                bid1Size=ticker.bid1Size,
+                ask1Price=ticker.ask1Price,
+                ask1Size=ticker.ask1Size,
+                lastPrice=ticker.lastPrice,
+                highPrice24h=ticker.highPrice24h,
+                lowPrice24h=ticker.lowPrice24h,
+                turnover24h=ticker.turnover24h,
+                volume24h=ticker.volume24h,
+            )
+            result.append(ticker_data)
+        self._handle_data_response(
+            data_type,
+            result,
+            correlation_id,
+        )
+
+    async def _request(self, data_type: DataType, correlation_id: UUID4) -> None:
+        if data_type.type == BybitTickerData:
+            symbol = data_type.metadata["symbol"]
+            await self._handle_ticker_data_request(symbol, correlation_id)
+
+

nautilus_trader/adapters/bybit/execution.py

@@ -154,6 +154,7 @@ async def _connect(self) -> None:
         # subscribe account updates
         await self._ws_client.subscribe_executions_update()
         await self._ws_client.subscribe_orders_update()
+        print("connect completed")
 
     async def generate_order_status_reports(
         self,
@@ -265,7 +266,13 @@ async def generate_position_status_reports(
                 position_report = position.parse_to_position_status_report(
                     account_id=self.account_id,
                     instrument_id=instrument_id,
+                    report_id=UUID4(),
+                    ts_init=self._clock.timestamp_ns(),
                 )
+                self._log.debug(f"Received {position_report}.")
+                reports.append(position_report)
+        return reports
+
 
 
     def _get_cache_active_symbols(self) -> set[str]:

nautilus_trader/adapters/bybit/schemas/market/ticker.py

@@ -16,6 +16,57 @@
 import msgspec
 
 from nautilus_trader.adapters.bybit.schemas.common import BybitListResult
+from nautilus_trader.core.data import Data
+
+
+class BybitTickerData(Data):
+    symbol: str
+    bid1Price: str
+    bid1Size: str
+    ask1Price: str
+    ask1Size: str
+    lastPrice: str
+    highPrice24h: str
+    lowPrice24h: str
+    turnover24h: str
+    volume24h: str
+
+    def __init__(
+        self,
+        symbol: str,
+        bid1Price: str,
+        bid1Size: str,
+        ask1Price: str,
+        ask1Size: str,
+        lastPrice: str,
+        highPrice24h: str,
+        lowPrice24h: str,
+        turnover24h: str,
+        volume24h: str,
+    ):
+        self.symbol = symbol
+        self.bid1Price = bid1Price
+        self.bid1Size = bid1Size
+        self.ask1Price = ask1Price
+        self.ask1Size = ask1Size
+        self.lastPrice = lastPrice
+        self.highPrice24h = highPrice24h
+        self.lowPrice24h = lowPrice24h
+        self.turnover24h = turnover24h
+        self.volume24h = volume24h
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}(" \
+               f"symbol={self.symbol!r}, " \
+               f"bid1Price={self.bid1Price!r}, " \
+               f"bid1Size={self.bid1Size!r}, " \
+               f"ask1Price={self.ask1Price!r}, " \
+               f"ask1Size={self.ask1Size!r}, " \
+               f"lastPrice={self.lastPrice!r}, " \
+               f"highPrice24h={self.highPrice24h!r}, " \
+               f"lowPrice24h={self.lowPrice24h!r}, " \
+               f"turnover24h={self.turnover24h!r}, " \
+               f"volume24h={self.volume24h!r})"
 
 
 class BybitTickerSpot(msgspec.Struct):

nautilus_trader/adapters/bybit/schemas/position.py

@@ -15,17 +15,20 @@
 
 import msgspec
 
+from nautilus_trader.adapters.bybit.common.enums import BybitPositionSide
 from nautilus_trader.adapters.bybit.schemas.common import BybitListResult
+from nautilus_trader.core.uuid import UUID4
 from nautilus_trader.execution.reports import PositionStatusReport
 from nautilus_trader.model.identifiers import AccountId, InstrumentId
+from nautilus_trader.model.objects import Quantity
 
 
 class BybitPositionStruct(msgspec.Struct):
     positionIdx: int
     riskId: int
     riskLimitValue: str
     symbol: str
-    side: str
+    side: BybitPositionSide
     size: str
     avgPrice: str
     positionValue: str
@@ -53,15 +56,20 @@ class BybitPositionStruct(msgspec.Struct):
     def parse_to_position_status_report(
         self,
         account_id: AccountId,
-        instrument_id: InstrumentId
+        instrument_id: InstrumentId,
+        report_id: UUID4,
+        ts_init: int
     ) -> PositionStatusReport:
-        position_side = BybitPositionSide(self.side).parse_to_position_side()
-
+        position_side = self.side.parse_to_position_side()
+        size = Quantity.from_str(self.positionValue)
         return PositionStatusReport(
             account_id=account_id,
             instrument_id=instrument_id,
-            position_side=
-
+            position_side=position_side,
+            quantity=size,
+            report_id=report_id,
+            ts_init=ts_init,
+            ts_last=ts_init,
         )