diff --git a/zipline/gens/composites.py b/zipline/gens/composites.py index 234db714..964bbe42 100644 --- a/zipline/gens/composites.py +++ b/zipline/gens/composites.py @@ -11,23 +11,19 @@ from zipline.gens.transform import stateful_transform SourceBundle = namedtuple("SourceBundle", ['source', 'args', 'kwargs']) TransformBundle = namedtuple("TransformBundle", ['tnfm', 'args', 'kwargs']) -def date_sorted_sources(bundles): +def date_sorted_sources(*sources): """ Takes an iterable of SortBundles, generating namestrings and initialized datasources for each before piping them into a date_sort. """ - assert isinstance(bundles, (list, tuple)) - for bundle in bundles: - assert isinstance(bundle, SourceBundle) - # Calculate namestring hashes to pass to date_sort. - names = [bundle.source.__name__ + hash_args(*bundle.args, **bundle.kwargs) - for bundle in bundles] + for source in sources: + assert source.__dict__.has_key('__init__') + assert source.__dict__.has_key('get_hash') + + # Get name hashes to pass to date_sort. + names = [source.get_hash for source in sources) - # Pass each source its arguments. - source_gens = [bundle.source(*bundle.args, **bundle.kwargs) - for bundle in bundles] - # Convert the list of generators into a flat stream by pulling # one element at a time from each. stream_in = roundrobin(source_gens, names) diff --git a/zipline/gens/examples.py b/zipline/gens/examples.py index 50b955e4..afc8ebe8 100644 --- a/zipline/gens/examples.py +++ b/zipline/gens/examples.py @@ -17,11 +17,11 @@ import zipline.protocol as zp if __name__ == "__main__": - filter = [2] + filter = [2,3] #Set up source a. One minute between events. args_a = tuple() kwargs_a = { - 'sids' : [2], + 'sids' : [1,2,3], 'start' : datetime(2012,1,3,15, tzinfo = pytz.utc), 'delta' : timedelta(minutes = 1), 'filter' : filter @@ -31,7 +31,7 @@ if __name__ == "__main__": #Set up source b. Two minutes between events. args_b = tuple() kwargs_b = { - 'sids' : [2], + 'sids' : [2,3,4], 'start' : datetime(2012,1,3,14, tzinfo = pytz.utc), 'delta' : timedelta(minutes = 1), 'filter' : filter @@ -39,8 +39,7 @@ if __name__ == "__main__": source_b = SpecificEquityTrades(*args_a, **kwargs_a) #Set up source c. Three minutes between events. - - # sort_out = date_sorted_sources(source_a, source_b) + sort_out = date_sorted_sources(source_a, source_b) # passthrough = TransformBundle(Passthrough, (), {}) # mavg_price = TransformBundle(MovingAverage, (timedelta(minutes = 20), ['price']), {}) @@ -60,4 +59,5 @@ if __name__ == "__main__": # pp(message) - + +