From 5aee03212d5e09b884a4c9864e5ab0fc4a6bb4ff Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 6 Apr 2012 12:39:27 -0400 Subject: [PATCH 1/7] Replayable error log. --- zipline/monitor.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/zipline/monitor.py b/zipline/monitor.py index bf50505e..f6b55037 100644 --- a/zipline/monitor.py +++ b/zipline/monitor.py @@ -169,6 +169,8 @@ class Controller(object): self.pub_socket = pub_socket self.route_socket = route_socket + self.error_replay = {} + if logging: self.logging = logging else: @@ -242,6 +244,13 @@ class Controller(object): #self.logging.info("[Controller] Tracking : %s" % ([c for c in self.tracked],)) pass + def replay_errors(self): + """ + Replay the errors in the order they were reported to the + controller. + """ + return [ a for a in sorted(self.replay_errors.keys())] + # ------------- # Publications # ------------- @@ -377,6 +386,7 @@ class Controller(object): self.logging.info('[Controller] Component "%s" done.' % component) def exception(self, component, failure): + self.error_replay[time.time()] = failure self.logging.error('Component "%s" in exception state' % component) # ----------------- From 80bfbd5dcbb1263f96caeabf09b5fd638d15b461 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Fri, 6 Apr 2012 13:31:39 -0400 Subject: [PATCH 2/7] Performance reports take either socket address or socket. --- zipline/finance/performance.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index e3564c54..3b784413 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -185,11 +185,14 @@ class PerformanceTracker(): Publish the performance results asynchronously to a socket. """ - ctx = context or zmq.Context.instance() - sock = ctx.socket(zmq.PUSH) - sock.connect(zmq_socket) + if isinstance(zmq_socket, zmq.Socket): + self.result_stream = zmq_socket + else: + ctx = context or zmq.Context.instance() + sock = ctx.socket(zmq.PUSH) + sock.connect(zmq_socket) - self.result_stream = sock + self.result_stream = sock def to_dict(self): """ From 16ceb68c7c66c7f8dfcbdd8a52615ff72ff779c7 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Sat, 7 Apr 2012 09:58:50 -0400 Subject: [PATCH 3/7] Move zmq topology mapper to dev/ folder --- {zipline => dev}/topos.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename {zipline => dev}/topos.py (100%) diff --git a/zipline/topos.py b/dev/topos.py similarity index 100% rename from zipline/topos.py rename to dev/topos.py From 1dbe0975fcc78b4518aa3742528c6c764f281f65 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 9 Apr 2012 07:55:11 -0400 Subject: [PATCH 4/7] NumpyChannel --- zipline/zmq_utils.py | 60 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/zipline/zmq_utils.py b/zipline/zmq_utils.py index e32931dc..5e0a529f 100644 --- a/zipline/zmq_utils.py +++ b/zipline/zmq_utils.py @@ -3,6 +3,9 @@ Misc ZeroMQ utilities. """ import gevent import msgpack +import numpy +from numpy import dtype +from pandas import DataFrame from gevent_zeromq import zmq from contextlib import closing @@ -92,3 +95,60 @@ def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None): import pdb; pdb.set_trace() return gevent.spawn(console) + +class NumpyChannel(zmq.Socket): + + def recv_pandas(self, flags=0, copy=True, track=False): + + # Pandas Metadata + index, columns, dtype_name, shape = msgpack.loads(self.recv(flags=flags)) + + # Pandas ndarray + ndbuffer = self.recv(flags=flags, copy=copy, track=track) + buf = buffer(ndbuffer) + + ndarray = numpy.frombuffer(buf, dtype=dtype(dtype_name)).reshape(shape) + return DataFrame(data=ndarray, index=index, + columns=columns, dtype=dtype_name) + + def send_pandas(self, df, flags=0, copy=True, track=False): + + # Pandas Metadata + index = df.index.tolist() + columns = df.columns.tolist() + dtype_name = df.values.dtype.name + shape = df.values.shape + + # Pandas ndarray + ndarray = df.values + + metadata = msgpack.dumps((index, columns, dtype_name, shape)) + + self.send(metadata, flags|zmq.SNDMORE) + return self.send(ndarray, flags, copy=copy, track=track) + +if __name__ == '__main__': + + from numpy.random import randn + df = DataFrame(randn(5,5)) + + ctx = zmq.Context.instance() + + def send(): + pub = NumpyChannel(ctx, zmq.PUSH) + pub.bind('inproc://a') + + for i in xrange(100): + pub.send_pandas(df, copy=False) + + def recv(): + sub = NumpyChannel(ctx, zmq.PULL) + sub.connect('inproc://a') + + for i in xrange(100): + sub.recv_pandas(copy=False) + + gevent.joinall([ + gevent.spawn(send), + gevent.spawn(recv) + ]) From 04d389b7748c98e3736a67246100681d1beaf91b Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Mon, 9 Apr 2012 23:43:53 -0400 Subject: [PATCH 5/7] Remove old pandas-zmq protocol. --- zipline/serial.py | 37 +------------------------------------ 1 file changed, 1 insertion(+), 36 deletions(-) diff --git a/zipline/serial.py b/zipline/serial.py index b0eba920..64faa9fd 100644 --- a/zipline/serial.py +++ b/zipline/serial.py @@ -6,11 +6,9 @@ ZeroMQ. :) """ import zlib -#import blosc import hmac import base64 -import numpy -import pandas +#import blosc import cPickle as pickle @@ -33,39 +31,6 @@ def recv_zipped_pickle(socket, flags=0, protocol=-1): p = zlib.uncompress(z) return pickle.loads(p, protocol=protocol) -# HDF5, Numpy Byte Strings, Pandas arrays should use -# blosc and reconstruct the Python container from the byte string -# on the other side. - -def send_numpy(socket, obj, flags=0): - packed = blosc.pack_array(obj) - return socket.send(packed, flags=flags) - -def recv_numpy(socket, flags=0): - packed = blosc.unpack_array(socket.recv(flags)) - return socket.send(packed, flags=flags) - -def send_pandas(socket, obj, flags=0): - ndarray = obj._data.blocks[0].values - socket.send_multipart(ndarray, flags=flags) - spec = ( - obj._data.index, - obj._data.columns, - obj._data.blocks[0].dtype - ) - return socket.send_multipart(spec, flags) - -def recv_pandas(socket, flags=0): - ndarray = socket.recv_multipart(flags) - spec = socket.recv_multipart(flags) - return pandas.DataFrame._init_ndarray(ndarray, *spec) - -def send_hdf5(self): - pass - -def recv_hdf5(self): - pass - # Cryptographically secure wire protocol for ZeroMQ Using HMAC. # Compare byte strings, backported from Python 3. From 6f27009e82cea92e1b26ad5e33bfb4c18c960263 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 10 Apr 2012 00:30:53 -0400 Subject: [PATCH 6/7] Started datetime utiles lib. --- zipline/date_utils.py | 55 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 zipline/date_utils.py diff --git a/zipline/date_utils.py b/zipline/date_utils.py new file mode 100644 index 00000000..ddbe278b --- /dev/null +++ b/zipline/date_utils.py @@ -0,0 +1,55 @@ +from collections import namedtuple + +import pytz +import calendar +from dateutil import rrule +from datetime import datetime, date, timedelta +from dateutil.relativedelta import * + +# Datetime Tuple +d_tuple = namedtuple('dt', ['year', 'month', 'day', 'hour', 'minute', 'second', 'micros']) + +WEEKDAYS = [rrule.MO, rrule.TU, rrule.WE, rrule.TH, rrule.FR] + +HOLIDAYS = { + 'new_years' : datetime(2008 , 1 , 1 ), + 'mlk_day' : datetime(2008 , 1 , 21), + 'presidents' : datetime(2008 , 2 , 18), + 'good_friday' : datetime(2008 , 3 , 21), + 'memorial_day' : datetime(2008 , 5 , 26), + 'july_4th' : datetime(2008 , 7 , 4 ), + 'labor_day' : datetime(2008 , 9 , 1 ), + 'tgiving' : datetime(2008 , 11 , 27), + 'christmas' : datetime(2008 , 5 , 25), +} + +# Create a rule to recur every weekday starting today +rule = rrule.rrule( + rrule.DAILY, + byweekday=WEEKDAYS, +) + +# Precompute the rule, so that dates are cached. +rs = rrule.rruleset() +rs.rrule(rule) + +# Add holidays as exclusion days +for holiday in HOLIDAYS.itervalues(): + rs.exdate(holiday) + +def trading_days(after, before, inclusive=False): + """ + Iterates over the NYSE trading days between the two given + dates. + """ + return rs.between(after, before, inc=inclusive) + +if __name__ == '__main__': + + now = datetime.now() + now30 = datetime.now() + timedelta(days=30) + + # Iterate over the trading days between any two arbitrary + # days, excluding the preset holidays. + for day in trading_days(now, now30): + print day From ea340b61860a63e88af0acfb7761b61f8965b7dc Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Tue, 10 Apr 2012 08:30:37 -0400 Subject: [PATCH 7/7] Datetime subclasses with tzinfo=UTC by default --- zipline/date_utils.py | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/zipline/date_utils.py b/zipline/date_utils.py index ddbe278b..966525c5 100644 --- a/zipline/date_utils.py +++ b/zipline/date_utils.py @@ -1,5 +1,6 @@ from collections import namedtuple +import time import pytz import calendar from dateutil import rrule @@ -7,8 +8,23 @@ from datetime import datetime, date, timedelta from dateutil.relativedelta import * # Datetime Tuple +# -------------- d_tuple = namedtuple('dt', ['year', 'month', 'day', 'hour', 'minute', 'second', 'micros']) +# UTC Datetime Subclasses +# ----------------------- +def utcnow(): + return datetime.now(pytz.utc) + +class utcdatetime(datetime): + def __new__(cls, *args, **kwargs): + kwargs['tzinfo'] = pytz.utc + dt = datetime.__new__(cls, *args, **kwargs) + return dt + +# Datetime Calculations +# --------------------- + WEEKDAYS = [rrule.MO, rrule.TU, rrule.WE, rrule.TH, rrule.FR] HOLIDAYS = { @@ -27,6 +43,7 @@ HOLIDAYS = { rule = rrule.rrule( rrule.DAILY, byweekday=WEEKDAYS, + cache=True, ) # Precompute the rule, so that dates are cached. @@ -53,3 +70,10 @@ if __name__ == '__main__': # days, excluding the preset holidays. for day in trading_days(now, now30): print day + + # Its now cached so if we do that traversal again it only + # takes like 1e-5 seconds. + tic = time.time() + for day in trading_days(now, now30): + print day + print time.time() - tic