From e1b376a49b23fb468e5eaf5cda0d1463fbf8975b Mon Sep 17 00:00:00 2001 From: Eddie Hebert Date: Thu, 14 Apr 2016 16:01:42 -0400 Subject: [PATCH] BUG: Add limit to memory growth on sliding windows Add a cap of 5 sliding windows (one per OHCLV column) to the history loader's cache of sliding windos. This prevents unbounded growth on algorithms that call history with a highly varied list of equities. To follow is splitting the cache up by column and by sid, so that the loader does not re-prefetch sids which have already been read with sufficient data; however this patch is enough to fix the issue where an algo with high rotation can add up a megabyte per day of memory on algorithms which rotate on a 5% dollar volume pipeline. With this cap those algorithms have more plateaus with regard to memory consumption. This patch requires new dependency of `cachetools` library. --- etc/requirements.txt | 2 ++ zipline/data/us_equity_loader.py | 18 +++++++++--------- 2 files changed, 11 insertions(+), 9 deletions(-) diff --git a/etc/requirements.txt b/etc/requirements.txt index 7ffa9527..6abd8169 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -58,3 +58,5 @@ sqlalchemy==1.0.8 # for intervaltree sortedcontainers==1.4.4 intervaltree==2.1.0 + +cachetools==1.1.5 diff --git a/zipline/data/us_equity_loader.py b/zipline/data/us_equity_loader.py index b1690960..cb9a9a14 100644 --- a/zipline/data/us_equity_loader.py +++ b/zipline/data/us_equity_loader.py @@ -17,6 +17,8 @@ from abc import ( abstractmethod, abstractproperty, ) + +from cachetools import LRUCache from numpy import dtype, around from pandas.tslib import normalize_date @@ -25,7 +27,7 @@ from six import iteritems, with_metaclass from zipline.pipeline.data.equity_pricing import USEquityPricing from zipline.lib._float64window import AdjustedArrayWindow as Float64Window from zipline.lib.adjustment import Float64Multiply -from zipline.utils.cache import CachedObject, Expired +from zipline.utils.cache import ExpiringCache from zipline.utils.memoize import lazyval @@ -82,7 +84,8 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): self.env = env self._reader = reader self._adjustments_reader = adjustment_reader - self._window_blocks = {} + # TODO: Split cache into 'by column' and 'by sid'. + self._window_blocks = ExpiringCache(LRUCache(maxsize=5)) @abstractproperty def _prefetch_length(self): @@ -218,11 +221,7 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): size = len(dts) assets_key = frozenset(assets) try: - block_cache = self._window_blocks[(assets_key, field, size)] - try: - return block_cache.unwrap(end) - except Expired: - pass + return self._window_blocks.get((assets_key, field, size), end) except KeyError: pass @@ -253,8 +252,9 @@ class USEquityHistoryLoader(with_metaclass(ABCMeta)): size ) block = SlidingWindow(window, size, start_ix, offset) - self._window_blocks[(assets_key, field, size)] = CachedObject( - block, prefetch_end) + self._window_blocks.set((assets_key, field, size), + block, + prefetch_end) return block def history(self, assets, dts, field):