Merge pull request #998 from ricequant/develop

release 6.1.4

Author: Cuizi7

CHANGELOG.rst

@@ -2,6 +2,30 @@
 CHANGELOG
 ==================
 
+6.1.4
+==================
+
+下单接口变更：
+- `order_target_portfolio_smart` API 支持 ETF/LOF/REITs 品种
+- `order_target_portfolio` API 支持 REITs 品种
+
+涨跌停边界撮合修正：
+- 是否触及涨跌停判断修改为按 tick_size 做容差处理
+- 获取到的涨跌停价为 NaN 时，认为标的没有涨跌停限制
+
+
+6.1.3
+==================
+
+- 修复 `all_instruments` 获取数据确实的问题
+
+
+6.1.2
+==================
+
+- 修复更新 bundle 数据时，由于数据量过大导致断开与 rqdata 的连接
+
+
 6.1.0
 ==================
 

rqalpha/mod/rqalpha_mod_sys_accounts/api/api_stock.py

@@ -319,7 +319,7 @@ def order_lots(
 
 
 ORDER_TARGET_PORTFOLIO_SUPPORTED_INS_TYPES = {
-    INSTRUMENT_TYPE.CS, INSTRUMENT_TYPE.ETF, INSTRUMENT_TYPE.LOF, INSTRUMENT_TYPE.INDX, INSTRUMENT_TYPE.CONVERTIBLE
+    INSTRUMENT_TYPE.CS, INSTRUMENT_TYPE.ETF, INSTRUMENT_TYPE.LOF, INSTRUMENT_TYPE.INDX, INSTRUMENT_TYPE.CONVERTIBLE, INSTRUMENT_TYPE.REITs
 }
 
 
@@ -384,7 +384,7 @@ def order_target_portfolio(
             ).format(id_or_ins, type(id_or_ins)))
         if ins.type not in ORDER_TARGET_PORTFOLIO_SUPPORTED_INS_TYPES:
             raise RQInvalidArgument(_(
-                "function order_target_portfolio: invalid instrument type, excepted CS/ETF/LOF/INDX, got {}"
+                "function order_target_portfolio: invalid instrument type, excepted CS/ETF/LOF/INDX/Convertible/REITs, got {}"
             ).format(ins.order_book_id))
         order_book_id = ins.order_book_id
         last_price = env.data_proxy.get_last_price(order_book_id)

rqalpha/mod/rqalpha_mod_sys_accounts/api/order_target_portfolio.py

@@ -28,6 +28,7 @@
 from rqalpha.utils.exception import RQApiNotSupportedError, RQInvalidArgument
 from rqalpha.utils.functools import lru_cache
 from rqalpha.utils.i18n import gettext as _, lazy_gettext
+from rqalpha.utils.price_limits import reaches_limit_down_vectorized, reaches_limit_up_vectorized
 
 
 class AdjustingResult(NamedTuple):
@@ -72,6 +73,15 @@ class DenialReason(CommentedEnum):
     closable_exceeded = 'closable_exceeded', lazy_gettext('Order creation failed: insufficient closable position')
 
 
