Merge pull request #51 from tigerfintech/feature_upgrade_stomp

upgrade stomp.py

Author: hotea

CHANGELOG.md

@@ -1,3 +1,18 @@
+## 2.1.2 (2022-06-14)
+### Modify
+- 升级 stomp.py 版本, 将之前的 4.1.24 升级到 8.0.1
+### Breaking
+- PushClient 去除 `auto_reconnect` 参数，如需自定义重连方式，可自定义方法并绑定 `disconnect_callback` 进行重连
+- 处理连接被踢的情况，如果有多台设备使用相同 tiger_id 连接, 新连接会将较早的连接踢掉，较早连接会抛出异常，停止接收消息
+
+## 2.1.1 (2022-05-25)
+### New
+- 新增批量分页获取k线接口
+  股票：`QuoteClient.get_bars_by_page`
+  期货：`QuoteClient.get_future_bars_by_page`
+- `QuoteClient.get_future_bars`, `QuoteClient.get_bars` 增加 `page_token` 参数，可用于分页请求定位下一页位置
+- `tigeropen.trade.domain.order.Order` 新增 `user_mark` 属性，用户下单时可传入一定长度的备注信息，该属性值在查询订单时会返回。(需用户提前向平台申请配置)
+
 ## 2.1.0 (2022-05-07)
 ### New
 - 动态获取服务域名；更改默认域名

requirements.txt

@@ -1,9 +1,10 @@
-simplejson==3.17.3
+simplejson==3.17.6
 delorean==1.0.0
 pandas
 python-dateutil==2.8.2
-pytz==2021.1
+pytz==2022.1
 pyasn1==0.4.8
-rsa==4.7.2
-stomp.py==4.1.24
-getmac==0.8.2
+rsa==4.8
+stomp.py==8.0.1
+getmac==0.8.3
+cryptography==37.0.2

tigeropen/__init__.py

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

tigeropen/examples/option_helpers/helpers.py

