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/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) 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.""" 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. 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 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!") diff --git a/zipline/toys/__init__.py b/zipline/toys/__init__.py deleted file mode 100644 index e69de29b..00000000 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 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') 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') 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) - ])