diff --git a/README.md b/README.md index ae9fb91b..58e3c0b4 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,6 @@ Scientific python on the Mac can be a bit confusing because of the many independ - umfpack - you need this to build scipy. ```brew install umfpack``` - swig - you need this to build scipy. ```brew install swig``` - hdf5 - you need this to build tables. ```brew install hdf5``` -- zeromq - you need this to run qbt. ```brew install zmq``` Data Sources diff --git a/docs/zipline.rst b/docs/zipline.rst index ac093601..e8a3b904 100644 --- a/docs/zipline.rst +++ b/docs/zipline.rst @@ -9,14 +9,6 @@ zipline Package :undoc-members: :show-inheritance: -:mod:`component` Module ------------------------ - -.. automodule:: zipline.component - :members: - :undoc-members: - :show-inheritance: - :mod:`lines` Module ------------------- @@ -25,22 +17,6 @@ zipline Package :undoc-members: :show-inheritance: -:mod:`messaging` Module ------------------------ - -.. automodule:: zipline.messaging - :members: - :undoc-members: - :show-inheritance: - -:mod:`monitor` Module ---------------------- - -.. automodule:: zipline.monitor - :members: - :undoc-members: - :show-inheritance: - :mod:`protocol` Module ---------------------- @@ -49,66 +25,18 @@ zipline Package :undoc-members: :show-inheritance: -:mod:`protocol_utils` Module ----------------------------- +:mod:`test_algorithms` Module +----------------------------- -.. automodule:: zipline.protocol_utils +.. automodule:: zipline.test_algorithms :members: :undoc-members: :show-inheritance: -:mod:`serial` Module --------------------- - -.. automodule:: zipline.serial - :members: - :undoc-members: - :show-inheritance: - -:mod:`simulator` Module ------------------------ - -.. automodule:: zipline.simulator - :members: - :undoc-members: - :show-inheritance: - -:mod:`sources` Module +:mod:`version` Module --------------------- -.. automodule:: zipline.sources - :members: - :undoc-members: - :show-inheritance: - -:mod:`topology` Module ----------------------- - -.. automodule:: zipline.topology - :members: - :undoc-members: - :show-inheritance: - -:mod:`topos` Module -------------------- - -.. automodule:: zipline.topos - :members: - :undoc-members: - :show-inheritance: - -:mod:`util` Module ------------------- - -.. automodule:: zipline.util - :members: - :undoc-members: - :show-inheritance: - -:mod:`zmq_utils` Module ------------------------ - -.. automodule:: zipline.zmq_utils +.. automodule:: zipline.version :members: :undoc-members: :show-inheritance: @@ -118,7 +46,10 @@ Subpackages .. toctree:: + zipline.core + zipline.data zipline.finance - zipline.test - zipline.transforms + zipline.gens + zipline.optimize + zipline.utils diff --git a/docs/zipline.test.rst b/docs/zipline.test.rst deleted file mode 100644 index b3da3bbb..00000000 --- a/docs/zipline.test.rst +++ /dev/null @@ -1,83 +0,0 @@ -test Package -============ - -:mod:`algorithms` Module ------------------------- - -.. automodule:: zipline.test.algorithms - :members: - :undoc-members: - :show-inheritance: - -:mod:`client` Module --------------------- - -.. automodule:: zipline.test.client - :members: - :undoc-members: - :show-inheritance: - -:mod:`factory` Module ---------------------- - -.. automodule:: zipline.test.factory - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_finance` Module --------------------------- - -.. automodule:: zipline.test.test_finance - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_monitor` Module --------------------------- - -.. automodule:: zipline.test.test_monitor - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_perf_tracking` Module --------------------------------- - -.. automodule:: zipline.test.test_perf_tracking - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_protocol` Module ---------------------------- - -.. automodule:: zipline.test.test_protocol - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_risk` Module ------------------------ - -.. automodule:: zipline.test.test_risk - :members: - :undoc-members: - :show-inheritance: - -:mod:`test_sanity` Module -------------------------- - -.. automodule:: zipline.test.test_sanity - :members: - :undoc-members: - :show-inheritance: - -:mod:`transform` Module ------------------------ - -.. automodule:: zipline.test.transform - :members: - :undoc-members: - :show-inheritance: - diff --git a/docs/zipline.transforms.rst b/docs/zipline.transforms.rst deleted file mode 100644 index aaf57f49..00000000 --- a/docs/zipline.transforms.rst +++ /dev/null @@ -1,19 +0,0 @@ -transforms Package -================== - -:mod:`transforms` Package -------------------------- - -.. automodule:: zipline.transforms - :members: - :undoc-members: - :show-inheritance: - -:mod:`technical` Module ------------------------ - -.. automodule:: zipline.transforms.technical - :members: - :undoc-members: - :show-inheritance: - diff --git a/etc/requirements.txt b/etc/requirements.txt index a620712d..9664715d 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -2,13 +2,6 @@ msgpack-python==0.1.12 humanhash==0.0.1 iso8601==0.1.4 -# ZeroMQ -pyzmq==2.1.11 -gevent-zeromq==0.2.2 - -# Unix -setproctitle==1.1.6 - # Logging Logbook==0.3 diff --git a/etc/requirements_sci.txt b/etc/requirements_sci.txt index f138534c..c7f5fcfd 100644 --- a/etc/requirements_sci.txt +++ b/etc/requirements_sci.txt @@ -7,7 +7,7 @@ numpy>=1.6.1 pandas==0.8.0 scipy>=0.10.0 -matplotlib==1.1.0 +#matplotlib==1.1.0 #http://sourceforge.net/projects/matplotlib/files/matplotlib/matplotlib-1.1.0/matplotlib-1.1.0.tar.gz numexpr==2.0.1 @@ -16,5 +16,3 @@ patsy==0.1.0 statsmodels>=0.5.0 scikit-learn==0.11 -# ZeroMQ -pyzmq==2.1.11 diff --git a/jenkins_setup.cfg b/jenkins_setup.cfg index 3798b1be..a6f24289 100644 --- a/jenkins_setup.cfg +++ b/jenkins_setup.cfg @@ -2,7 +2,7 @@ verbosity=2 detailed-errors=1 -#with-xcoverage=1 +with-xcoverage=1 cover-package=zipline cover-erase=1 cover-html=1 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 index 7b4fb624..be8117ce 100644 --- a/tests/test_exception_handling.py +++ b/tests/test_exception_handling.py @@ -1,15 +1,13 @@ -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.finance.slippage import FixedSlippage from zipline.lines import SimulatedTrading from zipline.gens.transform import StatefulTransform -from zipline.gens.tradesimulation import MAX_HEARTBEAT_INTERVALS +from zipline.utils.timeout import TimeoutException from zipline.utils.test_utils import ( drain_zipline, @@ -22,7 +20,6 @@ from zipline.utils.test_utils import ( DEFAULT_TIMEOUT = 15 # seconds EXTENDED_TIMEOUT = 90 -allocator = AddressAllocator(1000) class ExceptionTestCase(TestCase): @@ -31,14 +28,11 @@ class ExceptionTestCase(TestCase): def setUp(self): self.zipline_test_config = { 'sid' : 133, - 'results_socket_uri' : allocator.lease(1)[0], - 'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE + 'slippage' : FixedSlippage() } - self.ctx = zmq.Context() setup_logger(self) def tearDown(self): - self.ctx.term() teardown_logger(self) def test_datasource_exception(self): @@ -46,15 +40,15 @@ class ExceptionTestCase(TestCase): 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' + with self.assertRaises(ZeroDivisionError) as ctx: + output, _ = drain_zipline(self, zipline) + + self.assertEqual( + ctx.exception.message, + 'integer division or modulo by zero' + ) + def test_tranform_exception(self): exc_tnfm = StatefulTransform(ExceptionTransform) @@ -63,16 +57,11 @@ class ExceptionTestCase(TestCase): 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' + with self.assertRaises(AssertionError) as ctx: + output, _ = drain_zipline(self, zipline) + self.assertEqual(ctx.exception.message,'An assertion message') def test_exception_in_init(self): # Simulation @@ -86,16 +75,11 @@ class ExceptionTestCase(TestCase): 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') + with self.assertRaises(Exception) as ctx: + output, _ = drain_zipline(self, zipline) + + self.assertEqual(ctx.exception.message,'Algo exception in initialize') def test_exception_in_handle_data(self): # Simulation @@ -110,16 +94,10 @@ class ExceptionTestCase(TestCase): **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') + with self.assertRaises(Exception) as ctx: + output, _ = drain_zipline(self, zipline) + + self.assertEqual(ctx.exception.message,'Algo exception in handle_data') def test_zerodivision_exception_in_handle_data(self): @@ -134,36 +112,30 @@ class ExceptionTestCase(TestCase): **self.zipline_test_config ) - output, _ = drain_zipline(self, zipline) + with self.assertRaises(ZeroDivisionError) as ctx: + output, _ = drain_zipline(self, zipline) + + self.assertEqual(ctx.exception.message,'integer division or modulo by zero') - 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') + + with self.assertRaises(TimeoutException) as ctx: + output, _ = drain_zipline(self, zipline) + + self.assertEqual(ctx.exception.message,'Call to initialize timed out') def test_heartbeat(self): - + self.zipline_test_config['algorithm'] = \ TooMuchProcessingAlgorithm( self.zipline_test_config['sid'] @@ -171,20 +143,11 @@ class ExceptionTestCase(TestCase): 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' + with self.assertRaises(TimeoutException) as ctx: + output, _ = drain_zipline(self, zipline) - # 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') - + self.assertEqual( + ctx.exception.message, + 'Too much time spent in handle_data call' + ) diff --git a/tests/test_finance.py b/tests/test_finance.py index 87795609..8d3ab408 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 @@ -13,21 +12,20 @@ from nose.tools import timed import zipline.utils.factory as factory from zipline.finance.trading import TradingEnvironment -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 -from zipline.utils.test_utils import \ - setup_logger, \ - teardown_logger,\ +from zipline.finance.slippage import VolumeShareSlippage +from zipline.utils.test_utils import( + setup_logger, + teardown_logger, assert_single_position - +) DEFAULT_TIMEOUT = 15 # seconds EXTENDED_TIMEOUT = 90 -allocator = AddressAllocator(1000) class FinanceTestCase(TestCase): @@ -36,9 +34,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) @@ -70,7 +66,6 @@ class FinanceTestCase(TestCase): period_start = datetime(2008, 1, 1, tzinfo = pytz.utc), period_end = datetime(2008, 12, 31, tzinfo = pytz.utc), capital_base = 100000, - max_drawdown = 0.50 ) #holidays taken from: http://www.nyse.com/press/1191407641943.html new_years = datetime(2008, 1, 1, tzinfo = pytz.utc) @@ -263,7 +258,7 @@ class FinanceTestCase(TestCase): sid = 1 trading_environment = factory.create_trading_environment() - trade_sim = TransactionSimulator([sid]) + trade_sim = TransactionSimulator() price = [10.1] * trade_count volume = [100] * trade_count start_date = trading_environment.first_open @@ -320,12 +315,9 @@ class FinanceTestCase(TestCase): 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) - trade.TRANSACTION = txn - else: - trade.TRANSACTION = None + trade_sim.update(trade) + if trade.TRANSACTION: + transactions.append(trade.TRANSACTION) tracker.process_event(trade) diff --git a/tests/test_optimize.py b/tests/test_optimize.py index c16454b1..365a2a23 100644 --- a/tests/test_optimize.py +++ b/tests/test_optimize.py @@ -12,8 +12,6 @@ from zipline.core.devsimulator import AddressAllocator DEFAULT_TIMEOUT = 15 # seconds EXTENDED_TIMEOUT = 90 -allocator = AddressAllocator(1000) - from zipline.utils.test_utils import setup_logger, teardown_logger class TestUpDown(TestCase): @@ -26,7 +24,6 @@ class TestUpDown(TestCase): def setUp(self): self.zipline_test_config = { - 'allocator' : allocator, 'sid' : 133, 'trade_count' : 5, 'amplitude' : 30, diff --git a/tests/test_protocol.py b/tests/test_protocol.py deleted file mode 100644 index d1606ed4..00000000 --- a/tests/test_protocol.py +++ /dev/null @@ -1,73 +0,0 @@ -""" -Test the FRAME/UNFRAME functions in the sequence expected from ziplines. -""" -from unittest2 import TestCase -from datetime import datetime, timedelta -from collections import defaultdict - -from nose.tools import timed - -import zipline.utils.factory as factory -import zipline.protocol as zp - - -DEFAULT_TIMEOUT = 5 # seconds - -class ProtocolTestCase(TestCase): - - leased_sockets = defaultdict(list) - - def setUp(self): - #qutil.configure_logging() - self.trading_environment = factory.create_trading_environment() - - @timed(DEFAULT_TIMEOUT) - def test_trade_feed_protocol(self): - - sid = 133 - price = [10.0] * 4 - volume = [100] * 4 - - start_date = datetime.strptime("02/15/2012","%m/%d/%Y") - one_day_td = timedelta(days=1) - - trades = factory.create_trade_history( - sid, - price, - volume, - one_day_td, - self.trading_environment - ) - - for trade in trades: - #simulate data source sending frame - msg = zp.DATASOURCE_FRAME(zp.ndict(trade)) - #feed unpacking frame - recovered_trade = zp.DATASOURCE_UNFRAME(msg) - #feed sending frame - feed_msg = zp.FEED_FRAME(recovered_trade) - #transform unframing - recovered_feed = zp.FEED_UNFRAME(feed_msg) - #do a transform - trans_msg = zp.TRANSFORM_FRAME('helloworld', 2345.6) - #simulate passthrough transform -- passthrough shouldn't even - # unpack the msg, just resend. - - passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH,\ - feed_msg) - - #merge unframes transform and passthrough - trans_recovered = zp.TRANSFORM_UNFRAME(trans_msg) - pt_recovered = zp.TRANSFORM_UNFRAME(passthrough_msg) - #simulated merge - pt_recovered.PASSTHROUGH.merge(trans_recovered) - #frame the merged event - merged_msg = zp.MERGE_FRAME(pt_recovered.PASSTHROUGH) - #unframe the merge and validate values - event = zp.MERGE_UNFRAME(merged_msg) - - #check the transformed value, should only be in event, not trade. - self.assertTrue(event.helloworld == 2345.6) - event.delete('helloworld') - - self.assertEqual(zp.ndict(trade), event) diff --git a/zipline/__init__.py b/zipline/__init__.py index 31272fcb..4151cd02 100644 --- a/zipline/__init__.py +++ b/zipline/__init__.py @@ -5,10 +5,8 @@ Zipline # This is *not* a place to dump arbitrary classes/modules for convenience, # it is a place to expose the public interfaces. -import protocol # namespace from utils.protocol_utils import ndict __all__ = [ - protocol, ndict ] diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index a6dea38b..d0219895 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -41,10 +41,6 @@ Performance Tracking | | For details look at the comments for | | | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`| +-----------------+----------------------------------------------------+ - | exceeded_max_ | True if the simulation was stopped because single | - | loss | day losses exceeded the max_drawdown stipulated in | - | | trading_environment. | - +-----------------+----------------------------------------------------+ Position Tracking ================= @@ -131,7 +127,7 @@ import zipline.finance.risk as risk log = logbook.Logger('Performance') class PerformanceTracker(object): - + """ Tracks the performance of the zipline as it is running in the simulator, relays this out to the Deluge broker and then @@ -163,7 +159,6 @@ class PerformanceTracker(object): self.txn_count = 0 self.event_count = 0 self.last_dict = None - self.exceeded_max_loss = False # this performance period will span the entire simulation. self.cumulative_performance = PerformancePeriod( @@ -196,7 +191,7 @@ class PerformanceTracker(object): for sid in sid_list: self.cumulative_performance.positions[sid] = Position(sid) self.todays_performance.positions[sid] = Position(sid) - + def transform(self, stream_in): """ Main generator work loop. @@ -206,15 +201,6 @@ class PerformanceTracker(object): event.perf_message = self.handle_simulation_end() del event['TRANSACTION'] yield event - elif self.exceeded_max_loss: - # in case of max_loss, signal to downstream - # generators that we are done. - event.dt = "DONE" - event.perf_message = self.handle_simulation_end() - del event['TRANSACTION'] - yield event - # Cut off the rest of the stream. - raise StopIteration() else: event.perf_message = self.process_event(event) event.portfolio = self.get_portfolio() @@ -245,9 +231,6 @@ class PerformanceTracker(object): message = None - if self.exceeded_max_loss: - return message - assert isinstance(event, zp.ndict) self.event_count += 1 @@ -296,19 +279,6 @@ class PerformanceTracker(object): # browser. daily_update = self.to_dict() - if self.trading_environment.max_drawdown: - returns = self.todays_performance.returns - max_dd = -1 * self.trading_environment.max_drawdown - if returns < max_dd: - log.info(str(returns) + " broke through " + str(max_dd)) - log.info("Exceeded max drawdown.") - # mark the perf period with max loss flag, - # so it shows up in the update, but don't end the test - # here. Let the update go out before stopping - self.exceeded_max_loss = True - return daily_update - - #move the market day markers forward self.market_open = self.market_open + self.calendar_day @@ -343,14 +313,11 @@ class PerformanceTracker(object): # the stream will end on the last trading day, but will not trigger # an end of day, so we trigger the final market close here. - # In the case of max drawdown, we needn't close again. - if not self.exceeded_max_loss: - self.handle_market_close() + self.handle_market_close() self.risk_report = risk.RiskReport( self.returns, - self.trading_environment, - exceeded_max_loss = self.exceeded_max_loss + self.trading_environment ) risk_dict = self.risk_report.to_dict() diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 66825eba..aa7a4cd4 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -332,7 +332,7 @@ class RiskReport(): self, algorithm_returns, trading_environment, - exceeded_max_loss=False): + ): """ algorithm_returns needs to be a list of daily_return objects sorted in date ascending order @@ -340,7 +340,6 @@ class RiskReport(): self.algorithm_returns = algorithm_returns self.trading_environment = trading_environment - self.exceeded_max_loss = exceeded_max_loss self.created = epoch_now() if len(self.algorithm_returns) == 0: @@ -375,7 +374,6 @@ class RiskReport(): 'three_month' : [x.to_dict() for x in self.three_month_periods], 'six_month' : [x.to_dict() for x in self.six_month_periods], 'twelve_month' : [x.to_dict() for x in self.year_periods], - 'exceeded_max_loss' : self.exceeded_max_loss, 'created' : self.created } diff --git a/zipline/finance/slippage.py b/zipline/finance/slippage.py new file mode 100644 index 00000000..9e1decb0 --- /dev/null +++ b/zipline/finance/slippage.py @@ -0,0 +1,128 @@ +import pytz +import math +from datetime import timedelta + +import zipline.protocol as zp + +def create_transaction(sid, amount, price, dt, direction, commission): + txn = {'sid' : sid, + 'amount' : int(amount), + 'dt' : dt, + 'price' : price, + 'commission' : commission * amount * direction + } + return zp.ndict(txn) + +class VolumeShareSlippage(object): + + def __init__(self, + volume_limit=.25, + price_impact=0.1, + commission=0.03): + self.volume_limit = volume_limit + self.price_impact = price_impact + self.commission = commission + + def simulate(self, event, open_orders): + + if(event.volume == 0): + #there are zero volume events bc some stocks trade + #less frequently than once per minute. + return None + + if event.sid in open_orders: + orders = open_orders[event.sid] + orders = sorted(orders, key=lambda o: o.dt) + else: + return None + + dt = event.dt + total_order = 0 + simulated_amount = 0 + simulated_impact = 0.0 + direction = 1.0 + for order in orders: + + if(order.dt < event.dt): + + # orders are only good on the day they are issued + if order.dt.day < event.dt.day: + continue + + open_amount = order.amount - order.filled + + if(open_amount != 0): + direction = open_amount / math.fabs(open_amount) + else: + direction = 1 + + desired_order = total_order + open_amount + + volume_share = direction * (desired_order) / event.volume + if volume_share > self.volume_limit: + volume_share = self.volume_limit + simulated_amount = int(volume_share * event.volume * direction) + simulated_impact = (volume_share)**2 * self.price_impact * direction * event.price + + order.filled += (simulated_amount - total_order) + total_order = simulated_amount + + # we cap the volume share at configured % of a trade + if volume_share == self.volume_limit: + break + + + orders = [ x for x in orders if abs(x.amount - x.filled) > 0 and x.dt.day >= event.dt.day] + + open_orders[event.sid] = orders + + + if simulated_amount != 0: + return create_transaction( + event.sid, + simulated_amount, + event.price + simulated_impact, + dt.replace(tzinfo = pytz.utc), + direction, + self.commission + ) + +class FixedSlippage(object): + + def __init__(self, spread=0.0, commission=0.0): + """ + Use the fixed slippage model, which will just add/subtract a specified spread + spread/2 will be added on buys and subtracted on sells per share + commission will be charged per share + """ + self.spread = spread + self.commission = commission + + def simulate(self, event, open_orders): + if event.sid in open_orders: + orders = open_orders[event.sid] + orders = sorted(orders, key=lambda o: o.dt) + else: + return None + + amount = 0 + for order in orders: + amount += order.amount + + if(amount == 0): + return + + direction = amount / math.fabs(amount) + + txn = create_transaction( + event.sid, + amount, + event.price + self.spread/2.0, + event.dt, + direction, + self.commission + ) + + open_orders[event.sid] = [] + + return txn diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index a53b5f87..4258a4ff 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -1,33 +1,25 @@ import pytz -import math import logbook import datetime +from collections import defaultdict + import zipline.protocol as zp -from zipline.protocol import SIMULATION_STYLE +from zipline.finance.slippage import VolumeShareSlippage, FixedSlippage log = logbook.Logger('Transaction Simulator') class TransactionSimulator(object): - def __init__(self, sid_filter, style=SIMULATION_STYLE.PARTIAL_VOLUME): - self.open_orders = {} - self.txn_count = 0 - self.trade_window = datetime.timedelta(seconds=30) - self.orderTTL = datetime.timedelta(days=1) - self.commission = 0.03 + def __init__(self, slippage=None): + if slippage: + assert isinstance(slippage, (VolumeShareSlippage, FixedSlippage)) + self.slippage = slippage + else: + self.slippage = VolumeShareSlippage() - if not style or style == SIMULATION_STYLE.PARTIAL_VOLUME: - self.apply_trade_to_open_orders = self.simulate_with_partial_volume - elif style == SIMULATION_STYLE.BUY_ALL: - self.apply_trade_to_open_orders = self.simulate_buy_all - elif style == SIMULATION_STYLE.FIXED_SLIPPAGE: - self.apply_trade_to_open_orders = self.simulate_with_fixed_cost - elif style == SIMULATION_STYLE.NOOP: - self.apply_trade_to_open_orders = self.simulate_noop + self.open_orders = defaultdict(list) - for sid in sid_filter: - self.open_orders[sid] = [] def place_order(self, order): # initialized filled field. @@ -45,121 +37,9 @@ class TransactionSimulator(object): event.TRANSACTION = None # We only fill transactions on trade events. if event.type == zp.DATASOURCE_TYPE.TRADE: - event.TRANSACTION = self.apply_trade_to_open_orders(event) + event.TRANSACTION = self.slippage.simulate(event, self.open_orders) return event - def simulate_buy_all(self, event): - txn = self.create_transaction( - event.sid, - event.volume, - event.price, - event.dt, - 1 - ) - return txn - - def simulate_noop(self, event): - return None - - def simulate_with_fixed_cost(self, event): - if self.open_orders.has_key(event.sid): - orders = self.open_orders[event.sid] - orders = sorted(orders, key=lambda o: o.dt) - else: - return None - - amount = 0 - for order in orders: - amount += order.amount - - if(amount == 0): - return - - direction = amount / math.fabs(amount) - - txn = self.create_transaction( - event.sid, - amount, - event.price + 0.10, # Magic constant? - event.dt, - direction - ) - - self.open_orders[event.sid] = [] - - return txn - - def simulate_with_partial_volume(self, event): - if(event.volume == 0): - #there are zero volume events bc some stocks trade - #less frequently than once per minute. - return None - - if self.open_orders.has_key(event.sid): - orders = self.open_orders[event.sid] - orders = sorted(orders, key=lambda o: o.dt) - else: - return None - - dt = event.dt - expired = [] - total_order = 0 - simulated_amount = 0 - simulated_impact = 0.0 - direction = 1.0 - for order in orders: - - if(order.dt < event.dt): - - # orders are only good on the day they are issued - if order.dt.day < event.dt.day: - continue - - open_amount = order.amount - order.filled - - if(open_amount != 0): - direction = open_amount / math.fabs(open_amount) - else: - direction = 1 - - desired_order = total_order + open_amount - - volume_share = direction * (desired_order) / event.volume - if volume_share > .25: - volume_share = .25 - simulated_amount = int(volume_share * event.volume * direction) - simulated_impact = (volume_share)**2 * .1 * direction * event.price - - order.filled += (simulated_amount - total_order) - total_order = simulated_amount - - # we cap the volume share at 25% of a trade - if volume_share == .25: - break - - orders = [ x for x in orders if abs(x.amount - x.filled) > 0 and x.dt.day >= event.dt.day] - - self.open_orders[event.sid] = orders - - - if simulated_amount != 0: - return self.create_transaction( - event.sid, - simulated_amount, - event.price + simulated_impact, - dt.replace(tzinfo = pytz.utc), - direction - ) - - def create_transaction(self, sid, amount, price, dt, direction): - self.txn_count += 1 - txn = {'sid' : sid, - 'amount' : int(amount), - 'dt' : dt, - 'price' : price, - 'commission' : self.commission * amount * direction - } - return zp.ndict(txn) class TradingEnvironment(object): @@ -169,8 +49,7 @@ class TradingEnvironment(object): treasury_curves, period_start = None, period_end = None, - capital_base = None, - max_drawdown = None + capital_base = None ): self.trading_days = [] @@ -181,10 +60,9 @@ class TradingEnvironment(object): self.period_end = period_end self.capital_base = capital_base self.period_trading_days = None - self.max_drawdown = max_drawdown assert self.period_start <= self.period_end, \ - "Period start falls after period end." + "Period start falls after period end." for bm in benchmark_returns: self.trading_days.append(bm.date) @@ -197,7 +75,7 @@ class TradingEnvironment(object): self.first_open = self.calculate_first_open() self.last_close = self.calculate_last_close() - + self.prior_day_open = self.calculate_prior_day_open() def calculate_first_open(self): diff --git a/zipline/gens/tradesimulation.py b/zipline/gens/tradesimulation.py index 5627dc05..9c2df532 100644 --- a/zipline/gens/tradesimulation.py +++ b/zipline/gens/tradesimulation.py @@ -21,7 +21,7 @@ MAX_HEARTBEAT_INTERVALS = 15 #count class TradeSimulationClient(object): """ Generator-style class that takes the expected output of a merge, a - user algorithm, a trading environment, and a simulator style as + user algorithm, a trading environment, and a simulator slippage as arguments. Pipes the merge stream through a TransactionSimulator and a PerformanceTracker, which keep track of the current state of our algorithm's simulated universe. Results are fed to the user's @@ -52,14 +52,14 @@ class TradeSimulationClient(object): is sent to the algo. """ - def __init__(self, algo, environment, sim_style): + def __init__(self, algo, environment, slippage): self.algo = algo self.sids = algo.get_sid_filter() self.environment = environment - self.style = sim_style + self.slippage = slippage - self.ordering_client = TransactionSimulator(self.sids, style=self.style) + self.ordering_client = TransactionSimulator(self.slippage) self.perf_tracker = PerformanceTracker(self.environment, self.sids) self.algo_start = self.environment.first_open @@ -106,12 +106,12 @@ class TradeSimulationClient(object): yield message class AlgorithmSimulator(object): - + def __init__(self, order_book, algo, algo_start): - + # ========== # Algo Setup # ========== @@ -130,6 +130,9 @@ class AlgorithmSimulator(object): self.algolog = Logger("AlgoLog") self.algo.set_logger(self.algolog) + # Porived user algorithm with slippage override. + self.algo.set_slippage_override(self.override_slippage) + # Handler for heartbeats during calls to handle_data. def log_heartbeats(beat_count, stackframe): t = beat_count * HEARTBEAT_INTERVAL @@ -174,6 +177,9 @@ class AlgorithmSimulator(object): # to monkey patch sys.stdout with a logbook interface. self.stdout_capture = stdout_only_pipe + def override_slippage(self, slippage): + self.order_book.slippage = slippage + def order(self, sid, amount): """ Closure to pass into the user's algo to allow placing orders @@ -202,7 +208,7 @@ class AlgorithmSimulator(object): # simulator so that it can fill the placed order when it # receives its next message. self.order_book.place_order(order) - + def transform(self, stream_in): """ Main generator work loop. @@ -263,7 +269,7 @@ class AlgorithmSimulator(object): del event['perf_message'] self.update_universe(event) - + # Send the current state of the universe to the user's algo. self.simulate_snapshot(date) @@ -286,7 +292,7 @@ class AlgorithmSimulator(object): # Needs to be set so that we inject the proper date into algo # log/print lines. self.snapshot_dt = date - + start_tic = datetime.now() with self.heartbeat_monitor: self.algo.handle_data(self.universe) diff --git a/zipline/lines.py b/zipline/lines.py index 75c473f0..a3dfa9ba 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -59,16 +59,8 @@ 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 ( @@ -76,9 +68,9 @@ from zipline.gens.composites import ( sequential_transforms ) from zipline.gens.tradesimulation import TradeSimulationClient as tsc -from logbook import Logger +from zipline.finance.slippage import FixedSlippage -import zipline.protocol as zp +from logbook import Logger log = Logger('Lines') @@ -90,150 +82,41 @@ class SimulatedTrading(object): transforms, algorithm, environment, - style, - results_socket_uri, - context, - sim_id): + slippage): + """ + @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 + + @slippage - an object with a simulate method that takes a + trade event and returns a transaction + """ 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.trading_client = tsc(algorithm, environment, slippage) 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 - # 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): @@ -253,9 +136,10 @@ class SimulatedTrading(object): - trade_source - optional parameter to specify trades, if present. If not present :py:class:`zipline.sources.SpecificEquityTrades` is the source, with daily frequency in trades. - - simulation_style: optional parameter that configures the - :py:class:`zipline.finance.trading.TransactionSimulator`. Expects - a SIMULATION_STYLE as defined in + - slippage: optional parameter that configures the + :py:class:`zipline.gens.tradingsimulation.TransactionSimulator`. Expects + an object with a simulate mehod, such as + :py:class:`zipline.gens.tradingsimulation.FixedSlippage`. :py:mod:`zipline.finance.trading` - transforms: optional parameter that provides a list of StatefulTransform objects. @@ -293,13 +177,7 @@ class SimulatedTrading(object): # trade than order trade_count = 101 - simulation_style = config.get('simulation_style') - 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) + slippage = config.get('slippage', FixedSlippage()) #------------------- # Trade Source @@ -340,53 +218,8 @@ class SimulatedTrading(object): transforms, test_algo, trading_environment, - simulation_style, - results_socket_uri, - zmq_context, - simulation_id) + slippage, + ) #------------------- 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/optimize/factory.py b/zipline/optimize/factory.py index cb0de069..8967d28a 100644 --- a/zipline/optimize/factory.py +++ b/zipline/optimize/factory.py @@ -12,7 +12,7 @@ from zipline.utils.factory import get_next_trading_dt, create_trading_environmen from zipline.finance.sources import SpecificEquityTrades from zipline.optimize.algorithms import BuySellAlgorithm from zipline.lines import SimulatedTrading -from zipline.finance.trading import SIMULATION_STYLE +from zipline.finance.slippage import FixedSlippage from copy import copy from itertools import cycle @@ -128,7 +128,7 @@ def create_predictable_zipline(config, offset=0, simulate=True): config['trade_count'] = trade_count config['trade_source'] = source config['environment'] = trading_environment - config['simulation_style'] = SIMULATION_STYLE.FIXED_SLIPPAGE + config['slippage'] = FixedSlippage() config['devel'] = True zipline = SimulatedTrading.create_test_zipline(**config) diff --git a/zipline/protocol.py b/zipline/protocol.py index f9c1326b..f796693f 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -1,668 +1,4 @@ -""" -The messaging protocol for Zipline. - -Asserts are in place because any protocol error corresponds to a -programmer error so we want it to fail fast and in an obvious way -so it doesn't happen again. ZeroMQ follows the same philosophy. - -Notes -===== - -Msgpack -------- -Msgpack is the fastest serialization protocol in Python at the -moment. Its 100% C is typically orders of magnitude faster than -json and pickle making it awesome for ZeroMQ. - -You can only serialize Python structural primitives: strings, -numeric types, dicts, tuples and lists. Any any recursive -combinations of these. - -Basically every basestring in Python corresponds to valid -msgpack message since the protocol is highly error tolerant. -Just keep in mind that if you ever unpack a raw msgpack string -make sure it looks like what you intend and/or catch ValueError -and TypeError exceptions. - -It also has the nice benefit of never invoking ``eval`` ( unlike -json and pickle) which is a major security boon since it is -impossible to arbitrary code for evaluation through messages. - -UltraJSON ---------- -For anything going to the browser UltraJSON is the fastest -serializer, its mostly C as well. - -The same domain of serialization as msgpack applies: Python -structural primitives. It also has the additional constraint -that anything outside of UTF8 can cause serious problems, so if -you have a strong desire to JSON encode ancient Sanskrit -( admit it, we all do ), just say no. - -Data Structures -=============== - -Enum ----- - -Classic C style enumeration:: - - opts = Enum('FOO', 'BAR') - - opts.FOO # 0 - opts.BAR # 1 - opts.FOO = opts.BAR # False - -Oh, and if you do this:: - - protocol.Enum([1,2,3]) - -Your interpreter will segfault, think of this like an extreme assert. - -Namedict --------- - -Namedicts are dict like objects that have fields accessible by attribute lookup -as well as being indexable and iterable:: - - HEARTBEAT_PROTOCOL = ndict({ - 'REQ' : b'\x01', - 'REP' : b'\x02', - }) - - HEARTBEAT_PROTOCOL.REQ # syntactic sugar - HEARTBEAT_PROTOCOL.REP # oh suga suga - - HEARTBEAT_PROTOCOL['REQ'] # classic dictionary index - -Namedtuple ----------- - -From the standard library, namedtuples are great for specifying -containers for spec'ing data container objects:: - - from collections import namedtuple - - Person = namedtuple('Person', 'name age gender') - bob = Person(name='Bob', age=30, gender='male') - - bob.name # 'Bob' - bob.age # 30 - bob.gender # male - - # The slots on the tuple are also finite and read-only. This - # is a good thing, keeps us honest! - - bob.hobby = 'underwater archery' - # Will raise: - # AttributeError: 'Person' object has no attribute 'hobby' - - bob.name = 'joe' - # Will raise: - # AttributeError: can't set attribute - - # Namedtuples are normally read-only, but you can change the - # internals using a private operation. - bob._replace(gender='female') - - # You can also dump out to dictionary form: - OrderedDict([('name', 'Bob'), ('age', 30), ('gender', 'male')]) - - # Or JSON. - json.dumps(bob._asdict()) - '{"gender":"male","age":30,"name":"Bob"}' - -""" - -import msgpack -import numbers -import datetime -import pytz -import traceback -import re -import os - -from collections import namedtuple - -from utils.protocol_utils import Enum, FrameExceptionFactory, ndict, namelookup -from utils.date_utils import EPOCH, UN_EPOCH, epoch_now - -# ----------------------- -# Control Protocol -# ----------------------- - -PRODUCTION_PREFIXES = ['PERF', 'RISK', 'EXCEPTION','CANCEL','DONE', 'LOG'] -PRICE_FIELDS = ['price', 'open', 'close', 'high', 'low'] - -INVALID_CONTROL_FRAME = FrameExceptionFactory('CONTROL') - -CONTROL_STATES = Enum( - 'INIT', - 'SOURCES_READY', - 'RUNNING', - 'TERMINATE', -) - -CONTROL_PROTOCOL = Enum( - 'HEARTBEAT' , # 0 - req - 'SHUTDOWN' , # 1 - req - 'KILL' , # 2 - req - 'GO' , # - req - - 'OK' , # 3 - rep - 'DONE' , # 4 - rep - 'EXCEPTION' , # 5 - rep - 'READY' , # 6 - rep -) - -def CONTROL_FRAME(event, payload): - assert isinstance(event, int,) - assert isinstance(payload, basestring) - - return msgpack.dumps(tuple([event, payload])) - -def CONTROL_UNFRAME(msg): - """ - A status code and a message. - """ - assert isinstance(msg, basestring) - - try: - event, payload = msgpack.loads(msg) - assert isinstance(event, int) - assert isinstance(payload, basestring) - - return event, payload - except TypeError: - raise INVALID_CONTROL_FRAME(msg) - except ValueError: - raise INVALID_CONTROL_FRAME(msg) - -# ----------------------- -# Component State -# ----------------------- - -COMPONENT_TYPE = Enum( - 'SOURCE' , # 0 - 'CONDUIT' , # 1 - 'SINK' , # 2 -) - -COMPONENT_STATE = Enum( - 'OK' , # 0 - 'DONE' , # 1 - 'EXCEPTION' , # 2 -) - -# NOFAILURE - Component is either not running or has not failed -# ALGOEXCEPT - Exception thrown in the given algorithm -# HOSTEXCEPT - Exception thrown on our end. -# INTERRUPT - Manually interuptted by user - -COMPONENT_FAILURE = Enum( - 'NOFAILURE' , - 'ALGOEXCEPT' , - 'HOSTEXCEPT' , - 'INTERRUPT' , -) - -BACKTEST_STATE = Enum( - 'IDLE' , - 'QUEUED' , - 'INPROGRESS' , - 'CANCELLED' , # cancelled ( before natural completion ) - 'EXCEPTION' , # failure ( due to unnatural causes ) - 'DONE' , # done ( naturally completed ) -) - -# ----------------------- -# Datasource Protocol -# ----------------------- - -INVALID_DATASOURCE_FRAME = FrameExceptionFactory('DATASOURCE') - -def DATASOURCE_FRAME(event): - """ - Wraps any datasource payload with id and type, so that unpacking may choose - the write UNFRAME for the payload. - - :param event: ndict with following properties - - - *ds_id* an identifier that is unique to the datasource in the context of a component host (e.g. Simulator) - - *ds_type* a string denoting the datasource type. Must be on of: - - - TRADE - - DONE - - (others to follow soon) - - - *payload* a msgpack string carrying the payload for the frame - """ - assert isinstance(event.source_id, basestring) - assert isinstance(event.type, int), 'Unexpected type %s' % (event.type) - - #datasources will send sometimes send empty msgs to feel gaps - if (event.type == DATASOURCE_TYPE.EMPTY): - return msgpack.dumps(tuple([ - event.type, - event.source_id, - "EMPTY" - ])) - - elif(event.type == DATASOURCE_TYPE.TRADE): - return msgpack.dumps(tuple([ - event.type, - event.source_id, - TRADE_FRAME(event) - ])) - - elif(event.type == DATASOURCE_TYPE.DONE): - return msgpack.dumps(tuple([ - event.type, - event.source_id, - "DONE" - ])) - else: - raise INVALID_DATASOURCE_FRAME(str(event)) - -def DATASOURCE_UNFRAME(msg): - """ - - Extracts payload, and calls correct UNFRAME method based on the - datasource type passed along. - - Returns a dict containing at least: - - - source_id: instance-unique string - - type: datasource type - - dt: None, 'DONE' or a datetime object - - other properties are added based on the datasource type: - - - TRADE - - - sid - int security identifier - - price - float - - volume - int - - dt - a datetime object - - """ - try: - ds_type, source_id, payload = msgpack.loads(msg) - assert isinstance(ds_type, int) - - rval = ndict({'source_id':source_id}) - - if payload == DATASOURCE_TYPE.EMPTY: - child_value = ndict({'dt':None}) - elif(ds_type == DATASOURCE_TYPE.TRADE): - child_value = TRADE_UNFRAME(payload) - elif(ds_type == DATASOURCE_TYPE.DONE): - child_value = ndict({'dt' : 'DONE'}) - else: - raise INVALID_DATASOURCE_FRAME(msg) - - rval.merge(child_value) - return rval - - except TypeError: - raise INVALID_DATASOURCE_FRAME(msg) - except ValueError: - raise INVALID_DATASOURCE_FRAME(msg) - -# ----------------------- -# Feed Protocol -# ----------------------- - -INVALID_FEED_FRAME = FrameExceptionFactory('FEED') - -def FEED_FRAME(event): - """ - :param event: a ndict with at least - - - source_id - - type - - dt - """ - assert isinstance(event, ndict), 'unknown type %s' % str(event) - source_id = event.source_id - ds_type = event.type - PACK_DATE(event) - payload = event.as_dict() - return msgpack.dumps(payload) - -def FEED_UNFRAME(msg): - try: - payload = msgpack.loads(msg) - #TODO: anything we can do to assert more about the content of the dict? - assert isinstance(payload, dict) - rval = ndict(payload) - assert rval.source_id - assert rval.type in DATASOURCE_TYPE - assert rval.dt - UNPACK_DATE(rval) - return rval - except TypeError: - raise INVALID_FEED_FRAME(msg) - except ValueError: - raise INVALID_FEED_FRAME(msg) - -# ----------------------- -# Transform Protocol -# ----------------------- - -INVALID_TRANSFORM_FRAME = FrameExceptionFactory('TRANSFORM') - -def TRANSFORM_FRAME(name, value): - assert isinstance(name, basestring) - if value == None: - return msgpack.dumps(tuple([name, TRANSFORM_TYPE.EMPTY])) - return msgpack.dumps(tuple([name, value])) - -def TRANSFORM_UNFRAME(msg): - """ - :rtype: ndict with : - """ - try: - - name, value = msgpack.loads(msg) - if(value == TRANSFORM_TYPE.EMPTY): - return ndict({name : None}) - #TODO: anything we can do to assert more about the content of the dict? - assert isinstance(name, basestring) - if(name == TRANSFORM_TYPE.PASSTHROUGH): - value = FEED_UNFRAME(value) - - return ndict({name : value}) - except TypeError: - raise INVALID_TRANSFORM_FRAME(msg) - except ValueError: - raise INVALID_TRANSFORM_FRAME(msg) - -# ----------------------- -# Merge Protocol -# ----------------------- -INVALID_MERGE_FRAME = FrameExceptionFactory('MERGE') - -def MERGE_FRAME(event): - """ - :param event: a nameddict with at least: - - - source_id - - type - """ - assert isinstance(event, ndict) - PACK_DATE(event) - payload = event.as_dict() - return msgpack.dumps(payload) - -def MERGE_UNFRAME(msg): - try: - payload = msgpack.loads(msg) - #TODO: anything we can do to assert more about the content of the dict? - assert isinstance(payload, dict) - payload = ndict(payload) - UNPACK_DATE(payload) - return payload - except TypeError: - raise INVALID_MERGE_FRAME(msg) - except ValueError: - raise INVALID_MERGE_FRAME(msg) - - -# ----------------------- -# Trades -# ----------------------- -# -# - Should only be called from inside DATASOURCE_ (UN)FRAME. - -def TRADE_FRAME(event): - """ - :param event: should be a ndict with: - - - ds_id -- the datasource id sending this trade out - - sid -- the security id - - price -- float of the price printed for the trade - - volume -- int for shares in the trade - - dt -- datetime for the trade - - """ - assert isinstance(event, ndict) - assert event.type == DATASOURCE_TYPE.TRADE - assert isinstance(event.sid, int) - for field in PRICE_FIELDS: - assert isinstance(event[field], numbers.Real) - assert isinstance(event.volume, numbers.Integral) - PACK_DATE(event) - return msgpack.dumps(tuple([ - event.sid, - event.price, - event.open, - event.close, - event.high, - event.low, - event.volume, - event.dt, - event.type - ])) - -def TRADE_UNFRAME(msg): - try: - packed = msgpack.loads(msg) - sid, price, open, close, high, low, volume, dt, source_type = packed - - assert isinstance(sid, int) - assert isinstance(price, numbers.Real) - assert isinstance(volume, numbers.Integral) - rval = ndict({ - 'sid' : sid, - 'price' : price, - 'open' : open, - 'close' : close, - 'high' : high, - 'low' : low, - 'volume' : volume, - 'dt' : dt, - 'type' : source_type - }) - UNPACK_DATE(rval) - return rval - except TypeError: - raise INVALID_TRADE_FRAME(msg) - except ValueError: - raise INVALID_TRADE_FRAME(msg) - -# ----------------------- -# Performance and Risk -# ----------------------- - -def PERF_FRAME(perf): - """ - Frame the performance update created at the end of each simulated trading - day. The msgpack is a tuple with the first element statically set to 'PERF'. - Like RISK_FRAME, this method calls BT_UPDATE_FRAME internally, so that - clients can call BT_UPDATE_UNFRAME for all messages from the backtest. - - :param perf: the dictionary created by zipline.trade_client.perf - :rvalue: a msgpack string - """ - - #TODO: add asserts... - - assert isinstance(perf['started_at'], datetime.datetime) - assert isinstance(perf['period_start'], datetime.datetime) - assert isinstance(perf['period_end'], datetime.datetime) - - assert isinstance(perf['daily_perf'], dict) - assert isinstance(perf['cumulative_perf'], dict) - - tp = perf['daily_perf'] - cp = perf['cumulative_perf'] - - assert isinstance(tp['transactions'], list) - # we never want to send transactions for the cumulative period. - # performance.py should never send them, but just to be safe: - assert not cp.has_key('transactions') - assert isinstance(tp['positions'], list) - assert isinstance(cp['positions'], list) - assert isinstance(tp['period_close'], datetime.datetime) - assert isinstance(tp['period_open'], datetime.datetime) - assert isinstance(cp['period_close'], datetime.datetime) - assert isinstance(cp['period_open'], datetime.datetime) - - perf['started_at'] = EPOCH(perf['started_at']) - perf['period_start'] = EPOCH(perf['period_start']) - perf['period_end'] = EPOCH(perf['period_end']) - tp['period_close'] = EPOCH(tp['period_close']) - tp['period_open'] = EPOCH(tp['period_open']) - cp['period_close'] = EPOCH(cp['period_close']) - cp['period_open'] = EPOCH(cp['period_open']) - - tp['transactions'] = convert_transactions(tp['transactions']) - - return BT_UPDATE_FRAME('PERF', perf) - -def convert_transactions(transactions): - results = [] - for txn in transactions: - txn['date'] = EPOCH(txn['dt']) - del(txn['dt']) - results.append(txn) - return results - -def RISK_FRAME(risk): - return BT_UPDATE_FRAME('RISK', risk) - -def EXCEPTION_FRAME(exception_tb, name, message): - stack_list = traceback.extract_tb(exception_tb) - rlist = [] - for stack in stack_list: - filename = shorten_filename(stack[0]) - # default the line to empty string rather than None - line = '' - if stack[3]: - line = stack[3] - rstack = { - 'filename' : filename, - 'lineno' : stack[1], - 'method' : stack[2], - 'line' : line - } - rlist.append(rstack) - result = { - 'date' : epoch_now(), - 'stack' : rlist, - 'name' : name, - 'message' : message - } - - return BT_UPDATE_FRAME('EXCEPTION', result) - -def shorten_filename(filename): - if filename == None: - return None - - # check if the path contains zipeline - # looks for a zipline directory in the middle of the path - # this will work on - # /zipline/workspace/zipline/core/component.py, but fail for - # /home/fawce/projects/zipline/zipline/core/component.py - path_re = r'(.*)(/zipline/.*[.]py)' - match = re.search(path_re, filename) - - if match and match.lastindex == 2: - filepath = match.group(2) - return os.path.join('/zipline',filepath) - else: - # return just the filename. - head, tail = os.path.split(filename) - return tail - -def CANCEL_FRAME(date): - result = { - 'date' : EPOCH(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): - """ - Frames prepared by RISK_FRAME and PERF_FRAME methods are sent via the same - socket. This method provides a prefix to allow for muxing the messages - onto a single socket. - """ - return msgpack.dumps(tuple([prefix, payload])) - -def BT_UPDATE_UNFRAME(msg): - """ - Risk, Perf, and LOG framing methods prefix the payload with - a shorthand for their type. That way, all messages received from the socket - can be PERF_FRAMED(), whether they are risk, perf, or log. - """ - prefix, payload = msgpack.loads(msg, use_list=True) - return dict(prefix=prefix, payload=payload) - -# ----------------------- -# Date Helpers -# ----------------------- - -def PACK_DATE(event): - """ - Packs the datetime property of event into msgpack'able longs. - This function should be called purely for its side effects. - The event's 'dt' property is replaced by a tuple of integers - - - year, month, day, hour, minute, second, microsecond - - PACK_DATE and UNPACK_DATE are inverse operations. - - :param event: event must a ndict with a property named 'dt' that is a datetime. - :rtype: None - """ - assert isinstance(event.dt, datetime.datetime) - # utc only please - assert event.dt.tzinfo == pytz.utc - event['dt'] = date_to_tuple(event['dt']) - -def date_to_tuple(dt): - year, month, day, hour, minute, second = dt.timetuple()[0:6] - micros = dt.microsecond - return tuple([year, month, day, hour, minute, second, micros]) - -def UNPACK_DATE(event): - """ - Unpacks the datetime property of event from msgpack'able longs. - This function should be called purely for its side effects. - The event's 'dt' property is converted to a datetime by reading and then - combining a tuple of integers. - - UNPACK_DATE and PACK_DATE are inverse operations. - - :param tuple event: event must a ndict with: - - - a property named 'dt_tuple' that is a tuple of integers \ - representing the date and time in UTC. - - dt_tuple must have year, month, day, hour, minute, second, and microsecond - - :rtype: None - """ - assert isinstance(event.dt, tuple) - assert len(event.dt) == 7 - for item in event.dt: - assert isinstance(item, numbers.Integral) - event.dt = tuple_to_date(event.dt) - -def tuple_to_date(date_tuple): - year, month, day, hour, minute, second, micros = date_tuple - dt = datetime.datetime(year, month, day, hour, minute, second) - dt = dt.replace(microsecond = micros, tzinfo = pytz.utc) - return dt +from utils.protocol_utils import Enum, ndict, namelookup # Datasource type should completely determine the other fields of a # message with its type. @@ -688,68 +24,3 @@ FINANCE_COMPONENT = namelookup({ 'TRADING_CLIENT' : 'TRADING_CLIENT', 'PORTFOLIO_CLIENT' : 'PORTFOLIO_CLIENT', }) - - -# the simulation style enumerates the available transaction simulation -# strategies. -SIMULATION_STYLE = Enum( - 'PARTIAL_VOLUME', - 'BUY_ALL', - 'FIXED_SLIPPAGE', - 'NOOP' -) - -#Global variables for the fields we extract out of a standard logbook record. -LOG_FIELDS = set(['func_name', 'lineno', 'time', 'msg',\ - 'level', 'channel', ]) -LOG_EXTRA_FIELDS = set(['algo_dt',]) -LOG_DONE = "DONE" - -def LOG_FRAME(payload): - """ - Expects a dictionary of the form: - { - 'algo_dt' : 1199223000, #Algo simulation date. - 'time' : 1199223001, #Realtime date of log creation. - 'func_name' : 'foo', - 'lineno' : 46, - 'msg' : 'Successfully disintegrated llama #3', - 'level' : 4, #Logbook enum - 'channel' : 'MyLogger' - } - - Frame checks that we have all expected fields and exports an - event/payload dict as JSON. - """ - - assert isinstance(payload, dict), \ - "LOG_FRAME expected a dict" - - assert payload.has_key('algo_dt'), \ - "LOG_FRAME with no algo_dt" - assert payload.has_key('time'), \ - "LOG_FRAME with no time" - assert payload.has_key('channel'),\ - "LOG_FRAME with no channel" - assert payload.has_key('level'),\ - "LOG_FRAME with no level" - assert payload.has_key('msg'),\ - "LOG_FRAME with no message" - - # truncation will only work with strings and msgpack will - # preserve primitives. - payload['msg'] = str(payload['msg']) - - return BT_UPDATE_FRAME('LOG', payload) - -def LOG_UNFRAME(msg): - """ - msg should be a tuple of ('LOG',dict) - """ - record = msgpack.loads(msg) - assert isinstance(record, tuple) - assert len(record) == 2 - assert record[0] == 'LOG' - payload = record[1] - - return payload diff --git a/zipline/test_algorithms.py b/zipline/test_algorithms.py index eb883469..bcff7f81 100644 --- a/zipline/test_algorithms.py +++ b/zipline/test_algorithms.py @@ -44,6 +44,12 @@ The algorithm must expose methods: self.Portfolio[sid(133)]['cost_basis'] + - set_slippage_override: method that accepts a callable. Will + be set as the value of the set_slippage_override method of + the trading_client. This allows an algorithm to change the + slippage model used to predict transactions based on orders + and trade events. + """ @@ -90,7 +96,10 @@ class TestAlgorithm(): def get_sid_filter(self): return self.sid_filter -# + def set_slippage_override(self, slippage_callable): + pass + + # class HeavyBuyAlgorithm(): """ This algorithm will send a specified number of orders, to allow unit tests @@ -128,6 +137,9 @@ class HeavyBuyAlgorithm(): def get_sid_filter(self): return [self.sid] + def set_slippage_override(self, slippage_callable): + pass + class NoopAlgorithm(object): """ Dolce fa niente. @@ -151,6 +163,9 @@ class NoopAlgorithm(object): def get_sid_filter(self): return [] + def set_slippage_override(self, slippage_callable): + pass + class ExceptionAlgorithm(object): """ Throw an exception from the method name specified in the @@ -194,6 +209,9 @@ class ExceptionAlgorithm(object): else: return [self.sid] + def set_slippage_override(self, slippage_callable): + pass + class DivByZeroAlgorithm(): def __init__(self, sid): @@ -221,6 +239,9 @@ class DivByZeroAlgorithm(): def get_sid_filter(self): return [self.sid] + def set_slippage_override(self, slippage_callable): + pass + class InitializeTimeoutAlgorithm(): def __init__(self, sid): self.sid = sid @@ -229,7 +250,8 @@ class InitializeTimeoutAlgorithm(): def initialize(self): import time from zipline.gens.tradesimulation import INIT_TIMEOUT - time.sleep(INIT_TIMEOUT + 1) + time.sleep(INIT_TIMEOUT + 1000) + def set_order(self, order_callable): pass @@ -245,7 +267,10 @@ class InitializeTimeoutAlgorithm(): def get_sid_filter(self): return [self.sid] - + + def set_slippage_override(self, slippage_callable): + pass + class TooMuchProcessingAlgorithm(): def __init__(self, sid): self.sid = sid @@ -270,7 +295,10 @@ class TooMuchProcessingAlgorithm(): def get_sid_filter(self): return [self.sid] - + + def set_slippage_override(self, slippage_callable): + pass + class TimeoutAlgorithm(): def __init__(self, sid): @@ -298,6 +326,9 @@ class TimeoutAlgorithm(): def get_sid_filter(self): return [self.sid] + def set_slippage_override(self, slippage_callable): + pass + class TestPrintAlgorithm(): def __init__(self, sid): @@ -322,6 +353,9 @@ class TestPrintAlgorithm(): def get_sid_filter(self): return [self.sid] + def set_slippage_override(self, slippage_callable): + pass + class TestLoggingAlgorithm(): def __init__(self, sid): @@ -345,3 +379,6 @@ class TestLoggingAlgorithm(): def get_sid_filter(self): return [self.sid] + + def set_slippage_override(self, slippage_callable): + pass diff --git a/zipline/utils/date_utils.py b/zipline/utils/date_utils.py index 2819d4e9..8e2aeaaa 100644 --- a/zipline/utils/date_utils.py +++ b/zipline/utils/date_utils.py @@ -3,7 +3,6 @@ from collections import namedtuple import time import pytz import iso8601 -import calendar from dateutil import rrule from datetime import datetime, date, timedelta from dateutil.relativedelta import * @@ -141,3 +140,9 @@ def date_to_datetime(t): dt = datetime.fromordinal(t.toordinal()) dt = dt.replace(tzinfo = pytz.utc) return dt + +def tuple_to_date(date_tuple): + year, month, day, hour, minute, second, micros = date_tuple + dt = datetime(year, month, day, hour, minute, second) + dt = dt.replace(microsecond = micros, tzinfo = pytz.utc) + return dt diff --git a/zipline/utils/factory.py b/zipline/utils/factory.py index cf2168fb..7146c0de 100644 --- a/zipline/utils/factory.py +++ b/zipline/utils/factory.py @@ -9,8 +9,10 @@ from os.path import join, abspath, dirname from operator import attrgetter from datetime import datetime, timedelta +from zipline.utils.date_utils import tuple_to_date +from zipline.utils.protocol_utils import ndict + import zipline.finance.risk as risk -import zipline.protocol as zp from zipline.gens.tradegens import RandomEquityTrades from zipline.gens.tradegens import SpecificEquityTrades @@ -33,7 +35,7 @@ def load_market_data(): bm_list = msgpack.loads(fp_bm.read()) bm_returns = [] for packed_date, returns in bm_list: - event_dt = zp.tuple_to_date(packed_date) + event_dt = tuple_to_date(packed_date) #event_dt = event_dt.replace( # hour=0, # minute=0, @@ -49,7 +51,7 @@ def load_market_data(): tr_list = msgpack.loads(fp_tr.read()) tr_curves = {} for packed_date, curve in tr_list: - tr_dt = zp.tuple_to_date(packed_date) + tr_dt = tuple_to_date(packed_date) #tr_dt = tr_dt.replace(hour=0, minute=0, second=0, tzinfo=pytz.utc) tr_curves[tr_dt] = curve @@ -95,7 +97,7 @@ def create_trade_history(sid, prices, amounts, interval, trading_calendar): return trades def create_txn(sid, price, amount, datetime, btrid=None): - txn = zp.ndict({ + txn = ndict({ 'sid' : sid, 'amount' : amount, 'dt' : datetime, diff --git a/zipline/utils/log_utils.py b/zipline/utils/log_utils.py index ea1abf18..c7d7b9b9 100644 --- a/zipline/utils/log_utils.py +++ b/zipline/utils/log_utils.py @@ -1,20 +1,10 @@ import logbook -import zmq -import pytz -import datetime - -from logbook import NOTSET -from logbook.handlers import Handler, FileHandler - -from zipline.protocol import LOG_FRAME, LOG_FIELDS, \ - LOG_EXTRA_FIELDS - from contextlib import contextmanager log = logbook.Logger("LogUtils") -class redirecter(object): +class redirector(object): def __init__(self, logger, name): self.logger = logger self.buffer = bytes() @@ -33,7 +23,7 @@ class redirecter(object): self.logger.error(out_form) self.buffer = bytes() -class log_redirecter(object): +class log_redirector(object): def __init__(self, logger): self.logger = logger @@ -57,8 +47,8 @@ def stdout_pipe(logger, pipe_name): import sys orig_fds = sys.stdout, sys.stderr - sys.stderr = redirecter(logger, pipe_name) - sys.stdout = redirecter(logger, pipe_name) + sys.stderr = redirector(logger, pipe_name) + sys.stdout = redirector(logger, pipe_name) yield sys.stderr.flush() @@ -72,95 +62,8 @@ def stdout_only_pipe(logger, pipe_name): """ import sys orig_fd = sys.stdout - sys.stdout = log_redirecter(logger) + sys.stdout = log_redirector(logger) yield sys.stdout.flush() sys.stdout = orig_fd - -class ZeroMQLogHandler(Handler): - """ - A handler that takes messages captured from the user algorithm stdout - and transforms them into LOG_FRAMES suitable for database storage. - Setup is similar to logbook.queues.ZeroMQHandler, except we connect - instead of binding and we extract record fields into a dict. - """ - - def __init__(self, socket=None, level=NOTSET, filter=None, bubble=False, - context=None, fds = LOG_FIELDS, extra_fds = LOG_EXTRA_FIELDS): - Handler.__init__(self, level, filter, bubble) - - try: - import zmq - except ImportError: - raise RuntimeError('The pyzmq library is required for ' - 'the ZeroMQHandler.') - #: the zero mq context - self.context = context - #: the zero mq socket. - self.socket = socket #self.context.socket(zmq.PUSH) - - #self.uri = uri - #if uri is not None: - # self.socket.connect(uri) - - self.fds = fds - self.extra_fds = extra_fds - - def export_record(self, record): - """ - Extract relevant fields from a log record, fiddling with datetime - fields to make json happy. - """ - from zipline.utils.date_utils import EPOCH - - #Needed to extract record info from dictionary. - record.pull_information() - - #Logbook stores record times as datetime objects, which - #can't be serialized by JSON, so we need to convert to - #unix epoch representation. - - - #Do the same if algo_dt is a datetime object. - if record.extra.has_key('algo_dt'): - algo_dt = record.extra['algo_dt'] - - if isinstance(algo_dt, datetime.datetime): - algo_dt = EPOCH(algo_dt.replace(tzinfo = pytz.utc)) - record.extra['algo_dt'] = algo_dt - - data = {} - - #Extract all the fields we care about from LogRecord's internal - #dictionary. - - for field in iter(self.fds): - if record.__dict__.has_key(field): - data[field] = record.__dict__[field] - else: - data[field] = None - - for field in iter(self.extra_fds): - if record.extra.has_key(field): - data[field] = record.extra[field] - else: - data[field] = None - - if data['time']: - assert isinstance(data['time'], datetime.datetime) - - time = data['time'].replace(tzinfo = pytz.utc) - #logbook measures time in utc already, no need to convert. - data['time'] = EPOCH(time) - - return data - - def emit(self, record): - """Extract relevant fields and send info as JSON over a zmq socket.""" - payload = self.export_record(record) - self.socket.send(LOG_FRAME(payload)) - - def close(self): - pass - #self.socket.close() diff --git a/zipline/utils/protocol_utils.py b/zipline/utils/protocol_utils.py index c3ae9352..62f3aab9 100644 --- a/zipline/utils/protocol_utils.py +++ b/zipline/utils/protocol_utils.py @@ -5,7 +5,7 @@ from collections import MutableMapping def Enum(*options): """ - Fast enums are very important when we want really tight zmq + Fast enums are very important when we want really tight loops. These are probably going to evolve into pure C structs anyways so might as well get going on that. """ diff --git a/zipline/utils/test_utils.py b/zipline/utils/test_utils.py index 513352b1..4fdc3918 100644 --- a/zipline/utils/test_utils.py +++ b/zipline/utils/test_utils.py @@ -1,7 +1,3 @@ -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,80 +61,40 @@ 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 drain_zipline(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 'daily_perf' in update: transaction_count += \ - len(update['payload']['daily_perf']['transactions']) - elif update['prefix'] == 'EXCEPTION': - break - elif update['prefix'] == 'DONE': - break - - if count and msg_counter >= count: - break - - receiver.close() - del receiver + len(update['daily_perf']['transactions']) 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') +def assert_single_position(test, zipline): - test.assertEqual( - test.zipline_test_config['order_count'], - transaction_count - ) + output, transaction_count = drain_zipline(test, zipline) + + if 'expected_transactions' in test.zipline_test_config: + test.assertEqual( + test.zipline_test_config['expected_transactions'], + transaction_count + ) + else: + test.assertEqual( + test.zipline_test_config['order_count'], + transaction_count + ) # 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), @@ -153,17 +109,7 @@ def assert_single_position(test, zipline, blocking=False): "Portfolio should have one position in " + str(sid) ) - -def launch_component(component): - proc = multiprocessing.Process(target=component.run) - proc.start() - return proc - - -def launch_monitor(monitor): - proc = multiprocessing.Process(target=monitor.run) - proc.start() - return proc + return output, transaction_count class ExceptionSource(object):