wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-02 12:30:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

WIP: bundling 1min data

Browse Source
This commit is contained in:
Victor Grau Serrat
2017-09-20 09:15:39 -06:00
parent e761433d06
commit 91e71c5e38
1 changed files with 83 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/curate/poloniex.py
+83 -30
View File
@@ -1,15 +1,15 @@
import json, time, csv
from datetime import datetime
import pandas as pd
import os
import time
import requests
import logbook
import os, time, shutil, requests, logbook
DT_START = time.mktime(datetime(2010, 1, 1, 0, 0).timetuple())
DT_START = int(time.mktime(datetime(2010, 1, 1, 0, 0).timetuple()))
DT_END = int(time.time())
CSV_OUT_FOLDER = '/var/tmp/catalyst/data/poloniex/'
CSV_OUT_FOLDER = '/Volumes/enigma/data/poloniex/'
CONN_RETRIES = 2
COINS = ['USDT_BTC','USDT_DASH','USDT_ETC','USDT_ETH','USDT_LTC','USDT_NXT','USDT_REP','USDT_STR','USDT_XMR','USDT_XRP','USDT_ZEC']
COINS = ['USDT_BTC',]
logbook.StderrHandler().push_application()
@@ -116,36 +116,44 @@ class PoloniexCurator(object):
df = self.get_data( currencyPair, start, first, df )
return df
def retrieve_trade_history(self, currencyPair, start, end=9999999999):
def _retrieve_tradeID_date(self, row):
tId = int(row.split(',')[0])
d = pd.to_datetime( row.split(',')[1], infer_datetime_format=True).value // 10 ** 9
return tId, d
def retrieve_trade_history(self, currencyPair, start=DT_START, end=DT_END, temp=None):
csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
try:
with open(csv_fn, 'ab+') as f:
f.seek(0, os.SEEK_END)
if(f.tell() > 2): # First check file is not zero size
if(f.tell() > 2): # First check file is not zero size
f.seek(0) # Go to the beginning to read first line
last_tradeID, end_file = self._retrieve_tradeID_date(f.readline())
f.seek(-2, os.SEEK_END) # Jump to the second last byte.
while f.read(1) != b"\n": # Until EOL is found...
f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more.
lastrow = f.readline() # read last line
last_tradeID = int(lastrow.split(',')[0])
end = pd.to_datetime( lastrow.split(',')[1], infer_datetime_format=True).value // 10 ** 9
first_tradeID, start_file = self._retrieve_tradeID_date(f.readline())
if( first_tradeID == 1 and end_file + 3600 > DT_END ):
return
except Exception as e:
log.error('Error opening file: %s' % csv_fn)
log.exception(e)
'''
Poloniex API limits querying TradeHistory to intervals smaller than 1 year,
so we make sure that start date is never more than 1 year apart from end date
Poloniex API limits querying TradeHistory to intervals smaller than 1 month,
so we make sure that start date is never more than 1 month apart from end date
'''
if( end == 9999999999 and time.time() - start > 365*86400 ):
newstart = time.time() - 360*86400
elif( end != 9999999999 and end - start > 365*86400 ):
newstart = end - 360*86400
if( end - start > 2419200 ): # 60 s/min * 60 min/hr * 24 hr/day * 28 days
newstart = end - 2419200
else:
newstart = start
log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end))
log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end) +'\t '
+ time.ctime(newstart) + ' - '+ time.ctime(end))
url = self._api_path + 'command=returnTradeHistory&currencyPair=' + currencyPair + '&start=' + str(newstart) + '&end=' + str(end)
@@ -155,31 +163,63 @@ class PoloniexCurator(object):
log.error('Failed to retrieve trade history data for %s' % currencyPair)
log.exception(e)
return None
else:
if isinstance(response.json(), dict) and response.json()['error']:
log.error('Failed to to retrieve trade history data for %s: %s' % (currencyPair,response.json()['error']))
exit(1)
if('last_tradeID' in locals() and response.json()[-1]['tradeID'] == last_tradeID): # Got to the end of TradingHistory for this coin
if('first_tradeID' in locals() and response.json()[-1]['tradeID'] == first_tradeID): # Got to the end of TradingHistory for this coin
return
try:
with open(csv_fn, 'ab') as csvfile:
csvwriter = csv.writer(csvfile)
if( 'end_file' in locals() and end_file + 3600 < end):
if (temp is None):
temp = os.tmpfile()
tempcsv = csv.writer(temp)
for item in response.json():
if( 'last_tradeID' in locals() and item['tradeID'] >= last_tradeID ):
if( item['tradeID'] <= last_tradeID ):
continue
csvwriter.writerow([
tempcsv.writerow([
item['tradeID'],
item['date'],
item['type'],
item['rate'],
item['amount'],
item['total'],
item['globalTradeID']
item['globalTradeID']
])
if( response.json()[-1]['tradeID'] > last_tradeID ):
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
self.retrieve_trade_history(currencyPair, start, end, temp=temp)
else:
with open(csv_fn,'rb+') as f:
shutil.copyfileobj(f,temp)
f.seek(0)
temp.seek(0)
shutil.copyfileobj(temp,f)
temp.close()
end = start_file
else:
with open(csv_fn, 'ab') as csvfile:
csvwriter = csv.writer(csvfile)
for item in response.json():
if( 'first_tradeID' in locals() and item['tradeID'] >= first_tradeID ):
continue
csvwriter.writerow([
item['tradeID'],
item['date'],
item['type'],
item['rate'],
item['amount'],
item['total'],
item['globalTradeID']
])
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
except Exception as e:
log.error('Error opening %s' % csv_fn)
log.exception(e)
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
self.retrieve_trade_history(currencyPair, start, end) # If we get here, we aren't done. Repeat
def write_ohlcv_file(self, currencyPair):
@@ -189,7 +229,8 @@ class PoloniexCurator(object):
if( os.path.isfile(csv_1min) ):
log.debug(currencyPair+': 1min data already present. Delete the file if you want to rebuild it.')
else:
df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'] )
df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'],
dtype = {'tradeID': int, 'date': str, 'type': str, 'rate': float, 'amount': float, 'total': float, 'globalTradeID': int } )
df.drop(['tradeID','type','amount','globalTradeID'], axis=1, inplace=True)
df['date'] = pd.to_datetime(df['date'], infer_datetime_format=True)
ohlcv = self.generate_ohlcv(df)
@@ -288,11 +329,23 @@ class PoloniexCurator(object):
return df[datetime.fromtimestamp(start):datetime.fromtimestamp(end-1)]
def onemin_to_dataframe(self, currencyPair, start, end):
csv_fn = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv'
df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume'])
df['date'] = pd.to_datetime(df['date'],unit='s')
df.set_index('date', inplace=True)
return df[start : end]
if __name__ == '__main__':
pc = PoloniexCurator()
#pc.get_currency_pairs()
pc.get_currency_pairs()
#pc.append_data()
for coin in COINS:
# pc.retrieve_trade_history(coin,DT_START)
pc.write_ohlcv_file(coin)
#for coin in COINS:
for currencyPair in pc.currency_pairs:
#csv_1min = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv'
#if( os.path.isfile(csv_1min) ):
# log.debug(currencyPair+': 1min data already present. Delete the file if you want to rebuild it.')
#else:
pc.retrieve_trade_history(currencyPair)
pc.write_ohlcv_file(currencyPair)