From a3682ec59fed3440c748170535ae09ec9c66c7c8 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:32:38 -0400 Subject: [PATCH 01/54] Removes unused TestClient class. --- tests/client.py | 81 ------------------------------------------------- 1 file changed, 81 deletions(-) delete mode 100644 tests/client.py diff --git a/tests/client.py b/tests/client.py deleted file mode 100644 index 3ceede6f..00000000 --- a/tests/client.py +++ /dev/null @@ -1,81 +0,0 @@ -import logging - -import zipline.protocol as zp -from zipline.core.component import Component -from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE - -LOGGER = logging.getLogger('ZiplineLogger') - -class TestClient(Component): - - def init(self): - self.received_count = 0 - self.prev_dt = None - - self.result_streams = [] - - # Maximum outgoing result streams, really shouldn't ever - # need more than 1. - self.max_outgoing = 5 - - @property - def get_id(self): - return "TEST_CLIENT" - - @property - def get_type(self): - return COMPONENT_TYPE.SINK - - def open(self): - self.data_feed = self.connect_result() - - def result_stream(self, zmq_socket, context=None): - """ - Asynchronously grab a socket to stream results out on. - """ - ctx = context or zmq.Context.instance() - sock = ctx.socket(zmq.PULL) - sock.bind(zmq_socket) - - # Add - self.result_streams.append( sock ) - - def do_work(self): - socks = dict(self.poll.poll(self.heartbeat_timeout)) - - if socks.get(self.control_in) == self.zmq.POLLIN: - msg = self.control_in.recv() - - if socks.get(self.data_feed) == self.zmq.POLLIN: - msg = self.data_feed.recv() - #logger.info('msg:' + str(msg)) - - if msg == str(CONTROL_PROTOCOL.DONE): - LOGGER.info("Client is DONE!") - self.signal_done() - return - - self.received_count += 1 - - try: - event = self.unframe(msg) - - # deserialization error - except zp.INVALID_MERGE_FRAME as exc: - return self.signal_exception(exc) - - if self.prev_dt != None: - if not event['dt'] >= self.prev_dt: - raise Exception( - "Message out of order: {date} after {prev}".format( - date = event['dt'], prev = self.prev_dt - ) - ) - else: - self.prev_dt = event.dt - - if self.received_count % 100 == 0: - LOGGER.info("received {n} messages".format(n=self.received_count)) - - def unframe(self, msg): - return zp.MERGE_UNFRAME(msg) From f8fd508dfc5ef9e89816b71bb39bcfda32e910a9 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:38:14 -0400 Subject: [PATCH 02/54] Removes custom Exception that isn't found in code base. --- zipline/exceptions.py | 5 ----- 1 file changed, 5 deletions(-) delete mode 100644 zipline/exceptions.py diff --git a/zipline/exceptions.py b/zipline/exceptions.py deleted file mode 100644 index 0c05a67e..00000000 --- a/zipline/exceptions.py +++ /dev/null @@ -1,5 +0,0 @@ -from utils.exception_utils import CustomException - -class ComponentNoInit(CustomException): - argmap = ('classname',) - message = """Class {classname} does not define an init method.""" From eb6c83a7c3bc5b18d5728293e25069ba656c3694 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:40:19 -0400 Subject: [PATCH 03/54] Removes vestigial code for zmq based generators. --- zipline/gens/zmq_gens.py | 18 ------------------ zipline/gens/zmqgen.py | 19 ------------------- 2 files changed, 37 deletions(-) delete mode 100644 zipline/gens/zmq_gens.py delete mode 100644 zipline/gens/zmqgen.py diff --git a/zipline/gens/zmq_gens.py b/zipline/gens/zmq_gens.py deleted file mode 100644 index e60dae2b..00000000 --- a/zipline/gens/zmq_gens.py +++ /dev/null @@ -1,18 +0,0 @@ -import zmq - -import zipline.protocol as zp - -def gen_from_zmq(poller, unframe, namestring): - """ - A generator that takes an initialized zmq poller and yields - messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE. - """ - while True: - message = poller.recv() - # Done protocol should now be a message type so that - # done messages can also have source_ids. - if message.type == zp.CONTROL_PROTOCOL.DONE: - yield done_message(message.source_id) - break - else: - yield unframe(message) diff --git a/zipline/gens/zmqgen.py b/zipline/gens/zmqgen.py deleted file mode 100644 index 66dbdca3..00000000 --- a/zipline/gens/zmqgen.py +++ /dev/null @@ -1,19 +0,0 @@ -import zmq -import zipline.protocol as zp - -def gen_from_pull_socket(socket_uri, context, unframe): - """ - A generator that takes a socket_uri, and yields - messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE. - """ - pull_socket = context.socket(zmq.PULL) - pull_socket.connect(socket_uri) - poller = zmq.Poller() - poller.register(pull_socket, zmq.POLLIN) - - return gen_from_poller(poller, pull_socket, unframe) - - -# this generator needs to know about the source_ids coming in via -# the poller, and need to yield DONE messages for each -# source_id. From c09c786bcdd496f3bc21d0f179a725674ec10e95 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:41:07 -0400 Subject: [PATCH 04/54] Removes unused profiler for multiprocessor architecture. --- zipline/profile/prof_yappi.py | 93 ----------------------------------- 1 file changed, 93 deletions(-) delete mode 100644 zipline/profile/prof_yappi.py diff --git a/zipline/profile/prof_yappi.py b/zipline/profile/prof_yappi.py deleted file mode 100644 index 58878fe8..00000000 --- a/zipline/profile/prof_yappi.py +++ /dev/null @@ -1,93 +0,0 @@ -from __future__ import division - -import logging -from zipline.core.devsimulator import AddressAllocator -import zipline.finance -from zipline.optimize.factory import create_predictable_zipline -import pandas as pd -import numpy as np -import os.path - -def convert_ystats(ystats): - """Convert yappi.get_stats().func_stats object to pandas - DataFrame. - - """ - func_names = [os.path.split(item[0])[-1] for item in ystats] - ncall = [float(item[1]) for item in ystats] - ttot = [float(item[2]) for item in ystats] - tsub = [float(item[3]) for item in ystats] - tavg = [float(item[4]) for item in ystats] - stats = pd.DataFrame({'ncall': ncall, 'ttot': ttot, 'tsub': tsub, 'tavg': tavg}, index=func_names) - - return stats - - -allocator = AddressAllocator(1000) - -config = { 'allocator' :allocator, - 'sid' :133, - 'trade_count' :5000, - 'amplitude' :30, - 'base_price' :50 - } - -LOGGER = logging.getLogger('ZiplineLogger') - -import yappi - -def gen_single_stats(func, *args, **kwargs): - """Profile func(*args, **kwargs) with yappi. - - Returns DataFrame of statistics. - """ - yappi.start() - func(*args, **kwargs) - yappi.stop() - return convert_ystats(yappi.get_stats().func_stats) - -def gen_avg_stats(func, runs=1, *args, **kwargs): - """Profile func(*args, **kwargs) with yappi. Runs multiple times at computes the average. - - Returns DataFrame of average statistics. - """ - - avg_stats = pd.concat([gen_single_stats() for i in range(runs)], keys=range(runs)) - grouped = avg_stats.groupby(level=1) - - return grouped.aggregate(np.mean) - -def run_updown(fname='before_stats.csv'): - """Profile a zipline with the UpDown tradesource (does not require - DB access) and the buy/sell algorithm (requires no - computation). - - Saves output statics under fname. - - Returns Dataframe of statistics. - """ - zp, _ = create_predictable_zipline(config, simulate=False) - stats = gen_single_stats(zp.simulate, blocking=True) - stats.to_csv(fname) - - return stats - -def calc_speedup(before='before_stats.csv', after='after_stats.csv'): - """Calculate speed-up between two previously run and saved - statistics under filename before and after. - - Prints DataFrame of top 30 speed-ups and top 30 slow-downs. - - """ - old = pd.DataFrame.from_csv(before) - new = pd.DataFrame.from_csv(after) - speed_up = old / new - speed_up = speed_up.fillna(1) - speed_up = speed_up.sort(column='ttot', ascending=False) - slow_down = speed_up.sort(column='ttot', ascending=True) - print speed_up[:30] - print slow_down[:30] - -if __name__ == '__main__': - run_updown() - yappi.print_stats(sort_type=yappi.SORTTYPE_TTOT) \ No newline at end of file From 125eb36d427be49253a0a4f4da1dd98585f36f63 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:42:00 -0400 Subject: [PATCH 05/54] Removes unused zmq_utils now that zipline is less ZeroMQ heavy. --- zipline/utils/zmq_utils.py | 154 ------------------------------------- 1 file changed, 154 deletions(-) delete mode 100644 zipline/utils/zmq_utils.py diff --git a/zipline/utils/zmq_utils.py b/zipline/utils/zmq_utils.py deleted file mode 100644 index 49177ee4..00000000 --- a/zipline/utils/zmq_utils.py +++ /dev/null @@ -1,154 +0,0 @@ -""" -Misc ZeroMQ experimental tools -""" -import gevent -import msgpack -import numpy -from numpy import dtype -from pandas import DataFrame -from gevent_zeromq import zmq - -from contextlib import closing - -class ZmqDone(Exception): - - def __init__(self, socket, frame): - self.ident = socket.identity - self.frame = str(frame) - - def __str__(self): - return 'Socket ( %s ) finished with frame ( %s )' % \ - ( self.ident, self.frame ) - -class zs(object): - """ - A wrapper for the *very* common pattern of reading from a - upstream socket until you get a DONE or EXCEPTION frame. - - # Eliminates all the boilerplate serialization logic - # and error handling cases into 3 lines. - - halts = (ERROR_FRAME, CLOSE_FRAME) - stream = zs(socket, halts) - - stream.on_error(YouFailAtFailing) - - for msg in stream: - print msg - - """ - - def __init__(self, socket, halts, srl=msgpack): - self._socket = socket - self.exc_case = halts[0] - self.done_case = halts[1] - - self.loads = srl.loads - self.halt_method = 'exception' - self.exception = ZmqDone - self.function = None - - def __iter__(self): - self.last = msg = self.loads(self._socket.recv()) - - if msg == self.exc_case: - return self.halt() - - if msg == self.done_case: - raise StopIteration - - yield msg - - def last(self): - return self.last - - def halt(self): - if self.halt_method == 'exception': - raise self.exception - elif self.halt_method == 'function': - return self.function() - - def on_error(self, callee): - - if isinstance(callee, Exception): - self.halt_method = 'exception' - self.exception = callee - else: - self.halt_method = 'function' - self.function = callee - -def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None): - """ - A utility to drop into a ZeroMQ pdb console and inspect - messages as they come through. If you just want to pipe to - stdout, don't use this. - """ - - context = context or zmq.Context.instance() - socket = context.socket(zmq.PULL) - socket.bind(socket_addr) - - def console(): - while True: - msg = socket.recv_pyobj() - print msg - 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 6a6f874ed03ebc70f9c695be3b44af80cd5f27af Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:42:59 -0400 Subject: [PATCH 06/54] Removes unused toys directory. --- zipline/toys/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 zipline/toys/__init__.py diff --git a/zipline/toys/__init__.py b/zipline/toys/__init__.py deleted file mode 100644 index e69de29b..00000000 From c20f4062ce1eae904b0937eff7ee018bf4a8c7e1 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:51:39 -0400 Subject: [PATCH 07/54] Removes unused module. Looks like this was used for keeping state in the 0MQ-heavy world. --- zipline/transitions.py | 69 ------------------------------------------ 1 file changed, 69 deletions(-) delete mode 100644 zipline/transitions.py diff --git a/zipline/transitions.py b/zipline/transitions.py deleted file mode 100644 index e47d54c8..00000000 --- a/zipline/transitions.py +++ /dev/null @@ -1,69 +0,0 @@ -import types -from collections import Container, Hashable, Callable - -class Any(object): pass - -class Workflow(Container, Callable): - - def __init__(self, states, transitions, initial_state): - self.simple = set() - self.complx = [] - - if isinstance(states[0], tuple): - self.groups = {b for _,b in states} - else: - self.groups = set() - - matcher = lambda b: lambda f,t : t == b - - for (a, b) in transitions.itervalues(): - if a is Any: - self.complx.append(matcher(b)) - if isinstance(a, Hashable) and isinstance(b, Hashable): - self.simple.add((a,b)) - - def __call__(self, **kwargs): - if 'group' in kwargs: - return self.groups - - def __contains__(self, state): - if state in self.simple: - return True - for match in self.complx: - if match(*state): - return True - else: - return False - -class WorkflowMeta(type): - """ - Base metaclass component workflows. - """ - - def __new__(cls, name, mro, attrs): - base = 'Component' - - state = attrs.get('states', None) - transitions = attrs.get('transitions', None) - initial_state = attrs.get('initial_state', None) - - if not 'abstract' in attrs: - - if attrs.get('workflow'): - raise RuntimeError('`workflow` is a reserved attribute.') - - if not state: - raise RuntimeError('Must specify states') - - if not transitions: - raise RuntimeError('Must specify transitions') - - if not transitions: - raise RuntimeError('Must specify initial_state') - - new_class = super(WorkflowMeta, cls).__new__(cls, name, mro, attrs) - - if not 'abstract' in attrs: - new_class.workflow = Workflow(state, transitions, initial_state) - - return new_class From b96122f3557b7590b5ff196b96d03f287ea94541 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 13:57:09 -0400 Subject: [PATCH 08/54] Removes unused wiring for speedups. Separating this deletion into it's own patch, so that it can be used as a base to add C extensions back in, if we end up going that route. --- pavement.py | 73 +----------------------------------- zipline/speedups/__init__.py | 0 zipline/speedups/example.pyx | 3 -- 3 files changed, 1 insertion(+), 75 deletions(-) delete mode 100644 zipline/speedups/__init__.py delete mode 100644 zipline/speedups/example.pyx diff --git a/pavement.py b/pavement.py index d4b4bd7a..5353c8a3 100644 --- a/pavement.py +++ b/pavement.py @@ -1,36 +1,14 @@ import os import re import sys -import glob import time -from distutils.dep_util import newer -#from distutils.extension import Extension -from setuptools.extension import Extension - -from paver.easy import options, Bunch, task, needs, path, info +from paver.easy import options, Bunch, task, path from paver.setuputils import install_distutils_tasks, \ find_packages, find_package_data from subprocess import call install_distutils_tasks() -# ========= -# Compilers -# ========= - -try: - from Cython.Compiler.Main import compile - from Cython.Distutils import build_ext - have_cython = True -except ImportError: - have_cython = False - -try: - import numpy as np - have_numpy = True -except: - have_numpy = False - # =================== # Release Information # =================== @@ -69,11 +47,6 @@ def parse_requirements(file_name): requirements.append(line) return requirements -example = Extension( - "zipline/speedups/example", ["zipline/speedups/example.pyx"], - #include_dirs=[np.get_include()], -) - # ============ # Dependencies # ============ @@ -88,11 +61,6 @@ tests_require = install_requires + parse_requirements('./etc/requirements_dev.tx # seutp.py # ======== -if have_numpy and have_cython: - cext = [example] -else: - cext = [] - options( sphinx = Bunch( builddir="_build", @@ -127,10 +95,6 @@ options( 'Topic :: Scientific/Engineering :: Information Analysis', 'Topic :: System :: Distributed Computing', ], - ext_modules = cext, - cmdclass = { - 'build_ext': build_ext - }, entry_points = { 'console_scripts': [ 'zipline = zipline.core.interpreter:main', @@ -139,41 +103,6 @@ options( ), ) -# ============ -# C Extensions -# ============ - -@task -def clean_inplace(): - """ - Remove shared objects and C files from the extension - directory. - """ - for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.c')): - p = path(fn) - p.remove() - - for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.so')): - p = path(fn) - p.remove() - -@task -def build_cython(): - for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.pyx')): - p = path(fn) - - modname = p.splitext()[0].basename() - dest = p.splitext()[0] + '.c' - - if newer(p.abspath(), dest.abspath()): - info('cython %s -o %s'%(p, dest.basename())) - compile(p.abspath(), full_module_name=modname) - -@task -@needs(['build_cython', 'setuptools.command.build_ext']) -def build_ext(): - pass - # ====== # Tasks # ====== diff --git a/zipline/speedups/__init__.py b/zipline/speedups/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/zipline/speedups/example.pyx b/zipline/speedups/example.pyx deleted file mode 100644 index befd4ae7..00000000 --- a/zipline/speedups/example.pyx +++ /dev/null @@ -1,3 +0,0 @@ -from libc.stdio cimport printf - -printf("Hello World!") From 27e83ba1c68d011af396d617522afc93120df013 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 15:35:29 -0400 Subject: [PATCH 09/54] Removes module for serializing over 0MQ. zipline no longer serializes internally over 0ZM, so this is unneeded. --- zipline/utils/serial.py | 56 ----------------------------------------- 1 file changed, 56 deletions(-) delete mode 100644 zipline/utils/serial.py diff --git a/zipline/utils/serial.py b/zipline/utils/serial.py deleted file mode 100644 index 64faa9fd..00000000 --- a/zipline/utils/serial.py +++ /dev/null @@ -1,56 +0,0 @@ -""" -Format serializer for Zipline. - -Because I'm opinionated about how you should send things over -ZeroMQ. :) -""" - -import zlib -import hmac -import base64 -#import blosc - -import cPickle as pickle - -# Pickle does the equivelant of builtin ``eval``. Be afraid, be -# very afraid. - -def send_zipped_pickle(socket, obj, flags=0, protocol=-1): - """ - Pickle an object, and zip the pickle before sending it. - """ - p = pickle.dumps(obj, protocol) - z = zlib.compress(p) - return socket.send(z, flags=flags) - -def recv_zipped_pickle(socket, flags=0, protocol=-1): - """ - Unpickle and uncompress a received object. - """ - z = socket.recv(flags) - p = zlib.uncompress(z) - return pickle.loads(p, protocol=protocol) - -# Cryptographically secure wire protocol for ZeroMQ Using HMAC. - -# Compare byte strings, backported from Python 3. -def byte_eq(a, b): - return not sum(0 if x==y else 1 for x, y in zip(a, b)) and len(a) == len(b) - -def send_secure(socket, data, key, flags=0): - msg = base64.b64encode(data) - sig = base64.b64encode(hmac.new(key, msg).digest()) - return socket.send(bytes('!') + sig + bytes('?') + msg, flags=flags) - -def recv_secure(socket, data, key, flags): - data = socket.recv(flags=flags) - - try: - sig, msg = data.split(bytes('?'), 1) - except ValueError: - raise Exception('Invalid signature/message pair.') - - if byte_eq(sig[1:], base64.b64encode(hmac.new(key, msg).digest())): - return base64.b64decode(msg) - else: - raise Exception('Cryptographically invalid message received') From 1059f711deb7ff66b94f8eb9e04c44184487c38f Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 22 Aug 2012 16:21:36 -0400 Subject: [PATCH 10/54] Removes unused custom Exception handler. --- zipline/utils/exception_utils.py | 16 ---------------- 1 file changed, 16 deletions(-) delete mode 100644 zipline/utils/exception_utils.py diff --git a/zipline/utils/exception_utils.py b/zipline/utils/exception_utils.py deleted file mode 100644 index 67dc1ee9..00000000 --- a/zipline/utils/exception_utils.py +++ /dev/null @@ -1,16 +0,0 @@ -from textwrap import dedent - -class CustomException(Exception): - argmap = {0: 'classname'} - - def __init__(self, *args): - self.args = args - - def format(self): - assert len(self.args) == len(self.argmap), \ - """Wrong number of arguments passed to custom exception %s.""" \ - % self.__class__ - return self.message.format(**dict(zip(self.argmap, self.args))) - - def __str__(self): - return dedent(self.format()).strip('\n') From 4337a429ff0a14b2ecd2f1541c8467fadaefb388 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 14:48:05 -0400 Subject: [PATCH 11/54] Applies PEP-8 recommendations, mostly whitespace cleanup. Also, changes has_key to use `in` --- zipline/lines.py | 42 +++++++++++++++++++++++------------------- 1 file changed, 23 insertions(+), 19 deletions(-) diff --git a/zipline/lines.py b/zipline/lines.py index d340df5a..ba973943 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -87,6 +87,7 @@ class CancelSignal(Exception): def __init__(self): pass + class SimulatedTrading(object): def __init__(self, @@ -102,7 +103,8 @@ class SimulatedTrading(object): self.date_sorted = date_sorted_sources(*sources) self.transforms = transforms # Formerly merged_transforms. - self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms) + self.with_tnfms = sequential_transforms(self.date_sorted, + *self.transforms) self.trading_client = tsc(algorithm, environment, style) self.gen = self.trading_client.simulate(self.with_tnfms) self.results_uri = results_socket_uri @@ -149,7 +151,7 @@ class SimulatedTrading(object): "Results socket must exist to stream results" try: for event in self.gen: - if event.has_key('daily_perf'): + if 'daily_perf' in event: msg = zp.PERF_FRAME(event) else: msg = zp.RISK_FRAME(event) @@ -180,7 +182,7 @@ class SimulatedTrading(object): else: log.warning("Sending SIGINT") os.kill(ppid, SIGINT) - + def handle_exception(self, exc): if isinstance(exc, CancelSignal): # signal from monitor of an orderly shutdown, @@ -207,7 +209,7 @@ class SimulatedTrading(object): exc_type.__name__, exc_value.message ) - + self.results_socket.send(msg) except: log.exception("Exception while reporting simulation exception.") @@ -228,8 +230,8 @@ class SimulatedTrading(object): # bubbled. Since we do not want user logs in our system # logs, we set bubble to False. self.zmq_out = ZeroMQLogHandler( - socket = self.results_socket, - filter = lambda r, h: r.channel in ['Print', 'AlgoLog'], + socket=self.results_socket, + filter=lambda r, h: r.channel in ['Print', 'AlgoLog'], bubble=False ) @@ -263,7 +265,8 @@ class SimulatedTrading(object): is the source, with daily frequency in trades. - simulation_style: optional parameter that configures the :py:class:`zipline.finance.trading.TransactionSimulator`. Expects - a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading` + a SIMULATION_STYLE as defined in + :py:mod:`zipline.finance.trading` - transforms: optional parameter that provides a list of StatefulTransform objects. """ @@ -278,22 +281,22 @@ class SimulatedTrading(object): #-------------------- # Trading Environment #-------------------- - if config.has_key('environment'): + if 'environment' in config: trading_environment = config['environment'] else: trading_environment = factory.create_trading_environment() - if config.has_key('order_count'): + if 'order_count' in config: order_count = config['order_count'] else: order_count = 100 - if config.has_key('order_amount'): + if 'order_amount' in config: order_amount = config['order_amount'] else: order_amount = 100 - if config.has_key('trade_count'): + if 'trade_count' in config: trade_count = config['trade_count'] else: # to ensure all orders are filled, we provide one more @@ -304,14 +307,14 @@ class SimulatedTrading(object): if not simulation_style: simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE - zmq_context = config.get('zmq_context', None) - simulation_id = config.get('simulation_id', 'test_simulation') - results_socket_uri = config.get('results_socket_uri', None) + zmq_context = config.get('zmq_context', None) + simulation_id = config.get('simulation_id', 'test_simulation') + results_socket_uri = config.get('results_socket_uri', None) #------------------- # Trade Source #------------------- - if config.has_key('trade_source'): + if 'trade_source' in config: trade_source = config['trade_source'] else: trade_source = factory.create_daily_trade_source( @@ -321,7 +324,6 @@ class SimulatedTrading(object): concurrent=concurrent_trades ) - #------------------- # Transforms #------------------- @@ -330,7 +332,7 @@ class SimulatedTrading(object): #------------------- # Create the Algo #------------------- - if config.has_key('algorithm'): + if 'algorithm' in config: test_algo = config['algorithm'] else: test_algo = TestAlgorithm( @@ -356,6 +358,7 @@ class SimulatedTrading(object): return sim + class SimulatedTradingLite(object): """ SimulatedTrading without multiprocess and without zmq. @@ -372,9 +375,10 @@ class SimulatedTradingLite(object): self.date_sorted = date_sorted_sources(*sources) self.transforms = transforms # Formerly merged_transforms. - self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms) + self.with_tnfms = sequential_transforms(self.date_sorted, + *self.transforms) self.trading_client = tsc(algorithm, environment, style) self.gen = self.trading_client.simulate(self.with_tnfms) - + def get_results(self): return self.gen From 2201678375d17314600a4a01f92e6eb42cd3d189 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 15:47:53 -0400 Subject: [PATCH 12/54] Adds docstring for parameters of SimulatedTradingLite --- zipline/lines.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/zipline/lines.py b/zipline/lines.py index ba973943..5283756f 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -371,7 +371,25 @@ class SimulatedTradingLite(object): algorithm, environment, style): + """ + @sources - an iterable of iterables + These iterables must yield ndicts that contain: + - type :: a ziplines.protocol.DATASOURCE_TYPE + - dt :: a milliseconds since epoch timestamp in UTC + @transforms - An iterable of instances of StatefulTransform. + + @algorithm - An object that implements: + `def initialize(self)` + `def handle_data(self, data)` + `def get_sid_filter(self)` + `def set_logger(self, logger)` + `def set_order(self, order_callable)` + + @environment - An instance of finance.trading.TradingEnvironment + + @style - protocol.SIMULATION_STYLE + """ self.date_sorted = date_sorted_sources(*sources) self.transforms = transforms # Formerly merged_transforms. From f4f20dd2d13744b76811886ab7d61c955fc7eb47 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 15:51:37 -0400 Subject: [PATCH 13/54] Removes profiling tools that we currently don't run regularly. --- vb_suite/run_suite.py | 12 --- vb_suite/suite.py | 111 ---------------------------- vb_suite/zipline_bench_functions.py | 43 ----------- vb_suite/ziplines.py | 27 ------- 4 files changed, 193 deletions(-) delete mode 100644 vb_suite/run_suite.py delete mode 100644 vb_suite/suite.py delete mode 100644 vb_suite/zipline_bench_functions.py delete mode 100644 vb_suite/ziplines.py diff --git a/vb_suite/run_suite.py b/vb_suite/run_suite.py deleted file mode 100644 index c9b397e1..00000000 --- a/vb_suite/run_suite.py +++ /dev/null @@ -1,12 +0,0 @@ -from vbench.api import BenchmarkRunner -from suite import * - -def run_process(): - runner = BenchmarkRunner(benchmarks, REPO_PATH, REPO_URL, - BUILD, DB_PATH, TMP_DIR, PREPARE, - run_option='all', start_date=START_DATE, - module_dependencies=dependencies) - runner.run() - -if __name__ == '__main__': - run_process() diff --git a/vb_suite/suite.py b/vb_suite/suite.py deleted file mode 100644 index ed6c016d..00000000 --- a/vb_suite/suite.py +++ /dev/null @@ -1,111 +0,0 @@ -from vbench.api import Benchmark, GitRepo -from datetime import datetime - -import os - -modules = ['ziplines'] - -by_module = {} -benchmarks = [] - -for modname in modules: - ref = __import__(modname) - by_module[modname] = [v for v in ref.__dict__.values() - if isinstance(v, Benchmark)] - benchmarks.extend(by_module[modname]) - -for bm in benchmarks: - assert(bm.name is not None) - -import getpass -import sys - -USERNAME = getpass.getuser() - -if sys.platform == 'darwin': - HOME = '/Users/%s' % USERNAME -else: - HOME = '/home/%s' % USERNAME - -REPO_PATH = os.path.join(HOME, 'projects/qexec/zipline_repo') -REPO_URL = 'git@github.com:quantopian/zipline.git' -DB_PATH = os.path.join(REPO_PATH, 'vb_suite/benchmarks.db') -TMP_DIR = os.path.join(HOME, 'tmp/vb_zipline') - -PREPARE = """ -""" -BUILD = """ -""" -dependencies = ['zipline_bench_functions.py'] - -START_DATE = datetime(2011, 6, 1) - -repo = GitRepo(REPO_PATH) - -RST_BASE = 'source' - -# HACK! - -#timespan = [datetime(2011, 1, 1), datetime(2012, 1, 1)] - -def generate_rst_files(benchmarks): - import matplotlib as mpl - mpl.use('Agg') - import matplotlib.pyplot as plt - - vb_path = os.path.join(RST_BASE, 'vbench') - fig_base_path = os.path.join(vb_path, 'figures') - - if not os.path.exists(vb_path): - print 'creating %s' % vb_path - os.makedirs(vb_path) - - if not os.path.exists(fig_base_path): - print 'creating %s' % fig_base_path - os.makedirs(fig_base_path) - - for bmk in benchmarks: - print 'Generating rst file for %s' % bmk.name - rst_path = os.path.join(RST_BASE, 'vbench/%s.txt' % bmk.name) - - fig_full_path = os.path.join(fig_base_path, '%s.png' % bmk.name) - - # make the figure - plt.figure(figsize=(10, 6)) - ax = plt.gca() - bmk.plot(DB_PATH, ax=ax) - - start, end = ax.get_xlim() - - plt.xlim([start - 30, end + 30]) - plt.savefig(fig_full_path, bbox_inches='tight') - plt.close('all') - - fig_rel_path = 'vbench/figures/%s.png' % bmk.name - rst_text = bmk.to_rst(image_path=fig_rel_path) - with open(rst_path, 'w') as f: - f.write(rst_text) - - with open(os.path.join(RST_BASE, 'index.rst'), 'w') as f: - print >> f, """ -Performance Benchmarks -====================== - -These historical benchmark graphs were produced with `vbench -`__. - -.. toctree:: - :hidden: - :maxdepth: 3 -""" - for modname, mod_bmks in sorted(by_module.items()): - print >> f, ' vb_%s' % modname - modpath = os.path.join(RST_BASE, 'vb_%s.rst' % modname) - with open(modpath, 'w') as mh: - header = '%s\n%s\n\n' % (modname, '=' * len(modname)) - print >> mh, header - - for bmk in mod_bmks: - print >> mh, bmk.name - print >> mh, '-' * len(bmk.name) - print >> mh, '.. include:: vbench/%s.txt\n' % bmk.name diff --git a/vb_suite/zipline_bench_functions.py b/vb_suite/zipline_bench_functions.py deleted file mode 100644 index c970f055..00000000 --- a/vb_suite/zipline_bench_functions.py +++ /dev/null @@ -1,43 +0,0 @@ -try: - from zipline.simulator import AddressAllocator - pass -except Exception, e: - from zipline.core.devsimulator import AddressAllocator - -from zipline.lines import SimulatedTrading - -allocator = AddressAllocator(1001) - - -def get_zipline(): - zipline_test_config = { - 'allocator':allocator, - 'sid':133 - } - - zipline = SimulatedTrading.create_test_zipline( - **zipline_test_config - ) - - return zipline - -def run_basic_zipline(): - zipline = get_zipline() - zipline.simulate(blocking=True) - -def load_ndict(): - from zipline import ndict - nd = ndict({}) - keyname = 'a %i' - for i in xrange(1000000): - nd[keyname % i] = i - - for i in xrange(1000000): - nd[keyname % i] - -def mass_create_ndict(): - from zipline import ndict - data = dict(('a %d' % a,a) for a in xrange(1000)) - - for i in xrange(10000): - ndict(data) diff --git a/vb_suite/ziplines.py b/vb_suite/ziplines.py deleted file mode 100644 index 3874de70..00000000 --- a/vb_suite/ziplines.py +++ /dev/null @@ -1,27 +0,0 @@ -from vbench.api import Benchmark -from datetime import datetime - -setup = """ -from zipline_bench_functions import * -""" - -basic_zipline = Benchmark( - 'run_basic_zipline()', - setup=setup, - start_date=datetime(2012,5,15), - name='basic_zipline_test' -) - -load_ndict = Benchmark( - 'load_ndict()', - setup=setup, - start_date=datetime(2012,5,15), - name='load_ndict' -) - -mass_create_ndict = Benchmark( - 'mass_create_ndict()', - setup=setup, - start_date=datetime(2012,5,1), - name='create_ndict' -) From 49cac9ae9744a3d4015064fb84c53423a9fdac12 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 16:00:53 -0400 Subject: [PATCH 14/54] Removes ipython notebook that hasn't been touched in a while. --- notebooks/Experimenting with Frames.ipynb | 352 ---------------------- 1 file changed, 352 deletions(-) delete mode 100644 notebooks/Experimenting with Frames.ipynb diff --git a/notebooks/Experimenting with Frames.ipynb b/notebooks/Experimenting with Frames.ipynb deleted file mode 100644 index 50e5b5a5..00000000 --- a/notebooks/Experimenting with Frames.ipynb +++ /dev/null @@ -1,352 +0,0 @@ -{ - "metadata": { - "name": "Experimenting with Frames" - }, - "nbformat": 3, - "worksheets": [ - { - "cells": [ - { - "cell_type": "heading", - "source": [ - "Performance Tracking" - ] - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "import datetime", - "import pandas", - "import pytz", - "", - "import zipline.test.factory as factory", - "import zipline.finance.performance as perf", - "import zipline.protocol as zp", - "import zipline.finance.risk as risk", - "import zipline.finance.trading as trading" - ], - "language": "python", - "outputs": [], - "prompt_number": 38 - }, - { - "cell_type": "heading", - "source": [ - "Create a simulated trade history using the test factory" - ] - }, - { - "cell_type": "markdown", - "source": [ - "For any backtesting, zipline relies on a TradingEnvironment object. Trading environment holds essential facts: ", - " ", - " - start and end times for the simulation.", - " - historical daily returns for your benchmark.", - " - historical treasury curves", - " - an assumed capital base for your portfolio", - " - a calendar of trading days based on your benchmark", - "", - "zipline ships with a compressed archives of the S&P daily returns, and US treasury curves to facilitate standalone development and testing. In the next cell we instantiate the environment using these defaults. You can see more of this in zipline/test/test_perf_tracking.py" - ] - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "benchmark_returns, treasury_curves = factory.load_market_data()", - " ", - "trading_environment = risk.TradingEnvironment(benchmark_returns, treasury_curves)" - ], - "language": "python", - "outputs": [], - "prompt_number": 39 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "trade_count = 100", - "sid = 133", - "price = 10.1 ", - "price_list = [price] * trade_count", - "volume = [100] * trade_count", - "start_date = datetime.datetime.strptime(\"01/01/2011\",\"%m/%d/%Y\")", - "start_date = start_date.replace(tzinfo=pytz.utc)", - "trade_time_increment = datetime.timedelta(days=1)", - "", - "trade_history = factory.create_trade_history( ", - " sid, ", - " price_list, ", - " volume, ", - " start_date, ", - " trade_time_increment, ", - " trading_environment ", - ")", - "", - "sid2 = 134", - "price2 = 12.12", - "price2_list = [price2] * trade_count ", - "trade_history2 = factory.create_trade_history( ", - " sid2, ", - " price2_list, ", - " volume, ", - " start_date, ", - " trade_time_increment, ", - " trading_environment ", - ")", - " ", - "trade_history.extend(trade_history2) ", - "trade_history = sorted(trade_history, key=lambda x: x.dt)" - ], - "language": "python", - "outputs": [], - "prompt_number": 40 - }, - { - "cell_type": "markdown", - "source": [ - "Now that we have a simulated history of trades for two companies and a corresponding trading environment, we can create a dataframe of trades." - ] - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df = pandas.DataFrame(index = ['price', 'volume', 'dt'])", - "for event in trade_history:", - " series = event.as_series()", - " #df.index = df.index.tolist().append(event.sid)", - " #series.name = event.sid", - " df[event.sid] = series" - ], - "language": "python", - "outputs": [], - "prompt_number": 92 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 93, - "text": [ - " 133 134", - "price 10.1 12.12", - "volume 100 100", - "dt 2011-04-08 00:00:00+00:00 2011-04-08 00:00:00+00:00" - ] - } - ], - "prompt_number": 93 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t = df.transpose()", - "df_t" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 94, - "text": [ - " price volume dt", - "133 10.1 100 2011-04-08 00:00:00+00:00", - "134 12.12 100 2011-04-08 00:00:00+00:00" - ] - } - ], - "prompt_number": 94 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df[133]" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 56, - "text": [ - "sid 133", - "volume 100", - "dt 2011-04-08 00:00:00+00:00", - "price 10.1", - "changed NaN", - "Name: 133" - ] - } - ], - "prompt_number": 56 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t['price']" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 57, - "text": [ - "133 10.1", - "134 12.12", - "Name: price" - ] - } - ], - "prompt_number": 57 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t['price'].max()" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 50, - "text": [ - "12.12" - ] - } - ], - "prompt_number": 50 - }, - { - "cell_type": "code", - "collapsed": true, - "input": [ - "last = trade_history[23].dt" - ], - "language": "python", - "outputs": [], - "prompt_number": 51 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t['changed'] = df_t['dt'] > last" - ], - "language": "python", - "outputs": [], - "prompt_number": 53 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 54, - "text": [ - " sid volume dt price changed", - "133 133 100 2011-04-08 00:00:00+00:00 10.1 True", - "134 134 100 2011-04-08 00:00:00+00:00 12.12 True" - ] - } - ], - "prompt_number": 54 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t.index" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 59, - "text": [ - "Int64Index([133, 134])" - ] - } - ], - "prompt_number": 59 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df.index" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 60, - "text": [ - "Index([sid, volume, dt, price, changed], dtype=object)" - ] - } - ], - "prompt_number": 60 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df.columns" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 61, - "text": [ - "Int64Index([133, 134])" - ] - } - ], - "prompt_number": 61 - }, - { - "cell_type": "code", - "collapsed": false, - "input": [ - "df_t.columns" - ], - "language": "python", - "outputs": [ - { - "output_type": "pyout", - "prompt_number": 62, - "text": [ - "Index([sid, volume, dt, price, changed], dtype=object)" - ] - } - ], - "prompt_number": 62 - }, - { - "cell_type": "code", - "collapsed": true, - "input": [], - "language": "python", - "outputs": [] - } - ] - } - ] -} \ No newline at end of file From 0a0f35a14ec775915d5e2abc6d018f80e3517dcf Mon Sep 17 00:00:00 2001 From: Jonathan Kamens Date: Thu, 23 Aug 2012 17:25:58 -0400 Subject: [PATCH 15/54] Make ordered_pip.sh more robust Make ordered_pip.sh exit on error. Remove useless echo of empty string. Fix counting of the number of packages installed. Ignore blank lines in requirements file. --- etc/ordered_pip.sh | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/etc/ordered_pip.sh b/etc/ordered_pip.sh index ca3fa440..3db63eb1 100755 --- a/etc/ordered_pip.sh +++ b/etc/ordered_pip.sh @@ -1,11 +1,12 @@ -#!/bin/bash +#!/bin/bash -e -echo $hash +a=0 while read line do - if [[ $line != \#* ]] ; then + if [[ -n "$line" && "$line" != \#* ]] ; then #echo $line pip install $line fi + ((a = a + 1)) done < $1 -echo "Final line count is: $a"; +echo "$0: Final package count is $a"; From f14c6c7077d534bec51538a0fbfedc090238bed2 Mon Sep 17 00:00:00 2001 From: Jonathan Kamens Date: Thu, 23 Aug 2012 17:27:00 -0400 Subject: [PATCH 16/54] Fix statsmodels version dependencies to make pip happy Use statsmodels>=0.5.0 instead of statsmodels==0.5.0-tutorial-beta, because although the latter is the download version listed in pypi, it installs as 0.5.0, which means if we list that specific version as the requirement, then pip reinstalls it every time launch.sh is run. --- etc/requirements_sci.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/etc/requirements_sci.txt b/etc/requirements_sci.txt index a0104713..cc15764a 100644 --- a/etc/requirements_sci.txt +++ b/etc/requirements_sci.txt @@ -13,7 +13,7 @@ matplotlib==1.1.0 numexpr==2.0.1 Cython==0.15.1 patsy==0.1.0 -statsmodels==0.5.0-tutorial-beta +statsmodels>=0.5.0 # ZeroMQ From 215a64ed6a86d84d921efc07fb2ad3f7ba69b6a8 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 20:58:53 -0400 Subject: [PATCH 17/54] Removes unused import of zmq. --- zipline/finance/performance.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index 487c25de..a6dea38b 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -125,8 +125,6 @@ import datetime import pytz import math -import zmq - import zipline.protocol as zp import zipline.finance.risk as risk From 61ef8f434104451494611db72d358a2859a3fc5c Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 23 Aug 2012 23:48:05 -0400 Subject: [PATCH 18/54] Removes unused imports. --- tests/test_exception_handling.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index a9d6063e..7b4fb624 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -9,16 +9,15 @@ from zipline.finance.trading import SIMULATION_STYLE from zipline.core.devsimulator import AddressAllocator from zipline.lines import SimulatedTrading from zipline.gens.transform import StatefulTransform -from zipline.gens.tradesimulation import HEARTBEAT_INTERVAL, \ - MAX_HEARTBEAT_INTERVALS +from zipline.gens.tradesimulation import MAX_HEARTBEAT_INTERVALS -from zipline.utils.test_utils import \ - drain_zipline, \ - check, \ - setup_logger, \ - teardown_logger, \ - ExceptionSource, \ - ExceptionTransform +from zipline.utils.test_utils import ( + drain_zipline, + setup_logger, + teardown_logger, + ExceptionSource, + ExceptionTransform +) DEFAULT_TIMEOUT = 15 # seconds EXTENDED_TIMEOUT = 90 From 2800c472b857b849b37a4e0c92d8db2211b81615 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 00:05:58 -0400 Subject: [PATCH 19/54] Removes unused package of psutil. --- etc/requirements.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index bef0ebf6..4993d549 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -18,4 +18,3 @@ setproctitle==1.1.6 Logbook==0.3 blist==1.3.4 -psutil==0.4.1 From 20db5dedb55a6b5fe3809bcf35935949f51c6044 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 00:07:13 -0400 Subject: [PATCH 20/54] Removes unused package from requirements. --- etc/requirements.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index 4993d549..1c4a95b9 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -1,6 +1,5 @@ msgpack-python==0.1.12 humanhash==0.0.1 -ujson==1.18 iso8601==0.1.4 # ZeroMQ From 16aadc79ff5cc397ad221dc4b496be16f6dd4663 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 09:33:56 -0400 Subject: [PATCH 21/54] Removes unused module. --- zipline/utils/protocol_utils.py | 1 - 1 file changed, 1 deletion(-) diff --git a/zipline/utils/protocol_utils.py b/zipline/utils/protocol_utils.py index c74c81f2..c3ae9352 100644 --- a/zipline/utils/protocol_utils.py +++ b/zipline/utils/protocol_utils.py @@ -2,7 +2,6 @@ import copy import pandas from ctypes import Structure, c_ubyte from collections import MutableMapping -from itertools import izip def Enum(*options): """ From 33f42a9374e74f42d413fb42c28e5fbd477e3ecf Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 13:18:59 -0400 Subject: [PATCH 22/54] Removes unused imports. --- zipline/gens/composites.py | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 63d69d0f..a64d10ff 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -1,9 +1,6 @@ -import datetime -from itertools import tee, starmap, chain -from collections import namedtuple +from itertools import tee, chain -from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.utils import roundrobin, hash_args, done_message +from zipline.gens.utils import roundrobin, done_message from zipline.gens.sort import date_sort from zipline.gens.merge import merge from zipline.gens.transform import StatefulTransform From 59362fb78bd59428f4fe5688cea4972c227b0bc8 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 13:22:25 -0400 Subject: [PATCH 23/54] Applies PEP-8 fixes. Whitespace wrangling. --- zipline/gens/composites.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index a64d10ff..05839a28 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -5,6 +5,7 @@ from zipline.gens.sort import date_sort from zipline.gens.merge import merge from zipline.gens.transform import StatefulTransform + def date_sorted_sources(*sources): """ Takes an iterable of sources, generating namestrings and @@ -13,7 +14,7 @@ def date_sorted_sources(*sources): for source in sources: assert iter(source), "Source %s not iterable" % source - assert source.__class__.__dict__.has_key('get_hash'), "No get_hash" + assert 'get_hash' in source.__class__.__dict__, "No get_hash" # Get name hashes to pass to date_sort. names = [source.get_hash() for source in sources] @@ -28,6 +29,7 @@ def date_sorted_sources(*sources): return date_sort(stream_in, names) + def merged_transforms(sorted_stream, *transforms): """ A generator that takes the expected output of a date_sort, pipes @@ -42,7 +44,7 @@ def merged_transforms(sorted_stream, *transforms): assert isinstance(transform, StatefulTransform) transform.merged = True transform.sequential = False - + # Generate expected hashes for each transform namestrings = [tnfm.get_hash() for tnfm in transforms] @@ -66,6 +68,7 @@ def merged_transforms(sorted_stream, *transforms): # Return the merged events. return add_done(dt_aliased) + def sequential_transforms(stream_in, *transforms): """ Apply each transform in transforms sequentially to each event in stream_in. @@ -78,7 +81,7 @@ def sequential_transforms(stream_in, *transforms): for tnfm in transforms: tnfm.sequential = True tnfm.merged = False - + # Recursively apply all transforms to the stream. stream_out = reduce(lambda stream, tnfm: tnfm.transform(stream), transforms, @@ -87,6 +90,7 @@ def sequential_transforms(stream_in, *transforms): dt_aliased = alias_dt(stream_out) return add_done(dt_aliased) + def alias_dt(stream_in): """ Alias the dt field to datetime on each message. @@ -95,6 +99,7 @@ def alias_dt(stream_in): message['datetime'] = message['dt'] yield message + # Add a done message to a stream. def add_done(stream_in): return chain(stream_in, [done_message('Composite')]) From 9b664e54d867cbefe65c6d64ab487fd30abe187e Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 13:29:00 -0400 Subject: [PATCH 24/54] Removes unused imports in tests. --- tests/test_sorting.py | 18 +++++++++--------- tests/test_transforms.py | 5 +---- 2 files changed, 10 insertions(+), 13 deletions(-) diff --git a/tests/test_sorting.py b/tests/test_sorting.py index bec97e31..003fc889 100644 --- a/tests/test_sorting.py +++ b/tests/test_sorting.py @@ -1,22 +1,22 @@ import pytz from unittest2 import TestCase -from itertools import cycle, chain, izip, izip_longest +from itertools import chain, izip_longest from datetime import datetime, timedelta from collections import deque from zipline import ndict -from zipline.gens.sort import \ - date_sort, \ - ready, \ - done, \ - queue_is_ready,\ +from zipline.gens.sort import ( + date_sort, + ready, + done, + queue_is_ready, queue_is_done -from zipline.gens.utils import hash_args, alternate, done_message -from zipline.gens.tradegens import date_gen, SpecificEquityTrades +) +from zipline.gens.utils import alternate, done_message +from zipline.gens.tradegens import SpecificEquityTrades from zipline.gens.composites import date_sorted_sources -import zipline.protocol as zp class HelperTestCase(TestCase): diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 8e203664..3672c4e1 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -2,14 +2,11 @@ import pytz import numpy from datetime import timedelta, datetime -from collections import defaultdict from unittest2 import TestCase from zipline import ndict -from zipline.lines import SimulatedTrading - -from zipline.utils.test_utils import setup_logger, teardown_logger +from zipline.utils.test_utils import setup_logger from zipline.utils.date_utils import utcnow from zipline.gens.tradegens import SpecificEquityTrades From d36cdec35013aeb326e5a981e916c01da57099cd Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 14:04:03 -0400 Subject: [PATCH 25/54] Applies PEP-8 recommendations. Whitespace wrangling. --- tests/test_sorting.py | 93 ++++++++++++++++++++++--------------------- 1 file changed, 47 insertions(+), 46 deletions(-) diff --git a/tests/test_sorting.py b/tests/test_sorting.py index 003fc889..63455bbd 100644 --- a/tests/test_sorting.py +++ b/tests/test_sorting.py @@ -36,7 +36,6 @@ class HelperTestCase(TestCase): assert queue_is_ready(queue) assert not queue_is_done(queue) - queue.appendleft(to_dt('DONE')) assert queue_is_ready(queue) @@ -81,6 +80,7 @@ class HelperTestCase(TestCase): assert ready(sources) assert done(sources) + class DateSortTestCase(TestCase): def setUp(self): @@ -99,7 +99,7 @@ class DateSortTestCase(TestCase): assert m1 == m2 def test_single_source(self): - + # Just using the built-in defaults. See # zipline/gens/tradegens.py source = SpecificEquityTrades() @@ -111,28 +111,28 @@ class DateSortTestCase(TestCase): self.run_date_sort(with_done, expected, [source.get_hash()]) def test_multi_source(self): - - filter = [2,3] + + filter = [2, 3] args_a = tuple() kwargs_a = { - 'count' : 100, - 'sids' : [1,2,3], - 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 6), - 'filter' : filter + 'count': 100, + 'sids': [1, 2, 3], + 'start': datetime(2012, 1, 3, 15, tzinfo=pytz.utc), + 'delta': timedelta(minutes=6), + 'filter': filter } source_a = SpecificEquityTrades(*args_a, **kwargs_a) args_b = tuple() kwargs_b = { - 'count' : 100, - 'sids' : [2,3,4], - 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 5), - 'filter' : filter + 'count': 100, + 'sids': [2, 3, 4], + 'start': datetime(2012, 1, 3, 15, tzinfo=pytz.utc), + 'delta': timedelta(minutes=5), + 'filter': filter } source_b = SpecificEquityTrades(*args_b, **kwargs_b) - + all_events = list(chain(source_a, source_b)) # The expected output is all events, sorted by dt with @@ -150,76 +150,75 @@ class DateSortTestCase(TestCase): with_done_b = chain(source_b, [done_message(source_b.get_hash())]) interleaved = alternate(with_done_a, with_done_b) - + # Test sort with alternating messages from source_a and # source_b. self.run_date_sort(interleaved, expected, source_ids) - + source_a.rewind() source_b.rewind() with_done_a = chain(source_a, [done_message(source_a.get_hash())]) with_done_b = chain(source_b, [done_message(source_b.get_hash())]) - + sequential = chain(with_done_a, with_done_b) - + # Test sort with all messages from a, followed by all messages # from b. - - self.run_date_sort(sequential, expected, source_ids) + self.run_date_sort(sequential, expected, source_ids) def test_sort_composite(self): - filter = [1,2] + filter = [1, 2] #Set up source a. One hour between events. args_a = tuple() kwargs_a = { - 'count' : 100, - 'sids' : [1], - 'start' : datetime(2012,6,6,0), - 'delta' : timedelta(hours = 1), - 'filter' : filter + 'count': 100, + 'sids': [1], + 'start': datetime(2012, 6, 6, 0), + 'delta': timedelta(hours=1), + 'filter': filter } source_a = SpecificEquityTrades(*args_a, **kwargs_a) #Set up source b. One day between events. args_b = tuple() kwargs_b = { - 'count' : 50, - 'sids' : [2], - 'start' : datetime(2012,6,6,0), - 'delta' : timedelta(days = 1), - 'filter' : filter + 'count': 50, + 'sids': [2], + 'start': datetime(2012, 6, 6, 0), + 'delta': timedelta(days=1), + 'filter': filter } source_b = SpecificEquityTrades(*args_b, **kwargs_b) #Set up source c. One minute between events. args_c = tuple() kwargs_c = { - 'count' : 150, - 'sids' : [1,2], - 'start' : datetime(2012,6,6,0), - 'delta' : timedelta(minutes = 1), - 'filter' : filter + 'count': 150, + 'sids': [1, 2], + 'start': datetime(2012, 6, 6, 0), + 'delta': timedelta(minutes=1), + 'filter': filter } source_c = SpecificEquityTrades(*args_c, **kwargs_c) # Set up source d. This should produce no events because the # internal sids don't match the filter. args_d = tuple() kwargs_d = { - 'count' : 50, - 'sids' : [3], - 'start' : datetime(2012,6,6,0), - 'delta' : timedelta(minutes = 1), - 'filter' : filter + 'count': 50, + 'sids': [3], + 'start': datetime(2012, 6, 6, 0), + 'delta': timedelta(minutes=1), + 'filter': filter } source_d = SpecificEquityTrades(*args_d, **kwargs_d) sources = [source_a, source_b, source_c, source_d] hashes = [source.get_hash() for source in sources] - + sort_out = date_sorted_sources(*sources) - + # Read all the values from sort and assert that they arrive in # the correct sorting with the expected hash values. to_list = list(sort_out) @@ -239,7 +238,8 @@ class DateSortTestCase(TestCase): assert to_list == expected -def compare_by_dt_source_id(x,y): + +def compare_by_dt_source_id(x, y): if x.dt < y.dt: return -1 elif x.dt > y.dt: @@ -252,8 +252,9 @@ def compare_by_dt_source_id(x,y): else: return 0 -#Alias for ease of use +#Alias for ease of use comp = compare_by_dt_source_id + def to_dt(msg): return ndict({'dt': msg}) From 8edbc581c1b0e0d9f325ab0320af8b78359f0917 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 14:10:11 -0400 Subject: [PATCH 26/54] Applies PEP-8 recommendations. Whitespace wrangling. --- zipline/gens/utils.py | 63 ++++++++++++++++++++++++++----------------- 1 file changed, 39 insertions(+), 24 deletions(-) diff --git a/zipline/gens/utils.py b/zipline/gens/utils.py index b8ee6ac4..ce91d550 100644 --- a/zipline/gens/utils.py +++ b/zipline/gens/utils.py @@ -8,26 +8,29 @@ from itertools import izip_longest from zipline import ndict from zipline.protocol import DATASOURCE_TYPE + def mock_raw_event(sid, dt): event = { - 'sid' : sid, - 'dt' : dt, - 'price' : 1.0, - 'volume' : 1 + 'sid': sid, + 'dt': dt, + 'price': 1.0, + 'volume': 1 } return event + def mock_done(id): return ndict({ - 'dt' : "DONE", - "source_id" : id, - 'tnfm_id' : id, + 'dt': "DONE", + "source_id": id, + 'tnfm_id': id, 'tnfm_value': None, - 'type' : DATASOURCE_TYPE.DONE + 'type': DATASOURCE_TYPE.DONE }) done_message = mock_done + def alternate(g1, g2): """Specialized version of roundrobin for just 2 generators.""" for e1, e2 in izip_longest(g1, g2): @@ -36,6 +39,7 @@ def alternate(g1, g2): if e2 != None: yield e2 + def roundrobin(sources, namestrings): """ Takes N generators, pulling one element off each until all inputs @@ -56,32 +60,36 @@ def roundrobin(sources, namestrings): yield done_message(namestring) del mapping[namestring] + def hash_args(*args, **kwargs): """Define a unique string for any set of representable args.""" arg_string = '_'.join([str(arg) for arg in args]) - kwarg_string = '_'.join([str(key) + '=' + str(value) for key, value in kwargs.iteritems()]) + kwarg_string = '_'.join([str(key) + '=' + str(value) + for key, value in kwargs.iteritems()]) combined = ':'.join([arg_string, kwarg_string]) hasher = md5() hasher.update(combined) return hasher.hexdigest() -def create_trade(sid, price, amount, datetime, source_id = "test_factory"): + +def create_trade(sid, price, amount, datetime, source_id="test_factory"): row = ndict({ - 'source_id' : source_id, - 'type' : DATASOURCE_TYPE.TRADE, - 'sid' : sid, - 'dt' : datetime, - 'price' : price, - 'close' : price, - 'open' : price, - 'low' : price * .95, - 'high' : price * 1.05, - 'volume' : amount + 'source_id': source_id, + 'type': DATASOURCE_TYPE.TRADE, + 'sid': sid, + 'dt': datetime, + 'price': price, + 'close': price, + 'open': price, + 'low': price * .95, + 'high': price * 1.05, + 'volume': amount }) return row + def sum_true(bool_iterable): """ Takes an iterable of boolean values and returns the number of @@ -102,6 +110,7 @@ def assert_datasource_protocol(event): assert isinstance(event.dt, datetime) assert event.dt.tzinfo == pytz.utc + def assert_trade_protocol(event): """Assert that an event meets the protocol for datasource TRADE outputs.""" assert_datasource_protocol(event) @@ -113,32 +122,38 @@ def assert_trade_protocol(event): assert isinstance(event.volume, numbers.Integral) assert isinstance(event.dt, datetime) + def assert_datasource_unframe_protocol(event): """Assert that an event is valid output of zp.DATASOURCE_UNFRAME.""" assert isinstance(event, ndict) assert isinstance(event.source_id, basestring) assert event.type in DATASOURCE_TYPE - assert event.has_key('dt') + assert 'dt' in event + def assert_sort_protocol(event): """Assert that an event is valid input to zp.FEED_FRAME.""" assert isinstance(event, ndict) assert isinstance(event.source_id, basestring) assert event.type in DATASOURCE_TYPE - assert event.has_key('dt') + assert 'dt' in event + def assert_sort_unframe_protocol(event): """Same as above.""" assert isinstance(event, ndict) assert isinstance(event.source_id, basestring) assert event.type in DATASOURCE_TYPE - assert event.has_key('dt') + assert 'dt' in event + def assert_transform_protocol(event): """Transforms should return an ndict to be merged by merge.""" assert isinstance(event, ndict) + def assert_merge_protocol(tnfm_ids, message): - """Merge should output an ndict with a field for each id in its transform set.""" + """Merge should output an ndict with a field for each id + in its transform set.""" assert isinstance(message, ndict) assert set(tnfm_ids) == set(message.keys()) From 06742e27a1c09fcd824c8937b594d39aaa48fffa Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Fri, 24 Aug 2012 23:36:04 -0400 Subject: [PATCH 27/54] Removes unused import. --- tests/test_finance.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_finance.py b/tests/test_finance.py index 5562b671..87795609 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -12,7 +12,6 @@ from nose.tools import timed import zipline.utils.factory as factory -from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import TradingEnvironment from zipline.core.devsimulator import AddressAllocator from zipline.lines import SimulatedTrading From 6f195233f5d86a4d79c12259d8a9eb760b369160 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Sat, 25 Aug 2012 01:38:33 -0400 Subject: [PATCH 28/54] Applies PEP-8 recommendations. Whitespace wrangling. --- tests/test_logger.py | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/tests/test_logger.py b/tests/test_logger.py index 5a9b8e31..7a2ae780 100644 --- a/tests/test_logger.py +++ b/tests/test_logger.py @@ -13,23 +13,21 @@ from zipline.utils.test_utils import create_receiver, drain_receiver from unittest2 import TestCase - class LoggerTestCase(TestCase): def setUp(self): - configure_logging() - self.LOG = logging.getLogger("ZiplineLogger") + configure_logging() + self.LOG = logging.getLogger("ZiplineLogger") def test_log(self): test_msg = uuid.uuid1().hex self.LOG.info(test_msg) - logfile = open('/var/log/zipline/zipline.log','r') + logfile = open('/var/log/zipline/zipline.log', 'r') with logfile: last_line = tail(logfile, window=1) logged_msg = last_line.split(" - ")[1] self.assertEqual(test_msg, logged_msg) - def test_zmq_handler(self): socket_addr = 'tcp://127.0.0.1:10000' ctx = zmq.Context() @@ -37,8 +35,8 @@ class LoggerTestCase(TestCase): socket_push.connect(socket_addr) recv = create_receiver(socket_addr, ctx) zmq_out = ZeroMQLogHandler( - socket = socket_push, - filter = lambda r, h: r.channel in ['test zmq logger'], + socket=socket_push, + filter=lambda r, h: r.channel in ['test zmq logger'], context=ctx, #bubble=False ) @@ -50,7 +48,6 @@ class LoggerTestCase(TestCase): with zmq_out.threadbound(): log.info(ex.num) - output, _ = drain_receiver(recv, count=1) self.assertEqual(output[-1]['prefix'], 'LOG') self.assertTrue(isinstance(output[-1]['payload']['msg'], basestring)) From 1f5cb86940a7586f50b3dcc1a65704e0246427e1 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Sat, 25 Aug 2012 01:34:19 -0400 Subject: [PATCH 29/54] Applies PEP-8 recommendations. Whitespace wrangling. --- tests/test_monitor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 76bb6184..7f2daa94 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -3,16 +3,16 @@ from unittest2 import TestCase, skip from zipline.core.monitor import Monitor + class TestMonitor(TestCase): def setUp(self): setup_logger(self, '/var/log/qexec/qexec.log') - def tearDown(self): teardown_logger(self) def test_init(self): - pub_socket = 'tcp://127.0.0.1:5000' + pub_socket = 'tcp://127.0.0.1:5000' route_socket = 'tcp://127.0.0.1:5001' exception_socket = 'tcp://127.0.0.1:5002' @@ -20,9 +20,9 @@ class TestMonitor(TestCase): mon.manage([]) def test_init_topology(self): - pub_socket = 'tcp://127.0.0.1:5000' + pub_socket = 'tcp://127.0.0.1:5000' route_socket = 'tcp://127.0.0.1:5001' exception_socket = 'tcp://127.0.0.1:5002' mon = Monitor(pub_socket, route_socket, exception_socket) - mon.manage([ 'a', 'b', 'c', 'd' ]) + mon.manage(['a', 'b', 'c', 'd']) From 11b67970a69d5e115fa85880a8b88ae4232df223 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Sat, 25 Aug 2012 01:33:13 -0400 Subject: [PATCH 30/54] Removes import of unused module. --- tests/test_monitor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 7f2daa94..fa5de2b8 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -1,5 +1,5 @@ from zipline.utils.test_utils import setup_logger, teardown_logger -from unittest2 import TestCase, skip +from unittest2 import TestCase from zipline.core.monitor import Monitor From 72e64776b9c20a6e0ada49df39d195050ef31cfd Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Sat, 25 Aug 2012 01:21:02 -0400 Subject: [PATCH 31/54] Applies PEP-8 recommendations to test_utils. Whitespace wrangling. --- zipline/utils/test_utils.py | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 3a9a0906..9b01d87c 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -9,14 +9,17 @@ from itertools import izip from logbook import FileHandler from zipline.core.monitor import Monitor + def setup_logger(test, path='/var/log/zipline/zipline.log'): test.log_handler = FileHandler(path) test.log_handler.push_application() + def teardown_logger(test): test.log_handler.pop_application() test.log_handler.close() + def check_list(test, a, b, label): test.assertTrue(isinstance(a, (list, blist.blist))) test.assertTrue(isinstance(b, (list, blist.blist))) @@ -33,8 +36,8 @@ def check_dict(test, a, b, label): if key in ['progress']: continue - test.assertTrue(a.has_key(key), "missing key at: " + label + "." + key) - test.assertTrue(b.has_key(key), "missing key at: " + label + "." + key) + test.assertTrue(key in a, "missing key at: " + label + "." + key) + test.assertTrue(key in b, "missing key at: " + label + "." + key) a_val = a[key] b_val = b[key] check(test, a_val, b_val, label + "." + key) @@ -62,6 +65,7 @@ def check(test, a, b, label=None): else: test.assertEqual(a, b, "mismatch on path: " + label) + def drain_zipline(test, zipline, p_blocking=False): assert test.ctx, "method expects a valid zmq context" assert test.zipline_test_config, "method expects a valid test config" @@ -86,15 +90,17 @@ def drain_zipline(test, zipline, p_blocking=False): return output, transaction_count + def create_receiver(socket_addr, ctx): receiver = ctx.socket(zmq.PULL) receiver.bind(socket_addr) return receiver + def drain_receiver(receiver, count=None): output = [] - transaction_count = 0 + transaction_count = 0 msg_counter = 0 while True: msg = receiver.recv() @@ -119,7 +125,9 @@ def drain_receiver(receiver, count=None): def assert_single_position(test, zipline, blocking=False): - output, transaction_count = drain_zipline(test, zipline, p_blocking=blocking) + output, transaction_count = drain_zipline(test, + zipline, + p_blocking=blocking) test.assertEqual(output[-1]['prefix'], 'DONE') test.assertEqual( @@ -152,6 +160,7 @@ def launch_component(component): proc.start() return proc + def launch_monitor(monitor): proc = multiprocessing.Process(target=monitor.run) proc.start() @@ -176,6 +185,7 @@ def create_monitor(allocator): return mon + class ExceptionSource(object): def __init__(self): @@ -190,6 +200,7 @@ class ExceptionSource(object): def next(self): 5 / 0 + class ExceptionTransform(object): def __init__(self): From 3a054786f8c43f90753fd8ca5187289fc4e23a6d Mon Sep 17 00:00:00 2001 From: Thomas Wiecki Date: Wed, 29 Aug 2012 09:05:25 -0400 Subject: [PATCH 32/54] Added cvxopt and scikit-learn to requirements. --- etc/requirements_sci.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/etc/requirements_sci.txt b/etc/requirements_sci.txt index cc15764a..294459d9 100644 --- a/etc/requirements_sci.txt +++ b/etc/requirements_sci.txt @@ -14,7 +14,8 @@ numexpr==2.0.1 Cython==0.15.1 patsy==0.1.0 statsmodels>=0.5.0 - +scikit-learn==0.11 +cvxopt==1.1.5 # ZeroMQ pyzmq==2.1.11 From dc4c9082e9ba8122580edb07d3eb85e622e7f027 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 29 Aug 2012 14:36:10 -0400 Subject: [PATCH 33/54] Removes cvxopt from pip installs. It appears that cvxopt is uninstallable via pip in it's current state. --- etc/requirements_sci.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/etc/requirements_sci.txt b/etc/requirements_sci.txt index 294459d9..f138534c 100644 --- a/etc/requirements_sci.txt +++ b/etc/requirements_sci.txt @@ -15,7 +15,6 @@ Cython==0.15.1 patsy==0.1.0 statsmodels>=0.5.0 scikit-learn==0.11 -cvxopt==1.1.5 # ZeroMQ pyzmq==2.1.11 From 771e7da531854b3636f8c6252502fd8d0460e0fe Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Sat, 1 Sep 2012 23:01:44 -0400 Subject: [PATCH 34/54] Fixes spelling of burrito. --- zipline/core/monitor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index 3736b7e3..acc80132 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -379,7 +379,7 @@ class Monitor(object): # Noop exit func #sys.exitfunc = lambda: None - # Send SIGHUP to buritto + # Send SIGHUP to burrito self.signal_hangup() if not self.alive: From d656c914d5813543f5a4fd930f26e8036f56d364 Mon Sep 17 00:00:00 2001 From: scottsanderson Date: Tue, 28 Aug 2012 14:26:42 -0400 Subject: [PATCH 35/54] remove outdated examples file --- zipline/gens/examples.py | 100 --------------------------------------- 1 file changed, 100 deletions(-) delete mode 100644 zipline/gens/examples.py diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py deleted file mode 100644 index f3a0dd0b..00000000 --- a/zipline/gens/examples.py +++ /dev/null @@ -1,100 +0,0 @@ -import pytz -import time - -from time import sleep -from pprint import pprint as pp -from datetime import datetime, timedelta -from itertools import izip - -from zipline.utils.factory import create_trading_environment -from zipline.test_algorithms import TestAlgorithm - -from zipline.gens.composites import SourceBundle, TransformBundle, \ - date_sorted_sources, merged_transforms, sequential_transforms -from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform -from zipline.gens.tradesimulation import TradeSimulationClient as tsc - -import zipline.protocol as zp - -if __name__ == "__main__": - - filter = [2,3] - #Set up source a. Six minutes between events. - args_a = tuple() - kwargs_a = { - 'count' : 1000, - 'sids' : [1,2,3], - 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 6), - 'filter' : filter - } - source_a = SpecificEquityTrades(*args_a, **kwargs_a) - source_a_prime = SpecificEquityTrades(*args_a, **kwargs_a) - - #Set up source b. Five minutes between events. - args_b = tuple() - kwargs_b = { - 'count' : 1000, - 'sids' : [2,3,4], - 'start' : datetime(2012,1,3,14, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 5), - 'filter' : filter - } - source_b = SpecificEquityTrades(*args_b, **kwargs_b) - source_b_prime = SpecificEquityTrades(*args_b, **kwargs_b) - - sorted = date_sorted_sources(source_a, source_b) - sorted_prime = date_sorted_sources( - source_a_prime, - source_b_prime - ) - - passthrough = StatefulTransform(Passthrough) - mavg_price = StatefulTransform( - MovingAverage, - timedelta(minutes = 20), - ['price'] - ) - - passthrough_prime = StatefulTransform(Passthrough) - mavg_price_prime = StatefulTransform( - MovingAverage, - timedelta(minutes = 20), - ['price'] - ) - - merged = merged_transforms(sorted, passthrough, mavg_price) - start = time.time() - for message in merged: - assert 1 + 1 == 2 - stop = time.time() - merge_time = stop - start - print "Merge time: %s" % str(merge_time) - - sequential = sequential_transforms( - sorted_prime, - passthrough_prime, - mavg_price_prime - ) - - start = time.time() - for message in sequential: - assert 1 + 1 == 2 - stop = time.time() - seq_time = stop - start - print "Sequential time: %s" % str(seq_time) - print "Merge/Seq: %s" % (str(merge_time/seq_time)) - - -# merged = merged_transforms(sorted, passthrough, mavg_price) - - # algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3]) -# environment = create_trading_environment(year = 2012) -# style = zp.SIMULATION_STYLE.FIXED_SLIPPAGE - -# trading_client = tsc(algo, environment, style) - -# for message in trading_client.simulate(merged): -# pp(message) - From c0694827d655099f066f70ef5879b56a46b062be Mon Sep 17 00:00:00 2001 From: scottsanderson Date: Tue, 28 Aug 2012 14:24:59 -0400 Subject: [PATCH 36/54] add the TransformMeta metaclass --- zipline/gens/transform.py | 31 +++++++++++++++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index c19501ee..29c2e25e 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -29,6 +29,21 @@ class Passthrough(object): def update(self, event): pass +class TransformMeta(type): + """ + Metaclass that automatically packages a class inside of + StatefulTransform on initialization. Specifically, if Foo is a + class with its __metaclass__ attribute set to TransformMeta, then + calling Foo(*args, **kwargs) will return StatefulTransform(Foo, + *args, **kwargs) instead of an instance of Foo. (Note that you can + still recover an instance of a "raw" Foo by introspecting the + resulting StatefulTransform's 'state' field. + """ + + def __call__(cls, *args, **kwargs): + return StatefulTransform(cls, *args, **kwargs) + + class StatefulTransform(object): """ Generic transform generator that takes each message from an @@ -55,11 +70,23 @@ class StatefulTransform(object): self.merged = True # Create an instance of our transform class. - self.state = tnfm_class(*args, **kwargs) + if isinstance(tnfm_class, TransformMeta): + # Classes derived TransformMeta have their __call__ + # attribute overridden. Since this is what is usually + # used to create an instance, we have to delegate the + # responsibility of creating an instance to + # TransformMeta's parent class, which is 'type'. This is + # what is implicitly done behind the scenes by the python + # interpreter for most classes anyway, but here we have to + # be explicit because we've overridden the method that + # usually resolves to our super call. + self.state = super(TransformMeta, tnfm_class).__call__(*args, **kwargs) + # Normal object instantiation. + else: + self.state = tnfm_class(*args, **kwargs) # Create the string associated with this generator's output. self.namestring = tnfm_class.__name__ + hash_args(*args, **kwargs) - log.info('StatefulTransform [%s] initialized' % self.namestring) def get_hash(self): return self.namestring From 71839522cfd9675cb2ba6b31f6036801f2b9e8bb Mon Sep 17 00:00:00 2001 From: scottsanderson Date: Tue, 28 Aug 2012 14:25:17 -0400 Subject: [PATCH 37/54] switch current transforms over to the new metaclass --- tests/test_transforms.py | 11 ++++------- zipline/gens/mavg.py | 3 ++- zipline/gens/returns.py | 3 +++ zipline/gens/stddev.py | 3 ++- zipline/gens/vwap.py | 4 +++- 5 files changed, 14 insertions(+), 10 deletions(-) diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 3672c4e1..d617ce20 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -153,8 +153,7 @@ class FinanceTransformsTestCase(TestCase): def test_vwap(self): - vwap = StatefulTransform( - VWAP, + vwap = VWAP( market_aware = False, delta = timedelta(days = 2) ) @@ -177,7 +176,7 @@ class FinanceTransformsTestCase(TestCase): def test_returns(self): # Daily returns. - returns = StatefulTransform(Returns, 1) + returns = Returns(1) transformed = list(returns.transform(self.source)) tnfm_vals = [message.tnfm_value for message in transformed] @@ -218,8 +217,7 @@ class FinanceTransformsTestCase(TestCase): def test_moving_average(self): - mavg = StatefulTransform( - MovingAverage, + mavg = MovingAverage( market_aware = False, fields = ['price', 'volume'], delta = timedelta(days = 2), @@ -260,8 +258,7 @@ class FinanceTransformsTestCase(TestCase): self.trading_environment ) - stddev = StatefulTransform( - MovingStandardDev, + stddev = MovingStandardDev( market_aware = False, delta = timedelta(minutes = 150), ) diff --git a/zipline/gens/mavg.py b/zipline/gens/mavg.py index 21aa0bd0..3daba2bc 100644 --- a/zipline/gens/mavg.py +++ b/zipline/gens/mavg.py @@ -3,7 +3,7 @@ from datetime import datetime, timedelta from collections import defaultdict from zipline import ndict -from zipline.gens.transform import EventWindow +from zipline.gens.transform import EventWindow, TransformMeta class MovingAverage(object): """ @@ -12,6 +12,7 @@ class MovingAverage(object): averages over any number of distinct fields (For example, we can maintain a sid's average volume as well as its average price.) """ + __metaclass__ = TransformMeta def __init__(self, fields, market_aware, days = None, delta = None): diff --git a/zipline/gens/returns.py b/zipline/gens/returns.py index 49d3e9b5..a4932485 100644 --- a/zipline/gens/returns.py +++ b/zipline/gens/returns.py @@ -1,3 +1,4 @@ +from zipline.gens.transform import TransformMeta from collections import defaultdict, deque class Returns(object): @@ -5,6 +6,8 @@ class Returns(object): Class that maintains a dictionary from sids to the sid's closing price N trading days ago. """ + __metaclass__ = TransformMeta + def __init__(self, days): self.days = days self.mapping = defaultdict(self._create) diff --git a/zipline/gens/stddev.py b/zipline/gens/stddev.py index 1f46429a..71759b90 100644 --- a/zipline/gens/stddev.py +++ b/zipline/gens/stddev.py @@ -4,7 +4,7 @@ from collections import defaultdict from math import sqrt from zipline import ndict -from zipline.gens.transform import EventWindow +from zipline.gens.transform import EventWindow, TransformMeta class MovingStandardDev(object): """ @@ -13,6 +13,7 @@ class MovingStandardDev(object): standard deviation of all events falling within the specified window. """ + __metaclass__ = TransformMeta def __init__(self, market_aware, days = None, delta = None): diff --git a/zipline/gens/vwap.py b/zipline/gens/vwap.py index 5a0947d8..b2e72dfd 100644 --- a/zipline/gens/vwap.py +++ b/zipline/gens/vwap.py @@ -3,12 +3,14 @@ from datetime import datetime, timedelta from collections import defaultdict from zipline import ndict -from zipline.gens.transform import EventWindow +from zipline.gens.transform import EventWindow, TransformMeta class VWAP(object): """ Class that maintains a dictionary from sids to VWAPEventWindows. """ + __metaclass__ = TransformMeta + def __init__(self, market_aware, delta=None, days=None): self.market_aware = market_aware From 5bd57d148c8faa42df794e3bbbc8b9b8deb6839a Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Tue, 4 Sep 2012 18:53:58 -0400 Subject: [PATCH 38/54] Removes unused imports from vwap. --- zipline/gens/vwap.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/zipline/gens/vwap.py b/zipline/gens/vwap.py index b2e72dfd..e904d0e0 100644 --- a/zipline/gens/vwap.py +++ b/zipline/gens/vwap.py @@ -1,8 +1,6 @@ from numbers import Number -from datetime import datetime, timedelta from collections import defaultdict -from zipline import ndict from zipline.gens.transform import EventWindow, TransformMeta class VWAP(object): From 4d5457a3631d598967f2e44b0b288ed5ec016287 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Tue, 4 Sep 2012 18:57:01 -0400 Subject: [PATCH 39/54] Applies PEP-8 recommendations to `vwap` module. --- zipline/gens/vwap.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/zipline/gens/vwap.py b/zipline/gens/vwap.py index e904d0e0..9d908b46 100644 --- a/zipline/gens/vwap.py +++ b/zipline/gens/vwap.py @@ -3,6 +3,7 @@ from collections import defaultdict from zipline.gens.transform import EventWindow, TransformMeta + class VWAP(object): """ Class that maintains a dictionary from sids to VWAPEventWindows. @@ -14,7 +15,7 @@ class VWAP(object): self.market_aware = market_aware self.delta = delta self.days = days - + # Market-aware mode only works with full-day windows. if self.market_aware: assert self.days and not self.delta,\ @@ -28,13 +29,13 @@ class VWAP(object): # No way to pass arguments to the defaultdict factory, so we # need to define a method to generate the correct EventWindows. self.sid_windows = defaultdict(self.create_window) - + def create_window(self): """Factory method for self.sid_windows.""" return VWAPEventWindow( - self.market_aware, - days = self.days, - delta = self.delta + self.market_aware, + days=self.days, + delta=self.delta ) def update(self, event): @@ -48,6 +49,7 @@ class VWAP(object): window.update(event) return window.get_vwap() + class VWAPEventWindow(EventWindow): """ Iteratively maintains a vwap for a single sid over a given @@ -69,7 +71,7 @@ class VWAPEventWindow(EventWindow): def handle_remove(self, event): self.flux -= event.volume * event.price self.totalvolume -= event.volume - + def get_vwap(self): """ Return the calculated vwap for this sid. From bc8bb299df445118e8e220262900daff02489d67 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Tue, 4 Sep 2012 23:37:54 -0400 Subject: [PATCH 40/54] Applies PEP-8 recommendations before editing. --- zipline/core/devsimulator.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/zipline/core/devsimulator.py b/zipline/core/devsimulator.py index cbce4b66..6bc76815 100644 --- a/zipline/core/devsimulator.py +++ b/zipline/core/devsimulator.py @@ -14,6 +14,7 @@ THE DEVSIMULATOR IS DEPRECATED, IT WILL NOT BEHAVE LIKE ANY OTHER SYSTEM USED IN TESTS OR IN PRODUCTION """ + class AddressAllocator(object): """ Produces a iterator of 10000 sockets to allocate as needed. @@ -28,9 +29,10 @@ class AddressAllocator(object): ] def lease(self, n): - sockets = self.sockets[self.idx:self.idx+n] + sockets = self.sockets[self.idx: self.idx + n] self.idx += n return sockets def reaquire(self, *conn): - pass \ No newline at end of file + pass + From 380ba0bf2a096a659dbe3be7f6b8007343024c01 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 07:59:56 -0400 Subject: [PATCH 41/54] Removes unused module. --- zipline/core/devsimulator.py | 1 - 1 file changed, 1 deletion(-) diff --git a/zipline/core/devsimulator.py b/zipline/core/devsimulator.py index 6bc76815..c28d0e72 100644 --- a/zipline/core/devsimulator.py +++ b/zipline/core/devsimulator.py @@ -4,7 +4,6 @@ See :py:method"" """ import logbook -import threading log = logbook.Logger('Dev Simulator') From 294236af4dbd5d986fe4150029a8066f8640855a Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 08:00:13 -0400 Subject: [PATCH 42/54] Removes trailing line. --- zipline/core/devsimulator.py | 1 - 1 file changed, 1 deletion(-) diff --git a/zipline/core/devsimulator.py b/zipline/core/devsimulator.py index c28d0e72..6e273f8c 100644 --- a/zipline/core/devsimulator.py +++ b/zipline/core/devsimulator.py @@ -34,4 +34,3 @@ class AddressAllocator(object): def reaquire(self, *conn): pass - From a6fa1def3a7c3ada5ef9c2483ac64b6292fa3fdd Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 08:03:54 -0400 Subject: [PATCH 43/54] Removes unused imports from tradesimulation module. --- zipline/gens/tradesimulation.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 98aa727f..5627dc05 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -1,14 +1,11 @@ -import signal from logbook import Logger, Processor -from datetime import datetime, timedelta -from numbers import Integral +from datetime import datetime from itertools import groupby from zipline import ndict from zipline.utils.timeout import Heartbeat, Timeout -from zipline.gens.transform import StatefulTransform from zipline.finance.trading import TransactionSimulator from zipline.finance.performance import PerformanceTracker from zipline.utils.log_utils import stdout_only_pipe From 48476cb7a754bba47dc2d284befede6abf2286a4 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 08:08:05 -0400 Subject: [PATCH 44/54] Removes noop test. --- tests/test_sanity.py | 7 ------- 1 file changed, 7 deletions(-) delete mode 100644 tests/test_sanity.py diff --git a/tests/test_sanity.py b/tests/test_sanity.py deleted file mode 100644 index 416bba57..00000000 --- a/tests/test_sanity.py +++ /dev/null @@ -1,7 +0,0 @@ -from unittest2 import TestCase - -class TestEnviroment(TestCase): - - def test_universe(self): - # first order logic is working today. Yay! - self.assertTrue(True != False) From ff526f11c9bde67f5e5f2bbc0bdbbc7b33f0bb41 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 08:16:17 -0400 Subject: [PATCH 45/54] Removes unused `profile` module. --- zipline/profile/__init__.py | 0 zipline/profile/prof.py | 104 ------------------------------------ 2 files changed, 104 deletions(-) delete mode 100644 zipline/profile/__init__.py delete mode 100644 zipline/profile/prof.py diff --git a/zipline/profile/__init__.py b/zipline/profile/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/zipline/profile/prof.py b/zipline/profile/prof.py deleted file mode 100644 index d292b300..00000000 --- a/zipline/profile/prof.py +++ /dev/null @@ -1,104 +0,0 @@ -""" - -Viscosity - Tools for benchmarking ZeroMQ data flow. - -""" - -import time as timer -import logging -import pycounters -from contextlib import contextmanager, nested -from pycounters import base -from pycounters.shortcuts import frequency, time -from pycounters import shortcuts, reporters, start_auto_reporting, register_reporter -from pycounters import shortcuts,reporters,report_value, output_report, \ -counters, register_counter, _reporting_decorator_context_manager - -JSONFile = "counters.json" - -logger = logging.getLogger('simple_example') -logger.setLevel(logging.DEBUG) - -ch = logging.StreamHandler() -ch.setLevel(logging.DEBUG) -logger.addHandler(ch) - -reporter = reporters.JSONFileReporter(output_file=JSONFile) -logreport = reporters.LogReporter(logger) -register_reporter(logreport) -register_reporter(reporter) - -class timecontext: - - def __init__(self, name): - self.name = name - - def __enter__(self): - cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) - if not cntr: - counter = counters.AverageTimeCounter(self.name) - register_counter(counter) - self.tic = timer.time() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - if not exc_type: - shortcuts.value(self.name, timer.time() - self.tic) - -class ttimecontext: - - def __init__(self, name): - self.name = name - - def __enter__(self): - counter = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) - - if not counter: - counter = counters.EventCounter(self.name) - counter.value = 0 - register_counter(counter) - - self.counter = counter - self.tic = timer.time() - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - if not exc_type: - val = (timer.time() - self.tic) - if not self.counter.value: - self.counter.value = long(0.0) - self.counter.value += val - -class occurancecontext: - - def __init__(self, name): - self.name = name - - def __enter__(self): - cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False) - if not cntr: - cntr = counters.TotalCounter(self.name) - counter = counters.TotalCounter(self.name) - register_counter(counter) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - shortcuts.value(self.name, 1) - -if __name__ == '__main__': - - with timecontext('average time'): - for i in xrange(5): - x = [2] * 1000 - timer.sleep(0.01) - - with occurancecontext('totalcount'): - for i in xrange(5): - x = [2] * 1000 - - with ttimecontext('total time'): - for i in xrange(5): - x = [2] * 1000 - timer.sleep(1) - - pycounters.output_report() From d265a646962aa5cbdfdfe2733ef7c018f14fd7fc Mon Sep 17 00:00:00 2001 From: Jonathan Kamens Date: Wed, 5 Sep 2012 10:53:55 -0400 Subject: [PATCH 46/54] Don't list setuptools or distribute in requirements.txt There's no need to list setuptools or distribute in requirements.txt because they are installed automatically when you create a new virtualenv. Furthermore, listing them explicitly breaks if the version you list is different from the virtualenv version. --- etc/requirements.txt | 4 ---- 1 file changed, 4 deletions(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index 1c4a95b9..a620712d 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -6,10 +6,6 @@ iso8601==0.1.4 pyzmq==2.1.11 gevent-zeromq==0.2.2 -# Packaging -distribute==0.6.27 -setuptools==0.6c11 - # Unix setproctitle==1.1.6 From b19a4f79dfdc6ab98359cde4f19681c5e5db3307 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 08:28:01 -0400 Subject: [PATCH 47/54] Removes unused import and cleans whitespace in mavg. --- zipline/gens/mavg.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/zipline/gens/mavg.py b/zipline/gens/mavg.py index 3daba2bc..de572236 100644 --- a/zipline/gens/mavg.py +++ b/zipline/gens/mavg.py @@ -1,10 +1,10 @@ from numbers import Number -from datetime import datetime, timedelta from collections import defaultdict from zipline import ndict from zipline.gens.transform import EventWindow, TransformMeta + class MovingAverage(object): """ Class that maintains a dictionary from sids to @@ -14,7 +14,7 @@ class MovingAverage(object): """ __metaclass__ = TransformMeta - def __init__(self, fields, market_aware, days = None, delta = None): + def __init__(self, fields, market_aware, days=None, delta=None): self.fields = fields self.market_aware = market_aware @@ -31,7 +31,7 @@ class MovingAverage(object): else: assert self.delta and not self.days, \ "Non-market-aware mode requires a timedelta." - + # No way to pass arguments to the defaultdict factory, so we # need to define a method to generate the correct EventWindows. self.sid_windows = defaultdict(self.create_window) @@ -41,12 +41,12 @@ class MovingAverage(object): Factory method for self.sid_windows. """ return MovingAverageEventWindow( - self.fields, - self.market_aware, - self.days, + self.fields, + self.market_aware, + self.days, self.delta ) - + def update(self, event): """ Update the event window for this event's sid. Return an ndict @@ -58,6 +58,7 @@ class MovingAverage(object): window.update(event) return window.get_averages() + class MovingAverageEventWindow(EventWindow): """ Iteratively calculates moving averages for a particular sid over a @@ -103,7 +104,7 @@ class MovingAverageEventWindow(EventWindow): # Averages are None by convention if we have no ticks. if len(self.ticks) == 0: return 0.0 - + # Calculate and return the average. len(self.ticks) is O(1). else: return self.totals[field] / len(self.ticks) @@ -112,7 +113,7 @@ class MovingAverageEventWindow(EventWindow): """ Return an ndict of all our tracked averages. """ - out = ndict() + out = ndict() for field in self.fields: out[field] = self.average(field) return out @@ -125,4 +126,5 @@ class MovingAverageEventWindow(EventWindow): assert event.has_key(field), \ "Event missing [%s] in MovingAverageEventWindow" % field assert isinstance(event[field], Number), \ - "Got %s for %s in MovingAverageEventWindow" % (event[field], field) + "Got %s for %s in MovingAverageEventWindow" % (event[field], + field) From d505fba14897d0488ecb8d50d224859218447369 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 12:42:21 -0400 Subject: [PATCH 48/54] Removes unused component module from core. This had provided a Component class that was used in a ZeroMQ based strategy for moving around data that his since been replaced by the generator model. --- tests/test_components.py | 384 ---------------------- zipline/core/__init__.py | 2 - zipline/core/component.py | 651 -------------------------------------- 3 files changed, 1037 deletions(-) delete mode 100644 tests/test_components.py delete mode 100644 zipline/core/component.py diff --git a/tests/test_components.py b/tests/test_components.py deleted file mode 100644 index a9320dde..00000000 --- a/tests/test_components.py +++ /dev/null @@ -1,384 +0,0 @@ -import zmq -import pytz -from pprint import pformat as pf -from datetime import datetime, timedelta - -from unittest2 import TestCase, skip -from collections import defaultdict -from zipline.gens.composites import date_sorted_sources, merged_transforms - -from zipline.core.devsimulator import AddressAllocator -from zipline.gens.transform import Passthrough, StatefulTransform -from zipline.gens.mavg import MovingAverage -from zipline.gens.tradesimulation import TradeSimulationClient as tsc - -from zipline.utils.factory import create_trading_environment -from zipline.test_algorithms import TestAlgorithm - - -from zipline.utils.test_utils import ( - setup_logger, - teardown_logger, - create_monitor, - launch_monitor -) - -from zipline.core import Component -from zipline.protocol import ( - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - FEED_FRAME, - FEED_UNFRAME, - MERGE_FRAME, - MERGE_UNFRAME, - SIMULATION_STYLE, - PERF_FRAME, - BT_UPDATE_UNFRAME -) - -from zipline.gens.tradegens import SpecificEquityTrades - -import logbook -log = logbook.Logger('ComponentTestCase') - -allocator = AddressAllocator(1000) - - -class ComponentTestCase(TestCase): - - leased_sockets = defaultdict(list) - - def setUp(self): - self.zipline_test_config = { - 'allocator' : allocator, - 'sid' : 133, - 'devel' : False, - 'results_socket' : allocator.lease(1)[0], - 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE - } - self.ctx = zmq.Context() - setup_logger(self) - - count = 250 - filter = [2,3] - #Set up source a. One minute between events. - args_a = tuple() - kwargs_a = { - 'count' : 2*count, - 'sids' : [1,2,3], - 'start' : datetime(2002,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(hours = 6), - 'filter' : filter - } - self.source_a = SpecificEquityTrades(*args_a, **kwargs_a) - - #Set up source b. Two minutes between events. - args_b = tuple() - kwargs_b = { - 'count' : count, - 'sids' : [2,3,4], - 'start' : datetime(2002,1,3,14, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 5), - 'filter' : filter - } - self.source_b = SpecificEquityTrades(*args_b, **kwargs_b) - - self.environment = create_trading_environment(year = 2002) - - - - def tearDown(self): - teardown_logger(self) - - @skip - def test_source(self): - monitor = create_monitor(allocator) - socket_uri = allocator.lease(1)[0] - count = 100 - - filter = [1,2,3,4] - #Set up source a. One minute between events. - args_a = tuple() - kwargs_a = { - 'sids' : [1,2], - 'start' : datetime(2012,6,6,0,tzinfo=pytz.utc), - 'delta' : timedelta(minutes = 1), - 'filter' : filter, - 'count' : count - } - - trade_gen = SpecificEquityTrades(*args_a, **kwargs_a) - - - comp_a = Component( - trade_gen, - monitor, - socket_uri, - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - "source_a" - ) - - mon_proc = launch_monitor(monitor) - - for event in comp_a: - log.info(event) - - # wait for the sending process to exit - comp_a.proc.join() - mon_proc.join() - - @skip - def test_sort(self): - monitor = create_monitor(allocator) - socket_uris = allocator.lease(3) - count = 100 - - filter = [1,2,3,4] - #Set up source a. One minute between events. - args_a = tuple() - kwargs_a = { - 'sids' : [1,2], - 'start' : datetime(2012,6,6,0,tzinfo=pytz.utc), - 'delta' : timedelta(minutes = 1), - 'filter' : filter, - 'count' : count - } - trade_gen_a = SpecificEquityTrades(*args_a, **kwargs_a) - - #Set up source b. Two minutes between events. - args_b = tuple() - kwargs_b = { - 'sids' : [2], - 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 1), - 'filter' : filter, - 'count' : count - } - trade_gen_b = SpecificEquityTrades(*args_b, **kwargs_b) - - #Set up source c. Three minutes between events. - args_c = tuple() - kwargs_c = { - 'sids' : [3], - 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), - 'delta' : timedelta(minutes = 1), - 'filter' : filter, - 'count' : count - } - - trade_gen_c = SpecificEquityTrades(*args_c, **kwargs_c) - - - comp_a = Component( - trade_gen_a, - monitor, - socket_uris[0], - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - trade_gen_a.get_hash() - ) - - comp_b = Component( - trade_gen_b, - monitor, - socket_uris[1], - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - trade_gen_b.get_hash() - ) - - comp_c = Component( - trade_gen_c, - monitor, - socket_uris[2], - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - trade_gen_c.get_hash() - ) - - sources = [comp_a, comp_b, comp_c] - - sorted_out = date_sorted_sources(*sources) - - mon_proc = launch_monitor(monitor) - - prev = None - sort_count = 0 - for msg in sorted_out: - if prev: - self.assertTrue(msg.dt >= prev.dt, \ - "Messages should be in date ascending order") - prev = msg - sort_count += 1 - - self.assertEqual(count*3, sort_count) - - # wait for processes to finish - comp_a.proc.join() - comp_b.proc.join() - comp_c.proc.join() - mon_proc.join() - - @skip - def test_full(self): - monitor = create_monitor(allocator) - - # ------------------------ - # Run sources in dedicated processes - comp_a = Component( - self.source_a, - monitor, - allocator.lease(1)[0], - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - self.source_a.get_hash() - ) - - comp_b = Component( - self.source_b, - monitor, - allocator.lease(1)[0], - DATASOURCE_FRAME, - DATASOURCE_UNFRAME, - self.source_b.get_hash() - ) - - # Date sort the sources, and run the sort in a dedicated - # process - sources = [comp_a, comp_b] - - sorted_out = date_sorted_sources(*sources) - - sorted = Component( - sorted_out, - monitor, - allocator.lease(1)[0], - FEED_FRAME, - FEED_UNFRAME, - "sort" - ) - - - passthrough = StatefulTransform(Passthrough) - mavg_price = StatefulTransform( - MovingAverage, - ['price'], - market_aware = False, - delta=timedelta(minutes = 20) - ) - - merged_gen = merged_transforms(sorted, passthrough, mavg_price) - - merged = Component( - merged_gen, - monitor, - allocator.lease(1)[0], - MERGE_FRAME, - MERGE_UNFRAME, - "merge" - ) - - algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3]) - - style = SIMULATION_STYLE.FIXED_SLIPPAGE - - trading_client = tsc(algo, self.environment, style) - tsc_gen = trading_client.simulate(merged) - - tsc_comp = Component( - tsc_gen, - monitor, - allocator.lease(1)[0], - PERF_FRAME, - BT_UPDATE_UNFRAME, - "tsc" - ) - mon_proc = launch_monitor(monitor) - for message in tsc_comp: - log.info(pf(message)) - - - # wait for processes to finish - comp_a.proc.join() - comp_b.proc.join() - sorted.proc.join() - merged.proc.join() - tsc_comp.proc.join() - mon_proc.join() - return - - def test_single_thread(self): - - #Set up source c. Three minutes between events. - - sorted = date_sorted_sources(self.source_a, self.source_b) - - passthrough = StatefulTransform(Passthrough) - mavg_price = StatefulTransform( - MovingAverage, - ['price'], - market_aware=False, - delta=timedelta(minutes = 20), - ) - - merged = merged_transforms(sorted, passthrough, mavg_price) - - algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3]) - style = SIMULATION_STYLE.FIXED_SLIPPAGE - - trading_client = tsc(algo, self.environment, style) - for message in trading_client.simulate(merged): - log.info(pf(message)) - - @skip - def test_compound(self): - monitor = create_monitor(allocator) - - sorted_out = date_sorted_sources(self.source_a, self.source_b) - - sorted = Component( - sorted_out, - monitor, - allocator.lease(1)[0], - FEED_FRAME, - FEED_UNFRAME, - "feed" - ) - - passthrough = StatefulTransform(Passthrough) - mavg_price = StatefulTransform( - MovingAverage, - ['price'], - market_aware = False, - delta=timedelta(minutes = 20) - ) - - merged_gen = merged_transforms(sorted, passthrough, mavg_price) - - merged = Component( - merged_gen, - monitor, - allocator.lease(1)[0], - MERGE_FRAME, - MERGE_UNFRAME, - "merge" - ) - - algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3]) - style = SIMULATION_STYLE.FIXED_SLIPPAGE - - trading_client = tsc(algo, self.environment, style) - tsc_gen = trading_client.simulate(merged) - - - mon_proc = launch_monitor(monitor) - for message in tsc_gen: - log.info(pf(message)) - - - # wait for processes to finish - sorted.proc.join() - merged.proc.join() - mon_proc.join() - return diff --git a/zipline/core/__init__.py b/zipline/core/__init__.py index 6bdbf960..ea179376 100644 --- a/zipline/core/__init__.py +++ b/zipline/core/__init__.py @@ -1,7 +1,5 @@ -from component import Component from monitor import Monitor __all__ = [ - Component, Monitor, ] diff --git a/zipline/core/component.py b/zipline/core/component.py deleted file mode 100644 index c9ce56e6..00000000 --- a/zipline/core/component.py +++ /dev/null @@ -1,651 +0,0 @@ -""" -Contains the base class for all components. -""" - -import os -import sys -import uuid -import time -import socket -import logbook -import humanhash -import multiprocessing -from setproctitle import setproctitle -from collections import namedtuple - - -# pyzmq -import zmq - -from zipline.core.monitor import PARAMETERS - -from zipline.protocol import ( - CONTROL_PROTOCOL, - COMPONENT_STATE, - CONTROL_FRAME, - CONTROL_UNFRAME, - EXCEPTION_FRAME -) - - -log = logbook.Logger('Component') - -class KillSignal(Exception): - def __init__(self): - pass - -class ShutdownSignal(Exception): - def __init__(self): - pass - -ComponentSocketArgs = namedtuple('ComponentSocketArgs',['uri','style','bind']) - -class Component(object): - - # ------------ - # Construction - # ------------ - - def __init__(self, - generator, - monitor, - socket_uri, - frame, - unframe, - component_id - ): - - # ----------------- - # Generator - # ----------------- - self.generator = generator - self.frame = frame - self.component_id = component_id - - # lock for waiting on monitor "GO" - self.waiting = None - - # ----------------- - # ZMQ properties - # ----------------- - self.in_socket_args = ComponentSocketArgs( - uri = socket_uri, - style = zmq.PULL, - bind = False - ) - self.out_socket_args = ComponentSocketArgs( - uri = socket_uri, - style = zmq.PUSH, - bind = True - ) - self.zmq = None - self.context = None - self.out_socket = None - self.in_socket = None - self.monitor = monitor - self.unframe = unframe - self.prefix = "" - - # TODO: state_flag is deprecated, remove - self.state_flag = COMPONENT_STATE.OK - - # track time of last ping we received from monitor - self.last_ping = time.time() - - # Humanhashes make this way easier to debug because they stick - # in your mind unlike a 32 byte string of random hex. - self.guid = uuid.uuid4() - self.huid = humanhash.humanize(self.guid.hex) - - # first, start the generator in its own process. Once - # Monitor says "go", Events from the generator will be - # FRAME'd and PUSH'd to self.socket_uri. - monitor.add_to_topology(self.component_id) - - self.proc = multiprocessing.Process( - target=self.loop_send - ) - self.proc.start() - - # Placeholder for receive generator, which will be - # created in __iter__ - self.recv_gen = None - - - # ------------ - # Core Methods - # ------------ - - def loop_send(self): - """ - The main component loop. This is wrapped inside a - exception reporting context inside of run. - - The core logic of the all components is run here. - """ - try: - # The process title so you can watch it in top, ps. - self.prefix = "FORK-" - setproctitle(self.get_id) - - log.info("Start %r" % self) - log.info("Pid %s" % os.getpid()) - log.info("Group %s" % os.getpgrp()) - - self.open() - - self.signal_ready() - self.lock_ready() - - msg = None - for event in self.generator: - - if hasattr(event, 'dt') and event.dt == 'DONE': - continue - - self.wait_ready() - - self.heartbeat() - msg = self.frame(event) - self.out_socket.send(msg) - - self.signal_done() - - # keep heartbeating until we receive the shutdown - # message from the Monitor (raises a - # ShutdownSignal), or we don't hear from the Monitor - # for MAX_COMPONENT_WAIT. - while True: - self.heartbeat(timeout=1000) - - except Exception as exc: - self.handle_exception(exc) - finally: - log.info("Exiting %r" % self) - - - def create_recv_gen(self): - try: - # return the generator - return self.loop_recv() - except Exception as exc: - self.handle_exception(exc) - finally: - log.info("Created Recv Gen for %r" % self) - - def loop_recv(self): - try: - self.open(send=False) - self.signal_ready() - self.lock_ready() - - # we block on ready here until monitor sends the GO - # self.wait_ready() - for event in self.gen_from_poller(self.poll, self.in_socket, self.unframe): - yield event - - self.signal_done() - except Exception as exc: - self.handle_exception(exc) - finally: - log.info("Exiting %r" % self) - - def gen_from_poller(self, poller, in_socket, unframe): - - while True: - # Since we will yield None to avoid blocking, we need - # to have a small delay to give the poller a chance - # to receive a message from upstream. - socks = dict(poller.poll(100)) - self.heartbeat() - if socks.get(in_socket) == zmq.POLLIN: - message = in_socket.recv() - if message == str(CONTROL_PROTOCOL.DONE): - break - else: - event = unframe(message) - yield event - else: - yield - - def handle_exception(self, exc, re_raise=False): - if isinstance(exc, KillSignal): - # if we get a kill signal, forcibly close all the - # sockets. - self.teardown_sockets() - elif isinstance(exc, ShutdownSignal): - # signal from monitor of an orderly shutdown, - # do nothing. - pass - else: - self.signal_exception(exc) - - def __iter__(self): - return self - - def next(self): - if not self.recv_gen: - self.recv_gen = self.create_recv_gen() - return self.recv_gen.next() - - # ---------------------------- - # Cleanup & Modes of Failure - # ---------------------------- - - def teardown_sockets(self): - """ - Close all zmq sockets safely. This is universal, no matter where - this is running it will need the sockets closed. - """ - log.warn("{id} closing all sockets".format(id=self.get_id)) - #close all the sockets - for sock in self.sockets: - sock.close() - - def shutdown(self): - """ - Clean shutdown. - """ - raise ShutdownSignal() - - def kill(self): - """ - Unclean shutdown. - - Tear down ( fast ) as a mode of failure in the simulation or on - service halt. - """ - raise KillSignal() - - def signal_exception(self, exc=None, scope=None): - """ - All exceptions inside any component should boil back to - this handler. - - Will inform the system that the component has failed and how it - has failed. - """ - self.state_flag = COMPONENT_STATE.EXCEPTION - exc_type, exc_value, exc_traceback = sys.exc_info() - - # if a downstream component fails, this component may try - # sending when there are zero connections to the socket, - # which will raise ZMQError(EAGAIN). So, it doesn't make - # sense to relay this exception to Monitor and the rest - # of the zipline. - if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN: - log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\ - .format(id=self.get_id)) - return - - # sys.stdout.write(trace) - log.exception("Unexpected error in run for {id}.".format(id=self.get_id)) - - try: - log.info('{id} sending exception to monitor'\ - .format(id=self.get_id)) - msg = EXCEPTION_FRAME( - exc_traceback, - exc_type.__name__, - exc_value.message - ) - - exception_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.EXCEPTION, - msg - ) - self.control_out.send(exception_frame, self.zmq.NOBLOCK) - # The monitor should relay the exception back - # to all zipline components. Wait here until the - # notice arrives, and we can assume other zipline - # components have broken out of their message - # loops. - for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT): - self.heartbeat(timeout=1000) - log.warn("{id} never heard back from monitor."\ - .format(id=self.get_id)) - - except KillSignal: - log.info("{id} received confirmation from monitor"\ - .format(id=self.get_id)) - except: - log.exception("Exception waiting for monitor reply") - - - - # ---------------------- - # Internal Maintenance - # ---------------------- - - def lock_ready(self): - """ - Unlock the component, topology is now ready to run. - """ - self.waiting = True - - def unlock_ready(self): - """ - Unlock the component, topology is still pending. - """ - self.waiting = False - - def wait_ready(self): - # Implicit side-effect of unlocking the component iff - # the GO message is received from the monitor level. - # This then unlocks the barrier and proceeds to the - # do_work state. - - # Poll on a subset of the control protocol while we exist - # in the locked quasimode. Respond to HEARTBEAT and GO - # messages. - - start_wait = time.time() - - while self.waiting: - socks = dict(self.poll.poll(0)) - - assert self.control_in, \ - 'Component does not have a control_in socket' - - if socks.get(self.control_in) == zmq.POLLIN: - - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) - - # ==== - # Go - # ==== - - # A distributed lock from the monitor to ensure - # synchronized start. - - if event == CONTROL_PROTOCOL.HEARTBEAT: - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - self.control_out.send(heartbeat_frame) - log.info('Prestart Heartbeat ' + self.get_id) - - elif event == CONTROL_PROTOCOL.GO: - # Side effectful call from the monitor to unlock - # and begin doing work only when the entire topology - # of the system beings to come online - log.info('Unlocking ' + self.get_id) - self.unlock_ready() - - # ========= - # Soft Kill - # ========= - - # Try and clean up properly and send out any reports or - # data that are done during a clean shutdown. Inform the - # monitor that we're done. - elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.shutdown() - break - - # ========= - # Hard Kill - # ========= - - # Just exit. - elif event == CONTROL_PROTOCOL.KILL: - self.kill() - break - - elif time.time() - start_wait > PARAMETERS.MAX_COMPONENT_WAIT: - log.info('No go signal from monitor, %s exiting' \ - % self.get_id) - self.kill() - break - - def heartbeat(self, timeout=0): - # wait for synchronization reply from the host - socks = dict(self.poll.poll(timeout)) - - # ---------------- - # Control Dispatch - # ---------------- - assert self.control_in, 'Component does not have a control_in socket' - - if socks.get(self.control_in) == zmq.POLLIN: - msg = self.control_in.recv() - event, payload = CONTROL_UNFRAME(msg) - - # =========== - # Heartbeat - # =========== - - # The monitor will send out a single number packed in - # a CONTROL_FRAME with ``heartbeat`` event every - # (n)-seconds. The component then has n seconds to - # respond to it. If not then it will be considered as - # malfunctioning or maybe CPU bound. - - if event == CONTROL_PROTOCOL.HEARTBEAT: - # Heart outgoing - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - payload - ) - - self.last_ping = float(payload) - # Echo back the heartbeat identifier to tell the - # monitor that this component is still alive and - # doing work - self.control_out.send(heartbeat_frame) - - - # ========= - # Soft Kill - # ========= - - # Try and clean up properly and send out any reports or - # data that are done during a clean shutdown. Inform the - # monitor that we're done. - elif event == CONTROL_PROTOCOL.SHUTDOWN: - self.shutdown() - - # ========= - # Hard Kill - # ========= - - # Just exit. - elif event == CONTROL_PROTOCOL.KILL: - self.kill() - - # In case we didn't receive a ping, send a pre-emptive - # pong to the monitor. - elif time.time() - self.last_ping > 2: - # send a ping ahead of schedule - pre_pong = time.time() - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.OK, - str(pre_pong) - ) - - # Echo back the heartbeat identifier to tell the - # monitor that this component is still alive and - # doing work - self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK) - self.last_ping = pre_pong - elif time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT: - # monitor is gone without sending the shutdown - # signal, do a hard exit. - self.kill() - - - def signal_ready(self): - log.info(self.get_id + ' is ready') - frame = CONTROL_FRAME( - CONTROL_PROTOCOL.READY, - '' - ) - self.control_out.send(frame) - - def signal_done(self): - """ - Notify down stream components that we're done. - """ - - self.state_flag = COMPONENT_STATE.DONE - # notify internal work loop that we're done - self.done = True # TODO: use state flag - - if self.out_socket: - msg = zmq.Message(str(CONTROL_PROTOCOL.DONE)) - self.out_socket.send(msg) - - - # notify monitor we're done - done_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.DONE, - '' - ) - - self.control_out.send(done_frame) - log.info("[%s] sent control done" % self.get_id) - - # ----------- - # Messaging - # ----------- - - def open(self, send=True): - """ - Open the connections needed to start doing work. - Perform any setup that must be done within process. - """ - self.sockets = [] - self.zmq = zmq - self.context = self.zmq.Context() - self.poll = self.zmq.Poller() - - self.setup_control() - - if send: - self.out_socket = self.open_socket(self.out_socket_args) - self.sockets.extend([self.out_socket]) - else: - self.in_socket = self.open_socket(self.in_socket_args) - self.sockets.extend([self.in_socket]) - - def open_socket(self, sock_args): - if sock_args.bind: - return self.bind_socket(sock_args) - else: - return self.connect_socket(sock_args) - - def bind_socket(self, sock_args): - if sock_args.style == zmq.PULL: - return self.bind_pull_socket(sock_args.uri) - if sock_args.style == zmq.PUSH: - return self.bind_push_socket(sock_args.uri) - if sock_args.style == zmq.PUB: - return self.bind_pub_socket(sock_args.uri) - - raise Exception("Invalid socket arguments") - - def connect_socket(self, sock_args): - if sock_args.style == zmq.PULL: - return self.connect_pull_socket(sock_args.uri) - if sock_args.style == zmq.PUSH: - return self.connect_push_socket(sock_args.uri) - if sock_args.style == zmq.SUB: - return self.connect_sub_socket(sock_args.uri) - - raise Exception("Invalid socket arguments") - - def bind_push_socket(self, addr): - push_socket = self.context.socket(self.zmq.PUSH) - push_socket.bind(addr) - self.sockets.append(push_socket) - - return push_socket - - def connect_pull_socket(self, addr): - pull_socket = self.context.socket(self.zmq.PULL) - pull_socket.connect(addr) - self.sockets.append(pull_socket) - self.poll.register(pull_socket, self.zmq.POLLIN) - - return pull_socket - - - def bind_pull_socket(self, addr): - pull_socket = self.context.socket(self.zmq.PULL) - pull_socket.bind(addr) - self.poll.register(pull_socket, self.zmq.POLLIN) - self.sockets.append(pull_socket) - - return pull_socket - - def connect_push_socket(self, addr): - push_socket = self.context.socket(self.zmq.PUSH) - push_socket.connect(addr) - self.sockets.append(push_socket) - - return push_socket - - - def setup_control(self): - """ - Set up the control socket. Used to monitor the overall status - of the simulation and to forcefully tear down the simulation in - case of a failure. - """ - self.control_out = self.monitor.message_sender( - identity = self.get_id, - context = self.context, - ) - - self.control_in = self.monitor.message_listener( - context = self.context - ) - - self.poll.register(self.control_in, self.zmq.POLLIN) - self.sockets.extend([self.control_in, self.control_out]) - - # --------------------- - # Description and Debug - # --------------------- - - @property - def get_id(self): - """ - The time invariant name for this component. - Must be unique within this zipline. - """ - return self.prefix + self.component_id - - def get_hash(self): - return self.component_id - - def debug(self): - """ - Debug information about the component. - """ - return { - 'id' : self.get_id , - 'huid' : self.huid , - 'host' : socket.gethostname() , - 'pid' : os.getpid() , - 'memaddress' : hex(id(self)) , - 'ready' : self.successful() , - 'successful' : self.ready() , - } - - def __repr__(self): - """ - Return a useful string representation of the component to - indicate its type, unique identifier, and computational context - identifier name. - """ - - return "<{name} {uuid} at {host} {pid} {pointer}>".format( - name = self.get_id , - uuid = self.guid , - host = socket.gethostname() , - pid = os.getpid() , - pointer = hex(id(self)) , - ) From 98f4e4c76840930358bc4e4deedc1c05d7e8cc4b Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 12:53:12 -0400 Subject: [PATCH 49/54] Applies PEP-8 recommendations. White space tweaking. --- zipline/gens/returns.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/zipline/gens/returns.py b/zipline/gens/returns.py index a4932485..a407a1e0 100644 --- a/zipline/gens/returns.py +++ b/zipline/gens/returns.py @@ -1,6 +1,7 @@ from zipline.gens.transform import TransformMeta from collections import defaultdict, deque + class Returns(object): """ Class that maintains a dictionary from sids to the sid's @@ -16,8 +17,8 @@ class Returns(object): """ Update and return the calculated returns for this event's sid. """ - assert event.has_key('dt') - assert event.has_key('price') + assert 'dt' in event + assert 'price' in event tracker = self.mapping[event.sid] tracker.update(event) @@ -26,6 +27,7 @@ class Returns(object): def _create(self): return ReturnsFromPriorClose(self.days) + class ReturnsFromPriorClose(object): """ Records the last N closing events for a given security as well as the @@ -72,6 +74,5 @@ class ReturnsFromPriorClose(object): change = event.price - last_close self.returns = change / last_close - # the current event is now the last_event self.last_event = event From d90c53167960626841414869f3398f4f21481c98 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 15:24:30 -0400 Subject: [PATCH 50/54] Removes core.monitor module. The Monitor class that it provided was an artifact of a version that depended on ZeroMQ as a communication layer instead of the current generator bassed model. Also, removes tests that were exercising the Monitor. --- tests/test_monitor.py | 28 -- zipline/core/__init__.py | 4 - zipline/core/monitor.py | 654 ------------------------------------ zipline/utils/test_utils.py | 20 -- 4 files changed, 706 deletions(-) delete mode 100644 tests/test_monitor.py delete mode 100644 zipline/core/monitor.py diff --git a/tests/test_monitor.py b/tests/test_monitor.py deleted file mode 100644 index fa5de2b8..00000000 --- a/tests/test_monitor.py +++ /dev/null @@ -1,28 +0,0 @@ -from zipline.utils.test_utils import setup_logger, teardown_logger -from unittest2 import TestCase - -from zipline.core.monitor import Monitor - - -class TestMonitor(TestCase): - def setUp(self): - setup_logger(self, '/var/log/qexec/qexec.log') - - def tearDown(self): - teardown_logger(self) - - def test_init(self): - pub_socket = 'tcp://127.0.0.1:5000' - route_socket = 'tcp://127.0.0.1:5001' - exception_socket = 'tcp://127.0.0.1:5002' - - mon = Monitor(pub_socket, route_socket, exception_socket) - mon.manage([]) - - def test_init_topology(self): - pub_socket = 'tcp://127.0.0.1:5000' - route_socket = 'tcp://127.0.0.1:5001' - exception_socket = 'tcp://127.0.0.1:5002' - - mon = Monitor(pub_socket, route_socket, exception_socket) - mon.manage(['a', 'b', 'c', 'd']) diff --git a/zipline/core/__init__.py b/zipline/core/__init__.py index ea179376..8b137891 100644 --- a/zipline/core/__init__.py +++ b/zipline/core/__init__.py @@ -1,5 +1 @@ -from monitor import Monitor -__all__ = [ - Monitor, -] diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py deleted file mode 100644 index acc80132..00000000 --- a/zipline/core/monitor.py +++ /dev/null @@ -1,654 +0,0 @@ -import os -import zmq -import sys -import time -import itertools -import logbook -from setproctitle import setproctitle -from signal import SIGHUP, SIGINT - -from collections import Counter - -from zipline.protocol import ( - CONTROL_PROTOCOL, - CONTROL_FRAME, - CONTROL_UNFRAME, - CONTROL_STATES, - INVALID_CONTROL_FRAME -) - -from zipline.utils.protocol_utils import ndict - -INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES - -CONTROLLER_TRANSITIONS = frozenset([ - (-1 , INIT), - (INIT , SOURCES_READY), - (SOURCES_READY , RUNNING), - - (INIT , TERMINATE), # pseudo failure mode - (SOURCES_READY , TERMINATE), # pseudo failure mode - (RUNNING , TERMINATE), -]) - -class UnknownChatter(Exception): - def __init__(self, name): - self.named = name - def __str__(self): - return """Component calling itself "%s" talking on unexpected channel""" % self.named - - -log = logbook.Logger('Monitor') - -# The scalars determining the timing of the monitor behavior for -# the system. - -PARAMETERS = ndict(dict( - # time Monitor will wait for a heartbeat, in seconds - GENERATIONAL_PERIOD = 20, - # time Component will wait for GO and for a heartbeat before - # timing out. - MAX_COMPONENT_WAIT = 25, - ALLOWED_SKIPPED_HEARTBEATS = 10, - ALLOWED_INVALID_HEARTBEATS = 3, - PRESTART_HEARBEATS = 3, - SOURCES_START_HEARTBEATS = 3, - SYSTEM_TIMEOUT = 50, -)) - -class Monitor(object): - """ - A N to M messaging system for inter component communication. - - :param pub_socket: Socket to publish messages, the starting - point of :func message_listener: . - - :param route_socket: Socket to listen for status updates for - the individual components. - :func message_sender: . - - """ - - # Turn on debug for verbose logging of the system. - debug = True - period = PARAMETERS.GENERATIONAL_PERIOD - - def __init__( - self, - pub_socket, - route_socket, - exception_socket, - send_sighup=False): - - self.nosignals = False - self.context = None - self.zmq = None - self.zmq_poller = None - - self.running = False - self.alive = False - self.tracked = set() - self.finished = set() - - self.responses = set() - - self.ctime = 0 - self.tic = time.time() - self.freeform = False - self._state = -1 - - self.associated = [] - - self.pub_socket = pub_socket - self.route_socket = route_socket - self.exception_socket = exception_socket - - self.missed_beats = Counter() - - # start with an empty topology - self.topology = set([]) - - self.send_sighup = send_sighup - if self.send_sighup: - log.info("Request to send sighup/sigint") - - - def init_zmq(self): - self.zmq = zmq - self.context = self.zmq.Context() - self.zmq_poller = self.zmq.Poller - return - - def add_to_topology(self, component_id): - add = set([component_id, "FORK-" + component_id]) - self.topology.update(add) - - def freeze_topology(self): - if isinstance(self.topology, frozenset): - return - # we've been incrementally adding components. - # time to freeze. - self.manage(self.topology) - - def manage(self, topology): - """ - Give the controller a set set of components to manage and - a set of state transitions for the entire system. - """ - # A freeform topology is where we heartbeat with anything - # that shows up. - if topology == 'freeform': - self.freeform = True - self.topology = frozenset([]) - else: - self.freeform = False - self.topology = frozenset(topology) - self.alive = True - - @property - def state(self): - #log.info('returned %s' % self._state) - return self._state - - @state.setter - def state(self, new): - old = self._state - - if (old, new) in CONTROLLER_TRANSITIONS: - self._state = new - log.info("State Transition : %s -> %s" % (old, self._state)) - else: - raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new)) - - def run(self): - self.freeze_topology() - self.running = True - self.init_zmq() - setproctitle('Monitor') - - self.state = CONTROL_STATES.INIT - - # TODO: keep the exitfunc? the corresponding override on clean - # exit is commented out currently. - # - # Interpreter SIDE EFFECT - # ----------------------- - # The last breathe of the interpreter will assume that we've - # failed unless we specify otherwise. - log.info('registering exit function') - sys.exitfunc = self.signal_interrupt - # We overload this if ( and only if ) the topology exits - # cleanly. This prevents failure modes where the monitor - # dies. - - try: - return self._poll() # use a python loop - except KeyboardInterrupt: - log.info('Shutdown event loop') - - def log_status(self): - """ - Snapshot of the tracked components at every period. - """ - #log.info("Tracking component : %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 - # ------------- - - def send_go(self): - go_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.GO, - '' - ) - self.pub.send(go_frame) - - def send_heart(self): - if not self.running: - return - - heartbeat_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.HEARTBEAT, - str(self.ctime) - ) - self.pub.send(heartbeat_frame) - - def send_hardkill(self): - if not self.running: - return - - kill_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.KILL, - '' - ) - self.pub.send(kill_frame) - - def send_softkill(self): - if not self.running: - return - - soft_frame = CONTROL_FRAME( - CONTROL_PROTOCOL.SHUTDOWN, - '' - ) - self.pub.send(soft_frame) - - # ----------- - # Event Loops - # ----------- - - def _poll(self): - - assert self.route_socket - assert self.pub_socket - - assert self.topology,\ - """"Must define topology to monitor, call setup_controller() on - your Zipline. """ - - # -- Publish -- - # ============= - self.pub = self.context.socket(self.zmq.PUB) - self.pub.bind(self.pub_socket) - self.pub.setsockopt(zmq.LINGER, 0) - - # -- Router -- - # ============= - self.router = self.context.socket(self.zmq.ROUTER) - self.router.bind(self.route_socket) - self.router.setsockopt(zmq.LINGER, 0) - - # -- Exception Out -- - # =================== - self.ex_out = self.context.socket(self.zmq.PUSH) - self.ex_out.connect(self.exception_socket) - - poller = self.zmq.Poller() - poller.register(self.router, self.zmq.POLLIN) - #poller.register(self.cancel, self.zmq.POLLIN) - - self.associated += [self.pub, self.router] - - # TODO: actually do this - self.state = CONTROL_STATES.SOURCES_READY - self.state = CONTROL_STATES.RUNNING - - buffer = [] - - # =================== - # Heartbeat Iteration - # =================== - - for i in itertools.count(0): - self.log_status() - - # Reset the responses for this cycle - self.responses = set() - - # broadcast the heartbeat packet - self.ctime = time.time() - self.send_heart() - - # ============== - # Hearbeat Cycle - # ============== - - initializing = len(self.tracked) == 0 and len(self.finished) == 0 - - # Wait the responses - while self.alive: - - socks = dict(poller.poll(0)) - tic = time.time() - - if socks.get(self.router) == self.zmq.POLLIN: - rawmessage = self.router.recv() - - if rawmessage: - buffer.append(rawmessage) - try: - if not self.router.getsockopt(self.zmq.RCVMORE): - self.handle_recv(buffer[:]) - buffer = [] - - except INVALID_CONTROL_FRAME: - log.error('Invalid frame', rawmessage) - pass - - # We break out of this loop if the time between - # sending and receiving the heartbeat is more - # than our poll period. - - if tic - self.ctime > self.period: - log.info("heartbeat loop timedout: %s" % (tic - self.ctime)) - log.info(repr(self.responses)) - break - - # if this is the first time heartbeating, break - # out early if we get everything tracked no need - # to hold out for the full heartbeat. - if initializing and not self.freeform: - if len(self.responses) == len(self.topology): - log.info("breaking out of initial heartbeat") - break - - # Break out if the entire topology told us its DONE - if len(self.finished) == len(self.topology): - break - - - # ================ - # Heartbeat Stats - # ================ - - complete = self.beat() - - # ================ - # Topology Status - # ================ - - # Has the entire topology told us its DONE - done = len(self.finished) == len(self.topology) - - # Has the entire topology shown up to the party - complete = len(self.tracked) == len(self.topology) - - if complete: - self.send_go() - - log.info('Heartbeat (%s, %s)' % (done, complete)) - - # ================ - # Exit Strategies - # ================ - - # Will also fall out of loop when done, if using - # non-freeform topology - if done: - log.info('Entire topology exited cleanly') - self.shutdown() - - # Noop exit func - #sys.exitfunc = lambda: None - - # Send SIGHUP to burrito - self.signal_hangup() - - if not self.alive: - log.info('Breaking out of Monitor Loop') - break - - def signal_hangup(self): - """ - A clean exit, inform the burrito ( and arbiter ) that - we're good. The topology exited cleanly and we can prove - it. - """ - if not self.send_sighup: - log.warning("Skipping SIGHUP") - return - ppid = os.getppid() - log.warning("Sending SIGHUP") - os.kill(ppid, SIGHUP) - - def signal_interrupt(self): - """ - Send a SIGINT in the error mode that the monitor's - interpreter exits. If the monitor dies the system is - considered a failure. - """ - if not self.send_sighup: - log.warning("Skipping SIGINT") - return - ppid = os.getpid() - log.warning("Sending SIGINT") - os.kill(ppid, SIGINT) - - def beat(self): - """ - The tracking logic of the system. It's the "stethoscope" - that inspects to the heartbeats in a generation and - infers the state of the system from the responses. - """ - - # These the set overloaded operations - # A & B ~ set.intersection - # A - B ~ set.difference - - # * good - Components we are currently tracking and who just sent - # us back the right response. - # * bad - Components we are currently tracking but who did not - # send us back a response. - # * new - Components we haven't heard from yet, but sent back the - # right response. - # * finished - Components we were tracking but have now - # finished, when this set goes to zero this - # triggers the end of the topology. - - good = self.tracked & self.responses - bad = self.tracked - good - self.finished - new = self.responses - good - self.finished - - for component in new: - self.new(component) - - for component in bad: - self.timed_out(component) - - missing = self.topology - self.tracked - self.finished - - for component in missing: - if self.debug: - log.info('Missing component %r' % component) - - - for component in self.tracked: - if component not in self.topology: - log.info('Uninvited component %r' % component) - - # -------------- - # Init Handlers - # -------------- - - def new_universal(self): - pass - - # The various "states of being that a component can inform us - # of - def new(self, component): - if self.state is CONTROL_STATES.TERMINATE: - return - - if component in self.finished: - #log.info("Got heartbeat from supposedly finished component") - return - - log.info('Now Tracking "%s" ' % component) - - universal = self.new_universal - init_handlers = {} - - if component in (self.topology - self.finished) or self.freeform: - init_handlers.get(component, universal)() - self.tracked.add(component) - else: - # Some sort of socket collision has occurred, this is - # a very bad failure mode. - raise UnknownChatter(component) - - # ------------------ - # Epic Fail Handling - # ------------------ - - def timed_out(self, component): - if self.state is CONTROL_STATES.TERMINATE: - return - - if component in (self.topology - self.finished) or self.freeform: - log.warning('Component "%s" missed heartbeat' % component) - # we treat a time out as a severe failure, and - # conduct a rapid shutdown - self.kill() - - - # ------------------- - # Completion Handling - # ------------------- - - def done(self, component): - self.finished.add(component) - self.tracked.discard(component) - log.info('Component "%s" finished.' % component) - - # -------------- - # Error Handling - # -------------- - def exception(self, component, exception_data): - log.error('Component in exception state: %s. Shutting down system and sending exception data to listeners.'\ - % component) - # Send the exception message out to listeners. - self.ex_out.send(exception_data) - # An exception in one component is treated as a hard - # failure, and we conduct a rapid shutdown. - self.kill() - - # ----------------- - # Protocol Handling - # ----------------- - - def handle_recv(self, msg): - """ - Check for proper framing at the transport layer. - Seperates the proper frames from anything else that might - be coming over the wire. - """ - - identity = msg[0] # identity of the socket - id, status = CONTROL_UNFRAME(msg[1]) - - # I'm alive, condemned to be a free process in the cold - # cold dark absurd Zipline universe. - if id is CONTROL_PROTOCOL.READY: - self.responses.add(identity) - return - - # The heartbeat love song between a component and the - # controller - if id is CONTROL_PROTOCOL.OK: - - if status == str(self.ctime): - # Go to your bosom; knock there, and ask your heart what - # it doth know... - self.responses.add(identity) - elif float(status) < self.ctime: - # False face must hide what the false heart doth know. - log.warning('Delayed heartbeat received: %s' % msg) - elif float(status) > self.ctime: - # Pre-emptive heartbeat from the component - # log.info("pre-emptive pong: %s" % msg) - self.responses.add(identity) - else: - # Otherwise its something weird and we don't know - # what to do so just say so, probably line noise - # from ZeroMQ - - # What's in a name? that which we call a rose... - log.error("Weird heartbeat packet happened: %s" % msg) - return - - # A component is telling us it failed, and how - if id is CONTROL_PROTOCOL.EXCEPTION: - # status should be a msgpack emitted from - # EXCEPTION_FRAME - try: - exception_data = status - self.exception(identity, exception_data) - except: - # if an exception occurs when we try to handle - # the exception, signal the parent that we need - # to go down - # TODO: should we attempt to call self.exception? - log.exception("Unexpected exception sending exception data") - self.kill() - - return - - # A component is telling us its done with work and won't - # be talking to us anymore - if id is CONTROL_PROTOCOL.DONE: - self.done(identity) - return - - # ------------------- - # Hooks for Endpoints - # ------------------- - - # These are all connects so no complex allocation logic is - # needed. Dealers and Subscribers can all come and go as a - # function of time without impacting flow of the whole - # system. - - def message_sender(self, identity, context = None): - """ - Spin off a socket used for sending messages to this - controller. - """ - - if not context: - context = self.zmq.Context.instance() - - s = context.socket(zmq.DEALER) - s.setsockopt(zmq.IDENTITY, identity) - s.connect(self.route_socket) - - self.associated.append(s) - return s - - def message_listener(self, context = None): - """ - Spin off a socket used for receiving messages from this - controller. - """ - - if not context: - context = self.zmq.Context.instance() - - s = context.socket(zmq.SUB) - s.connect(self.pub_socket) - s.setsockopt(zmq.SUBSCRIBE, '') - - self.associated.append(s) - return s - - def kill(self): - """Aggressively exit the whole zipline. - """ - if self.state is CONTROL_STATES.TERMINATE: - return - - log.info('Hard Shutdown') - self.send_hardkill() - self.state = CONTROL_STATES.TERMINATE - self.alive = False - # send burrito an interrupt, instructing it to kill all - # child processes assocated with this zipline. - time.sleep(3) - self.signal_interrupt() - - def shutdown(self): - - if self.state is CONTROL_STATES.TERMINATE: - return - - log.info('Soft Shutdown') - self.send_softkill() - self.state = CONTROL_STATES.TERMINATE - self.alive = False diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 9b01d87c..513352b1 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -7,7 +7,6 @@ import blist from zipline.utils.date_utils import EPOCH from itertools import izip from logbook import FileHandler -from zipline.core.monitor import Monitor def setup_logger(test, path='/var/log/zipline/zipline.log'): @@ -167,25 +166,6 @@ def launch_monitor(monitor): return proc -def create_monitor(allocator): - sockets = allocator.lease(3) - mon = Monitor( - # pub socket - sockets[0], - # route socket - sockets[1], - # exception socket to match tradesimclient's result - # socket, because we want to relay exceptions to the - # same listener - sockets[2], - # this controller is expected to run in a test, so no - # need to signal the parent process on success or error. - send_sighup=False - ) - - return mon - - class ExceptionSource(object): def __init__(self): From a4b19f66cbe8e98bcefcc83451e32aa32fb5a54c Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Wed, 5 Sep 2012 16:36:27 -0400 Subject: [PATCH 51/54] Removes unused, deprecated `logger` module. --- tests/test_logger.py | 59 ----------------------------------------- zipline/utils/logger.py | 52 ------------------------------------ 2 files changed, 111 deletions(-) delete mode 100644 tests/test_logger.py delete mode 100644 zipline/utils/logger.py diff --git a/tests/test_logger.py b/tests/test_logger.py deleted file mode 100644 index 7a2ae780..00000000 --- a/tests/test_logger.py +++ /dev/null @@ -1,59 +0,0 @@ -import logging -import logbook -import uuid -import zmq - -from zipline import ndict - -from zipline.utils.logger import configure_logging, tail -from zipline.utils.log_utils import ZeroMQLogHandler - -from zipline.utils.test_utils import create_receiver, drain_receiver - -from unittest2 import TestCase - - -class LoggerTestCase(TestCase): - - def setUp(self): - configure_logging() - self.LOG = logging.getLogger("ZiplineLogger") - - def test_log(self): - test_msg = uuid.uuid1().hex - self.LOG.info(test_msg) - logfile = open('/var/log/zipline/zipline.log', 'r') - with logfile: - last_line = tail(logfile, window=1) - logged_msg = last_line.split(" - ")[1] - self.assertEqual(test_msg, logged_msg) - - def test_zmq_handler(self): - socket_addr = 'tcp://127.0.0.1:10000' - ctx = zmq.Context() - socket_push = ctx.socket(zmq.PUSH) - socket_push.connect(socket_addr) - recv = create_receiver(socket_addr, ctx) - zmq_out = ZeroMQLogHandler( - socket=socket_push, - filter=lambda r, h: r.channel in ['test zmq logger'], - context=ctx, - #bubble=False - ) - - log = logbook.Logger('test zmq logger') - x = ndict({}) - x.a = 1 - ex = example(133) - with zmq_out.threadbound(): - log.info(ex.num) - - output, _ = drain_receiver(recv, count=1) - self.assertEqual(output[-1]['prefix'], 'LOG') - self.assertTrue(isinstance(output[-1]['payload']['msg'], basestring)) - - -class example(object): - - def __init__(self, num): - self.num = num diff --git a/zipline/utils/logger.py b/zipline/utils/logger.py deleted file mode 100644 index b044642f..00000000 --- a/zipline/utils/logger.py +++ /dev/null @@ -1,52 +0,0 @@ -""" -Small classes to assist with timezone calculations, LOGGER configuration, -and other common operations. -""" - -# DEPRECATED DO NOT USE - -import logging -import logging.config -from os.path import join, abspath, dirname - -def configure_logging(): - logging.config.fileConfig( - logger_path(), - disable_existing_loggers = False - ) - -def logger_path(): - import zipline - log_path = dirname(abspath(zipline.__file__)) - return join(log_path, 'logging.cfg') - - -# utility for tailing a log file. -def tail( f, window=20 ): - """ - from - http://stackoverflow.com/questions/136168/get-last-n-lines-of-a-file- \ - with-python-similar-to-tail - """ - BUFSIZ = 1024 - f.seek(0, 2) - bytes = f.tell() - size = window - block = -1 - data = [] - while size > 0 and bytes > 0: - if (bytes - BUFSIZ > 0): - # Seek back one whole BUFSIZ - f.seek(block*BUFSIZ, 2) - # read BUFFER - data.append(f.read(BUFSIZ)) - else: - # file too small, start from begining - f.seek(0,0) - # only read what was not read - data.append(f.read(bytes)) - linesFound = data[-1].count('\n') - size -= linesFound - bytes -= BUFSIZ - block -= 1 - return '\n'.join(''.join(data).splitlines()[-window:]) From d1c490204820769b6d20c7eec4435a806ad07810 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 6 Sep 2012 02:35:37 -0400 Subject: [PATCH 52/54] Removes unused CancelSignal. This was only triggered by the now removed Monitor. --- zipline/lines.py | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/zipline/lines.py b/zipline/lines.py index 5283756f..75c473f0 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -83,11 +83,6 @@ import zipline.protocol as zp log = Logger('Lines') -class CancelSignal(Exception): - def __init__(self): - pass - - class SimulatedTrading(object): def __init__(self, @@ -184,12 +179,7 @@ class SimulatedTrading(object): os.kill(ppid, SIGINT) def handle_exception(self, exc): - if isinstance(exc, CancelSignal): - # signal from monitor of an orderly shutdown, - # do nothing. - pass - else: - self.signal_exception(exc) + self.signal_exception(exc) def signal_exception(self, exc=None): """ From 680f86ceb3763001afcf6e5f4d90f99e8864d6ea Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 6 Sep 2012 02:50:11 -0400 Subject: [PATCH 53/54] Removes unused imports. --- zipline/gens/transform.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 29c2e25e..44b9bdc4 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -2,19 +2,16 @@ Generator versions of transforms. """ import types -import pytz import logbook from copy import deepcopy -from datetime import datetime, timedelta -from collections import deque, defaultdict -from numbers import Number +from datetime import datetime +from collections import deque from abc import ABCMeta, abstractmethod from zipline import ndict from zipline.utils.tradingcalendar import non_trading_days -from zipline.gens.utils import assert_sort_unframe_protocol, \ - assert_transform_protocol, hash_args +from zipline.gens.utils import assert_sort_unframe_protocol, hash_args log = logbook.Logger('Transform') From d6ddc1390daeda29addbdd1379e572a6d5a74571 Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 6 Sep 2012 02:51:02 -0400 Subject: [PATCH 54/54] Ensures that EventWindow is a new style object. --- zipline/gens/transform.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index 44b9bdc4..03eb41ea 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -150,7 +150,7 @@ class StatefulTransform(object): log.info('Finished StatefulTransform [%s]' % self.get_hash()) -class EventWindow: +class EventWindow(object): """ Abstract base class for transform classes that calculate iterative metrics on events within a given timedelta. Maintains a list of