从自动收报机数据生成CandleSticks

Question

我有一个算法可以将股票行情数据转换为CandleSticks。我有一个多次调用这个函数的代码，我正在尝试优化这个函数，让它运行得更快。因此，我希望您阅读代码，并给我一些建议，如何让它更快

对于这个问题，您可以将市场代码视为两个列表。某一股票的价格表

stock_price = [ 5, 5.1, 5, 4.9 , ... ]

以及与每个价格相关联的时间戳的列表。

timestamps  = [ 1534339504.36133 , 1534339704.36133, 1534339804.36133, 1534340504.36133, ... ]

您会注意到采样率是可变的，有时可能是几秒，有时可能是几分钟。输入列表按递增的时间戳排序。

所以我给出了我想要计算的数字N。如果我要10支持续时间为5分钟的蜡烛，而我没有足够的时间戳，那么第一支蜡烛就是NAN。另一方面，如果我有来自过去几周的大量时间戳，则只考虑最后的样本来计算最后的10根蜡烛，其余的将被忽略。

还有另一个细节。我用一种细微不同的方式来计算蜡烛数。通常，它们被引用到UTC，我认为我的列表中的最后一个元素是我最后一支蜡烛的收盘价和时间

最后，我需要一个列表或numpy数组，其中包含candle的开盘价、最高价、最低价、收盘价以及N个candles的时间间隔T

因此，为了将这两个列表转换为candlecharts，我执行以下操作

# time_interval is the size of the candle: 1, 5, 10... minutes, hours, etc
# nb_candles is the number of candles that I want to extract ( for example the last 5 candles )
def convert_samples_to_candles( stock_price , times , time_interval , nb_candles=-1 ):
    #If no data return NaNs        
    if( len(stock_price) == 0 or len(times) == 0 ):
        NO_RESPONSE = [np.NaN]
        return NO_RESPONSE, NO_RESPONSE, NO_RESPONSE, NO_RESPONSE, NO_RESPONSE

    last_time = times[-1]
    last_val  = stock_price[-1]

    #if nb_candles is not specified compute all the candles
    if( nb_candles==-1 ):
        nb_candles = int((last_time - times[0])/time_interval) + 1

    candles_open  = [np.NaN]*nb_candles
    candles_close = [np.NaN]*nb_candles
    candles_high  = [np.NaN]*nb_candles 
    candles_low   = [np.NaN]*nb_candles 
    candles_time  = [np.NaN]*nb_candles

    k=1
    last_candle = -1

    #Initialize the last candles with the last value
    candles_open[-1]  = last_val
    candles_close[-1] = last_val
    candles_high[-1]  = last_val
    candles_low[-1]   = last_val
    candles_time[-1]  = last_time

    #Iterate and fill each candle from the last one to the first one
    nb_times = len(times)
    while( k < nb_times and times[-1*k] + nb_candles*time_interval >  last_time ):

        a_last       = stock_price[-1*k]
        a_timestamp  = times[-1*k]
        candle_index = (-1*int((last_time - a_timestamp)/time_interval) -1)

        if( candle_index > -1 ):
            k += 1
            continue

        if( candle_index < last_candle ):
            candles_time[ candle_index ]  = a_timestamp
            candles_close[ candle_index ] = a_last
            candles_high[ candle_index ]  = a_last
            candles_low[ candle_index ]   = a_last
            candles_open[ candle_index ]  = a_last

            last_candle = candle_index

        else:
            #print candle_index, candles_open 
            candles_open[ candle_index ]  = a_last

            if( candles_high[ candle_index ]  < a_last ):
                candles_high[ candle_index ]  = a_last

            if( candles_low[ candle_index ]   > a_last ):
                candles_low[ candle_index ]   = a_last

        k += 1


    return candles_open, candles_close, candles_high, candles_low, candles_time

非常感谢您的宝贵时间！

Answer 1

因此，在一些研究之后，我试图给出一种不同的方法来计算蜡烛。

我定义了一个Candle_Handler类，并迭代地插入样本，然后更新candles。

当您迭代地重新计算candles时，此代码比问题中的代码略快。

class Candle_Handler(  ):

    def __init__(self, time_interval, nb_candles=5 ):

        self.nb_candles    = nb_candles
        self.time_interval = time_interval

        self.times  = []
        self.values = []

        self.candles_t = [ [] for _ in range(nb_candles) ]
        self.candles_v = [ [] for _ in range(nb_candles) ]       


    def insert_sample( self, value, time ):
        self.candles_t[-1].append(time)
        self.candles_v[-1].append(value)

        for i in range( self.nb_candles ):

            candle_index = -1*(i+1)
            if( len(self.candles_t[candle_index]) == 0 ): continue

            candle_time_interval = (i+1)*self.time_interval

            if( i + 1 == self.nb_candles ):
                while( len(self.candles_t[candle_index])> 0 and  time - self.candles_t[candle_index][0] > candle_time_interval ):
                    del self.candles_t[candle_index][0]
                    del self.candles_v[candle_index][0]

            else:

                while( len(self.candles_t[candle_index])> 0 and  time - self.candles_t[candle_index][0] > candle_time_interval ):
                    ltime  = self.candles_t[candle_index].pop(0)
                    lvalue = self.candles_v[candle_index].pop(0)

                    self.candles_t[candle_index-1].append( ltime )
                    self.candles_v[candle_index-1].append( lvalue )


    def get_all_candles(self, delta=1.0 ):

        last_time = self.candles_t[-1][-1]

        candles_open  = [ c[0]   if len(c)>0 else np.NAN for c in self.candles_v ] 
        candles_close = [ c[-1]  if len(c)>0 else np.NAN for c in self.candles_v ] 
        candles_high  = [ max(c) if len(c)>0 else np.NAN for c in self.candles_v ]  
        candles_low   = [ min(c) if len(c)>0 else np.NAN for c in self.candles_v ]  
        #candles_time  = [ c[-1]  if len(c)>0 else np.NAN for c in self.candles_t ]        
        candles_time  = [ last_time - (self.nb_candles - (c+1) )*self.time_interval for c in range(self.nb_candles) ]         


        for i in range( 1, self.nb_candles ):
            if( np.isnan( candles_close[i-1] ) ): continue

            if( np.isnan( candles_open[i] ) ):
                candles_open[i]  = candles_close[i-1]
                candles_close[i] = candles_close[i-1]
                candles_high[i]  = candles_close[i-1]
                candles_low[i]   = candles_close[i-1]


        if( not delta == 1.0 ):
            candles_close[-1] = candles_close[-1]*delta
            if( candles_high[-1] < candles_close[-1] ):
                candles_high[-1] = candles_close[-1]
            if( candles_low[-1]  > candles_close[-1] ):
                candles_low[-1]  = candles_close[-1]

            if( len(self.candles_v[-1]) == 1 ):
                candles_open[-1] = candles_close[-1]

        return candles_open, candles_close, candles_high, candles_low, candles_time