Refactor following jobs

Author: sinall

shipane_sdk/base_quant_client.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+
+from datetime import date, timedelta, datetime
+
+
+class BaseQuantClient(object):
+    def __init__(self, name):
+        self._name = name
+        self._last_login_time = datetime.now() - timedelta(1)
+
+    @property
+    def name(self):
+        return self._name
+
+    def login(self):
+        self._last_login_time = datetime.now()
+
+    def is_login(self):
+        return self._last_login_time >= datetime.combine(date.today(), datetime.min.time())
+
+    def query(self):
+        return []

shipane_sdk/jobs/joinquant_following.py renamed to shipane_sdk/jobs/online_quant_following.py

@@ -3,20 +3,15 @@
 import logging
 from datetime import datetime
 
-from shipane_sdk.joinquant.client import JoinQuantClient
-from shipane_sdk.joinquant.transaction import JoinQuantTransaction
 from shipane_sdk.market_utils import MarketUtils
 
 
-class JoinQuantFollowingJob(object):
-    def __init__(self, config, client):
+class OnlineQuantFollowingJob(object):
+    def __init__(self, shipane_client, quant_client, name=None):
         self._log = logging.getLogger()
-        self._config = config
-        self._shipane_client = client
-        self._jq_client = JoinQuantClient(username=self._config.get('JoinQuant', 'username'),
-                                          password=self._config.get('JoinQuant', 'password'),
-                                          backtest_id=self._config.get('JoinQuant', 'backtest_id'))
-        self._jq_client.login()
+        self._shipane_client = shipane_client
+        self._quant_client = quant_client
+        self._name = name
         self._start_datatime = datetime.now()
         self._processed_transactions = []
 
@@ -27,19 +22,21 @@ def __call__(self):
                 del self._processed_transactions[:]
             return
 
+        if not self._quant_client.is_login():
+            self._log.info("登录 {}".format(self._quant_client.name))
+            self._quant_client.login()
+
         self._log.info("********** 开始跟单 **********")
         try:
-            transaction_detail = self._jq_client.query()
-            raw_transactions = transaction_detail['data']['transaction']
-            self._log.info("获取到 {} 条委托".format(len(raw_transactions)))
+            all_transactions = self._quant_client.query()
+            self._log.info("获取到 {} 条委托".format(len(all_transactions)))
 
             transactions = []
-            for raw_transaction in raw_transactions:
-                transaction = JoinQuantTransaction(raw_transaction).normalize()
+            for transaction in all_transactions:
                 if self._is_expired(transaction):
                     continue
-
                 transactions.append(transaction)
+            
             self._log.info("获取到 {} 条有效委托".format(len(transactions)))
 
             for tx in transactions:
@@ -59,6 +56,10 @@ def __call__(self):
             self._log.exception("跟单异常")
         self._log.info("********** 结束跟单 **********\n")
 
+    @property
+    def name(self):
+        return self._name
+
     def _is_expired(self, transaction):
         if transaction.completed_at < self._start_datatime:
             return True

shipane_sdk/jobs/ricequant_following.py

@@ -4,11 +4,16 @@
 
 import requests
 
+from shipane_sdk.base_quant_client import BaseQuantClient
+from shipane_sdk.joinquant.transaction import JoinQuantTransaction
 
-class JoinQuantClient(object):
+
+class JoinQuantClient(BaseQuantClient):
     BASE_URL = 'https://www.joinquant.com'
 
     def __init__(self, **kwargs):
+        super(JoinQuantClient, self).__init__('JoinQuant')
+
         self._session = requests.Session()
         self._username = kwargs.pop('username')
         self._password = kwargs.pop('password')
@@ -35,11 +40,20 @@ def login(self):
             'cookie': response.headers['Set-Cookie']
         })
 
+        super(JoinQuantClient, self).login()
+
     def query(self):
         today_str = datetime.today().strftime('%Y-%m-%d')
         response = self._session.get('{}/algorithm/live/transactionDetail'.format(self.BASE_URL), params={
             'backtestId': self._backtest_id,
             'data': today_str,
             'ajax': 1
         })
-        return response.json()
+        transaction_detail = response.json()
+        raw_transactions = transaction_detail['data']['transaction']
+        transactions = []
+        for raw_transaction in raw_transactions:
+            transaction = JoinQuantTransaction(raw_transaction).normalize()
+            transactions.append(transaction)
+
+        return transactions

shipane_sdk/joinquant/client.py

@@ -2,14 +2,27 @@
 
 from rqopen_client import RQOpenClient
 
+from shipane_sdk.base_quant_client import BaseQuantClient
+from shipane_sdk.ricequant.transaction import RiceQuantTransaction
 
-class RiceQuantClient(object):
+
+class RiceQuantClient(BaseQuantClient):
     def __init__(self, **kwargs):
+        super(RiceQuantClient, self).__init__('RiceQuant')
+
         self._run_id = kwargs.pop('run_id')
         self._rq_client = RQOpenClient(kwargs.pop('username'), kwargs.pop('password'))
 
     def login(self):
         self._rq_client.login()
+        super(RiceQuantClient, self).login()
 
     def query(self):
