rqalpha.portfolio.position 源代码

# -*- coding: utf-8 -*-
# 版权所有 2019 深圳米筐科技有限公司(下称“米筐科技”)
#
# 除非遵守当前许可,否则不得使用本软件。
#
#     * 非商业用途(非商业用途指个人出于非商业目的使用本软件,或者高校、研究所等非营利机构出于教育、科研等目的使用本软件):
#         遵守 Apache License 2.0(下称“Apache 2.0 许可”),
#         您可以在以下位置获得 Apache 2.0 许可的副本:http://www.apache.org/licenses/LICENSE-2.0。
#         除非法律有要求或以书面形式达成协议,否则本软件分发时需保持当前许可“原样”不变,且不得附加任何条件。
#
#     * 商业用途(商业用途指个人出于任何商业目的使用本软件,或者法人或其他组织出于任何目的使用本软件):
#         未经米筐科技授权,任何个人不得出于任何商业目的使用本软件(包括但不限于向第三方提供、销售、出租、出借、转让本软件、
#         本软件的衍生产品、引用或借鉴了本软件功能或源代码的产品或服务),任何法人或其他组织不得出于任何目的使用本软件,
#         否则米筐科技有权追究相应的知识产权侵权责任。
#         在此前提下,对本软件的使用同样需要遵守 Apache 2.0 许可,Apache 2.0 许可与本许可冲突之处,以本许可为准。
#         详细的授权流程,请联系 public@ricequant.com 获取。

from typing import Iterable, Tuple, Dict, Type, Union
from datetime import date
from collections import UserDict

from rqalpha.interface import AbstractPosition
from rqalpha.environment import Environment
from rqalpha.const import POSITION_EFFECT, POSITION_DIRECTION
from rqalpha.model.order import Order
from rqalpha.model.trade import Trade
from rqalpha.model.instrument import Instrument
from rqalpha.utils.i18n import gettext as _
from rqalpha.utils.repr import property_repr, PropertyReprMeta
from rqalpha.utils import is_valid_price


def new_position_meta():

    type_map = {}

    class Meta(PropertyReprMeta):
        def __new__(mcs, *args, **kwargs):
            cls = super(Meta, mcs).__new__(mcs, *args, **kwargs)
            try:
                instrument_types = cls.__instrument_types__
            except AttributeError:
                pass
            else:
                for instrument_type in instrument_types:
                    type_map[instrument_type] = cls
            return cls

    return type_map, Meta


POSITION_TYPE_MAP, PositionMeta = new_position_meta()
POSITION_PROXY_TYPE_MAP, PositionProxyMeta = new_position_meta()


