wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-06 05:14:38 +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
This commit is contained in:
fawce
2012-08-03 23:01:11 -04:00
parent f3c81d6731 8437a28c14
commit 4a3773848a
10 changed files with 270 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/core/component.py
-1
View File
@@ -114,7 +114,6 @@ class Component(object):
# Core Methods
# ------------
def loop_send(self):
"""
The main component loop. This is wrapped inside a
zipline/finance/performance.py
+14 -11
View File
@@ -203,10 +203,15 @@ class PerformanceTracker(object):
self.todays_performance.positions[sid] = Position(sid)
def update(self, event):
event.perf_message = self.process_event(event)
event.portfolio = self.get_portfolio()
del event['TRANSACTION']
return event
if event.dt == "DONE":
event.perf_message = self.handle_simulation_end()
del event['TRANSACTION']
return event
else:
event.perf_message = self.process_event(event)
event.portfolio = self.get_portfolio()
del event['TRANSACTION']
return event
def get_portfolio(self):
return self.cumulative_performance.as_portfolio()
@@ -270,6 +275,7 @@ class PerformanceTracker(object):
#calculate performance as of last trade
self.cumulative_performance.calculate_performance()
self.todays_performance.calculate_performance()
return message
@@ -296,7 +302,8 @@ class PerformanceTracker(object):
# calculate progress of test
self.progress = self.day_count / self.total_days
#TODO TODO TODO!!
# Take a snapshot of our current peformance to return to the
# browser.
daily_update = self.to_dict()
if self.trading_environment.max_drawdown:
@@ -356,12 +363,8 @@ class PerformanceTracker(object):
exceeded_max_loss = self.exceeded_max_loss
)
if self.results_socket:
log.info("about to stream the risk report...")
risk_dict = self.risk_report.to_dict()
msg = zp.RISK_FRAME(risk_dict)
self.results_socket.send(msg)
risk_dict = self.risk_report.to_dict()
return risk_dict
class Position(object):
zipline/finance/trading.py
+9 -2
View File
@@ -11,8 +11,8 @@ log = logbook.Logger('Transaction Simulator')
class TransactionSimulator(object):
UPDATER = True
def __init__(self, open_orders, style=SIMULATION_STYLE.PARTIAL_VOLUME):
self.open_orders = open_orders
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)
@@ -27,8 +27,15 @@ class TransactionSimulator(object):
elif style == SIMULATION_STYLE.NOOP:
self.apply_trade_to_open_orders = self.simulate_noop
for sid in sid_filter:
self.open_orders[sid] = []
def place_order(self, order):
self.open_orders[order.sid].append(order)
def update(self, event):
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)
return event
zipline/gens/composites.py
+22 -21
View File
@@ -3,7 +3,7 @@ from itertools import tee, starmap
from collections import namedtuple
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.utils import roundrobin, hash_args
from zipline.gens.utils import roundrobin, hash_args, done_message
from zipline.gens.sort import date_sort
from zipline.gens.merge import merge
from zipline.gens.transform import StatefulTransform
@@ -15,7 +15,7 @@ def date_sorted_sources(*sources):
"""
Takes an iterable of SortBundles, generating namestrings and initialized datasources
for each before piping them into a date_sort.
n """
"""
for source in sources:
assert iter(source), "Source %s not iterable" % source
@@ -34,8 +34,7 @@ n """
return date_sort(stream_in, names)
def merged_transforms(sorted_stream, bundles):
def merged_transforms(sorted_stream, *transforms):
"""
A generator that takes the expected output of a date_sort, pipes it
through a given set of transforms, and runs the results throught a
@@ -45,32 +44,34 @@ def merged_transforms(sorted_stream, bundles):
tnfm_kwargs should be a list of dictionaries representing keyword
arguments to each transform.
"""
for transform in transforms:
assert isinstance(transform, StatefulTransform)
# Generate expected hashes for each transform
namestrings = [bundle.tnfm.__name__ + hash_args(*bundle.args, **bundle.kwargs)
for bundle in bundles]
namestrings = [tnfm.get_hash() for tnfm in transforms]
# Create a copy of the stream for each transform.
split = tee(sorted_stream, len(bundles))
# Package a stream copy with each bundle
tnfms_with_streams = zip(split, bundles)
split = tee(sorted_stream, len(transforms))
# Package a stream copy with each StatefulTransform instance.
bundles = zip(transforms, split)
# Convert the copies into transform streams.
tnfms = [
StatefulTransform(
stream_copy,
bundle.tnfm,
*bundle.args,
**bundle.kwargs
)
for stream_copy, bundle in tnfms_with_streams
]
tnfm_gens = [tnfm.gen() for tnfm in tnfms]
tnfm_gens = [tnfm.transform(stream) for tnfm, stream in bundles]
# Roundrobin the outputs of our transforms to create a single flat stream.
# Roundrobin the outputs of our transforms to create a single flat
# stream.
to_merge = roundrobin(tnfm_gens, namestrings)
# Pipe the stream into merge.
merged = merge(to_merge, namestrings)
# Return the merged events.
return merged
def zipline(sources, transforms, endpoint):
assert isinstance(sources, (list, tuple))
assert isinstance(transforms, (list, tuple))
zipline/gens/examples.py
+13 -12
View File
@@ -11,7 +11,7 @@ from zipline.gens.composites import SourceBundle, TransformBundle, \
date_sorted_sources, merged_transforms
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.transform import MovingAverage, Passthrough, StatefulTransform
from zipline.gens.tradesimulation import trade_simulation_client as tsc
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
import zipline.protocol as zp
@@ -21,9 +21,10 @@ if __name__ == "__main__":
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'count' : 325,
'sids' : [1,2,3],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 10),
'delta' : timedelta(hours = 6),
'filter' : filter
}
source_a = SpecificEquityTrades(*args_a, **kwargs_a)
@@ -31,29 +32,29 @@ if __name__ == "__main__":
#Set up source b. Two minutes between events.
args_b = tuple()
kwargs_b = {
'count' : 7500,
'sids' : [2,3,4],
'start' : datetime(2012,1,3,14, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 10),
'delta' : timedelta(minutes = 5),
'filter' : filter
}
source_b = SpecificEquityTrades(*args_b, **kwargs_b)
#Set up source c. Three minutes between events.
sort_out = date_sorted_sources(source_a, source_b)
sorted = date_sorted_sources(source_a, source_b)
passthrough = TransformBundle(Passthrough, (), {})
mavg_price = TransformBundle(MovingAverage, (timedelta(minutes = 20), ['price']), {})
tnfm_bundles = (passthrough, mavg_price)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(MovingAverage, timedelta(minutes = 20), ['price'])
merge_out = merged_transforms(sort_out, tnfm_bundles)
merged = merged_transforms(sorted, passthrough, mavg_price)
algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3])
environment = create_trading_environment(year = 2012)
style = zp.SIMULATION_STYLE.FIXED_SLIPPAGE
client_out = tsc(merge_out, algo, environment, style)
for message in client_out:
pp(message)
sleep(1)
trading_client = tsc(algo, environment, style)
for message in trading_client.simulate(merged):
pp(message)
zipline/gens/merge.py
+2 -1
View File
@@ -6,7 +6,7 @@ from collections import deque
from zipline import ndict
from zipline.gens.utils import hash_args, \
assert_merge_protocol
assert_merge_protocol, done_message
from itertools import repeat
def merge(stream_in, tnfm_ids):
@@ -51,6 +51,7 @@ def merge(stream_in, tnfm_ids):
assert len(queue) == 1, "Bad queue in merge on exit: %s" % queue
assert queue[0].dt == "DONE", \
"Bad last message in merge on exit: %s" % queue
yield done_message('Merge')
def merge_one(sources):
dict_primer = zip(sources.keys(), repeat(None))
zipline/gens/tradegens.py
+1 -1
View File
@@ -84,7 +84,7 @@ class SpecificEquityTrades(object):
self.generator = self.create_fresh_generator()
def __iter__(self):
return self.generator
return self
def next(self):
return self.generator.next()
zipline/gens/tradesimulation.py
+197 -124
View File
@@ -9,7 +9,7 @@ from zipline.gens.transform import StatefulTransform
from zipline.finance.trading import TransactionSimulator
from zipline.finance.performance import PerformanceTracker
def trade_simulation_client(stream_in, algo, environment, sim_style):
class TradeSimulationClient(object):
"""
Generator that takes the expected output of a merge, a user
algorithm, a trading environment, and a simulator style as
@@ -42,61 +42,118 @@ def trade_simulation_client(stream_in, algo, environment, sim_style):
overwritten so that only the most recent snapshot of the universe
is sent to the algo.
"""
#============
# Algo Setup
#============
# Initialize txn_sim's dictionary of orders here so that we can
# reference it from within the user's algorithm.
sids = algo.get_sid_filter()
open_orders = {}
def __init__(self, algo, environment, sim_style):
for sid in sids:
open_orders[sid] = []
self.algo = algo
self.sids = algo.get_sid_filter()
self.environment = environment
self.style = sim_style
def get_hash(self):
"""
There should only ever be one TSC in the system.
"""
return self.__class__.__name__ + hash_args()
def simulate(self, stream_in):
"""
Main generator work loop.
"""
# Simulate filling any open orders made by the previous run of
# the user's algorithm. Sets the txn field to true on any
# event that results in a filled order.
ordering_client = StatefulTransform(
TransactionSimulator,
self.sids,
style = self.style
)
with_filled_orders = ordering_client.transform(stream_in)
# Pipe the events with transactions to perf. This will remove
# the txn field added by TransactionSimulator and replace it
# with a portfolio object to be passed to the user's
# algorithm. Also adds a PERF_MESSAGE field which is usually
# none, but contains an update message once per day.
perf_tracker = StatefulTransform(
PerformanceTracker,
self.environment,
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(
with_portfolio,
ordering_client.state,
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:
yield message
class AlgorithmSimulator(object):
# Pipe the in stream into the transaction simulator.
# Creates a txn field on the event containing transaction
# information if we filled any pending orders on the event's sid.
# TRANSACTION is None if we didn't fill any orders.
with_txns = StatefulTransform(
stream_in,
TransactionSimulator,
open_orders,
style = sim_style
)
# Pipe the events with transactions to perf. This will remove the
# txn field added by TransactionSimulator and replace it with
# a portfolio object to be passed to the user's algorithm. Also adds
# a PERF_MESSAGE field which is usually none, but contains an update
# message once per day.
with_portfolio_and_perf_msg = StatefulTransform(
with_txns,
PerformanceTracker,
environment,
sids
)
# 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, sids, algo, open_orders)
for message in perf_messages:
yield message
def algo_simulator(stream_in, sids, algo, order_book):
def __init__(self, stream_in, order_book, algo):
simulation_dt = None
self.stream_in = stream_in
# 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
# We extract the order book from the txn client so that
# the algo can place new orders.
self.order_book = order_book
self.algo = algo
self.sids = algo.get_sid_filter()
# 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"))
# Call the user's initialize method.
self.algo.initialize()
# 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
self.this_snapshot_dt = None
self.__generator = None
def __iter__(self):
return self
def next(self):
if self.__generator:
return self.__generator.next()
else:
self.__generator = self._gen()
return self.__generator.next()
def order(self, sid, amount):
"""
Closure to pass into the user's algo to allow placing orders
into the txn_sim's dict of open orders.
"""
assert sid in self.sids, "Order on invalid sid: %i" % sid
order = ndict({
'dt' : simulation_dt,
'dt' : self.simulation_dt,
'sid' : sid,
'amount' : int(amount),
'filled' : 0
@@ -104,91 +161,107 @@ def algo_simulator(stream_in, sids, algo, order_book):
# 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(
zero_message = "Requested to trade zero shares of {sid}".format(
sid=event.sid
)
log.debug(log)
log.debug(zero_message)
# Don't bother placing orders for 0 shares.
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"))
# 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
# receives its next message.
self.order_book.place_order(order)
# Call user-defined initialize method before we process any
# events.
algo.initialize()
universe = ndict()
for sid in sids:
universe[sid] = ndict()
universe.portfolio = None
this_snapshot_dt = None
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
del 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
def _gen(self):
"""
Internal generator work loop.
"""
for event in self.stream_in:
# Yield any perf messages received to be relayed back to the browser.
if event.perf_message:
yield event.perf_message
del event['perf_message']
if event.dt == "DONE":
break
# Update the universe with the new event.
update_universe(event, universe)
# This should only happen for the first event we run.
if self.simulation_dt == None:
self.simulation_dt = event.dt
# ======================
# Time Compression Logic
# ======================
if self.this_snapshot_dt != None:
self.update_current_snapshot(event)
# 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.
# The algorithm has been missing events because it took
# 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:
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
"""
# 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.
else:
update_universe(event, universe)
if event.dt >= simulation_dt:
this_snapshot_dt = event.dt
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(event, universe):
universe.portfolio = event.portfolio
del event['portfolio']
def update_universe(self, event):
"""
Update the universe with new event information.
"""
# Update our portfolio.
self.universe.portfolio = event.portfolio
event_sid = event.sid
del event['sid']
for field in event.keys():
universe[event_sid][field] = event[field]
# Update our knowledge of this event's sid
for field in event.keys():
self.universe[event.sid][field] = event[field]
zipline/gens/transform.py
+11 -11
View File
@@ -42,13 +42,13 @@ def functional_transform(stream_in, func, *args, **kwargs):
class StatefulTransform(object):
"""
Generic transform generator that takes each message from an
in-stream and passes it to a state class. For each call to
in-stream and passes it to a state object. For each call to
update, the state class must produce a message to be fed
downstream. Any transform class with the FORWARDER class variable
set to true will forward all fields in the original message.
Otherwise only dt, tnfm_id, and tnfm_value are forwarded.
"""
def __init__(self, stream_in, tnfm_class, *args, **kwargs):
def __init__(self, tnfm_class, *args, **kwargs):
assert isinstance(tnfm_class, (types.ObjectType, types.ClassType)), \
"Stateful transform requires a class."
assert tnfm_class.__dict__.has_key('update'), \
@@ -56,26 +56,26 @@ class StatefulTransform(object):
self.forward_all = tnfm_class.__dict__.get('FORWARDER', False)
self.update_in_place = tnfm_class.__dict__.get('UPDATER', False)
# You can't be both a forwarded and an updater.
assert not all([self.forward_all, self.update_in_place])
self.stream_in = stream_in
# Create an instance of our transform class.
self.state = tnfm_class(*args, **kwargs)
# Generate the string associated with this generator's output.
# Create the string associated with this generator's output.
self.namestring = tnfm_class.__name__ + hash_args(*args, **kwargs)
def get_hash(self):
return self.namestring
def __iter__(self):
return self.gen()
def gen(self):
def transform(self, stream_in):
return self._gen(stream_in)
def _gen(self, stream_in):
# IMPORTANT: Messages may contain pointers that are shared with
# other streams, so we only manipulate copies.
for message in self.stream_in:
for message in stream_in:
assert_sort_unframe_protocol(message)
message_copy = deepcopy(message)
zipline/gens/utils.py
+1 -1
View File
@@ -23,7 +23,7 @@ def mock_done(id):
"source_id" : id,
'tnfm_id' : id,
'tnfm_value': None,
'type' : 0
'type' : DATASOURCE_TYPE.DONE
})
done_message = mock_done