AutoParser.py

Go to the documentation of this file.

## @package serverportfolio.AutoParser
# Module for updating the dictionary of stocks using independent threads every TIMEPARSER sec,\n
#   and parsing data on demand from the server.
#
# Module loops over all stocks and record value in a 'CSV' style (in /data/historical/*INST.dcsv).
# To extend to different action, may use threads for other parsing actions (FORCE...)
#
# Implement a queueing system of stocks which are regularly 'parsed' by independent threads:
# - the threading loop every TIMEPARSER second.  
# -  put/remove stocks in the queue depending on states: is CLOSED, WAIT_OPEN, OPEN
#
# The parser threads waits for data in the queue
#   implements a simple lock if Server wait for an answer
#
# The queue is run in a separate thread ( no daemon ),
#     which creates 2 threads ( "daemon-ised" ) for parsing
# this way, main thread is still free
#
# Add list_error_stock, easy to deal, may check when Dictionary is reloaded for instance
#     may try to define temporary problem ( internet connection / slow ) and error in key with error value... 
#
# Rules for the states
# States for stock CLOSED(inactive)->WAIT_OPEN(active,but not write)->OPEN(active and write)->CLOSED(inactive)
#
# LastChanged $Id: AutoParser.py 9 2015-04-02 23:27:25Z michael $

import sys, logging, time
# for threading and queue
import Queue, threading

import serverportfolio.GlobalDicts as GlobalDicts
from   serverportfolio.DictionaryStocks import DictionaryStocks
from   serverportfolio.UpdateStocks import UpdateStocks
# need module for parsing, will import Bourso or Yahoo parser. Included 
#from serverportfolio.Parsers.ParserStocks import ParserStocks

## @brief Wait X seconds between parsing again
TIMEPARSER=30
## @brief Number of threads waiting for data in the queue. It is fixed at the initialisation.
NB_THREADS_PARSER=3

## @class thread_parser
# @brief Independent thread parsing the web pages, consuming data from the queue.
class thread_parser(threading.Thread):

    # may implement a counter of instances, check when there is crash...
    # count=0
    
    ## @brief Constructor for one thread_parser.
    # @param queue with Stock waiting to be parsed
    # @param dict_stock ...
    # @param name set the name of the thread
    def __init__( self,queue, name ):

        # parent class initialization
        threading.Thread.__init__(self,name=name)
        
        self.logger = logging.getLogger('SP.AutoParser.ThreadParser')
        self.logger.debug("init thread parser")
        
        ## queue to read the jobs to do, what to put ? a Stock or the symbol ?
        self._queue_jobs = queue
        ## internal reference to the singleton DictionaryStocks
        # should be an access to UpdateStocks now, one class for each parser certainly better
        #self._dict_stocks = dict_stocks
        
        # nice to have an unique object with an unique parser
        self._update_stock_th = UpdateStocks( 'InstValue' ) # fixed_parser='InstValue' )
        # set writing mode. done by UpdateStocks as post-process
        self._update_stock_th.set_write_to_file(True)
        # could make a link to the dictionary in UpdateStocks, used for get_stocks, 
        
        ## use to stop the threads 
        # __prefix with name class, safer or stupid !
        # needed ?? no just wait something in the queue, to check ?
        # @todo check if necessary
        self._update = True
        
    # test
    def __del__(self):
        print "destructor of thread parser ", self.name
    
    ## @brief Consume a stock and an action to perform form the queue
    # Action implemented : InstValue\n
    # To extend to more\n
    def run(self):

        # here add condition on the queue, just execute the command
        # tests should be done previously
        self.logger.debug("thread _parser %s run() " % self.name)
        
        # update allows to finish the thread from the outside
        count=0
        while self._update:

            # try to get a job to do
            try:
                #print "before reading queue"
                tuple_item = self._queue_jobs.get()
                #print "get from queue namestock ",namestock, action
                self.logger.debug("get from queue namestock %s, action %s" % tuple_item)
                namestock, action = tuple_item
                #print "get from queue namestock, action ",namestock, action
                
            # except never seen executed
            except:
                self.logger.debug("Catch exception in thread_parser")
                raise
            
            # happens if WAIT_OPEN only, 
            # should not if error check before, never in list_active !!
