Merge branch 'level2' into 'dev'

Level2

See merge request server/openapi/openapi-python-sdk!67

Author: zhanggoo9527

tigeropen/common/consts/push_destinations.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+
+QUOTE = 'quote'
+QUOTE_DEPTH = 'quotedepth'
+QUOTE_FUTURE = 'future'
+QUOTE_OPTION = 'option'
+
+TRADE_ASSET = 'trade/asset'
+TRADE_POSITION = 'trade/position'
+TRADE_ORDER = 'trade/order'

tigeropen/common/consts/quote_keys.py

@@ -17,6 +17,8 @@ class QuoteChangeKey(Enum):
     bid_price = 'bidPrice'
     bid_size = 'bidSize'
     minute = 'mi'
+    bid_depth = 'bidDepth'
+    ask_depth = 'askDepth'
 
 
 @unique

tigeropen/common/consts/service_types.py

@@ -42,6 +42,7 @@
 QUOTE_REAL_TIME = "quote_real_time"
 QUOTE_SHORTABLE_STOCKS = "quote_shortable_stocks"
 QUOTE_STOCK_TRADE = "quote_stock_trade"
+QUOTE_DEPTH = "quote_depth"  # level2 深度行情
 GRAB_QUOTE_PERMISSION = "grab_quote_permission"  # 抢占行情
 
 # 期权行情

tigeropen/examples/push_client_demo.py

@@ -42,6 +42,18 @@ def on_quote_changed(symbol, items, hour_trading):
          ('high', 345.0), ('prev_close', 342.4), ('low', 339.2), ('open', 344.0), ('volume', 7361440),
          ('minute', {'p': 339.8, 'a': 341.084, 't': 1568098440000, 'v': 7000, 'h': 340.0, 'l': 339.8}),
          ('timestamp', '1568098469463')]
+    深度行情 items 数据示例. 最多只推送前 40 档, 由 3 个等长数组构成，分别代表：price(价格)/volume(数量)/count(单量),
+    相邻档位价格可能一样, 其中 count 是可选的
+        [('bid_depth',
+            '[[127.87,127.86,127.86,127.86,127.85,127.85,127.85,127.84,127.84,127.84,127.84,127.83,127.83, 127.83,127.83,127.83,127.82,127.81,127.8,127.8,127.8,127.8,127.8,127.8,127.8,127.79,127.79,127.78, 127.78, 127.75,127.68,127.6,127.6,127.55,127.5,127.5,127.5,127.5,127.29,127.28],
+              [69,2,5,20,1,1,1,18,1,70,80,40,2,330,330,1,40,80,20,10,131,2,30,50,300,1,38,1,1,15,6,20,1,3,100,15,25,30,49,43]
+             ]'),
+        ('ask_depth',
+            '[[127.91,127.94,127.95,127.95,127.95,127.95,127.95,127.96,127.98,127.98,127.98,127.98,127.99,127.99, 128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0, 128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0,128.0],
+              [822,4,98,50,5,5,500,642,300,40,1,36,19,1,1,1,1,50,1,1,50,1,100,10,1,1,10,1,1,1,1,5,1,8,1,1,120,70,1,4]
+            ]'),
+        ('timestamp', 1621933454191)]
+
     """
     print(symbol, items, hour_trading)
 
@@ -158,6 +170,10 @@ def unsubscribe_callback(destination, content):
     push_client.subscribe_quote(['AAPL', 'GOOG'])
     # 可以指定关注的行情key的类型, QuoteKeyType.TRADE 为成交数据, QuoteKeyType.QUOTE 为盘口数据
     # push_client.subscribe_quote(['MSFT', 'AMD'], quote_key_type=QuoteKeyType.TRADE)
+
+    # 订阅深度行情
+    push_client.subscribe_depth_quote(['AMD', 'BABA'])
+
     # 订阅资产变动
     push_client.subscribe_asset()
     # 订阅订单变动

tigeropen/examples/quote_client_demo.py

@@ -43,6 +43,9 @@ def get_quote():
     print(ticks)
     short_interest = openapi_client.get_short_interest(['GOOG', 'AAPL', '00700'])
     print(short_interest)
+    # 获取深度行情
+    order_book = openapi_client.get_depth_quote(['02828'], Market.HK)
+    print(order_book)
     # 股票详情
     stock_details = openapi_client.get_stock_details(['AAPL', '03690'])
     print(stock_details)

tigeropen/push/push_client.py

@@ -6,10 +6,15 @@
 """
 import sys
 import json
+from collections import defaultdict
+
 import stomp
 import six
 import logging
 from stomp.exception import ConnectFailedException
+
+from tigeropen.common.consts.push_destinations import QUOTE, QUOTE_DEPTH, QUOTE_FUTURE, QUOTE_OPTION, TRADE_ASSET, \
+    TRADE_ORDER, TRADE_POSITION
 from tigeropen.common.util.signature_utils import sign_with_rsa
 from tigeropen.common.util.order_utils import get_order_status
 from tigeropen.common.consts.push_types import RequestType, ResponseType
