Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@staticmethod
def _frozen_cash_of_order(order):
order_cost = order.frozen_price * order.quantity if order.side == SIDE.BUY else 0
return order_cost + Environment.get_instance().get_order_transaction_cost(DEFAULT_ACCOUNT_TYPE.STOCK, order)
def _stock_validator(self, account, order):
if order.side == SIDE.SELL:
return True
frozen_value = order.frozen_price * order.quantity
cost_money = frozen_value + self._env.get_order_transaction_cost(DEFAULT_ACCOUNT_TYPE.STOCK, order)
if cost_money <= account.cash:
return True
user_system_log.warn(
_("Order Creation Failed: not enough money to buy {order_book_id}, needs {cost_money:.2f}, "
"cash {cash:.2f}").format(
order_book_id=order.order_book_id,
cost_money=cost_money,
cash=account.cash,
)
)
return False
def _to_account_record(self, date, account):
data = {
'date': date,
'cash': self._safe_convert(account.cash),
'transaction_cost': self._safe_convert(account.transaction_cost),
'market_value': self._safe_convert(account.market_value),
'total_value': self._safe_convert(account.total_value),
}
for f in self.ACCOUNT_FIELDS_MAP[account.type]:
data[f] = self._safe_convert(getattr(account, f))
return data
POSITION_FIELDS_MAP = {
DEFAULT_ACCOUNT_TYPE.STOCK.name: [
'quantity', 'last_price', 'avg_price', 'market_value'
],
DEFAULT_ACCOUNT_TYPE.FUTURE.name: [
'margin', 'margin_rate', 'contract_multiplier', 'last_price',
'buy_pnl', 'buy_margin', 'buy_quantity', 'buy_avg_open_price',
'sell_pnl', 'sell_margin', 'sell_quantity', 'sell_avg_open_price'
],
DEFAULT_ACCOUNT_TYPE.BOND.name: [
'quantity', 'last_price', 'avg_price', 'market_value'
],
}
def _to_position_record(self, date, order_book_id, position):
data = {
'order_book_id': order_book_id,
'symbol': self._symbol(order_book_id),
def value_percent(self):
"""
[float] 获得该持仓的实时市场价值在总投资组合价值中所占比例,取值范围[0, 1]
"""
accounts = Environment.get_instance().portfolio.accounts
if DEFAULT_ACCOUNT_TYPE.STOCK.name not in accounts:
return 0
total_value = accounts[DEFAULT_ACCOUNT_TYPE.STOCK.name].total_value
return 0 if total_value == 0 else self.market_value / total_value
def can_submit_order(self, order, account=None):
if account is None:
return True
if account.type == DEFAULT_ACCOUNT_TYPE.STOCK.name:
return self._stock_validator(account, order)
elif account.type == DEFAULT_ACCOUNT_TYPE.FUTURE.name:
return self._future_validator(account, order)
else:
return True
def get_position(self, order_book_id, direction):
account_type = get_account_type(order_book_id)
if account_type == DEFAULT_ACCOUNT_TYPE.STOCK.name:
return StockSimulationBookingPosition(
self._positions[order_book_id], direction
)
elif account_type == DEFAULT_ACCOUNT_TYPE.FUTURE.name:
return FutureSimulationBookingPosition(
self._positions[order_book_id], direction
)
else:
raise NotImplementedError
def start_up(self, env, mod_config):
if mod_config.commission_multiplier < 0:
raise patch_user_exc(ValueError(_(u"invalid commission multiplier value: value range is [0, +∞)")))
if env.config.base.market == MARKET.CN:
env.set_transaction_cost_decider(DEFAULT_ACCOUNT_TYPE.STOCK, CNStockTransactionCostDecider(
mod_config.commission_multiplier, mod_config.cn_stock_min_commission
))
env.set_transaction_cost_decider(DEFAULT_ACCOUNT_TYPE.FUTURE, CNFutureTransactionCostDecider(
mod_config.commission_multiplier
))
elif env.config.base.market == MARKET.HK:
env.set_transaction_cost_decider(DEFAULT_ACCOUNT_TYPE.STOCK, HKStockTransactionCostDecider(
mod_config.commission_multiplier, mod_config.hk_stock_min_commission
))
def get_future_trading_points(env, trading_date, frequency):
if frequency == "1m":
trading_minutes = set()
universe = env.get_universe()
for order_book_id in universe:
if get_account_type(order_book_id) == DEFAULT_ACCOUNT_TYPE.STOCK:
continue
trading_minutes.update(env.data_proxy.get_trading_minutes_for(order_book_id, trading_date))
return set([convert_int_to_datetime(minute) for minute in trading_minutes])
# TODO future hours
return set()
def stock_account(self):
"""
[StockAccount] 股票账户
"""
return self._accounts.get(DEFAULT_ACCOUNT_TYPE.STOCK.name, None)