-        return self._rq_client.get_day_trades(self._run_id)
+        response = self._rq_client.get_day_trades(self._run_id)
+        raw_transactions = response['resp']['trades']
+        transactions = []
+        for raw_transaction in raw_transactions:
+            transaction = RiceQuantTransaction(raw_transaction).normalize()
+            transactions.append(transaction)
+
+        return transactions

shipane_sdk/ricequant/client.py

@@ -11,9 +11,10 @@
 
 from shipane_sdk import Client
 from shipane_sdk.ap import APCronParser
-from shipane_sdk.jobs.joinquant_following import JoinQuantFollowingJob
 from shipane_sdk.jobs.new_stock_purchase import NewStockPurchaseJob
-from shipane_sdk.jobs.ricequant_following import RiceQuantFollowingJob
+from shipane_sdk.jobs.online_quant_following import OnlineQuantFollowingJob
+from shipane_sdk.joinquant.client import JoinQuantClient
+from shipane_sdk.ricequant.client import RiceQuantClient
 
 if six.PY2:
     ConfigParser = configparser.RawConfigParser
@@ -34,10 +35,16 @@ def __init__(self):
         self._client = Client(host=self._config.get('ShiPanE', 'host'),
                               port=self._config.get('ShiPanE', 'port'),
                               key=self._config.get('ShiPanE', 'key'))
+        self._jq_client = JoinQuantClient(username=self._config.get('JoinQuant', 'username'),
+                                          password=self._config.get('JoinQuant', 'password'),
+                                          backtest_id=self._config.get('JoinQuant', 'backtest_id'))
+        self._rq_client = RiceQuantClient(username=self._config.get('RiceQuant', 'username'),
+                                          password=self._config.get('RiceQuant', 'password'),
+                                          run_id=self._config.get('RiceQuant', 'run_id'))
 
         self._new_stock_purchase_job = NewStockPurchaseJob(self._config, self._client)
-        self._joinquant_following_job = JoinQuantFollowingJob(self._config, self._client)
-        self._ricequant_following_job = RiceQuantFollowingJob(self._config, self._client)
+        self._jq_following_job = OnlineQuantFollowingJob(self._client, self._jq_client, 'JoinQuantFollowingJob')
+        self._rq_following_job = OnlineQuantFollowingJob(self._client, self._rq_client, 'RiceQuantFollowingJob')
 
     def start(self):
         scheduler = BackgroundScheduler()
@@ -49,14 +56,16 @@ def start(self):
             self._log.warning('New stock purchase job is not enabled')
 
         if self._config.getboolean('JoinQuant', 'enabled'):
-            scheduler.add_job(self._joinquant_following_job,
-                              APCronParser.parse(self._config.get('JoinQuant', 'schedule')))
+            scheduler.add_job(self._jq_following_job,
+                              APCronParser.parse(self._config.get('JoinQuant', 'schedule')),
+                              None, None, None, self._jq_following_job.name)
         else:
             self._log.warning('JoinQuant following job is not enabled')
 
         if self._config.getboolean('RiceQuant', 'enabled'):
-            scheduler.add_job(self._ricequant_following_job,
-                              APCronParser.parse(self._config.get('RiceQuant', 'schedule')))
+            scheduler.add_job(self._rq_following_job,
+                              APCronParser.parse(self._config.get('RiceQuant', 'schedule')),
+                              None, None, None, self._rq_following_job.name)
         else:
             self._log.warning('RiceQuant following job is not enabled')
 

shipane_sdk/scheduler.py

@@ -19,11 +19,11 @@ def setUp(self):
         config = ConfigParser()
         dir_path = os.path.dirname(os.path.realpath(__file__))
         config.read('{}/../../config/config.ini'.format(dir_path))
-        self._jqClient = JoinQuantClient(username=config.get('JoinQuant', 'username'),
-                                         password=config.get('JoinQuant', 'password'),
-                                         backtest_id=config.get('JoinQuant', 'backtestId'))
+        self._jq_client = JoinQuantClient(username=config.get('JoinQuant', 'username'),
+                                          password=config.get('JoinQuant', 'password'),
+                                          backtest_id=config.get('JoinQuant', 'backtestId'))
 
     def test_query(self):
-        self._jqClient.login()
-        transaction_detail = self._jqClient.query()
-        self.assertTrue('data' in transaction_detail)
+        self._jq_client.login()
+        transactions = self._jq_client.query()
+        self.assertTrue(isinstance(transactions, list))

tests/shipane_sdk/joinquant/test_client.py

@@ -19,11 +19,11 @@ def setUp(self):
         config = ConfigParser()
         dir_path = os.path.dirname(os.path.realpath(__file__))
         config.read('{}/../../config/config.ini'.format(dir_path))
-        self._rqClient = RiceQuantClient(username=config.get('RiceQuant', 'username'),
-                                         password=config.get('RiceQuant', 'password'),
-                                         run_id=config.get('RiceQuant', 'run_id'))
+        self._rq_client = RiceQuantClient(username=config.get('RiceQuant', 'username'),
+                                          password=config.get('RiceQuant', 'password'),
+                                          run_id=config.get('RiceQuant', 'run_id'))
 
     def test_query(self):
-        self._rqClient.login()
-        response = self._rqClient.query()
-        self.assertTrue('resp' in response)
+        self._rq_client.login()
+        transactions = self._rq_client.query()
+        self.assertTrue(isinstance(transactions, list))