diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index e92a9a11..f1ba76f1 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -67,18 +67,68 @@ class PoloniexCurator(object): return DT_START - def get_data(self, currencyPair, start, end=9999999999, period=300): - url = self._api_path + 'command=returnChartData¤cyPair=' + currencyPair + '&start=' + str(start) + '&end=' + str(end) + '&period=' + str(period) + def get_data(self, currencyPair, start, end=9999999999, prev_df=None): + log.debug(currencyPair+': Retrieving from '+str(start)+' to '+str(end)) + + ''' + Poloniex limits a single query to returnTradeHistory to less than a year between start and end + ''' + 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 + else: + newstart = start + + url = self._api_path + 'command=returnTradeHistory¤cyPair=' + currencyPair + '&start=' + str(newstart) + '&end=' + str(end) try: response = requests.get(url) except Exception as e: - log.error('Failed to retrieve candlestick chart data for %s' % currencyPair) + log.error('Failed to retrieve trade history data for %s' % currencyPair) log.exception(e) return None - return response.json() + log.debug(currencyPair+': Received '+str(len(response.json()))+' trades.') + if(len(response.json())==1 and not isinstance(response.json(),list)): + r = response.json() + print(r) + if(r['error']): + log.error(r['error']) + return None + df = pd.DataFrame(data=response.json(), columns = ['date','rate', 'total', 'tradeID']) + df['rate'] = pd.to_numeric( df['rate'], errors='coerce') # Convert rate to float + df['total'] = pd.to_numeric( df['total'], errors='coerce') # Convert vol to float + df['tradeID'] = pd.to_numeric( df['tradeID'], errors='coerce') # Convert vol to float + df['date'] = pd.to_datetime(df['date'], infer_datetime_format=True) # Convert date + df.set_index('tradeID', inplace=True) # Index by tradeID + df = df.iloc[::-1] # Reverse timeseries as TradeHistory is provided newest to oldest + + if(prev_df is not None): + if(prev_df.index[0] == df.index[0]): + return prev_df + df = prev_df.combine_first(df) + + first = df['date'].iloc[0].value // 10 ** 9 + df = self.get_data( currencyPair, start, first, df ) + return df + + + 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 5min 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 + ''' Pulls latest data for a single pair ''' @@ -88,21 +138,24 @@ class PoloniexCurator(object): start = self._get_start_date(csv_fn) # Only fetch data if more than 5min have passed since last fetch if (time.time() > start): - data = self.get_data(currencyPair, start) + data = self.get_data(currencyPair, start) + if data is not None: + ohlcv = self.generate_ohlcv(data) + try: with open(csv_fn, 'ab') as csvfile: csvwriter = csv.writer(csvfile) - for item in data: - if item['date'] == 0: + for item in ohlcv.itertuples(): + if item.Index == 0: continue csvwriter.writerow([ - item['date'], - item['open'], - item['high'], - item['low'], - item['close'], - item['volume'], + item.Index.value // 10 ** 9, + item.open, + item.high, + item.low, + item.close, + item.volume, ]) except Exception as e: log.error('Error opening %s' % csv_fn)