Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_account_type_enum(order_book_id):
from rqalpha.environment import Environment
instrument = Environment.get_instance().get_instrument(order_book_id)
enum_type = instrument.enum_type
if enum_type in INST_TYPE_IN_STOCK_ACCOUNT:
return DEFAULT_ACCOUNT_TYPE.STOCK
elif enum_type == INSTRUMENT_TYPE.FUTURE:
return DEFAULT_ACCOUNT_TYPE.FUTURE
elif enum_type == INSTRUMENT_TYPE.BOND:
return DEFAULT_ACCOUNT_TYPE.BOND
else:
raise NotImplementedError
def _frozen_cash_of_order(order):
order_cost = margin_of(
order.order_book_id, order.quantity, order.frozen_price
) if order.position_effect == POSITION_EFFECT.OPEN else 0
return order_cost + Environment.get_instance().get_order_transaction_cost(
DEFAULT_ACCOUNT_TYPE.FUTURE, order
)
def get_account_type_enum(order_book_id):
from rqalpha.environment import Environment
instrument = Environment.get_instance().get_instrument(order_book_id)
enum_type = instrument.enum_type
if enum_type in INST_TYPE_IN_STOCK_ACCOUNT:
return DEFAULT_ACCOUNT_TYPE.STOCK
elif enum_type == INSTRUMENT_TYPE.FUTURE:
return DEFAULT_ACCOUNT_TYPE.FUTURE
elif enum_type == INSTRUMENT_TYPE.BOND:
return DEFAULT_ACCOUNT_TYPE.BOND
else:
raise NotImplementedError
def _get_trading_points(self, trading_date, frequency):
indexer = self._indexer
trading_points = set()
for account_type in self._config.base.accounts:
if account_type == DEFAULT_ACCOUNT_TYPE.STOCK.name:
trading_points.update(indexer.get_a_stock_trading_points(trading_date, frequency))
elif account_type == DEFAULT_ACCOUNT_TYPE.FUTURE.name:
trading_points.update(indexer.get_future_trading_points(self._env, trading_date, frequency))
return sorted(list(trading_points))
def _get_trading_minutes(self, trading_date):
trading_minutes = set()
for account_type in self._config.base.accounts:
if account_type == DEFAULT_ACCOUNT_TYPE.STOCK.name:
trading_minutes = trading_minutes.union(self._get_stock_trading_minutes(trading_date))
elif account_type == DEFAULT_ACCOUNT_TYPE.FUTURE.name:
trading_minutes = trading_minutes.union(self._get_future_trading_minutes(trading_date))
return sorted(list(trading_minutes))
# [END] minute event helper
def __init__(self, multiplier, cn_stock_min_commission, hk_stock_min_commission):
self.deciders = dict()
self.deciders[DEFAULT_ACCOUNT_TYPE.STOCK.name] = StockCommission(
multiplier, cn_stock_min_commission, hk_stock_min_commission
)
self.deciders[DEFAULT_ACCOUNT_TYPE.FUTURE.name] = FutureCommission(multiplier)
def can_submit_order(self, order, account=None):
if account is not None and account.type == DEFAULT_ACCOUNT_TYPE.FUTURE.name:
return self._future_validator(account, order)
return True
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()