From 9b6c83e3e5f4a5a2045b619bb296e1f2a430a395 Mon Sep 17 00:00:00 2001 From: Stephen Diehl Date: Wed, 16 May 2012 16:40:25 -0400 Subject: [PATCH] Removed more amguity over source_id --- zipline/components/datasource.py | 24 ++++++++++--------- zipline/core/host.py | 28 +++++++++------------- zipline/finance/sources.py | 41 ++++++++++++++++---------------- zipline/lines.py | 2 +- 4 files changed, 46 insertions(+), 49 deletions(-) diff --git a/zipline/components/datasource.py b/zipline/components/datasource.py index d6238e9b..d9e7da96 100644 --- a/zipline/components/datasource.py +++ b/zipline/components/datasource.py @@ -12,9 +12,9 @@ LOGGER = logging.getLogger('ZiplineLogger') class DataSource(Component): """ - Abstract baseclass for data sources. Subclass and implement send_all - - usually this means looping through all records in a store, converting - to a dict, and calling send(map). + Abstract baseclass for data sources. Subclass and implement send_all + - usually this means looping through all records in a store, + converting to a dict, and calling send(map). Every datasource has a dict property to hold filters:: - key -- name of the filter, e.g. SID @@ -23,20 +23,19 @@ class DataSource(Component): Modify the datasource's filters via the set_filter(name, value) """ - def init(self, source_id): - self.id = source_id - self.filter = {} - self.cur_event = None - def set_filter(self, name, value): self.filter[name] = value + def setup_source(self): + self.filter = {} + self.cur_event = None + @property def get_id(self): """ - Returns this component id, this is fixed at a class - level. This should not and cannot be contingent on - arguments to the init function. Examples: + Returns this component id, this is fixed at a class level. This + should not and cannot be contingent on arguments to the init + function. Examples: - "TradeDataSource" - "RandomEquityTrades" @@ -70,3 +69,6 @@ class DataSource(Component): def frame(self, event): return zp.DATASOURCE_FRAME(event) + + def do_work(self): + raise NotImplementedError() diff --git a/zipline/core/host.py b/zipline/core/host.py index 250daf96..4a6f790a 100644 --- a/zipline/core/host.py +++ b/zipline/core/host.py @@ -12,22 +12,18 @@ LOGGER = logging.getLogger('ZiplineLogger') class ComponentHost(Component): """ - Components that can launch multiple sub-components, synchronize their - start, and then wait for all components to be finished. + Components that can launch multiple sub-components, synchronize + their start, and then wait for all components to be finished. """ - def __init__(self, addresses): - Component.__init__(self) + def init(self, addresses): + assert hasattr(self, 'zmq_flavor'), \ + """ You must specify a flavor of ZeroMQ for all ComponentHost + subclasses. """ + self.addresses = addresses self.running = False - self.init() - - def init(self): - assert hasattr(self, 'zmq_flavor'), """ - You must specify a flavor of ZeroMQ for all - ComponentHost subclasses. """ - # Component Registry, keyed by get_id # ---------------------- self.components = {} @@ -81,7 +77,7 @@ class ComponentHost(Component): self.sync_register[component.get_id] = datetime.datetime.utcnow() if isinstance(component, DataSource): - self.feed.add_source(component.get_id) + self.feed.add_source(component.source_id) if isinstance(component, BaseTransform): self.merge.add_source(component.get_id) @@ -104,7 +100,7 @@ class ComponentHost(Component): self.sockets.append(self.sync_socket) def open(self): - for component in self.components.values(): + for component in self.components.itervalues(): self.launch_component(component) self.launch_controller() @@ -126,9 +122,9 @@ class ComponentHost(Component): while self.is_running(): # wait for synchronization request at start, and DONE at end. # don't timeout. - socks = dict(self.sync_poller.poll()) + socks = dict(self.sync_poller.poll()) - if self.sync_socket in socks and socks[self.sync_socket] == self.zmq.POLLIN: + if socks.get(self.sync_socket) == self.zmq.POLLIN: msg = self.sync_socket.recv() try: @@ -160,5 +156,3 @@ class ComponentHost(Component): def teardown_component(self, component): raise NotImplementedError - - diff --git a/zipline/finance/sources.py b/zipline/finance/sources.py index 889e7f01..52a13c75 100644 --- a/zipline/finance/sources.py +++ b/zipline/finance/sources.py @@ -13,6 +13,10 @@ import zipline.protocol as zp class TradeDataSource(DataSource): + def init(self, source_id): + self.source_id = source_id + self.setup_source() + @property def get_id(self): return 'TradeDataSource' @@ -25,13 +29,14 @@ class TradeDataSource(DataSource): :rtype: None """ - event.source_id = self.get_id + event.source_id = self.source_id + if event.sid in self.filter['SID']: message = zp.DATASOURCE_FRAME(event) else: blank = ndict({ "type" : zp.DATASOURCE_TYPE.TRADE, - "source_id" : self.get_id + "source_id" : self.source_id }) message = zp.DATASOURCE_FRAME(blank) @@ -43,8 +48,8 @@ class RandomEquityTrades(TradeDataSource): Generates a random stream of trades for testing. """ - def __init__(self, sid, source_id, count): - DataSource.__init__(self, source_id) + def init(self, sid, source_id, count): + self.source_id = source_id self.count = count self.incr = 0 self.sid = sid @@ -52,14 +57,12 @@ class RandomEquityTrades(TradeDataSource): self.day = datetime.timedelta(days=1) self.price = random.uniform(5.0, 50.0) + self.setup_source() + @property def get_id(self): return 'RandomEquityTrades' - @property - def get_type(self): - zp.COMPONENT_TYPE.SOURCE - def do_work(self): if not self.incr < self.count: self.signal_done() @@ -84,27 +87,25 @@ class SpecificEquityTrades(TradeDataSource): Generates a random stream of trades for testing. """ - def __init__(self, source_id, event_list): + def init(self, source_id, event_list): """ :param event_list: should be a chronologically ordered list of - dictionaries in the following form: + dictionaries in the following form:: - event = { - 'sid' : an integer for security id, - 'dt' : datetime object, - 'price' : float for price, - 'volume' : integer for volume - } + event = { + 'sid' : an integer for security id, + 'dt' : datetime object, + 'price' : float for price, + 'volume' : integer for volume + } """ - DataSource.__init__(self, source_id) + self.source_id = source_id self.event_list = event_list self.count = 0 # TODO temporary hack self.control_out = Mock() - - def get_type(self): - zp.COMPONENT_TYPE.SOURCE + self.setup_source() @property def get_id(self): diff --git a/zipline/lines.py b/zipline/lines.py index b7c26b3c..5e0b2f8f 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -287,7 +287,7 @@ class SimulatedTrading(object): self.sim.register_components([source]) # ``id`` is name of source_id, ``get_id`` is the class name - self.sources[source.id] = source + self.sources[source.source_id] = source def add_transform(self, transform): assert isinstance(transform, BaseTransform)