@@ -73,10 +78,7 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconn
         self._private_key = None
         self._sign = None
         self._stomp_connection = None
-        self._quote_counter = 0
-        self._asset_counter = 0
-        self._position_counter = 0
-        self._order_counter = 0
+        self._destination_counter_map = defaultdict(lambda: 0)
 
         self.subscribed_symbols = None
         self.quote_changed = None
@@ -243,98 +245,53 @@ def on_error(self, headers, body):
         if self.error_callback:
             self.error_callback(body)
 
-    @staticmethod
-    def _get_subscribe_id(counter):
-        return 'sub-' + str(counter)
+    def _update_subscribe_id(self, destination):
+        self._destination_counter_map[destination] += 1
+
+    def _get_subscribe_id(self, destination):
+        return 'sub-' + str(self._destination_counter_map[destination])
 
     def subscribe_asset(self, account=None):
         """
         订阅账户资产更新
         :return:
         """
-        self._asset_counter += 1
-        sub_id = self._get_subscribe_id(self._asset_counter)
-        headers = dict()
-        headers['destination'] = 'trade/asset'
-        headers['subscription'] = 'Asset'
-        headers['id'] = sub_id
-        if account:
-            headers['account'] = account
-
-        self._stomp_connection.subscribe('trade/asset', id=sub_id, headers=headers)
-        return sub_id
+        return self._handle_trade_subscribe(TRADE_ASSET, 'Asset', account)
 
     def unsubscribe_asset(self, id=None):
         """
         退订账户资产更新
         :return:
         """
-        headers = dict()
-        headers['destination'] = 'trade/asset'
-        headers['subscription'] = 'Asset'
-        sub_id = id if id else self._get_subscribe_id(self._asset_counter)
-        headers['id'] = sub_id
-        self._stomp_connection.unsubscribe(id=sub_id, headers=headers)
+        self._handle_trade_unsubscribe(TRADE_ASSET, 'Asset', sub_id=id)
 
     def subscribe_position(self, account=None):
         """
         订阅账户持仓更新
         :return:
         """
-        self._position_counter += 1
-        sub_id = self._get_subscribe_id(self._position_counter)
-        headers = dict()
-        headers['destination'] = 'trade/position'
-        headers['subscription'] = 'Position'
-        headers['id'] = sub_id
-        if account:
-            headers['account'] = account
-
-        self._stomp_connection.subscribe('trade/position', id=sub_id, headers=headers)
-        return sub_id
+        return self._handle_trade_subscribe(TRADE_POSITION, 'Position', account)
 
     def unsubscribe_position(self, id=None):
         """
         退订账户持仓更新
         :return:
         """
-        headers = dict()
-        headers['destination'] = 'trade/position'
-        headers['subscription'] = 'Position'
-        sub_id = id if id else self._get_subscribe_id(self._position_counter)
-        headers['id'] = sub_id
-
-        self._stomp_connection.unsubscribe(id=sub_id, headers=headers)
+        self._handle_trade_unsubscribe(TRADE_POSITION, 'Position', sub_id=id)
 
     def subscribe_order(self, account=None):
         """
         订阅账户订单更新
         :return:
         """
-        self._order_counter += 1
-        sub_id = self._get_subscribe_id(self._order_counter)
-        headers = dict()
-        headers['destination'] = 'trade/order'
-        headers['subscription'] = 'OrderStatus'
-        headers['id'] = sub_id
-        if account:
-            headers['account'] = account
-
-        self._stomp_connection.subscribe('trade/order', id=sub_id, headers=headers)
-        return sub_id
+        return self._handle_trade_subscribe(TRADE_ORDER, 'OrderStatus', account)
 
     def unsubscribe_order(self, id=None):
         """
         退订账户订单更新
         :return:
         """
-        headers = dict()
-        headers['destination'] = 'trade/order'
-        headers['subscription'] = 'OrderStatus'
-        sub_id = id if id else self._get_subscribe_id(self._order_counter)
-        headers['id'] = sub_id
-
-        self._stomp_connection.unsubscribe(id=sub_id, headers=headers)
+        self._handle_trade_unsubscribe(TRADE_ORDER, 'OrderStatus', sub_id=id)
 
     def subscribe_quote(self, symbols, quote_key_type=QuoteKeyType.TRADE, focus_keys=None):
         """
@@ -344,48 +301,112 @@ def subscribe_quote(self, symbols, quote_key_type=QuoteKeyType.TRADE, focus_keys
         :param focus_keys: 行情 key
         :return:
         """
-        self._quote_counter += 1
-        sub_id = self._get_subscribe_id(self._quote_counter)
-        headers = dict()
-        headers['destination'] = 'quote'
-        headers['subscription'] = 'Quote'
-        headers['id'] = sub_id
-        if symbols:
-            headers['symbols'] = ','.join(symbols)
+        extra_headers = dict()
         if focus_keys:
             keys = list()
             for key in focus_keys:
                 if isinstance(key, six.string_types):
                     keys.append(key)
                 else:
                     keys.append(key.value)
