收集来自比特币衍生品交易所的近期交易数据

Question

作为一些交易软件的组成部分，我编写了函数parse_ticks()，作为模拟比特币衍生品交易所BitMEX的Exchange类的一部分。

parse_ticks()的目的是将前几分钟存储的滴答数据(通过调用all_ticks = self.ws.get_ticks()获得的单个交易列表)转换为字典，从以前的分钟滴答列表中聚合OHLCV值。交易列表是一个dict的列表，每个dict包含一个时间戳，交易的大小，如果是买入或卖出，等等。

当分钟过去时，每个仪器都会调用一次parse_ticks()。在这个场景中，我正在观看两种乐器，所以parse_ticks()在每分钟开始时被打了两次电话。

我为parse_ticks()运行时间计时了2400分钟(2400次观察)，尽可能多的后台进程和程序被禁用，并获得了这些结果(以秒计)：

平均运行时间: 0.0318425

最低运行时间: 0.00458

最大运行时间: 0.07958

性病发展: 0.02276709988

这里有一个很大的范围，它的最小值和最大值有很大的不同，并且std的发展速度几乎和平均值一样大。

我怎么降低性病。parse_ticks()运行时的发展，使平均运行时间更接近0.00458

的最小观测值。

这种优化是否属于python的范围？

编辑:在评论中回答问题(谢谢大家的帮助)：

ws.get_ticks()没有出站API调用，如下所示：

def get_ticks(self):
        return self.data['trade']

其中，data['trade']是一个包含实时从websocket流中保存的蜱的列表。如果'trade'列表超过10000个元素，它将被削减30%，因此一旦达到极限，这个方面(要解析的数据量不断增加)是不变的。因此，当调用parse_ticks()时，数据已经可用。

滴答的数量是不一样的，所以有些差异可以用它来解释。但肯定不是0.075的极限范围吗？(min - max)

运行时间看起来是随机的，在所有的观察中都有长的和短的运行。

self.bars = {}
self.symbols = ["XBTUSD", "ETHUSD"]
self.ws = Bitmex_WS()

def parse_ticks(self):
    """Return a 1-min OHLCV dict, given a list of the previous
       minutes tick data."""
    
    all_ticks = self.ws.get_ticks()
    target_minute = datetime.datetime.utcnow().minute - 1
    ticks_target_minute = []
    tcount = 0

    # search from end of tick list to grab newest ticks first
    for i in reversed(all_ticks):
        try:
            ts = i['timestamp']
            if type(ts) is not datetime.datetime:
                ts = parser.parse(ts)
        except Exception:
            self.logger.debug(traceback.format_exc())
        
        # scrape prev minutes ticks
        if ts.minute == target_minute:
            ticks_target_minute.append(i)
            ticks_target_minute[tcount]['timestamp'] = ts
            tcount += 1

        # store the previous-to-target-minute bar's last
        # traded price to use as the open price for target bar
        if ts.minute == target_minute - 1:
            ticks_target_minute.append(i)
            ticks_target_minute[tcount]['timestamp'] = ts
            break
    ticks_target_minute.reverse()

    # reset bar dict ready for new bars
    self.bars = {i: [] for i in self.symbols}

    # build 1 min bars for each symbol
    for symbol in self.symbols:
        ticks = [i for i in ticks_target_minute if i['symbol'] == symbol]
        bar = self.build_OHLCV(ticks, symbol)
        self.bars[symbol].append(bar)
        # self.logger.debug(bar)

Answer 1

我们(听起来也像你一样)没有足够的信息来回答这个问题。但首先有一些好消息:我看不到您的方法显然在浪费时间的任何区域(比如使用in搜索不断增长的列表中的某个项目)。

