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_historical_data():
days = datetime.datetime.today() - datetime.timedelta(days=7)
print(days)
data = historical_data(
"USDT", "BNB", "1d", days.timestamp())
assert len(data) == 7
def test_historical_data_ploniex_long():
""" this one is awesome since you can download years worth of data"""
days = datetime.datetime.today() - datetime.timedelta(days=365)
data = historical_data(
"BTC", "ETC", "1d", days.timestamp(), "poloniex")
assert len(data) == 365
def test_historical_data_ploniex():
""" this one is awesome since you can download years worth of data"""
days = datetime.datetime.today() - datetime.timedelta(days=90)
data = historical_data(
"BTC", "ETH", "1d", days.timestamp(), "poloniex")
assert len(data) == 90
latest_time = OHLCV.session.query(func.max(OHLCV.timestamp)).filter(
OHLCV.exchange == ccxt_api.name,
OHLCV.pair == "{}/{}".format(asset.upper(), stake.upper()),
OHLCV.interval == interval
).one()[0]
if force:
print("forcing database refresh and downloading all data!")
latest_time = None
# add additional data on top
if latest_time is None:
# store data for all
for row in historical_data(stake, asset, interval, from_date, ccxt_api):
o = OHLCV(
id="{}-{}-{}/{}:{}".format(ccxt_api.name, interval, asset.upper(), stake.upper(), row[0]),
exchange=ccxt_api.name,
pair="{}/{}".format(asset.upper(), stake.upper()),
interval=interval,
open=row[1],
close=row[4],
high=row[2],
low=row[3],
volume=row[5],
timestamp=row[0]
)
OHLCV.session.merge(o)
else:
# calculate the difference in days and download and merge the data files
exchange=ccxt_api.name,
pair="{}/{}".format(asset.upper(), stake.upper()),
interval=interval,
open=row[1],
close=row[4],
high=row[2],
low=row[3],
volume=row[5],
timestamp=row[0]
)
OHLCV.session.merge(o)
else:
# calculate the difference in days and download and merge the data files
for row in historical_data(stake, asset, interval, latest_time, ccxt_api):
o = OHLCV(
id="{}-{}-{}/{}:{}".format(ccxt_api.name, interval, asset.upper(), stake.upper(), row[0]),
exchange=ccxt_api.name,
pair="{}/{}".format(asset.upper(), stake.upper()),
interval=interval,
open=row[1],
close=row[4],
high=row[2],
low=row[3],
volume=row[5],
timestamp=row[0]
)
OHLCV.session.merge(o)
# return all data to user
result = []