diff --git a/tests/test_components.py b/tests/test_components.py index 33255883..f351dac6 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] @@ -262,8 +267,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) @@ -312,7 +318,12 @@ class ComponentTestCase(TestCase): sorted = date_sorted_sources(self.source_a, self.source_b) passthrough = StatefulTransform(Passthrough) - mavg_price = StatefulTransform(MovingAverage, timedelta(minutes = 20), ['price']) + mavg_price = StatefulTransform( + MovingAverage, + ['price'], + market_aware=False, + delta=timedelta(minutes = 20), + ) merged = merged_transforms(sorted, passthrough, mavg_price) @@ -340,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) diff --git a/tests/test_exception_handling.py b/tests/test_exception_handling.py index 144568d1..6a091106 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -26,11 +26,9 @@ 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 + 'sid' : 133, + 'results_socket_uri' : allocator.lease(1)[0], + 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE } self.ctx = zmq.Context() setup_logger(self) @@ -52,18 +50,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 +71,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 +92,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 +109,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 +123,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 +152,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/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 7d417ba6..0835cb71 100644 --- a/tests/test_transforms.py +++ b/tests/test_transforms.py @@ -20,7 +20,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, @@ -77,7 +77,6 @@ class FinanceTransformsTestCase(TestCase): fields = ['price', 'volume'], delta = timedelta(days = 2), ) - transformed = list(mavg.transform(self.source)) # Output values. tnfm_prices = [message.tnfm_value.price for message in transformed] @@ -100,4 +99,3 @@ class FinanceTransformsTestCase(TestCase): assert tnfm_prices == expected_prices assert tnfm_volumes == expected_volumes - 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/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/gens/tradegens.py b/zipline/gens/tradegens.py index 9a0f7406..2e8f6bea 100644 --- a/zipline/gens/tradegens.py +++ b/zipline/gens/tradegens.py @@ -8,8 +8,7 @@ import pytz 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, tzinfo=pytz.utc), delta = timedelta(minutes = 1), diff --git a/zipline/gens/transform.py b/zipline/gens/transform.py index fdf545a1..dbb7c75b 100644 --- a/zipline/gens/transform.py +++ b/zipline/gens/transform.py @@ -63,8 +63,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. @@ -82,14 +82,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) @@ -118,7 +118,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 @@ -150,8 +150,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 @@ -179,7 +179,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: @@ -215,14 +215,14 @@ class EventWindow: # | | # V V 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 @@ -239,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]) - - - diff --git a/zipline/gens/utils.py b/zipline/gens/utils.py index 83372753..1ac85df6 100644 --- a/zipline/gens/utils.py +++ b/zipline/gens/utils.py @@ -66,6 +66,17 @@ 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 sum_true(bool_iterable): """ Takes an iterable of boolean values and returns the number of diff --git a/zipline/lines.py b/zipline/lines.py index a5a3858e..4a54fb5d 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -59,112 +59,180 @@ before invoking simulate. | __init__. | +---------------------------------+ """ - -import inspect -import logbook -#import zipline.utils.factory as factory - -from zipline.components import DataSource -from zipline.transforms import BaseTransform +import sys +import zmq +import multiprocessing 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.log_utils import ZeroMQLogHandler, stdout_only_pipe +from zipline.utils import factory -log = logbook.Logger('Lines') +from zipline.test_algorithms import TestAlgorithm + +from zipline.gens.composites import \ + date_sorted_sources, merged_transforms +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 = Logger('Lines') + +class CancelSignal(Exception): + def __init__(self): + pass 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. + def __init__(self, + sources, + transforms, + algorithm, + environment, + style, + results_socket_uri, + context, + sim_id): - 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. - """ + self.date_sorted = date_sorted_sources(*sources) + self.transforms = transforms + 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.simulate(self.merged) + self.results_uri = results_socket_uri + self.results_socket = None + self.context = context + self.sim_id = sim_id - def __init__(self, **config): + # optional process if we fork simulate into an + # independent process. + self.proc = None + self.logger = Logger(sim_id) + + + 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(target=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 + 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.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 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 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: {id}".format(id=self.sim_id)) + + 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): """ - :param config: a dict with the following required properties:: + All exceptions inside any component should boil back to + this handler. - - 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` + Will inform the system that the component has failed and how it + has failed. """ - 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) + 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.") - self.leased_sockets = [] - self.sim_context = None + 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() - 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], - } + def setup_logging(self, socket = None): + sock = socket or self.results_socket - 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.zmq_out = ZeroMQLogHandler( + socket = sock, ) - 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 ) + # This is a class, which is instantiated later + # 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): @@ -173,7 +241,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 +255,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'] #-------------------- @@ -223,6 +291,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 #------------------- @@ -236,6 +308,12 @@ class SimulatedTrading(object): trade_count, trading_environment ) + + #------------------- + # Transforms + #------------------- + transforms = config.get('transforms', []) + #------------------- # Create the Algo #------------------- @@ -248,157 +326,19 @@ 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( + [trade_source], + transforms, + test_algo, + trading_environment, + simulation_style, + results_socket_uri, + zmq_context, + simulation_id) #------------------- - zipline.add_source(trade_source) - - 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): - self.name = zipline_name - self.message = msg - - def __str__(self): - return "Unexpected exception {line}: {msg}".format( - line=self.name, - msg=self.message - ) + return sim 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 004f542a..1b881329 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.gens.tradegens 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 diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index eda5a133..036ebe02 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 @@ -81,8 +81,7 @@ def drain_zipline(test, zipline): # 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 @@ -97,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 @@ -117,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 @@ -128,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),