+SUPPORTED_INSTRUMENT_TYPES = {
+    INSTRUMENT_TYPE.CS,
+    INSTRUMENT_TYPE.CONVERTIBLE,
+    INSTRUMENT_TYPE.ETF,
+    INSTRUMENT_TYPE.LOF,
+    INSTRUMENT_TYPE.REITs,
+}
+
+
 class OrderTargetPortfolio:
     def __init__(
         self,
@@ -101,10 +111,11 @@ def __init__(
 
         instruments = env.data_proxy.get_active_instruments(index, env.trading_dt)
         for i in instruments.values():
-            if i.type != INSTRUMENT_TYPE.CS:
+            if i.type not in SUPPORTED_INSTRUMENT_TYPES:
                 raise RQApiNotSupportedError(_('instrument type {} is not supported').format(i.type))
 
         self._market = Series({i.order_book_id: i.market for i in instruments.values()}, dtype='object')
+        self._tick_sizes = Series({i: env.data_proxy.get_tick_size(i) for i in index}, dtype=float)
         self._min_qty = Series(
             {i.order_book_id: i.min_order_quantity for i in instruments.values()},
             dtype='int64',
@@ -136,25 +147,19 @@ def __init__(
             else:
                 prev_date = env.data_proxy.get_previous_trading_date(env.trading_dt)
                 for order_book_id in index:
-                    bars = env.data_proxy.history_bars(order_book_id, 1, '1d', ['close'], prev_date)
-                    if bars is None:
+                    bar = env.data_proxy.get_bar(order_book_id, prev_date, '1d')
+                    if bar.isnan:
                         raise RuntimeError('missing valuation prices: {}'.format(order_book_id))
-                    self._prices.loc[order_book_id, 'last'] = bars['close'][0]
+                    self._prices.loc[order_book_id, 'last'] = bar.close
         elif phase == EXECUTION_PHASE.ON_BAR:
             if env.config.base.frequency == '1d':
                 # TODO：根据算法时间选择最近的分钟线作为估值
                 # 当前先选择开盘价
                 for order_book_id in index:
-                    bars = env.data_proxy.history_bars(
-                        order_book_id,
-                        1,
-                        '1d',
-                        ['open', 'limit_up', 'limit_down'],
-                        env.trading_dt,
-                    )
-                    if bars is None:
+                    bar = env.data_proxy.get_bar(order_book_id, env.trading_dt, '1d')
+                    if bar.isnan:
                         raise RuntimeError('missing valuation prices: {}'.format(order_book_id))
-                    self._prices.loc[order_book_id] = list(bars[0])
+                    self._prices.loc[order_book_id] = [bar.open, bar.limit_up, bar.limit_down]
                 if valuation_prices is not None:
                     self._prices['last'] = valuation_prices
             elif env.config.base.frequency == '1m':
@@ -212,8 +217,8 @@ def _calc_adjusting(
         denials[DenialReason.suspended_sell] = (diff < 0) & self._suspended
         denials[DenialReason.no_price] = prices.isna() & (diff != 0)
 
-        limit_up = ~(limit_up.isna()) & (prices >= limit_up)
-        limit_down = ~(limit_down.isna()) & (prices <= limit_down)
+        limit_up = reaches_limit_up_vectorized(prices, limit_up, self._tick_sizes)
+        limit_down = reaches_limit_down_vectorized(prices, limit_down, self._tick_sizes)
         if direction == POSITION_DIRECTION.LONG:
             # 涨停不能开、跌停不能平
             limit_buy = (diff > 0) & limit_up
@@ -304,6 +309,7 @@ def __call__(self, direction: POSITION_DIRECTION = POSITION_DIRECTION.LONG) -> A
             safety -= min(max(proportion_diff / 10, 0.0001), 0.002)
         return AdjustingResult(adjustments=last_diff, denials=self._format_denials(last_denials))
 
+
 @export_as_api
 @ExecutionContext.enforce_phase(
     EXECUTION_PHASE.OPEN_AUCTION,

rqalpha/mod/rqalpha_mod_sys_risk/validators/price_validator.py

@@ -17,7 +17,6 @@
 from rqalpha.const import ORDER_TYPE, POSITION_EFFECT
 from rqalpha.model.order import Order
 from rqalpha.portfolio.account import Account
-from rqalpha.utils.logger import user_system_log
 
 from rqalpha.utils.i18n import gettext as _
 
@@ -29,9 +28,10 @@ def __init__(self, env):
     def validate_submission(self, order: Order, account: Optional[Account] = None) -> Optional[str]:
         if (order.type != ORDER_TYPE.LIMIT) or (order.position_effect == POSITION_EFFECT.EXERCISE):
             return None
-        
+
         # FIXME: it may be better to round price in data source
         limit_up = round(self._env.price_board.get_limit_up(order.order_book_id), 4)
+        # 此处不使用 price_limits.reaches_limit，以为事前风控宜松不宜紧，要对挂单做无罪推定。
         if order.price > limit_up:
             reason = _(
                 "Order Creation Failed: limit order price {limit_price} is higher "
@@ -57,4 +57,4 @@ def validate_submission(self, order: Order, account: Optional[Account] = None) -
         return None
 
     def validate_cancellation(self, order: Order, account: Optional[Account] = None) -> Optional[str]:
-        return None
+        return None

rqalpha/mod/rqalpha_mod_sys_simulation/matcher.py

@@ -16,7 +16,6 @@
 #         详细的授权流程，请联系 public@ricequant.com 获取。
 import datetime
 from collections import defaultdict
-import math
 from rqalpha.const import MATCHING_TYPE, ORDER_TYPE, POSITION_EFFECT, SIDE
 from rqalpha.environment import Environment
 from rqalpha.core.events import EVENT, Event
@@ -25,26 +24,12 @@
 from rqalpha.model.tick import TickObject
 from rqalpha.portfolio.account import Account
 from rqalpha.utils import is_valid_price
-from rqalpha.interface import AbstractPriceBoard
+from rqalpha.utils.price_limits import reaches_limit
 from typing import Dict
 from rqalpha.utils.i18n import gettext as _
 from .slippage import SlippageDecider
 
 
-LIMIT_PRICE_VALID_THRESHOLD = 1e-7
-
-
-def _price_reaches_limit(order_book_id: str, side: SIDE, deal_price: float, price_board: AbstractPriceBoard):
-    if side == SIDE.BUY:
-        limit_price = price_board.get_limit_up(order_book_id)
-        return deal_price >= limit_price or math.isclose(deal_price, limit_price, abs_tol=LIMIT_PRICE_VALID_THRESHOLD)
-    elif side == SIDE.SELL:
-        limit_price = price_board.get_limit_down(order_book_id)
-        return deal_price <= limit_price or math.isclose(deal_price, limit_price, abs_tol=LIMIT_PRICE_VALID_THRESHOLD)
-    else:
-        raise ValueError(f"Unsupport side: {side}")
-
-
 class AbstractMatcher:
     def match(self, account, order, open_auction):
         # type: (Account, Order, bool) -> None
@@ -125,6 +110,7 @@ def match(self, account, order, open_auction):
             raise NotImplementedError
         order_book_id = order.order_book_id
         instrument = self._env.get_instrument(order_book_id)
+        tick_size = self._env.data_proxy.get_tick_size(order_book_id)
 
         deal_price = self._get_deal_price(order, open_auction)
 
@@ -154,11 +140,11 @@ def match(self, account, order, open_auction):
                 return
             # 是否限制涨跌停不成交
             if self._price_limit:
-                if _price_reaches_limit(order_book_id, order.side, deal_price, price_board):
+                if reaches_limit(order_book_id, deal_price, order.side, price_board, tick_size):
                     return
         else:
             if self._price_limit:
-                if _price_reaches_limit(order_book_id, order.side, deal_price, price_board):
+                if reaches_limit(order_book_id, deal_price, order.side, price_board, tick_size):
                     reason = _(
                         "Order Cancelled: current bar [{order_book_id}] reach the {limit_up_or_down} price."
                     ).format(order_book_id=order.order_book_id, limit_up_or_down="limit_up" if order.side == SIDE.BUY else "limit_down")
@@ -330,6 +316,7 @@ def match(self, account, order, open_auction):  # type: (Account, Order, bool) -
         # 标的信息
         order_book_id = order.order_book_id
         instrument = self._env.get_instrument(order_book_id)
+        tick_size = self._env.data_proxy.get_tick_size(order_book_id)
 
         # 获取tick数据
         _cur_tick = self._cur_tick.get(order_book_id)
@@ -375,7 +362,7 @@ def match(self, account, order, open_auction):  # type: (Account, Order, bool) -
                 return
             # 是否限制涨跌停不成交
             if self._price_limit:
-                if _price_reaches_limit(order_book_id, order.side, deal_price, price_board):
+                if reaches_limit(order_book_id, deal_price, order.side, price_board, tick_size):
                     return
             if self._liquidity_limit:
                 if order.side == SIDE.BUY and price_board.get_a1(order_book_id) == 0:
@@ -384,7 +371,7 @@ def match(self, account, order, open_auction):  # type: (Account, Order, bool) -
                     return
         else:
             if self._price_limit:
-                if _price_reaches_limit(order_book_id, order.side, deal_price, price_board):
+                if reaches_limit(order_book_id, deal_price, order.side, price_board, tick_size):
                     reason = _(
                         "Order Cancelled: current tick [{order_book_id}] reach the {limit_up_or_down} price."
                     ).format(order_book_id=order.order_book_id, limit_up_or_down="limit_up" if order.side == SIDE.BUY else "limit_down")

rqalpha/mod/rqalpha_mod_sys_simulation/signal_broker.py

@@ -24,6 +24,7 @@
 from rqalpha.utils.logger import user_system_log
 from rqalpha.utils.i18n import gettext as _
 from rqalpha.utils import is_valid_price
+from rqalpha.utils.price_limits import reaches_limit
 from rqalpha.core.events import EVENT, Event
 from rqalpha.model.trade import Trade
 from rqalpha.model.order import ALGO_ORDER_STYLES
@@ -59,6 +60,7 @@ def cancel_order(self, order):
     def _match(self, account, order):
         order_book_id = order.order_book_id
         price_board = self._env.price_board
+        tick_size = self._env.data_proxy.get_tick_size(order_book_id)
 
         last_price = price_board.get_last_price(order_book_id)
 
@@ -91,19 +93,19 @@ def _match(self, account, order):
 
         if self._price_limit:
             if order.position_effect != POSITION_EFFECT.EXERCISE:
-                if order.side == SIDE.BUY and deal_price >= price_board.get_limit_up(order_book_id):
+                if reaches_limit(
+                    order_book_id,
+                    deal_price,
+                    order.side,
+                    price_board,
+                    tick_size,
+                ):
                     order.mark_rejected(_(
-                        "Order Cancelled: current bar [{order_book_id}] reach the limit_up price."
-                    ).format(order_book_id=order.order_book_id))
-                    self._env.event_bus.publish_event(Event(
-                        EVENT.ORDER_UNSOLICITED_UPDATE, account=account, order=copy(order)
+                        "Order Cancelled: current bar [{order_book_id}] reach the {limit_up_or_down} price."
+                    ).format(
+                        order_book_id=order.order_book_id,
+                        limit_up_or_down="limit_up" if order.side == SIDE.BUY else "limit_down",
                     ))
-                    return
-
-                if order.side == SIDE.SELL and deal_price <= price_board.get_limit_down(order_book_id):
-                    order.mark_rejected(_(
-                        "Order Cancelled: current bar [{order_book_id}] reach the limit_down price."
-                    ).format(order_book_id=order.order_book_id))
                     self._env.event_bus.publish_event(Event(
                         EVENT.ORDER_UNSOLICITED_UPDATE, account=account, order=copy(order)
                     ))

rqalpha/model/instrument.py

@@ -397,8 +397,10 @@ def min_order_quantity(self):
     @cached_property
     def order_step_size(self):
         # 下单递进股数
-        if self.board_type == "KSH" or self.board_type == "BJS":
-            return 1
+        if self.type == INSTRUMENT_TYPE.CS:
+            # 对于其他资产类型 instrument 中（如可转债）可能没有 board_type 字段
+            if self.board_type == "KSH" or self.board_type == "BJS":
+                return 1
         return self.round_lot
 
     def during_call_auction(self, dt):
@@ -445,7 +447,7 @@ def tick_size(self):
         # type: () -> float
         if self.type in (INSTRUMENT_TYPE.CS, INSTRUMENT_TYPE.INDX):
             return 0.01
-        elif self.type in ("ETF", "LOF"):
+        elif self.type in ("ETF", "LOF", "REITs"):
             return 0.001
         elif self.type == INSTRUMENT_TYPE.FUTURE:
             return self._futures_tick_size_getter(self)

rqalpha/utils/price_limits.py

@@ -0,0 +1,56 @@
+from typing import Union
+
+from pandas import Series
+
+from rqalpha.const import SIDE
+from rqalpha.interface import AbstractPriceBoard
+from rqalpha.utils import is_valid_price
+
+PriceLike = Union[int, float]
+PRICE_LIMIT_TOLERANCE = 1e-6
+
+
+def reaches_limit_up(price: PriceLike, limit_up: PriceLike, tick_size: PriceLike) -> bool:
+    if not (is_valid_price(price) and is_valid_price(limit_up)):
+        return False
+    return price >= limit_up - tick_size + PRICE_LIMIT_TOLERANCE
+
+
+def reaches_limit_down(price: PriceLike, limit_down: PriceLike, tick_size: PriceLike) -> bool:
+    if not (is_valid_price(price) and is_valid_price(limit_down)):
+        return False
+    return price <= limit_down + tick_size - PRICE_LIMIT_TOLERANCE
+
+
+def reaches_limit(
+    order_book_id: str,
+    price: PriceLike,
+    side: SIDE,
+    price_board: AbstractPriceBoard,
+    tick_size: PriceLike,
+) -> bool:
+    if side == SIDE.BUY:
+        return reaches_limit_up(price, price_board.get_limit_up(order_book_id), tick_size)
+    if side == SIDE.SELL:
+        return reaches_limit_down(price, price_board.get_limit_down(order_book_id), tick_size)
+    raise ValueError("unsupport side: {}".format(side))
+
+
+def _is_valid_price_series(values: Series) -> Series:
+    return values.notna() & values.gt(0)
+
+
+def reaches_limit_up_vectorized(prices: Series, limit_up: Series, tick_sizes: Series) -> Series:
+    valid_price = _is_valid_price_series(prices)
+    valid_limit = _is_valid_price_series(limit_up)
+
+    thresholds = limit_up - tick_sizes + PRICE_LIMIT_TOLERANCE
+    return valid_price & valid_limit & prices.ge(thresholds)
+
+
+def reaches_limit_down_vectorized(prices: Series, limit_down: Series, tick_sizes: Series) -> Series:
+    valid_price = _is_valid_price_series(prices)
+    valid_limit = _is_valid_price_series(limit_down)
+
+    thresholds = limit_down + tick_sizes - PRICE_LIMIT_TOLERANCE
+    return valid_price & valid_limit & prices.le(thresholds)