switched from pub/sub to push/pull for merge -> client socket. also purged source_id from the constructors of all datasources.

tests/test_finance.py

@@ -210,6 +210,7 @@ class FinanceTestCase(TestCase):
             "The merge should be drained of all messages, found {n} remaining." \
             .format(n=zipline.sim.merge.pending_messages())
         )
+
         self.assertEqual(
             zipline.algorithm.count,
             zipline.algorithm.incr,

zipline/components/tradesimulation.py

@@ -49,7 +49,6 @@ class TradeSimulationClient(Component):
         self.perf.open(self.context)
 
     def do_work(self):
-        from rdb import set_trace; set_trace()
         # poll all the sockets
         socks = dict(self.poll.poll(self.heartbeat_timeout))
 

zipline/core/host.py

@@ -93,7 +93,7 @@ class ComponentHost(object):
             self.components[component.get_id] = component
 
             if isinstance(component, DataSource):
-                self.feed.add_source(component.source_id)
+                self.feed.add_source(component.get_id)
             if isinstance(component, BaseTransform):
                 self.merge.add_source(component.get_id)
 

zipline/finance/sources.py

@@ -25,9 +25,9 @@ class TradeDataSource(DataSource):
         self.source_id = source_id
         self.setup_source()
 
-    @property
-    def get_id(self):
-        return 'TradeDataSource'
+    #@property
+    #def get_id(self):
+    #    return 'TradeDataSource'
 
     def send(self, event):
         """
@@ -56,8 +56,7 @@ class RandomEquityTrades(TradeDataSource):
     Generates a random stream of trades for testing.
     """
 
-    def init(self, sid, source_id, count):
-        self.source_id      = source_id
+    def init(self, sid, count):
         self.count          = count
         self.incr           = 0
         self.sid            = sid
@@ -95,7 +94,7 @@ class SpecificEquityTrades(TradeDataSource):
     Generates a random stream of trades for testing.
     """
 
-    def init(self, source_id, event_list):
+    def init(self, event_list):
         """
         :param event_list: should be a chronologically ordered list of
         dictionaries in the following form::
@@ -107,7 +106,6 @@ class SpecificEquityTrades(TradeDataSource):
                 'volume' : integer for volume
             }
         """
-        self.source_id = source_id
         self.event_list = event_list
         self.count = 0
 

zipline/lines.py

@@ -279,7 +279,7 @@ class SimulatedTrading(object):
         self.sim.register_components([source])
 
         # ``id`` is name of source_id, ``get_id`` is the class name
-        self.sources[source.source_id] = source
+        self.sources[source.get_id] = source
 
     def add_transform(self, transform):
         assert isinstance(transform, BaseTransform)

zipline/optimize/factory.py

@@ -65,7 +65,7 @@ def create_updown_trade_source(sid, trade_count, trading_environment, base_price
 
     trading_environment.period_end = cur
 
-    source = SpecificEquityTrades("updown_" + str(sid), events)
+    source = SpecificEquityTrades(events)
 
     return source
 

zipline/utils/factory.py

@@ -171,7 +171,7 @@ def create_returns_from_list(returns, trading_calendar):
 
 def create_random_trade_source(sid, trade_count, trading_environment):
     # create the source
-    source = RandomEquityTrades(sid, "rand-"+str(sid), trade_count)
+    source = RandomEquityTrades(sid, trade_count)
 
     # make the period_end of trading_environment match
     cur = trading_environment.first_open
@@ -244,5 +244,5 @@ def create_trade_source(sids, trade_count, trade_time_increment, trading_environ
     #history.
     trading_environment.period_end = trade_history[-1].dt
 
-    source = SpecificEquityTrades("flat", trade_history)
+    source = SpecificEquityTrades(trade_history)
     return source