Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _fake_trade(order_book_id, quantity, price):
return Trade.__from_create__(0, price, abs(quantity),
SIDE.BUY if quantity > 0 else SIDE.SELL,
POSITION_EFFECT.OPEN, order_book_id)
@ExecutionContext.enforce_phase(EXECUTION_PHASE.ON_BAR,
EXECUTION_PHASE.ON_TICK,
EXECUTION_PHASE.SCHEDULED,
EXECUTION_PHASE.GLOBAL)
@apply_rules(verify_that('id_or_ins').is_valid_future(),
verify_that('amount').is_number().is_greater_or_equal_than(0),
verify_that('side').is_in([SIDE.BUY, SIDE.SELL]),
verify_that('position_effect').is_in([POSITION_EFFECT.OPEN, POSITION_EFFECT.CLOSE]),
verify_that('style').is_instance_of((LimitOrder, MarketOrder, type(None))))
def order(id_or_ins, amount, side, position_effect, style):
if not isinstance(style, OrderStyle):
raise RuntimeError
if amount < 0:
raise RuntimeError
if amount == 0:
user_system_log.warn(_(u"Order Creation Failed: Order amount is 0."))
return None
if isinstance(style, LimitOrder) and style.get_limit_price() <= 0:
raise RQInvalidArgument(_(u"Limit order price should be positive"))
order_book_id = assure_future_order_book_id(id_or_ins)
env = Environment.get_instance()
if env.config.base.run_type != RUN_TYPE.BACKTEST:
from rqalpha.interface import AbstractMod
from rqalpha.model.trade import Trade
from rqalpha.const import SIDE, POSITION_EFFECT, RUN_TYPE
from rqalpha.utils.logger import system_log
from rqalpha.events import EVENT
from .booking_account import BookingAccount
# noinspection PyUnresolvedReferences
from . import api_booking
SIDE_DICT = {
1: SIDE.BUY,
2: SIDE.SELL,
}
POSITION_EFFECT_DICT = {
1: POSITION_EFFECT.OPEN,
2: POSITION_EFFECT.CLOSE,
3: POSITION_EFFECT.CLOSE_TODAY,
}
class BookingMod(AbstractMod):
def __init__(self):
self.env = None
self.mod_config = None
self.booking_account = None
else:
# 平昨仓
quantity *= -1
buy_old_quantity, buy_today_quantity = position.buy_old_quantity, position.buy_today_quantity
if buy_old_quantity > 0:
orders.append(
order(order_book_id, min(quantity, buy_old_quantity), SIDE.SELL, POSITION_EFFECT.CLOSE, style))
quantity -= min(quantity, buy_old_quantity)
if quantity <= 0:
return orders
# 平今仓
if buy_today_quantity > 0:
orders.append(order(
order_book_id,
min(quantity, buy_today_quantity),
SIDE.SELL,
POSITION_EFFECT.CLOSE_TODAY,
style
))
quantity -= buy_today_quantity
if quantity <= 0:
return orders
# 开空仓
orders.append(order(order_book_id, quantity, SIDE.SELL, POSITION_EFFECT.OPEN, style))
return orders
if trade.side not in (SIDE.BUY, SIDE.SELL):
raise RuntimeError("unknown side, trade {}".format(trade))
long_positions = self._positions_dict[POSITION_DIRECTION.LONG]
short_positions = self._positions_dict[POSITION_DIRECTION.SHORT]
if trade.position_effect == POSITION_EFFECT.OPEN:
if trade.side == SIDE.BUY:
position = long_positions.get_or_create(order_book_id)
elif trade.side == SIDE.SELL:
position = short_positions.get_or_create(order_book_id)
elif trade.position_effect in (POSITION_EFFECT.CLOSE, POSITION_EFFECT.CLOSE_TODAY):
if trade.side == SIDE.BUY:
position = short_positions.get_or_create(order_book_id)
elif trade.side == SIDE.SELL:
position = long_positions.get_or_create(order_book_id)
else:
# NOTE: 股票如果没有position_effect就特殊处理
position = long_positions.get_or_create(order_book_id)
position.apply_trade(trade)
self._backward_trade_set.add(trade.exec_id)
def order_shares(id_or_ins, amount, price=None, style=None):
order_book_id = assure_order_book_id(id_or_ins)
env = Environment.get_instance()
if amount > 0:
side = SIDE.BUY
position_effect = POSITION_EFFECT.OPEN
else:
amount = abs(amount)
side = SIDE.SELL
position_effect = POSITION_EFFECT.CLOSE
round_lot = int(env.get_instrument(order_book_id).round_lot)
try:
amount = int(Decimal(amount) / Decimal(round_lot)) * round_lot
except ValueError:
amount = 0
order = Order.__from_create__(order_book_id, amount, side, style, position_effect)
if amount == 0:
# 如果计算出来的下单量为0, 则不生成Order, 直接返回None
# 因为很多策略会直接在handle_bar里面执行order_target_percent之类的函数,经常会出现下一个量为0的订单,如果这些订单都生成是没有意义的。
user_system_log.warn(_(u"Order Creation Failed: 0 order quantity"))
return
trades = []
position_list = self._get_old_position_list()
for position_dict in position_list:
if position_dict["buy_quantity"] != 0:
trade = self._create_trade(
position_dict["order_book_id"],
position_dict["buy_quantity"],
SIDE.BUY,
POSITION_EFFECT.OPEN,
)
trades.append(trade)
if position_dict["sell_quantity"] != 0:
trade = self._create_trade(
position_dict["order_book_id"],
position_dict["sell_quantity"],
SIDE.SELL,
POSITION_EFFECT.OPEN,
)
trades.append(trade)
system_log.info("yesterday positions trades")
for trade in trades:
system_log.info("trade: {:9}, qtx {}, side {}", trade.order_book_id, trade.last_quantity, trade.side)
self.booking_account.fast_forward([], trades=trades)
self.booking_account._settlement(None, check_delist=False)
# 计算今仓
trades = []
trade_list = self._get_trade_list()
for trade_dict in trade_list:
verify_that('side').is_in([SIDE.BUY, SIDE.SELL]),
verify_that('position_effect').is_in([POSITION_EFFECT.OPEN, POSITION_EFFECT.CLOSE, POSITION_EFFECT.CLOSE_TODAY]),
verify_that('style').is_instance_of((LimitOrder, MarketOrder, type(None))))
def send_order(order_book_id, amount, side, position_effect, style):
env = Environment.get_instance()
order = Order.__from_create__(order_book_id, amount, side, style, position_effect)
env.broker.submit_order(order)
return order
def sell_open_order_quantity(self):
"""
[int] 卖方向挂单量
"""
return sum(order.unfilled_quantity for order in self.open_orders if
order.side == SIDE.SELL and order.position_effect == POSITION_EFFECT.OPEN)
def closable_buy_quantity(self):
"""
[float] 可平多方向持仓
"""
sell_close_order_quantity = sum(o.unfilled_quantity for o in self.open_orders if o.side == SIDE.SELL and
o.position_effect in (POSITION_EFFECT.CLOSE, POSITION_EFFECT.CLOSE_TODAY))
return self.buy_quantity - sell_close_order_quantity