#             if self.dict_stocks.GetError( namestock ) != 0:
#                 print "AutoParser found error BEFORE GetInstValues ", namestock
#                 # let info until problem solved
#                 self.logger.info("AutoParser found error %s BEFORE GetInstValues" % namestock)
#                 # make difference ??
#                 break
                #continue
            
    ######################################################################
            # Main point, getting instant. values of a stock
# now parser created at start, e_action fixed (or variable?), should re-use the parser, still need to wait it finishes         
            if action == "InstValue" :
                
                
                # get last values from the web
                # exception problem with connection or parsing, should not kill the thread !
                # new all update is done there (except ML)
                try :
                    self._update_stock_th.update_data( namestock )
                except :
                    print "AutoParser Got exception, just break "
                    self.logger.warning("AutoParser got exception in GetInstValues %s" % namestock )
                    break

                # here ok, active, need a last_modification/last_saved 
                # avoid some call to wrapper, read file. Could be done in Stock as well !
                # 
                # post-process nothing to do, except SavingToFile
                # obj_stock = DictStocks().get_stocks( namestock )
                # obj_stock.save_inst_value()
                
                # check about error given by GetInstValues, problem with connection or parsing, should not kill the thread
#                 if self.dict_stocks.GetError( namestock ) != 0:
#                     # let info until problem solved
#                     self.logger.warning( "AutoParser found error after GetInstValues %s" % namestock )
#                     print "AutoParser found error after GetInstValues ", namestock
#                     break

# post processing, could be done by Stock itself, inside GetInstValue, 
# or inside a Parser function ( parser belong to the thread, can be re-used ), now call run_parser( namestock ),
# or post-processing called by thread, updated_stock ( e_action ) : Dictionary calls write file, send TCP message...
# release lock...
                # test should not happen
                # if dict_data == None:
                #    self.logger.critical("dict_data is null ",namestock)
                
                
            # before test if new state is closed, 
                # check state if 'CLOSED', no need to deal with file opening and closing file for nothing
#                 if (dict_data['state']=='CLOSED'):
#                     
#                     # means stock was opened (or WAIT_OPEN) and has closed, change state in dict_stocks
#                     if self.dict_stocks.GetState( namestock ) == 'OPEN':
#                         self.dict_stocks.SetState( namestock, 'CLOSED' )
#                     
#                         # copy a last time the value to be sure to have final data of the day
#                         
#                         # could check if different..., function test_change ??
#                         #DictStocks[namestock]['data']=dict_data
#                         self.dict_stocks.AddVolatileData( namestock, dict_data )
#                         self.logger.info("%s OPEN->CLOSED", namestock)
#                     
#                     # means stock wait for opening, let in the state
#                     #else if DictStocks[namestock]['state'] == 'WAIT_OPEN' :
#                         #print "stock %s remains WAIT_OPEN" % (namestock)
#                     
#                 # means stock is 'OPEN'
#                 else:
#                 
#                     # test, should never happen !! assert
#                     #if dict_data['state'] != 'OPEN':
#                     if obj_stock.state != 'OPEN':
#                         #print "ERROR WITH STATE IN THREAD_PARSER"
#                         self.logger.critical("%s ERROR WITH STATE IN THREAD_PARSER", namestock)
#                         assert ( 1 == 2 ), "ERROR WITH STATE IN THREAD_PARSER"
#                         sys.exit(0)
# 
#                     # update WAIT_OPEN->OPEN, no change for the queue already in active_list
#                     #if self.dict_stocks.GetState( namestock ) == 'WAIT_OPEN':
#                     #    self.dict_stocks.SetState( namestock, 'OPEN' )
#                     if obj_stock.state == 'WAIT_OPEN':
#                     
#                         # write anyway in this case because of the ML (otherwise need to compare date and time)
#                         # done by the parser now
#                         # self.dict_stocks.SaveInstValue( namestock, dict_data )
#                     
#                     # should be in Stock all this, 
#                         # assure the next test will true, not recorded twice
#                         # last modif not necessary because 'time'
#                         self.dict_stocks.SetLastModification( namestock, dict_data['time'] )
#                         self.dict_stocks.AddVolatileData( namestock, dict_data )
#                     
#                         # set to log
#                         self.logger.info("WAIT_OPEN->OPEN %s", namestock )
#                     
#         # to put in Stock
#         # test for saving into file
#                     # test if different time, if WAIT_OPEN->OPEN, will be true ( last is different or null )
#                     if dict_data['time'] != self.dict_stocks.GetLastModification( namestock ):
#                         #print "save record stock %s" % namestock
#                         #print "last_modif :",DictStocks[namestock]['last_modification']
#                         #print "time :",dict_data['time']
#                         self.dict_stocks.SetLastModification ( namestock, dict_data['time'] )
#                         self.dict_stocks.AddVolatileData( namestock, dict_data )
# 
#                         # write into file if modified
#                         #self.dict_stocks.SaveInstValue( namestock, dict_data )
#                         self.dict_stocks.SaveInstValue2( namestock, dict_data )
#                     # uncomment for tests    
#                     #else :
#                     #    print "%s time has not evolved or first 'OPEN' after 'WAIT_OPEN', do not record " % (namestock)

           
                    
    ################################################################                     
            else :
                print "a different action to implement ", action
                assert ( 1 == 2 ), "a different action to implement"
                
