Configuring Parallelization
PyBroker uses Joblib to compute indicators, train models, and optimize parameters in parallel.
Setting Workers
set_parallel updates the global Joblib configuration. The n_jobs parameter specifies the number of worker jobs: -1 (the default) uses all available CPU cores, and 1 runs sequentially. Read the current settings with
get_parallel_config:
[1]:
from pybroker import set_parallel, get_parallel_config
# Use a fixed number of workers.
set_parallel(n_jobs=4)
print(get_parallel_config())
# Or disable parallel execution entirely.
set_parallel(n_jobs=1)
print(get_parallel_config())
ParallelConfig(n_jobs=4, backend='loky', parallel=None)
ParallelConfig(n_jobs=1, backend='loky', parallel=None)
Parallel Indicators
Indicators are computed per symbol: all indicators for a given symbol are grouped into a single task, and one task is dispatched per symbol. Passing parallel_indicators=True to backtest, walkforward, or
optimize runs these tasks across the configured workers (defaulting to False).
To see this in action, let’s backtest a moving average crossover:
[2]:
import numpy as np
import pybroker
from pybroker import Strategy, YFinance, sumv
pybroker.enable_data_source_cache("parallelization")
set_parallel(n_jobs=-1)
def sma(bar_data, period):
return sumv(bar_data.close, period) / period
sma_20 = pybroker.indicator("sma_20", sma, period=20)
def sma_cross(ctx):
sma_vals = ctx.indicator("sma_20")
if np.isnan(sma_vals[-1]):
return
pos = ctx.long_pos()
if not pos and ctx.close[-1] > sma_vals[-1]:
ctx.buy_shares = 100
elif pos and ctx.close[-1] < sma_vals[-1]:
ctx.sell_all_shares()
yfinance = YFinance()
strategy = Strategy(yfinance, start_date="1/1/2021", end_date="1/1/2026")
strategy.add_execution(sma_cross, ["V", "MA", "AXP"], indicators=sma_20)
result = strategy.backtest(parallel_indicators=True, warmup=20)
result.metrics_df.head()
Backtesting: 2021-01-01 00:00:00 to 2026-01-01 00:00:00
Loading bar data...
[*********************100%***********************] 3 of 3 completed
Loaded bar data: 0:00:00
Computing indicators...
100% (3 of 3) |##########################| Elapsed Time: 0:00:01 Time: 0:00:01
Test split: 2021-01-04 00:00:00 to 2025-12-31 00:00:00
100% (1255 of 1255) |####################| Elapsed Time: 0:00:00 Time: 0:00:00
Finished backtest: 0:00:02
[2]:
| name | value | |
|---|---|---|
| 0 | trade_count | 218 |
| 1 | initial_market_value | 100000.0 |
| 2 | end_market_value | 116119.9 |
| 3 | total_pnl | 15837.68 |
| 4 | unrealized_pnl | 282.22 |
Standalone indicator computation with an IndicatorSet accepts the same parallel_indicators flag:
[3]:
from pybroker import IndicatorSet
df = yfinance.query(
["V", "MA", "AXP"], start_date="1/1/2021", end_date="1/1/2026"
)
ind_set = IndicatorSet()
ind_set.add(sma_20)
ind_set(df, parallel_indicators=True).tail()
Loaded cached bar data.
Computing indicators...
100% (3 of 3) |##########################| Elapsed Time: 0:00:01 Time: 0:00:01
[3]:
| symbol | date | sma_20 | |
|---|---|---|---|
| 3760 | V | 2025-12-24 | 339.050002 |
| 3761 | V | 2025-12-26 | 340.110501 |
| 3762 | V | 2025-12-29 | 341.119000 |
| 3763 | V | 2025-12-30 | 342.280499 |
| 3764 | V | 2025-12-31 | 343.334999 |
Parallel Model Training
Model training runs serially by default. Passing parallel_models=True to backtest or walkforward trains each model in its own task across the configured workers.
This example adapts the linear regression model from Training a Model:
[4]:
from sklearn.linear_model import LinearRegression
from pybroker.indicator import close_minus_ma
cmma_20 = close_minus_ma("cmma_20", lookback=20, atr_length=20)
def train_slr(symbol, train_data, test_data):
# Predict the next bar's return given the 20-day CMMA.
prev_close = train_data["close"].shift(1)
daily_returns = (train_data["close"] - prev_close) / prev_close
train_data["pred"] = daily_returns.shift(-1)
train_data = train_data.dropna()
model = LinearRegression()
model.fit(train_data[["cmma_20"]], train_data[["pred"]])
# Return the trained model and columns to use as input data.
return model, ["cmma_20"]
model_slr = pybroker.model("slr", train_slr, indicators=[cmma_20])
def hold_long(ctx):
if not ctx.long_pos():
if ctx.preds("slr")[-1] > 0:
ctx.buy_shares = 100
elif ctx.preds("slr")[-1] < 0:
ctx.sell_all_shares()
model_strategy = Strategy(yfinance, start_date="1/1/2021", end_date="1/1/2026")
model_strategy.add_execution(hold_long, ["V", "MA", "AXP"], models=model_slr)
result = model_strategy.walkforward(
windows=2,
train_size=0.5,
lookahead=1,
warmup=20,
parallel_models=True,
)
result.metrics_df.head()
Backtesting: 2021-01-01 00:00:00 to 2026-01-01 00:00:00
Loaded cached bar data.
Computing indicators...
100% (3 of 3) |##########################| Elapsed Time: 0:00:00 Time: 0:00:00
Train split: 2021-01-05 00:00:00 to 2022-08-31 00:00:00
Finished training models: 0:00:01
Test split: 2022-09-01 00:00:00 to 2024-05-01 00:00:00
100% (418 of 418) |######################| Elapsed Time: 0:00:00 Time: 0:00:00
Train split: 2022-09-01 00:00:00 to 2024-05-01 00:00:00
Finished training models: 0:00:01
Test split: 2024-05-02 00:00:00 to 2025-12-31 00:00:00
100% (418 of 418) |######################| Elapsed Time: 0:00:00 Time: 0:00:00
Finished backtest: 0:00:03
[4]:
| name | value | |
|---|---|---|
| 0 | trade_count | 52 |
| 1 | initial_market_value | 100000.0 |
| 2 | end_market_value | 146555.0 |
| 3 | total_pnl | 16058.0 |
| 4 | unrealized_pnl | 30497.0 |
Using Ray as the Backend
Ray can distribute the same work across many cores or an entire cluster. Install it with pip install ray and then register it with Joblib via register_ray (from ray.util.joblib):
[5]:
import ray
from ray.util.joblib import register_ray
ray.init(num_cpus=2, include_dashboard=False, ignore_reinit_error=True)
register_ray()
2026-08-11 13:30:35,114 INFO worker.py:2024 -- Started a local Ray instance.
After registering Ray, pass backend="ray" to set_parallel to make it available as a backend:
[6]:
set_parallel(backend="ray", n_jobs=-1)
print(get_parallel_config())
ParallelConfig(n_jobs=-1, backend='ray', parallel=None)
PyBroker will now use the Ray backend for all parallel tasks. For example, calling backtest with parallel_indicators=True:
[7]:
result = strategy.backtest(parallel_indicators=True, warmup=20)
print(f"Total return: {result.metrics.total_return_pct:.2f}%")
ray.shutdown()
Backtesting: 2021-01-01 00:00:00 to 2026-01-01 00:00:00
Loaded cached bar data.
Computing indicators...
100% (3 of 3) |##########################| Elapsed Time: 0:00:02 Time: 0:00:02
Test split: 2021-01-04 00:00:00 to 2025-12-31 00:00:00
100% (1255 of 1255) |####################| Elapsed Time: 0:00:00 Time: 0:00:00
Finished backtest: 0:00:02
Total return: 15.84%
We will explore parameter optimization in the next notebook, which can also be parallelized in certain scenarios.