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"; diff --git a/etc/requirements.txt b/etc/requirements.txt index bef0ebf6..a620712d 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -1,16 +1,11 @@ msgpack-python==0.1.12 humanhash==0.0.1 -ujson==1.18 iso8601==0.1.4 # ZeroMQ pyzmq==2.1.11 gevent-zeromq==0.2.2 -# Packaging -distribute==0.6.27 -setuptools==0.6c11 - # Unix setproctitle==1.1.6 @@ -18,4 +13,3 @@ setproctitle==1.1.6 Logbook==0.3 blist==1.3.4 -psutil==0.4.1 diff --git a/etc/requirements_sci.txt b/etc/requirements_sci.txt index a0104713..f138534c 100644 --- a/etc/requirements_sci.txt +++ b/etc/requirements_sci.txt @@ -13,8 +13,8 @@ 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 +scikit-learn==0.11 # ZeroMQ pyzmq==2.1.11 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 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/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/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 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 diff --git a/tests/test_logger.py b/tests/test_logger.py deleted file mode 100644 index 5a9b8e31..00000000 --- a/tests/test_logger.py +++ /dev/null @@ -1,62 +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/tests/test_monitor.py b/tests/test_monitor.py deleted file mode 100644 index 76bb6184..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, 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' - 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/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) diff --git a/tests/test_sorting.py b/tests/test_sorting.py index bec97e31..63455bbd 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): @@ -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}) diff --git a/tests/test_transforms.py b/tests/test_transforms.py index 8e203664..d617ce20 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 @@ -156,8 +153,7 @@ class FinanceTransformsTestCase(TestCase): def test_vwap(self): - vwap = StatefulTransform( - VWAP, + vwap = VWAP( market_aware = False, delta = timedelta(days = 2) ) @@ -180,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] @@ -221,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), @@ -263,8 +258,7 @@ class FinanceTransformsTestCase(TestCase): self.trading_environment ) - stddev = StatefulTransform( - MovingStandardDev, + stddev = MovingStandardDev( market_aware = False, delta = timedelta(minutes = 150), ) 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' -) diff --git a/zipline/core/__init__.py b/zipline/core/__init__.py index 6bdbf960..8b137891 100644 --- a/zipline/core/__init__.py +++ b/zipline/core/__init__.py @@ -1,7 +1 @@ -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)) , - ) diff --git a/zipline/core/devsimulator.py b/zipline/core/devsimulator.py index cbce4b66..6e273f8c 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') @@ -14,6 +13,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 +28,9 @@ 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 diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py deleted file mode 100644 index 3736b7e3..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 buritto - 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/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/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 diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 27be9148..ef97bfd1 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -1,13 +1,11 @@ -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 + def date_sorted_sources(*sources): """ Takes an iterable of sources, generating namestrings and @@ -31,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 @@ -45,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] @@ -69,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. @@ -81,28 +81,6 @@ 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, - stream_in) - - dt_aliased = alias_dt(stream_out) - return add_done(dt_aliased) - -def sequential_transforms_dict(stream_in, transforms): - """ - Apply each transform in transforms sequentially to each event in stream_in. - Each transform application will add a new entry indexed to the transform's - hash string. - """ - - assert isinstance(transforms, dict) - - for tnfm in transforms.itervalues(): - tnfm.forward_all = False - tnfm.update_in_place = False - tnfm.append_value = True # Recursively apply all transforms to the stream. stream_out = reduce(lambda stream, tnfm: tnfm.transform(stream), @@ -121,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')]) diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py deleted file mode 100644 index 21d9cde4..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 date_sorted_sources, merged_transforms, sequential_transforms -from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.mavg import MovingAverage -from zipline.gens.transform import 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) - diff --git a/zipline/gens/mavg.py b/zipline/gens/mavg.py index 21aa0bd0..de572236 100644 --- a/zipline/gens/mavg.py +++ b/zipline/gens/mavg.py @@ -1,9 +1,9 @@ from numbers import Number -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,8 +12,9 @@ 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): + def __init__(self, fields, market_aware, days=None, delta=None): self.fields = fields self.market_aware = market_aware @@ -30,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) @@ -40,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 @@ -57,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 @@ -102,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) @@ -111,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 @@ -124,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) diff --git a/zipline/gens/returns.py b/zipline/gens/returns.py index 49d3e9b5..a407a1e0 100644 --- a/zipline/gens/returns.py +++ b/zipline/gens/returns.py @@ -1,10 +1,14 @@ +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 closing price N trading days ago. """ + __metaclass__ = TransformMeta + def __init__(self, days): self.days = days self.mapping = defaultdict(self._create) @@ -13,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) @@ -23,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 @@ -69,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 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/tradegens.py b/zipline/gens/tradegens.py index 09685cba..e7c4e375 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -214,36 +214,4 @@ class DataFrameSource(SpecificEquityTrades): # Return the filtered event stream. - return _generator() - - -# !!!!!!! Deprecated for now !!!!!!!!! - -def RandomEquityTrades(object): - - def __init__(self): - # We shouldn't get any positional args. - assert args == () - - self.count = config.get('count', 500) - self.sids = config.get('sids', [1,2]) - self.filter = config.get('filter') - - dates = fuzzy_dates(count) - prices = mock_prices(count, rand = True) - volumes = mock_volumes(count, rand = True) - sids = cycle(sids) - - arg_gen = izip(sids, prices, volumes, dates) - - unfiltered = (create_trade(*args) for args in arg_gen) - - if filter: - filtered = ifilter(lambda event: event.sid in filter, unfiltered) - else: - filtered = unfiltered - return filtered - -# if __name__ == "__main__": -# import nose.tools; nose.tools.set_trace() -# trades = SpecificEquityTrades(filter = [1]) + return _generator() \ No newline at end of file diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 2f60c7e0..fecf994d 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 diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index c19501ee..03eb41ea 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') @@ -29,6 +26,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 +67,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 @@ -126,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 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()) diff --git a/zipline/gens/vwap.py b/zipline/gens/vwap.py index 5a0947d8..9d908b46 100644 --- a/zipline/gens/vwap.py +++ b/zipline/gens/vwap.py @@ -1,20 +1,21 @@ from numbers import Number -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 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. 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/lines.py b/zipline/lines.py index 9f2a8153..cba2d094 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -68,26 +68,21 @@ from setproctitle import setproctitle from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import SIMULATION_STYLE -from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe +from zipline.utils.log_utils import ZeroMQLogHandler from zipline.utils import factory -from zipline.test_algorithms import TestAlgorithm - -from zipline.gens.composites import \ - date_sorted_sources, merged_transforms, sequential_transforms -from zipline.gens.transform import Passthrough, StatefulTransform +from zipline.gens.composites import ( + date_sorted_sources, + sequential_transforms +) from zipline.gens.tradesimulation import TradeSimulationClient as tsc -from logbook import Logger, NestedSetup, Processor +from logbook import Logger import zipline.protocol as zp log = Logger('Lines') -class CancelSignal(Exception): - def __init__(self): - pass - class SimulatedTrading(object): def __init__(self, @@ -103,7 +98,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 @@ -150,7 +146,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) @@ -171,6 +167,8 @@ class SimulatedTrading(object): def close(self): log.info("Closing Simulation: {id}".format(id=self.sim_id)) + if self.results_socket: + self.results_socket.close() if self.proc and self.send_sighup: ppid = os.getppid() if self.success: @@ -181,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): """ @@ -227,8 +220,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 ) @@ -242,12 +235,6 @@ class SimulatedTrading(object): else: return [] - def __iter__(self): - return self - - def next(self): - return self.gen.next() - @staticmethod def create_test_zipline(**config): """ @@ -268,7 +255,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. """ @@ -283,22 +271,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 @@ -309,22 +297,21 @@ 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 #------------------- - sids = [sid] - #------------------- - if config.has_key('trade_source'): + if 'trade_source' in config: trade_source = config['trade_source'] else: trade_source = factory.create_daily_trade_source( - sids, + sid_list, trade_count, - trading_environment + trading_environment, + concurrent=concurrent_trades ) #------------------- @@ -335,7 +322,7 @@ class SimulatedTrading(object): #------------------- # Create the Algo #------------------- - if config.has_key('algorithm'): + if 'algorithm' in config: test_algo = config['algorithm'] else: test_algo = TestAlgorithm( @@ -361,6 +348,7 @@ class SimulatedTrading(object): return sim + class SimulatedTradingLite(object): """ SimulatedTrading without multiprocess and without zmq. @@ -373,15 +361,33 @@ 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. - 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 @@ -389,4 +395,4 @@ class SimulatedTradingLite(object): return self def next(self): - return self.gen.next() + return self.gen.next() \ No newline at end of file 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() 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/factory.py b/zipline/utils/factory.py index 8b2565c4..4f2d4ff7 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -12,7 +12,6 @@ from datetime import datetime, timedelta import zipline.finance.risk as risk import zipline.protocol as zp -from zipline.gens.tradegens import RandomEquityTrades from zipline.gens.tradegens import SpecificEquityTrades from zipline.gens.utils import create_trade from zipline.finance.trading import TradingEnvironment 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:]) 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): """ 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/test_utils.py b/zipline/utils/test_utils.py index 3a9a0906..513352b1 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -7,16 +7,18 @@ 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'): 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 +35,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 +64,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 +89,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 +124,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,30 +159,13 @@ def launch_component(component): proc.start() return proc + def launch_monitor(monitor): proc = multiprocessing.Process(target=monitor.run) proc.start() 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): @@ -190,6 +180,7 @@ class ExceptionSource(object): def next(self): 5 / 0 + class ExceptionTransform(object): def __init__(self): 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) - ])