mirror of
https://github.com/wassname/catalyst.git
synced 2026-07-18 12:20:12 +08:00
Merge pull request #110 from quantopian/remove-dead-code
Removes seemingly unused modules
This commit is contained in:
Eddie Hebert
+1
-72
@@ -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
|
||||
# ======
|
||||
|
||||
@@ -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)
|
||||
@@ -1,5 +0,0 @@
|
||||
from utils.exception_utils import CustomException
|
||||
|
||||
class ComponentNoInit(CustomException):
|
||||
argmap = ('classname',)
|
||||
message = """Class {classname} does not define an init method."""
|
||||
@@ -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)
|
||||
@@ -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.
|
||||
@@ -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)
|
||||
@@ -1,3 +0,0 @@
|
||||
from libc.stdio cimport printf
|
||||
|
||||
printf("Hello World!")
|
||||
@@ -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
|
||||
@@ -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')
|
||||
@@ -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')
|
||||
@@ -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)
|
||||
])
|
||||
Reference in New Issue
Block a user