代码之家 › 专栏 › 技术社区 › Gabriel M RandayO

从ticker数据生成烛台

candlestick-chart quantitative-finance finance optimization python

1

Gabriel M RandayO · 技术社区 · 6 年前

我有一个算法,把股市行情数据转换成烛台。所以我想让你读一下代码,并给我一些建议,让它更快

股票价格表

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

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

您会注意到采样率是可变的,有时可能是几秒,有时可能是几分钟。输入列表通过增加时间戳进行排序。

所以我给了一些我想计算的蜡烛。每支蜡烛的持续时间为T。如果我要求10支蜡烛的持续时间为5分钟,我没有足够的时间戳,第一支蜡烛将是NAN。另一方面,如果我有大量的时间戳从过去几周,只有最后的样本将被考虑到计算最后10蜡烛,其余的将是忽略。

通常,它们是指UTC,我认为列表中的最后一个元素是最后一支蜡烛的收盘价和时间

所以为了把这两个列表转换成烛光图,我做了如下的工作

# 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

非常感谢你的时间!

1 回复 | 直到 6 年前

1

0

Gabriel M RandayO 6 年前

因此,经过一些研究,我试图给出一种不同的方法来计算蜡烛。

当您迭代地重新计算蜡烛时,此代码比问题中的代码稍微快一些。

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