Removed more amguity over source_id

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()

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
-
-

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):

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)