Merge pull request #54 from tigerfintech/dev

push tick; subscribe quote query callback

Author: hotea

CHANGELOG.md

@@ -1,3 +1,13 @@
+## 2.1.3 (2022-07-01)
+### New
+- 长连接新增逐笔订阅: `PushClient.subscribe_tick`, 退订 `PushClient.unsubscribe_tick`
+- 新增已订阅查询回调方法 `PushClient.query_subscribed_callback` 取代旧有的 `Pushclient.subscribed_symbols`
+- `QuoteClient.get_trade_ticks` 新增 `trade_session` 参数，可指定该参数查询盘前盘后数据
+
+### Modify
+- `Pushclient.subscribed_symbols` 标记为废弃
+
+
 ## 2.1.2 (2022-06-14)
 ### Modify
 - 升级 stomp.py 版本, 将之前的 4.1.24 升级到 8.0.1

setup.py

@@ -16,7 +16,7 @@
     name='tigeropen',
     version=__VERSION__,
     description='TigerBrokers Open API',
-    packages=find_packages(exclude=[]),
+    packages=find_packages(exclude=["tests"]),
     author='TigerBrokers',
     author_email='[email protected]',
     license='Apache License v2',

tigeropen/__init__.py

@@ -4,4 +4,4 @@
 
 @author: gaoan
 """
-__VERSION__ = '2.1.2'
+__VERSION__ = '2.1.3'

tigeropen/common/consts/push_destinations.py

@@ -4,6 +4,7 @@
 QUOTE_DEPTH = 'quotedepth'
 QUOTE_FUTURE = 'future'
 QUOTE_OPTION = 'option'
+TRADE_TICK = 'tradetick'
 
 TRADE_ASSET = 'trade/asset'
 TRADE_POSITION = 'trade/position'

tigeropen/common/consts/push_subscriptions.py

@@ -4,6 +4,7 @@
 SUBSCRIPTION_QUOTE_DEPTH = 'QuoteDepth'
 SUBSCRIPTION_QUOTE_FUTURE = 'Future'
 SUBSCRIPTION_QUOTE_OPTION = 'Option'
+SUBSCRIPTION_TRADE_TICK = 'TradeTick'
 
 SUBSCRIPTION_TRADE_ASSET = 'Asset'
 SUBSCRIPTION_TRADE_POSITION = 'Position'

tigeropen/common/consts/service_types.py

@@ -50,6 +50,7 @@
 QUOTE_DEPTH = "quote_depth"  # level2 深度行情
 GRAB_QUOTE_PERMISSION = "grab_quote_permission"  # 抢占行情
 GET_QUOTE_PERMISSION = "get_quote_permission"
+TRADING_CALENDAR = "trading_calendar"
 
 # 期权行情
 OPTION_EXPIRATION = "option_expiration"

tigeropen/examples/push_client_demo.py

@@ -6,12 +6,29 @@
 """
 import time
 # from tigeropen.common.consts import QuoteKeyType
+import pandas as pd
+
 from tigeropen.push.push_client import PushClient
 from tigeropen.examples.client_config import get_client_config
 
 