## @class Thread_Handle_Queue
# @brief Thread to handle the queueing system and create the thread_parsers.
class Thread_Handle_Queue(threading.Thread):

    ## @brief Constructor.
    # Size of the FIFO queue set to 10.
    def __init__( self ): #, dict_stocks ) :
        # init baseclass first
        threading.Thread.__init__(self, name='queue_thread')

        self.logger = logging.getLogger("SP.AutoParser.Queue")
        self.logger.info('init thread queue')
        
        #print "init thread_handle_queue"
        # control for stopping the run loop
        self._running=True
        
        # init a FIFO queue of size max size 10
        self._queue_jobs=Queue.Queue( 10 )
        
    # not necessary, only to dispatch oactive/inactive in init
        # Copy of an internal list of stock symbols loaded in DictionaryStocks
        # maybe even not necessary, convenient, but may need to be updated !
        #self._list_stocks = self.dict_stocks.get_stock_keys()
        #self._list_stocks = DictionaryStocks().get_stock_keys()
        
        # initialise only for the check_open in init 
        update_stock = UpdateStocks( 'InstValue' )
        # will write the first value in check_open(), only if the stock is OPEN
        update_stock.set_write_to_file(True)
        
        # make local reference to the singleton object, maybe not too bad...used akk the time
        #self.dict_stocks = dict_stocks
       
        self.logger.info("list_stocks: %s" % update_stock.get_stock_keys()) #self._list_stocks)

        ## @brief Important list, a stock should only belongs to one of theses lists !
        # considered as active: 5 minutes before opening and until closed is detected
        ## list of temporarly closed stocks.
        self.list_inactive_stock=[]
        ## list of open ( or wait_open ) stocks.
        self.list_active_stock=[]
        
        # list of stock which cannot be parsed, to move to Stock, only exception can make them move to inactive
        # but must be done by threads 
        #self.list_error_stock=[]
        
        # initialize the list_active / inactive / error stocks
        # here do not use thread ( how to get back the answer ? by an other queue !!
        # here again error could be associated to a stock : error_parser, error_X
        #for symbol in self._list_stocks:
        for symbol in update_stock.get_stock_keys():
            # check if 'state' = OPEN or CLOSE
            # may set 'state' = ERROR if cannot parse 
            
            #if self.dict_stocks.get_stocks( symbol).check_open():
            if update_stock.check_open( symbol ):
                #print "check open running..."
                self.list_active_stock.append(symbol)
                
            else:
                self.list_inactive_stock.append(symbol)
                
            # to initialize the error state
            #else:
                #if self.dict_stocks.dictstocks[stock]['error'] == 0 :
                #    self.list_inactive_stock.append(stock)
                #else :
                #self.list_error_stock.append(stock)

        all_lines = update_stock.print_dict_stocks('InstValue')
        self.logger.debug("\n%s",all_lines)
        #time.sleep(5)
        
        ## @brief Contains the threads who will listen to the queue.
        # possibility to add an other queue to receive results of the threads...
        # lock problem ?? when update, should not be the same threads
        self._list_thread_parser=[]
        ## Create the parsers and append to _list_thread_parser.
        self.create_threads_parser(NB_THREADS_PARSER)
           
        #if GlobalDicts.DEBUG_MODE : 
        print "init handler_queue"
        print "list_active_stock"
        print self.list_active_stock
        print "list_inactive_stock"
        print self.list_inactive_stock
        #print "list_error_stock"
        #print self.list_error_stock
        self.logger.info("list active %s" % ([stock for stock in self.list_active_stock]) )
        self.logger.info("list inactive %s" % ([stock for stock in self.list_inactive_stock]) )

    # test          
    def __del__(self):
        print "\ndestructor handler_queue\n"

        #print "list_thread before deletion"
        #for t in self.list_thread_parser:
        #    print t
        ##stop first trhaed_parser
        #for t in self.list_thread_parser:
            ##t.__update=False
        #del self.list_thread_parser[0]
        #print "sleep(10)"
        #time.sleep(10)
        #del self.list_thread_parser[0]
        #time.sleep(10)
        #print "list_thread"
        #for t in self.list_thread_parser:
        #    print t

    ## @brief Create and start the threads, called at initialisation.
    # Threads set as deamon, stored self.list_thread_parser (should check about problem with deamon)
    # @param nb_threads_parser number of threads to create (NB_THREADS_PARSER)               
    def create_threads_parser(self, nb_threads_parser):
        
        for i in range(nb_threads_parser):
            name_thread = "t_parser-" + str(i)
            t = thread_parser( self._queue_jobs, name_thread )
            self._list_thread_parser.append(t)
            # main does not stop if deleted
            t.setDaemon(True)
            t.start()
            
    ## @brief Fill the queue every TIMEPARSER sec.
    # Run until self._running is set to False
    # Check and change the state CLOSED, WAIT_OPEN of the Stocks.
    def run(self):

        print ""
        print "thread_handle_queue run() will update every ", TIMEPARSER, " sec."
        print ""
        
        while self._running:

            # loop over inactive stocks, check if they will soon open
            for symbol in self.list_inactive_stock:
                
                # maybe should keep a link, or ask for the list ?
                obj_stock = DictionaryStocks().get_stocks( symbol )
                
                # if state error should not add directly, but should check before with CheckOpen, or GetInstValue
                # better create list_error_stock...
                #if DictStocks[stock]['error']

                # get time of opening
                time_to_open = obj_stock.time_to_open()
                self.logger.debug("time_to_open %s %d" % (symbol,time_to_open.seconds))
                
                # test will open in less than 5 minutes-> send to list_active_stock
                # if (time_to_open.days==0) and (time_to_open.seconds < 300):
                if (time_to_open.days==0) and (time_to_open.seconds < 3600): #for test 1 hour
                    
                    # move to active_list
                    self.list_inactive_stock.remove(symbol)
                    self.list_active_stock.append(symbol)
                    
                    # set as 'WAIT_OPEN' in the queue, will stay to WAIT_OPEN with update
                    obj_stock.state = 'WAIT_OPEN'
                    self.logger.info("CLOSED->WAIT_OPEN, %s will open in %d seconds" % (symbol,time_to_open.seconds))
                    
                    # DO NOT insert to the queue, next loop of list_active_stock will do it

            # loop over active stock, check if they need to be rerun
            for symbol in self.list_active_stock:
                
                obj_stock = DictionaryStocks().get_stocks( symbol )
                
                # still open, put again in the queue
                # need specific test for ML
                if obj_stock.state == 'OPEN':
                    #print "found OPEN symbol ", obj_stock.symbol
                    #print 'id ', id(obj_stock)