@@ -268,6 +268,7 @@ def __init__(self,
     else:
         helper_class = FDAmericanDividendOptionHelper
 
+    ql.Settings.instance().evaluationDate = settlement_date
     helper = helper_class(option_type=ql_option_type,
                           underlying=args.underlying,
                           strike=args.strike,

tigeropen/examples/push_client_demo.py

@@ -120,12 +120,12 @@ def unsubscribe_callback(destination, content):
     print('subscribe:{}, callback content:{}'.format(destination, content))
 
 
-def error_callback(content):
+def error_callback(frame):
     """错误回调"""
-    print(content)
+    print(frame)
 
 
-def connect_callback():
+def connect_callback(frame):
     """连接建立回调"""
     print('connected')
 

tigeropen/examples/quote_client_demo.py

@@ -5,9 +5,11 @@
 @author: gaoan
 """
 import logging
+import time
+
 import pandas as pd
 from tigeropen.common.consts import Market, QuoteRight, FinancialReportPeriodType, Valuation, \
-    Income, Balance, CashFlow, BalanceSheetRatio, Growth, Leverage, Profitability, IndustryLevel
+    Income, Balance, CashFlow, BalanceSheetRatio, Growth, Leverage, Profitability, IndustryLevel, BarPeriod
 from tigeropen.quote.domain.filter import OptionFilter
 
 from tigeropen.quote.quote_client import QuoteClient
@@ -61,6 +63,17 @@ def get_quote():
     print(delay_brief)
 
 
+def test_gat_bars_by_page():
+    bars = openapi_client.get_bars_by_page(['AAPL'], period=BarPeriod.DAY,
+                                           end_time='2022-05-01',
+                                           total=10,
+                                           page_size=4,
+                                           )
+    bars['cn_date'] = pd.to_datetime(bars['time'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('Asia/Shanghai')
+    bars['us_date'] = pd.to_datetime(bars['time'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('US/Eastern')
+    print(bars)
+
+
 def get_option_quote():
     symbol = 'AAPL'
     expirations = openapi_client.get_option_expirations(symbols=[symbol])
@@ -85,11 +98,11 @@ def get_option_quote():
     chains = openapi_client.get_option_chain('AAPL', '2023-01-20', implied_volatility_min=0.5, open_interest_min=200,
                                              vega_min=0.1, rho_max=0.9)
     # convert expiry date to US/Eastern
-    chains['expiry_date'] = pd.to_datetime(chains['expiry'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('US/Eastern')
+    chains['expiry_date'] = pd.to_datetime(chains['expiry'], unit='ms').dt.tz_localize('UTC').dt.tz_convert(
+        'US/Eastern')
     print(chains)
 
 
-
 def get_future_quote():
     exchanges = openapi_client.get_future_exchanges()
     print(exchanges)
@@ -107,18 +120,27 @@ def get_future_quote():
     print(briefs)
 
 
+def test_get_future_bars_by_page():
+    bars = openapi_client.get_future_bars_by_page('CLmain',
+                                                  end_time=1648526400000,
+                                                  total=10,
+                                                  page_size=4)
+    bars['cn_date'] = pd.to_datetime(bars['time'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('Asia/Shanghai')
+    bars['us_date'] = pd.to_datetime(bars['time'], unit='ms').dt.tz_localize('UTC').dt.tz_convert('US/Eastern')
+    print(bars)
+
 
 def get_fundamental():
     """获取基础数据"""
-    
+
     # 日级财务数据
     financial_daily = openapi_client.get_financial_daily(symbols=['AAPL', 'MSFT'],
                                                          market=Market.US,
                                                          fields=[Valuation.shares_outstanding],
                                                          begin_date='2019-01-01',
                                                          end_date='2019-01-10')
     print(financial_daily)
-    
+
     # 财报数据(季报或年报)
     financial_report = openapi_client.get_financial_report(symbols=['AAPL', 'GOOG'],
                                                            market=Market.US,

tigeropen/push/__init__.py

@@ -3,4 +3,26 @@
 Created on 2018/10/29
 
 @author: gaoan
-"""
+"""
+import ssl
+import time
+
+
+def _patch_ssl(wait=0.01):
+    def new_wrap_socket(self, sock, server_side=False,
+                        do_handshake_on_connect=True,
+                        suppress_ragged_eofs=True,
+                        server_hostname=None, session=None):
+        time.sleep(wait)
+        return self.sslsocket_class._create(
+            sock=sock,
+            server_side=server_side,
+            do_handshake_on_connect=do_handshake_on_connect,
+            suppress_ragged_eofs=suppress_ragged_eofs,
+            server_hostname=server_hostname,
+            context=self,
+            session=session
+        )
+    ssl.SSLContext.old_wrap_socket = ssl.SSLContext.wrap_socket
+    ssl.SSLContext.wrap_socket = new_wrap_socket
+

tigeropen/push/push_client.py

@@ -7,6 +7,7 @@
 import json
 import logging
 import sys
+import os
 from collections import defaultdict
 
 import stomp
@@ -22,10 +23,12 @@
     SUBSCRIPTION_TRADE_ORDER
 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.push import _patch_ssl
 
 HOUR_TRADING_QUOTE_KEYS_MAPPINGS = {'hourTradingLatestPrice': 'latest_price', 'hourTradingPreClose': 'pre_close',
                                     'hourTradingLatestTime': 'latest_time', 'hourTradingVolume': 'volume',
@@ -67,14 +70,12 @@
 
 
 class PushClient(stomp.ConnectionListener):
-    def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconnect=True,
-                 heartbeats=(30 * 1000, 30 * 1000)):
+    def __init__(self, host, port, use_ssl=True, connection_timeout=120, heartbeats=(30 * 1000, 30 * 1000)):
         """
         :param host:
         :param port:
         :param use_ssl:
-        :param connection_timeout: second
-        :param auto_reconnect:
+        :param connection_timeout: unit: second. The timeout value should be greater the heartbeats interval
         :param heartbeats: tuple of millisecond
         """
         self.host = host
@@ -97,8 +98,8 @@ def __init__(self, host, port, use_ssl=True, connection_timeout=120, auto_reconn
         self.unsubscribe_callback = None
         self.error_callback = None
         self._connection_timeout = connection_timeout
-        self._auto_reconnect = auto_reconnect
         self._heartbeats = heartbeats
+        _patch_ssl()
 
     def _connect(self):
         try:
@@ -108,12 +109,13 @@ def _connect(self):
         except:
             pass
 
-        self._stomp_connection = stomp.Connection12(host_and_ports=[(self.host, self.port), ], use_ssl=self.use_ssl,
+        self._stomp_connection = stomp.Connection12(host_and_ports=[(self.host, self.port)],
                                                     keepalive=KEEPALIVE, timeout=self._connection_timeout,
                                                     heartbeats=self._heartbeats)
         self._stomp_connection.set_listener('push', self)
-        self._stomp_connection.start()
         try:
+            if self.use_ssl:
+                self._stomp_connection.set_ssl([(self.host, self.port)])
             self._stomp_connection.connect(self._tiger_id, self._sign, wait=True, headers=self._generate_headers())
         except ConnectFailedException as e:
             raise e
@@ -128,23 +130,26 @@ def disconnect(self):
         if self._stomp_connection:
             self._stomp_connection.disconnect()
 
-    def on_connected(self, headers, body):
+    def on_connected(self, frame):
         if self.connect_callback:
-            self.connect_callback()
+            self.connect_callback(frame)
 
     def on_disconnected(self):
         if self.disconnect_callback:
             self.disconnect_callback()
-        elif self._auto_reconnect:
-            self._connect()
 
-    def on_message(self, headers, body):
+    def on_message(self, frame):
         """
         Called by the STOMP connection when a MESSAGE frame is received.
 
-        :param dict headers: a dictionary containing all headers sent by the server as key/value pairs.
-        :param body: the frame's payload - the message body.
+        :param Frame frame: the stomp frame. stomp.utils.Frame
+            A STOMP frame's attributes:
+                cmd: the protocol command
+                headers: a map of headers for the frame
+                body: the content of the frame.
         """
+        headers = frame.headers
+        body = frame.body
         try:
             response_type = headers.get('ret-type')
             if response_type == str(ResponseType.GET_SUB_SYMBOLS_END.value):
@@ -255,11 +260,17 @@ def on_message(self, headers, body):
         except Exception as e:
             logging.error(e, exc_info=True)
 
-    def on_error(self, headers, body):
+    def on_error(self, frame):
+        body = json.loads(frame.body)
+        if body.get('code') == 4001:
+            logging.error(body)
+            self.disconnect_callback = None
+            raise ApiException(4001, body.get('message'))
+
         if self.error_callback:
-            self.error_callback(body)
+            self.error_callback(frame)
         else:
-            logging.error(body)
+            logging.error(frame.body)
 
     def _update_subscribe_id(self, destination):
         self._destination_counter_map[destination] += 1