fix(memory): streaming on-disk APIs + worker recycling for backtest pipelines

MDUYN · MDUYN · commit 6ef9cc7483a0 · 2026-05-05T23:00:31.000+02:00
The previous v8.7.1 bounded in-flight tasks but two underlying causes
of OOM remained for users with thousands of large backtests:

1. `recalculate_backtests(List[Backtest])` requires the entire batch
   to be resident in the parent process before the call. With
   thousands of backtests holding portfolio snapshots, trades and
   timeseries, that alone is tens of GB.
2. ProcessPoolExecutor reuses worker processes across tasks. Heavy
   tasks (pandas/polars/numpy intermediates) leave behind cached
   buffers the worker's allocator never returns to the OS, so RSS
   grows over time even with bounded in-flight tasks.

This release adds two new public functions:

- `recalculate_backtests_in_directory(src_dir, dst_dir=None, ...)`:
  streams from disk. The Backtest never crosses the process boundary —
  workers load, recalc, save back, return only the destination path
  and a small index row. Parent memory stays flat regardless of batch
  size.
- `iter_backtests_from_directory(...)`: generator yielding one
  Backtest at a time so callers can process and discard sequentially
  without materialising a List[Backtest].

Also adds worker recycling via `max_tasks_per_child=16` to:
- `recalculate_backtests` (with cross-version fallback)
- `load_backtests_from_directory`
- `migrate_backtests`

On Python 3.11+ this is the native ProcessPoolExecutor parameter; on
3.10 it is emulated by closing and re-opening the executor every
N completions.

Bump version to v8.7.2.
diff --git a/investing_algorithm_framework/__init__.py b/investing_algorithm_framework/__init__.py
@@ -17,6 +17,7 @@
     Position, TimeFrame, INDEX_DATETIME, MarketCredential, TakeProfitRule, \
     PortfolioConfiguration, RESOURCE_DIRECTORY, AWS_LAMBDA_LOGGING_CONFIG, \
     Trade, APP_MODE, AppMode, DATETIME_FORMAT, load_backtests_from_directory, \
+    iter_backtests_from_directory, \
     BacktestDateRange, convert_polars_to_pandas, BacktestRun, \
     DEFAULT_LOGGING_CONFIG, DataType, DataProvider, StopLossRule, \
     ScalingRule, TradingCost, \
@@ -71,7 +72,8 @@
     get_current_average_trade_loss, get_negative_trades, \
     get_positive_trades, get_number_of_trades, get_current_win_rate, \
     get_current_win_loss_ratio, create_backtest_metrics_for_backtest, \
-    recalculate_backtests, TradeTakeProfitService, TradeStopLossService, \
+    recalculate_backtests, recalculate_backtests_in_directory, \
+    TradeTakeProfitService, TradeStopLossService, \
     get_cv_consistency, get_normalized_stability, \
     get_consistency_score, get_stability_score
 
@@ -220,12 +222,14 @@
     "get_number_of_trades",
     "BacktestRun",
     "load_backtests_from_directory",
+    "iter_backtests_from_directory",
     "save_backtests_to_directory",
     "retag_backtests",
     "migrate_backtests",
     "DataError",
     "create_backtest_metrics_for_backtest",
     "recalculate_backtests",
+    "recalculate_backtests_in_directory",
     "TakeProfitRule",
     "StopLossRule",
     "ScalingRule",
diff --git a/investing_algorithm_framework/domain/__init__.py b/investing_algorithm_framework/domain/__init__.py
@@ -44,6 +44,7 @@
     BacktestDateRange, Backtest, BacktestMetrics, combine_backtests, \
     BacktestPermutationTest, BacktestEvaluationFocus, \
     generate_backtest_summary_metrics, load_backtests_from_directory, \
+    iter_backtests_from_directory, \
     save_backtests_to_directory, retag_backtests, migrate_backtests, \
     BacktestIndex, save_bundle, open_bundle, BUNDLE_EXT, \
     BUNDLE_FORMAT_VERSION
@@ -174,6 +175,7 @@
     'ScalingRule',
     'TradingCost',
     "load_backtests_from_directory",
+    "iter_backtests_from_directory",
     "save_backtests_to_directory",
     "retag_backtests",
     "migrate_backtests",