#            To move to Stock as well
                    # Particularity ML on Bourso, Closed until 15h, but stay open after until ???18h ??
                    # Test if state='OPEN' and date=today(), work has been done
                    if obj_stock.get_market() == 'ml_euro':
                        
                        # if true work has been done for today, to put as inactive
                        if obj_stock.test_ML_done():
                            #print "ML Market already done"
                            self.list_active_stock.remove(symbol)
                            self.list_inactive_stock.append(symbol)
                            # may forced to close as well, here better than parser
                            self.logger.info("ML market %s done, move to inactive", stock)
                            
                    # all the other wait to be closed, send into the queue for a next round
                    else:
                        self.logger.debug("put stock, action %s %s" % (symbol,"InstValue"))
                        self._queue_jobs.put ( (symbol,"InstValue") )
                        
                elif obj_stock.state == 'WAIT_OPEN':
                    self.logger.debug("WAIT_OPEN insert to the queue")
                    self._queue_jobs.put ( (symbol,"InstValue") )
                
               
                # Possible only if it was OPEN before (or WAIT_OPEN, not anymore) and seen closed at the last update
                # move into inactive stock
                elif obj_stock.state == 'CLOSED':
                    self.list_active_stock.remove(symbol)
                    self.list_inactive_stock.append(symbol)
                    self.logger.info("%s OPEN->CLOSED, moved to inactive_stock", symbol)
                
            # no special loop for list_error_stock
            # need function to retest on RELOAD for instance
            # or function to retest depending on code error value
            
            #print "list active stock "
            #print self.list_active_stock
            #print "list_inactive_stock"
            #print self.list_inactive_stock
            
            # use a elapsed time of X sec. to make the next loop
            time.sleep ( TIMEPARSER )
            
