wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-14 11:15:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

tests passing using the process simulator.

Browse Source
This commit is contained in:
fawce
2012-07-24 14:00:33 -04:00
parent 582e7abd13
commit d0e987a8e8
9 changed files with 175 additions and 244 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_finance.py
+50 -120
View File
@@ -2,6 +2,7 @@
Tests for the zipline.finance package
"""
import pytz
import zmq
from unittest2 import TestCase
from datetime import datetime, timedelta
@@ -33,10 +34,13 @@ class FinanceTestCase(TestCase):
def setUp(self):
self.zipline_test_config = {
'allocator' : allocator,
'sid' : 133,
'devel' : True
'allocator' : allocator,
'sid' : 133,
'devel' : True,
'results_socket' : allocator.lease(1)[0]
}
self.ctx = zmq.Context()
self.log_handler = LoggingHandler()
self.log_handler.push_application()
@@ -109,147 +113,64 @@ class FinanceTestCase(TestCase):
self.assertTrue(env.last_close.month == 12)
self.assertTrue(env.last_close.day == 31)
# The following two tests appear broken no that the order source is
# non blocking. HUNCH: The trades are streaming through before the orders
# are placed.
def drain_zipline(self):
self.receiver = self.ctx.socket(zmq.PULL)
self.receiver.bind(self.zipline_test_config['results_socket'])
#@timed(EXTENDED_TIMEOUT)
def test_orders(self):
output = []
transaction_count = 0
while True:
msg = self.receiver.recv()
if msg == str(zp.CONTROL_PROTOCOL.DONE):
break
else:
update = zp.BT_UPDATE_UNFRAME(msg)
output.append(update)
if update['prefix'] == 'PERF':
transaction_count += \
len(update['payload']['daily_perf']['transactions'])
# Simulation
# ----------
self.zipline_test_config['simulation_style'] = \
SIMULATION_STYLE.FIXED_SLIPPAGE
zipline = SimulatedTrading.create_test_zipline(
**self.zipline_test_config
)
zipline.simulate(blocking=True)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
# TODO: Make more assertions about the final state of the components.
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
# the trading client should receive one transaction for every
# order placed.
self.assertEqual(
zipline.trading_client.txn_count,
zipline.trading_client.order_count
)
#@timed(DEFAULT_TIMEOUT)
def test_aggressive_buying(self):
# Simulation
# ----------
# TODO: for some reason the orders aren't filled without an extra
# trade.
trade_count = 5000
self.zipline_test_config['order_count'] = trade_count - 1
self.zipline_test_config['trade_count'] = trade_count
self.zipline_test_config['order_amount'] = 1
# tell the simulator to fill the orders in individual transactions
# matching the order volume exactly.
self.zipline_test_config['simulation_style'] = \
SIMULATION_STYLE.FIXED_SLIPPAGE
self.zipline_test_config['environment'] = factory.create_trading_environment()
sid_list = [self.zipline_test_config['sid']]
self.zipline_test_config['trade_source'] = factory.create_minutely_trade_source(
sid_list,
trade_count,
self.zipline_test_config['environment']
)
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
zipline.simulate(blocking=True)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
#
# the trading client should receive one transaction for every
# order placed.
self.assertEqual(
zipline.trading_client.txn_count,
zipline.trading_client.order_count
)
del self.receiver
return output, transaction_count
@timed(EXTENDED_TIMEOUT)
def test_performance(self):
def test_full_zipline(self):
#provide enough trades to ensure all orders are filled.
self.zipline_test_config['order_count'] = 100
self.zipline_test_config['trade_count'] = 200
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
zipline.simulate(blocking=True)
zipline.simulate(blocking=False)
output, transaction_count = self.drain_zipline()
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
self.assertEqual(
zipline.sim.feed.pending_messages(),
0,
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages())
self.zipline_test_config['order_count'],
transaction_count
)
self.assertEqual(
zipline.sim.merge.pending_messages(),
0,
"The merge should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.merge.pending_messages())
)
# 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']
self.assertEqual(
zipline.algorithm.count,
zipline.algorithm.incr,
"The test algorithm should send as many orders as specified.")
transaction_sim = zipline.trading_client.txn_sim
self.assertEqual(
transaction_sim.txn_count,
zipline.trading_client.perf.txn_count,
"The perf tracker should handle the same number of transactions \
as the simulator emits."
)
self.assertEqual(
len(zipline.get_positions()),
len(closing_positions),
1,
"Portfolio should have one position."
)
sid = self.zipline_test_config['sid']
self.assertEqual(
zipline.get_positions()[sid]['sid'],
closing_positions[0]['sid'],
sid,
"Portfolio should have one position in " + str(sid)
)
self.assertEqual(
zipline.sources['SpecificEquityTrades'].count,
self.zipline_test_config['trade_count'],
"The simulated trade source should send all trades."
)
self.assertEqual(
zipline.algorithm.frame_count,
self.zipline_test_config['trade_count'],
"The algorithm should receive all trades."
)
#@timed(DEFAULT_TIMEOUT)
def test_sid_filter(self):
"""Ensure the algorithm's filter prevents events from arriving."""
@@ -271,11 +192,17 @@ class FinanceTestCase(TestCase):
**self.zipline_test_config
)
zipline.simulate(blocking=True)
zipline.simulate(blocking=False)
output, transaction_count = self.drain_zipline()
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
#check that the algorithm received no events
self.assertEqual(
0,
test_algo.frame_count,
transaction_count,
"The algorithm should not receive any events due to filtering."
)
@@ -402,6 +329,9 @@ class FinanceTestCase(TestCase):
self.transaction_sim(**params1)
def transaction_sim(self, **params):
""" This is a utility method that asserts expected
results for conversion of orders to transactions given a
trade history"""
trade_count = params['trade_count']
trade_amount = params['trade_amount']
zipline/components/tradesimulation.py
+9 -4
View File
@@ -80,7 +80,7 @@ class TradeSimulationClient(Component):
self.algo_log = Logger("AlgoLog")
self.algorithm.set_logger(self.algo_log)
self.run_logged_op(self.algorithm.initialize)
self.do_op(self.algorithm.initialize)
def setup_logging(self, socket = None):
sock = socket or self.results_socket
@@ -183,11 +183,16 @@ class TradeSimulationClient(Component):
# data injection pipeline for log rerouting
# any fields injected here should be added to
# LOG_EXTRA_FIELDS in zipline/protocol.py
self.run_logged_op(self.algorithm.handle_data, data)
self.do_op(self.algorithm.handle_data, data)
def run_logged_op(self, callable_op, *args, **kwargs):
def exception_callback(self, trace):
log.info(trace)
pass
def do_op(self, callable_op, *args, **kwargs):
""" Wrap a callable operation with the zmq logbook
handler if it exits."""
handler if it exits. Also wrap the invocation of callable
op in a try/except, and relay """
if self.zmq_out:
def inject_event_data(record):
zipline/core/component.py
+17 -12
View File
@@ -208,8 +208,8 @@ class Component(object):
self.setup_poller()
self.open()
self.setup_control()
self.open()
self.signal_ready()
self.lock_ready()
@@ -326,7 +326,9 @@ class Component(object):
# In case we didn't receive a ping, send a pre-emptive
# pong to the monitor.
elif self.last_ping and time.time() - self.last_ping > 1:
elif hasattr(self, 'control_out') and \
self.last_ping and \
time.time() - self.last_ping > 1:
# send a ping ahead of schedule
pre_pong = time.time()
heartbeat_frame = CONTROL_FRAME(
@@ -452,7 +454,7 @@ class Component(object):
def signal_exception(self, exc=None, scope=None):
"""
This is *very* important error tracking handler.
This is a *very* important error tracking handler.
Will inform the system that the component has failed and how it
has failed.
@@ -474,6 +476,9 @@ class Component(object):
trace = ''.join(traceback.format_exception(exc_type, exc_value, exc_traceback))
sys.stdout.write(trace)
if hasattr(self, 'exception_callback'):
self.exception_callback(trace)
if hasattr(self, 'control_out'):
exception_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.EXCEPTION,
@@ -490,20 +495,20 @@ class Component(object):
self.state_flag = COMPONENT_STATE.DONE
if self.out_socket:
if hasattr(self, 'out_socket') and self.out_socket:
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
if hasattr(self, 'control_out'):
# notify controller we're done
done_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.DONE,
''
)
# notify controller we're done
done_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.DONE,
''
)
self.control_out.send(done_frame)
log.info("[%s] sent control done" % self.get_id)
self.control_out.send(done_frame)
log.info("[%s] sent control done" % self.get_id)
# there is a narrow race condition where we finish just
# after the Monitor accepts our prior heartbeat, but just
zipline/core/devsimulator.py
+1 -69
View File
@@ -5,7 +5,6 @@ See :py:method""
import logbook
import threading
from zipline.core.simulatorref import SimulatorBase
log = logbook.Logger('Dev Simulator')
@@ -34,71 +33,4 @@ class AddressAllocator(object):
return sockets
def reaquire(self, *conn):
pass
class Simulator(SimulatorBase):
zmq_flavor = 'thread'
def __init__(self, addresses):
# TODO: rethink this
SimulatorBase.__init__(self, addresses)
self.subthreads = []
self.running = False
log.warn(DEPRECATION_WARNING)
@property
def get_id(self):
return 'Simple Simulator'
def launch_controller(self):
thread = threading.Thread(
target=self.controller.run,
)
thread.start()
self.subthreads.append(thread)
return thread
def simulate(self):
thread = threading.Thread(target=self.run)
thread.start()
self.subthreads.append(thread)
self.running = True
return thread
def did_clean_shutdown(self):
return not any([t.isAlive() for t in self.subthreads])
def shutdown(self):
"""
Destroy all tracked components.
"""
if not self.running:
return
for component in self.components.itervalues():
component.shutdown()
for thread in self.subthreads:
if thread.is_alive():
thread._Thread__stop()
#self.controller.shutdown()
self.running = False
assert self.did_clean_shutdown()
def launch_component(self, component):
thread = threading.Thread(target=component.run)
thread.start()
self.subthreads.append(thread)
return thread
pass
zipline/core/monitor.py
+1 -1
View File
@@ -42,7 +42,7 @@ log = logbook.Logger('Controller')
# the system.
PARAMETERS = ndict(dict(
GENERATIONAL_PERIOD = 10, #seconds
GENERATIONAL_PERIOD = 30, #seconds
ALLOWED_SKIPPED_HEARTBEATS = 10,
ALLOWED_INVALID_HEARTBEATS = 3,
PRESTART_HEARBEATS = 3,
zipline/core/process.py
+89
View File
@@ -0,0 +1,89 @@
"""
The process simulator. Each component in threading.Thread
"""
import logbook
import multiprocessing
from zipline.core.host import ComponentHost
log = logbook.Logger('Process Simulator')
class ProcessSimulator(ComponentHost):
"""
The process simulator.
"""
zmq_flavor = 'mp'
def __init__(self, addresses):
ComponentHost.__init__(self, addresses)
self.subprocesses = []
self.running = False
self.mapping = {}
def define(self, key, val):
"""
Returns the mapping between a component and its
pid.
"""
self.mapping[key] = val
@property
def get_id(self):
return 'Multiprocess Simulator'
# =========
# Launchers
# =========
#
# invoked by the host's open()
def launch_controller(self):
proc = multiprocessing.Process(target=self.controller.run)
proc.start()
self.con = proc
# Process specific
self.controller_process = proc
self.mapping[proc.pid] = 'Controller'
def launch_component(self, component):
proc = multiprocessing.Process(target=component.run)
proc.start()
self.subprocesses.append(proc)
self.mapping[proc.pid] = component.get_id
return proc
def simulate(self):
"""
Kick off the simulation
"""
self.run()
def did_clean_shutdown(self):
cleanly = not any([s.is_alive() for s in self.subprocesses])
if not cleanly:
for process in self.subprocesses:
if process.is_alive():
log.error('Failed to Yield', self.mapping[process.pid])
return cleanly
def shutdown(self, ensure_clean=True):
"""
Shutdown the simulation.
"""
for component in self.components.itervalues():
component.shutdown()
for process in self.subprocesses:
process.join(timeout=1)
process.terminate()
self.controller.shutdown(soft=True)
self.running = False
self.con.terminate()
if ensure_clean:
assert self.did_clean_shutdown()
zipline/core/simulatorref.py
+1 -1
View File
@@ -17,7 +17,7 @@ Subclasses:
import abc
from zipline.core.host import ComponentHost
class SimulatorBase(ComponentHost):
class SimulatorBaseDeprecated(ComponentHost):
__metaclass__ = abc.ABCMeta
zipline/lines.py
+6 -36
View File
@@ -69,7 +69,7 @@ from zipline.transforms import BaseTransform
from zipline.test_algorithms import TestAlgorithm
from zipline.components import TradeSimulationClient
from zipline.core.devsimulator import Simulator
from zipline.core.process import ProcessSimulator
from zipline.core.monitor import Controller
from zipline.finance.trading import SIMULATION_STYLE
@@ -105,8 +105,6 @@ class SimulatedTrading(object):
:py:class:`zipline.trading.TradingEnvironment`
- allocator: an instance of
:py:class:`zipline.simulator.AddressAllocator`
- simulator_class: a :py:class:`zipline.core.host.ComponentHost`
subclass (not an instance)
- 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`
@@ -139,12 +137,9 @@ class SimulatedTrading(object):
devel = self.devel
)
# TODO: Not freeform
self.con.manage('freeform')
self.started = False
self.sim = config['simulator_class'](addresses)
self.sim = ProcessSimulator(addresses)
self.clients = {}
@@ -179,9 +174,6 @@ class SimulatedTrading(object):
- order_amount - the number of shares per order, defaults to 100
- trade_count - the number of trades to simulate, defaults to 101
to ensure all orders are processed.
- simulator_class - optional parameter that provides an alternative
subclass of ComponentHost to hold the whole zipline. Defaults to
:py:class:`zipline.simulator.Simulator`
- algorithm - optional parameter providing an algorithm. defaults
to :py:class:`zipline.test.algorithms.TestAlgorithm`
- trade_source - optional parameter to specify trades, if present.
@@ -221,11 +213,6 @@ class SimulatedTrading(object):
# trade than order
trade_count = 101
if config.has_key('simulator_class'):
simulator_class = config['simulator_class']
else:
simulator_class = Simulator
simulation_style = config.get('simulation_style')
if not simulation_style:
simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE
@@ -266,7 +253,6 @@ class SimulatedTrading(object):
'algorithm' : test_algo,
'trading_environment' : trading_environment,
'allocator' : allocator,
'simulator_class' : simulator_class,
'simulation_style' : simulation_style,
'results_socket' : results_socket,
'devel' : config.get('devel', False)
@@ -277,7 +263,7 @@ class SimulatedTrading(object):
# Save us from needless debugging
inside_test = 'nose' in inspect.stack()[-1][1]
if inside_test and not config.get('devel', False):
if False and inside_test and not config.get('devel', False):
assert False, """
You need to run the SimulatedTrading inside a test with devel=True
"""
@@ -366,7 +352,7 @@ class SimulatedTrading(object):
def setup_controller(self):
"""
Prepare the controller tro manage the topology specified
Prepare the controller to manage the topology specified
by this line.
"""
self.con.manage(self.topology)
@@ -377,14 +363,6 @@ class SimulatedTrading(object):
self.started = True
self.sim_context = self.sim.simulate()
# If we're in development mode then flag all the
# components in the topology as devel so as to indicate
# that they won't poll on the control channels for
# anything other than the synchronized start.
if self.devel:
for component in self.components:
component.devel = True
# 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
@@ -392,16 +370,8 @@ class SimulatedTrading(object):
# TODO: better way of identifying concurrency substrate
if blocking:
if self.sim.zmq_flavor == 'thread':
log.debug('Blocking')
for thread in self.sim.subthreads:
#log.debug('Waiting on %r' % thread)
log.debug('Waiting on %r' % thread)
thread.join()
log.debug('Yielded on %r' % thread)
else:
for process in self.sim.subprocesses:
process.join()
for process in self.sim.subprocesses:
process.join()
@property
def is_success(self):
zipline/test_algorithms.py
+1 -1
View File
@@ -154,7 +154,7 @@ class ExceptionAlgorithm(object):
"""
def __init__(self, throw_from):
self.throw_from == throw_from
self.throw_from = throw_from
def initialize(self):
if self.throw_from == "initialize":