wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-08-02 12:30:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into batch_window

Browse Source
This commit is contained in:
Thomas Wiecki
2012-09-17 12:49:02 -04:00
parent ea5cbaba56 71c1f755b3
commit a729435964
27 changed files with 340 additions and 1671 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
-1
View File
@@ -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
docs/zipline.rst
+10 -79
View File
@@ -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
docs/zipline.test.rst
-83
View File
@@ -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:
docs/zipline.transforms.rst
-19
View File
@@ -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:
etc/requirements.txt
-7
View File
@@ -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
etc/requirements_sci.txt
+1 -3
View File
@@ -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
jenkins_setup.cfg
+1 -1
View File
@@ -2,7 +2,7 @@
verbosity=2
detailed-errors=1
#with-xcoverage=1
with-xcoverage=1
cover-package=zipline
cover-erase=1
cover-html=1
pavement.py
-1
View File
@@ -110,7 +110,6 @@ options(
# Because I'm lazy
stuff_i_want_in_my_debug_shell = [
('qutil', 'zipline.util', []),
('zmq', 'zmq', []),
]
@task
tests/test_exception_handling.py
+40 -77
View File
@@ -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'
)
tests/test_finance.py
+9 -17
View File
@@ -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)
tests/test_optimize.py
-3
View File
@@ -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,
tests/test_protocol.py
-73
View File
@@ -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)
zipline/__init__.py
-2
View File
@@ -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
]
zipline/finance/performance.py
+4 -37
View File
@@ -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()
zipline/finance/risk.py
+1 -3
View File
@@ -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
}
zipline/finance/slippage.py
+128
View File
@@ -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
zipline/finance/trading.py
+14 -136
View File
@@ -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):
zipline/gens/tradesimulation.py
+15 -9
View File
@@ -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)
zipline/lines.py
+35 -202
View File
@@ -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
zipline/optimize/factory.py
+2 -2
View File
@@ -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)
zipline/protocol.py
+1 -730
View File
@@ -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 <transform_name>:<transform_value>
"""
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
zipline/test_algorithms.py
+41 -4
View File
@@ -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
zipline/utils/date_utils.py
+6 -1
View File
@@ -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
zipline/utils/factory.py
+6 -4
View File
@@ -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,
zipline/utils/log_utils.py
+5 -102
View File
@@ -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()
zipline/utils/protocol_utils.py
+1 -1
View File
@@ -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.
"""
zipline/utils/test_utils.py
+20 -74
View File
@@ -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):