在你更多地关注这一点之前，考虑到即使方差很大，对于两个仪器来说，你的最大时间小于十分之一秒。换句话说，你可以监视超过1500台仪器，然后你才能以你的频率(1分钟)的顺序接近最大处理时间。因此，问问自己，您是否正在执行不必要的过早优化。这段代码会在\mathcal{O}(1000)仪器上运行，持续时间超过两天吗？如果没有，你可以在这里停下来。如果是，或者作为一项学术练习，继续下去。

据我所见，这段代码的运行时间在很大程度上取决于两个因素：

1. self.ws.get_ticks()需要多长时间。这种方法是否连接到互联网以获取其数据？然后，差异可能实际上是建立连接和获取数据的差异。这可能受到您的internet连接的影响，但也受您试图连接到的服务器上的当前负载的影响。在这种情况下，你什么也做不了。
2. 返回多少个元素。获取数据的实际函数可能要花费更长的时间才能获得更多的元素，但处理也需要更长的时间，因为您需要遍历列表中的所有元素。

唯一知道的方法就是收集更多的数据。单独计时对self.ws.get_ticks()的调用，收集len(all_ticks)并按时间顺序绘制所有内容。也许这会帮助你发现一些有趣的东西。

以下是您可以发现的一些可能性：

• 您获取信息的服务器也有一些更新频率，所以每五分钟就会有数据。
• 它们有一个速率限制，使所有请求在失败之间运行(这比传输一组数据更快)。
• 实际上，每次调用都会返回所有数据，当您到达最后一分钟时，您只需要返回break。在这种情况下，每次连续调用都比较昂贵，因此这当然会增加标准偏差。尝试找到一种将开始日期传递给呼叫的方法。

下面是一个小班，它可以帮助你跟踪我刚刚想出的不同计时器：

from time import perf_counter
from statistics import mean, median, stdev
from collections import defaultdict

class Timer:
    durations = defaultdict(list)

    def __init__(self, name=""):
        self.name = name

    def __enter__(self):
        self.start = perf_counter()

    def __exit__(self, *args):
        Timer.durations[self.name].append(perf_counter() - self.start)

    @staticmethod
    def calc_stats(x):
        return {"mean": mean(x),
                "std": stdev(x),
                "median": median(x),
                "min": min(x),
                "max": max(x)}

    @staticmethod
    def stats():
        return {name: Timer.calc_stats(x) for name, x in Timer.durations.items()}

通过一些示例用法：

from time import sleep
import pandas as pd
import matplotlib.pyplot as plt

from timer import Timer

for n in range(10):
    with Timer("a"):
        sleep(0.5)
    with Timer("b"):
        sleep(0.1 * n)

print(pd.DataFrame(Timer.stats()))
#                a         b
# mean    0.500559  0.450504
# std     0.000021  0.303079
# median  0.500552  0.450502
# min     0.500534  0.000006
# max     0.500602  0.901003

plt.plot(Timer.durations["a"], label="a")
plt.plot(Timer.durations["b"], label="b")
plt.legend()
plt.xlabel("Iteration")
plt.ylabel("Time [s]")
plt.show()

​

其他一些评论：

• 您可以使用self.bars = {i: [] for i in self.symbols}代替self.bars = defaultdict(list)，就像我在Timer类中所做的那样。
• 尽量避免单字母变量。在一些情况下，它们是可以的，但是i (和n)通常是为整数保留的。使用x，或者更好地使用tick。
• 不要直接比较type，而是使用isinstance(datetime.datetime, s)。这也允许子类。
• 别躺在你的文件里。你说“返回一个1分钟的OHLCV，给出前几分钟的滴答数据列表”，但几乎没有一个是正确的。该方法不返回任何内容，也不接受任何参数作为输入！
• 如果你知道预期的异常，那就只抓到它。至少您没有一个普通的except，但是except KeyError和解析器中的一些特定错误会更好。这样你就不会错过一个意想不到的错误。您还应该问问自己，如果发生错误，您的代码会做什么。我认为它将只使用以前的迭代ts，这将重复数据。只希望你在第一次时间戳上没有问题！