使用分块与缓存的并行回测

如何突破单核和单机内存限制，扩展研究规模？VBT 将任务分块，在线程、进程或 Ray 集群上运行，并复用已完成结果。您可以控制同时运行的工作量及每次运行保留的内容，让 Python 并行回测扩展到数百万个参数组合。

从自己的回测函数开始

将数据准备、信号和投资组合分析保留在一个函数中，再让 VBT 为每个参数组合运行它。这个小示例在线程中测试三个入场阈值，每个阈值返回一个度量值：

Run a small threshold sweep across threads

close = pd.Series([100.0, 102.0, 101.0, 104.0, 103.0, 106.0])

@vbt.parameterized(merge_func="concat", engine="threadpool", chunk_len=2, show_progress=False)
def test_entry_price(close, entry_price):
    pf = vbt.PF.from_signals(
        close,
        entries=close.vbt.crossed_above(entry_price),
        exits=close.vbt.crossed_below(entry_price),
        fees=0.001,
    )
    return pf.total_return

result = test_entry_price(close, vbt.Param([100.0, 102.0, 104.0], name="entry_price"))
result.round(4)

entry_price
100.0    0.0000
102.0    0.0182
104.0   -0.0010
dtype: float64

这里，chunk_len=2 每次提交两个参数调用。每次调用返回小数形式的总收益率，VBT 为合并结果添加参数值标签。设置 engine="serial" 可顺序运行同一函数。下方一千万组合示例展示如何将此模式扩展到更大任务与搜索空间。更大搜索空间参见策略优化。

一千万个组合，精简结果

Backtest 10,000,000 combinations and keep the best stop per window pair

close = vbt.YFData.pull("BTC-USD", start="2020-01-01", end="2025-01-01").close

@vbt.parameterized(
    merge_func="row_stack",
    engine="threadpool",  
    chunk_len="auto",
    warmup=True,
)
@njit(nogil=True)
def best_stop_nb(close, fast_window, slow_window, sl_stops):
    fast = vbt.nb.rolling_mean_1d_nb(close, fast_window)
    slow = vbt.nb.rolling_mean_1d_nb(close, slow_window)
    entries = vbt.nb.crossed_above_1d_nb(fast, slow)
    exits = vbt.nb.crossed_above_1d_nb(slow, fast)
    n = len(sl_stops)
    sim_out = vbt.pf_nb.from_signals_nb(
        target_shape=(len(close), n),
        group_lens=np.full(n, 1),
        close=close,
        long_entries=entries,
        long_exits=exits,
        sl_stop=sl_stops.reshape(1, -1),
        fees=0.001,
        save_returns=True,
    )
    sharpe = vbt.ret_nb.sharpe_ratio_nb(sim_out.in_outputs.returns, 365.0)
    best = 0
    for i in range(n):  
        if sharpe[i] > sharpe[best]:
            best = i
    return np.array([sharpe[best], sl_stops[best]])

sl_stops = np.linspace(0.001, 0.30, 1000)
best = best_stop_nb(
    close.values,
    vbt.Param(np.arange(2, 102), name="fast_window"),
    vbt.Param(np.arange(105, 305, 2), name="slow_window"),
    sl_stops,
)
best = best.unstack().set_axis(["sharpe", "sl_stop"], axis=1)
best.shape

(10000, 2)

best.nlargest(3, "sharpe").round(3)

                         sharpe  sl_stop
fast_window slow_window
2           107           1.612    0.011
            119           1.603    0.023
            117           1.594    0.050

同一扫描分别用 engine="threadpool" 和 engine="pathos" 运行，后者将分块发送到独立进程。8 核 Apple M3 实测，两者得到相同结果：

引擎	秒
`threadpool`	330
`pathos`	246

线程运行的内存峰值为 802 MB，仅导入 VBT 和加载数据就占用 412 MB，因此 10,000,000 次回测只增加约 390 MB：每个工作单元一次持有 1,000 次回测的一个分块，每对窗口最后仅保留两个数值。尽管每个工作进程都从头启动并自行编译函数，进程方案在这里仍然更快。更快的引擎取决于任务，因此长时间运行前，应先用各引擎对小部分任务计时。

分块将函数参数拆成若干部分，逐块调用函数，再合并结果。@vbt.parameterized 对参数组合分块，@vbt.chunked 沿任意轴对数组分块；VBT 中多数编译函数，包括模拟器，都有每个参数的分块规则。因此在投资组合方法上设置 chunked="threadpool"，无须额外代码即可按列拆分，详见下方分块功能介绍。分块可以嵌套：外层按批次循环控制内存，内层并行运行每个批次。

围绕独立研究任务选择切分方式，例如参数组合或投资组合。内置模拟规则会将共享现金组保持在一起，合并记录时调整列索引，使订单仍属于正确的资产或策略。因此，需要逐笔交易分析时可以保留完整记录，只需排名时则可将每块归约为度量指标。

执行引擎

vbt.execute 在按名称选择的引擎上运行任务列表：

