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

end to end zipline with pure generators

Browse Source
This commit is contained in:
scottsanderson
2012-08-02 00:49:48 -04:00
parent 14067d8323
commit 56177a7c4f
3 changed files with 124 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/performance.py
+14 -14
View File
@@ -197,13 +197,12 @@ class PerformanceTracker(object):
# save the transactions for the daily periods
keep_transactions = True
)
for sid in sid_list:
self.cumulative_performance.positions[sid] = Position(sid)
self.todays_performance.positions[sid] = Position(sid)
def update(self, event):
import nose.tools; nose.tools.set_trace()
event.perf_message = self.process_event(event)
event.portfolio = self.get_portfolio()
del event['TRANSACTION']
@@ -247,6 +246,8 @@ class PerformanceTracker(object):
def process_event(self, event):
message = None
if self.exceeded_max_loss:
return
@@ -255,7 +256,7 @@ class PerformanceTracker(object):
self.event_count += 1
if(event.dt >= self.market_close):
self.handle_market_close()
message = self.handle_market_close()
if event.TRANSACTION:
self.txn_count += 1
@@ -270,8 +271,10 @@ class PerformanceTracker(object):
self.cumulative_performance.calculate_performance()
self.todays_performance.calculate_performance()
def handle_market_close(self):
return message
def handle_market_close(self):
# add the return results from today to the list of DailyReturn objects.
todays_date = self.market_close.replace(hour=0, minute=0, second=0)
todays_return_obj = risk.DailyReturn(
@@ -293,14 +296,9 @@ class PerformanceTracker(object):
# calculate progress of test
self.progress = self.day_count / self.total_days
# TODO!!!!
#TODO TODO TODO!!
daily_update = self.to_dict()
# Output results
if self.results_socket:
msg = zp.PERF_FRAME(self.to_dict())
self.results_socket.send(msg)
#
if self.trading_environment.max_drawdown:
returns = self.todays_performance.returns
max_dd = -1 * self.trading_environment.max_drawdown
@@ -311,7 +309,7 @@ class PerformanceTracker(object):
# 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
return daily_update
#move the market day markers forward
@@ -333,6 +331,8 @@ class PerformanceTracker(object):
self.market_close,
keep_transactions = True
)
return daily_update
def handle_simulation_end(self):
"""
@@ -369,8 +369,8 @@ class Position(object):
self.sid = sid
self.amount = 0
self.cost_basis = 0.0 ##per share
self.last_sale_price = None
self.last_sale_date = None
self.last_sale_price = 0.0
self.last_sale_date = 0.0
def update(self, txn):
if(self.sid != txn.sid):
zipline/gens/examples.py
+11 -8
View File
@@ -14,13 +14,13 @@ import zipline.protocol as zp
if __name__ == "__main__":
filter = [1,2,3,4]
filter = [2]
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'sids' : [1],
'sids' : [1,2,3,4],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 1),
'delta' : timedelta(hours = 1),
'filter' : filter
}
bundle_a = SourceBundle(SpecificEquityTrades, args_a, kwargs_a)
@@ -28,9 +28,9 @@ if __name__ == "__main__":
#Set up source b. Two minutes between events.
args_b = tuple()
kwargs_b = {
'sids' : [2],
'sids' : [1,2,3,4],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 1),
'delta' : timedelta(hours = 1),
'filter' : filter
}
bundle_b = SourceBundle(SpecificEquityTrades, args_b, kwargs_b)
@@ -38,9 +38,9 @@ if __name__ == "__main__":
#Set up source c. Three minutes between events.
args_c = tuple()
kwargs_c = {
'sids' : [3],
'sids' : [1,2,3,4],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 1),
'delta' : timedelta(hours = 1),
'filter' : filter
}
bundle_c = SourceBundle(SpecificEquityTrades, args_c, kwargs_c)
@@ -63,5 +63,8 @@ if __name__ == "__main__":
style = zp.SIMULATION_STYLE.PARTIAL_VOLUME
client_out = tsc(merge_out, algo, environment, style)
client_out.next()
import nose.tools; nose.tools.set_trace()
for message in client_out:
pass
zipline/gens/tradesimulation.py
+99 -57
View File
@@ -55,37 +55,6 @@ def trade_simulation_client(stream_in, algo, environment, sim_style):
for sid in sids:
open_orders[sid] = []
# Closure to pass into the user's algo to allow placing orders
# into the txn_sim's dict of open orders.
def order(self, sid, amount):
assert sid in sids, "Order on invalid sid: %i" % sid
order = ndict({
'dt' : self.current_dt,
'sid' : sid,
'amount' : int(amount),
'filled' : 0
})
# Tell the user if they try to buy 0 shares of something.
if order.amount == 0:
log = "requested to trade zero shares of {sid}".format(
sid=event.sid
)
log.debug(log)
return
open_orders[sid].append(event)
# Set the algo's order method.
algo.set_order(order)
# Provide a logbook logging interface to user code.
algo.set_logger(logbook.Logger("Algolog"))
# Call user-defined initialize method before we process any
# events.
algo.initialize()
# Pipe the in stream into the transaction simulator.
# Creates a txn field on the event containing transaction
@@ -113,44 +82,117 @@ def trade_simulation_client(stream_in, algo, environment, sim_style):
# Batch the event stream by dt to be processed by the user's algo.
# Yields perf messages whenever it encounters them.
perf_messages = algo_simulator(with_portfolio_and_perf_msg, algo)
perf_messages = algo_simulator(with_portfolio_and_perf_msg, sids, algo, open_orders)
for message in perf_messages:
yield message
def algo_simulator(stream_in, sids, algo, order_book):
simulation_dt = None
# Closure to pass into the user's algo to allow placing orders
# into the txn_sim's dict of open orders.
def order(sid, amount):
assert sid in sids, "Order on invalid sid: %i" % sid
order = ndict({
'dt' : simulation_dt,
'sid' : sid,
'amount' : int(amount),
'filled' : 0
})
def algo_simulator(stream_in, sids, algo):
# Tell the user if they try to buy 0 shares of something.
if order.amount == 0:
log = "requested to trade zero shares of {sid}".format(
sid=event.sid
)
log.debug(log)
return
order_book[sid].append(order)
# Set the algo's order method.
algo.set_order(order)
# Provide a logbook logging interface to user code.
algo.set_logger(logbook.Logger("Algolog"))
# Call user-defined initialize method before we process any
# events.
algo.initialize()
this_snapshot_dt = None
current_dt = None
universe = ndict()
for sid in sids:
universe[sid] = None
universe[sid] = ndict()
universe.portfolio = None
for update in stream_in:
#Yield perf messages to be relayed back to the browser.
if update.perf_message:
yield perf_message
if current_dt = None:
current_dt = update.dt
# If this message is newer than the algorithm's simulated dt,
# call handle data on a snapshot of the current algo universe,
# then
if message.dt >= current_dt + last_delta:
start_tic = datetime.now()
algo.handle_data(universe)
stop_tic = datetime.now()
last_delta = datetime
current_dt = message.dt + last_delta
batch.data[message.sid] = message
batch.data.portfolio = message.portfolio
for event in stream_in:
# Yield any perf messages received to be relayed back to the browser.
if event.perf_message:
yield event.perf_message
# This should only happen for the first event we run.
if simulation_dt == None:
simulation_dt = event.dt
# If we are currently creating a new message and this update
# matches the message dt, update the state of the universe.
if this_snapshot_dt != None:
if event.dt == this_snapshot_dt:
update_universe(event, universe)
# If we are constructing a snapshot and we hit a new dt, call
# handle_data and record how long it takes.
else:
start_tic = datetime.now()
algo.handle_data(universe)
stop_tic = datetime.now()
# How long did you take?
delta = stop_tic - start_tic
# Update the simulation time.
simulation_dt = this_snapshot_dt + delta
# Update the universe with the new event.
update_universe(event, universe)
# If the current event is later than the simulation
# time, update the universe and start constructing
# another snapshot.
if event.dt >= simulation_dt:
this_snapshot_dt = event.dt
else:
this_snapshot_dt = None
# We have been fastforwarding. Update the universe
# and check if we can start a new snapshot.
else:
update_universe(event, universe)
if event.dt >= simulation_dt:
this_snapshot_dt = event.dt
def update_universe(event, universe):
universe.portfolio = event.portfolio
del event['portfolio']
event_sid = event.sid
del event['sid']
for field in event.keys():
universe[event_sid][field] = event[field]