feat(backtesting): streaming migrate_backtests with resume support

migrate_backtests() now fuses load+save inside each worker process
instead of loading every backtest into the parent first. Memory
usage stays roughly constant regardless of source size.

New options:
- include_ohlcv: include OHLCV data in destination bundles
- skip_existing (default True): skip backtests whose destination
  bundle already exists, making interrupted migrations resumable.

The CLI 'iaf migrate-backtests' exposes both flags. Documented
the function in Getting Started/backtesting.md.

Author: MDUYN

docusaurus/docs/Getting Started/backtesting.md

@@ -89,6 +89,39 @@ backtests = load_backtests_from_directory("./my_backtests")
 Backtests are persisted in the framework's optimized **`.iafbt` bundle format** — a single binary file per backtest using zstd compression + MessagePack encoding. Compared to the legacy directory layout it is ~21× smaller, ~27× fewer files, and ~3× faster to load. Both `save_backtests_to_directory` and `load_backtests_from_directory` support parallel I/O via `workers=N`. Existing legacy directories keep working transparently; use `iaf migrate-backtests --src ... --dst ...` to convert them.
 :::
 
+### Migrating Existing Backtests
+
+Convert a directory of legacy backtest folders to the new bundle
+format with `migrate_backtests`. The migration is **streamed**
+(load+save fused per worker), so memory usage stays roughly
+constant regardless of how many backtests you migrate, and an
+interrupted run can simply be re-invoked to resume — destination
+bundles that already exist are skipped by default.
+
+```python
+from investing_algorithm_framework import migrate_backtests
+
+n = migrate_backtests(
+    src_dir="./old_backtests",       # legacy folders and/or .iafbt
+    dst_dir="./bundled_backtests",   # destination
+    workers=8,                        # parallel; None = min(8, cpu)
+    show_progress=True,
+    write_index=True,                 # also build index.parquet
+    include_ohlcv=False,
+    skip_existing=True,               # resume-safe (default)
+)
+print(f"migrated {n} backtests")
+```
+
+Or from the CLI:
+
+```bash
+iaf migrate-backtests \
+    --src ./old_backtests \
+    --dst ./bundled_backtests \
+    --workers 8
+```
+
 ## Reporting
 
 Use [Backtest Reports](/docs/Getting%20Started/backtest-reports) to turn

investing_algorithm_framework/cli/cli.py

@@ -273,13 +273,28 @@ def mcp(directory):
     "--no-index", is_flag=True, default=False,
     help="Skip writing index.parquet at the destination.",
 )
-def migrate_backtests_cmd(src, dst, workers, no_index):
+@click.option(
+    "--include-ohlcv", is_flag=True, default=False,
+    help="Include OHLCV data in the destination bundles.",
+)
+@click.option(
+    "--no-skip-existing", is_flag=True, default=False,
+    help="Re-migrate even if the destination bundle already exists.",
+)
+def migrate_backtests_cmd(
+    src, dst, workers, no_index, include_ohlcv, no_skip_existing
+):
     """Convert a directory of legacy backtest folders into the bundled
     binary format introduced in issue #487.
 
     The new ``.iafbt`` format is a single zstd-compressed MessagePack
     file per backtest. Loading bundled directories is dramatically
     faster than the legacy multi-file layout for large batches.
+
+    Migration is streamed (load+save fused per worker) so memory
+    usage stays roughly constant regardless of source size, and
+    interrupted runs can be resumed (existing destination bundles
+    are skipped by default).
     """
     from investing_algorithm_framework.domain import migrate_backtests
 
@@ -289,6 +304,8 @@ def migrate_backtests_cmd(src, dst, workers, no_index):
         workers=workers,
         show_progress=True,
         write_index=not no_index,
+        include_ohlcv=include_ohlcv,
+        skip_existing=not no_skip_existing,
     )
     click.echo(f"Migrated {n} backtest(s) from {src} to {dst}")
 

investing_algorithm_framework/domain/backtesting/backtest_utils.py