-            headers['keys'] = ','.join(keys)
+            extra_headers['keys'] = ','.join(keys)
         elif quote_key_type and quote_key_type.value:
-            headers['keys'] = quote_key_type.value
-        self._stomp_connection.subscribe('quote', id=sub_id, headers=headers)
-        return sub_id
+            extra_headers['keys'] = quote_key_type.value
+        return self._handle_quote_subscribe(destination=QUOTE, subscription='Quote', symbols=symbols,
+                                            extra_headers=extra_headers)
+
+    def subscribe_depth_quote(self, symbols):
+        """
+        订阅深度行情
+        :param symbols: symbol列表
+        :return:
+        """
+        return self._handle_quote_subscribe(destination=QUOTE_DEPTH, subscription='QuoteDepth', symbols=symbols)
+
+    def subscribe_option(self, symbols):
+        """
+        订阅期权行情
+        :param symbols: symbol列表
+        :return:
+        """
+        return self._handle_quote_subscribe(destination=QUOTE_OPTION, subscription='Option', symbols=symbols)
+
+    def subscribe_future(self, symbols):
+        """
+        订阅期货行情
+        :param symbols: symbol列表
+        :return:
+        """
+        return self._handle_quote_subscribe(destination=QUOTE_FUTURE, subscription='Future', symbols=symbols)
 
     def query_subscribed_quote(self):
         """
         查询已订阅行情的合约
         :return:
         """
         headers = dict()
-        headers['destination'] = 'quote'
+        headers['destination'] = QUOTE
         headers['req-type'] = RequestType.REQ_SUB_SYMBOLS.value
-        self._stomp_connection.send('quote', "{}", headers=headers)
+        self._stomp_connection.send(QUOTE, "{}", headers=headers)
 
     def unsubscribe_quote(self, symbols=None, id=None):
         """
         退订行情更新
         :return:
         """
+        self._handle_quote_unsubscribe(destination=QUOTE, subscription='Quote', sub_id=id, symbols=symbols)
+
+    def unsubscribe_depth_quote(self, symbols=None, id=None):
+        """
+        退订深度行情更新
+        :return:
+        """
+        self._handle_quote_unsubscribe(destination=QUOTE_DEPTH, subscription='AskBid', sub_id=id, symbols=symbols)
+
+    def _handle_trade_subscribe(self, destination, subscription, account=None, extra_headers=None):
+        if extra_headers is None:
+            extra_headers = dict()
+        if account is not None:
+            extra_headers['account'] = account
+        return self._handle_subscribe(destination=destination, subscription=subscription, extra_headers=extra_headers)
+
+    def _handle_quote_subscribe(self, destination, subscription, symbols=None, extra_headers=None):
+        if extra_headers is None:
+            extra_headers = dict()
+        if symbols is not None:
+            extra_headers['symbols'] = ','.join(symbols)
+        return self._handle_subscribe(destination=destination, subscription=subscription, extra_headers=extra_headers)
+
+    def _handle_trade_unsubscribe(self, destination, subscription, sub_id=None):
+        self._handle_unsubscribe(destination=destination, subscription=subscription, sub_id=sub_id)
+
+    def _handle_quote_unsubscribe(self, destination, subscription, sub_id=None, symbols=None):
+        extra_headers = dict()
+        if symbols is not None:
+            extra_headers['symbols'] = ','.join(symbols)
+        self._handle_unsubscribe(destination=destination, subscription=subscription, sub_id=sub_id,
+                                 extra_headers=extra_headers)
+
+    def _handle_subscribe(self, destination, subscription, extra_headers=None):
         headers = dict()
-        headers['destination'] = 'quote'
-        headers['subscription'] = 'Quote'
-        sub_id = id if id else self._get_subscribe_id(self._quote_counter)
+        headers['destination'] = destination
+        headers['subscription'] = subscription
+        self._update_subscribe_id(destination)
+        sub_id = self._get_subscribe_id(destination)
         headers['id'] = sub_id
-        if symbols:
-            headers['symbols'] = ','.join(symbols)
 
-        self._stomp_connection.unsubscribe(id=sub_id, headers=headers)
+        if extra_headers is not None:
+            headers.update(extra_headers)
+        self._stomp_connection.subscribe(destination, id=sub_id, headers=headers)
+        return sub_id
+
+    def _handle_unsubscribe(self, destination, subscription, sub_id=None, extra_headers=None):
+        headers = dict()
+        headers['destination'] = destination
+        headers['subscription'] = subscription
+        id_ = sub_id if sub_id is not None else self._get_subscribe_id(destination)
+        headers['id'] = id_
+        if extra_headers:
+            headers.update(extra_headers)
+
+        self._stomp_connection.unsubscribe(id=id_, headers=headers)