wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-13 01:15:36 +08:00
Code Issues Packages Projects Releases Wiki Activity

MAINT: handful of coins with first tradeID > 1 and PEP8

Browse Source
This commit is contained in:
Victor Grau Serrat
2017-11-08 11:31:56 -07:00
parent 9229809b05
commit 9a80a488cd
1 changed files with 145 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
catalyst/curate/poloniex.py
+145 -76
View File
@@ -6,7 +6,7 @@ from catalyst.exchange.exchange_utils import get_exchange_symbols_filename
DT_START = int(time.mktime(datetime(2010, 1, 1, 0, 0).timetuple()))
DT_END = int(time.time())
DT_END = pd.to_datetime('today').value // 10 ** 9
CSV_OUT_FOLDER = '/var/tmp/catalyst/data/poloniex/'
CSV_OUT_FOLDER = '/Volumes/enigma/data/poloniex/'
CONN_RETRIES = 2
@@ -30,10 +30,11 @@ class PoloniexCurator(object):
log.error('Failed to create data folder: %s' % CSV_OUT_FOLDER)
log.exception(e)
'''
Retrieves and returns all currency pairs from the exchange
'''
def get_currency_pairs(self):
'''
Retrieves and returns all currency pairs from the exchange
'''
url = self._api_path + 'command=returnTicker'
try:
@@ -49,30 +50,40 @@ class PoloniexCurator(object):
self.currency_pairs.append(ticker)
self.currency_pairs.sort()
log.debug('Currency pairs retrieved successfully: %d' % (len(self.currency_pairs)))
log.debug('Currency pairs retrieved successfully: %d'.format(
len(self.currency_pairs)
))
'''
Helper function that reads tradeID and date fields from CSV readline
'''
def _retrieve_tradeID_date(self, row):
'''
Helper function that reads tradeID and date fields from CSV readline
'''
tId = int(row.split(',')[0])
d = pd.to_datetime( row.split(',')[1], infer_datetime_format=True).value // 10 ** 9
d = pd.to_datetime(row.split(',')[1],
infer_datetime_format=True).value // 10 ** 9
return tId, d
'''
Retrieves TradeHistory from exchange for a given currencyPair between start and end dates.
If no start date is provided, uses a system-wide one (beginning of time for cryptotrading)
If no end date is provided, 'now' is used
def retrieve_trade_history(self, currencyPair, start=DT_START,
end=DT_END, temp=None):
'''
Retrieves TradeHistory from exchange for a given currencyPair
between start and end dates. If no start date is provided, uses
a system-wide one (beginning of time for cryptotrading).
If no end date is provided, 'now' is used.
Stores results in CSV file on disk.
This function is called recursively to work around the limitations imposed by the provider API.
'''
def retrieve_trade_history(self, currencyPair, start=DT_START, end=DT_END, temp=None):
This function is called recursively to work around the
limitations imposed by the provider API.
'''
csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
'''
Check what data we already have on disk, reading first and last lines from file.
Data is stored on file from NEWEST to OLDEST.
Check what data we already have on disk, reading first and last
lines from file. Data is stored on file from NEWEST to OLDEST.
'''
try:
with open(csv_fn, 'ab+') as f:
@@ -85,7 +96,12 @@ class PoloniexCurator(object):
f.seek(-2, os.SEEK_CUR) # ...jump back the read byte plus one more.
first_tradeID, start_file = self._retrieve_tradeID_date(f.readline())
if( first_tradeID == 1 and end_file + 3600 > DT_END ):
if( end_file + 3600 * 6 > DT_END and ( first_tradeID == 1
or (currencyPair == 'BTC_HUC' and first_tradeID == 2)
or (currencyPair == 'BTC_RIC' and first_tradeID == 2)
or (currencyPair == 'BTC_XCP' and first_tradeID == 2)
or (currencyPair == 'BTC_NAV' and first_tradeID == 4569)
or (currencyPair == 'BTC_POT' and first_tradeID == 23511) ) ):
return
except Exception as e:
@@ -93,45 +109,63 @@ class PoloniexCurator(object):
log.exception(e)
'''
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
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 - start > 2419200 ): # 60 s/min * 60 min/hr * 24 hr/day * 28 days
if( end - start > 2419200 ): # 60s/min * 60min/hr * 24hr/day * 28days
newstart = end - 2419200
else:
newstart = start
log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end) +'\t '
+ time.ctime(newstart) + ' - '+ time.ctime(end))
log.debug('{}: Retrieving from {} to {}\t {} - {}'.format(
currencyPair, str(newstart), str(end),
time.ctime(newstart), time.ctime(end)))
url = self._api_path + 'command=returnTradeHistory&currencyPair=' + currencyPair + '&start=' + str(newstart) + '&end=' + str(end)
url = '{path}command=returnTradeHistory&currencyPair={pair}' \
'&start={start}&end={end}'.format(
path = self._api_path,
pair = currencyPair,
start = str(newstart),
end = str(end)
)
print url
try:
response = requests.get(url)
except Exception as e:
log.error('Failed to retrieve trade history data for %s' % currencyPair)
log.error('Failed to retrieve trade history data for {}'.format(
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']))
log.error('Failed to to retrieve trade history data '
'for {}: {}'.format(
currencyPair,
response.json()['error']
))
exit(1)
'''
If we get to transactionId == 1, and we already have that on disk,
we got to the end of TradeHistory for this coin.
If we get to transactionId == 1, and we already have that on
disk, we got to the end of TradeHistory for this coin.
'''
if('first_tradeID' in locals() and response.json()[-1]['tradeID'] == first_tradeID):
if('first_tradeID' in locals()
and response.json()[-1]['tradeID'] == first_tradeID):
return
'''
There are primarily two scenarios:
a) There is newer data available that we need to add at the beginning
of the file. We'll retrieve all what we need until we get to what
we already have, writing it to a temporary file; and we will write
that at the beginning of our existing file.
b) We are going back in time, appending at the end of our existing
TradeHistory until the first transaction for this currencyPair
a) There is newer data available that we need to add at
the beginning of the file. We'll retrieve all what we
need until we get to what we already have, writing it
to a temporary file; and we will write that at the
beginning of our existing file.
b) We are going back in time, appending at the end of
our existing TradeHistory until the first transaction
for this currencyPair
'''
try:
if( 'end_file' in locals() and end_file + 3600 < end):
@@ -151,8 +185,10 @@ class PoloniexCurator(object):
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)
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)
@@ -165,7 +201,8 @@ class PoloniexCurator(object):
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 ):
if( 'first_tradeID' in locals()
and item['tradeID'] >= first_tradeID ):
continue
csvwriter.writerow([
item['tradeID'],
@@ -176,48 +213,67 @@ class PoloniexCurator(object):
item['total'],
item['globalTradeID']
])
end = pd.to_datetime( response.json()[-1]['date'], infer_datetime_format=True).value // 10 ** 9
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)
'''
If we got here, we aren't done yet. Call recursively with 'end' times
that go sequentially back in time.
If we got here, we aren't done yet. Call recursively with
'end' times that go sequentially back in time.
'''
self.retrieve_trade_history(currencyPair, start, end)
'''
def generate_ohlcv(self, df):
'''
Generates OHLCV dataframe from a dataframe containing all TradeHistory
by resampling with 1-minute period
'''
def generate_ohlcv(self, df):
df.set_index('date', inplace=True) # Index by date
vol = df['total'].to_frame('volume') # Will deal with vol separately, as ohlc() messes it up
df.drop('total', axis=1, inplace=True) # Drop volume data from dataframe
ohlc = df.resample('T').ohlc() # Resample OHLC in 1min bins
ohlc.columns = ohlc.columns.map(lambda t: t[1]) # Raname columns by dropping 'rate'
closes = ohlc['close'].fillna(method='pad') # Pad forward missing 'close'
ohlc = ohlc.apply(lambda x: x.fillna(closes)) # Fill N/A with last close
vol = vol.resample('T').sum().fillna(0) # Add volumes by bin
ohlcv = pd.concat([ohlc,vol], axis=1) # Concatenate OHLC + Volume
'''
df.set_index('date', inplace=True) # Index by date
vol = df['total'].to_frame('volume') # Will deal with vol separately, as ohlc() messes it up
df.drop('total', axis=1, inplace=True) # Drop volume data from dataframe
ohlc = df.resample('T').ohlc() # Resample OHLC in 1min bins
ohlc.columns = ohlc.columns.map(lambda t: t[1]) # Raname columns by dropping 'rate'
closes = ohlc['close'].fillna(method='pad') # Pad forward missing 'close'
ohlc = ohlc.apply(lambda x: x.fillna(closes)) # Fill N/A with last close
vol = vol.resample('T').sum().fillna(0) # Add volumes by bin
ohlcv = pd.concat([ohlc,vol], axis=1) # Concatenate OHLC + Volume
return ohlcv
'''
def write_ohlcv_file(self, currencyPair):
'''
Generates OHLCV data file with 1minute bars from TradeHistory on disk
'''
def write_ohlcv_file(self, currencyPair):
'''
csv_trades = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
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.')
# 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'],
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 = 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)
try:
@@ -240,20 +296,29 @@ class PoloniexCurator(object):
log.debug(currencyPair+': Generated 1min OHLCV data.')
'''
Returns a data frame for a given currencyPair from data on disk
'''
def onemin_to_dataframe(self, currencyPair, start, end):
'''
Returns a data frame for a given currencyPair from data on disk
'''
csv_fn = CSV_OUT_FOLDER + 'crypto_1min-' + currencyPair + '.csv'
df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume'])
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]
'''
Generates a symbols.json file with corresponding start_date for each currencyPair
'''
def generate_symbols_json(self, filename=None):
'''
Generates a symbols.json file with corresponding start_date
for each currencyPair
'''
symbol_map = {}
if(filename is None):
@@ -262,14 +327,16 @@ class PoloniexCurator(object):
with open(filename, 'w') as symbols:
for currencyPair in self.currency_pairs:
start = None
csv_fn = CSV_OUT_FOLDER + 'crypto_trades-' + currencyPair + '.csv'
csv_fn = '{}crypto_trades-{}.csv'.format(
CSV_OUT_FOLDER, currencyPair)
with open(csv_fn, 'r') as f:
f.seek(0, os.SEEK_END)
if(f.tell() > 2): # First check file is not zero size
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.
start = pd.to_datetime( f.readline().split(',')[1], infer_datetime_format=True)
if(f.tell() > 2): # First check file is not zero size
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.
start = pd.to_datetime( f.readline().split(',')[1],
infer_datetime_format=True)
if(start is None):
start = time.gmtime()
@@ -279,7 +346,8 @@ class PoloniexCurator(object):
symbol = symbol,
start_date = start.strftime("%Y-%m-%d")
)
json.dump(symbol_map, symbols, sort_keys=True, indent=2, separators=(',',':'))
json.dump(symbol_map, symbols, sort_keys=True, indent=2,
separators=(',',':'))
if __name__ == '__main__':
@@ -289,6 +357,7 @@ if __name__ == '__main__':
for currencyPair in pc.currency_pairs:
pc.retrieve_trade_history(currencyPair)
pc.write_ohlcv_file(currencyPair)
log.debug('{} up to date.'.format(currencyPair))
#pc.write_ohlcv_file(currencyPair)