# called by Server option FORCE, but how to get the return ? certainly a test...
# Ok, found the point is to add a new stock in the queue, then updated normally, can call GET
# certainly for a method ADD_QUEUE

    ## @brief Function to force an update, insert directly the stock in the queue.
    # Called by the Server ?\n
    # Implement simple lock than the caller must check before returning ?
    # @param stock  name of the stock
    # @param action only InstValue implemented
    def InsertToQueue( self, stock, action ):
        #print "InsertToQueue to put in the queue ", stock, action
        action = "InstValue"
        
        self.logger.info("%s Force InsertToQueue", stock)
        
        # first must test it is correct
        obj_stock = DictionaryStocks().get_stocks( stock )
        
        # may return an Invalid Stock
        if obj_stock.is_valid():
            
            # insert to the queue, but did not insert in the list of active/inactive stock
            # self._queue_jobs.put((stock,action))
            
            # insert to the list, dangerous into the queue AND the list
            if (DictionaryStocks().get_stocks( stock ).state == 'OPEN') | (DictionaryStocks().get_stocks( stock ).state == 'WAIT_OPEN'):
                self.list_active_stock.append( stock )
            elif DictionaryStocks().get_stocks( stock ).state == 'CLOSED':
                self.list_inactive_stock.append( stock )
                
            print "Insert To Queue new list active ", self.list_active_stock
            print "Insert To Queue new list inactive ", self.list_inactive_stock
            return True
        else:
            self.logger.info("InvalidStock discarded")
            return False
        
        #Create a manual lock
        #self.dict_stocks.dictstocks[stock]['lock'] = True
        #if GlobalDicts.DEBUG_MODE:
        #    print 'lock stock ', stock, self.dict_stocks.dictstocks[stock]['lock']

# ############# Main  ################################
# to add, test when deleting queue, threads....

if __name__ == "__main__":
    
    # if commented, will appear in doxygen documentation, see usage of EXCLUDE_SYMBOL
    # http://stackoverflow.com/questions/4306133/how-to-exclude-variables-from-doxygen-output
    
    logging.basicConfig(level=logging.DEBUG)
    # logging.basicConfig(filename='myapp.log', level=logging.INFO) # filemode='w' to not append to file
    logger = logging.getLogger('SP')
    main_logger = logging.getLogger("SP.main")
    
    # Load the dictionary and the configuration file
    dict_stocks = DictionaryStocks( load_all_stocks=True )
    #dict_stocks = DictionaryStocks()
    # not a nice way to load a few stock, but it is not the point anyway
    #dict_stocks.get_stocks(['CAC40','FSLR','EURUS'])
    
    time.sleep(2)
    
    print "FIRST: threading.enumerate ",threading.enumerate()   
    # run a thread for the queue
    # to suppress ? to see level of interaction
    tqh=Thread_Handle_Queue() # dict_stocks )
    print "tqh.name ", tqh.name, " is Alive:", tqh.is_alive(), " is daemon:", tqh.daemon
    time.sleep(5)
    tqh.start()
    # Alive after start()
    print "tqh.name ", tqh.name, " is Alive:", tqh.is_alive(), " is daemon:", tqh.daemon

    time.sleep(20)
    # print "dict_stocks in main"
    # print dict_stocks

    print "SECOND: enumerate ",threading.enumerate()

    time.sleep(25)
    # test for reloading dictstocks with a new value
    # dict_stocks.ReadStocks('dictstocks.add1.txt')
    # print dict_stocks
    
    # test force update...

    tqh._running=False
    
    # should try to set flag and join the thread here
    del tqh

    time.sleep(3)
    print "LAST : enumerate ",threading.enumerate()