From 3b6a293019be17c82d0b583d6753fde1690da7e4 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Tue, 20 Jun 2017 16:37:04 -0400 Subject: [PATCH] WIP: Poloniex bundle --- catalyst/data/bundles/__init__.py | 2 + catalyst/data/bundles/poloniex.py | 190 ++++++++++++++++++++++++++++++ curate/crypto_price_generator.py | 16 ++- 3 files changed, 204 insertions(+), 4 deletions(-) create mode 100644 catalyst/data/bundles/poloniex.py diff --git a/catalyst/data/bundles/__init__.py b/catalyst/data/bundles/__init__.py index e79d35b1..0c90c772 100644 --- a/catalyst/data/bundles/__init__.py +++ b/catalyst/data/bundles/__init__.py @@ -13,6 +13,7 @@ from .core import ( unregister, ) from .yahoo import yahoo_equities +from .poloniex import poloniex_cryptoassets __all__ = [ 'UnknownBundle', @@ -26,4 +27,5 @@ __all__ = [ 'to_bundle_ingest_dirname', 'unregister', 'yahoo_equities', + 'poloniex_cryptoassets', ] diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py new file mode 100644 index 00000000..26344b22 --- /dev/null +++ b/catalyst/data/bundles/poloniex.py @@ -0,0 +1,190 @@ +import os +from datetime import datetime + +import numpy as np +import pandas as pd +from pandas_datareader.data import DataReader +import requests + +from catalyst.utils.calendars import register_calendar_alias +from catalyst.utils.cli import maybe_show_progress +from .core import register + + +def _cachpath(symbol, type_): + return '-'.join((symbol.replace(os.path.sep, '_'), type_)) + + +def poloniex_cryptoassets(symbols, start=None, end=None): + """Create a data bundle ingest function from a set of symbols loaded from + poloniex + + Parameters + ---------- + symbols : iterable[str] + The ticker symbols to load data for. + start : datetime, optional + The start date to query for. By default this pulls the full history + for the calendar. + end : datetime, optional + The end date to query for. By default this pulls the full history + for the calendar. + + Returns + ------- + ingest : callable + The bundle ingest function for the given set of symbols. + + Examples + -------- + This code should be added to ~/.catalyst/extension.py + + .. code-block:: python + + from catalyst.data.bundles import poloniex_cryptoassets, register + + symbols = ( + 'USDT_BTC', + 'USDT_ETH', + 'USDT_LTC', + ) + register('my_bundle', poloniex_cryptoassets(symbols)) + + Notes + ----- + The sids for each symbol will be the index into the symbols sequence. + """ + # strict this in memory so that we can reiterate over it + symbols = tuple(symbols) + + def ingest(environ, + asset_db_writer, + minute_bar_writer, # unused + daily_bar_writer, + adjustment_writer, + calendar, + start_session, + end_session, + cache, + show_progress, + output_dir, + # pass these as defaults to make them 'nonlocal' in py2 + start=start, + end=end): + if start is None: + start = start_session + if end is None: + end = None + + metadata = pd.DataFrame(np.empty(len(symbols), dtype=[ + ('start_date', 'datetime64[ns]'), + ('end_date', 'datetime64[ns]'), + ('auto_close_date', 'datetime64[ns]'), + ('symbol', 'object'), + ])) + + def _pricing_iter(): + sid = 0 + + for symbol in symbols: + #def to_dataframe(self, start, end, currencyPair=None): + csv_fn = '/var/tmp/' + 'crypto_prices-' + symbol + '.csv' # TODO: DIR as parameter + #last_date = self._get_start_date(csv_fn) + #if last_date + 300 < end or not os.path.exists(csv_fn): + # get latest data + #self.append_data_single_pair(currencyPair) + + # CSV holds the latest snapshot + df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume']) + df['date']=pd.to_datetime(df['date'], utc=True, unit='s') + df.set_index('date', inplace=True) + + df = df.resample('D').mean() + + # ToDo: we assume that the source is always up to date and complete, otherwise fetch + if(pd.to_datetime(start).tz_convert(None) < df.index[0]): df_start = df.index[0] + else: df_start = pd.to_datetime(start).tz_convert(None) + if(pd.to_datetime(end).tz_convert(None) > df.index[-1]): df_end = df.index[-1] + else: df_end = pd.to_datetime(end).tz_convert(None) + + df = df.loc[ df_start : df_end ] + + # the start date is the date of the first trade and + # the end date is the date of the last trade + start_date = df.index[0] + end_date = df.index[-1] + # The auto_close date is the day after the last trade. + ac_date = end_date + pd.Timedelta(days=1) + metadata.iloc[sid] = start_date, end_date, ac_date, symbol + + yield sid, df + sid += 1 + + ''' + with maybe_show_progress( + symbols, + show_progress, + label='Downloading Yahoo pricing data: ') as it, \ + requests.Session() as session: + for symbol in it: + path = _cachpath(symbol, 'ohlcv') + try: + df = cache[path] + except KeyError: + df = cache[path] = DataReader( + symbol, + 'yahoo', + start, + end, + session=session, + ).sort_index() + + # the start date is the date of the first trade and + # the end date is the date of the last trade + start_date = df.index[0] + end_date = df.index[-1] + # The auto_close date is the day after the last trade. + ac_date = end_date + pd.Timedelta(days=1) + metadata.iloc[sid] = start_date, end_date, ac_date, symbol + + df.rename( + columns={ + 'Open': 'open', + 'High': 'high', + 'Low': 'low', + 'Close': 'close', + 'Volume': 'volume', + }, + inplace=True, + ) + yield sid, df + sid += 1 + ''' + daily_bar_writer.write(_pricing_iter(), show_progress=show_progress) + + symbol_map = pd.Series(metadata.symbol.index, metadata.symbol) + + # Hardcode the exchange to "POLONIEX" for all assets and (elsewhere) + # register "YAHOO" to resolve to the OPEN calendar, because these are + # all cryptoassets and thus use the OPEN calendar. + metadata['exchange'] = "POLONIEX" + asset_db_writer.write(equities=metadata) + + return ingest + + +# bundle used when creating test data +register( + '.test-poloniex', + poloniex_cryptoassets( + ( + 'USDT_BTC', + 'USDT_ETH', + 'USDT_LTC', + ), + pd.Timestamp('2010-01-01', tz='utc'), + pd.Timestamp('2015-01-01', tz='utc'), + ), +) + +register_calendar_alias("POLONIEX", "OPEN") diff --git a/curate/crypto_price_generator.py b/curate/crypto_price_generator.py index b8fdc6ac..49ab18a9 100644 --- a/curate/crypto_price_generator.py +++ b/curate/crypto_price_generator.py @@ -6,6 +6,8 @@ import time import requests import logbook +import catalyst.data.bundles.core as bundles + DT_START = time.mktime(datetime(2010, 01, 01, 0, 0).timetuple()) # DT_START = time.mktime(datetime(2017, 06, 13, 0, 0).timetuple()) # TODO: remove temp CSV_OUT_FOLDER = 'data/' @@ -125,13 +127,19 @@ class PoloniexDataGenerator(object): # CSV holds the latest snapshot df = pd.read_csv(csv_fn, names=['date', 'open', 'high', 'low', 'close', 'volume']) - df.columns = ['date', 'open', 'high', 'low', 'close', 'volume'] - return df.loc[(df['date'] > start) & (df['date'] <= end)] + df['date']=pd.to_datetime(df['date'],unit='s') + df.set_index('date', inplace=True) + + #return df.loc[(df.index > start) & (df.index <= end)] + return df[datetime.fromtimestamp(start):datetime.fromtimestamp(end-1)] if __name__ == '__main__': pdg = PoloniexDataGenerator() - pdg.get_currency_pairs() - pdg.append_data() +# pdg.get_currency_pairs() +# pdg.append_data() + df = pdg.to_dataframe(time.mktime(datetime(2017, 6, 01, 0, 0).timetuple()),time.mktime(datetime(2017, 6, 02, 0, 0).timetuple()),'USDT_BTC') + print(df) + # from zipline.utils.calendars import get_calendar