diff --git a/investing_algorithm_framework/domain/backtesting/__init__.py b/investing_algorithm_framework/domain/backtesting/__init__.py
@@ -9,6 +9,7 @@
     generate_backtest_summary_metrics
 from .backtest_utils import (
     load_backtests_from_directory,
+    iter_backtests_from_directory,
     save_backtests_to_directory,
     retag_backtests,
     migrate_backtests,
@@ -33,6 +34,7 @@
     "combine_backtests",
     "generate_backtest_summary_metrics",
     "load_backtests_from_directory",
+    "iter_backtests_from_directory",
     "save_backtests_to_directory",
     "retag_backtests",
     "migrate_backtests",
diff --git a/investing_algorithm_framework/domain/backtesting/backtest_utils.py b/investing_algorithm_framework/domain/backtesting/backtest_utils.py
@@ -410,22 +410,47 @@ def load_backtests_from_directory(
     else:
         # Process pool only handles bundle loads efficiently; loading
         # legacy directories from workers also works but the win is
-        # smaller. We still parallelise both for simplicity.
-        with ProcessPoolExecutor(max_workers=n_workers) as ex:
-            futures = {
-                ex.submit(_load_one_dispatch, it): it for it in sources
-            }
-            for fut in as_completed(futures):
-                src = futures[fut]
+        # smaller. We still parallelise both for simplicity. Submission
+        # is bounded to ``n_workers`` in flight so the executor's
+        # feeder thread does not buffer the entire source list, and
+        # workers are recycled (Python 3.11+) so RSS stays bounded.
+        import sys
+        pool_kwargs = {"max_workers": n_workers}
+        if sys.version_info >= (3, 11):
+            pool_kwargs["max_tasks_per_child"] = 16
+
+        sources_iter = iter(sources)
+        inflight = {}
+
+        with ProcessPoolExecutor(**pool_kwargs) as ex:
+            def _submit_next():
                 try:
-                    backtests.append(fut.result())
-                except Exception as e:
-                    logger.error(
-                        f"Failed to load backtest from {src[0]}: {e}"
-                    )
-                finally:
-                    if pbar is not None:
-                        pbar.update(1)
+                    it = next(sources_iter)
+                except StopIteration:
+                    return False
+                inflight[ex.submit(_load_one_dispatch, it)] = it
+                return True
+
+            for _ in range(n_workers):
+                if not _submit_next():
+                    break
+
+            while inflight:
+                done_set, _unused = wait(
+                    inflight.keys(), return_when=FIRST_COMPLETED
+                )
+                for fut in done_set:
+                    src = inflight.pop(fut)
+                    try:
+                        backtests.append(fut.result())
+                    except Exception as e:
+                        logger.error(
+                            f"Failed to load backtest from {src[0]}: {e}"
+                        )
+                    finally:
+                        if pbar is not None:
+                            pbar.update(1)
+                    _submit_next()
 
     if pbar is not None:
         pbar.close()
@@ -439,6 +464,87 @@ def load_backtests_from_directory(
     return backtests
 
 
+def iter_backtests_from_directory(
+    directory_path: Union[str, Path],
+    filter_function: Callable[[Backtest], bool] = None,
+    number_of_backtests_to_load: int = None,
+    show_progress: bool = False,
+):
+    """Yield :class:`Backtest` objects one at a time from *directory_path*.
+
+    Memory-stable alternative to :func:`load_backtests_from_directory`:
+    each backtest is loaded only when the caller asks for the next
+    item, so peak memory equals one backtest at a time. Use this when
+    a directory contains so many large backtests that the full list
+    would not fit in RAM.
+
+    Example:
+        >>> for bt in iter_backtests_from_directory(path):
+        ...     process(bt)
+        ...     # bt is dropped here; its memory is reclaimed before
+        ...     # the next one is read from disk.
+
+    Args:
+        directory_path: Source directory.
+        filter_function: Optional predicate; backtests for which it
+            returns False are skipped.
+        number_of_backtests_to_load: Optional cap on the number of
+            backtests yielded.
+        show_progress: Display a tqdm progress bar.
+    """
+    directory_path = str(directory_path)
+    if not os.path.exists(directory_path):
+        logger.warning(
+            f"Directory {directory_path} does not exist. "
+            "No backtests loaded."
+        )
+        return
+
+    sources: List[tuple] = []
+    for entry in sorted(os.listdir(directory_path)):
+        if entry == "checkpoints.json" or entry.endswith(".py"):
+            continue
+        if entry == "index.parquet" or entry == "ohlcv":
+            continue
+        full = os.path.join(directory_path, entry)
+        if os.path.isfile(full):
+            if entry.endswith(BUNDLE_EXT) or is_bundle_file(full):
+                sources.append((full, "bundle"))
+        elif os.path.isdir(full):
+            sources.append((full, "directory"))
+
+    if number_of_backtests_to_load is not None:
+        sources = sources[: max(0, int(number_of_backtests_to_load))]
+
+    pbar = tqdm(
+        total=len(sources),
+        desc="Loading backtests",
+        disable=not show_progress,
+    )
+    try:
+        for item in sources:
+            try:
+                bt = _load_one_dispatch(item)
+            except Exception as e:
+                logger.error(
+                    f"Failed to load backtest from {item[0]}: {e}"
+                )
+                pbar.update(1)
+                continue
+            try:
+                if filter_function is not None and not filter_function(bt):
+                    pbar.update(1)
+                    continue
+            except Exception as fe:  # pragma: no cover - defensive
+                logger.error(f"Error in filter_function: {fe}")
+                pbar.update(1)
+                continue
+            yield bt
+            pbar.update(1)
+    finally:
+        pbar.close()
+
+
 # ---------------------------------------------------------------------------
 # Index helpers (#487 step 4)
 # ---------------------------------------------------------------------------
@@ -699,8 +805,14 @@ def migrate_backtests(
             # Bound in-flight tasks to ``resolved_workers`` so we don't
             # buffer the full plan inside the executor's feeder, and
             # consume results as they finish to keep memory flat.
+            # Recycle workers (3.11+) so RSS stays bounded across long
+            # migrations.
+            import sys
+            pool_kwargs = {"max_workers": resolved_workers}
+            if sys.version_info >= (3, 11):
+                pool_kwargs["max_tasks_per_child"] = 16
             plan_iter = iter(plan)
-            with ProcessPoolExecutor(max_workers=resolved_workers) as ex:
+            with ProcessPoolExecutor(**pool_kwargs) as ex:
                 inflight = {}
                 for _ in range(resolved_workers):
                     try:
diff --git a/investing_algorithm_framework/services/__init__.py b/investing_algorithm_framework/services/__init__.py
@@ -38,7 +38,8 @@
     get_current_win_rate, get_current_average_trade_return, \
     get_current_average_trade_loss, get_current_average_trade_duration, \
     get_current_average_trade_gain, create_backtest_metrics_for_backtest, \
-    recalculate_backtests, get_cv_consistency, get_normalized_stability, \
+    recalculate_backtests, recalculate_backtests_in_directory, \
+    get_cv_consistency, get_normalized_stability, \
     get_consistency_score, get_stability_score
 
 __all__ = [
@@ -133,6 +134,7 @@
     "get_current_average_trade_return",
     "create_backtest_metrics_for_backtest",
     "recalculate_backtests",
+    "recalculate_backtests_in_directory",
     "TradeStopLossService",
     "TradeTakeProfitService",
     "get_mean_yearly_return",
diff --git a/investing_algorithm_framework/services/metrics/__init__.py b/investing_algorithm_framework/services/metrics/__init__.py
@@ -28,7 +28,8 @@
     get_current_win_loss_ratio
 from .calmar_ratio import get_calmar_ratio
 from .generate import create_backtest_metrics, \
-    create_backtest_metrics_for_backtest, recalculate_backtests
+    create_backtest_metrics_for_backtest, recalculate_backtests, \
+    recalculate_backtests_in_directory
 from .risk_free_rate import get_risk_free_rate_us
 from .trades import get_negative_trades, get_positive_trades, \
     get_number_of_trades, get_number_of_closed_trades, \
diff --git a/investing_algorithm_framework/services/metrics/generate.py b/investing_algorithm_framework/services/metrics/generate.py
diff --git a/pyproject.toml b/pyproject.toml
