# -*- 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 rqalpha.utils.functools import lru_cache
from typing import Dict, Union, Callable, List, Tuple
from itertools import chain
import six
import jsonpickle
import numpy as np
from rqalpha.environment import Environment
from rqalpha.const import DAYS_CNT, DEFAULT_ACCOUNT_TYPE, POSITION_DIRECTION
from rqalpha.utils import merge_dicts
from rqalpha.utils.repr import PropertyReprMeta
from rqalpha.core.events import EVENT
from rqalpha.model.order import OrderStyle, Order
from rqalpha.interface import AbstractPosition
from .account import Account
from .position import PositionType, PositionProxyType
OrderApiType = Callable[[str, Union[int, float], OrderStyle, bool], List[Order]]
[文档]class Portfolio(object, metaclass=PropertyReprMeta):
"""
投资组合，策略所有账户的集合
"""
__repr_properties__ = (
"total_value", "unit_net_value", "daily_pnl", "daily_returns", "total_returns", "annualized_returns", "accounts"
)
def __init__(self, starting_cash, init_positions):
# type: (Dict[str, float], List[Tuple[str, int]]) -> Portfolio
self._static_unit_net_value = 1
self._last_unit_net_value = 1
account_args = {}
for account_type, cash in starting_cash.items():
account_args[account_type] = {"type": account_type, "total_cash": cash, "init_positions": {}}
for order_book_id, quantity in init_positions:
account_type = self.get_account_type(order_book_id)
if account_type in account_args:
account_args[account_type]["init_positions"][order_book_id] = quantity
self._accounts = {account_type: Account(**args) for account_type, args in account_args.items()}
self._units = sum(account.total_value for account in six.itervalues(self._accounts))
self._register_event()
def get_state(self):
return jsonpickle.encode({
'static_unit_net_value': self._static_unit_net_value,
'last_unit_net_value': self._last_unit_net_value,
'units': self._units,
'accounts': {
name: account.get_state() for name, account in self._accounts.items()
}
}).encode('utf-8')
def set_state(self, state):
state = state.decode('utf-8')
value = jsonpickle.decode(state)
self._static_unit_net_value = value['static_unit_net_value']
self._last_unit_net_value = value.get('last_unit_net_value', self._static_unit_net_value)
self._units = value['units']
for k, v in value['accounts'].items():
self._accounts[k].set_state(v)
def get_positions(self):
return list(chain(*(a.get_positions() for a in six.itervalues(self._accounts))))
def get_position(self, order_book_id, direction):
# type: (str, POSITION_DIRECTION) -> AbstractPosition
account = self._accounts[self.get_account_type(order_book_id)]
return account.get_position(order_book_id, direction)
@classmethod
def get_account_type(cls, order_book_id):
instrument = Environment.get_instance().data_proxy.instruments(order_book_id)
return instrument.account_type
def get_account(self, order_book_id):
return self._accounts[self.get_account_type(order_book_id)]
@property
def accounts(self):
# type: () -> Dict[DEFAULT_ACCOUNT_TYPE, Account]
"""
账户字典
"""
return self._accounts
@property
def stock_account(self):
"""
[StockAccount] 股票账户
"""
return self._accounts.get(DEFAULT_ACCOUNT_TYPE.STOCK.name, None)
@property
def future_account(self):
"""
[FutureAccount] 期货账户
"""
return self._accounts.get(DEFAULT_ACCOUNT_TYPE.FUTURE.name, None)
@property
def start_date(self):
"""
[datetime.datetime] 策略投资组合的开始日期
"""
return Environment.get_instance().config.base.start_date
@property
def units(self):
"""
[float] 份额
"""
return self._units
@property
def unit_net_value(self):
"""
[float] 实时净值
"""
if self._units == 0:
return np.nan
return self.total_value / self._units
@property
def static_unit_net_value(self):
"""
[float] 昨日净值
"""
return self._static_unit_net_value
@property
def daily_pnl(self):
"""
[float] 当日盈亏
"""
return sum(account.daily_pnl for account in six.itervalues(self._accounts))
@property
def daily_returns(self):
"""
[float] 当前最新一天的日收益
"""
return np.nan if self._static_unit_net_value == 0 else self.unit_net_value / self._static_unit_net_value - 1
@property
def total_returns(self):
"""
[float] 累计收益率
"""
return self.unit_net_value - 1
@property
def annualized_returns(self):
"""
[float] 累计年化收益率
"""
if self.unit_net_value <= 0:
return -1
env = Environment.get_instance()
date_count = float(env.data_proxy.count_trading_dates(env.config.base.start_date, env.trading_dt.date()))
return self.unit_net_value ** (DAYS_CNT.TRADING_DAYS_A_YEAR / date_count) - 1
@property
def total_value(self):
"""
[float]总权益
"""
return sum(account.total_value for account in six.itervalues(self._accounts))
@property
def portfolio_value(self):
"""
[Deprecated] 总权益
"""
return self.total_value
@property
@lru_cache()
def positions(self):
"""
[dict] 持仓字典
"""
return MixedPositions(self._accounts)
@property
def cash(self):
"""
[float] 可用资金
"""
return sum(account.cash for account in six.itervalues(self._accounts))
@property
def transaction_cost(self):
"""
[float] 交易成本（税费）
"""
return sum(account.transaction_cost for account in six.itervalues(self._accounts))
@property
def market_value(self):
"""
[float] 市值
"""
return sum(account.market_value for account in six.itervalues(self._accounts))
@property
def pnl(self):
"""
[float] 收益
"""
return (self.unit_net_value - 1) * self.units
@property
def starting_cash(self):
"""
[float] 初始资金
"""
return self.units
@property
def frozen_cash(self):
"""
[float] 冻结资金
"""
return sum(account.frozen_cash for account in six.itervalues(self._accounts))
def _pre_before_trading(self, _):
if not np.isnan(self.unit_net_value):
self._static_unit_net_value = self.unit_net_value
else:
self._static_unit_net_value = self._last_unit_net_value
def _post_settlement(self, event):
self._last_unit_net_value = self.unit_net_value
def _register_event(self):
event_bus = Environment.get_instance().event_bus
event_bus.prepend_listener(EVENT.PRE_BEFORE_TRADING, self._pre_before_trading)
event_bus.prepend_listener(EVENT.POST_SETTLEMENT, self._post_settlement)
class MixedPositions(dict):
def __init__(self, accounts):
super(MixedPositions, self).__init__()
self._accounts = accounts
def __missing__(self, key):
account_type = Portfolio.get_account_type(key)
for a_type in self._accounts:
if a_type == account_type:
return self._accounts[a_type].positions[key]
return None
def __contains__(self, item):
return item in self.keys()
def __repr__(self):
keys = []
for account in six.itervalues(self._accounts):
keys += account.positions.keys()
return str(sorted(keys))
def __len__(self):
return sum(len(account.positions) for account in six.itervalues(self._accounts))
def __iter__(self):
keys = []
for account in six.itervalues(self._accounts):
keys += account.positions.keys()
for key in sorted(keys):
yield key
def items(self):
items = merge_dicts(*[account.positions.items() for account in six.itervalues(self._accounts)])
for k in sorted(items.keys()):
yield k, items[k]
def keys(self):
keys = []
for account in six.itervalues(self._accounts):
keys += list(account.positions.keys())
return sorted(keys)