引擎	任务运行方式	适用于
`serial`	依次运行	调试和限制内存
`threadpool`	同一进程中的多个线程	释放 GIL 的编译代码
`pathos`、`processpool`、`mpire`	独立进程	持有 GIL 的 Python 代码
`dask`	Dask 线程或工作进程	已有 Dask 环境
`ray`	本地或远程 Ray 集群	超过单机能力的任务

执行框架提供统一控制，包括分块大小、进度条、每块后的缓存清理或垃圾回收，以及每块前后运行的钩子。应确保工作函数及其依赖可被工作进程访问，例如在可导入模块中定义。启动工作进程可能需要导入 VBT 和编译回调，因此应让每个进程处理多个组合，例如使用 distribute="chunks"，而非一次一个。只选择一层并行：若任务已经在线程池或 Dask 中运行，再在内部使用 Numba 或 Rust parallel=True，会让线程竞争同一批核心。参见下方多线程和多进程功能介绍。

从笔记本到 Ray 集群

同一参数化函数既可使用本地引擎，也可通过 init_kwargs 将 Ray 连接到已有集群。VBT 通过 Ray 提交任务，由 Ray 管理集群。每个任务应批量包含足够多的计算，以抵消工作进程启动和数据传输成本。这适合超出单机能力的独立策略试验、数据处理作业或验证运行。

对于重复作业，可以将现有 vbt.Executor 再传给 vbt.execute，复用其配置。执行钩子还可在分块前后准备共享资源或保存摘要，使同一框架能够处理完整研究流程。

复用结果

投资组合、指标和记录会缓存属性与方法结果，因此 pf.stats() 只计算一次交易或回撤，之后复用。中央注册表跟踪每个缓存，报告命中、未命中和大小，并可按对象或类清理，详见下方缓存。对于长任务，cache_chunks=True 将每个已完成分块保存到磁盘，使重启后的运行可复用匹配的已完成分块，详见下方分块缓存。

这些缓存解决不同需求。对象缓存加速当前进程内的重复分析，分块缓存则在重启后保留已完成工作。Numba 编译缓存复用编译代码，详见计算后端。可以检查并清理选中的对象缓存，或在一次性计算时关闭缓存，避免保留结果只增加内存。

内存

先运行小批次，再同时选择分块大小和工作单元数量。多个工作单元各持有一个分块，可能比单个串行工作单元占用更多内存。仅返回所需度量也可使最终合并结果保持精简，如一千万组合示例所示。

在函数内归约： 像上例一样返回度量或摘要，而非投资组合。
限制记录： 已知每列可能产生多少订单时，用 max_order_records 和 max_log_records 设置模拟所分配的记录数组大小。
跳过结果： 函数返回 vbt.NoResult 会将该组合排除在合并结果之外；skip_errored=True 对抛出错误的组合做同样处理。所有组合都跳过时，raise_no_results=False 返回空结果而非抛出异常。
移除不需要的内容： pf.replace(open=None, high=None, low=None) 让保存的投资组合不再携带三个价格数组，代价是依赖这些数组的图表和交易统计不可用，例如最大有利与不利偏移。
在分块间释放内存： 执行选项中的 clear_cache 和 collect_garbage 在每个分块后运行。进程任务应在拥有内存的工作进程内清理。Pathos 等引擎还支持处理指定数量任务后回收工作进程。

教程

会员专享的超快速 SuperTrend（英文）教程从 pandas 到 Numba 构建指标，再通过线程和分块运行。

分块缓存

新增于v2024.5.15

✅ 大多数使用 VBT 执行框架的工作流，包括数据拉取、分块、参数化、切分和优化，现在都可以将中间结果转存磁盘，并在崩溃重启后重新加载。您可以在云实例上放心测试数十亿个参数组合，无须担心丢失数据。

Execute a basic range with random fallouts while caching successful attempts

np.random.seed(42)

@vbt.parameterized(cache_chunks=True, chunk_len=1)  
def basic_iterator(i):
    print("i:", i)
    rand_number = np.random.uniform()
    if rand_number < 0.2:
        print("failed ⛔")
        raise ValueError
    return i

attempt = 0
while True:
    attempt += 1
    print("attempt", attempt)
    try:
        basic_iterator(vbt.Param(np.arange(10)))
        print("completed 🎉")
        break
    except ValueError:
        pass

attempt 1
i: 0
i: 1
i: 2
i: 3
i: 4
failed ⛔
attempt 2
i: 4
failed ⛔
attempt 3
i: 4
failed ⛔
attempt 4
i: 4
i: 5
i: 6
i: 7
failed ⛔
attempt 5
i: 7
i: 8
i: 9
completed 🎉

分块

新增于1.0.0

✅ 新的创新分块机制可指定参数如何拆分。它自动拆分类数组参数，将每块传给函数执行，再合并结果。因此可以拆分大数组，以分布式方式运行任意函数。VBT 还提供中央注册表，并为大多数 Numba 编译函数（包括模拟函数）的所有参数提供分块规范。只需一条命令即可启用。将大规模搜索拆成可容纳于现有内存的小块！🎉

Backtest at most 100 parameter combinations at once

