monday coding

src/rethink/portfolio_item.py

@@ -316,21 +316,24 @@ class Portfolio(AbstractTableModel):
         return self.port['g_table']['header'][col][0]
 
     def get_value_at(self, row, col):
-        ckey = self.port['g_table']['row_to_ckey_index'][row]
-        p_item = self.port['port_items'][ckey]
+#         ckey = self.port['g_table']['row_to_ckey_index'][row]
+#         p_item = self.port['port_items'][ckey]
         raise NotImplementedError
     
     def get_values_at(self, row):
         ckey = self.port['g_table']['row_to_ckey_index'][row]
         p_item = self.port['port_items'][ckey]
-        return self.port_item_to_row_fields(p_item)
+        return self.port_item_to_row_fields((None, p_item))
     
     def port_item_to_row_fields(self, x):
         
         def handle_NaN(n):
             # the function JSON.parse will fail at the javascript side if it encounters
             # a NaN value in the json string. Convert Nan to null to circumvent the issue 
-            return None if math.isnan(n) else n 
+            try:
+                return None if math.isnan(n) else n
+            except:
+                return None 
         
         rf = [{'v': '%s-%s-%s' % (x[1].get_symbol_id(), x[1].get_expiry(), x[1].get_strike())}, 
              {'v': x[1].get_right()},

src/rethink/portfolio_monitor.py

@@ -12,12 +12,13 @@ from finopt.instrument import Symbol, Option
 from rethink.option_chain import OptionsChain
 from rethink.tick_datastore import TickDataStore
 from rethink.portfolio_item import PortfolioItem, PortfolioRules, Portfolio
+from rethink.table_model import AbstractTableModel, AbstractPortfolioTableModelListener
 from comms.ibc.tws_client_lib import TWS_client_manager, AbstractGatewayListener
 
 
 
 
-class PortfolioMonitor(AbstractGatewayListener):
+class PortfolioMonitor(AbstractGatewayListener, AbstractPortfolioTableModelListener):
 
 
     def __init__(self, kwargs):
@@ -31,7 +32,7 @@ class PortfolioMonitor(AbstractGatewayListener):
         
         
         
-        self.ws_server =  None
+        
         '''
             portfolios: {<account>: <portfolio>}
         '''
@@ -87,10 +88,10 @@ class PortfolioMonitor(AbstractGatewayListener):
             logging.info('PortfolioMonitor: Service shut down complete...')               
     
         
-    def kproducer(self):
+    def get_kproducer(self):
         # returns a reference to the kafka base producer that we can 
         # use for sending messages
-        return self.twsc.gw_message_handler()
+        return self.twsc.gw_message_handler
     
                 
     def get_portfolio(self, account):
@@ -334,13 +335,24 @@ class PortfolioMonitor(AbstractGatewayListener):
         
     
     def notify_table_model_changes(self, account, port, contract_key, mode):
-            row = port.ckey_to_row(contract_key)
-            rvs = port.get_values_at(row)
-            port.fire_table_row_updated(row, rvs)
-            event_type = 'table_row_updated' if mode == 'U' else 'table_row_inserted'
-            self.kproducer().send_message(event_type, json.dumps({'source': 'dt_port-%s' % account, 'row': row, 'row_values': rvs}))
-    
+            
+        row = port.ckey_to_row(contract_key)
+        rvs = port.get_values_at(row)
+        #logging.info('---- %s' % str(rvs))
+        port.fire_table_row_updated(row, rvs)
+        event_type = AbstractTableModel.EVENT_TM_TABLE_ROW_UPDATED if mode == 'U' else AbstractTableModel.EVENT_TM_TABLE_ROW_INSERTED
+        self.get_kproducer().send_message(event_type, json.dumps({'source': 'dt_port-%s' % account, 'row': row, 'row_values': rvs}))
     
+    # implment AbstractPortfolioTableModelListener
+    # handle requests to get data table json
+    def event_tm_request_table_structure(self, event, request_id, account):
+        try:
+            self.get_kproducer().send_message(AbstractTableModel.EVENT_TM_TABLE_STRUCTURE_CHANGED, 
+                                          request_id, self.portfolios[account].get_JSON())
+        except:
+            logging.error("PortfolioMonitor:event_tm_request_table_structure. Error invoking get_JSON[%s]. Client request id:%s, %s" %
+                            account, request_id, ', '.join(e for e in sys.exc_info()))
+            
         
 if __name__ == '__main__':
     
@@ -360,7 +372,7 @@ if __name__ == '__main__':
       'session_timeout_ms': 10000,
       'clear_offsets':  False,
       'logconfig': {'level': logging.INFO, 'filemode': 'w', 'filename': '/tmp/pm.log'},
-      'topics': ['position', 'positionEnd', 'tickPrice', 'update_portfolio_account'],
+      'topics': ['position', 'positionEnd', 'tickPrice', 'update_portfolio_account', 'event_tm_request_table_structure'],
       'seek_to_end': ['*'],
       'ws_port': 9001,
       

src/rethink/table_model.py

@@ -1,4 +1,5 @@
 from misc2.observer import Subscriber, Publisher
+from comms.ibgw.base_messaging import BaseMessageListener
 from misc2.observer import NotImplementedException
 import logging
 
@@ -8,8 +9,12 @@ class AbstractTableModel(Publisher):
     EVENT_TM_TABLE_CELL_UPDATED = 'event_tm_table_cell_updated'
     EVENT_TM_TABLE_ROW_INSERTED = 'event_tm_table_row_inserted'
     EVENT_TM_TABLE_ROW_UPDATED = 'event_tm_table_row_updated'
+    EVENT_TM_TABLE_STRUCTURE_CHANGED = 'event_tm_table_structure_changed'
     TM_EVENTS = [EVENT_TM_TABLE_CELL_UPDATED, EVENT_TM_TABLE_ROW_INSERTED, EVENT_TM_TABLE_ROW_UPDATED]
     
+    EVENT_TM_REQUEST_TABLE_STRUCTURE = "event_tm_request_table_structure"
+    TM_REQUESTS = [EVENT_TM_REQUEST_TABLE_STRUCTURE]
+    
     def __init__(self):
         Publisher.__init__(self, AbstractTableModel.TM_EVENTS)
     
@@ -52,7 +57,30 @@ class AbstractTableModel(Publisher):
         raise NotImplementedException
 
 
+class AbstractPortfolioTableModelListener(BaseMessageListener):
+    '''
+    
+    '''
+    
+    def __init__(self, name, ws_server):
+        BaseMessageListener.__init__(self, name)
+        self.ws_server = ws_server
+
+    def event_tm_table_cell_updated(self, event, source, row, row_values):
+        logging.info("[%s] received %s content:[%s]" % (self.name, event, vars()))
+
+        
+    def event_tm_table_row_inserted(self, event, source, row, row_values):
+        logging.info("[%s] received %s content:[%s]" % (self.name, event, vars()))
+
+    def event_tm_table_row_updated(self, event, source, row, row_values):   
+        logging.info("[%s] received %s content:[%s]" % (self.name, event, vars()))
+    
+    def event_tm_table_structure_changed(self, event, origin_request_id, account, data_table_json):
+        logging.info("[%s] received %s content:[%s]" % (self.name, event, vars()))
         
+    def event_tm_request_table_structure(self, event, request_id, account):
+        logging.info("[%s] received %s content:[%s]" % (self.name, event, vars()))