+def query_subscribed_callback(data):
+    """
+    callback of PushClient.query_subscribed_quote
+    :param data:
+        example:
+        {'subscribed_symbols': ['QQQ'], 'limit': 1200, 'used': 1, 'symbol_focus_keys': {'qqq': ['open', 'prev_close', 'low', 'volume', 'latest_price', 'close', 'high']},
+         'subscribed_quote_depth_symbols': ['NVDA'], 'quote_depth_limit': 20, 'quote_depth_used': 1,
+         'subscribed_trade_tick_symbols': ['QQQ', 'AMD', '00700'], 'trade_tick_limit': 1200, 'trade_tick_used': 3
+         }
+    :return:
+    """
+    print(data)
+
+
 def on_query_subscribed_quote(symbols, focus_keys, limit, used):
     """
+    deprecated. Use query_subscribed_callback instead.
     查询已订阅symbol回调
     :param symbols: 订阅合约的列表
     :param focus_keys: 每个合约订阅的 key 列表
@@ -58,6 +75,28 @@ def on_quote_changed(symbol, items, hour_trading):
     print(symbol, items, hour_trading)
 
 
+def on_tick_changed(symbol, items):
+    """
+
+    :param symbol:
+    :param items:
+      items example:
+        [{'tick_type': '*', 'price': 293.87, 'volume': 102, 'part_code': 'NSDQ',
+            'part_code_name': 'NASDAQ Stock Market, LLC (NASDAQ)', 'cond': 'US_FORM_T', 'time': 1656405615779,
+            'server_timestamp': 1656405573461, 'type': 'TradeTick', 'quote_level': 'usStockQuote', 'sn': 342,
+            'timestamp': 1656405617385},
+         {'tick_type': '*', 'price': 293.87, 'volume': 102, 'part_code': 'NSDQ',
+          'part_code_name': 'NASDAQ Stock Market, LLC (NASDAQ)', 'cond': 'US_FORM_T', 'time': 1656405616573,
+          'server_timestamp': 1656405573461,
+          'type': 'TradeTick', 'quote_level': 'usStockQuote', 'sn': 343, 'timestamp': 1656405617385}]
+    :return:
+    """
+    print(symbol, items)
+    # convert to DataFrame
+    # frame = pd.DataFrame(items)
+    # print(frame)
+
+
 def on_order_changed(account, items):
     """
 
@@ -152,8 +191,12 @@ def disconnect_callback():
 
     # 行情变动回调
     push_client.quote_changed = on_quote_changed
+    # 逐笔数据回调
+    push_client.tick_changed = on_tick_changed
     # 已订阅 symbol 查询回调
-    push_client.subscribed_symbols = on_query_subscribed_quote
+    push_client.query_subscribed_callback = query_subscribed_callback
+    # 已订阅 symbol 查询回调(已废弃)
+    # push_client.subscribed_symbols = on_query_subscribed_quote
     # 订单变动回调
     push_client.order_changed = on_order_changed
     # 资产变动回调
@@ -182,6 +225,9 @@ def disconnect_callback():
     # 订阅深度行情
     push_client.subscribe_depth_quote(['AMD', 'BABA'])
 
+    # 订阅逐笔数据
+    push_client.subscribe_tick(['AMD', 'QQQ'])
+
     # 订阅资产变动
     push_client.subscribe_asset()
     # 订阅订单变动

tigeropen/examples/quote_client_demo.py

@@ -62,6 +62,10 @@ def get_quote():
     delay_brief = openapi_client.get_stock_delay_briefs(['AAPL', 'GOOG'])
     print(delay_brief)
 