@vbt.chunked(
    chunk_len=100,
    merge_func="concat",  
    execute_kwargs=dict(  
        clear_cache=True,
        collect_garbage=True
    )
)
def backtest(data, fast_windows, slow_windows):  
    fast_ma = vbt.MA.run(data.close, fast_windows, short_name="fast")
    slow_ma = vbt.MA.run(data.close, slow_windows, short_name="slow")
    entries = fast_ma.ma_crossed_above(slow_ma)
    exits = fast_ma.ma_crossed_below(slow_ma)
    pf = vbt.PF.from_signals(data.close, entries, exits)
    return pf.total_return

param_product = vbt.combine_params(  
    dict(
        fast_window=vbt.Param(range(2, 100), condition="fast_window < slow_window"),
        slow_window=vbt.Param(range(2, 100)),
    ),
    build_index=False
)
backtest(
    vbt.YFData.pull(["BTC-USD", "ETH-USD"]),  
    vbt.Chunked(param_product["fast_window"]),  
    vbt.Chunked(param_product["slow_window"])
)

分块 48/48

fast_window  slow_window  symbol
2            3            BTC-USD    193.124482
                          ETH-USD     12.247315
             4            BTC-USD    159.600953
                          ETH-USD     15.825041
             5            BTC-USD    124.703676
                                            ...
97           98           ETH-USD      3.947346
             99           BTC-USD     25.551881
                          ETH-USD      3.442949
98           99           BTC-USD     27.943574
                          ETH-USD      3.540720
Name: total_return, Length: 9506, dtype: float64

多线程

新增于1.0.0

✅ 集成 concurrent.futures 的 ThreadPoolExecutor、pathos 的 ThreadPool，以及 Dask 后端，可在多个线程中运行多个分块。最适合加速释放 GIL 的重型函数，例如 Numba 和 C 函数。多线程 + 分块 + Numba = 💪

Benchmark 1000 random portfolios without and with multithreading

data = vbt.YFData.pull(["BTC-USD", "ETH-USD"])

%timeit vbt.PF.from_random_signals(data.close, n=[100] * 1000, seed=42)

613 ms ± 37.2 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit vbt.PF.from_random_signals(data.close, n=[100] * 1000, seed=42, chunked="threadpool")

294 ms ± 8.91 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

多进程

新增于1.0.0

✅ 集成 concurrent.futures 的 ProcessPoolExecutor、pathos 的 ProcessPool 与 ParallelPool、mpire 的 WorkerPool，以及 Ray 后端，可跨多个进程运行多个分块。最适合加速不释放 GIL 的重型函数，例如普通 Python 函数，以及容易序列化的轻量参数。想跨多台机器测试数十亿个超参数组合？可通过 Ray 集群在云端扩展函数和整个应用 👀

Benchmark running a slow function on each column without and with multiprocessing

@vbt.chunked(
    size=vbt.ArraySizer(arg_query="items", axis=1),
    arg_take_spec=dict(
        items=vbt.ArraySelector(axis=1)
    ),
    merge_func=np.column_stack
)
def bubble_sort(items):
    items = items.copy()
    for i in range(len(items)):
        for j in range(len(items) - 1 - i):
            if items[j] > items[j + 1]:
                items[j], items[j + 1] = items[j + 1], items[j]
    return items

np.random.seed(42)
items = np.random.uniform(size=(1000, 3))

%timeit bubble_sort(items)

456 ms ± 1.36 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

%timeit bubble_sort(items, _execute_kwargs=dict(engine="pathos"))

165 ms ± 1.51 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

缓存

新增于1.0.0

✅ 缓存已完全重新实现，现由中央注册表管理。由此可以跟踪 VBT 所有可缓存部分的实用统计量，例如显示以 MB 计的缓存总大小。完整控制，清晰可见 🪟

Get the cache statistics after computing the statistics of a random portfolio

data = vbt.YFData.pull("BTC-USD")
pf = vbt.PF.from_random_signals(data.close, n=5, seed=42)
_ = pf.stats()

pf.get_ca_setup().get_status_overview(
    filter_func=lambda setup: setup.caching_enabled,
    include=["hits", "misses", "total_size"]
)

                                 hits  misses total_size
object
portfolio:0.drawdowns               0       1    70.9 kB
portfolio:0.exit_trades             0       1    70.5 kB
portfolio:0.filled_close            6       1    24.3 kB
portfolio:0.init_cash               3       1   32 Bytes
portfolio:0.init_position           0       1   32 Bytes
portfolio:0.init_position_value     0       1   32 Bytes
portfolio:0.init_value              5       1   32 Bytes
portfolio:0.input_value             1       1   32 Bytes
portfolio:0.orders                  9       1    69.7 kB
portfolio:0.total_profit            1       1   32 Bytes
portfolio:0.trades                  0       1    70.5 kB

并行执行与缓存

从自己的回测函数开始

一千万个组合，精简结果

将任务分块

执行引擎

从笔记本到 Ray 集群

复用结果

内存

分块缓存

分块

多线程

多进程

缓存

本页目录