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_init():
wallets = [
Wallet(exchange, 10000 * USD),
Wallet(exchange, 0 * BTC)
]
portfolio = Portfolio(base_instrument=USD, wallets=wallets)
base_wallet = portfolio.get_wallet(exchange.id, USD)
quantity = (1 / 10) * base_wallet.balance
order = Order(side=TradeSide.BUY,
trade_type=TradeType.MARKET,
pair=USD/BTC,
quantity=quantity,
portfolio=portfolio)
assert order
assert order.id
assert order.path_id
assert order.quantity.instrument == USD
assert order.filled_size == 0
def test_sell_on_exchange():
wallets = [
Wallet(exchange, 0 * USD),
Wallet(exchange, 10 * BTC)
]
portfolio = Portfolio(base_instrument=USD, wallets=wallets)
base_wallet = portfolio.get_wallet(exchange.id, USD)
quote_wallet = portfolio.get_wallet(exchange.id, BTC)
quantity = (1 / 10) * quote_wallet.balance
order = Order(side=TradeSide.SELL,
trade_type=TradeType.MARKET,
pair=USD / BTC,
quantity=quantity,
portfolio=portfolio)
current_price = 12390.90
def test_valid_iadd():
# Add to free unlocked balance
wallet = Wallet(exchange, 10000 * USD)
wallet += 500 * USD
assert wallet.balance == 10500 * USD
assert len(wallet.locked) == 0
# Add to balance with locked path_id
wallet = Wallet(exchange, 10000 * USD)
wallet += Quantity(USD, 500, path_id=path_id)
assert wallet.balance == 10000 * USD
assert wallet.locked[path_id] == 500 * USD
# Add to more balance with locked path_id
wallet += Quantity(USD, 500, path_id=path_id)
assert wallet.balance == 10000 * USD
assert wallet.locked[path_id] == 1000 * USD
# Add to balance that has another locked path_id
wallet += Quantity(USD, 500, path_id=other_id)
assert wallet.balance == 10000 * USD
assert wallet.locked[path_id] == 1000 * USD
assert wallet.locked[other_id] == 500 * USD
def test_order_runs_through_broker():
wallets = [
Wallet(exchange, 10000 * USD),
Wallet(exchange, 0 * BTC)
]
portfolio = Portfolio(base_instrument=USD, wallets=wallets)
exchange.reset()
portfolio.reset()
broker.reset()
base_wallet = portfolio.get_wallet(exchange.id, USD)
quantity = (1 / 10) * base_wallet.balance
order = Order(side=TradeSide.BUY,
trade_type=TradeType.MARKET,
pair=USD / BTC,
quantity=quantity,
portfolio=portfolio)
def test_path_order_runs_though_broker():
wallets = [
Wallet(exchange, 10000 * USD),
Wallet(exchange, 0 * BTC)
]
portfolio = Portfolio(base_instrument=USD, wallets=wallets)
exchange.reset()
portfolio.reset()
broker.reset()
base_wallet = portfolio.get_wallet(exchange.id, USD)
quantity = (1 / 10) * base_wallet.balance
order = Order(side=TradeSide.BUY,
trade_type=TradeType.MARKET,
pair=USD / BTC,
quantity=quantity,
portfolio=portfolio)
def test_invalid_isub():
# Add to balance with locked path_id
wallet = Wallet(exchange, 10000 * USD)
wallet += Quantity(USD, 500, path_id=path_id)
wallet += Quantity(USD, 700, path_id=other_id)
with pytest.raises(InsufficientFundsForAllocation):
wallet -= 11000 * USD
with pytest.raises(InsufficientFundsForAllocation):
wallet -= Quantity(USD, 750, path_id)
with pytest.raises(InsufficientFundsForAllocation):
wallet -= Quantity(USD, 750, path_id)
with pytest.raises(IncompatibleInstrumentOperation):
wallet -= 500 * 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 test_add(portfolio, exchange):
wallet_bch = Wallet(exchange, 1000 * BCH)
portfolio.add(wallet_bch)
assert wallet_bch in portfolio.wallets
def test_invalid_iadd():
# Add to free unlocked balance with different instrument
wallet = Wallet(exchange, 10000 * USD)
with pytest.raises(IncompatibleInstrumentOperation):
wallet += 500 * BTC