def pull()

in 2_Strategies/model/algo_live_feed.py [0:0]

• 29 lines of code
• 4 McCabe index (conditional complexity)

    def pull(self):
        #print("B:%s" % self.lines.datetime.array)
        if math.isnan(self.lines.datetime[0]):
            now = datetime.datetime.now()
            self.lines.datetime[0]=date2num(now)
        now=datetime.datetime.now()
        try:
            item={}
            res=self.lambda_client.invoke(
                FunctionName='algo_market_data',
                InvocationType='RequestResponse',
                Payload=json.dumps(item)
            )
            t=res['Payload']
            l=json.loads(t.read().decode('utf-8'))
            print("load:%s" % l)
            
            #print(self.lines.datetime.array)
            #print(self.lines.close.array)
            
            for x in l:
                dt=pd.to_datetime(x['date'], format = "%Y-%m-%d")
                #print(dt)
                close=x['close']
                
                self.lines.datetime[0] = date2num(datetime.datetime.now())
                self.lines.open[0] = close
                self.lines.high[0] = close
                self.lines.low[0] = close
                self.lines.close[0] = close
                self.lines.volume[0] = 0
                
                self.connected=True
                self._laststatus=self.LIVE
                #print("connected")
        except Exception as e:
            print("err:%s" % e)
            time.sleep(5)