From 06dc6f7acb84cb4e73440bb39a5bdf762997ab12 Mon Sep 17 00:00:00 2001 From: fawce Date: Mon, 6 Aug 2012 13:11:20 -0400 Subject: [PATCH 1/6] beginning refactor to use single threaded simulator. --- zipline/__init__.py | 6 - zipline/gens/examples.py | 16 +-- zipline/gens/tradegens.py | 3 +- zipline/gens/utils.py | 12 ++ zipline/lines.py | 272 ++++++-------------------------------- zipline/utils/factory.py | 49 +++---- 6 files changed, 76 insertions(+), 282 deletions(-) diff --git a/zipline/__init__.py b/zipline/__init__.py index a84cd345..31272fcb 100644 --- a/zipline/__init__.py +++ b/zipline/__init__.py @@ -6,15 +6,9 @@ Zipline # it is a place to expose the public interfaces. import protocol # namespace -from core.monitor import Monitor -from lines import SimulatedTrading -from core.host import ComponentHost from utils.protocol_utils import ndict __all__ = [ - SimulatedTrading, - Monitor, - ComponentHost, protocol, ndict ] diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py index a6a95f59..2f003230 100644 --- a/zipline/gens/examples.py +++ b/zipline/gens/examples.py @@ -1,6 +1,5 @@ import pytz -from time import sleep from pprint import pprint as pp from datetime import datetime, timedelta @@ -16,7 +15,7 @@ from zipline.gens.tradesimulation import TradeSimulationClient as tsc import zipline.protocol as zp if __name__ == "__main__": - + filter = [2,3] #Set up source a. One minute between events. args_a = tuple() @@ -42,19 +41,18 @@ if __name__ == "__main__": #Set up source c. Three minutes between events. - sorted = date_sorted_sources(source_a, source_b) - + sorted = date_sorted_sources(source_a, source_b) + passthrough = StatefulTransform(Passthrough) mavg_price = StatefulTransform(MovingAverage, timedelta(minutes = 20), ['price']) - + 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/tradegens.py b/zipline/gens/tradegens.py index b1a0ed96..5c1bacd7 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -6,8 +6,7 @@ import random from itertools import chain, cycle, ifilter, izip from datetime import datetime, timedelta -from zipline.utils.factory import create_trade -from zipline.gens.utils import hash_args +from zipline.gens.utils import hash_args, create_trade def date_gen(start = datetime(2006, 6, 6, 12), delta = timedelta(minutes = 1), diff --git a/zipline/gens/utils.py b/zipline/gens/utils.py index 071ce5fc..45e31de4 100644 --- a/zipline/gens/utils.py +++ b/zipline/gens/utils.py @@ -66,6 +66,18 @@ def hash_args(*args, **kwargs): hasher.update(combined) return hasher.hexdigest() +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, + 'volume' : amount + }) + return row + + def assert_datasource_protocol(event): """Assert that an event meets the protocol for datasource outputs.""" diff --git a/zipline/lines.py b/zipline/lines.py index a5a3858e..fbc4d155 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -60,110 +60,41 @@ before invoking simulate. +---------------------------------+ """ -import inspect import logbook -#import zipline.utils.factory as factory - -from zipline.components import DataSource -from zipline.transforms import BaseTransform from zipline.test_algorithms import TestAlgorithm -from zipline.components import TradeSimulationClient -from zipline.core.process import ProcessSimulator -from zipline.core.monitor import Monitor from zipline.finance.trading import SIMULATION_STYLE +from zipline.utils import factory +import pytz + +from pprint import pprint as pp +from datetime import datetime, timedelta + +from zipline.utils.factory import create_trading_environment +from zipline.test_algorithms import TestAlgorithm + +from zipline.gens.composites import SourceBundle, TransformBundle, \ + date_sorted_sources, merged_transforms +from zipline.gens.tradegens import SpecificEquityTrades +from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform +from zipline.gens.tradesimulation import TradeSimulationClient as tsc + +import zipline.protocol as zp + log = logbook.Logger('Lines') class SimulatedTrading(object): - """ - Zipline with:: - - _no_ data sources. - - Trade simulation client, which is available to send callbacks on - events and also accept orders to be simulated. - - An order data source, which will receive orders from the trade - simulation client, and feed them into the event stream to be - serialized and order alongside all other data source events. - - transaction simulation transformation, which receives the order - events and estimates a theoretical execution price and volume. + @staticmethod + def create_simulation(sources, transforms, algorithm, environment, style): - All components in this zipline are subject to heartbeat checks and - a control monitor, which can kill the entire zipline in the event of - exceptions in one of the components or an external request to end the - simulation. - """ + sorted = date_sorted_sources(*sources) + passthrough = StatefulTransform(Passthrough) - def __init__(self, **config): - """ - :param config: a dict with the following required properties:: - - - algorithm: a class that follows the algorithm protocol. See - :py:meth:`zipline.finance.trading.TradeSimulationClient.add_algorithm - for details. - - trading_environment: an instance of - :py:class:`zipline.trading.TradingEnvironment` - - allocator: an instance of - :py:class:`zipline.simulator.AddressAllocator` - - 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` - """ - assert isinstance(config, dict) - self.algorithm = config['algorithm'] - self.allocator = config['allocator'] - self.trading_environment = config['trading_environment'] - self.sim_style = config.get('simulation_style') - self.send_sighup = config.get('send_sighup', False) - - - self.leased_sockets = [] - self.sim_context = None - - sockets = self.allocate_sockets(7) - addresses = { - 'sync_address' : sockets[0], - 'data_address' : sockets[1], - 'feed_address' : sockets[2], - 'merge_address' : sockets[3], - # TODO: this refers to the results of the merge, a - # horribly confusing name for the socket. - 'results_address' : sockets[4], - } - - self.monitor = Monitor( - # pub socket - sockets[5], - # route socket - sockets[6], - # exception socket to match tradesimclient's result - # socket, because we want to relay exceptions to the - # same listener - config['results_socket'], - send_sighup=self.send_sighup - ) - - self.started = False - - self.sim = ProcessSimulator(addresses) - - self.clients = {} - - self.trading_client = TradeSimulationClient( - self.trading_environment, - self.sim_style, - config['results_socket'], - self.algorithm - ) - self.add_client(self.trading_client) - - # setup all sources - self.sources = {} - - #setup transforms - self.transforms = {} - - self.sim.register_monitor( self.monitor ) + merged = merged_transforms(sorted, passthrough, *transforms) + trading_client = tsc(algorithm, environment, style) + return trading_client.simluate(merged) @staticmethod @@ -173,7 +104,6 @@ class SimulatedTrading(object): - environment - a \ :py:class:`zipline.finance.trading.TradingEnvironment` - - allocator - a :py:class:`zipline.simulator.AddressAllocator` - sid - an integer, which will be used as the security ID. - order_count - the number of orders the test algo will place, defaults to 100 @@ -188,10 +118,11 @@ class SimulatedTrading(object): - 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` + - transforms: optional parameter that provides a list + of StatefulTransform objects. """ assert isinstance(config, dict) - allocator = config['allocator'] sid = config['sid'] #-------------------- @@ -236,6 +167,12 @@ class SimulatedTrading(object): trade_count, trading_environment ) + + #------------------- + # Transforms + #------------------- + transforms = config.get('transforms', []) + #------------------- # Create the Algo #------------------- @@ -248,149 +185,20 @@ class SimulatedTrading(object): order_count ) - if config.has_key('results_socket'): - results_socket = config['results_socket'] - else: - results_socket = None #------------------- # Simulation #------------------- - zipline = SimulatedTrading(**{ - 'algorithm' : test_algo, - 'trading_environment' : trading_environment, - 'allocator' : allocator, - 'simulation_style' : simulation_style, - 'results_socket' : results_socket, - }) + + sim = SimulatedTrading.create_simulation( + [trade_source], + transforms, + test_algo, + trading_environment, + simulation_style) #------------------- - zipline.add_source(trade_source) + return sim - return zipline - - def add_source(self, source): - """ - Adds the source to the zipline, sets the sid filter of the - source to the algorithm's sid filter. - """ - assert isinstance(source, DataSource) - self.check_started() - source.set_filter('sid', self.algorithm.get_sid_filter()) - self.sim.register_components([source]) - - # ``id`` is name of source_id, ``get_id`` is the class name - self.sources[source.get_id] = source - - def add_transform(self, transform): - assert isinstance(transform, BaseTransform) - self.check_started() - self.sim.register_components([transform]) - self.transforms[transform.get_id] = transform - - def add_client(self, client): - assert isinstance(client, TradeSimulationClient) - self.check_started() - self.sim.register_components([client]) - self.clients[client.get_id] = client - - def check_started(self): - if self.started: - raise ZiplineException("TradeSimulation", "You cannot add \ - components after the simulation has begun.") - - def get_cumulative_performance(self): - return self.trading_client.perf.cumulative_performance.to_dict() - - def allocate_sockets(self, n): - """ - Allocate sockets local to this line, track them so - we can gc after test run. - """ - - assert isinstance(n, int) - assert n > 0 - - leased = self.allocator.lease(n) - self.leased_sockets.extend(leased) - - return leased - - @property - def components(self): - """ - Return the component instances inside of this topology - """ - - base = set(self.sim.components.values()) - transforms = set(self.transforms.values()) - sources = set(self.sources.values()) - - return base | transforms | sources - - @property - def topology(self): - """ - Returns the Component names in the topology of the - backtest. - """ - - # A complete topology is the union of three classes of - # components added individually to the simulation client - # at various places. - # - # base : ['FEED', 'MERGE', 'TRADING_CLIENT', 'PASSTHROUGH'] - # transforms : ['vwap__01', ... ] - # sources : ['MongoTradeHistory', ... ] - - base = set(self.sim.components.keys()) - transforms = set(self.transforms.keys()) - sources = set(self.sources.keys()) - - return base | transforms | sources - - def setup_monitor(self): - """ - Prepare the monitor to manage the topology specified - by this line. - """ - self.monitor.manage(self.topology) - - def simulate(self, blocking=True): - self.setup_monitor() - - self.started = True - self.sim_context = self.sim.simulate() - - # 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 - # the supervisory layer - - # TODO: better way of identifying concurrency substrate - if blocking: - for process in self.sim.subprocesses: - process.join() - - @property - def is_success(self): - # TODO: other assertions? - if self.sim.did_clean_shutdown(): - return True - else: - return False - - #-------------------------------- - # Component property accessors - #-------------------------------- - - def get_positions(self): - """ - returns current positions as a dict. draws from the cumulative - performance period in the performance tracker. - """ - perf = self.trading_client.perf.cumulative_performance - positions = perf.get_positions() - return positions class ZiplineException(Exception): def __init__(self, zipline_name, msg): diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index 004f542a..e001cf07 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -12,7 +12,9 @@ from datetime import datetime, timedelta import zipline.finance.risk as risk import zipline.protocol as zp -from zipline.finance.sources import SpecificEquityTrades, RandomEquityTrades +from zipline.finance.sources import RandomEquityTrades +from zipline.gens.tradegens import SpecificEquityTrades +from zipline.gens.utils import create_trade from zipline.finance.trading import TradingEnvironment # TODO @@ -69,16 +71,6 @@ def create_trading_environment(year=2006): return trading_environment -def create_trade(sid, price, amount, datetime, source_id = "test_factory"): - row = zp.ndict({ - 'source_id' : source_id, - 'type' : zp.DATASOURCE_TYPE.TRADE, - 'sid' : sid, - 'dt' : datetime, - 'price' : price, - 'volume' : amount - }) - return row def get_next_trading_dt(current, interval, trading_calendar): next = current @@ -220,29 +212,20 @@ def create_minutely_trade_source(sids, trade_count, trading_environment): ) def create_trade_source(sids, trade_count, trade_time_increment, trading_environment): - trade_history = [] - price = [10.1] * trade_count - volume = [100] * trade_count + #Set up source a. One minute between events. + args = tuple() + kwargs = { + 'count' : trade_count, + 'sids' : sids, + 'start' : trading_environment.first_open, + 'delta' : trade_time_increment, + 'filter' : sids + } + source = SpecificEquityTrades(*args, **kwargs) - for sid in sids: - start_date = trading_environment.first_open + # TODO: do we need to set the trading environment's end to same dt as + # the last trade in the history? + #trading_environment.period_end = trade_history[-1].dt - generated_trades = create_trade_history( - sid, - price, - volume, - trade_time_increment, - trading_environment - ) - - trade_history.extend(generated_trades) - - trade_history = sorted(trade_history, key=attrgetter('dt')) - - #set the trading environment's end to same dt as the last trade in the - #history. - trading_environment.period_end = trade_history[-1].dt - - source = SpecificEquityTrades(trade_history) return source From 3f4d772e4c04e4c6cee7c29d8769a65f13712d2c Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Aug 2012 11:15:14 -0400 Subject: [PATCH 2/6] first draft of lines returning new SimulatedTrading object. --- tests/test_components.py | 15 ++- tests/test_exception_handling.py | 10 +- zipline/lines.py | 193 ++++++++++++++++++++++++++----- 3 files changed, 176 insertions(+), 42 deletions(-) diff --git a/tests/test_components.py b/tests/test_components.py index 33255883..7ab0d3b2 100644 --- a/tests/test_components.py +++ b/tests/test_components.py @@ -8,7 +8,8 @@ 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 MovingAverage, Passthrough, StatefulTransform +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 @@ -113,7 +114,8 @@ class ComponentTestCase(TestCase): monitor, socket_uri, DATASOURCE_FRAME, - DATASOURCE_UNFRAME + DATASOURCE_UNFRAME, + "source_a" ) launch_monitor(monitor) @@ -171,7 +173,8 @@ class ComponentTestCase(TestCase): monitor, socket_uris[0], DATASOURCE_FRAME, - DATASOURCE_UNFRAME + DATASOURCE_UNFRAME, + trade_gen_a.get_hash() ) comp_b = Component( @@ -179,7 +182,8 @@ class ComponentTestCase(TestCase): monitor, socket_uris[1], DATASOURCE_FRAME, - DATASOURCE_UNFRAME + DATASOURCE_UNFRAME, + trade_gen_b.get_hash() ) comp_c = Component( @@ -187,7 +191,8 @@ class ComponentTestCase(TestCase): monitor, socket_uris[2], DATASOURCE_FRAME, - DATASOURCE_UNFRAME + DATASOURCE_UNFRAME, + trade_gen_c.get_hash() ) sources = [comp_a, comp_b, comp_c] diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index 144568d1..f278ef34 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -26,11 +26,11 @@ class ExceptionTestCase(TestCase): 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 + 'allocator' : allocator, + 'sid' : 133, + 'devel' : False, + 'results_socket_uri' : allocator.lease(1)[0], + 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE } self.ctx = zmq.Context() setup_logger(self) diff --git a/zipline/lines.py b/zipline/lines.py index fbc4d155..6b96d1ec 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -59,44 +59,178 @@ before invoking simulate. | __init__. | +---------------------------------+ """ - -import logbook +import sys +import zmq +import multiprocessing 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 import factory -import pytz -from pprint import pprint as pp -from datetime import datetime, timedelta - -from zipline.utils.factory import create_trading_environment from zipline.test_algorithms import TestAlgorithm -from zipline.gens.composites import SourceBundle, TransformBundle, \ +from zipline.gens.composites import \ date_sorted_sources, merged_transforms -from zipline.gens.tradegens import SpecificEquityTrades -from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform +from zipline.gens.transform import Passthrough, StatefulTransform from zipline.gens.tradesimulation import TradeSimulationClient as tsc +from logbook import Logger, NestedSetup, Processor import zipline.protocol as zp -log = logbook.Logger('Lines') +log = Logger('Lines') + +class CancelSignal(Exception): + def __init__(self): + pass class SimulatedTrading(object): - @staticmethod - def create_simulation(sources, transforms, algorithm, environment, style): + def __init__(self, + sources, + transforms, + algorithm, + environment, + style, + results_socket_uri, + context, + sim_id): - sorted = date_sorted_sources(*sources) - passthrough = StatefulTransform(Passthrough) + self.date_sorted = date_sorted_sources(*sources) + self.transforms = transforms + self.transforms.extend(StatefulTransform(Passthrough)) + self.merged = merged_transforms(self.date_sorted, *self.transforms) + self.trading_client = tsc(algorithm, environment, style) + self.gen = self.trading_client.simluate(self.merged) + self.results_uri = results_socket_uri + self.results_socket = None + self.context = context + self.sim_id = sim_id - merged = merged_transforms(sorted, passthrough, *transforms) - trading_client = tsc(algorithm, environment, style) - return trading_client.simluate(merged) + # optional process if we fork simulate into an + # independent process. + self.proc = None + def simulate(self, blocking=True): + + # for non-blocking, + if blocking: + self.run_gen() + else: + return self.fork_and_sim() + + def fork_and_sim(self): + self.proc = multiprocessing.Process(self.run_gen) + self.proc.start() + return self.proc + + def run_gen(self): + + self.open() + if self.zmq_out: + + def inject_event_data(record): + # Record the simulation time. + record.extra['algo_dt'] = self.current_dt + + data_injector = Processor(inject_event_data) + log_pipeline = NestedSetup([self.zmq_out,data_injector]) + with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''): + self.drain_gen() + # if no log socket, just run the algo normally + else: + self.drain_gen() + + def stream_results(self): + assert self.results_socket, \ + "Results socket must exist to stream results" + try: + for event in self.gen: + if event.has_key('daily_perf'): + msg = zp.PERF_FRAME(event) + else: + msg = zp.RISK_FRAME(event) + self.results_socket.send(msg) + self.signal_done() + except Exception as exc: + self.handle_exception(exc) + finally: + self.close() + + def close(self): + log.info("Closing Simulation") + + def cancel(self): + if self.proc and self.proc.is_alive(): + self.proc.terminate() + else: + self.gen.throw(CancelSignal()) + + def handle_exception(self, exc): + if isinstance(exc, CancelSignal): + # signal from monitor of an orderly shutdown, + # do nothing. + pass + else: + 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() + + log.exception("Unexpected error in run for {id}.".format(id=self.sim_id)) + + try: + log.info('{id} sending exception to monitor'\ + .format(id=self.sim_id)) + msg = zp.EXCEPTION_FRAME( + exc_traceback, + exc_type.__name__, + exc_value.message + ) + + exception_frame = zp.CONTROL_FRAME( + zp.CONTROL_PROTOCOL.EXCEPTION, + msg + ) + self.results_socket.send(exception_frame) + + 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.sockets.append(sock) + self.results_socket = sock + + self.setup_logging() + + def setup_logging(self, socket = None): + sock = socket or self.results_socket + + self.zmq_out = ZeroMQLogHandler( + socket = sock, + ) + + + # This is a class, which is instantiated later + # in run_algorithm. The class provides a generator. + self.stdout_capture = stdout_only_pipe + @staticmethod def create_test_zipline(**config): """ @@ -154,6 +288,10 @@ class SimulatedTrading(object): if not simulation_style: simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE + zmq_context = config.get('zmq_context', None) + simulation_id = config.get('simumlation_id', 'test_simulation') + results_socket_uri = config.get('results_socket_uri', None) + #------------------- # Trade Source #------------------- @@ -189,24 +327,15 @@ class SimulatedTrading(object): # Simulation #------------------- - sim = SimulatedTrading.create_simulation( + sim = SimulatedTrading( [trade_source], transforms, test_algo, trading_environment, - simulation_style) + simulation_style, + zmq_context, + results_socket_uri, + simulation_id) #------------------- return sim - - -class ZiplineException(Exception): - def __init__(self, zipline_name, msg): - self.name = zipline_name - self.message = msg - - def __str__(self): - return "Unexpected exception {line}: {msg}".format( - line=self.name, - msg=self.message - ) From 804bcb4e0c10afb4e3344da6cee47ccab62bd5ee Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Aug 2012 13:16:42 -0400 Subject: [PATCH 3/6] exceptions tests passing --- tests/test_exception_handling.py | 116 ++++++++++++++----------------- zipline/lines.py | 36 +++++----- zipline/utils/test_utils.py | 9 ++- 3 files changed, 72 insertions(+), 89 deletions(-) diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index f278ef34..53d24732 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -52,18 +52,14 @@ class ExceptionTestCase(TestCase): **self.zipline_test_config ) output, _ = drain_zipline(self, zipline) - self.assertEqual(len(output), 1) + self.assertEqual(len(output), 2) self.assertEqual(output[-1]['prefix'], 'EXCEPTION') payload = output[-1]['payload'] self.assertTrue(payload['date']) del payload['date'] check(self, payload, INITIALIZE_TB) - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) - def test_exception_in_handle_data(self): - # Simulation # ---------- self.zipline_test_config['algorithm'] = \ @@ -77,15 +73,12 @@ class ExceptionTestCase(TestCase): ) output, _ = drain_zipline(self, zipline) - - self.assertEqual(len(output), 1) + self.assertEqual(len(output), 3) self.assertEqual(output[-1]['prefix'], 'EXCEPTION') payload = output[-1]['payload'] self.assertTrue(payload['date']) del payload['date'] check(self, payload, HANDLE_DATA_TB) - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) def test_zerodivision_exception_in_handle_data(self): @@ -101,14 +94,12 @@ class ExceptionTestCase(TestCase): ) output, _ = drain_zipline(self, zipline) - self.assertEqual(len(output), 5) + self.assertEqual(len(output), 6) self.assertEqual(output[-1]['prefix'], 'EXCEPTION') payload = output[-1]['payload'] self.assertTrue(payload['date']) del payload['date'] check(self, payload, ZERO_DIV_TB) - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) # TODO: # - define more zipline failure modes: exception in other @@ -120,21 +111,12 @@ class ExceptionTestCase(TestCase): INITIALIZE_TB =\ {'message': 'Algo exception in initialize', 'name': 'Exception', - 'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.initialize_algo()', - 'lineno': 91, - 'method': 'do_work'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.do_op(self.algorithm.initialize)', - 'lineno': 74, - 'method': 'initialize_algo'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'callable_op(*args, **kwargs)', - 'lineno': 194, - 'method': 'do_op'}, + 'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'}, + {'filename': '/zipline/gens/tradesimulation.py', 'line': 'self.algo,', 'lineno': 93, 'method': 'simulate'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.algo.initialize()', + 'lineno': 123, + 'method': '__init__'}, {'filename': '/zipline/test_algorithms.py', 'line': 'raise Exception("Algo exception in initialize")', 'lineno': 166, @@ -143,25 +125,27 @@ INITIALIZE_TB =\ HANDLE_DATA_TB =\ {'message': 'Algo exception in handle_data', 'name': 'Exception', - 'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.process_event(event)', - 'lineno': 110, - 'method': 'do_work'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.run_algorithm()', - 'lineno': 158, - 'method': 'process_event'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.do_op(self.algorithm.handle_data, data)', - 'lineno': 180, - 'method': 'run_algorithm'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'callable_op(*args, **kwargs)', - 'lineno': 194, - 'method': 'do_op'}, + 'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'for message in algo_results:', + 'lineno': 100, + 'method': 'simulate'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'return self.__generator.next()', + 'lineno': 144, + 'method': 'next'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.update_current_snapshot(event)', + 'lineno': 199, + 'method': '_gen'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.simulate_current_snapshot()', + 'lineno': 221, + 'method': 'update_current_snapshot'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.algo.handle_data(self.universe)', + 'lineno': 246, + 'method': 'simulate_current_snapshot'}, {'filename': '/zipline/test_algorithms.py', 'line': 'raise Exception("Algo exception in handle_data")', 'lineno': 187, @@ -170,23 +154,25 @@ HANDLE_DATA_TB =\ ZERO_DIV_TB= \ {'message': 'integer division or modulo by zero', 'name': 'ZeroDivisionError', - 'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'}, - {'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.process_event(event)', - 'lineno': 110, - 'method': 'do_work'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.run_algorithm()', - 'lineno': 158, - 'method': 'process_event'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'self.do_op(self.algorithm.handle_data, data)', - 'lineno': 180, - 'method': 'run_algorithm'}, - {'filename': '/zipline/components/tradesimulation.py', - 'line': 'callable_op(*args, **kwargs)', - 'lineno': 194, - 'method': 'do_op'}, + 'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'for message in algo_results:', + 'lineno': 100, + 'method': 'simulate'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'return self.__generator.next()', + 'lineno': 144, + 'method': 'next'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.update_current_snapshot(event)', + 'lineno': 199, + 'method': '_gen'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.simulate_current_snapshot()', + 'lineno': 221, + 'method': 'update_current_snapshot'}, + {'filename': '/zipline/gens/tradesimulation.py', + 'line': 'self.algo.handle_data(self.universe)', + 'lineno': 246, + 'method': 'simulate_current_snapshot'}, {'filename': '/zipline/test_algorithms.py', 'line': '5/0', 'lineno': 218, 'method': 'handle_data'}]} diff --git a/zipline/lines.py b/zipline/lines.py index 6b96d1ec..b4db5de6 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -99,10 +99,10 @@ class SimulatedTrading(object): self.date_sorted = date_sorted_sources(*sources) self.transforms = transforms - self.transforms.extend(StatefulTransform(Passthrough)) + self.transforms.append(StatefulTransform(Passthrough)) self.merged = merged_transforms(self.date_sorted, *self.transforms) self.trading_client = tsc(algorithm, environment, style) - self.gen = self.trading_client.simluate(self.merged) + self.gen = self.trading_client.simulate(self.merged) self.results_uri = results_socket_uri self.results_socket = None self.context = context @@ -111,6 +111,7 @@ class SimulatedTrading(object): # optional process if we fork simulate into an # independent process. self.proc = None + self.logger = Logger(sim_id) def simulate(self, blocking=True): @@ -122,7 +123,7 @@ class SimulatedTrading(object): return self.fork_and_sim() def fork_and_sim(self): - self.proc = multiprocessing.Process(self.run_gen) + self.proc = multiprocessing.Process(target=self.run_gen) self.proc.start() return self.proc @@ -133,15 +134,16 @@ class SimulatedTrading(object): def inject_event_data(record): # Record the simulation time. - record.extra['algo_dt'] = self.current_dt + #record.extra['algo_dt'] = self.current_dt + pass data_injector = Processor(inject_event_data) log_pipeline = NestedSetup([self.zmq_out,data_injector]) with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''): - self.drain_gen() + self.stream_results() # if no log socket, just run the algo normally else: - self.drain_gen() + self.stream_results() def stream_results(self): assert self.results_socket, \ @@ -153,7 +155,8 @@ class SimulatedTrading(object): else: msg = zp.RISK_FRAME(event) self.results_socket.send(msg) - self.signal_done() + + self.signal_done() except Exception as exc: self.handle_exception(exc) finally: @@ -186,10 +189,8 @@ class SimulatedTrading(object): """ exc_type, exc_value, exc_traceback = sys.exc_info() - log.exception("Unexpected error in run for {id}.".format(id=self.sim_id)) - try: - log.info('{id} sending exception to monitor'\ + log.exception('{id} sending exception to result stream.'\ .format(id=self.sim_id)) msg = zp.EXCEPTION_FRAME( exc_traceback, @@ -197,11 +198,7 @@ class SimulatedTrading(object): exc_value.message ) - exception_frame = zp.CONTROL_FRAME( - zp.CONTROL_PROTOCOL.EXCEPTION, - msg - ) - self.results_socket.send(exception_frame) + self.results_socket.send(msg) except: log.exception("Exception while reporting simulation exception.") @@ -214,9 +211,6 @@ class SimulatedTrading(object): sock = self.context.socket(zmq.PUSH) sock.connect(self.results_uri) self.results_socket = sock - self.sockets.append(sock) - self.results_socket = sock - self.setup_logging() def setup_logging(self, socket = None): @@ -231,6 +225,10 @@ class SimulatedTrading(object): # in run_algorithm. The class provides a generator. self.stdout_capture = stdout_only_pipe + def join(self): + if self.proc: + self.proc.join() + @staticmethod def create_test_zipline(**config): """ @@ -333,8 +331,8 @@ class SimulatedTrading(object): test_algo, trading_environment, simulation_style, - zmq_context, results_socket_uri, + zmq_context, simulation_id) #------------------- diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index eda5a133..03442002 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -65,10 +65,10 @@ def drain_zipline(test, zipline): 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'], \ + 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'], + test.zipline_test_config['results_socket_uri'], test.ctx ) # Bind and connect are asynch, so allow time for bind before @@ -76,13 +76,12 @@ def drain_zipline(test, zipline): time.sleep(1) # start the simulation - zipline.simulate(blocking=False) + zipline.simulate(blocking=True) 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. - for process in zipline.sim.subprocesses: - process.join() + zipline.join() return output, transaction_count From aeb50da170fa412979c84aed92a0c8a73400355f Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Aug 2012 14:42:43 -0400 Subject: [PATCH 4/6] fixes for unit tests, back to 50/51 passing. --- tests/test_exception_handling.py | 2 -- tests/test_finance.py | 25 ++++++++----------------- tests/test_monitor.py | 6 ++++-- tests/test_perf_tracking.py | 5 ++++- tests/test_protocol.py | 4 ---- tests/test_transforms.py | 21 ++++++++++----------- zipline/finance/returns.py | 7 +++---- zipline/finance/trading.py | 4 +++- zipline/lines.py | 7 ++++++- zipline/protocol.py | 6 ++++++ zipline/utils/factory.py | 2 +- zipline/utils/test_utils.py | 25 +++++++++++-------------- 12 files changed, 56 insertions(+), 58 deletions(-) diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index 53d24732..6a091106 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -26,9 +26,7 @@ class ExceptionTestCase(TestCase): def setUp(self): self.zipline_test_config = { - 'allocator' : allocator, 'sid' : 133, - 'devel' : False, 'results_socket_uri' : allocator.lease(1)[0], 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE } diff --git a/tests/test_finance.py b/tests/test_finance.py index d108e019..e5f26240 100644 --- a/tests/test_finance.py +++ b/tests/test_finance.py @@ -11,7 +11,6 @@ from collections import defaultdict from nose.tools import timed import zipline.utils.factory as factory -import zipline.protocol as zp from zipline.test_algorithms import TestAlgorithm from zipline.finance.trading import TradingEnvironment @@ -19,10 +18,9 @@ from zipline.core.devsimulator import AddressAllocator from zipline.lines import SimulatedTrading from zipline.finance.performance import PerformanceTracker from zipline.utils.protocol_utils import ndict -from zipline.finance.trading import TransactionSimulator, SIMULATION_STYLE +from zipline.finance.trading import TransactionSimulator from zipline.utils.test_utils import \ drain_zipline, \ - check, \ setup_logger, \ teardown_logger,\ assert_single_position @@ -39,10 +37,8 @@ class FinanceTestCase(TestCase): def setUp(self): self.zipline_test_config = { - 'allocator' : allocator, - 'sid' : 133, - #'devel' : True, - 'results_socket' : allocator.lease(1)[0] + 'sid' : 133, + 'results_socket_uri' : allocator.lease(1)[0] } self.ctx = zmq.Context() @@ -60,7 +56,7 @@ class FinanceTestCase(TestCase): trading_environment ) prev = None - for trade in trade_source.event_list: + for trade in trade_source: if prev: self.assertTrue(trade.dt > prev.dt) prev = trade @@ -123,7 +119,6 @@ class FinanceTestCase(TestCase): self.zipline_test_config['order_count'] = 100 self.zipline_test_config['trade_count'] = 200 zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config) - assert_single_position(self, zipline) #@timed(DEFAULT_TIMEOUT) @@ -148,9 +143,6 @@ class FinanceTestCase(TestCase): ) output, transaction_count = drain_zipline(self, zipline) - self.assertTrue(zipline.sim.ready()) - self.assertFalse(zipline.sim.exception) - #check that the algorithm received no events self.assertEqual( 0, @@ -301,12 +293,12 @@ class FinanceTestCase(TestCase): # if present, expect transaction amounts to match orders exactly. complete_fill = params.get('complete_fill') + sid = 1 trading_environment = factory.create_trading_environment() - trade_sim = TransactionSimulator() + trade_sim = TransactionSimulator([sid]) price = [10.1] * trade_count volume = [100] * trade_count start_date = trading_environment.first_open - sid = 1 generated_trades = factory.create_trade_history( sid, @@ -330,7 +322,7 @@ class FinanceTestCase(TestCase): 'dt' : order_date }) - trade_sim.add_open_order(order) + trade_sim.place_order(order) order_date = order_date + order_interval # move after market orders to just after market next @@ -353,14 +345,13 @@ class FinanceTestCase(TestCase): self.assertEqual(order.amount, order_amount * alternator**i) - tracker = PerformanceTracker(trading_environment) + tracker = PerformanceTracker(trading_environment, [sid]) # this approximates the loop inside TradingSimulationClient transactions = [] for trade in generated_trades: if trade_delay: trade.dt = trade.dt + trade_delay - txn = trade_sim.apply_trade_to_open_orders(trade) if txn: transactions.append(txn) diff --git a/tests/test_monitor.py b/tests/test_monitor.py index 3d063954..76bb6184 100644 --- a/tests/test_monitor.py +++ b/tests/test_monitor.py @@ -14,13 +14,15 @@ class TestMonitor(TestCase): 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) + 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, ) + mon = Monitor(pub_socket, route_socket, exception_socket) mon.manage([ 'a', 'b', 'c', 'd' ]) diff --git a/tests/test_perf_tracking.py b/tests/test_perf_tracking.py index 1a77818c..2f9c1df8 100644 --- a/tests/test_perf_tracking.py +++ b/tests/test_perf_tracking.py @@ -543,7 +543,10 @@ shares in position" self.trading_environment.capital_base = 1000.0 self.trading_environment.frame_index = ['sid', 'volume', 'dt', \ 'price', 'changed'] - perf_tracker = perf.PerformanceTracker(self.trading_environment) + perf_tracker = perf.PerformanceTracker( + self.trading_environment, + [sid, sid2] + ) for event in trade_history: #create a transaction for all but diff --git a/tests/test_protocol.py b/tests/test_protocol.py index c90b09dc..d1606ed4 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -1,8 +1,6 @@ """ Test the FRAME/UNFRAME functions in the sequence expected from ziplines. """ -import pytz - from unittest2 import TestCase from datetime import datetime, timedelta from collections import defaultdict @@ -10,10 +8,8 @@ from collections import defaultdict from nose.tools import timed import zipline.utils.factory as factory -from zipline.utils import logger import zipline.protocol as zp -from zipline.finance.sources import SpecificEquityTrades DEFAULT_TIMEOUT = 5 # seconds diff --git a/tests/test_transforms.py b/tests/test_transforms.py index b9420633..27de0626 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -37,7 +37,7 @@ class ZiplineWithTransformsTestCase(TestCase): zipline = SimulatedTrading.create_test_zipline( **self.zipline_test_config ) - vwap = VWAPTransform("vwap_10", daycount=10) + vwap = VWAP("vwap_10", daycount=10) zipline.add_transform(vwap) zipline.simulate(blocking=True) @@ -49,7 +49,7 @@ class FinanceTransformsTestCase(TestCase): def setUp(self): self.trading_environment = factory.create_trading_environment() - setup_logger(self, '/var/log/qexec/qexec.log') + setup_logger(self) trade_history = factory.create_trade_history( 133, @@ -74,11 +74,11 @@ class FinanceTransformsTestCase(TestCase): ((10.0 * 100) + (10.0 * 100)) / (200.0), ((10.0 * 100) + (10.0 * 100) + (11.0 * 100)) / (300.0), # First event should get droppped here. - ((10.0 * 100) + (11.0 * 100) + (11.0 * 300)) / (500.0)] + ((10.0 * 100) + (11.0 * 100) + (11.0 * 300)) / (500.0)] # Output should match the expected. assert tnfm_vals == expected - + def test_returns(self): trade_history = factory.create_trade_history( @@ -98,13 +98,13 @@ class FinanceTransformsTestCase(TestCase): def test_moving_average(self): - + mavg = StatefulTransform( - MovingAverage, - timedelta(days = 2), + MovingAverage, + timedelta(days = 2), ['price', 'volume'] - ) - + ) + transformed = list(mavg.transform(self.source)) # Output values. tnfm_prices = [message.tnfm_value.price for message in transformed] @@ -120,7 +120,6 @@ class FinanceTransformsTestCase(TestCase): ((100.0 + 100.0 + 100.0) / 3.0), # First event should get dropped here. ((100.0 + 100.0 + 300.0) / 3.0)] - + assert tnfm_prices == expected_prices assert tnfm_volumes == expected_volumes - diff --git a/zipline/finance/returns.py b/zipline/finance/returns.py index 6e390364..2973029f 100644 --- a/zipline/finance/returns.py +++ b/zipline/finance/returns.py @@ -1,15 +1,14 @@ from collections import defaultdict -from zipline.transforms.base import BaseTransform class Returns(object): """ Class that maintains a dictionary from sids to the event representing the most recent closing price. """ - def __init__(self, days == 1): + def __init__(self, days = 1): self.days = days self.mapping = defaultdict(self._create) - + def update(self, event): """ Update and return the calculated returns for this event's sid. @@ -18,7 +17,7 @@ class Returns(object): return sid_returns def _create(self): - return ReturnsFromPriorClose(days) + return ReturnsFromPriorClose(self.days) class ReturnsFromPriorClose(object): """ diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 7bd8c7c3..baa21e58 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -31,6 +31,8 @@ class TransactionSimulator(object): self.open_orders[sid] = [] def place_order(self, order): + # initialized filled field. + order.filled = 0 self.open_orders[order.sid].append(order) def update(self, event): @@ -39,7 +41,7 @@ class TransactionSimulator(object): if event.type == zp.DATASOURCE_TYPE.TRADE: event.TRANSACTION = self.apply_trade_to_open_orders(event) return event - + def simulate_buy_all(self, event): txn = self.create_transaction( event.sid, diff --git a/zipline/lines.py b/zipline/lines.py index b4db5de6..4a54fb5d 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -162,8 +162,13 @@ class SimulatedTrading(object): finally: self.close() + def signal_done(self): + # notify monitor we're done + done_frame = zp.DONE_FRAME('succes') + self.results_socket.send(done_frame) + def close(self): - log.info("Closing Simulation") + log.info("Closing Simulation: {id}".format(id=self.sim_id)) def cancel(self): if self.proc and self.proc.is_alive(): diff --git a/zipline/protocol.py b/zipline/protocol.py index 7aa503d7..bbf80f98 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -570,6 +570,12 @@ def CANCEL_FRAME(date): return BT_UPDATE_FRAME('CANCEL', result) +def DONE_FRAME(msg): + assert isinstance(msg, basestring), \ + "Done message must be a string." + + return BT_UPDATE_FRAME('DONE', msg) + def BT_UPDATE_FRAME(prefix, payload): """ diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index e001cf07..1b881329 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -12,7 +12,7 @@ from datetime import datetime, timedelta import zipline.finance.risk as risk import zipline.protocol as zp -from zipline.finance.sources import RandomEquityTrades +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/test_utils.py b/zipline/utils/test_utils.py index 03442002..036ebe02 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -76,7 +76,7 @@ def drain_zipline(test, zipline): time.sleep(1) # start the simulation - zipline.simulate(blocking=True) + zipline.simulate(blocking=False) 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 @@ -96,16 +96,15 @@ def drain_receiver(receiver): transaction_count = 0 while True: msg = receiver.recv() - if msg == str(zp.CONTROL_PROTOCOL.DONE): + update = zp.BT_UPDATE_UNFRAME(msg) + output.append(update) + if update['prefix'] == 'PERF': + transaction_count += \ + len(update['payload']['daily_perf']['transactions']) + elif update['prefix'] == 'EXCEPTION': + break + elif update['prefix'] == 'DONE': break - else: - update = zp.BT_UPDATE_UNFRAME(msg) - output.append(update) - if update['prefix'] == 'PERF': - transaction_count += \ - len(update['payload']['daily_perf']['transactions']) - elif update['prefix'] == 'EXCEPTION': - break receiver.close() del receiver @@ -116,9 +115,6 @@ def drain_receiver(receiver): def assert_single_position(test, zipline): output, transaction_count = drain_zipline(test, zipline) - test.assertTrue(zipline.sim.ready()) - test.assertFalse(zipline.sim.exception) - test.assertEqual( test.zipline_test_config['order_count'], transaction_count @@ -127,7 +123,8 @@ def assert_single_position(test, zipline): # 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'] + perfs = [x for x in output if x['prefix'] == 'PERF'] + closing_positions = perfs[-2]['payload']['daily_perf']['positions'] test.assertEqual( len(closing_positions), From acc88793ade4ac063ab1e8b151a2431ae79837e6 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Aug 2012 15:46:56 -0400 Subject: [PATCH 5/6] patching test to match new MovingAverage init. --- tests/test_components.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/test_components.py b/tests/test_components.py index 83b96f0c..f351dac6 100644 --- a/tests/test_components.py +++ b/tests/test_components.py @@ -351,8 +351,9 @@ class ComponentTestCase(TestCase): passthrough = StatefulTransform(Passthrough) mavg_price = StatefulTransform( MovingAverage, - timedelta(minutes = 20), - ['price'] + ['price'], + market_aware = False, + delta=timedelta(minutes = 20) ) merged_gen = merged_transforms(sorted, passthrough, mavg_price) From 172ed2aafec5786d46423298b0b5da827c14b898 Mon Sep 17 00:00:00 2001 From: fawce Date: Tue, 7 Aug 2012 15:49:04 -0400 Subject: [PATCH 6/6] removed the set_trace --- zipline/gens/transform.py | 24 ++++++++++-------------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index ee3d1621..79719263 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -62,8 +62,8 @@ class StatefulTransform(object): self.append_value = tnfm_class.__dict__.get('APPENDER', False) # You only one special behavior mode can be set. - assert sum(map(int, [self.forward_all, - self.update_in_place, + assert sum(map(int, [self.forward_all, + self.update_in_place, self.append_value])) <= 1 # Create an instance of our transform class. @@ -81,14 +81,14 @@ class StatefulTransform(object): def _gen(self, stream_in): # IMPORTANT: Messages may contain pointers that are shared with # other streams, so we only manipulate copies. - + for message in stream_in: # allow upstream generators to yield None to avoid # blocking. if message == None: continue - + #TODO: refactor this to avoid unnecessary copying. assert_sort_unframe_protocol(message) @@ -117,7 +117,7 @@ class StatefulTransform(object): # TransactionSimulator. elif self.update_in_place: yield tnfm_value - + # APPENDER flag should be used to add a single new # key-value pair to the event. The new key is this # transform's namestring, and it's value is the value @@ -149,8 +149,8 @@ class EventWindow: tick. Calls self.handle_add(event) for each event added to the window. Calls self.handle_remove(event) for each event removed from the window. Subclass these methods along with init(*args, - **kwargs) to calculate metrics over the window. - + **kwargs) to calculate metrics over the window. + The market_aware flag is used to toggle whether the eventwindow calculates @@ -178,7 +178,7 @@ class EventWindow: else: assert self.delta and not self.days, \ "Non-market-aware mode requires a timedelta." - + # Set the behavior for dropping events from the back of the # event window. if self.market_aware: @@ -213,16 +213,15 @@ class EventWindow: # oldest newest # | | # V V - import nose.tools; nose.tools.set_trace() while self.drop_condition(self.ticks[0].dt, self.ticks[-1].dt): - + # popleft removes and returns the oldest tick in self.ticks popped = self.ticks.popleft() # Subclasses should override handle_remove to define # behavior for removing ticks. self.handle_remove(popped) - + def out_of_market_window(self, oldest, newest): return trading_days_between(oldest, newest) >= self.days @@ -240,6 +239,3 @@ class EventWindow: # Something is wrong if new event is older than previous. assert event.dt >= self.ticks[-1].dt, \ "Events arrived out of order in EventWindow: %s -> %s" % (event, self.ticks[0]) - - -