diff --git a/tests/test_finance.py b/tests/test_finance.py index f841e37f..d85959ba 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -2,6 +2,7 @@ Tests for the zipline.finance package """ import pytz +import zmq from unittest2 import TestCase from datetime import datetime, timedelta @@ -33,10 +34,13 @@ class FinanceTestCase(TestCase): def setUp(self): self.zipline_test_config = { - 'allocator' : allocator, - 'sid' : 133, - 'devel' : True + 'allocator' : allocator, + 'sid' : 133, + 'devel' : True, + 'results_socket' : allocator.lease(1)[0] } + self.ctx = zmq.Context() + self.log_handler = LoggingHandler() self.log_handler.push_application() @@ -109,147 +113,64 @@ class FinanceTestCase(TestCase): self.assertTrue(env.last_close.month == 12) self.assertTrue(env.last_close.day == 31) - # The following two tests appear broken no that the order source is - # non blocking. HUNCH: The trades are streaming through before the orders - # are placed. + def drain_zipline(self): + self.receiver = self.ctx.socket(zmq.PULL) + self.receiver.bind(self.zipline_test_config['results_socket']) - #@timed(EXTENDED_TIMEOUT) - def test_orders(self): + output = [] + transaction_count = 0 + while True: + msg = self.receiver.recv() + if msg == str(zp.CONTROL_PROTOCOL.DONE): + break + else: + update = zp.BT_UPDATE_UNFRAME(msg) + output.append(update) + if update['prefix'] == 'PERF': + transaction_count += \ + len(update['payload']['daily_perf']['transactions']) - # Simulation - # ---------- - - self.zipline_test_config['simulation_style'] = \ - SIMULATION_STYLE.FIXED_SLIPPAGE - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - zipline.simulate(blocking=True) - - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) - - # TODO: Make more assertions about the final state of the components. - self.assertEqual(zipline.sim.feed.pending_messages(), 0, \ - "The feed should be drained of all messages, found {n} remaining." \ - .format(n=zipline.sim.feed.pending_messages())) - - # the trading client should receive one transaction for every - # order placed. - self.assertEqual( - zipline.trading_client.txn_count, - zipline.trading_client.order_count - ) - - - #@timed(DEFAULT_TIMEOUT) - def test_aggressive_buying(self): - - # Simulation - # ---------- - - # TODO: for some reason the orders aren't filled without an extra - # trade. - trade_count = 5000 - self.zipline_test_config['order_count'] = trade_count - 1 - self.zipline_test_config['trade_count'] = trade_count - self.zipline_test_config['order_amount'] = 1 - - # tell the simulator to fill the orders in individual transactions - # matching the order volume exactly. - self.zipline_test_config['simulation_style'] = \ - SIMULATION_STYLE.FIXED_SLIPPAGE - self.zipline_test_config['environment'] = factory.create_trading_environment() - - sid_list = [self.zipline_test_config['sid']] - - self.zipline_test_config['trade_source'] = factory.create_minutely_trade_source( - sid_list, - trade_count, - self.zipline_test_config['environment'] - ) - - zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config) - zipline.simulate(blocking=True) - - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) - - self.assertEqual(zipline.sim.feed.pending_messages(), 0, \ - "The feed should be drained of all messages, found {n} remaining." \ - .format(n=zipline.sim.feed.pending_messages())) - - # - # the trading client should receive one transaction for every - # order placed. - self.assertEqual( - zipline.trading_client.txn_count, - zipline.trading_client.order_count - ) + del self.receiver + return output, transaction_count @timed(EXTENDED_TIMEOUT) - def test_performance(self): + def test_full_zipline(self): #provide enough trades to ensure all orders are filled. self.zipline_test_config['order_count'] = 100 self.zipline_test_config['trade_count'] = 200 zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config) - zipline.simulate(blocking=True) + zipline.simulate(blocking=False) + + output, transaction_count = self.drain_zipline() + + self.assertTrue(zipline.sim.ready()) + self.assertFalse(zipline.sim.exception) self.assertEqual( - zipline.sim.feed.pending_messages(), - 0, - "The feed should be drained of all messages, found {n} remaining." \ - .format(n=zipline.sim.feed.pending_messages()) + self.zipline_test_config['order_count'], + transaction_count ) - self.assertEqual( - zipline.sim.merge.pending_messages(), - 0, - "The merge should be drained of all messages, found {n} remaining." \ - .format(n=zipline.sim.merge.pending_messages()) - ) + # the final message is the risk report, the second to + # last is the final day's results. Positions is a list of + # dicts. + closing_positions = output[-2]['payload']['daily_perf']['positions'] self.assertEqual( - zipline.algorithm.count, - zipline.algorithm.incr, - "The test algorithm should send as many orders as specified.") - - - transaction_sim = zipline.trading_client.txn_sim - self.assertEqual( - transaction_sim.txn_count, - zipline.trading_client.perf.txn_count, - "The perf tracker should handle the same number of transactions \ - as the simulator emits." - ) - - self.assertEqual( - len(zipline.get_positions()), + len(closing_positions), 1, "Portfolio should have one position." ) sid = self.zipline_test_config['sid'] self.assertEqual( - zipline.get_positions()[sid]['sid'], + closing_positions[0]['sid'], sid, "Portfolio should have one position in " + str(sid) ) - self.assertEqual( - zipline.sources['SpecificEquityTrades'].count, - self.zipline_test_config['trade_count'], - "The simulated trade source should send all trades." - ) - - self.assertEqual( - zipline.algorithm.frame_count, - self.zipline_test_config['trade_count'], - "The algorithm should receive all trades." - ) - #@timed(DEFAULT_TIMEOUT) def test_sid_filter(self): """Ensure the algorithm's filter prevents events from arriving.""" @@ -271,11 +192,17 @@ class FinanceTestCase(TestCase): **self.zipline_test_config ) - zipline.simulate(blocking=True) + zipline.simulate(blocking=False) + + output, transaction_count = self.drain_zipline() + + self.assertTrue(zipline.sim.ready()) + self.assertFalse(zipline.sim.exception) + #check that the algorithm received no events self.assertEqual( 0, - test_algo.frame_count, + transaction_count, "The algorithm should not receive any events due to filtering." ) @@ -402,6 +329,9 @@ class FinanceTestCase(TestCase): self.transaction_sim(**params1) def transaction_sim(self, **params): + """ This is a utility method that asserts expected + results for conversion of orders to transactions given a + trade history""" trade_count = params['trade_count'] trade_amount = params['trade_amount'] diff --git a/zipline/components/tradesimulation.py b/zipline/components/tradesimulation.py index 12b4c8d7..9fcefdd7 100644 --- a/zipline/components/tradesimulation.py +++ b/zipline/components/tradesimulation.py @@ -80,7 +80,7 @@ class TradeSimulationClient(Component): self.algo_log = Logger("AlgoLog") self.algorithm.set_logger(self.algo_log) - self.run_logged_op(self.algorithm.initialize) + self.do_op(self.algorithm.initialize) def setup_logging(self, socket = None): sock = socket or self.results_socket @@ -183,11 +183,16 @@ class TradeSimulationClient(Component): # data injection pipeline for log rerouting # any fields injected here should be added to # LOG_EXTRA_FIELDS in zipline/protocol.py - self.run_logged_op(self.algorithm.handle_data, data) + self.do_op(self.algorithm.handle_data, data) - def run_logged_op(self, callable_op, *args, **kwargs): + def exception_callback(self, trace): + log.info(trace) + pass + + def do_op(self, callable_op, *args, **kwargs): """ Wrap a callable operation with the zmq logbook - handler if it exits.""" + handler if it exits. Also wrap the invocation of callable + op in a try/except, and relay """ if self.zmq_out: def inject_event_data(record): diff --git a/zipline/core/component.py b/zipline/core/component.py index 841800e1..2b8d425b 100644 --- a/zipline/core/component.py +++ b/zipline/core/component.py @@ -208,8 +208,8 @@ class Component(object): self.setup_poller() - self.open() self.setup_control() + self.open() self.signal_ready() self.lock_ready() @@ -326,7 +326,9 @@ class Component(object): # In case we didn't receive a ping, send a pre-emptive # pong to the monitor. - elif self.last_ping and time.time() - self.last_ping > 1: + elif hasattr(self, 'control_out') and \ + self.last_ping and \ + time.time() - self.last_ping > 1: # send a ping ahead of schedule pre_pong = time.time() heartbeat_frame = CONTROL_FRAME( @@ -452,7 +454,7 @@ class Component(object): def signal_exception(self, exc=None, scope=None): """ - This is *very* important error tracking handler. + This is a *very* important error tracking handler. Will inform the system that the component has failed and how it has failed. @@ -474,6 +476,9 @@ class Component(object): trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback)) sys.stdout.write(trace) + if hasattr(self, 'exception_callback'): + self.exception_callback(trace) + if hasattr(self, 'control_out'): exception_frame = CONTROL_FRAME( CONTROL_PROTOCOL.EXCEPTION, @@ -490,20 +495,20 @@ class Component(object): self.state_flag = COMPONENT_STATE.DONE - if self.out_socket: + if hasattr(self, 'out_socket') and self.out_socket: msg = zmq.Message(str(CONTROL_PROTOCOL.DONE)) self.out_socket.send(msg) + if hasattr(self, 'control_out'): + # notify controller we're done + done_frame = CONTROL_FRAME( + CONTROL_PROTOCOL.DONE, + '' + ) - # notify controller 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) + self.control_out.send(done_frame) + log.info("[%s] sent control done" % self.get_id) # there is a narrow race condition where we finish just # after the Monitor accepts our prior heartbeat, but just diff --git a/zipline/core/devsimulator.py b/zipline/core/devsimulator.py index 8835cb07..cbce4b66 100644 --- a/zipline/core/devsimulator.py +++ b/zipline/core/devsimulator.py @@ -5,7 +5,6 @@ See :py:method"" import logbook import threading -from zipline.core.simulatorref import SimulatorBase log = logbook.Logger('Dev Simulator') @@ -34,71 +33,4 @@ class AddressAllocator(object): return sockets def reaquire(self, *conn): - pass - - -class Simulator(SimulatorBase): - - zmq_flavor = 'thread' - - def __init__(self, addresses): - # TODO: rethink this - SimulatorBase.__init__(self, addresses) - self.subthreads = [] - self.running = False - - log.warn(DEPRECATION_WARNING) - - @property - def get_id(self): - return 'Simple Simulator' - - def launch_controller(self): - thread = threading.Thread( - target=self.controller.run, - ) - thread.start() - - self.subthreads.append(thread) - return thread - - def simulate(self): - thread = threading.Thread(target=self.run) - thread.start() - - self.subthreads.append(thread) - self.running = True - - return thread - - def did_clean_shutdown(self): - return not any([t.isAlive() for t in self.subthreads]) - - def shutdown(self): - """ - Destroy all tracked components. - """ - - if not self.running: - return - - - for component in self.components.itervalues(): - component.shutdown() - - for thread in self.subthreads: - if thread.is_alive(): - thread._Thread__stop() - - #self.controller.shutdown() - - self.running = False - - assert self.did_clean_shutdown() - - def launch_component(self, component): - thread = threading.Thread(target=component.run) - thread.start() - - self.subthreads.append(thread) - return thread + pass \ No newline at end of file diff --git a/zipline/core/monitor.py b/zipline/core/monitor.py index efcc6314..c9c66fda 100644 --- a/zipline/core/monitor.py +++ b/zipline/core/monitor.py @@ -42,7 +42,7 @@ log = logbook.Logger('Controller') # the system. PARAMETERS = ndict(dict( - GENERATIONAL_PERIOD = 10, #seconds + GENERATIONAL_PERIOD = 30, #seconds ALLOWED_SKIPPED_HEARTBEATS = 10, ALLOWED_INVALID_HEARTBEATS = 3, PRESTART_HEARBEATS = 3, diff --git a/zipline/core/process.py b/zipline/core/process.py new file mode 100644 index 00000000..1bac176e --- /dev/null +++ b/zipline/core/process.py @@ -0,0 +1,89 @@ +""" +The process simulator. Each component in threading.Thread +""" + +import logbook +import multiprocessing +from zipline.core.host import ComponentHost + +log = logbook.Logger('Process Simulator') + +class ProcessSimulator(ComponentHost): + """ + The process simulator. + """ + + zmq_flavor = 'mp' + + def __init__(self, addresses): + ComponentHost.__init__(self, addresses) + self.subprocesses = [] + self.running = False + self.mapping = {} + + def define(self, key, val): + """ + Returns the mapping between a component and its + pid. + """ + self.mapping[key] = val + + @property + def get_id(self): + return 'Multiprocess Simulator' + + # ========= + # Launchers + # ========= + # + # invoked by the host's open() + + def launch_controller(self): + proc = multiprocessing.Process(target=self.controller.run) + proc.start() + self.con = proc + + # Process specific + self.controller_process = proc + self.mapping[proc.pid] = 'Controller' + + def launch_component(self, component): + proc = multiprocessing.Process(target=component.run) + proc.start() + self.subprocesses.append(proc) + + self.mapping[proc.pid] = component.get_id + return proc + + def simulate(self): + """ + Kick off the simulation + """ + self.run() + + def did_clean_shutdown(self): + cleanly = not any([s.is_alive() for s in self.subprocesses]) + if not cleanly: + for process in self.subprocesses: + if process.is_alive(): + log.error('Failed to Yield', self.mapping[process.pid]) + return cleanly + + def shutdown(self, ensure_clean=True): + """ + Shutdown the simulation. + """ + for component in self.components.itervalues(): + component.shutdown() + + for process in self.subprocesses: + process.join(timeout=1) + process.terminate() + + self.controller.shutdown(soft=True) + self.running = False + + self.con.terminate() + + if ensure_clean: + assert self.did_clean_shutdown() diff --git a/zipline/core/simulatorref.py b/zipline/core/simulatorref.py index abe87379..6c63e27b 100644 --- a/zipline/core/simulatorref.py +++ b/zipline/core/simulatorref.py @@ -17,7 +17,7 @@ Subclasses: import abc from zipline.core.host import ComponentHost -class SimulatorBase(ComponentHost): +class SimulatorBaseDeprecated(ComponentHost): __metaclass__ = abc.ABCMeta diff --git a/zipline/lines.py b/zipline/lines.py index f56c1eb3..99f69e80 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -69,7 +69,7 @@ from zipline.transforms import BaseTransform from zipline.test_algorithms import TestAlgorithm from zipline.components import TradeSimulationClient -from zipline.core.devsimulator import Simulator +from zipline.core.process import ProcessSimulator from zipline.core.monitor import Controller from zipline.finance.trading import SIMULATION_STYLE @@ -105,8 +105,6 @@ class SimulatedTrading(object): :py:class:`zipline.trading.TradingEnvironment` - allocator: an instance of :py:class:`zipline.simulator.AddressAllocator` - - simulator_class: a :py:class:`zipline.core.host.ComponentHost` - subclass (not an instance) - 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` @@ -139,12 +137,9 @@ class SimulatedTrading(object): devel = self.devel ) - # TODO: Not freeform - self.con.manage('freeform') - self.started = False - self.sim = config['simulator_class'](addresses) + self.sim = ProcessSimulator(addresses) self.clients = {} @@ -179,9 +174,6 @@ class SimulatedTrading(object): - order_amount - the number of shares per order, defaults to 100 - trade_count - the number of trades to simulate, defaults to 101 to ensure all orders are processed. - - simulator_class - optional parameter that provides an alternative - subclass of ComponentHost to hold the whole zipline. Defaults to - :py:class:`zipline.simulator.Simulator` - algorithm - optional parameter providing an algorithm. defaults to :py:class:`zipline.test.algorithms.TestAlgorithm` - trade_source - optional parameter to specify trades, if present. @@ -221,11 +213,6 @@ class SimulatedTrading(object): # trade than order trade_count = 101 - if config.has_key('simulator_class'): - simulator_class = config['simulator_class'] - else: - simulator_class = Simulator - simulation_style = config.get('simulation_style') if not simulation_style: simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE @@ -266,7 +253,6 @@ class SimulatedTrading(object): 'algorithm' : test_algo, 'trading_environment' : trading_environment, 'allocator' : allocator, - 'simulator_class' : simulator_class, 'simulation_style' : simulation_style, 'results_socket' : results_socket, 'devel' : config.get('devel', False) @@ -277,7 +263,7 @@ class SimulatedTrading(object): # Save us from needless debugging inside_test = 'nose' in inspect.stack()[-1][1] - if inside_test and not config.get('devel', False): + if False and inside_test and not config.get('devel', False): assert False, """ You need to run the SimulatedTrading inside a test with devel=True """ @@ -366,7 +352,7 @@ class SimulatedTrading(object): def setup_controller(self): """ - Prepare the controller tro manage the topology specified + Prepare the controller to manage the topology specified by this line. """ self.con.manage(self.topology) @@ -377,14 +363,6 @@ class SimulatedTrading(object): self.started = True self.sim_context = self.sim.simulate() - # If we're in development mode then flag all the - # components in the topology as devel so as to indicate - # that they won't poll on the control channels for - # anything other than the synchronized start. - if self.devel: - for component in self.components: - component.devel = True - # If we're using a threaded simulator block on the pool # of thread since we're only ever in a test and we don't # generally monitor the state of the system as a hold at @@ -392,16 +370,8 @@ class SimulatedTrading(object): # TODO: better way of identifying concurrency substrate if blocking: - if self.sim.zmq_flavor == 'thread': - log.debug('Blocking') - for thread in self.sim.subthreads: - #log.debug('Waiting on %r' % thread) - log.debug('Waiting on %r' % thread) - thread.join() - log.debug('Yielded on %r' % thread) - else: - for process in self.sim.subprocesses: - process.join() + for process in self.sim.subprocesses: + process.join() @property def is_success(self): diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index aa07fa54..7011fa53 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -154,7 +154,7 @@ class ExceptionAlgorithm(object): """ def __init__(self, throw_from): - self.throw_from == throw_from + self.throw_from = throw_from def initialize(self): if self.throw_from == "initialize":