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

Merge branch 'new_world_order' of github.com:quantopian/zipline into new_world_order

Browse Source 
Conflicts:
	tests/test_transforms.py
	zipline/finance/returns.py
This commit is contained in:
scottsanderson
2012-08-07 17:08:08 -04:00
parent 1d9da39fbb 172ed2aafe
commit 23a62a942a
16 changed files with 330 additions and 416 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_components.py
+22 -10
View File
@@ -8,7 +8,8 @@ from collections import defaultdict
from zipline.gens.composites import date_sorted_sources, merged_transforms
from zipline.core.devsimulator import AddressAllocator
from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform
from zipline.gens.transform import Passthrough, StatefulTransform
from zipline.gens.mavg import MovingAverage
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from zipline.utils.factory import create_trading_environment
@@ -113,7 +114,8 @@ class ComponentTestCase(TestCase):
monitor,
socket_uri,
DATASOURCE_FRAME,
DATASOURCE_UNFRAME
DATASOURCE_UNFRAME,
"source_a"
)
launch_monitor(monitor)
@@ -171,7 +173,8 @@ class ComponentTestCase(TestCase):
monitor,
socket_uris[0],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME
DATASOURCE_UNFRAME,
trade_gen_a.get_hash()
)
comp_b = Component(
@@ -179,7 +182,8 @@ class ComponentTestCase(TestCase):
monitor,
socket_uris[1],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME
DATASOURCE_UNFRAME,
trade_gen_b.get_hash()
)
comp_c = Component(
@@ -187,7 +191,8 @@ class ComponentTestCase(TestCase):
monitor,
socket_uris[2],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME
DATASOURCE_UNFRAME,
trade_gen_c.get_hash()
)
sources = [comp_a, comp_b, comp_c]
@@ -262,8 +267,9 @@ class ComponentTestCase(TestCase):
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
timedelta(minutes = 20),
['price']
['price'],
market_aware = False,
delta=timedelta(minutes = 20)
)
merged_gen = merged_transforms(sorted, passthrough, mavg_price)
@@ -312,7 +318,12 @@ class ComponentTestCase(TestCase):
sorted = date_sorted_sources(self.source_a, self.source_b)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(MovingAverage, timedelta(minutes = 20), ['price'])
mavg_price = StatefulTransform(
MovingAverage,
['price'],
market_aware=False,
delta=timedelta(minutes = 20),
)
merged = merged_transforms(sorted, passthrough, mavg_price)
@@ -340,8 +351,9 @@ class ComponentTestCase(TestCase):
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
timedelta(minutes = 20),
['price']
['price'],
market_aware = False,
delta=timedelta(minutes = 20)
)
merged_gen = merged_transforms(sorted, passthrough, mavg_price)
tests/test_exception_handling.py
+54 -70
View File
@@ -26,11 +26,9 @@ class ExceptionTestCase(TestCase):
def setUp(self):
self.zipline_test_config = {
'allocator' : allocator,
'sid' : 133,
'devel' : False,
'results_socket' : allocator.lease(1)[0],
'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE
'sid' : 133,
'results_socket_uri' : allocator.lease(1)[0],
'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE
}
self.ctx = zmq.Context()
setup_logger(self)
@@ -52,18 +50,14 @@ class ExceptionTestCase(TestCase):
**self.zipline_test_config
)
output, _ = drain_zipline(self, zipline)
self.assertEqual(len(output), 1)
self.assertEqual(len(output), 2)
self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
payload = output[-1]['payload']
self.assertTrue(payload['date'])
del payload['date']
check(self, payload, INITIALIZE_TB)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
def test_exception_in_handle_data(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
@@ -77,15 +71,12 @@ class ExceptionTestCase(TestCase):
)
output, _ = drain_zipline(self, zipline)
self.assertEqual(len(output), 1)
self.assertEqual(len(output), 3)
self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
payload = output[-1]['payload']
self.assertTrue(payload['date'])
del payload['date']
check(self, payload, HANDLE_DATA_TB)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
def test_zerodivision_exception_in_handle_data(self):
@@ -101,14 +92,12 @@ class ExceptionTestCase(TestCase):
)
output, _ = drain_zipline(self, zipline)
self.assertEqual(len(output), 5)
self.assertEqual(len(output), 6)
self.assertEqual(output[-1]['prefix'], 'EXCEPTION')
payload = output[-1]['payload']
self.assertTrue(payload['date'])
del payload['date']
check(self, payload, ZERO_DIV_TB)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
# TODO:
# - define more zipline failure modes: exception in other
@@ -120,21 +109,12 @@ class ExceptionTestCase(TestCase):
INITIALIZE_TB =\
{'message': 'Algo exception in initialize',
'name': 'Exception',
'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'},
{'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'},
{'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.initialize_algo()',
'lineno': 91,
'method': 'do_work'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.do_op(self.algorithm.initialize)',
'lineno': 74,
'method': 'initialize_algo'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'callable_op(*args, **kwargs)',
'lineno': 194,
'method': 'do_op'},
'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'},
{'filename': '/zipline/gens/tradesimulation.py', 'line': 'self.algo,', 'lineno': 93, 'method': 'simulate'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.algo.initialize()',
'lineno': 123,
'method': '__init__'},
{'filename': '/zipline/test_algorithms.py',
'line': 'raise Exception("Algo exception in initialize")',
'lineno': 166,
@@ -143,25 +123,27 @@ INITIALIZE_TB =\
HANDLE_DATA_TB =\
{'message': 'Algo exception in handle_data',
'name': 'Exception',
'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'},
{'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'},
{'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.process_event(event)',
'lineno': 110,
'method': 'do_work'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.run_algorithm()',
'lineno': 158,
'method': 'process_event'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.do_op(self.algorithm.handle_data, data)',
'lineno': 180,
'method': 'run_algorithm'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'callable_op(*args, **kwargs)',
'lineno': 194,
'method': 'do_op'},
'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'for message in algo_results:',
'lineno': 100,
'method': 'simulate'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'return self.__generator.next()',
'lineno': 144,
'method': 'next'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.update_current_snapshot(event)',
'lineno': 199,
'method': '_gen'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.simulate_current_snapshot()',
'lineno': 221,
'method': 'update_current_snapshot'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.algo.handle_data(self.universe)',
'lineno': 246,
'method': 'simulate_current_snapshot'},
{'filename': '/zipline/test_algorithms.py',
'line': 'raise Exception("Algo exception in handle_data")',
'lineno': 187,
@@ -170,23 +152,25 @@ HANDLE_DATA_TB =\
ZERO_DIV_TB= \
{'message': 'integer division or modulo by zero',
'name': 'ZeroDivisionError',
'stack': [{'filename': '/zipline/core/component.py', 'line': 'self._run()', 'lineno': 210, 'method': 'run'},
{'filename': '/zipline/core/component.py', 'line': 'self.loop()', 'lineno': 201, 'method': '_run'},
{'filename': '/zipline/core/component.py', 'line': 'self.do_work()', 'lineno': 241, 'method': 'loop'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.process_event(event)',
'lineno': 110,
'method': 'do_work'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.run_algorithm()',
'lineno': 158,
'method': 'process_event'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'self.do_op(self.algorithm.handle_data, data)',
'lineno': 180,
'method': 'run_algorithm'},
{'filename': '/zipline/components/tradesimulation.py',
'line': 'callable_op(*args, **kwargs)',
'lineno': 194,
'method': 'do_op'},
'stack': [{'filename': '/zipline/lines.py', 'line': 'for event in self.gen:', 'lineno': 152, 'method': 'stream_results'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'for message in algo_results:',
'lineno': 100,
'method': 'simulate'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'return self.__generator.next()',
'lineno': 144,
'method': 'next'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.update_current_snapshot(event)',
'lineno': 199,
'method': '_gen'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.simulate_current_snapshot()',
'lineno': 221,
'method': 'update_current_snapshot'},
{'filename': '/zipline/gens/tradesimulation.py',
'line': 'self.algo.handle_data(self.universe)',
'lineno': 246,
'method': 'simulate_current_snapshot'},
{'filename': '/zipline/test_algorithms.py', 'line': '5/0', 'lineno': 218, 'method': 'handle_data'}]}
tests/test_finance.py
+8 -17
View File
@@ -11,7 +11,6 @@ from collections import defaultdict
from nose.tools import timed
import zipline.utils.factory as factory
import zipline.protocol as zp
from zipline.test_algorithms import TestAlgorithm
from zipline.finance.trading import TradingEnvironment
@@ -19,10 +18,9 @@ from zipline.core.devsimulator import AddressAllocator
from zipline.lines import SimulatedTrading
from zipline.finance.performance import PerformanceTracker
from zipline.utils.protocol_utils import ndict
from zipline.finance.trading import TransactionSimulator, SIMULATION_STYLE
from zipline.finance.trading import TransactionSimulator
from zipline.utils.test_utils import \
drain_zipline, \
check, \
setup_logger, \
teardown_logger,\
assert_single_position
@@ -39,10 +37,8 @@ class FinanceTestCase(TestCase):
def setUp(self):
self.zipline_test_config = {
'allocator' : allocator,
'sid' : 133,
#'devel' : True,
'results_socket' : allocator.lease(1)[0]
'sid' : 133,
'results_socket_uri' : allocator.lease(1)[0]
}
self.ctx = zmq.Context()
@@ -60,7 +56,7 @@ class FinanceTestCase(TestCase):
trading_environment
)
prev = None
for trade in trade_source.event_list:
for trade in trade_source:
if prev:
self.assertTrue(trade.dt > prev.dt)
prev = trade
@@ -123,7 +119,6 @@ class FinanceTestCase(TestCase):
self.zipline_test_config['order_count'] = 100
self.zipline_test_config['trade_count'] = 200
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
assert_single_position(self, zipline)
#@timed(DEFAULT_TIMEOUT)
@@ -148,9 +143,6 @@ class FinanceTestCase(TestCase):
)
output, transaction_count = drain_zipline(self, zipline)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
#check that the algorithm received no events
self.assertEqual(
0,
@@ -301,12 +293,12 @@ class FinanceTestCase(TestCase):
# if present, expect transaction amounts to match orders exactly.
complete_fill = params.get('complete_fill')
sid = 1
trading_environment = factory.create_trading_environment()
trade_sim = TransactionSimulator()
trade_sim = TransactionSimulator([sid])
price = [10.1] * trade_count
volume = [100] * trade_count
start_date = trading_environment.first_open
sid = 1
generated_trades = factory.create_trade_history(
sid,
@@ -330,7 +322,7 @@ class FinanceTestCase(TestCase):
'dt' : order_date
})
trade_sim.add_open_order(order)
trade_sim.place_order(order)
order_date = order_date + order_interval
# move after market orders to just after market next
@@ -353,14 +345,13 @@ class FinanceTestCase(TestCase):
self.assertEqual(order.amount, order_amount * alternator**i)
tracker = PerformanceTracker(trading_environment)
tracker = PerformanceTracker(trading_environment, [sid])
# this approximates the loop inside TradingSimulationClient
transactions = []
for trade in generated_trades:
if trade_delay:
trade.dt = trade.dt + trade_delay
txn = trade_sim.apply_trade_to_open_orders(trade)
if txn:
transactions.append(txn)
tests/test_monitor.py
+4 -2
View File
@@ -14,13 +14,15 @@ class TestMonitor(TestCase):
def test_init(self):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
exception_socket = 'tcp://127.0.0.1:5002'
mon = Monitor(pub_socket, route_socket)
mon = Monitor(pub_socket, route_socket, exception_socket)
mon.manage([])
def test_init_topology(self):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
exception_socket = 'tcp://127.0.0.1:5002'
mon = Monitor(pub_socket, route_socket, )
mon = Monitor(pub_socket, route_socket, exception_socket)
mon.manage([ 'a', 'b', 'c', 'd' ])
tests/test_perf_tracking.py
+4 -1
View File
@@ -543,7 +543,10 @@ shares in position"
self.trading_environment.capital_base = 1000.0
self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
'price', 'changed']
perf_tracker = perf.PerformanceTracker(self.trading_environment)
perf_tracker = perf.PerformanceTracker(
self.trading_environment,
[sid, sid2]
)
for event in trade_history:
#create a transaction for all but
tests/test_protocol.py
-4
View File
@@ -1,8 +1,6 @@
"""
Test the FRAME/UNFRAME functions in the sequence expected from ziplines.
"""
import pytz
from unittest2 import TestCase
from datetime import datetime, timedelta
from collections import defaultdict
@@ -10,10 +8,8 @@ from collections import defaultdict
from nose.tools import timed
import zipline.utils.factory as factory
from zipline.utils import logger
import zipline.protocol as zp
from zipline.finance.sources import SpecificEquityTrades
DEFAULT_TIMEOUT = 5 # seconds
tests/test_transforms.py
+1 -3
View File
@@ -20,7 +20,7 @@ class FinanceTransformsTestCase(TestCase):
def setUp(self):
self.trading_environment = factory.create_trading_environment()
setup_logger(self, '/var/log/qexec/qexec.log')
setup_logger(self)
trade_history = factory.create_trade_history(
133,
@@ -77,7 +77,6 @@ class FinanceTransformsTestCase(TestCase):
fields = ['price', 'volume'],
delta = timedelta(days = 2),
)
transformed = list(mavg.transform(self.source))
# Output values.
tnfm_prices = [message.tnfm_value.price for message in transformed]
@@ -100,4 +99,3 @@ class FinanceTransformsTestCase(TestCase):
assert tnfm_prices == expected_prices
assert tnfm_volumes == expected_volumes
zipline/__init__.py
-6
View File
@@ -6,15 +6,9 @@ Zipline
# it is a place to expose the public interfaces.
import protocol # namespace
from core.monitor import Monitor
from lines import SimulatedTrading
from core.host import ComponentHost
from utils.protocol_utils import ndict
__all__ = [
SimulatedTrading,
Monitor,
ComponentHost,
protocol,
ndict
]
zipline/finance/trading.py
+3 -1
View File
@@ -31,6 +31,8 @@ class TransactionSimulator(object):
self.open_orders[sid] = []
def place_order(self, order):
# initialized filled field.
order.filled = 0
self.open_orders[order.sid].append(order)
def update(self, event):
@@ -39,7 +41,7 @@ class TransactionSimulator(object):
if event.type == zp.DATASOURCE_TYPE.TRADE:
event.TRANSACTION = self.apply_trade_to_open_orders(event)
return event
def simulate_buy_all(self, event):
txn = self.create_transaction(
event.sid,
zipline/gens/tradegens.py
+1 -2
View File
@@ -8,8 +8,7 @@ import pytz
from itertools import chain, cycle, ifilter, izip
from datetime import datetime, timedelta
from zipline.utils.factory import create_trade
from zipline.gens.utils import hash_args
from zipline.gens.utils import hash_args, create_trade
def date_gen(start = datetime(2006, 6, 6, 12, tzinfo=pytz.utc),
delta = timedelta(minutes = 1),
zipline/gens/transform.py
+10 -13
View File
@@ -63,8 +63,8 @@ class StatefulTransform(object):
self.append_value = tnfm_class.__dict__.get('APPENDER', False)
# You only one special behavior mode can be set.
assert sum(map(int, [self.forward_all,
self.update_in_place,
assert sum(map(int, [self.forward_all,
self.update_in_place,
self.append_value])) <= 1
# Create an instance of our transform class.
@@ -82,14 +82,14 @@ class StatefulTransform(object):
def _gen(self, stream_in):
# IMPORTANT: Messages may contain pointers that are shared with
# other streams, so we only manipulate copies.
for message in stream_in:
# allow upstream generators to yield None to avoid
# blocking.
if message == None:
continue
#TODO: refactor this to avoid unnecessary copying.
assert_sort_unframe_protocol(message)
@@ -118,7 +118,7 @@ class StatefulTransform(object):
# TransactionSimulator.
elif self.update_in_place:
yield tnfm_value
# APPENDER flag should be used to add a single new
# key-value pair to the event. The new key is this
# transform's namestring, and it's value is the value
@@ -150,8 +150,8 @@ class EventWindow:
tick. Calls self.handle_add(event) for each event added to the
window. Calls self.handle_remove(event) for each event removed
from the window. Subclass these methods along with init(*args,
**kwargs) to calculate metrics over the window.
**kwargs) to calculate metrics over the window.
The market_aware flag is used to toggle whether the eventwindow
calculates
@@ -179,7 +179,7 @@ class EventWindow:
else:
assert self.delta and not self.days, \
"Non-market-aware mode requires a timedelta."
# Set the behavior for dropping events from the back of the
# event window.
if self.market_aware:
@@ -215,14 +215,14 @@ class EventWindow:
# | |
# V V
while self.drop_condition(self.ticks[0].dt, self.ticks[-1].dt):
# popleft removes and returns the oldest tick in self.ticks
popped = self.ticks.popleft()
# Subclasses should override handle_remove to define
# behavior for removing ticks.
self.handle_remove(popped)
def out_of_market_window(self, oldest, newest):
return trading_days_between(oldest, newest) >= self.days
@@ -239,6 +239,3 @@ class EventWindow:
# Something is wrong if new event is older than previous.
assert event.dt >= self.ticks[-1].dt, \
"Events arrived out of order in EventWindow: %s -> %s" % (event, self.ticks[0])
zipline/gens/utils.py
+11
View File
@@ -66,6 +66,17 @@ def hash_args(*args, **kwargs):
hasher.update(combined)
return hasher.hexdigest()
def create_trade(sid, price, amount, datetime, source_id = "test_factory"):
row = ndict({
'source_id' : source_id,
'type' : DATASOURCE_TYPE.TRADE,
'sid' : sid,
'dt' : datetime,
'price' : price,
'volume' : amount
})
return row
def sum_true(bool_iterable):
"""
Takes an iterable of boolean values and returns the number of
zipline/lines.py
+177 -237
View File
@@ -59,112 +59,180 @@ before invoking simulate.
| __init__. |
+---------------------------------+
"""
import inspect
import logbook
#import zipline.utils.factory as factory
from zipline.components import DataSource
from zipline.transforms import BaseTransform
import sys
import zmq
import multiprocessing
from zipline.test_algorithms import TestAlgorithm
from zipline.components import TradeSimulationClient
from zipline.core.process import ProcessSimulator
from zipline.core.monitor import Monitor
from zipline.finance.trading import SIMULATION_STYLE
from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe
from zipline.utils import factory
log = logbook.Logger('Lines')
from zipline.test_algorithms import TestAlgorithm
from zipline.gens.composites import \
date_sorted_sources, merged_transforms
from zipline.gens.transform import Passthrough, StatefulTransform
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from logbook import Logger, NestedSetup, Processor
import zipline.protocol as zp
log = Logger('Lines')
class CancelSignal(Exception):
def __init__(self):
pass
class SimulatedTrading(object):
"""
Zipline with::
- _no_ data sources.
- Trade simulation client, which is available to send callbacks on
events and also accept orders to be simulated.
- An order data source, which will receive orders from the trade
simulation client, and feed them into the event stream to be
serialized and order alongside all other data source events.
- transaction simulation transformation, which receives the order
events and estimates a theoretical execution price and volume.
def __init__(self,
sources,
transforms,
algorithm,
environment,
style,
results_socket_uri,
context,
sim_id):
All components in this zipline are subject to heartbeat checks and
a control monitor, which can kill the entire zipline in the event of
exceptions in one of the components or an external request to end the
simulation.
"""
self.date_sorted = date_sorted_sources(*sources)
self.transforms = transforms
self.transforms.append(StatefulTransform(Passthrough))
self.merged = merged_transforms(self.date_sorted, *self.transforms)
self.trading_client = tsc(algorithm, environment, style)
self.gen = self.trading_client.simulate(self.merged)
self.results_uri = results_socket_uri
self.results_socket = None
self.context = context
self.sim_id = sim_id
def __init__(self, **config):
# optional process if we fork simulate into an
# independent process.
self.proc = None
self.logger = Logger(sim_id)
def simulate(self, blocking=True):
# for non-blocking,
if blocking:
self.run_gen()
else:
return self.fork_and_sim()
def fork_and_sim(self):
self.proc = multiprocessing.Process(target=self.run_gen)
self.proc.start()
return self.proc
def run_gen(self):
self.open()
if self.zmq_out:
def inject_event_data(record):
# Record the simulation time.
#record.extra['algo_dt'] = self.current_dt
pass
data_injector = Processor(inject_event_data)
log_pipeline = NestedSetup([self.zmq_out,data_injector])
with log_pipeline.threadbound(), self.stdout_capture(self.logger, ''):
self.stream_results()
# if no log socket, just run the algo normally
else:
self.stream_results()
def stream_results(self):
assert self.results_socket, \
"Results socket must exist to stream results"
try:
for event in self.gen:
if event.has_key('daily_perf'):
msg = zp.PERF_FRAME(event)
else:
msg = zp.RISK_FRAME(event)
self.results_socket.send(msg)
self.signal_done()
except Exception as exc:
self.handle_exception(exc)
finally:
self.close()
def signal_done(self):
# notify monitor we're done
done_frame = zp.DONE_FRAME('succes')
self.results_socket.send(done_frame)
def close(self):
log.info("Closing Simulation: {id}".format(id=self.sim_id))
def cancel(self):
if self.proc and self.proc.is_alive():
self.proc.terminate()
else:
self.gen.throw(CancelSignal())
def handle_exception(self, exc):
if isinstance(exc, CancelSignal):
# signal from monitor of an orderly shutdown,
# do nothing.
pass
else:
self.signal_exception(exc)
def signal_exception(self, exc=None):
"""
:param config: a dict with the following required properties::
All exceptions inside any component should boil back to
this handler.
- algorithm: a class that follows the algorithm protocol. See
:py:meth:`zipline.finance.trading.TradeSimulationClient.add_algorithm
for details.
- trading_environment: an instance of
:py:class:`zipline.trading.TradingEnvironment`
- allocator: an instance of
:py:class:`zipline.simulator.AddressAllocator`
- simulation_style: optional parameter that configures the
:py:class:`zipline.finance.trading.TransactionSimulator`. Expects
a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading`
Will inform the system that the component has failed and how it
has failed.
"""
assert isinstance(config, dict)
self.algorithm = config['algorithm']
self.allocator = config['allocator']
self.trading_environment = config['trading_environment']
self.sim_style = config.get('simulation_style')
self.send_sighup = config.get('send_sighup', False)
exc_type, exc_value, exc_traceback = sys.exc_info()
try:
log.exception('{id} sending exception to result stream.'\
.format(id=self.sim_id))
msg = zp.EXCEPTION_FRAME(
exc_traceback,
exc_type.__name__,
exc_value.message
)
self.results_socket.send(msg)
except:
log.exception("Exception while reporting simulation exception.")
self.leased_sockets = []
self.sim_context = None
def open(self):
if not self.context:
self.context = zmq.Context()
if self.results_uri:
sock = self.context.socket(zmq.PUSH)
sock.connect(self.results_uri)
self.results_socket = sock
self.setup_logging()
sockets = self.allocate_sockets(7)
addresses = {
'sync_address' : sockets[0],
'data_address' : sockets[1],
'feed_address' : sockets[2],
'merge_address' : sockets[3],
# TODO: this refers to the results of the merge, a
# horribly confusing name for the socket.
'results_address' : sockets[4],
}
def setup_logging(self, socket = None):
sock = socket or self.results_socket
self.monitor = Monitor(
# pub socket
sockets[5],
# route socket
sockets[6],
# exception socket to match tradesimclient's result
# socket, because we want to relay exceptions to the
# same listener
config['results_socket'],
send_sighup=self.send_sighup
self.zmq_out = ZeroMQLogHandler(
socket = sock,
)
self.started = False
self.sim = ProcessSimulator(addresses)
self.clients = {}
self.trading_client = TradeSimulationClient(
self.trading_environment,
self.sim_style,
config['results_socket'],
self.algorithm
)
self.add_client(self.trading_client)
# setup all sources
self.sources = {}
#setup transforms
self.transforms = {}
self.sim.register_monitor( self.monitor )
# This is a class, which is instantiated later
# in run_algorithm. The class provides a generator.
self.stdout_capture = stdout_only_pipe
def join(self):
if self.proc:
self.proc.join()
@staticmethod
def create_test_zipline(**config):
@@ -173,7 +241,6 @@ class SimulatedTrading(object):
- environment - a \
:py:class:`zipline.finance.trading.TradingEnvironment`
- allocator - a :py:class:`zipline.simulator.AddressAllocator`
- sid - an integer, which will be used as the security ID.
- order_count - the number of orders the test algo will place,
defaults to 100
@@ -188,10 +255,11 @@ class SimulatedTrading(object):
- simulation_style: optional parameter that configures the
:py:class:`zipline.finance.trading.TransactionSimulator`. Expects
a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading`
- transforms: optional parameter that provides a list
of StatefulTransform objects.
"""
assert isinstance(config, dict)
allocator = config['allocator']
sid = config['sid']
#--------------------
@@ -223,6 +291,10 @@ class SimulatedTrading(object):
if not simulation_style:
simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE
zmq_context = config.get('zmq_context', None)
simulation_id = config.get('simumlation_id', 'test_simulation')
results_socket_uri = config.get('results_socket_uri', None)
#-------------------
# Trade Source
#-------------------
@@ -236,6 +308,12 @@ class SimulatedTrading(object):
trade_count,
trading_environment
)
#-------------------
# Transforms
#-------------------
transforms = config.get('transforms', [])
#-------------------
# Create the Algo
#-------------------
@@ -248,157 +326,19 @@ class SimulatedTrading(object):
order_count
)
if config.has_key('results_socket'):
results_socket = config['results_socket']
else:
results_socket = None
#-------------------
# Simulation
#-------------------
zipline = SimulatedTrading(**{
'algorithm' : test_algo,
'trading_environment' : trading_environment,
'allocator' : allocator,
'simulation_style' : simulation_style,
'results_socket' : results_socket,
})
sim = SimulatedTrading(
[trade_source],
transforms,
test_algo,
trading_environment,
simulation_style,
results_socket_uri,
zmq_context,
simulation_id)
#-------------------
zipline.add_source(trade_source)
return zipline
def add_source(self, source):
"""
Adds the source to the zipline, sets the sid filter of the
source to the algorithm's sid filter.
"""
assert isinstance(source, DataSource)
self.check_started()
source.set_filter('sid', self.algorithm.get_sid_filter())
self.sim.register_components([source])
# ``id`` is name of source_id, ``get_id`` is the class name
self.sources[source.get_id] = source
def add_transform(self, transform):
assert isinstance(transform, BaseTransform)
self.check_started()
self.sim.register_components([transform])
self.transforms[transform.get_id] = transform
def add_client(self, client):
assert isinstance(client, TradeSimulationClient)
self.check_started()
self.sim.register_components([client])
self.clients[client.get_id] = client
def check_started(self):
if self.started:
raise ZiplineException("TradeSimulation", "You cannot add \
components after the simulation has begun.")
def get_cumulative_performance(self):
return self.trading_client.perf.cumulative_performance.to_dict()
def allocate_sockets(self, n):
"""
Allocate sockets local to this line, track them so
we can gc after test run.
"""
assert isinstance(n, int)
assert n > 0
leased = self.allocator.lease(n)
self.leased_sockets.extend(leased)
return leased
@property
def components(self):
"""
Return the component instances inside of this topology
"""
base = set(self.sim.components.values())
transforms = set(self.transforms.values())
sources = set(self.sources.values())
return base | transforms | sources
@property
def topology(self):
"""
Returns the Component names in the topology of the
backtest.
"""
# A complete topology is the union of three classes of
# components added individually to the simulation client
# at various places.
#
# base : ['FEED', 'MERGE', 'TRADING_CLIENT', 'PASSTHROUGH']
# transforms : ['vwap__01', ... ]
# sources : ['MongoTradeHistory', ... ]
base = set(self.sim.components.keys())
transforms = set(self.transforms.keys())
sources = set(self.sources.keys())
return base | transforms | sources
def setup_monitor(self):
"""
Prepare the monitor to manage the topology specified
by this line.
"""
self.monitor.manage(self.topology)
def simulate(self, blocking=True):
self.setup_monitor()
self.started = True
self.sim_context = self.sim.simulate()
# If we're using a threaded simulator block on the pool
# of thread since we're only ever in a test and we don't
# generally monitor the state of the system as a hold at
# the supervisory layer
# TODO: better way of identifying concurrency substrate
if blocking:
for process in self.sim.subprocesses:
process.join()
@property
def is_success(self):
# TODO: other assertions?
if self.sim.did_clean_shutdown():
return True
else:
return False
#--------------------------------
# Component property accessors
#--------------------------------
def get_positions(self):
"""
returns current positions as a dict. draws from the cumulative
performance period in the performance tracker.
"""
perf = self.trading_client.perf.cumulative_performance
positions = perf.get_positions()
return positions
class ZiplineException(Exception):
def __init__(self, zipline_name, msg):
self.name = zipline_name
self.message = msg
def __str__(self):
return "Unexpected exception {line}: {msg}".format(
line=self.name,
msg=self.message
)
return sim
zipline/protocol.py
+6
View File
@@ -570,6 +570,12 @@ def CANCEL_FRAME(date):
return BT_UPDATE_FRAME('CANCEL', result)
def DONE_FRAME(msg):
assert isinstance(msg, basestring), \
"Done message must be a string."
return BT_UPDATE_FRAME('DONE', msg)
def BT_UPDATE_FRAME(prefix, payload):
"""
zipline/utils/factory.py
+16 -33
View File
@@ -12,7 +12,9 @@ from datetime import datetime, timedelta
import zipline.finance.risk as risk
import zipline.protocol as zp
from zipline.finance.sources import SpecificEquityTrades, RandomEquityTrades
from zipline.gens.tradegens import RandomEquityTrades
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.utils import create_trade
from zipline.finance.trading import TradingEnvironment
# TODO
@@ -69,16 +71,6 @@ def create_trading_environment(year=2006):
return trading_environment
def create_trade(sid, price, amount, datetime, source_id = "test_factory"):
row = zp.ndict({
'source_id' : source_id,
'type' : zp.DATASOURCE_TYPE.TRADE,
'sid' : sid,
'dt' : datetime,
'price' : price,
'volume' : amount
})
return row
def get_next_trading_dt(current, interval, trading_calendar):
next = current
@@ -220,29 +212,20 @@ def create_minutely_trade_source(sids, trade_count, trading_environment):
)
def create_trade_source(sids, trade_count, trade_time_increment, trading_environment):
trade_history = []
price = [10.1] * trade_count
volume = [100] * trade_count
#Set up source a. One minute between events.
args = tuple()
kwargs = {
'count' : trade_count,
'sids' : sids,
'start' : trading_environment.first_open,
'delta' : trade_time_increment,
'filter' : sids
}
source = SpecificEquityTrades(*args, **kwargs)
for sid in sids:
start_date = trading_environment.first_open
# TODO: do we need to set the trading environment's end to same dt as
# the last trade in the history?
#trading_environment.period_end = trade_history[-1].dt
generated_trades = create_trade_history(
sid,
price,
volume,
trade_time_increment,
trading_environment
)
trade_history.extend(generated_trades)
trade_history = sorted(trade_history, key=attrgetter('dt'))
#set the trading environment's end to same dt as the last trade in the
#history.
trading_environment.period_end = trade_history[-1].dt
source = SpecificEquityTrades(trade_history)
return source
zipline/utils/test_utils.py
+13 -17
View File
@@ -65,10 +65,10 @@ def drain_zipline(test, zipline):
assert test.ctx, "method expects a valid zmq context"
assert test.zipline_test_config, "method expects a valid test config"
assert isinstance(test.zipline_test_config, dict)
assert test.zipline_test_config['results_socket'], \
assert test.zipline_test_config['results_socket_uri'], \
"need to specify a socket address for logs/perf/risk"
test.receiver = create_receiver(
test.zipline_test_config['results_socket'],
test.zipline_test_config['results_socket_uri'],
test.ctx
)
# Bind and connect are asynch, so allow time for bind before
@@ -81,8 +81,7 @@ def drain_zipline(test, zipline):
# some processes will exit after the message stream is
# finished. We block here to avoid collisions with subsequent
# ziplines.
for process in zipline.sim.subprocesses:
process.join()
zipline.join()
return output, transaction_count
@@ -97,16 +96,15 @@ def drain_receiver(receiver):
transaction_count = 0
while True:
msg = receiver.recv()
if msg == str(zp.CONTROL_PROTOCOL.DONE):
update = zp.BT_UPDATE_UNFRAME(msg)
output.append(update)
if update['prefix'] == 'PERF':
transaction_count += \
len(update['payload']['daily_perf']['transactions'])
elif update['prefix'] == 'EXCEPTION':
break
elif update['prefix'] == 'DONE':
break
else:
update = zp.BT_UPDATE_UNFRAME(msg)
output.append(update)
if update['prefix'] == 'PERF':
transaction_count += \
len(update['payload']['daily_perf']['transactions'])
elif update['prefix'] == 'EXCEPTION':
break
receiver.close()
del receiver
@@ -117,9 +115,6 @@ def drain_receiver(receiver):
def assert_single_position(test, zipline):
output, transaction_count = drain_zipline(test, zipline)
test.assertTrue(zipline.sim.ready())
test.assertFalse(zipline.sim.exception)
test.assertEqual(
test.zipline_test_config['order_count'],
transaction_count
@@ -128,7 +123,8 @@ def assert_single_position(test, zipline):
# the final message is the risk report, the second to
# last is the final day's results. Positions is a list of
# dicts.
closing_positions = output[-2]['payload']['daily_perf']['positions']
perfs = [x for x in output if x['prefix'] == 'PERF']
closing_positions = perfs[-2]['payload']['daily_perf']['positions']
test.assertEqual(
len(closing_positions),