Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"old_quantity": 1, "today_quantity": 3
}},
"short_positions": {
"TF1812": {
"today_quantity": 4
}
}
}).encode('utf-8'))
self.assertPositions({
(POSITION_DIRECTION.LONG, "RB1812", 3, 1),
(POSITION_DIRECTION.SHORT, "TF1812", 4, 0)
})
self.env.event_bus.publish_event(Event(EVENT.TRADE, trade=Trade.__from_create__(
0, 0, 2, SIDE.SELL, POSITION_EFFECT.OPEN, "RB1812"
)))
self.assertPositions({
(POSITION_DIRECTION.LONG, "RB1812", 3, 1),
(POSITION_DIRECTION.SHORT, "RB1812", 2, 0),
(POSITION_DIRECTION.SHORT, "TF1812", 4, 0)
})
self.env.event_bus.publish_event(Event(EVENT.TRADE, trade=Trade.__from_create__(
0, 0, 3, SIDE.SELL, POSITION_EFFECT.CLOSE, "RB1812"
)))
self.assertPositions({
(POSITION_DIRECTION.LONG, "RB1812", 1, 0),
(POSITION_DIRECTION.SHORT, "RB1812", 2, 0),
(POSITION_DIRECTION.SHORT, "TF1812", 4, 0)
})
with self.mock_data_proxy_method("instruments", mock_get_instrument):
def buy_open_order_quantity(self):
"""
[int] 买方向挂单量
"""
return sum(order.unfilled_quantity for order in self.open_orders if
order.side == SIDE.BUY and order.position_effect == POSITION_EFFECT.OPEN)
def _cal_realized_pnl(self, cost_price, trade_price, side, consumed_quantity):
if side == SIDE.BUY:
return (cost_price - trade_price) * consumed_quantity * self.contract_multiplier
else:
return (trade_price - cost_price) * consumed_quantity * self.contract_multiplier
def validate_available_position(self, order, position):
if not self.config.available_position:
return True
if order.position_effect != POSITION_EFFECT.CLOSE:
return True
if order.side == SIDE.BUY and order.quantity > position.closable_sell_quantity:
order.mark_rejected(_(
"Order Rejected: not enough securities {order_book_id} to buy close, target sell quantity is {quantity}, sell_closable_quantity {closable}").format(
order_book_id=order.order_book_id,
quantity=order.quantity,
closable=position.closable_sell_quantity,
))
return False
elif order.side == SIDE.SELL and order.quantity > position.closable_buy_quantity:
order.mark_rejected(_(
"Order Rejected: not enough securities {order_book_id} to sell close, target sell quantity is {quantity}, buy_closable_quantity {closable}").format(
order_book_id=order.order_book_id,
quantity=order.quantity,
closable=position.closable_buy_quantity,
))
return False
return True
def set_state(self, d):
self._order_id = d['order_id']
if 'secondary_order_id' in d:
self._secondary_order_id = d['secondary_order_id']
self._calendar_dt = d['calendar_dt']
self._trading_dt = d['trading_dt']
self._order_book_id = d['order_book_id']
self._quantity = d['quantity']
self._side = self._str_to_enum(SIDE, d['side'])
if d['position_effect'] is None:
self._position_effect = None
else:
self._position_effect = self._str_to_enum(POSITION_EFFECT, d['position_effect'])
self._message = d['message']
self._filled_quantity = d['filled_quantity']
self._status = self._str_to_enum(ORDER_STATUS, d['status'])
self._frozen_price = d['frozen_price']
self._type = self._str_to_enum(ORDER_TYPE, d['type'])
self._transaction_cost = d['transaction_cost']
self._avg_price = d['avg_price']
def validate_available_position(self, order, position):
if not self.config.available_position:
return True
if order.side != SIDE.SELL:
return True
if order.quantity > position.sellable:
order.mark_rejected(_(
"Order Rejected: not enough stock {order_book_id} to sell, you want to sell {quantity}, sellable {sellable}").format(
order_book_id=order.order_book_id,
quantity=order.quantity,
sellable=position.sellable,
))
return False
return True
@staticmethod
def _stock_validator(account, order):
if order.side != SIDE.SELL:
return True
position = account.positions[order.order_book_id]
if order.quantity <= position.sellable:
return True
user_system_log.warn(_(
"Order Creation Failed: not enough stock {order_book_id} to sell, you want to sell {quantity},"
" sellable {sellable}").format(
order_book_id=order.order_book_id,
quantity=order.quantity,
sellable=position.sellable,
))
return False
raise RQInvalidArgument(_(u"Index Future contracts[99] are not supported in paper trading."))
price = env.get_last_price(order_book_id)
if not is_valid_price(price):
user_system_log.warn(
_(u"Order Creation Failed: [{order_book_id}] No market data").format(order_book_id=order_book_id)
)
return
amount = int(amount)
env = Environment.get_instance()
orders = []
if position_effect == POSITION_EFFECT.CLOSE:
if side == SIDE.BUY:
if env.portfolio:
position = env.portfolio.positions[order_book_id]
sell_quantity, sell_old_quantity = position.sell_quantity, position.sell_old_quantity
else:
position = env.booking.get_position(order_book_id, POSITION_DIRECTION.SHORT)
sell_quantity, sell_old_quantity = position.quantity, position.old_quantity
# 如果平仓量大于持仓量,则 Warning 并 取消订单创建
if amount > sell_quantity:
user_system_log.warn(
_(u"Order Creation Failed: close amount {amount} is larger than position "
u"quantity {quantity}").format(amount=amount, quantity=sell_quantity)
)
return []
sell_old_quantity = sell_old_quantity
if amount > sell_old_quantity:
def _get_direction(self, side, position_effect):
direction = None
if position_effect in (POSITION_EFFECT.CLOSE, POSITION_EFFECT.CLOSE_TODAY):
if side == SIDE.BUY:
direction = POSITION_DIRECTION.SHORT
elif side == SIDE.SELL:
direction = POSITION_DIRECTION.LONG
elif position_effect == POSITION_EFFECT.OPEN:
if side == SIDE.BUY:
direction = POSITION_DIRECTION.LONG
elif side == SIDE.SELL:
direction = POSITION_DIRECTION.SHORT
return direction