如何修复运行时错误：在Python中迭代期间deque发生了变化

Question

我正在尝试为算法交易实现市场模拟，我在github https://github.com/DrAshBooth/PyLOB上找到了这段代码。 问题是当我运行我的小窗口代码，例如，2天，一切都很好，我得到了我预期的结果。但当我将窗口增加到20天或更长时间时，我会得到“RuntimeError:deque在迭代期间发生变化”。 我检查了我的代码，但我从来没有发现任何东西可以在我的运行中改变deque。 下面是产生错误的代码部分：

    self.tape = deque(maxlen=None)
    .
    .
    .
    def avg_volume_traded(self):
       if self.tape != None and len(self.tape) > 0:
          num = 0
          total_volume = 0
          for entry in self.tape:
             total_volume = entry['qty'] + total_volume
             num += 1
          if num != 0 and total_volume != None:
             return total_volume, num
       else:
          return None, None

这是实际的错误消息：

^{pr2}$

这是将线程分为两部分使用的主要部分（类周期和日周期）：

class Periodic(object):
    def __init__(self, object, compression_factor, args=[], kwargs={}):
        self.compression_factor = compression_factor
        self.object = object
        self.func = object.trade
        self.args = args
        self.kwargs = kwargs
        self.seppuku = Event()
    def start(self):
        self.seppuku.clear()
        self.proc = Thread(target=self._doit)
        self.proc.start()
    def stop(self):
        self.seppuku.set()
        self.proc.join()
    def _doit(self):
        while True:
            self.seppuku.wait(self.object.interval / self.compression_factor)
            if self.seppuku.is_set():
                break
            self.func(self.args[0], self.args[1])

class day_period(object):
    def __init__(self, object, compression_factor, args=[], kwargs={}):
        self.period = (3600 * 4) / compression_factor
        self.func = object.run
        self.args = args
        self.kwargs = kwargs
        self.seppuku = Event()
    def start(self):
        self.seppuku.clear()
        self.proc = Thread(target=self._doit)
        self.proc.start()
    def stop(self):
        self.seppuku.set()
        self.proc.join()
    def _doit(self):
        while True:
            self.seppuku.wait(self.period)
            if self.seppuku.is_set():
                break
            self.func(self.args)

class intra_day_traders_mng(object):
    def __init__(self, simulation_config):
        self.config = simulation_config
        self.agents_list = []
        self.agents_dict = {}
        self.p_list = []
        self.compression_factor = simulation_config['simulation_config']['compression_factor']
        self.trader_creator()
        self.first_time = True
        self.day_of_simulation  = simulation_config['simulation_config']['day_number']

    def trader_creator(self):
        for agent_name in self.config['agents']['intra_day']:
            for config in self.config['agents']['intra_day'][agent_name]:
                if agent_name == 'nonclassified_trader':
                    for k in range(config['n_traders']):
                        self.agents_list.append(NON_CLASSIFIED_TRADER_INTRADAY(config))
                        time.sleep(.1)
        for agent_name in self.config['agents']['daily']:
            for config in self.config['agents']['daily'][agent_name]:
                if agent_name == 'nonclassified_trader':
                    for k in range(config['n_traders']):
                        self.agents_list.append(NON_CLASSIFIED_TRADER_DAILY(config))
                        time.sleep(0.1)
                if agent_name == "market_maker_trader":
                    for k in range(config['n_traders']):
                        self.agents_list.append(MARKET_MAKER_TRADER_DAILY(config))
                        time.sleep(0.1)
        for agent in self.agents_list:
            self.agents_dict.update({agent.id: agent})
        for agent in self.agents_list:
            agent.set_trader_dict(self.agents_dict)

    def random_initial(self):
        agents_random_list = random.choices(self.agents_list, k=len(self.agents_list))
        return agents_random_list

    def run(self, args):
        lob = args[0]
        reporter_obj = args[1]
        # when the trader running for first time
        if self.first_time == True:
            lob.time_obj.reset()
            agents_random_list = self.random_initial()
            for agent in agents_random_list:
                self.p_list.append(Periodic(agent, self.compression_factor, args=(lob,reporter_obj)))
                self.p_list[-1].start()
                time.sleep(.1)
            self.first_time = False
        else:
            for proc in self.p_list:
                proc.stop()
            for agent in self.agents_list:
                agent.reset_trader(lob)
            time_series = lob.ohcl()
            if len(time_series) == self.day_of_simulation :
                out = {'out':time_series}
                with open('output.json', 'w') as outfile:
                    json.dump(out, outfile)
                reporter_obj.save_as_csv()
                trade_summary = lob.trade_summary()
                with open('trade_report.csv', 'w') as csvFile:
                    writer = csv.writer(csvFile)
                    writer.writerows(trade_summary)
                csvFile.close()
                sys.exit()
            print("***********************************************************************************")
            print("day is:",lob.time_obj.day)
            lob.time_obj.reset()
            for proc in self.p_list:
                proc.start()
                time.sleep(.1)

