diff --git a/pavement.py b/pavement.py index 5353c8a3..e46e5cc5 100644 --- a/pavement.py +++ b/pavement.py @@ -110,7 +110,6 @@ options( # Because I'm lazy stuff_i_want_in_my_debug_shell = [ ('qutil', 'zipline.util', []), - ('zmq', 'zmq', []), ] @task diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py deleted file mode 100644 index 7b4fb624..00000000 --- a/tests/test_exception_handling.py +++ /dev/null @@ -1,190 +0,0 @@ -import zmq - -from unittest2 import TestCase -from collections import defaultdict - -from zipline.test_algorithms import ExceptionAlgorithm, DivByZeroAlgorithm, \ - InitializeTimeoutAlgorithm, TooMuchProcessingAlgorithm -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 MAX_HEARTBEAT_INTERVALS - -from zipline.utils.test_utils import ( - drain_zipline, - setup_logger, - teardown_logger, - ExceptionSource, - ExceptionTransform -) - -DEFAULT_TIMEOUT = 15 # seconds -EXTENDED_TIMEOUT = 90 - -allocator = AddressAllocator(1000) - -class ExceptionTestCase(TestCase): - - leased_sockets = defaultdict(list) - - def setUp(self): - self.zipline_test_config = { - 'sid' : 133, - 'results_socket_uri' : allocator.lease(1)[0], - 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE - } - self.ctx = zmq.Context() - setup_logger(self) - - def tearDown(self): - self.ctx.term() - teardown_logger(self) - - def test_datasource_exception(self): - self.zipline_test_config['trade_source'] = ExceptionSource() - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - output, _ = drain_zipline(self, zipline) - assert len(output) == 1 - assert output[0]['prefix'] == 'EXCEPTION' - message = output[0]['payload'] - for field in ['date', 'message', 'name', 'stack']: - assert field in message.keys() - - assert message['message'] == 'integer division or modulo by zero' - assert message['name'] == 'ZeroDivisionError' - - def test_tranform_exception(self): - exc_tnfm = StatefulTransform(ExceptionTransform) - self.zipline_test_config['transforms'] = [exc_tnfm] - - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - output, _ = drain_zipline(self, zipline) - assert len(output) == 1 - assert output[0]['prefix'] == 'EXCEPTION' - message = output[0]['payload'] - for field in ['date', 'message', 'name', 'stack']: - assert field in message.keys() - - assert message['message'] == 'An assertion message' - assert message['name'] == 'AssertionError' - - - def test_exception_in_init(self): - # Simulation - # ---------- - self.zipline_test_config['algorithm'] = \ - ExceptionAlgorithm( - 'initialize', - self.zipline_test_config['sid'] - ) - - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - output, _ = drain_zipline(self, zipline) - - self.assertEqual(output[-1]['prefix'], 'EXCEPTION') - payload = output[-1]['payload'] - self.assertTrue(payload['date']) - self.assertEqual(payload['message'],'Algo exception in initialize') - self.assertEqual(payload['name'],'Exception') - # make sure our path shortening is working - self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py') - self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py') - - def test_exception_in_handle_data(self): - # Simulation - # ---------- - self.zipline_test_config['algorithm'] = \ - ExceptionAlgorithm( - 'handle_data', - self.zipline_test_config['sid'] - ) - - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - - output, _ = drain_zipline(self, zipline) - self.assertEqual(output[-1]['prefix'], 'EXCEPTION') - payload = output[-1]['payload'] - self.assertTrue(payload['date']) - del payload['date'] - self.assertEqual(payload['message'],'Algo exception in handle_data') - self.assertEqual(payload['name'],'Exception') - # make sure our path shortening is working - self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py') - self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py') - - def test_zerodivision_exception_in_handle_data(self): - - # Simulation - # ---------- - self.zipline_test_config['algorithm'] = \ - DivByZeroAlgorithm( - self.zipline_test_config['sid'] - ) - - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - - output, _ = drain_zipline(self, zipline) - - self.assertEqual(output[-1]['prefix'], 'EXCEPTION') - payload = output[-1]['payload'] - self.assertTrue(payload['date']) - del payload['date'] - self.assertEqual(payload['message'],'integer division or modulo by zero') - self.assertEqual(payload['name'],'ZeroDivisionError') - # make sure our path shortening is working - self.assertEqual(payload['stack'][0]['filename'], '/zipline/lines.py') - self.assertEqual(payload['stack'][-1]['filename'], '/zipline/test_algorithms.py') - - def test_initialize_timeout(self): - - self.zipline_test_config['algorithm'] = \ - InitializeTimeoutAlgorithm( - self.zipline_test_config['sid'] - ) - - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - output, _ = drain_zipline(self, zipline) - self.assertEqual(output[-1]['prefix'], 'EXCEPTION') - payload = output[-1]['payload'] - self.assertEqual(payload['name'],'TimeoutException') - self.assertEqual(payload['message'], 'Call to initialize timed out') - - def test_heartbeat(self): - - self.zipline_test_config['algorithm'] = \ - TooMuchProcessingAlgorithm( - self.zipline_test_config['sid'] - ) - zipline = SimulatedTrading.create_test_zipline( - **self.zipline_test_config - ) - output, _ = drain_zipline(self, zipline) - - # There should be a message for each hearbeat, plus a message - # for the final timeout. - assert len(output) == MAX_HEARTBEAT_INTERVALS + 1 - - # Assert that everything but the last message is a heartbeat log. - for message in output[0:-1]: - assert message['prefix'] == 'LOG' - assert message['payload']['func_name'] == 'log_heartbeats' - - # Assert that the last message is a timeout exception. - self.assertEqual(output[-1]['prefix'], 'EXCEPTION') - payload = output[-1]['payload'] - self.assertEqual(payload['name'],'TimeoutException') - self.assertEqual(payload['message'], 'Too much time spent in handle_data call') - diff --git a/tests/test_finance.py b/tests/test_finance.py index 87795609..1168902f 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -2,7 +2,6 @@ Tests for the zipline.finance package """ import pytz -import zmq from unittest2 import TestCase from datetime import datetime, timedelta @@ -36,9 +35,7 @@ class FinanceTestCase(TestCase): def setUp(self): self.zipline_test_config = { 'sid' : 133, - 'results_socket_uri' : allocator.lease(1)[0] } - self.ctx = zmq.Context() setup_logger(self) diff --git a/zipline/lines.py b/zipline/lines.py index 75c473f0..46fe5fa7 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -59,16 +59,9 @@ before invoking simulate. | __init__. | +---------------------------------+ """ -import sys -import zmq -import os -from signal import SIGHUP, SIGINT -import multiprocessing -from setproctitle import setproctitle from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import SIMULATION_STYLE -from zipline.utils.log_utils import ZeroMQLogHandler from zipline.utils import factory from zipline.gens.composites import ( @@ -78,8 +71,6 @@ from zipline.gens.composites import ( from zipline.gens.tradesimulation import TradeSimulationClient as tsc from logbook import Logger -import zipline.protocol as zp - log = Logger('Lines') @@ -90,10 +81,27 @@ class SimulatedTrading(object): transforms, algorithm, environment, - style, - results_socket_uri, - context, - sim_id): + 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 @@ -102,138 +110,12 @@ class SimulatedTrading(object): *self.transforms) self.trading_client = tsc(algorithm, environment, style) self.gen = self.trading_client.simulate(self.with_tnfms) - self.results_uri = results_socket_uri - self.results_socket = None - self.context = context - self.sim_id = sim_id - # optional process if we fork simulate into an - # independent process. - self.proc = None - self.send_sighup = False - self.logger = Logger(sim_id) - self.print_logger = Logger('Print') + def __iter__(self): + return self.gen - # exit status flag - self.success = False - - def simulate(self, blocking=True, send_sighup=False): - - # for non-blocking, - if blocking: - self.run_gen() - else: - self.send_sighup = send_sighup - return self.fork_and_sim() - - def fork_and_sim(self): - self.proc = multiprocessing.Process(target=self.run_gen) - self.proc.start() - return self.proc - - def run_gen(self): - setproctitle(self.sim_id) - self.open() - if self.zmq_out: - with self.zmq_out.threadbound(): - self.stream_results() - # if no log socket, just run the algo normally - else: - self.stream_results() - - def stream_results(self): - assert self.results_socket, \ - "Results socket must exist to stream results" - try: - for event in self.gen: - if 'daily_perf' in event: - msg = zp.PERF_FRAME(event) - else: - msg = zp.RISK_FRAME(event) - self.results_socket.send(msg) - - self.signal_done() - self.success = True - except Exception as exc: - self.handle_exception(exc) - finally: - # not much to do besides log our exit. - self.close() - - def signal_done(self): - # notify monitor we're done - done_frame = zp.DONE_FRAME('success') - self.results_socket.send(done_frame) - - 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: - log.warning("Sending SIGHUP") - os.kill(ppid, SIGHUP) - else: - log.warning("Sending SIGINT") - os.kill(ppid, SIGINT) - - def handle_exception(self, exc): - self.signal_exception(exc) - - def signal_exception(self, exc=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. - """ - exc_type, exc_value, exc_traceback = sys.exc_info() - - try: - log.exception('{id} sending exception to result stream.'\ - .format(id=self.sim_id)) - msg = zp.EXCEPTION_FRAME( - exc_traceback, - exc_type.__name__, - exc_value.message - ) - - self.results_socket.send(msg) - except: - log.exception("Exception while reporting simulation exception.") - - def open(self): - if not self.context: - self.context = zmq.Context() - if self.results_uri: - sock = self.context.socket(zmq.PUSH) - sock.connect(self.results_uri) - self.results_socket = sock - self.setup_logging() - - def setup_logging(self): - assert self.results_socket - # The filter behavior is: matches are logged, mismatches - # are bubbled. If bubble is True, matches are also - # 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'], - bubble=False - ) - - def join(self): - if self.proc: - self.proc.join() - - def get_pids(self): - if self.proc: - return [self.proc.pid] - else: - return [] + def __next__(self): + return self.gen.next() @staticmethod def create_test_zipline(**config): @@ -297,10 +179,6 @@ 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) - #------------------- # Trade Source #------------------- @@ -341,52 +219,7 @@ class SimulatedTrading(object): test_algo, trading_environment, simulation_style, - results_socket_uri, - zmq_context, - simulation_id) + ) #------------------- return sim - - -class SimulatedTradingLite(object): - """ - SimulatedTrading without multiprocess and without zmq. - Useful for profiling the core logic and for rapid testing - of new features. - """ - def __init__(self, - sources, - transforms, - 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.trading_client = tsc(algorithm, environment, style) - self.gen = self.trading_client.simulate(self.with_tnfms) - - def get_results(self): - return self.gen diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 513352b1..ad085ba4 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -1,7 +1,4 @@ import multiprocessing -import zmq -import time -import zipline.protocol as zp from datetime import datetime import blist from zipline.utils.date_utils import EPOCH @@ -65,69 +62,18 @@ def check(test, a, b, label=None): 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" - assert isinstance(test.zipline_test_config, dict) - assert test.zipline_test_config['results_socket_uri'], \ - "need to specify a socket address for logs/perf/risk" - test.receiver = create_receiver( - test.zipline_test_config['results_socket_uri'], - test.ctx - ) - # Bind and connect are asynch, so allow time for bind before - # starting the zipline (TSC connects internally). - time.sleep(1) - - # start the simulation - zipline.simulate(blocking=p_blocking) - output, transaction_count = drain_receiver(test.receiver) - # some processes will exit after the message stream is - # finished. We block here to avoid collisions with subsequent - # ziplines. - zipline.join() - - 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): +def assert_single_position(test, zipline): output = [] transaction_count = 0 msg_counter = 0 - while True: - msg = receiver.recv() + # start the simulation + for update in zipline: msg_counter += 1 - update = zp.BT_UPDATE_UNFRAME(msg) output.append(update) - if update['prefix'] == 'PERF': + if update.has_key('daily_perf'): transaction_count += \ - len(update['payload']['daily_perf']['transactions']) - elif update['prefix'] == 'EXCEPTION': - break - elif update['prefix'] == 'DONE': - break + len(update['daily_perf']['transactions']) - if count and msg_counter >= count: - break - - receiver.close() - del receiver - - return output, transaction_count - - -def assert_single_position(test, zipline, blocking=False): - output, transaction_count = drain_zipline(test, - zipline, - p_blocking=blocking) - test.assertEqual(output[-1]['prefix'], 'DONE') test.assertEqual( test.zipline_test_config['order_count'], @@ -137,8 +83,7 @@ def assert_single_position(test, zipline, blocking=False): # the final message is the risk report, the second to # last is the final day's results. Positions is a list of # dicts. - perfs = [x for x in output if x['prefix'] == 'PERF'] - closing_positions = perfs[-2]['payload']['daily_perf']['positions'] + closing_positions = output[-2]['daily_perf']['positions'] test.assertEqual( len(closing_positions),