diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index ef97bfd1..7effcf19 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -1,10 +1,7 @@ -from itertools import tee, chain +from itertools import chain from zipline.gens.utils import roundrobin, done_message from zipline.gens.sort import date_sort -from zipline.gens.merge import merge -from zipline.gens.transform import StatefulTransform - def date_sorted_sources(*sources): """ @@ -29,46 +26,6 @@ def date_sorted_sources(*sources): return date_sort(stream_in, names) - -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 - through a merge to output a unified stream. tnfms should be a - list of pointers to generator functions. tnfm_args should be a - list of tuples, representing the arguments to be passed to each - transform. tnfm_kwargs should be a list of dictionaries - representing keyword arguments to each transform. - """ - for transform in transforms: - assert isinstance(transform, StatefulTransform) - transform.merged = True - transform.sequential = False - - # Generate expected hashes for each transform - namestrings = [tnfm.get_hash() for tnfm in transforms] - - # Create a copy of the stream for each transform. - split = tee(sorted_stream, len(transforms)) - - # Package a stream copy with each StatefulTransform instance. - bundles = zip(transforms, split) - - # Convert the copies into transform streams. - tnfm_gens = [tnfm.transform(stream) for tnfm, stream in bundles] - - # 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) - - dt_aliased = alias_dt(merged) - # Return the merged events. - return add_done(dt_aliased) - - def sequential_transforms(stream_in, *transforms): """ Apply each transform in transforms sequentially to each event in stream_in. @@ -87,6 +44,7 @@ def sequential_transforms(stream_in, *transforms): transforms, stream_in) + dt_aliased = alias_dt(stream_out) return add_done(dt_aliased)