Merge pull request #686 from ricequant/dev-rqsdk-493

定时器适配期货交易场景；更新测试数据

rqalpha/mod/rqalpha_mod_sys_scheduler/scheduler.py

@@ -27,22 +27,28 @@
 from rqalpha.core.events import EVENT
 from inspect import signature
 from rqalpha.utils.exception import patch_user_exc, ModifyExceptionFromType
+from rqalpha.utils.logger import system_log
+from rqalpha.utils.datetime_func import convert_dt_to_int, convert_date_to_int
 
 
 def market_close(hour=0, minute=0):
+    if Environment.get_instance().config.base.accounts.get("STOCK") is None:
+        system_log.warning("market_close using in stock market")
     minutes_since_midnight = 15 * 60 - hour * 60 - minute
-    if minutes_since_midnight < 13 * 60:
-        minutes_since_midnight -= 90
     return minutes_since_midnight
 
 
 def market_open(hour=0, minute=0):
+    if Environment.get_instance().config.base.accounts.get("STOCK") is None:
+        system_log.warning("market_open using in stock market")
     minutes_since_midnight = 9 * 60 + 31 + hour * 60 + minute
-    if minutes_since_midnight > 11 * 60 + 30:
-        minutes_since_midnight += 90
     return minutes_since_midnight
 
 
+def physical_time(hour=0, minute=0):
+    return hour * 60 + minute
+
+
 def _verify_function(name, func):
     if not callable(func):
         raise patch_user_exc(ValueError('scheduler.{}: func should be callable'.format(name)))
@@ -58,8 +64,8 @@ def __init__(self, frequency):
         self._today = None        # type: Optional[datetime.date]
         self._this_week = None    # type: Optional[List[datetime.date]]
         self._this_month = None   # type: Optional[List[datetime.date]]
-        self._last_minute = 0     # type: Optional[int]
-        self._current_minute = 0  # type: Optional[int]
+        self._last_dt = 0         # type: Optional[int]
+        self._current_dt = 0      # type: Optional[int]
         self._stage = None
         self._frequency = frequency
         self._trading_calendar = None
@@ -106,11 +112,22 @@ def _is_nth_trading_day_in_month(self, n):
             return False
 
     def _should_trigger(self, n):
-        # 非股票交易时间段不触发
-        if self._current_minute < 9*60+31 or self._current_minute > 15*60:
+        if self._stage == "before_trading":
+            # 只处理交易时间段内的定时任务
             return False
-
-        return self._last_minute < n <= self._current_minute
+        if self._frequency == "1d":
+            # 日频直接触发
+            return True
+        if n > self.ucontext.now.hour * 60 + self.ucontext.now.minute:
+            """
+            当前位置主要是针对股票账户下的 15 - 24 点设置的时间点的延后执行。
+            举例：用户设置 n 为 21:00，在当天收盘后(15:00)，下一个bar是第二天的09:31，则 dt 需要表示为当天的21:00。所以需要在日期上 - 1。
+            期货交易场景下早盘也会进入，但会被 self._last_dt < df 条件所限制
+            """
+            dt = convert_date_to_int(self.ucontext.now - datetime.timedelta(days=1)) + n // 60 * 10000 + n % 60 * 100
+        else:
+            dt = convert_date_to_int(self.ucontext.now) + n // 60 * 10000 + n % 60 * 100
+        return self._last_dt < dt <= self._current_dt
 
     def _is_before_trading(self):
         return self._stage == 'before_trading'
@@ -123,8 +140,8 @@ def _time_rule_for(self, time_rule):
             raise patch_user_exc(ValueError(
                 'invalid time_rule, "before_trading" or int expected, got {}'.format(repr(time_rule))
             ))
-
-        time_rule = time_rule if time_rule else self._minutes_since_midnight(9, 31)
+        # 期货交易的交易时段存在0点
+        time_rule = time_rule if time_rule is not None else self._minutes_since_midnight(9, 31)
         return lambda: self._should_trigger(time_rule)
 
     @ExecutionContext.enforce_phase(EXECUTION_PHASE.ON_INIT)
@@ -183,8 +200,6 @@ def next_day_(self, event):
             return
 
         self._today = Environment.get_instance().trading_dt.date()
-        self._last_minute = 0
-        self._current_minute = 0
         if not self._this_week or self._today > self._this_week[-1]:
             self._fill_week()
         if not self._this_month or self._today > self._this_month[-1]:
@@ -196,13 +211,13 @@ def _minutes_since_midnight(hour, minute):
 
     def next_bar_(self, event):
         bars = event.bar_dict
-        self._current_minute = self._minutes_since_midnight(self.ucontext.now.hour, self.ucontext.now.minute)
+        self._current_dt = convert_dt_to_int(self.ucontext.now)
         for day_rule, time_rule, func in self._registry:
             if day_rule() and time_rule():
                 with ExecutionContext(EXECUTION_PHASE.SCHEDULED):
                     with ModifyExceptionFromType(EXC_TYPE.USER_EXC):
                         func(self.ucontext, bars)
-        self._last_minute = self._current_minute
+        self._last_dt = self._current_dt
 
     def before_trading_(self, event):
         self._stage = 'before_trading'
@@ -212,6 +227,12 @@ def before_trading_(self, event):
                     with ModifyExceptionFromType(EXC_TYPE.USER_EXC):
                         func(self.ucontext, None)
         self._stage = None
+        if self._last_dt == 0:
+            # 回测刚开始时将_last_df定义为前一个交易日的收盘时间，为0时，期货夜盘会触发前一个交易日早盘的任务
+            if self._today == self.ucontext.now.date():
+                self._last_dt = convert_date_to_int(self.ucontext.now - datetime.timedelta(days=1)) + 15 * 10000 + 30 * 100
+            else:
+                self._last_dt = convert_date_to_int(self.ucontext.now) + 15 * 10000 + 30 * 100
 
     def _fill_week(self):
         weekday = self._today.isoweekday()
@@ -236,7 +257,16 @@ def _fill_month(self):
     def set_state(self, state):
         r = json.loads(state.decode('utf-8'))
         self._today = parse(r['today']).date()
-        self._last_minute = r['last_minute']
+        if r.get("last_dt"):
+            self._last_dt = r['last_dt']
+        elif r.get("last_minute"):
+            # 保持向前兼容
+            if r['last_minute'] > 15 * 60:
+                # 针对期货交易夜盘
+                dt = convert_date_to_int(self._today - datetime.timedelta(days=1)) + r['last_minute'] // 60 * 10000 + r['last_minute'] % 60 * 100
+            else:
+                dt = convert_date_to_int(self._today) + r['last_minute'] // 60 * 10000 + r['last_minute'] % 60 * 100
+            self._last_dt = dt
         self._fill_month()
         self._fill_week()
 
@@ -246,5 +276,5 @@ def get_state(self):
 
         return json.dumps({
             'today': self._today.strftime('%Y-%m-%d'),
-            'last_minute': self._last_minute
+            'last_dt': self._last_dt
         }).encode('utf-8')