Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_valid_isub():
# Add to remove from unlocked balance
wallet = Wallet(exchange, 10000 * USD)
wallet -= 500 * USD
assert wallet.balance == 9500 * USD
assert len(wallet.locked) == 0
# Add to balance with locked path_id
wallet = Wallet(exchange, 10000 * USD)
wallet += Quantity(USD, 750, path_id=path_id)
wallet += Quantity(USD, 1000, path_id=other_id)
wallet -= Quantity(USD, 500, path_id=path_id)
assert wallet.balance == 10000 * USD
assert wallet.locked[path_id] == 250 * USD
wallet -= Quantity(USD, 500, path_id=other_id)
assert wallet.balance == 10000 * USD
assert wallet.locked[other_id] == 500 * USD
def test_valid_rmul():
BTC = Instrument("BTC", 8, "Bitcoin")
# int
q = 8*BTC
assert isinstance(q, Quantity)
assert q.size == 8
assert q.instrument == BTC
# float
q = 8.0*BTC
assert isinstance(q, Quantity)
assert q.size == 8.0
assert q.instrument == BTC
def test_locked_balance():
wallet = Wallet(exchange, 10000 * USD)
wallet += Quantity(USD, 500, path_id=path_id)
wallet += Quantity(USD, 700, path_id=other_id)
assert wallet.locked_balance == 1200 * USD
def _execute_sell_order(self, order: 'Order', base_wallet: 'Wallet', quote_wallet: 'Wallet', current_price: float) -> Trade:
price = self._contain_price(current_price)
if order.type == TradeType.LIMIT and order.price > current_price:
return None
commission = Quantity(order.pair.base, order.size * self._commission, order.path_id)
size = self._contain_size(order.size - commission.size)
quantity = Quantity(order.pair.base, size, order.path_id)
trade = Trade(order_id=order.id,
exchange_id=self.id,
step=self.clock.step,
pair=order.pair,
side=TradeSide.SELL,
trade_type=order.type,
quantity=quantity,
price=price,
commission=commission)
# self._slippage_model.adjust_trade(trade)
quote_size = trade.size / trade.price * (trade.price / order.price)
exchange_id=self.id,
step=self.clock.step,
pair=order.pair,
side=TradeSide.BUY,
trade_type=order.type,
quantity=quantity,
price=price,
commission=commission)
# self._slippage_model.adjust_trade(trade)
quote_size = trade.size / trade.price * (order.price / trade.price)
base_wallet -= quantity
base_wallet -= commission
quote_wallet += Quantity(order.pair.quote, quote_size, order.path_id)
return trade
def balance(self, instrument: Instrument) -> Quantity:
"""The total balance of the portfolio in a specific instrument available for use."""
balance = Quantity(instrument, 0)
for (_, symbol), wallet in self._wallets.items():
if symbol == instrument.symbol:
balance += wallet.balance
return balance
def _execute_buy_order(self, order: 'Order', base_wallet: 'Wallet', quote_wallet: 'Wallet', current_price: float) -> Trade:
price = self._contain_price(current_price)
if order.type == TradeType.LIMIT and order.price < current_price:
return None
commission = Quantity(order.pair.base, order.size * self._commission, order.path_id)
size = self._contain_size(order.size - commission.size)
if order.type == TradeType.MARKET:
size = self._contain_size(order.price / price * order.size - commission.size)
quantity = Quantity(order.pair.base, size, order.path_id)
trade = Trade(order_id=order.id,
exchange_id=self.id,
step=self.clock.step,
pair=order.pair,
side=TradeSide.BUY,
trade_type=order.type,
quantity=quantity,
price=price,
commission=commission)
def balance(self, instrument: Instrument) -> Quantity:
"""The total balance of the portfolio in a specific instrument available for use."""
balance = Quantity(instrument, 0)
for (_, symbol), wallet in self._wallets.items():
if symbol == instrument.symbol:
balance += wallet.balance
return balance
def __init__(self, exchange: 'Exchange', instrument: 'Instrument', balance: float = 0):
self._id = uuid.uuid4()
self._exchange = exchange
self._instrument = instrument
self._balance = Quantity(balance, instrument, wallet_id=str(self.id))
self._locked = {}