+    # 获取市场交易日历
+    calendar = openapi_client.get_trading_calendar(Market.US, begin_date='2022-07-01', end_date='2022-09-02')
+    print(calendar)
+
 
 def test_gat_bars_by_page():
     bars = openapi_client.get_bars_by_page(['AAPL'], period=BarPeriod.DAY,

tigeropen/push/push_client.py

@@ -7,8 +7,8 @@
 import json
 import logging
 import sys
-import os
 from collections import defaultdict
+from itertools import accumulate, zip_longest
 
 import stomp
 from stomp.exception import ConnectFailedException
@@ -17,17 +17,19 @@
 from tigeropen.common.consts import OrderStatus
 from tigeropen.common.consts.params import P_SDK_VERSION, P_SDK_VERSION_PREFIX
 from tigeropen.common.consts.push_destinations import QUOTE, QUOTE_DEPTH, QUOTE_FUTURE, QUOTE_OPTION, TRADE_ASSET, \
-    TRADE_ORDER, TRADE_POSITION
+    TRADE_ORDER, TRADE_POSITION, TRADE_TICK
 from tigeropen.common.consts.push_subscriptions import SUBSCRIPTION_QUOTE, SUBSCRIPTION_QUOTE_DEPTH, \
     SUBSCRIPTION_QUOTE_OPTION, SUBSCRIPTION_QUOTE_FUTURE, SUBSCRIPTION_TRADE_ASSET, SUBSCRIPTION_TRADE_POSITION, \
-    SUBSCRIPTION_TRADE_ORDER
+    SUBSCRIPTION_TRADE_ORDER, SUBSCRIPTION_TRADE_TICK
 from tigeropen.common.consts.push_types import RequestType, ResponseType
 from tigeropen.common.consts.quote_keys import QuoteChangeKey, QuoteKeyType
 from tigeropen.common.exceptions import ApiException
-from tigeropen.common.util.string_utils import camel_to_underline
 from tigeropen.common.util.common_utils import get_enum_value
 from tigeropen.common.util.order_utils import get_order_status
 from tigeropen.common.util.signature_utils import sign_with_rsa
+from tigeropen.common.util.string_utils import camel_to_underline, camel_to_underline_obj
+from tigeropen.common.util.tick_util import get_part_code, get_part_code_name, get_trade_condition_map, \
+    get_trade_condition
 from tigeropen.push import _patch_ssl
 
 HOUR_TRADING_QUOTE_KEYS_MAPPINGS = {'hourTradingLatestPrice': 'latest_price', 'hourTradingPreClose': 'pre_close',
@@ -88,7 +90,9 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, heartbeats=
         self._destination_counter_map = defaultdict(lambda: 0)
 
         self.subscribed_symbols = None
+        self.query_subscribed_callback = None
         self.quote_changed = None
+        self.tick_changed = None
         self.asset_changed = None
         self.position_changed = None
         self.order_changed = None
@@ -99,6 +103,7 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, heartbeats=
         self.error_callback = None
         self._connection_timeout = connection_timeout
         self._heartbeats = heartbeats
+        self.logger = logging.getLogger('tiger_openapi')
         _patch_ssl()
 
     def _connect(self):
@@ -153,17 +158,29 @@ def on_message(self, frame):
         try:
             response_type = headers.get('ret-type')
             if response_type == str(ResponseType.GET_SUB_SYMBOLS_END.value):
-                if self.subscribed_symbols:
+                if self.subscribed_symbols or self.query_subscribed_callback:
                     data = json.loads(body)
-                    limit = data.get('limit')
-                    symbols = data.get('subscribedSymbols')
-                    used = data.get('used')
-                    symbol_focus_keys = data.get('symbolFocusKeys')
+                    formatted_data = camel_to_underline_obj(data)
+
+                    limit = formatted_data.get('limit')
+                    subscribed_symbols = formatted_data.get('subscribed_symbols')
+                    used = formatted_data.get('used')
+                    symbol_focus_keys = formatted_data.get('symbol_focus_keys')
                     focus_keys = dict()
                     for sym, keys in symbol_focus_keys.items():
                         keys = set(QUOTE_KEYS_MAPPINGS.get(key, camel_to_underline(key)) for key in keys)
                         focus_keys[sym] = list(keys)
-                    self.subscribed_symbols(symbols, focus_keys, limit, used)
+                    formatted_data['symbol_focus_keys'] = focus_keys
+                    formatted_data['subscribed_quote_depth_symbols'] = formatted_data.pop('subscribed_ask_bid_symbols')
+                    formatted_data['quote_depth_limit'] = formatted_data.pop('ask_bid_limit')
+                    formatted_data['quote_depth_used'] = formatted_data.pop('ask_bid_used')
+                    if self.subscribed_symbols:
+                        self.logger.warning('PushClient.subscribed_symbols is deprecated, '
+                                            'use PushClient.query_subscribed_callback instead.')
+                        self.subscribed_symbols(subscribed_symbols, focus_keys, limit, used)
+                    if self.query_subscribed_callback:
+                        self.query_subscribed_callback(formatted_data)
+
             elif response_type == str(ResponseType.GET_QUOTE_CHANGE_END.value):
                 if self.quote_changed:
                     data = json.loads(body)
@@ -204,6 +221,11 @@ def on_message(self, frame):
                                     items.append((camel_to_underline(key), value))
                         if items:
                             self.quote_changed(symbol, items, hour_trading)
+            elif response_type == str(ResponseType.GET_TRADING_TICK_END.value):
+                if self.tick_changed:
+                    symbol, items = self._convert_tick(body)
+                    self.tick_changed(symbol, items)
+
             elif response_type == str(ResponseType.SUBSCRIBE_ASSET.value):
                 if self.asset_changed:
                     data = json.loads(body)
@@ -258,19 +280,19 @@ def on_message(self, frame):
                 if self.error_callback:
                     self.error_callback(body)
         except Exception as e:
-            logging.error(e, exc_info=True)
+            self.logger.error(e, exc_info=True)
 
     def on_error(self, frame):
         body = json.loads(frame.body)
         if body.get('code') == 4001:
-            logging.error(body)
+            self.logger.error(body)
             self.disconnect_callback = None
             raise ApiException(4001, body.get('message'))
 
         if self.error_callback:
             self.error_callback(frame)
         else:
-            logging.error(frame.body)
+            self.logger.error(frame.body)
 
     def _update_subscribe_id(self, destination):
         self._destination_counter_map[destination] += 1
@@ -342,6 +364,15 @@ def subscribe_quote(self, symbols, quote_key_type=QuoteKeyType.TRADE, focus_keys
         return self._handle_quote_subscribe(destination=QUOTE, subscription=SUBSCRIPTION_QUOTE, symbols=symbols,
                                             extra_headers=extra_headers)
 
+    def subscribe_tick(self, symbols):
+        """
+        subscribe trade tick
+        :param symbols: symbol列表
+        :return:
+        """
+        return self._handle_quote_subscribe(destination=TRADE_TICK, subscription=SUBSCRIPTION_TRADE_TICK,
+                                            symbols=symbols)
+
     def subscribe_depth_quote(self, symbols):
         """
         订阅深度行情
@@ -383,6 +414,14 @@ def unsubscribe_quote(self, symbols=None, id=None):
         """
         self._handle_quote_unsubscribe(destination=QUOTE, subscription=SUBSCRIPTION_QUOTE, sub_id=id, symbols=symbols)
 
+    def unsubscribe_tick(self, symbols=None, id=None):
+        """
+        退订行情更新
+        :return:
+        """
+        self._handle_quote_unsubscribe(destination=TRADE_TICK, subscription=SUBSCRIPTION_TRADE_TICK, sub_id=id,
+                                       symbols=symbols)
+
     def unsubscribe_depth_quote(self, symbols=None, id=None):
         """
         退订深度行情更新
@@ -440,3 +479,40 @@ def _generate_headers(self, extra_headers=None):
             headers.update(extra_headers)
         return headers
 
+    @staticmethod
+    def _convert_tick(tick):
+        data = json.loads(tick)
+        symbol = data.pop('symbol')
+        data = camel_to_underline_obj(data)
+        price_offset = 10 ** data.pop('price_offset')
+        price_base = data.pop('price_base')
+        # The latter time is equal to the sum of all previous values
+        price_items = [('price', (item + price_base) / price_offset) for item in data.pop('prices')]
+        time_items = [('time', item) for item in accumulate(data.pop('times'))]
+        volumes = [('volume', item) for item in data.pop('volumes')]
+        tick_type_items = [('tick_type', item) for item in data.pop('tick_type')]
+        part_codes = data.pop('part_code')
+        if part_codes:
+            part_code_items = [('part_code', get_part_code(item)) for item in part_codes]
+            part_code_name_items = [('part_code_name', get_part_code_name(item)) for item in part_codes]
+        else:
+            part_code_items = [('part_code', None) for _ in range(len(time_items))]
+            part_code_name_items = [('part_code_name', None) for _ in range(len(time_items))]
+
+        conds = data.pop('cond')
+        cond_map = get_trade_condition_map(data.get('quote_level'))
+        if conds:
+            cond_items = [('cond', get_trade_condition(item, cond_map)) for item in conds]
+        else:
+            cond_items = [('cond', None) for _ in range(len(time_items))]
+        sn = data.pop('sn')
+        sn_list = [('sn', sn + i) for i in range(len(time_items))]
+        tick_data = zip_longest(tick_type_items, price_items, volumes, part_code_items,
+                                part_code_name_items, cond_items, time_items, sn_list)
+        items = []
+        for item in tick_data:
+            item_dict = dict(item)
+            item_dict.update(data)
+            items.append(item_dict)
+        return symbol, items
+