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_call_with_buy_order(mock_order_class, mock_exchange_class):
exchange = mock_exchange_class.return_value
exchange.quote_price = mock.Mock(return_value=7000.00)
order = mock_order_class.return_value
order.pair = USD/BTC
order.side = TradeSide.BUY
criteria = Limit(limit_price=6800.00)
assert not criteria(order, exchange)
criteria = Limit(limit_price=7000.00)
assert criteria(order, exchange)
criteria = Limit(limit_price=7200.00)
assert criteria(order, exchange)
def wallet_btc(exchange):
return Wallet(exchange, 1 * BTC)
def test_default_checks_on_call(mock_order_class, mock_exchange_class):
order = mock_order_class.return_value
order.pair = USD/BTC
exchange = mock_exchange_class.return_value
exchange.is_pair_tradable = lambda pair: str(pair) in ["USD/BTC", "USD/ETH", "USD/XRP"]
criteria = ConcreteCriteria()
# Pair being traded is supported by exchange
exchange.is_pair_tradable = lambda pair: str(pair) in ["USD/BTC", "USD/ETH", "USD/XRP"]
assert criteria(order, exchange)
# Pair being traded is not supported by exchange
exchange.is_pair_tradable = lambda pair: str(pair) in ["USD/ETH", "USD/XRP"]
assert not criteria(order, exchange)
# 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_call_with_direction_down(mock_order_class, mock_exchange_class):
exchange = mock_exchange_class.return_value
order = mock_order_class.return_value
order.pair = USD/BTC
order.price = 7000.00
criteria = Stop(direction=StopDirection.DOWN, percent=0.03)
# Greater than 3.00% below order price
exchange.quote_price.return_value = 0.95 * order.price
assert criteria(order, exchange)
# Equal to 3.00% below order price
exchange.quote_price.return_value = 0.969 * order.price
assert criteria(order, exchange)
# Less than 3.00% below order price
exchange.quote_price.return_value = 0.98 * order.price
assert not criteria(order, exchange)
def test_balance(portfolio_locked):
assert portfolio_locked.balance(USD) == 9850
assert portfolio_locked.balance(BTC) == 0.25
assert portfolio_locked.balance(ETH) == 3
assert portfolio_locked.balance(XRP) == 4738
def __init__(self, data_frame: pd.DataFrame = None, **kwargs):
self._commission = self.default('commission', 0.003, kwargs)
self._base_instrument = self.default('base_instrument', USD, kwargs)
self._quote_instrument = self.default('quote_instrument', BTC, kwargs)
self._initial_balance = self.default('initial_balance', 10000, kwargs)
self._min_trade_size = self.default('min_trade_size', 1e-6, kwargs)
self._max_trade_size = self.default('max_trade_size', 1e6, kwargs)
self._min_trade_price = self.default('min_trade_price', 1e-8, kwargs)
self._max_trade_price = self.default('max_trade_price', 1e8, kwargs)
self._randomize_time_slices = self.default('randomize_time_slices', False, kwargs)
self._min_time_slice = self.default('min_time_slice', 128, kwargs)
self._price_column = self.default('price_column', 'close', kwargs)
self.data_frame = self.default('data_frame', data_frame)
slippage_model = self.default('slippage_model', 'uniform', kwargs)
self._slippage_model = slippage.get(slippage_model) if isinstance(
slippage_model, str) else slippage_model()
self.reset()
# limitations under the License
import time
import ccxt
import numpy as np
import pandas as pd
from typing import Dict, List, Union
from gym.spaces import Space, Box
from ccxt import BadRequest
from tensortrade.trades import Trade, TradeType, TradeSide
from tensortrade.instruments import TradingPair, BTC, ETH
from tensortrade.exchanges import Exchange
BTC_ETH_PAIR = TradingPair(BTC, ETH)
class CCXTExchange(Exchange):
"""An exchange for trading on CCXT-supported cryptocurrency exchanges."""
def __init__(self, exchange: Union[ccxt.Exchange, str] = 'coinbase', **kwargs):
super(CCXTExchange, self).__init__(**kwargs)
exchange = self.default('exchange', exchange)
self._exchange = getattr(ccxt, exchange)() if \
isinstance(exchange, str) else exchange
self._credentials = self.default('credentials', None, kwargs)
self._timeframe = self.default('timeframe', '10m', kwargs)
self._observation_type = self.default('observation_type', 'ohlcv', kwargs)