196 lines
7.9 KiB
Python
196 lines
7.9 KiB
Python
import datetime
|
|
|
|
from candles import Candles
|
|
from Configuration import Configuration
|
|
from exchangeinterface import ExchangeInterface
|
|
|
|
|
|
def test_sqlite():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
# Configuration and settings for the user app and charts
|
|
conf = Configuration()
|
|
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
|
assert candle_obj is not None
|
|
# candle_obj.set_candle_history(symbol='BTCUSDT',
|
|
# timeframe='15m',
|
|
# exchange_name='binance_futures')
|
|
|
|
|
|
def test_ts_of_n_minutes_ago():
|
|
ca = 12
|
|
lc = 5
|
|
print(f'\ncandles ago is {ca}, length of candle is {lc}:')
|
|
result = ts_of_n_minutes_ago(ca, lc)
|
|
print(f'Result: TYPE({type(result)}): {result}')
|
|
print(f'The time right now is: {datetime.datetime.now()}')
|
|
difference = datetime.datetime.now() - result
|
|
print(f'The difference is: {difference}')
|
|
# Result is supposed to be one candle length extra.
|
|
assert difference == datetime.timedelta(hours=1, minutes=5, seconds=00)
|
|
|
|
|
|
def test_timeframe_to_minutes():
|
|
result = timeframe_to_minutes('1w')
|
|
print(f'\nnumber of minutes: {result}')
|
|
assert result == 10080
|
|
|
|
|
|
def test_fetch_exchange_id():
|
|
exchange_id = fetch_exchange_id('binance_coin')
|
|
print(f'\nexchange_id: {exchange_id}')
|
|
assert exchange_id == 3
|
|
|
|
|
|
def test_fetch_market_id_from_db():
|
|
market_id = fetch_market_id_from_db('ETHUSDT', 'binance_spot')
|
|
print(f'\nmarket_id: {market_id}')
|
|
assert market_id == 1
|
|
|
|
|
|
def test_fetch_candles_from_exchange():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
candles_obj = Candles(exchanges)
|
|
|
|
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSDT', interval='15m', exchange_name='binance_spot',
|
|
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
|
|
|
print(f'\n{result.head().to_string()}')
|
|
|
|
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSDT', interval='15m', exchange_name='binance_futures',
|
|
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
|
print(f'\n{result.head().to_string()}')
|
|
|
|
result = candles_obj._fetch_candles_from_exchange(symbol='BTCUSD_PERP', interval='15m',
|
|
exchange_name='binance_coin',
|
|
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
|
print(f'\n{result.head().to_string()}')
|
|
|
|
result = candles_obj._fetch_candles_from_exchange(symbol='BTC/USDT', interval='15m', exchange_name='alpaca',
|
|
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
|
print(f'\n{result.head().to_string()}')
|
|
assert not result.empty
|
|
|
|
|
|
def test_insert_candles_into_db():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
candles_obj = Candles(exchanges)
|
|
symbol = 'ETHUSDT'
|
|
interval = '15m'
|
|
exchange_name = 'binance_spot'
|
|
result = candles_obj._fetch_candles_from_exchange(symbol=symbol, interval=interval, exchange_name=exchange_name,
|
|
start_datetime={'year': 2023, 'month': 3, 'day': 15})
|
|
|
|
insert_candles_into_db(result, symbol=symbol, interval=interval, exchange_name=exchange_name)
|
|
assert True
|
|
|
|
|
|
def test_get_db_records_since():
|
|
timestamp = datetime.datetime.timestamp(datetime.datetime(year=2023, month=3, day=14, hour=1, minute=0)) * 1000
|
|
result = get_db_records_since(table_name='ETHUSDT_15m_alpaca', timestamp=timestamp)
|
|
print(result.head())
|
|
assert result is not None
|
|
|
|
|
|
def test_get_from_cache_since():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
candles_obj = Candles(exchanges)
|
|
symbol = 'ETHUSDT'
|
|
interval = '15m'
|
|
exchange_name = 'binance_spot'
|
|
timestamp = datetime.datetime.timestamp(datetime.datetime(year=2023, month=3, day=16, hour=1, minute=0)) * 1000
|
|
key = f'{symbol}_{interval}_{exchange_name}'
|
|
|
|
result = candles_obj.get_from_cache_since(key=key, start_datetime=timestamp)
|
|
print(result)
|
|
assert result is not None
|
|
|
|
|
|
def test_get_records_since():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
candles_obj = Candles(exchanges)
|
|
symbol = 'BTCUSDT'
|
|
interval = '2h'
|
|
exchange_name = 'binance_spot'
|
|
|
|
start_time = datetime.datetime(year=2023, month=3, day=14, hour=1, minute=0)
|
|
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
|
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
|
exchange_name=exchange_name, start_time=start_time)
|
|
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.open_time.min() / 1000)}')
|
|
|
|
start_time = datetime.datetime(year=2023, month=3, day=16, hour=1, minute=0)
|
|
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
|
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
|
exchange_name=exchange_name, start_time=start_time)
|
|
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.open_time.min() / 1000)}')
|
|
|
|
start_time = datetime.datetime(year=2023, month=3, day=13, hour=1, minute=0)
|
|
print(f'\ntest_candles_get_records() starting @: {start_time}')
|
|
result = candles_obj.get_records_since(symbol=symbol, timeframe=interval,
|
|
exchange_name=exchange_name, start_time=start_time)
|
|
print(f'test_candles_received records from: {datetime.datetime.fromtimestamp(result.open_time.min() / 1000)}')
|
|
|
|
assert result is not None
|
|
|
|
|
|
def test_get_last_n_candles():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
candles_obj = Candles(exchanges)
|
|
symbol = 'ETHUSDT'
|
|
interval = '1m'
|
|
exchange_name = 'binance_spot'
|
|
print(f'\n Here we go!')
|
|
result = candles_obj.get_last_n_candles(5, symbol, interval, exchange_name)
|
|
print(result)
|
|
result = candles_obj.get_last_n_candles(10, symbol, interval, exchange_name)
|
|
print(result)
|
|
result = candles_obj.get_last_n_candles(40, symbol, interval, exchange_name)
|
|
print(result)
|
|
result = candles_obj.get_last_n_candles(20, symbol, interval, exchange_name)
|
|
print(result)
|
|
result = candles_obj.get_last_n_candles(15, symbol, interval, exchange_name)
|
|
print(result)
|
|
assert result is not None
|
|
|
|
|
|
def test_get_latest_values():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
# Configuration and settings for the user app and charts
|
|
conf = Configuration()
|
|
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
|
|
|
symbol = 'ETHUSDT'
|
|
interval = '1m'
|
|
exchange_name = 'binance_spot'
|
|
result = candle_obj.get_latest_values('open', symbol=symbol, timeframe=interval,
|
|
exchange=exchange_name, num_record=10)
|
|
print(result)
|
|
assert result is not None
|
|
|
|
|
|
def test_get_colour_coded_volume():
|
|
# Object that interacts and maintains exchange_interface and account data
|
|
exchanges = ExchangeInterface()
|
|
# Configuration and settings for the user app and charts
|
|
conf = Configuration()
|
|
candle_obj = Candles(exchanges=exchanges, config_obj=conf)
|
|
|
|
symbol = 'ETHUSDT'
|
|
interval = '1h'
|
|
exchange_name = 'binance_spot'
|
|
result = candle_obj.get_latest_values('volume', symbol=symbol, timeframe=interval,
|
|
exchange=exchange_name, num_record=10)
|
|
print(result)
|
|
assert result is not None
|
|
|
|
|
|
def test_get_last_n_candles():
|
|
assert False
|