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

working on logging

Browse Source
This commit is contained in:
fawce
2012-08-08 14:00:36 -04:00
parent 82f0c03720
commit 427cfd53ec
4 changed files with 72 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/gens/tradesimulation.py
+51 -43
View File
@@ -1,4 +1,4 @@
import logbook
from logbook import Logger, Processor
from datetime import datetime, timedelta
from numbers import Integral
@@ -9,6 +9,9 @@ from zipline.gens.transform import StatefulTransform
from zipline.finance.trading import TransactionSimulator
from zipline.finance.performance import PerformanceTracker
from zipline.utils.log_utils import stdout_only_pipe
from zipline.gens.utils import hash_args
log = Logger('Trade Simulation')
class TradeSimulationClient(object):
"""
@@ -43,14 +46,15 @@ class TradeSimulationClient(object):
overwritten so that only the most recent snapshot of the universe
is sent to the algo.
"""
def __init__(self, algo, environment, sim_style):
self.algo = algo
self.sids = algo.get_sid_filter()
self.environment = environment
self.style = sim_style
self.algo_sim = None
def get_hash(self):
"""
There should only ever be one TSC in the system.
@@ -83,36 +87,36 @@ class TradeSimulationClient(object):
self.sids
)
with_portfolio = perf_tracker.transform(with_filled_orders)
# Pass the messages from perf along with the trading client's
# state into the algorithm for simulation. We provide the
# trading client so that the algorithm can place new orders
# into the client's order book.
algo_results = AlgorithmSimulator(
self.algo_sim = AlgorithmSimulator(
with_portfolio,
ordering_client.state,
self.algo,
self.algo,
)
# The algorithm will yield a daily_results message (as
# calculated by the performance tracker) at the end of each
# day. It will also yield a risk report at the end of the
# simulation.
for message in algo_results:
for message in self.algo_sim:
yield message
class AlgorithmSimulator(object):
def __init__(self, stream_in, order_book, algo):
def __init__(self, stream_in, order_book, algo):
self.stream_in = stream_in
# ==========
# Algo Setup
# ==========
# We extract the order book from the txn client so that
# We extract the order book from the txn client so that
# the algo can place new orders.
self.order_book = order_book
@@ -122,10 +126,8 @@ class AlgorithmSimulator(object):
# Monkey patch the user algorithm to place orders in the
# TransactionSimulator's order book.
self.algo.set_order(self.order)
self.algo.set_logger(logbook.Logger("AlgoLog"))
self.algo.set_logger(Logger("AlgoLog"))
# Call the user's initialize method.
self.algo.initialize()
# ==============
# Snapshot Setup
@@ -133,11 +135,11 @@ class AlgorithmSimulator(object):
# The algorithm's universe as of our most recent event.
self.universe = ndict()
for sid in self.sids:
self.universe[sid] = ndict()
self.universe.portfolio = None
# We don't have a datetime for the current snapshot until we
# receive a message.
self.simulation_dt = None
@@ -146,22 +148,22 @@ class AlgorithmSimulator(object):
# =============
# Logging Setup
# =============
# Processor function for injecting the algo_dt into
# user prints/logs.
def inject_algo_dt(record):
record.extra['algo_dt'] = self.this_snapshot_dt
self.processor = logbook.Processor(inject_algo_dt)
self.processor = Processor(inject_algo_dt)
# This is a class, which is instantiated later
# in run_algorithm. The class provides a generator.
self.stdout_capture = stdout_only_pipe
self.__generator = None
def __iter__(self):
return self
def next(self):
if self.__generator:
return self.__generator.next()
@@ -181,17 +183,17 @@ class AlgorithmSimulator(object):
'amount' : int(amount),
'filled' : 0
})
# Tell the user if they try to buy 0 shares of something.
if order.amount == 0:
zero_message = "Requested to trade zero shares of {sid}".format(
sid=event.sid
sid=order.sid
)
log.debug(zero_message)
# Don't bother placing orders for 0 shares.
return
return
# Add non-zero orders to the order book.
# Add non-zero orders to the order book.
# !!!IMPORTANT SIDE-EFFECT!!!
# This modifies the internal state of the transaction
# simulator so that it can fill the placed order when it
@@ -207,7 +209,9 @@ class AlgorithmSimulator(object):
# snapshot time to any log record generated.
with self.processor.threadbound(), self.stdout_capture(Logger('Print'),''):
# Call the user's initialize method.
self.algo.initialize()
for event in self.stream_in:
# Yield any perf messages received to be relayed back to
# the browser.
@@ -215,6 +219,12 @@ class AlgorithmSimulator(object):
yield event.perf_message
del event['perf_message']
if event.dt == "DONE":
if self.this_snapshot_dt:
# stop iteration happened
# mid-snapshot, so we have a universe
# snapshot that is not yet processed
# by the algorithm.
self.simulate_current_snapshot()
break
# This should only happen for the first event we run.
@@ -232,67 +242,65 @@ class AlgorithmSimulator(object):
# too long processing. Update the universe with data from
# this event, then check if enough time has passed that we
# can start a new snapshot.
else:
else:
self.update_universe(event)
if event.dt >= self.simulation_dt:
self.this_snapshot_dt = event.dt
def update_current_snapshot(self, event):
"""
Update our current snapshot of the universe. Call handle_data if
Update our current snapshot of the universe. Call handle_data if
"""
# The new event matches our snapshot dt. Just update the
# universe and move on.
if event.dt == self.this_snapshot_dt:
self.update_universe(event)
# The new event does not match our snapshot.
# The new event does not match our snapshot.
else:
self.simulate_current_snapshot()
# Once we've finished simulating the old snapshot,
# we can update the universe with the new event.
self.update_universe(event)
# The current event is later than the simulation time,
# which means the algorithm finished quickly enough to
# receive the new event. Start a new snapshot with this
# event's dt.
if event.dt >= self.simulation_dt:
self.this_snapshot_dt = event.dt
# The algorithm spent enough time processing that it
# missed the new event. Wait to start a new snapshot until
# the events catch up to the algo's simulated dt.
else:
self.this_snapshot_dt = None
def simulate_current_snapshot(self):
"""
Run the user's algo against our current snapshot and update the algo's
simulated time.
"""
"""
start_tic = datetime.now()
self.algo.handle_data(self.universe)
stop_tic = datetime.now()
# How long did you take?
delta = stop_tic - start_tic
# Update the simulation time.
self.simulation_dt = self.this_snapshot_dt + delta
def update_universe(self, event):
"""
Update the universe with new event information.
"""
# Update our portfolio.
self.universe.portfolio = event.portfolio
# Update our knowledge of this event's sid
for field in event.keys():
self.universe[event.sid][field] = event[field]
zipline/lines.py
+11 -8
View File
@@ -103,7 +103,7 @@ class SimulatedTrading(object):
self.date_sorted = date_sorted_sources(*sources)
self.transforms = transforms
# Formerly merged_transforms.
self.with_tnfms = sequential_transforms(self.date_sorted, *self.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)
self.results_uri = results_socket_uri
@@ -137,8 +137,7 @@ class SimulatedTrading(object):
setproctitle(self.sim_id)
self.open()
if self.zmq_out:
log_pipeline = NestedSetup([self.zmq_out,data_injector])
with log_pipeline.threadbound(), self.stdout_capture(self.print_logger, ''):
with self.zmq_out.threadbound():
self.stream_results()
# if no log socket, just run the algo normally
else:
@@ -148,13 +147,13 @@ class SimulatedTrading(object):
assert self.results_socket, \
"Results socket must exist to stream results"
try:
for event in self.gen:
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)
@@ -207,7 +206,7 @@ class SimulatedTrading(object):
)
self.results_socket.send(msg)
except:
log.exception("Exception while reporting simulation exception.")
@@ -222,12 +221,16 @@ class SimulatedTrading(object):
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 = True
bubble=False
)
def join(self):
if self.proc:
self.proc.join()
zipline/utils/log_utils.py
+8 -6
View File
@@ -120,12 +120,6 @@ class ZeroMQLogHandler(Handler):
#can't be serialized by JSON, so we need to convert to
#unix epoch representation.
if record.time:
assert isinstance(record.time, datetime.datetime)
time = record.time.replace(tzinfo = pytz.utc)
#logbook measures time in utc already, no need to convert.
record.time = EPOCH(time)
#Do the same if algo_dt is a datetime object.
if record.extra.has_key('algo_dt'):
@@ -151,6 +145,14 @@ class ZeroMQLogHandler(Handler):
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):
zipline/utils/test_utils.py
+2 -2
View File
@@ -61,7 +61,7 @@ def check(test, a, b, label=None):
test.assertEqual(a, b, "mismatch on path: " + label)
def drain_zipline(test, zipline):
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)
@@ -76,7 +76,7 @@ def drain_zipline(test, zipline):
time.sleep(1)
# start the simulation
zipline.simulate(blocking=False)
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