[文档]class Position(AbstractPosition, metaclass=PositionMeta): __repr_properties__ = ( "order_book_id", "direction", "quantity", "market_value", "trading_pnl", "position_pnl" ) # 用于注册该 Position 类型适用的 instrument_type __instrument_types__ = [] def __new__(cls, order_book_id, direction, init_quantity=0): if cls == Position: ins_type = Environment.get_instance().data_proxy.instruments(order_book_id).type try: position_cls = POSITION_TYPE_MAP[ins_type] except KeyError: raise NotImplementedError("") return position_cls.__new__(position_cls, order_book_id, direction, init_quantity) else: return object.__new__(cls) def __init__(self, order_book_id, direction, init_quantity=0): self._env = Environment.get_instance() self._order_book_id = order_book_id self._instrument = self._env.data_proxy.instruments(order_book_id) # type: Instrument self._direction = direction self._old_quantity = init_quantity self._logical_old_quantity = 0 self._today_quantity = 0 self._avg_price = 0 self._trade_cost = 0 self._transaction_cost = 0 self._prev_close = None self._last_price = float("NaN") self._direction_factor = 1 if direction == POSITION_DIRECTION.LONG else -1 @property def order_book_id(self): # type: () -> str return self._order_book_id @property def direction(self): # type: () -> POSITION_DIRECTION return self._direction @property def quantity(self): # type: () -> int return self._old_quantity + self._today_quantity @property def transaction_cost(self): # type: () -> float return self._transaction_cost @property def avg_price(self): # type: () -> float return self._avg_price @property def trading_pnl(self): # type: () -> float trade_quantity = self._today_quantity + (self._old_quantity - self._logical_old_quantity) if trade_quantity == 0: return 0 return (trade_quantity * self.last_price - self._trade_cost) * self._direction_factor @property def position_pnl(self): # type: () -> float if self._logical_old_quantity == 0: return 0 return self._logical_old_quantity * (self.last_price - self.prev_close) * self._direction_factor @property def pnl(self): # type: () -> float """ 返回该持仓的累积盈亏 """ if self.quantity == 0: return 0 return (self.last_price - self.avg_price) * self.quantity * self._direction_factor @property def market_value(self): # type: () -> float return self.last_price * self.quantity if self.quantity != 0 else 0 @property def margin(self): # type: () -> float return 0 @property def equity(self): # type: () -> float return self.last_price * self.quantity if self.quantity != 0 else 0 @property def prev_close(self): if not is_valid_price(self._prev_close): env = Environment.get_instance() self._prev_close = env.data_proxy.get_prev_close(self._order_book_id, env.trading_dt) return self._prev_close @property def last_price(self): if self._last_price != self._last_price: env = Environment.get_instance() self._last_price = env.data_proxy.get_last_price(self._order_book_id) if self._last_price != self._last_price: raise RuntimeError(_("last price of position {} is not supposed to be nan").format(self._order_book_id)) return self._last_price @property def closable(self): # type: () -> int """ 可平仓位 """ order_quantity = sum(o for o in self._open_orders if o.position_effect in ( POSITION_EFFECT.CLOSE, POSITION_EFFECT.CLOSE_TODAY, POSITION_EFFECT.EXERCISE )) return self.quantity - order_quantity @property def today_closable(self): # type: () -> int return self._today_quantity - sum( o.unfilled_quantity for o in self._open_orders if o.position_effect == POSITION_EFFECT.CLOSE_TODAY ) def get_state(self): """""" return { "old_quantity": self._old_quantity, "logical_old_quantity": self._logical_old_quantity, "today_quantity": self._today_quantity, "avg_price": self._avg_price, "trade_cost": self._trade_cost, "transaction_cost": self._transaction_cost, "prev_close": self._prev_close } def set_state(self, state): """""" self._old_quantity = state.get("old_quantity", 0) self._logical_old_quantity = state.get("logical_old_quantity", self._old_quantity) self._today_quantity = state.get("today_quantity", 0) self._avg_price = state.get("avg_price", 0) self._trade_cost = state.get("trade_cost", 0) self._transaction_cost = state.get("transaction_cost", 0) self._prev_close = state.get("prev_close") def before_trading(self, trading_date): # type: (date) -> float # 返回该阶段导致总资金的变化量 return 0 def apply_trade(self, trade): # type: (Trade) -> float # 返回总资金的变化量 self._transaction_cost += trade.transaction_cost if trade.position_effect == POSITION_EFFECT.OPEN: if self.quantity < 0: self._avg_price = trade.last_price if self.quantity + trade.last_quantity > 0 else 0 else: cost = self.quantity * self._avg_price + trade.last_quantity * trade.last_price self._avg_price = cost / (self.quantity + trade.last_quantity) self._today_quantity += trade.last_quantity self._trade_cost += trade.last_price * trade.last_quantity return (-1 * trade.last_price * trade.last_quantity) - trade.transaction_cost elif trade.position_effect == POSITION_EFFECT.CLOSE: self._today_quantity -= max(trade.last_quantity - self._old_quantity, 0) self._old_quantity -= min(trade.last_quantity, self._old_quantity) self._trade_cost -= trade.last_price * trade.last_quantity return trade.last_price * trade.last_quantity - trade.transaction_cost else: raise NotImplementedError("{} does not support position effect {}".format( self.__class__.__name__, trade.position_effect )) def settlement(self, trading_date): # type: (date) -> float # 返回该阶段导致总资金的变化量以及反映该阶段引起其他持仓变化的虚拟交易,虚拟交易用于换代码,转股等操作 self._old_quantity += self._today_quantity self._logical_old_quantity = self._old_quantity self._today_quantity = self._trade_cost = self._transaction_cost = self._non_closable = 0 self._prev_close = self.last_price return 0 def update_last_price(self, price): self._last_price = price def calc_close_today_amount(self, trade_amount): return 0 @property def _open_orders(self): # type: () -> Iterable[Order] for order in Environment.get_instance().broker.get_open_orders(self.order_book_id): if order.position_direction == self._direction: yield order
class PositionProxy(metaclass=PositionProxyMeta): __repr_properties__ = ( "order_book_id", "positions" ) # 用于注册该 Position 类型适用的 instrument_type __instrument_types__ = [] def __init__(self, long, short): # type: (Position, Position) -> PositionProxy self._long = long self._short = short __repr__ = property_repr @property def type(self): raise NotImplementedError @property def order_book_id(self): return self._long.order_book_id @property def last_price(self): return self._long.last_price @property def market_value(self): return self._long.market_value - self._short.market_value # -- PNL 相关 @property def position_pnl(self): """ [float] 昨仓盈亏,当前交易日盈亏中来源于昨仓的部分 多方向昨仓盈亏 = 昨日收盘时的持仓 * 合约乘数 * (最新价 - 昨收价) 空方向昨仓盈亏 = 昨日收盘时的持仓 * 合约乘数 * (昨收价 - 最新价) """ return self._long.position_pnl + self._short.position_pnl @property def trading_pnl(self): """ [float] 交易盈亏,当前交易日盈亏中来源于当日成交的部分 单比买方向成交的交易盈亏 = 成交量 * (最新价 - 成交价) 单比卖方向成交的交易盈亏 = 成交量 * (成交价 - 最新价) """ return self._long.trading_pnl + self._short.trading_pnl @property def daily_pnl(self): """ [float] 当日盈亏 当日盈亏 = 昨仓盈亏 + 交易盈亏 """ return self._long.position_pnl + self._long.trading_pnl + self._short.position_pnl +\ self._short.trading_pnl - self.transaction_cost @property def pnl(self): return self._long.pnl + self._short.pnl # -- Quantity 相关 @property def open_orders(self): return Environment.get_instance().broker.get_open_orders(self.order_book_id) # -- Margin 相关 @property def margin(self): """ [float] 保证金 保证金 = 持仓量 * 最新价 * 合约乘数 * 保证金率 股票保证金 = 市值 = 持仓量 * 最新价 """ return self._long.margin + self._short.margin @property def transaction_cost(self): """ [float] 交易费率 """ return self._long.transaction_cost + self._short.transaction_cost @property def positions(self): return [self._long, self._short] @property def long(self): return self._long @property def short(self): return self._short PositionType = Type[Position] PositionProxyType = Type[PositionProxy] PositionDictType = Dict[str, Dict[POSITION_DIRECTION, Position]] class PositionProxyDict(UserDict): def __init__(self, positions): super(PositionProxyDict, self).__init__() self._positions = positions # type: PositionDictType def keys(self): return self._positions.keys() def __getitem__(self, order_book_id): position_type, position_proxy_type = self._get_position_types(order_book_id) if order_book_id not in self._positions: long = position_type(order_book_id, POSITION_DIRECTION.LONG) short = position_type(order_book_id, POSITION_DIRECTION.SHORT) else: positions = self._positions[order_book_id] long = positions[POSITION_DIRECTION.LONG] short = positions[POSITION_DIRECTION.SHORT] return position_proxy_type(long, short) def __contains__(self, item): return item in self._positions def __iter__(self): return iter(self._positions) def __len__(self): return len(self._positions) def __setitem__(self, key, value): raise TypeError("{} object does not support item assignment".format(self.__class__.__name__)) def __delitem__(self, key): raise TypeError("{} object does not support item deletion".format(self.__class__.__name__)) def __repr__(self): return repr({k: self[k] for k in self._positions.keys()}) def _get_position_types(self, order_book_id): # type: (str) -> Tuple[Type[Position], Type[PositionProxy]] instrument_type = Environment.get_instance().data_proxy.instruments(order_book_id).type position_type = POSITION_TYPE_MAP.get(instrument_type, Position) position_proxy_type = POSITION_PROXY_TYPE_MAP.get(instrument_type, PositionProxy) return position_type, position_proxy_type