From ad95369028acf81219b094e42e784867ececc8e3 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 4 Sep 2017 11:23:40 -0400 Subject: [PATCH 01/19] Improved handling of insufficient funds on bittrex --- catalyst/exchange/bittrex/bittrex.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/catalyst/exchange/bittrex/bittrex.py b/catalyst/exchange/bittrex/bittrex.py index dfd1a496..cdd2d18a 100644 --- a/catalyst/exchange/bittrex/bittrex.py +++ b/catalyst/exchange/bittrex/bittrex.py @@ -119,7 +119,14 @@ class Bittrex(Exchange): ) return order else: - raise CreateOrderError(exchange=self.name, error=order_status) + if order_status == 'INSUFFICIENT_FUNDS': + log.warn('not enough funds to create order') + return None + else: + raise CreateOrderError( + exchange=self.name, + error=order_status + ) else: raise InvalidOrderStyle(exchange=self.name, style=style.__class__.__name__) From a58e3522a9bd4d9693e7d18bd65c23c3720501aa Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 4 Sep 2017 11:30:32 -0400 Subject: [PATCH 02/19] Improved handling of insufficient funds on bittrex --- catalyst/exchange/bittrex/bittrex.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/catalyst/exchange/bittrex/bittrex.py b/catalyst/exchange/bittrex/bittrex.py index cdd2d18a..f56073c5 100644 --- a/catalyst/exchange/bittrex/bittrex.py +++ b/catalyst/exchange/bittrex/bittrex.py @@ -122,6 +122,10 @@ class Bittrex(Exchange): if order_status == 'INSUFFICIENT_FUNDS': log.warn('not enough funds to create order') return None + elif order_status == 'DUST_TRADE_DISALLOWED_MIN_VALUE_50K_SAT': + log.warn('Your order is too small, order at least 50K' + ' Satoshi') + return None else: raise CreateOrderError( exchange=self.name, From 7247b761d51e15cc95f9e3e16aec0202e8d55c41 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Mon, 4 Sep 2017 11:33:21 -0400 Subject: [PATCH 03/19] Fixed an issue with failed orders --- catalyst/exchange/algorithm_exchange.py | 5 +++-- catalyst/exchange/exchange.py | 9 +++++---- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index 4fcc2517..80b1c5c5 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -401,8 +401,9 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): if order_id is not None: order = self.portfolio.open_orders[order_id] self.perf_tracker.process_order(order) - - return order + return order + else: + return None def round_order(self, amount): """ diff --git a/catalyst/exchange/exchange.py b/catalyst/exchange/exchange.py index 33712813..b0d0f5e2 100644 --- a/catalyst/exchange/exchange.py +++ b/catalyst/exchange/exchange.py @@ -531,10 +531,11 @@ class Exchange: ) ) order = self.create_order(asset, amount, is_buy, style) - - self._portfolio.create_order(order) - - return order.id + if order: + self._portfolio.create_order(order) + return order.id + else: + return None @abstractmethod def get_open_orders(self, asset): From 6e3017010f223fbd735da05b4967b7ceb95eb28b Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 5 Sep 2017 00:51:38 -0400 Subject: [PATCH 04/19] Added the initial version of a live graph --- catalyst/__main__.py | 12 +- catalyst/exchange/algorithm_exchange.py | 96 +++++++++- catalyst/exchange/exchange_utils.py | 32 ++++ catalyst/exchange/live_graph_clock.py | 177 ++++++++++++++++++ .../{exchange_clock.py => simple_clock.py} | 2 +- catalyst/utils/run_algo.py | 14 +- 6 files changed, 317 insertions(+), 16 deletions(-) create mode 100644 catalyst/exchange/live_graph_clock.py rename catalyst/exchange/{exchange_clock.py => simple_clock.py} (98%) diff --git a/catalyst/__main__.py b/catalyst/__main__.py index 36593bab..7a9e6316 100644 --- a/catalyst/__main__.py +++ b/catalyst/__main__.py @@ -210,6 +210,12 @@ def ipython_only(option): help='The base currency used to calculate statistics ' '(e.g. usd, btc, eth).', ) +@click.option( + '--live-graph/--no-live-graph', + is_flag=True, + default=False, + help='Display live graph.', +) @click.pass_context def run(ctx, algofile, @@ -227,7 +233,8 @@ def run(ctx, live, exchange_name, algo_namespace, - base_currency): + base_currency, + live_graph): """Run a backtest for the given algorithm. """ @@ -283,7 +290,8 @@ def run(ctx, live=live, exchange=exchange_name, algo_namespace=algo_namespace, - base_currency=base_currency + base_currency=base_currency, + live_graph=live_graph ) if output == '-': diff --git a/catalyst/exchange/algorithm_exchange.py b/catalyst/exchange/algorithm_exchange.py index 80b1c5c5..e7bced7e 100644 --- a/catalyst/exchange/algorithm_exchange.py +++ b/catalyst/exchange/algorithm_exchange.py @@ -19,6 +19,7 @@ from time import sleep from os import listdir from os.path import isfile, join from collections import deque +import numpy as np import logbook import pandas as pd @@ -28,14 +29,16 @@ from catalyst.algorithm import TradingAlgorithm from catalyst.data.minute_bars import BcolzMinuteBarWriter, \ BcolzMinuteBarReader from catalyst.errors import OrderInBeforeTradingStart -from catalyst.exchange.exchange_clock import ExchangeClock +from catalyst.exchange.simple_clock import SimpleClock +from catalyst.exchange.live_graph_clock import LiveGraphClock from catalyst.exchange.exchange_errors import ( ExchangeRequestError, ExchangePortfolioDataError, ExchangeTransactionError ) from catalyst.exchange.exchange_utils import get_exchange_minute_writer_root, \ - save_algo_object, get_algo_object, get_algo_folder + save_algo_object, get_algo_object, get_algo_folder, get_algo_df, \ + save_algo_df from catalyst.exchange.stats_utils import get_pretty_stats from catalyst.finance.performance.period import calc_period_stats from catalyst.gens.tradesimulation import AlgorithmSimulator @@ -56,8 +59,19 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): def __init__(self, *args, **kwargs): self.exchange = kwargs.pop('exchange', None) self.algo_namespace = kwargs.pop('algo_namespace', None) - self.orders = {} + self.live_graph = kwargs.pop('live_graph', None) + + self._clock = None self.minute_stats = deque(maxlen=60) + + self.pnl_stats = get_algo_df(self.algo_namespace, 'pnl_stats') + + self.custom_signals_stats = \ + get_algo_df(self.algo_namespace, 'custom_signals_stats') + + self.exposure_stats = \ + get_algo_df(self.algo_namespace, 'exposure_stats') + self.is_running = True self.retry_check_open_orders = 5 @@ -122,6 +136,13 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): sys.exit(0) + @property + def clock(self): + if self._clock is None: + return self._create_clock() + else: + return self._clock + def _create_clock(self): # The calendar's execution times are the minutes over which we actually @@ -137,10 +158,21 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): # This method is taken from TradingAlgorithm. # The clock has been replaced to use RealtimeClock # TODO: should we apply a time skew? not sure to understand the utility. - return ExchangeClock( - self.sim_params.sessions, - time_skew=self.exchange.time_skew - ) + + log.debug('creating clock') + if self.live_graph: + self._clock = LiveGraphClock( + self.sim_params.sessions, + time_skew=self.exchange.time_skew, + context=self + ) + else: + self._clock = SimpleClock( + self.sim_params.sessions, + time_skew=self.exchange.time_skew + ) + + return self._clock def _create_generator(self, sim_params): if self.perf_tracker is None: @@ -156,7 +188,7 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): self, sim_params, self.data_portal, - self._create_clock(), + self.clock, self._create_benchmark_source(), self.restrictions, universe_func=self._calculate_universe @@ -222,6 +254,49 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): error=e ) + def add_pnl_stats(self, period_stats): + starting = period_stats['starting_cash'] + current = period_stats['portfolio_value'] + appreciation = (current / starting) - 1 + perc = (appreciation * 100) if current != 0 else 0 + + log.debug('adding pnl stats: {:6f}%'.format(perc)) + + df = pd.DataFrame( + data=[dict(performance=perc)], + index=[period_stats['period_close']] + ) + self.pnl_stats = pd.concat([self.pnl_stats, df]) + + save_algo_df(self.algo_namespace, 'pnl_stats', self.pnl_stats) + + def add_custom_signals_stats(self, period_stats): + log.debug('adding custom signals stats: {}'.format(self.recorded_vars)) + df = pd.DataFrame( + data=[self.recorded_vars], + index=[period_stats['period_close']], + ) + self.custom_signals_stats = pd.concat([self.custom_signals_stats, df]) + + save_algo_df(self.algo_namespace, 'custom_signals_stats', + self.custom_signals_stats) + + def add_exposure_stats(self, period_stats): + data = dict( + long_exposure=period_stats['long_exposure'], + base_currency=period_stats['ending_cash'] + ) + log.debug('adding exposure stats: {}'.format(data)) + + df = pd.DataFrame( + data=[data], + index=[period_stats['period_close']], + ) + self.exposure_stats = pd.concat([self.exposure_stats, df]) + + save_algo_df(self.algo_namespace, 'exposure_stats', + self.exposure_stats) + def prepare_period_stats(self, start_dt, end_dt): """ Creates a dictionary representing the state of the tracker. @@ -314,9 +389,14 @@ class ExchangeTradingAlgorithm(TradingAlgorithm): minute_stats = self.prepare_period_stats( data.current_dt, data.current_dt + timedelta(minutes=1)) + # Saving the last hour in memory self.minute_stats.append(minute_stats) + self.add_pnl_stats(minute_stats) + self.add_custom_signals_stats(minute_stats) + self.add_exposure_stats(minute_stats) + print_df = pd.DataFrame(list(self.minute_stats)) log.debug( 'statistics for the last {stats_minutes} minutes:\n{stats}'.format( diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index 3b559726..cc91b470 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -3,6 +3,7 @@ import os import pickle import urllib from datetime import date, datetime +import pandas as pd from catalyst.exchange.exchange_errors import ExchangeAuthNotFound, \ ExchangeSymbolsNotFound @@ -116,6 +117,37 @@ def append_algo_object(algo_name, key, obj, environ=None): pickle.dump(obj, handle, protocol=pickle.HIGHEST_PROTOCOL) +def get_algo_df(algo_name, key, environ=None, rel_path=None): + folder = get_algo_folder(algo_name, environ) + + if rel_path is not None: + folder = os.path.join(folder, rel_path) + + filename = os.path.join(folder, key + '.csv') + + if os.path.isfile(filename): + try: + with open(filename, 'rb') as handle: + return pd.read_csv(handle, index_col=0, parse_dates=True) + except IOError: + return pd.DataFrame() + else: + return pd.DataFrame() + + +def save_algo_df(algo_name, key, df, environ=None, rel_path=None): + folder = get_algo_folder(algo_name, environ) + + if rel_path is not None: + folder = os.path.join(folder, rel_path) + ensure_directory(folder) + + filename = os.path.join(folder, key + '.csv') + + with open(filename, 'wb') as handle: + df.to_csv(handle) + + def get_exchange_minute_writer_root(exchange_name, environ=None): exchange_folder = get_exchange_folder(exchange_name, environ) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py new file mode 100644 index 00000000..589446ff --- /dev/null +++ b/catalyst/exchange/live_graph_clock.py @@ -0,0 +1,177 @@ +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from time import sleep + +import numpy as np +from matplotlib import pyplot as plt +from matplotlib import style +import pandas as pd +from catalyst.gens.sim_engine import ( + BAR, + SESSION_START, + MINUTE_END, + SESSION_END +) +from logbook import Logger +import matplotlib.dates as mdates + +log = Logger('LiveGraphClock') + +style.use('dark_background') + +days = mdates.DayLocator() +hours = mdates.HourLocator() +fmt = mdates.DateFormatter('%Y-%m-%d %H:%M') + + +class LiveGraphClock(object): + """Realtime clock for live trading. + + This class is a drop-in replacement for + :class:`zipline.gens.sim_engine.MinuteSimulationClock`. + + This is a stripped down version because crypto exchanges run around the clock. + + The :param:`time_skew` parameter represents the time difference between + the Broker and the live trading machine's clock. + """ + + def __init__(self, sessions, context, time_skew=pd.Timedelta('0s')): + + self.sessions = sessions + self.time_skew = time_skew + self._last_emit = None + self._before_trading_start_bar_yielded = True + self.context = context + + fig = plt.figure() + fig.canvas.set_window_title('Enigma Catalyst: {}'.format( + self.context.algo_namespace)) + + self.ax_pnl = fig.add_subplot(311) + self.draw_pnl() + + self.ax_custom_signals = fig.add_subplot(312, sharex=self.ax_pnl) + self.draw_custom_signals() + + self.ax_exposure = fig.add_subplot(313, sharex=self.ax_pnl) + self.draw_exposure() + + # rotates and right aligns the x labels, and moves the bottom of the + # axes up to make room for them + fig.autofmt_xdate() + fig.subplots_adjust(hspace=0.5) + + plt.tight_layout() + plt.ion() + plt.show() + + def format_ax(self, ax): + ax.xaxis.set_major_locator(hours) + ax.xaxis.set_major_formatter(fmt) + ax.xaxis.set_minor_locator(days) + ax.grid(True) + + def draw_pnl(self): + ax = self.ax_pnl + df = self.context.pnl_stats + + ax.clear() + ax.set_title('Performance') + ax.plot(df.index, df['performance'], '-', + color='green', + linewidth=1.0, + label='Performance' + ) + + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + + def perc(val): + return '{:2f}'.format(val) + + ax.format_ydata = perc + + self.format_ax(ax) + + def draw_custom_signals(self): + ax = self.ax_custom_signals + df = self.context.custom_signals_stats + + colors = ['blue', 'green', 'red', 'black', 'orange', 'yellow', 'pink'] + + ax.clear() + ax.set_title('Custom Signals') + for index, column in enumerate(df.columns.values.tolist()): + ax.plot(df.index, df[column], '-', + color=colors[index], + linewidth=1.0, + label=column + ) + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + self.format_ax(ax) + + def draw_exposure(self): + ax = self.ax_exposure + context = self.context + df = context.exposure_stats + + ax.clear() + ax.set_title('Exposure') + ax.plot(df.index, df['base_currency'], '-', + color='green', + linewidth=1.0, + label='Base Currency: {}'.format( + context.exchange.base_currency.upper() + ) + ) + + positions = context.exchange.portfolio.positions + symbols = [] + for position in positions: + symbols.append(position.symbol) + + ax.plot(df.index, df['long_exposure'], '-', + color='blue', + linewidth=1.0, + label='Long Exposure: {}'.format( + ', '.join(symbols).upper() + ) + ) + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + self.format_ax(ax) + + def __iter__(self): + yield pd.Timestamp.utcnow(), SESSION_START + + while True: # plot x^1, x^2, ..., x^4 + current_time = pd.Timestamp.utcnow() + current_minute = current_time.floor('1 min') + + if self._last_emit is None or current_minute > self._last_emit: + log.debug('emitting minutely bar: {}'.format(current_minute)) + + self._last_emit = current_minute + yield current_minute, BAR + + self.draw_pnl() + self.draw_custom_signals() + self.draw_exposure() + + plt.draw() + + else: + # I can't use the "animate" reactive approach here because + # I need to yield from the main loop. + # Workaround: https://stackoverflow.com/a/33050617/814633 + plt.pause(0.001) diff --git a/catalyst/exchange/exchange_clock.py b/catalyst/exchange/simple_clock.py similarity index 98% rename from catalyst/exchange/exchange_clock.py rename to catalyst/exchange/simple_clock.py index 4e180816..1f4d3ffa 100644 --- a/catalyst/exchange/exchange_clock.py +++ b/catalyst/exchange/simple_clock.py @@ -25,7 +25,7 @@ from logbook import Logger log = Logger('ExchangeClock') -class ExchangeClock(object): +class SimpleClock(object): """Realtime clock for live trading. This class is a drop-in replacement for diff --git a/catalyst/utils/run_algo.py b/catalyst/utils/run_algo.py index 42f4f349..fc9f6354 100644 --- a/catalyst/utils/run_algo.py +++ b/catalyst/utils/run_algo.py @@ -95,7 +95,8 @@ def _run(handle_data, live, exchange, algo_namespace, - base_currency): + base_currency, + live_graph): """Run a backtest for the given algorithm. This is shared between the cli and :func:`catalyst.run_algo`. @@ -277,7 +278,8 @@ def _run(handle_data, ) env = TradingEnvironment( - load=partial(load_crypto_market_data, bundle=b, bundle_data=bundle_data, environ=environ), + load=partial(load_crypto_market_data, bundle=b, + bundle_data=bundle_data, environ=environ), bm_symbol='USDT_BTC', trading_calendar=open_calendar, asset_db_path=connstr, @@ -338,7 +340,7 @@ def _run(handle_data, TradingAlgorithmClass = ( partial(ExchangeTradingAlgorithm, exchange=exchange, - algo_namespace=algo_namespace) + algo_namespace=algo_namespace, live_graph=live_graph) if live and exchange else TradingAlgorithm) perf = TradingAlgorithmClass( @@ -439,7 +441,8 @@ def run_algorithm(initialize, live=False, exchange_name=None, base_currency=None, - algo_namespace=None): + algo_namespace=None, + live_graph=False): """Run a trading algorithm. Parameters @@ -552,5 +555,6 @@ def run_algorithm(initialize, live=live, exchange=exchange_name, algo_namespace=algo_namespace, - base_currency=base_currency + base_currency=base_currency, + live_graph=live_graph ) From c4900af08887fb434f8c5f189d1a696a3aaae2a1 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 5 Sep 2017 00:58:11 -0400 Subject: [PATCH 05/19] Minor fix in the graph logic --- catalyst/exchange/live_graph_clock.py | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py index 589446ff..1dfa7cb2 100644 --- a/catalyst/exchange/live_graph_clock.py +++ b/catalyst/exchange/live_graph_clock.py @@ -11,20 +11,15 @@ # See the License for the specific language governing permissions and # limitations under the License. -from time import sleep - -import numpy as np -from matplotlib import pyplot as plt -from matplotlib import style +import matplotlib.dates as mdates import pandas as pd from catalyst.gens.sim_engine import ( BAR, - SESSION_START, - MINUTE_END, - SESSION_END + SESSION_START ) from logbook import Logger -import matplotlib.dates as mdates +from matplotlib import pyplot as plt +from matplotlib import style log = Logger('LiveGraphClock') @@ -164,11 +159,14 @@ class LiveGraphClock(object): self._last_emit = current_minute yield current_minute, BAR - self.draw_pnl() - self.draw_custom_signals() - self.draw_exposure() + try: + self.draw_pnl() + self.draw_custom_signals() + self.draw_exposure() - plt.draw() + plt.draw() + except Exception as e: + log.warn('Unable to update the graph: {}'.format(e)) else: # I can't use the "animate" reactive approach here because From 6a98a937dd3146d1e82d72709e0aec9756373468 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 5 Sep 2017 01:15:40 -0400 Subject: [PATCH 06/19] Minor fix in the graph logic --- catalyst/exchange/live_graph_clock.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py index 1dfa7cb2..bd76a38d 100644 --- a/catalyst/exchange/live_graph_clock.py +++ b/catalyst/exchange/live_graph_clock.py @@ -172,4 +172,4 @@ class LiveGraphClock(object): # I can't use the "animate" reactive approach here because # I need to yield from the main loop. # Workaround: https://stackoverflow.com/a/33050617/814633 - plt.pause(0.001) + plt.pause(1) From 6e6c62533b61639c00b99226f912ec1c26cc2ba9 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Tue, 5 Sep 2017 11:15:34 -0400 Subject: [PATCH 07/19] Fixes to the graph timeline axis --- catalyst/exchange/live_graph_clock.py | 63 +++++++++++++++++++++------ 1 file changed, 49 insertions(+), 14 deletions(-) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py index bd76a38d..877f8115 100644 --- a/catalyst/exchange/live_graph_clock.py +++ b/catalyst/exchange/live_graph_clock.py @@ -10,6 +10,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. +from datetime import timedelta import matplotlib.dates as mdates import pandas as pd @@ -25,8 +26,6 @@ log = Logger('LiveGraphClock') style.use('dark_background') -days = mdates.DayLocator() -hours = mdates.HourLocator() fmt = mdates.DateFormatter('%Y-%m-%d %H:%M') @@ -36,10 +35,24 @@ class LiveGraphClock(object): This class is a drop-in replacement for :class:`zipline.gens.sim_engine.MinuteSimulationClock`. - This is a stripped down version because crypto exchanges run around the clock. + This mixes the clock with a live graph. + + Note + ---- + This seemingly awkward approach allows us to run the program using a single + thread. This is important because Matplotlib does not play nice with + multi-threaded environments. Zipline probably does not either. + + + Matplotlib has a pause() method which is a wrapper around time.sleep() + used in the SimpleClock. The key difference is that users + can still interact with the chart during the pause cycles. This is + what enables us to keep a single thread. This is also why we are not using + the 'animate' callback of Matplotlib. We need to direct access to the + __iter__ method in order to yield events to Zipline. The :param:`time_skew` parameter represents the time difference between - the Broker and the live trading machine's clock. + the exchange and the live trading machine's clock. It's not used currently. """ def __init__(self, sessions, context, time_skew=pd.Timedelta('0s')): @@ -55,13 +68,15 @@ class LiveGraphClock(object): self.context.algo_namespace)) self.ax_pnl = fig.add_subplot(311) - self.draw_pnl() self.ax_custom_signals = fig.add_subplot(312, sharex=self.ax_pnl) - self.draw_custom_signals() self.ax_exposure = fig.add_subplot(313, sharex=self.ax_pnl) - self.draw_exposure() + + if len(context.minute_stats) > 0: + self.draw_pnl() + self.draw_custom_signals() + self.draw_exposure() # rotates and right aligns the x labels, and moves the bottom of the # axes up to make room for them @@ -73,11 +88,29 @@ class LiveGraphClock(object): plt.show() def format_ax(self, ax): - ax.xaxis.set_major_locator(hours) + """ + Trying to assign reasonable parameters to the time axis. + + TODO: room for improvement + + :param ax: + :return: + """ + ax.xaxis.set_major_locator(mdates.DayLocator(interval=1)) ax.xaxis.set_major_formatter(fmt) - ax.xaxis.set_minor_locator(days) + + locator = mdates.HourLocator(interval=4) + locator.MAXTICKS = 5000 + ax.xaxis.set_minor_locator(locator) + + datemin = pd.Timestamp.utcnow() + ax.set_xlim(datemin) + ax.grid(True) + def set_legend(self, ax): + ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + def draw_pnl(self): ax = self.ax_pnl df = self.context.pnl_stats @@ -90,13 +123,12 @@ class LiveGraphClock(object): label='Performance' ) - ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) - def perc(val): return '{:2f}'.format(val) ax.format_ydata = perc + self.set_legend(ax) self.format_ax(ax) def draw_custom_signals(self): @@ -113,7 +145,8 @@ class LiveGraphClock(object): linewidth=1.0, label=column ) - ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + + self.set_legend(ax) self.format_ax(ax) def draw_exposure(self): @@ -143,13 +176,14 @@ class LiveGraphClock(object): ', '.join(symbols).upper() ) ) - ax.legend(loc='upper left', ncol=1, fontsize=10, numpoints=1) + + self.set_legend(ax) self.format_ax(ax) def __iter__(self): yield pd.Timestamp.utcnow(), SESSION_START - while True: # plot x^1, x^2, ..., x^4 + while True: current_time = pd.Timestamp.utcnow() current_minute = current_time.floor('1 min') @@ -171,5 +205,6 @@ class LiveGraphClock(object): else: # I can't use the "animate" reactive approach here because # I need to yield from the main loop. + # Workaround: https://stackoverflow.com/a/33050617/814633 plt.pause(1) From 8850657f261d05b11213fc04ca97783a36148821 Mon Sep 17 00:00:00 2001 From: fredfortier Date: Thu, 7 Sep 2017 14:26:55 -0400 Subject: [PATCH 08/19] Fixed path issue with obsolete branch --- catalyst/exchange/exchange_utils.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/catalyst/exchange/exchange_utils.py b/catalyst/exchange/exchange_utils.py index cc91b470..2638c1c9 100644 --- a/catalyst/exchange/exchange_utils.py +++ b/catalyst/exchange/exchange_utils.py @@ -9,8 +9,9 @@ from catalyst.exchange.exchange_errors import ExchangeAuthNotFound, \ ExchangeSymbolsNotFound from catalyst.utils.paths import data_root, ensure_directory +# TODO: move to aws SYMBOLS_URL = 'https://raw.githubusercontent.com/enigmampc/catalyst/' \ - 'exchange-trading/catalyst/exchange/{exchange}/symbols.json' + 'master/catalyst/exchange/{exchange}/symbols.json' def get_exchange_folder(exchange_name, environ=None): From 48143d321266c4374731ccb1267a6ce70065c491 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Wed, 13 Sep 2017 17:22:11 -0600 Subject: [PATCH 09/19] WIP: Fixes 1/1000 price issue in history, and works with full coins. Requires matching-version 'catalyst ingest' --- catalyst/data/_equities.pyx | 2 +- catalyst/data/bundles/poloniex.py | 2 +- catalyst/data/history_loader.py | 2 +- catalyst/data/us_equity_pricing.py | 11 +++++++---- 4 files changed, 10 insertions(+), 7 deletions(-) diff --git a/catalyst/data/_equities.pyx b/catalyst/data/_equities.pyx index 81f9a66e..563fa56a 100644 --- a/catalyst/data/_equities.pyx +++ b/catalyst/data/_equities.pyx @@ -217,7 +217,7 @@ cpdef _read_bcolz_data(ctable_t table, if column_name in ['open', 'high', 'low', 'close']: where_nan = (outbuf == 0) - outbuf_as_float = outbuf.astype(float64) * .000001 + outbuf_as_float = outbuf.astype(float64) * .000000001 outbuf_as_float[where_nan] = NAN results.append(outbuf_as_float) elif column_name != 'volume': diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py index 4dcdf7bc..eb4fc735 100644 --- a/catalyst/data/bundles/poloniex.py +++ b/catalyst/data/bundles/poloniex.py @@ -105,7 +105,7 @@ class PoloniexBundle(BaseCryptoPricingBundle): # BcolzDailyBarReader introduces a 1/1000 factor in the way pricing is stored # on disk, which we compensate here to get the right pricing amounts # ref: data/us_equity_pricing.py - scale = 1000 + scale = 1 raw.loc[:, 'open'] /= scale raw.loc[:, 'high'] /= scale raw.loc[:, 'low'] /= scale diff --git a/catalyst/data/history_loader.py b/catalyst/data/history_loader.py index 8bc00563..e33f5f6e 100644 --- a/catalyst/data/history_loader.py +++ b/catalyst/data/history_loader.py @@ -38,7 +38,7 @@ from catalyst.utils.numpy_utils import float64_dtype from catalyst.utils.pandas_utils import find_in_sorted_index # Default number of decimal places used for rounding asset prices. -DEFAULT_ASSET_PRICE_DECIMALS = 3 +DEFAULT_ASSET_PRICE_DECIMALS = 9 class HistoryCompatibleUSEquityAdjustmentReader(object): diff --git a/catalyst/data/us_equity_pricing.py b/catalyst/data/us_equity_pricing.py index 03d01a4e..901a0e60 100644 --- a/catalyst/data/us_equity_pricing.py +++ b/catalyst/data/us_equity_pricing.py @@ -11,6 +11,9 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. + +from __future__ import division # Python2 req to have division of ints yield float + from errno import ENOENT from functools import partial from os import remove @@ -80,7 +83,6 @@ from catalyst.utils.cli import ( from ._equities import _compute_row_slices, _read_bcolz_data from ._adjustments import load_adjustments_from_sqlite - logger = logbook.Logger('UsEquityPricing') OHLC = frozenset(['open', 'high', 'low', 'close']) @@ -116,6 +118,8 @@ SQLITE_STOCK_DIVIDEND_PAYOUT_COLUMN_DTYPES = { UINT32_MAX = iinfo(uint32).max UINT64_MAX = iinfo(uint64).max +PRICE_ADJUSTMENT_FACTOR = 1000000000 # Provides 9 decimals resolution. Also affects _equities.pyx L220 + def check_uint32_safe(value, colname): if value >= UINT32_MAX: @@ -433,7 +437,7 @@ class BcolzDailyBarWriter(object): return raw_data winsorise_uint64(raw_data, invalid_data_behavior, 'volume', *OHLC) - processed = (raw_data[list(OHLC)] * 1000000).astype('uint64') + processed = (raw_data[list(OHLC)] * PRICE_ADJUSTMENT_FACTOR).astype('uint64') dates = raw_data.index.values.astype('datetime64[s]') check_uint32_safe(dates.max().view(np.int64), 'day') processed['day'] = dates.astype('uint32') @@ -519,7 +523,6 @@ class BcolzDailyBarReader(SessionBarReader): # Need to test keeping the entire array in memory for the course of a # process first. self._spot_cols = {} - self.PRICE_ADJUSTMENT_FACTOR = 0.001 self._read_all_threshold = read_all_threshold @lazyval @@ -763,7 +766,7 @@ class BcolzDailyBarReader(SessionBarReader): if price == 0: return nan else: - return price * 0.001 + return price / PRICE_ADJUSTMENT_FACTOR else: return price From e5a137f205fc55760832abd5db3fb906b27ffb44 Mon Sep 17 00:00:00 2001 From: Andrew Campbell Date: Fri, 21 Jul 2017 18:05:32 -0500 Subject: [PATCH 10/19] ENH: Bound trade amount with asset specific min trade size --- catalyst/algorithm.py | 17 +++++++---------- catalyst/assets/_assets.pyx | 15 +++++++++++---- catalyst/assets/asset_db_schema.py | 3 ++- catalyst/assets/asset_writer.py | 1 + catalyst/finance/slippage.py | 7 +++++-- catalyst/finance/transaction.py | 6 +----- catalyst/utils/math_utils.py | 2 ++ 7 files changed, 29 insertions(+), 22 deletions(-) diff --git a/catalyst/algorithm.py b/catalyst/algorithm.py index ed10fedd..5601f133 100644 --- a/catalyst/algorithm.py +++ b/catalyst/algorithm.py @@ -125,6 +125,7 @@ from catalyst.utils.factory import create_simulation_parameters from catalyst.utils.math_utils import ( tolerant_equals, round_if_near_integer, + round_nearest ) from catalyst.utils.pandas_utils import clear_dataframe_indexer_caches from catalyst.utils.preprocess import preprocess @@ -1488,7 +1489,7 @@ class TradingAlgorithm(object): def _calculate_order(self, asset, amount, limit_price=None, stop_price=None, style=None): - amount = self.round_order(amount) + amount = self.round_order(amount, asset) # Raises a ZiplineError if invalid parameters are detected. self.validate_order_params(asset, @@ -1505,16 +1506,13 @@ class TradingAlgorithm(object): return amount, style @staticmethod - def round_order(amount): + def round_order(amount, asset): """ - Convert number of shares to an integer. - - By default, truncates to the integer share count that's either within - .0001 of amount or closer to zero. - - E.g. 3.9999 -> 4.0; 5.5 -> 5.0; -5.5 -> -5.0 + Converts the number of shares to the smallest tradable lot size for + the asset being ordered. + """ - return int(round_if_near_integer(amount)) + return round_nearest(amount, asset.min_trade_size) def validate_order_params(self, asset, @@ -1550,7 +1548,6 @@ class TradingAlgorithm(object): self.updated_portfolio(), self.get_datetime(), self.trading_client.current_data) - @staticmethod def __convert_order_params_for_blotter(limit_price, stop_price, style): """ diff --git a/catalyst/assets/_assets.pyx b/catalyst/assets/_assets.pyx index cdc34008..05144223 100644 --- a/catalyst/assets/_assets.pyx +++ b/catalyst/assets/_assets.pyx @@ -59,6 +59,7 @@ cdef class Asset: cdef readonly object exchange cdef readonly object exchange_full + cdef readonly object min_trade_size _kwargnames = frozenset({ 'sid', @@ -70,6 +71,7 @@ cdef class Asset: 'auto_close_date', 'exchange', 'exchange_full', + 'min_trade_size', }) def __init__(self, @@ -81,7 +83,8 @@ cdef class Asset: object end_date=None, object first_traded=None, object auto_close_date=None, - object exchange_full=None): + object exchange_full=None, + object min_trade_size=None): self.sid = sid self.sid_hash = hash(sid) @@ -94,6 +97,7 @@ cdef class Asset: self.end_date = end_date self.first_traded = first_traded self.auto_close_date = auto_close_date + self.min_trade_size = min_trade_size def __int__(self): return self.sid @@ -148,7 +152,8 @@ cdef class Asset: def __repr__(self): attrs = ('symbol', 'asset_name', 'exchange', - 'start_date', 'end_date', 'first_traded', 'auto_close_date') + 'start_date', 'end_date', 'first_traded', 'auto_close_date', + 'min_trade_size') tuples = ((attr, repr(getattr(self, attr, None))) for attr in attrs) strings = ('%s=%s' % (t[0], t[1]) for t in tuples) @@ -170,7 +175,8 @@ cdef class Asset: self.end_date, self.first_traded, self.auto_close_date, - self.exchange_full)) + self.exchange_full, + self.min_trade_size)) cpdef to_dict(self): """ @@ -186,6 +192,7 @@ cdef class Asset: 'auto_close_date': self.auto_close_date, 'exchange': self.exchange, 'exchange_full': self.exchange_full, + 'min_trade_size': self.min_trade_size } @classmethod @@ -234,7 +241,7 @@ cdef class Equity(Asset): def __repr__(self): attrs = ('symbol', 'asset_name', 'exchange', 'start_date', 'end_date', 'first_traded', 'auto_close_date', - 'exchange_full') + 'exchange_full', 'min_trade_size') tuples = ((attr, repr(getattr(self, attr, None))) for attr in attrs) strings = ('%s=%s' % (t[0], t[1]) for t in tuples) diff --git a/catalyst/assets/asset_db_schema.py b/catalyst/assets/asset_db_schema.py index 9c9c98c2..35e062f9 100644 --- a/catalyst/assets/asset_db_schema.py +++ b/catalyst/assets/asset_db_schema.py @@ -39,7 +39,8 @@ equities = sa.Table( sa.Column('first_traded', sa.Integer), sa.Column('auto_close_date', sa.Integer), sa.Column('exchange', sa.Text), - sa.Column('exchange_full', sa.Text) + sa.Column('exchange_full', sa.Text), + sa.Column('min_trade_size', sa.Float) ) equity_symbol_mappings = sa.Table( diff --git a/catalyst/assets/asset_writer.py b/catalyst/assets/asset_writer.py index 9910db72..1ea5afb2 100644 --- a/catalyst/assets/asset_writer.py +++ b/catalyst/assets/asset_writer.py @@ -73,6 +73,7 @@ _equities_defaults = { 'exchange': None, # optional, something like "New York Stock Exchange" 'exchange_full': None, + 'min_trade_size': None } # Default values for the futures DataFrame diff --git a/catalyst/finance/slippage.py b/catalyst/finance/slippage.py index 118a5b58..cfba2683 100644 --- a/catalyst/finance/slippage.py +++ b/catalyst/finance/slippage.py @@ -41,6 +41,7 @@ DEFAULT_EQUITY_VOLUME_SLIPPAGE_BAR_LIMIT = 0.025 DEFAULT_FUTURE_VOLUME_SLIPPAGE_BAR_LIMIT = 0.05 + class LiquidityExceeded(Exception): pass @@ -205,12 +206,14 @@ class VolumeShareSlippage(SlippageModel): def process_order(self, data, order): volume = data.current(order.asset, "volume") + min_trade_size = order.asset.min_trade_size + max_volume = self.volume_limit * volume # price impact accounts for the total volume of transactions # created against the current minute bar remaining_volume = max_volume - self.volume_for_bar - if remaining_volume < 1: + if remaining_volume < min_trade_size: # we can't fill any more transactions raise LiquidityExceeded() @@ -218,7 +221,7 @@ class VolumeShareSlippage(SlippageModel): # volume available in the bar or the open amount. cur_volume = int(min(remaining_volume, abs(order.open_amount))) - if cur_volume < 1: + if cur_volume < min_trade_size: return None, None # tally the current amount into our total amount ordered. diff --git a/catalyst/finance/transaction.py b/catalyst/finance/transaction.py index 3a388890..8d215dcc 100644 --- a/catalyst/finance/transaction.py +++ b/catalyst/finance/transaction.py @@ -65,14 +65,10 @@ def create_transaction(order, dt, price, amount): # floor the amount to protect against non-whole number orders # TODO: Investigate whether we can add a robust check in blotter # and/or tradesimulation, as well. - amount_magnitude = int(abs(amount)) - - if amount_magnitude < 1: - raise Exception("Transaction magnitude must be at least 1.") transaction = Transaction( asset=order.asset, - amount=int(amount), + amount=amounts, dt=dt, price=price, order_id=order.id diff --git a/catalyst/utils/math_utils.py b/catalyst/utils/math_utils.py index 16fdb99d..7981a52b 100644 --- a/catalyst/utils/math_utils.py +++ b/catalyst/utils/math_utils.py @@ -17,6 +17,8 @@ import math from numpy import isnan +def round_nearest(x, a): + return round(round(x / a) * a, -int(math.floor(math.log10(a)))) def tolerant_equals(a, b, atol=10e-7, rtol=10e-7, equal_nan=False): """Check if a and b are equal with some tolerance. From c3897cfa5a1f598a6575337102c48dadb82c6f74 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Thu, 14 Sep 2017 15:22:51 -0600 Subject: [PATCH 11/19] ENH: wrapping up asset min_trade_size for fractional coinsup to 1/100000000th of a coin --- catalyst/assets/asset_writer.py | 4 +++- catalyst/data/bundles/base_pricing.py | 1 + catalyst/data/bundles/poloniex.py | 2 ++ catalyst/finance/slippage.py | 2 +- catalyst/finance/transaction.py | 2 +- 5 files changed, 8 insertions(+), 3 deletions(-) diff --git a/catalyst/assets/asset_writer.py b/catalyst/assets/asset_writer.py index 1ea5afb2..ec30ffab 100644 --- a/catalyst/assets/asset_writer.py +++ b/catalyst/assets/asset_writer.py @@ -73,7 +73,7 @@ _equities_defaults = { 'exchange': None, # optional, something like "New York Stock Exchange" 'exchange_full': None, - 'min_trade_size': None + 'min_trade_size': 1 } # Default values for the futures DataFrame @@ -391,6 +391,8 @@ class AssetDBWriter(object): The date on which to close any positions in this asset. exchange : str The exchange where this asset is traded. + min_trade_size: float, optional + The minimum denomination this asset can be traded. The index of this dataframe should contain the sids. futures : pd.DataFrame, optional diff --git a/catalyst/data/bundles/base_pricing.py b/catalyst/data/bundles/base_pricing.py index a0abd51a..c5281fdd 100644 --- a/catalyst/data/bundles/base_pricing.py +++ b/catalyst/data/bundles/base_pricing.py @@ -24,6 +24,7 @@ class BasePricingBundle(BaseBundle): ('start_date', 'datetime64[ns]'), ('end_date', 'datetime64[ns]'), ('ac_date', 'datetime64[ns]'), + ('min_trade_size', 'float'), ] @lazyval diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py index eb4fc735..8b5628d5 100644 --- a/catalyst/data/bundles/poloniex.py +++ b/catalyst/data/bundles/poloniex.py @@ -75,12 +75,14 @@ class PoloniexBundle(BaseCryptoPricingBundle): start_date = sym_data.index[0] end_date = sym_data.index[-1] ac_date = end_date + pd.Timedelta(days=1) + min_trade_size = 0.00000001 return ( sym_md.symbol, start_date, end_date, ac_date, + min_trade_size, ) def fetch_raw_symbol_frame(self, diff --git a/catalyst/finance/slippage.py b/catalyst/finance/slippage.py index cfba2683..36de4ec7 100644 --- a/catalyst/finance/slippage.py +++ b/catalyst/finance/slippage.py @@ -219,7 +219,7 @@ class VolumeShareSlippage(SlippageModel): # the current order amount will be the min of the # volume available in the bar or the open amount. - cur_volume = int(min(remaining_volume, abs(order.open_amount))) + cur_volume = min(remaining_volume, abs(order.open_amount)) if cur_volume < min_trade_size: return None, None diff --git a/catalyst/finance/transaction.py b/catalyst/finance/transaction.py index 8d215dcc..f9d72c38 100644 --- a/catalyst/finance/transaction.py +++ b/catalyst/finance/transaction.py @@ -68,7 +68,7 @@ def create_transaction(order, dt, price, amount): transaction = Transaction( asset=order.asset, - amount=amounts, + amount=amount, dt=dt, price=price, order_id=order.id From 72e07e242f39765599f1846d241c92610f2f2774 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Thu, 20 Jul 2017 01:37:55 -0400 Subject: [PATCH 12/19] WIP: curating 1min Poloniex data - no append --- catalyst/curate/poloniex.py | 79 +++++++++++++++++++++++++++++++------ 1 file changed, 66 insertions(+), 13 deletions(-) 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) From d1241252585646e7bced357c72472e1e159df3b9 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Fri, 21 Jul 2017 16:22:29 -0400 Subject: [PATCH 13/19] WIP: trades to disk - no append/no ingestion --- catalyst/curate/poloniex.py | 109 ++++++++++++++++++++++++++++++++++-- 1 file changed, 105 insertions(+), 4 deletions(-) diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index f1ba76f1..33eb46c9 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -9,6 +9,8 @@ import logbook DT_START = time.mktime(datetime(2010, 1, 1, 0, 0).timetuple()) CSV_OUT_FOLDER = '/var/tmp/catalyst/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'] + logbook.StderrHandler().push_application() log = logbook.Logger(__name__) @@ -114,6 +116,101 @@ class PoloniexCurator(object): df = self.get_data( currencyPair, start, first, df ) return df + def retrieve_trade_history(self, currencyPair, start, end=9999999999): + 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 + 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 + + 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 + ''' + 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 + + log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end)) + + 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 trade history data for %s' % currencyPair) + log.exception(e) + return None + + if('last_tradeID' in locals() and response.json()[-1]['tradeID'] == last_tradeID): # Got to the end of TradingHistory for this coin + return + + try: + with open(csv_fn, 'ab') as csvfile: + csvwriter = csv.writer(csvfile) + for item in response.json(): + if( 'last_tradeID' in locals() and item['tradeID'] >= last_tradeID ): + continue + csvwriter.writerow([ + item['tradeID'], + item['date'], + item['type'], + item['rate'], + item['amount'], + item['total'], + item['globalTradeID'] + ]) + 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): + + 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.') + else: + df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'] ) + 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: + with open(csv_1min, 'ab') as csvfile: + csvwriter = csv.writer(csvfile) + for item in ohlcv.itertuples(): + if item.Index == 0: + continue + csvwriter.writerow([ + 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) + log.exception(e) + log.debug(currencyPair+': Generated 1min OHLCV data.') def generate_ohlcv(self, df): @@ -171,10 +268,10 @@ class PoloniexCurator(object): for currencyPair in self.currency_pairs: self.append_data_single_pair(currencyPair) # Rate limit is 6 calls per second, sleep 1sec/6 to be safe - time.sleep(0.17) + #time.sleep(0.17) ''' - Returns a data frame for all pairs, or for the requests currency pair. + Returns a data frame for all pairs, or for the requested currency pair. Makes sure data is up to date ''' def to_dataframe(self, start, end, currencyPair=None): @@ -193,5 +290,9 @@ class PoloniexCurator(object): if __name__ == '__main__': pc = PoloniexCurator() - pc.get_currency_pairs() - pc.append_data() + #pc.get_currency_pairs() + #pc.append_data() + + for coin in COINS: + # pc.retrieve_trade_history(coin,DT_START) + pc.write_ohlcv_file(coin) From 01eefd67e01595a31b96e455e4b451ceab754550 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Fri, 15 Sep 2017 10:51:04 -0600 Subject: [PATCH 14/19] ingestion switch to create_writers when ingesting locally --- catalyst/data/bundles/poloniex.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py index eb4fc735..85ee6b8d 100644 --- a/catalyst/data/bundles/poloniex.py +++ b/catalyst/data/bundles/poloniex.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +import sys + from datetime import datetime import pandas as pd @@ -166,4 +168,9 @@ register_bundle(PoloniexBundle, ['USDT_BTC',]) For a production environment make sure to use (to bundle all pairs): register_bundle(PoloniexBundle) ''' -register_bundle(PoloniexBundle, create_writers=False) + +if 'ingest' in sys.argv and '-c' in sys.argv: + register_bundle(PoloniexBundle) +else: + register_bundle(PoloniexBundle, create_writers=False) + From 4a4277d9d1f26031b42e6b4a0c84b814b4c7429e Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Thu, 20 Jul 2017 01:37:55 -0400 Subject: [PATCH 15/19] WIP: curating 1min Poloniex data - no append --- catalyst/curate/poloniex.py | 79 +++++++++++++++++++++++++++++++------ 1 file changed, 66 insertions(+), 13 deletions(-) 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) From 6fddb925637d06cc7f3b3263e3e2172cc1338a62 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Fri, 21 Jul 2017 16:22:29 -0400 Subject: [PATCH 16/19] WIP: trades to disk - no append/no ingestion --- catalyst/curate/poloniex.py | 109 ++++++++++++++++++++++++++++++++++-- 1 file changed, 105 insertions(+), 4 deletions(-) diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index f1ba76f1..33eb46c9 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -9,6 +9,8 @@ import logbook DT_START = time.mktime(datetime(2010, 1, 1, 0, 0).timetuple()) CSV_OUT_FOLDER = '/var/tmp/catalyst/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'] + logbook.StderrHandler().push_application() log = logbook.Logger(__name__) @@ -114,6 +116,101 @@ class PoloniexCurator(object): df = self.get_data( currencyPair, start, first, df ) return df + def retrieve_trade_history(self, currencyPair, start, end=9999999999): + 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 + 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 + + 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 + ''' + 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 + + log.debug(currencyPair+': Retrieving from '+str(newstart)+' to '+str(end)) + + 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 trade history data for %s' % currencyPair) + log.exception(e) + return None + + if('last_tradeID' in locals() and response.json()[-1]['tradeID'] == last_tradeID): # Got to the end of TradingHistory for this coin + return + + try: + with open(csv_fn, 'ab') as csvfile: + csvwriter = csv.writer(csvfile) + for item in response.json(): + if( 'last_tradeID' in locals() and item['tradeID'] >= last_tradeID ): + continue + csvwriter.writerow([ + item['tradeID'], + item['date'], + item['type'], + item['rate'], + item['amount'], + item['total'], + item['globalTradeID'] + ]) + 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): + + 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.') + else: + df = pd.read_csv(csv_trades, names=['tradeID','date','type','rate','amount','total','globalTradeID'] ) + 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: + with open(csv_1min, 'ab') as csvfile: + csvwriter = csv.writer(csvfile) + for item in ohlcv.itertuples(): + if item.Index == 0: + continue + csvwriter.writerow([ + 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) + log.exception(e) + log.debug(currencyPair+': Generated 1min OHLCV data.') def generate_ohlcv(self, df): @@ -171,10 +268,10 @@ class PoloniexCurator(object): for currencyPair in self.currency_pairs: self.append_data_single_pair(currencyPair) # Rate limit is 6 calls per second, sleep 1sec/6 to be safe - time.sleep(0.17) + #time.sleep(0.17) ''' - Returns a data frame for all pairs, or for the requests currency pair. + Returns a data frame for all pairs, or for the requested currency pair. Makes sure data is up to date ''' def to_dataframe(self, start, end, currencyPair=None): @@ -193,5 +290,9 @@ class PoloniexCurator(object): if __name__ == '__main__': pc = PoloniexCurator() - pc.get_currency_pairs() - pc.append_data() + #pc.get_currency_pairs() + #pc.append_data() + + for coin in COINS: + # pc.retrieve_trade_history(coin,DT_START) + pc.write_ohlcv_file(coin) From 91e71c5e380c2d270b7428849d695185bd594237 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Wed, 20 Sep 2017 09:15:39 -0600 Subject: [PATCH 17/19] WIP: bundling 1min data --- catalyst/curate/poloniex.py | 113 ++++++++++++++++++++++++++---------- 1 file changed, 83 insertions(+), 30 deletions(-) diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index 33eb46c9..d63be3f9 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -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¤cyPair=' + 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) From 36c2564bb0769b122ea6de2c45ce0462830d02a1 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Wed, 20 Sep 2017 11:05:09 -0600 Subject: [PATCH 18/19] splitting plot styles: dark for live, default for backtesting --- catalyst/exchange/live_graph_clock.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/catalyst/exchange/live_graph_clock.py b/catalyst/exchange/live_graph_clock.py index 877f8115..60e7503b 100644 --- a/catalyst/exchange/live_graph_clock.py +++ b/catalyst/exchange/live_graph_clock.py @@ -24,8 +24,6 @@ from matplotlib import style log = Logger('LiveGraphClock') -style.use('dark_background') - fmt = mdates.DateFormatter('%Y-%m-%d %H:%M') @@ -63,6 +61,8 @@ class LiveGraphClock(object): self._before_trading_start_bar_yielded = True self.context = context + style.use('dark_background') + fig = plt.figure() fig.canvas.set_window_title('Enigma Catalyst: {}'.format( self.context.algo_namespace)) From a1bc174740bc629aeb43a984583cea09771e2615 Mon Sep 17 00:00:00 2001 From: Victor Grau Serrat Date: Wed, 20 Sep 2017 15:11:18 -0600 Subject: [PATCH 19/19] Wrapping up 1min data for Poloniex in backtesting --- catalyst/curate/poloniex.py | 218 +++++++++--------------------- catalyst/data/bundles/poloniex.py | 32 +++-- 2 files changed, 82 insertions(+), 168 deletions(-) diff --git a/catalyst/curate/poloniex.py b/catalyst/curate/poloniex.py index d63be3f9..911ac25f 100644 --- a/catalyst/curate/poloniex.py +++ b/catalyst/curate/poloniex.py @@ -6,19 +6,15 @@ import os, time, shutil, requests, logbook 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() log = logbook.Logger(__name__) class PoloniexCurator(object): - """ + ''' OHLCV data feed generator for crypto data. Based on Poloniex market data - """ + ''' _api_path = 'https://poloniex.com/public?' currency_pairs = [] @@ -31,6 +27,9 @@ 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): url = self._api_path + 'command=returnTicker' @@ -49,82 +48,29 @@ class PoloniexCurator(object): log.debug('Currency pairs retrieved successfully: %d' % (len(self.currency_pairs))) - def _get_start_date(self, csv_fn): - ''' Function returns latest appended date, if the file has been previously written - the last line is an empty one, so we have to read the second to last line - ''' - try: - with open(csv_fn, 'ab+') as f: - f.seek(0, os.SEEK_END) # First check file is not zero size - if(f.tell() > 2): - 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() - return int(lastrow.split(',')[0]) + 300 - - except Exception as e: - log.error('Error opening file: %s' % csv_fn) - log.exception(e) - - return DT_START - - 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 trade history data for %s' % currencyPair) - log.exception(e) - return None - - 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 + ''' + Helper function that reads tradeID and date fields from CSV readline + ''' 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 - + ''' + 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): 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. + ''' try: with open(csv_fn, 'ab+') as f: f.seek(0, os.SEEK_END) @@ -168,9 +114,22 @@ class PoloniexCurator(object): log.error('Failed to to retrieve trade history data for %s: %s' % (currencyPair,response.json()['error'])) exit(1) - if('first_tradeID' in locals() and response.json()[-1]['tradeID'] == first_tradeID): # Got to the end of TradingHistory 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): 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 + ''' try: if( 'end_file' in locals() and end_file + 3600 < end): if (temp is None): @@ -220,10 +179,34 @@ class PoloniexCurator(object): log.error('Error opening %s' % csv_fn) log.exception(e) - self.retrieve_trade_history(currencyPair, start, end) # If we get here, we aren't done. Repeat + ''' + 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 write_ohlcv_file(self, currencyPair): - + + ''' + 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 + return ohlcv + + + ''' + 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) ): @@ -253,82 +236,10 @@ class PoloniexCurator(object): log.exception(e) log.debug(currencyPair+': Generated 1min OHLCV data.') - 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 - ''' - def append_data_single_pair(self, currencyPair, repeat=0): - log.debug('Getting data for %s' % currencyPair) - csv_fn = CSV_OUT_FOLDER + 'crypto_prices-' + currencyPair + '.csv' - 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) - - 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 ohlcv.itertuples(): - if item.Index == 0: - continue - csvwriter.writerow([ - 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) - log.exception(e) - elif (repeat < CONN_RETRIES): - log.debug('Retrying: attemt %d' % (repeat+1) ) - self.append_data_single_pair(currencyPair, repeat + 1) ''' - Pulls latest data for all currency pairs + Returns a data frame for a given currencyPair from data on disk ''' - def append_data(self): - for currencyPair in self.currency_pairs: - self.append_data_single_pair(currencyPair) - # Rate limit is 6 calls per second, sleep 1sec/6 to be safe - #time.sleep(0.17) - - ''' - Returns a data frame for all pairs, or for the requested currency pair. - Makes sure data is up to date - ''' - def to_dataframe(self, start, end, currencyPair=None): - csv_fn = CSV_OUT_FOLDER + 'crypto_prices-' + currencyPair + '.csv' - 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'],unit='s') - df.set_index('date', inplace=True) - - 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']) @@ -336,16 +247,11 @@ class PoloniexCurator(object): df.set_index('date', inplace=True) return df[start : end] + if __name__ == '__main__': pc = PoloniexCurator() pc.get_currency_pairs() - #pc.append_data() - #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) diff --git a/catalyst/data/bundles/poloniex.py b/catalyst/data/bundles/poloniex.py index 9ad60d01..60bbac95 100644 --- a/catalyst/data/bundles/poloniex.py +++ b/catalyst/data/bundles/poloniex.py @@ -25,6 +25,8 @@ from catalyst.data.bundles.core import register_bundle from catalyst.data.bundles.base_pricing import BaseCryptoPricingBundle from catalyst.utils.memoize import lazyval +from catalyst.curate.poloniex import PoloniexCurator + class PoloniexBundle(BaseCryptoPricingBundle): @lazyval def name(self): @@ -38,7 +40,7 @@ class PoloniexBundle(BaseCryptoPricingBundle): def frequencies(self): return set(( 'daily', - #'5-minute', + 'minute', )) @lazyval @@ -94,17 +96,23 @@ class PoloniexBundle(BaseCryptoPricingBundle): start_date, end_date, frequency): - raw = pd.read_json( - self._format_data_url( - api_key, - symbol, - start_date, - end_date, - frequency, - ), - orient='records', - ) - raw.set_index('date', inplace=True) + + if(frequency == 'minute'): + pc = PoloniexCurator() + raw = pc.onemin_to_dataframe(symbol, start_date, end_date) + + else: + raw = pd.read_json( + self._format_data_url( + api_key, + symbol, + start_date, + end_date, + frequency, + ), + orient='records', + ) + raw.set_index('date', inplace=True) # BcolzDailyBarReader introduces a 1/1000 factor in the way pricing is stored # on disk, which we compensate here to get the right pricing amounts