if __name__ == '__main__':
    with open('config.json', 'r') as f:
        simulation_config = json.load(f)
    intra_day_mng_obj = intra_day_traders_mng(simulation_config)
    reporter_obj = REPORTER()
    # for synchronization of time
    time_obj = TIME_MANAGE(compression_factor=simulation_config['simulation_config']['compression_factor'])
    lob = OrderBook(time_obj, tick_size=simulation_config['simulation_config']['tickSize'])
    day_period(intra_day_mng_obj, simulation_config['simulation_config']['compression_factor'], args=(lob,reporter_obj)).start()

最后是定义自我录音带“在下面的代码中：

class OrderBook():
    def __init__(self, time_obj, tick_size=0.0001):
        self.tape = deque(maxlen=None)  # Index [0] is most recent trade
        self.bids = OrderTree()
        self.asks = OrderTree()
        self.lastTick = None
        self.lastTimestamp = 0
        self.tickSize = tick_size
        self.time = 0
        self.nextQuoteID = 0
        self.time_series = []
        self.time_obj = time_obj

    def clipPrice(self, price):
        return round(price, int(math.log10(1 / self.tickSize)))

    def updateTime(self):
        self.time = int(self.time_obj.now()['time'])

    def processOrder(self, quote, fromData, verbose):
        orderType = quote['type']
        orderInBook = None
        if fromData:
            self.time = quote['timestamp']
        else:
            self.updateTime()
            quote['timestamp'] = self.time
        if quote['qty'] <= 0:
            sys.exit('processLimitOrder() given order of qty <= 0')
        if not fromData: self.nextQuoteID += 1
        if orderType == 'market':
            trades = self.processMarketOrder(quote, verbose)
        elif orderType == 'limit':
            quote['price'] = self.clipPrice(quote['price'])
            trades, orderInBook = self.processLimitOrder(quote, fromData, verbose)
        else:
            sys.exit("processOrder() given neither 'market' nor 'limit'")
        return trades, orderInBook

    def processOrderList(self, side, orderlist,
                         qtyStillToTrade, quote, verbose):
        trades = []
        qtyToTrade = qtyStillToTrade
        while len(orderlist) > 0 and qtyToTrade > 0:
            headOrder = orderlist.getHeadOrder()
            tradedPrice = headOrder.price
            counterparty = headOrder.tid
            if qtyToTrade < headOrder.qty:
                tradedQty = qtyToTrade
                newBookQty = headOrder.qty - qtyToTrade
                headOrder.updateQty(newBookQty, headOrder.timestamp)
                qtyToTrade = 0
            elif qtyToTrade == headOrder.qty:
                tradedQty = qtyToTrade
                if side == 'bid':
                    self.bids.removeOrderById(headOrder.idNum)
                else:
                    self.asks.removeOrderById(headOrder.idNum)
                qtyToTrade = 0
            else:
                tradedQty = headOrder.qty
                if side == 'bid':
                    self.bids.removeOrderById(headOrder.idNum)
                else:
                    self.asks.removeOrderById(headOrder.idNum)
                qtyToTrade -= tradedQty
            if verbose: print('>>> TRADE \nt=%d $%f n=%d p1=%d p2=%d' %
                              (self.time, tradedPrice, tradedQty,
                               counterparty, quote['tid']))

            transactionRecord = {'timestamp': self.time,
                                 'price': tradedPrice,
                                 'qty': tradedQty,
                                 'time': self.time,
                                 'day': self.time_obj.now()['day']}
            if side == 'bid':
                transactionRecord['party1'] = [counterparty,
                                               'bid',
                                               headOrder.idNum]
                transactionRecord['party2'] = [quote['tid'],
                                               'ask',
                                               None]
            else:
                transactionRecord['party1'] = [counterparty,
                                               'ask',
                                               headOrder.idNum]
                transactionRecord['party2'] = [quote['tid'],
                                               'bid',
                                               None]
            self.tape.append(transactionRecord)
            trades.append(transactionRecord)
        return qtyToTrade, trades

    def processMarketOrder(self, quote, verbose):
        trades = []
        qtyToTrade = quote['qty']
        side = quote['side']
        if side == 'bid':
            while qtyToTrade > 0 and self.asks:
                bestPriceAsks = self.asks.minPriceList()
                qtyToTrade, newTrades = self.processOrderList('ask',
                                                              bestPriceAsks,
                                                              qtyToTrade,
                                                              quote, verbose)
                trades += newTrades
        elif side == 'ask':
            while qtyToTrade > 0 and self.bids:
                bestPriceBids = self.bids.maxPriceList()
                qtyToTrade, newTrades = self.processOrderList('bid',
                                                              bestPriceBids,
                                                              qtyToTrade,
                                                              quote, verbose)
                trades += newTrades
        else:
            sys.exit('processMarketOrder() received neither "bid" nor "ask"')
        return trades

    def processLimitOrder(self, quote, fromData, verbose):
        orderInBook = None
        trades = []
        qtyToTrade = quote['qty']
        side = quote['side']
        price = quote['price']
        if side == 'bid':
            while (self.asks and
                   price >= self.asks.minPrice() and
                   qtyToTrade > 0):
                bestPriceAsks = self.asks.minPriceList()
                qtyToTrade, newTrades = self.processOrderList('ask',
                                                              bestPriceAsks,
                                                              qtyToTrade,
                                                              quote, verbose)

                trades += newTrades
            if qtyToTrade > 0:
                if not fromData:
                    quote['idNum'] = self.nextQuoteID
                quote['qty'] = qtyToTrade
                self.bids.insertOrder(quote)
                orderInBook = quote
        elif side == 'ask':
            while (self.bids and
                   price <= self.bids.maxPrice() and
                   qtyToTrade > 0):
                bestPriceBids = self.bids.maxPriceList()
                qtyToTrade, newTrades = self.processOrderList('bid',
                                                              bestPriceBids,
                                                              qtyToTrade,
                                                              quote, verbose)
                trades += newTrades
            if qtyToTrade > 0:
                if not fromData:
                    quote['idNum'] = self.nextQuoteID
                quote['qty'] = qtyToTrade
                self.asks.insertOrder(quote)
                orderInBook = quote
        else:
            sys.exit('processLimitOrder() given neither bid nor ask')
        return trades, orderInBook

    def avg_volume_traded(self):
        if self.tape != None and len(self.tape) > 0:
            num = 0
            total_volume = 0
            for entry in self.tape:
                total_volume = entry['qty'] + total_volume
                num += 1
            if num != 0 and total_volume != None:
                return total_volume, num
        else:
            return None, None

Answer 1

问题是由于processOrderList（）和avg_volume_traded（）之间的访问竞争造成的。前者修改deque，后者迭代deque。在

一个简单的解决方案是让list（）从avg_volume_traded（）中的deque提取数据：

def avg_volume_traded(self):
   if self.tape != None and len(self.tape) > 0:
      num = 0
      total_volume = 0
      for entry in list(self.tape):     # <  atomic extraction step
         total_volume = entry['qty'] + total_volume
         num += 1
      if num != 0 and total_volume != None:
         return total_volume, num
   else:
      return None, None