@@ -565,26 +565,130 @@ def load_backtests(
 # ---------------------------------------------------------------------------
 
 
+def _migrate_one(args):
+    """Worker entry point: open *src* (legacy dir or bundle), write
+    *dst* as a bundle, return the destination path.
+
+    Doing load+save in one worker call keeps each backtest's data in
+    a single process — avoiding the cost of pickling fully-decoded
+    Backtest objects back to the parent process. This roughly halves
+    peak memory usage for large migrations and is faster end-to-end.
+    """
+    src, dst, include_ohlcv, ohlcv_store = args
+    bt = _open_bundle(src) if is_bundle_file(src) else Backtest.open(src)
+    return str(_save_bundle(
+        bt, dst,
+        include_ohlcv=include_ohlcv,
+        ohlcv_store=ohlcv_store,
+    ))
+
+
 def migrate_backtests(
     src_dir: Union[str, Path],
     dst_dir: Union[str, Path],
     *,
     workers: Optional[int] = None,
     show_progress: bool = False,
     write_index: bool = True,
+    include_ohlcv: bool = False,
+    skip_existing: bool = True,
 ) -> int:
-    """Rewrite a directory of legacy backtest folders as ``.iafbt``
-    bundles in *dst_dir*. Returns the number of backtests migrated.
+    """Rewrite a directory of legacy backtest folders (or existing
+    ``.iafbt`` bundles) as ``.iafbt`` bundles in *dst_dir*.
+
+    The migration is streamed: each backtest is loaded and re-saved
+    inside a single worker process, so memory usage stays roughly
+    constant regardless of how many backtests are being migrated.
+
+    Args:
+        src_dir: Source directory containing legacy backtest folders
+            and/or ``.iafbt`` bundles. Walked recursively.
+        dst_dir: Destination directory. Created if it does not exist.
+        workers: Number of parallel workers. ``None`` picks
+            ``min(8, cpu_count)``. Pass ``1`` to force serial.
+        show_progress: Display a progress bar.
+        write_index: Write a sibling ``index.parquet`` summarising the
+            destination bundles for fast filtering with
+            :class:`BacktestIndex`.
+        include_ohlcv: Include OHLCV data in the destination bundles.
+        skip_existing: Skip backtests whose destination bundle already
+            exists. Allows resuming an interrupted migration.
+
+    Returns:
+        Number of backtests migrated (excluding skipped ones).
     """
-    backtests = load_backtests_from_directory(
-        src_dir, workers=workers, show_progress=show_progress,
+    src_dir = Path(src_dir)
+    dst_dir = Path(dst_dir)
+    dst_dir.mkdir(parents=True, exist_ok=True)
+
+    # Discover sources: any *.iafbt file or any directory shaped like
+    # a legacy backtest (algorithm_id.json + runs/).
+    sources: List[str] = []
+    for root, dirs, files in os.walk(src_dir):
+        for fname in files:
+            if fname.endswith(BUNDLE_EXT):
+                sources.append(os.path.join(root, fname))
+        for dname in list(dirs):
+            d = os.path.join(root, dname)
+            if (
+                os.path.isfile(os.path.join(d, "algorithm_id.json"))
+                and os.path.isdir(os.path.join(d, "runs"))
+            ):
+                sources.append(d)
+                # Don't descend into a recognised backtest dir.
+                dirs.remove(dname)
+
+    if not sources:
+        return 0
+
+    # Build (src, dst) pairs, optionally skipping ones that already
+    # exist in dst_dir.
+    ohlcv_store = (
+        str(dst_dir / "ohlcv") if include_ohlcv else None
     )
-    save_backtests_to_directory(
-        backtests,
-        dst_dir,
-        format=FORMAT_BUNDLE,
-        workers=workers,
-        show_progress=show_progress,
-        write_index=write_index,
-    )
-    return len(backtests)
+    plan = []
+    for src in sources:
+        base = os.path.basename(os.path.normpath(src))
+        if base.endswith(BUNDLE_EXT):
+            base = base[: -len(BUNDLE_EXT)]
+        dst = str(dst_dir / f"{base}{BUNDLE_EXT}")
+        if skip_existing and os.path.isfile(dst):
+            continue
+        plan.append((src, dst, include_ohlcv, ohlcv_store))
+
+    if not plan:
+        return 0
+
+    n = len(plan)
+    resolved_workers = min(_resolve_workers(workers), n)
+
+    iterator: object
+    if resolved_workers > 1:
+        with ProcessPoolExecutor(max_workers=resolved_workers) as ex:
+            results = ex.map(_migrate_one, plan)
+            iterator = tqdm(
+                results,
+                total=n,
+                desc="Migrating backtests",
+                disable=not show_progress,
+            )
+            for _ in iterator:
+                pass
+    else:
+        for args in tqdm(
+            plan,
+            total=n,
+            desc="Migrating backtests",
+            disable=not show_progress,
+        ):
+            _migrate_one(args)
+
+    if write_index:
+        # Re-open the freshly written bundles (cheap header reads only
+        # for the index) to build the parquet manifest.
+        migrated = load_backtests_from_directory(
+            dst_dir, workers=workers, show_progress=False,
+        )
+        _write_index(dst_dir, migrated)
+
+    return n