python: fix examples

python/bbgo/data/__init__.py

@@ -1,3 +1,4 @@
 from .balance import Balance
+from .error import ErrorMessage
 from .kline import KLine
 from .order import Order

python/bbgo/data/error.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import bbgo_pb2
+
+
+@dataclass
+class ErrorMessage:
+    code: int
+    message: str
+
+    @classmethod
+    def from_pb(cls, obj: bbgo_pb2.Error) -> ErrorMessage:
+        return cls(
+            code=obj.error_code,
+            message=obj.error_message,
+        )

python/bbgo/data/kline.py

@@ -1,6 +1,10 @@
+from __future__ import annotations
+
 from dataclasses import dataclass
 from datetime import datetime
 
+import bbgo_pb2
+
 
 @dataclass
 class KLine:
@@ -16,3 +20,19 @@ class KLine:
     end_time: datetime = None
     quote_volume: float = None
     closed: bool = None
+
+    @classmethod
+    def from_pb(cls, obj: bbgo_pb2.KLine) -> KLine:
+        return cls(
+            exchange=obj.exchange,
+            symbol=obj.symbol,
+            open=float(obj.open),
+            high=float(obj.high),
+            low=float(obj.low),
+            close=float(obj.close),
+            volume=float(obj.volume),
+            quote_volume=float(obj.quote_volume),
+            start_time=datetime.fromtimestamp(obj.start_time / 1000),
+            end_time=datetime.fromtimestamp(obj.end_time / 1000),
+            closed=obj.closed,
+        )

python/bbgo/services.py

@@ -1,8 +1,12 @@
 from typing import List
+from typing import Tuple
 
 import bbgo_pb2
 import bbgo_pb2_grpc
 
+from .data import ErrorMessage
+from .data import KLine
+
 
 class UserDataService(object):
 
@@ -27,16 +31,25 @@ class MarketService(object):
                      exchange: str,
                      symbol: str,
                      limit: int = 30,
-                     interval: int = 1,
-                     timestamp: int = None) -> bbgo_pb2.QueryKLinesResponse:
+                     interval: str = '1m',
+                     start_time: int = None,
+                     end_time: int = None) -> Tuple[List[KLine], ErrorMessage]:
         request = bbgo_pb2.QueryKLinesRequest(exchange=exchange,
                                               symbol=symbol,
                                               limit=limit,
                                               interval=interval,
-                                              timestamp=timestamp)
+                                              start_time=start_time,
+                                              end_time=end_time)
 
-        response = self.market_data_stub.QueryKLines(request)
-        return response
+        response = self.stub.QueryKLines(request)
+
+        klines = []
+        for kline in response.klines:
+            klines.append(KLine.from_pb(kline))
+
+        error = ErrorMessage.from_pb(response.error)
+
+        return klines, error
 
 
 class TradingService(object):

python/bbgo/stream.py

@@ -9,10 +9,11 @@ import grpc
 
 class Stream(object):
 
-    def __init__(self, host: str, port: int, subscriptions: List[bbgo_pb2.Subscription]):
+    def __init__(self, host: str, port: int, subscriptions: List[bbgo_pb2.Subscription], user_data: bool = False):
         self.host = host
         self.port = port
         self.subscriptions = subscriptions
+        self.user_data = user_data
 
         # callbacks for public channel
         self.book_event_callbacks = []
@@ -48,10 +49,11 @@ class Stream(object):
                 self.dispatch_user_events(response)
 
     def start(self):
-        group = asyncio.gather(
-            self.subscribe(),
-            self.subscribe_user_data(),
-        )
+        coroutines = [self.subscribe()]
+        if self.user_data:
+            coroutines.append(self.subscribe_user_data())
+
+        group = asyncio.gather(*coroutines)
         loop = asyncio.get_event_loop()
         loop.run_until_complete(group)
         loop.close()

python/examples/query_klines.py

@@ -0,0 +1,27 @@
+import click
+import grpc
+import bbgo_pb2
+import bbgo_pb2_grpc
+from bbgo import MarketService
+
+
+@click.command()
+@click.option('--host', default='127.0.0.1')
+@click.option('--port', default=50051)
+def main(host, port):
+    address = f'{host}:{port}'
+    channel = grpc.insecure_channel(address)
+
+    stub = bbgo_pb2_grpc.MarketDataServiceStub(channel)
+    service = MarketService(stub)
+
+    klines, error = service.query_klines(exchange='binance', symbol='BTCUSDT', interval='1m', limit=10)
+
+    for kline in klines:
+        print(kline)
+
+    print(error)
+
+
+if __name__ == '__main__':
+    main()

python/examples/stream.py

@@ -0,0 +1,27 @@
+import click
+from loguru import logger
+
+import bbgo_pb2
+from bbgo import Stream
+
+
+@click.command()
+@click.option('--host', default='127.0.0.1')
+@click.option('--port', default=50051)
+def main(host, port):
+    subscriptions = [
+        bbgo_pb2.Subscription(exchange='max', channel=bbgo_pb2.Channel.BOOK, symbol='BTCUSDT', depth='full'),
+        bbgo_pb2.Subscription(exchange='binance', channel=bbgo_pb2.Channel.BOOK, symbol='BTCUSDT', depth='full'),
+    ]
+
+    def book_event_callback(event):
+        logger.info(event)
+
+    stream = Stream(host, port, subscriptions)
+    stream.on_book_event(book_event_callback)
+
+    stream.start()
+
+
+if __name__ == '__main__':
+    main()

python/tests/test_data.py

@@ -0,0 +1,55 @@
+from datetime import datetime
+
+import bbgo_pb2
+from bbgo.data import KLine, ErrorMessage
+
+
+def test_kline_from_pb():
+    exchange = "binance"
+    symbol = "BTCUSDT"
+    open = "39919.31"
+    high = "39919.32"
+    low = "39919.31"
+    close = "39919.31"
+    volume = "0.27697"
+    quote_volume = "11056.4530226"
+    start_time = 1649833260000
+    end_time = 1649833319999
+    closed = True
+
+    kline_pb = bbgo_pb2.KLine(exchange=exchange,
+                              symbol=symbol,
+                              open=open,
+                              high=high,
+                              low=low,
+                              close=close,
+                              volume=volume,
+                              quote_volume=quote_volume,
+                              start_time=start_time,
+                              end_time=end_time,
+                              closed=closed)
+
+    kline = KLine.from_pb(kline_pb)
+
+    assert kline.exchange == exchange
+    assert kline.symbol == symbol
+    assert kline.open == float(open)
+    assert kline.high == float(high)
+    assert kline.low == float(low)
+    assert kline.close == float(close)
+    assert kline.volume == float(volume)
+    assert kline.quote_volume == float(quote_volume)
+    assert kline.start_time == datetime.fromtimestamp(start_time / 1000)
+    assert kline.end_time == datetime.fromtimestamp(end_time / 1000)
+    assert closed == closed
+
+
+def test_order_from_pb():
+    error_code = 123
+    error_message = "error message 123"
+
+    error_pb = bbgo_pb2.Error(error_code=error_code, error_message=error_message)
+    error = ErrorMessage.from_pb(error_pb)
+
+    assert error.code == error_code
+    assert error.message == error_message