基于時間軸資料合成 bar的位置計算算法實作

前言： Pyhton 結束循環的方法是抛出

StopIteration

異常，涉及到異常棧操作，可見其性能非常的低。

當采用多重循環且循環基數巨大的算法時。 額，喝完一杯咖啡之後，還沒有任何輸出。

當你試圖運作一下測試Code時， 你會哭的:

def func7(loop0, loop1):
    for i in range(loop0):
        for j in range(loop1):
            a=
>>> %timeit func7(,)
           

在我的機器上運作同樣的 C 程式用時 19~20 秒之間。 好吧，我承認，這樣寫是很愚蠢的。

但是我們在設計程式時，未預期參數值大小的情況下就會寫成類似代碼。當同僚運作你的代碼時，肯定會皺眉的。

• • • 場景導入
    • 分析
    • 實作
    • 總結
    • 加好友

場景導入

閑話少說，來看看最近我遇到的一個場景。當我們使用股票相關軟體時，就會有分時圖：

1min, 5min, 15min 1day

, 友善我們對不同頻率的資料檢視與分析。

合成bar是基于時間軸的，當資料不足夠合成的一根完整 bar 時。我們采用已有的資料合成，直至合成一根完整的bar。保證每一個時間點都對應一個合成的bar。 若等資料完整才合成時一根bar，可能會導緻結尾的時間段為空。

假設已經擁有 1min 和 5min 的時間軸，現在要根據 1min 的資料來合成 5min 的bar， 我們要做的就是在時間軸計算處合成的位置。

分析

基于時間軸的資料合成。 請看以下資料：

1min	5min
9:01	9:05
9:02	9:10
9:03
9:04
9:05
9:06
9:07
9:08

合成9:02之間連續bar的資料包括[9:01,9:02]

合成9:05之間連續bar的資料包括[9:01, 9:02, 9:03, 9:04, 9:05]

合成9:08之間連續bar的資料包括[9:06, 9:07, 9:08]

合成9:10之間連續bar的資料包括[9:06, 9:07, 9:08]

已知資料是按照時間排序的，若合成每一根 bar 的起始位置确定了，就可以分别合成bar了。是以我們單獨抽出一個函數計算 begin_pos 和 end_pos。 有的人可能會問， 為什麼不一次性合成呢，因為還需在其它地方使用這兩個輸出。

尋找合成bar的位置就轉化為 高頻向低頻左插入位置查找， 了解這一點非常的重要。

實作

資料有序時尋找插入位置的算法有：

bisect.bisect_left(a, x, lo=0, hi=len(a))

Locate the insertion point for x in a to maintain sorted order. The parameters lo and hi may be used to specify a subset of the list which should be considered; by default the entire list is used. If x is already present in a, the insertion point will be before (to the left of) any existing entries. The return value is suitable for use as the first parameter to list.insert() assuming that a is already sorted.

The returned insertion point i partitions the array a into two halves so that all(val < x for val in a[lo:i]) for the left side and all(val >= x for val in a[i:hi]) for the right side.

bisect.bisect_left

numpy.searchsorted(a, v, side=’left’, sorter=None)

Find indices where elements should be inserted to maintain order.

Find the indices into a sorted array a such that, if the corresponding elements in v were inserted before the

indices, the order of a would be preserved.

numpy.searchsorted

ok 知道關鍵算法之後，看看實作過程

import numpy as np
from bisect import bisect_left

def calc_begin_end_pos(high_freq_tl, low_freq_tl): 
    """ 基于時間軸，計算時間合成 bar 的位置

    :param high_freq_tl: numpy.ndarray, 高頻時間軸 eg: 1min 
    :param low_freq_tl: numpy.ndarray, 低頻時間軸 eg: 2month

    :return: begin_pos, end_pos
    """
    # from bisect import bisect_left
    # c = [bisect_left(low_freq_tl, i) for i in high_freq_tl]
    c = np.searchsorted(low_freq_tl, high_freq_tl, side='left')
    a,b = np.unique(c, return_counts=True)
    d = dict(zip(a, np.add.accumulate(b)-b[]))

    begin_pos = np.array([d[i] for i in c])
    end_pos = np.arange(high_freq_tl.size)

    return begin_pos, end_pos
           

此代碼比較含蓄

d = dict(zip(a, np.add.accumulate(b)-b[]))
           

a 是 c 中 唯一的且按照順序排好的值，b 是 a 在 c 中的個數。

先對 b 做了一次累加求和，然後減去 b[0] 保證從 0 開始

這樣就構造了一個詞典，每個值在時間軸對應的開始計算的起始位置。

友善計算，一般會把時間點轉為整型或者浮點型。

假裝這是時間軸

h = np.arange(,)
l = np.arange(,, )

calc_begin_end_pos(h, l)
           

趕緊測試下性能吧，這個任務就交給讀者您了， 筆者是測過了的！

總結

之前采用一個比較 Low 的方法，主要思路是将 M:N 計算位置轉為 M:1:N的算法，1代表一個中間變量，沒計算一個位置要經過雙重循環，當高低頻跨度較小時，無謂的循環較少。跨度大時，每比較一個時間點跨度範圍的線性計算量。總之，應對場景開發與優化是最好不過了。

加好友

如果你和我有共同愛好，加好友一起學習吧！