Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _get(self, ep, symbol, start_date, end_date, retry, retry_wait, freq='6H'):
dates = [None]
if start_date:
if not end_date:
end_date = pd.Timestamp.utcnow()
dates = pd.interval_range(API._timestamp(start_date), API._timestamp(end_date), freq=freq).tolist()
if len(dates) == 0:
dates.append(pd.Interval(left=API._timestamp(start_date), right=API._timestamp(end_date)))
elif dates[-1].right < API._timestamp(end_date):
dates.append(pd.Interval(dates[-1].right, API._timestamp(end_date)))
@request_retry(self.ID, retry, retry_wait)
def helper(start, start_date, end_date):
if start_date and end_date:
endpoint = f'/api/v1/{ep}?symbol={symbol}&count={API_MAX}&reverse=false&start={start}&startTime={start_date}&endTime={end_date}'
else:
endpoint = f'/api/v1/{ep}?symbol={symbol}&reverse=true'
header = {}
if self.key_id and self.key_secret:
header = self._generate_signature("GET", endpoint)
header['Accept'] = 'application/json'
return requests.get('{}{}'.format(self.api, endpoint), headers=header)
for interval in dates:
def _get(self, ep, symbol, start_date, end_date, retry, retry_wait, freq='6H'):
dates = [None]
if start_date:
if not end_date:
end_date = pd.Timestamp.utcnow()
dates = pd.interval_range(API._timestamp(start_date), API._timestamp(end_date), freq=freq).tolist()
if len(dates) == 0:
dates.append(pd.Interval(left=API._timestamp(start_date), right=API._timestamp(end_date)))
elif dates[-1].right < API._timestamp(end_date):
dates.append(pd.Interval(dates[-1].right, API._timestamp(end_date)))
@request_retry(self.ID, retry, retry_wait)
def helper(start, start_date, end_date):
if start_date and end_date:
endpoint = f'/api/v1/{ep}?symbol={symbol}&count={API_MAX}&reverse=false&start={start}&startTime={start_date}&endTime={end_date}'
else:
endpoint = f'/api/v1/{ep}?symbol={symbol}&reverse=true'
header = {}
if self.key_id and self.key_secret:
header = self._generate_signature("GET", endpoint)
header['Accept'] = 'application/json'
return requests.get('{}{}'.format(self.api, endpoint), headers=header)
for interval in dates:
start = 0
def _get_trades(self, instrument, start_date, end_date, retry, retry_wait):
start = None
end = None
if start_date:
if not end_date:
end_date = pd.Timestamp.utcnow()
start = API._timestamp(start_date)
end = API._timestamp(end_date) - pd.Timedelta(nanoseconds=1)
start = int(start.timestamp() * 1000)
end = int(end.timestamp() * 1000)
@request_retry(self.ID, retry, retry_wait)
def helper(start, end):
if start and end:
return requests.get(f"{self.api}get_last_trades_by_instrument_and_time?&start_timestamp={start}&end_timestamp={end}&instrument_name={instrument}&include_old=true&count={REQUEST_LIMIT}")
else:
return requests.get(f"{self.api}get_last_trades_by_instrument_and_time/")
while True:
r = helper(start, end)
if r.status_code == 429:
sleep(int(r.headers['Retry-After']))
def _funding_normalization(self, funding: dict, symbol: str) -> dict:
ts = pd.to_datetime(funding['time'], format="%Y-%m-%dT%H:%M:%S%z")
return {
'timestamp': API._timestamp(funding['time']).timestamp(),
'pair': funding['future'],
'feed': self.ID,
'rate': funding['rate']
}
def _get_trades_hist(self, symbol, start_date, end_date, retry, retry_wait):
last = []
start = None
end = None
if end_date and not start_date:
start_date = '2019-01-01'
if start_date:
if not end_date:
end_date = pd.Timestamp.utcnow()
start = API._timestamp(start_date)
end = API._timestamp(end_date)
start = int(start.timestamp())
end = int(end.timestamp())
@request_retry(self.ID, retry, retry_wait)
def helper(start, end):
if start and end:
return requests.get(f"{self.api}/markets/{symbol}/trades?limit=100&start_time={start}&end_time={end}")
else:
return requests.get(f"{self.api}/markets/{symbol}/trades")
while True:
r = helper(start, end)
if r.status_code == 429:
sleep(RATE_LIMIT_SLEEP)
def _trade_normalization(self, trade: dict, symbol: str) -> dict:
return {
'timestamp': API._timestamp(trade['time']).timestamp(),
'pair': symbol,
'id': trade['id'],
'feed': self.ID,
'side': SELL if trade['side'] == 'sell' else BUY,
'amount': trade['size'],
'price': trade['price']
}