pybroker.optimize 模块

使用 Optuna 进行超参数声明和优化。

超参数为指标和执行函数声明可调节的值。每个超参数都会通过 hyperparam() 以名称全局注册，并在回测或优化时被解析为一个具体的 int 或 float。

将超参数作为关键字参数传递给 pybroker.indicator.indicator()，或将其列在 pybroker.strategy.Strategy.add_execution() 上，以便在执行函数内通过 ctx.hyperparam(name) 读取。

class Hyperparam(name: str, default: int | float, low: int | float, high: int | float, step: int | float)[源代码]

基类：object

声明一个带有边界和步长的具名超参数。

通过 hyperparam() 创建，并以 name 全局注册。

name

用于指标关键字参数、执行函数超参数列表和优化结果中的唯一标识符。

类型:: str

default

用于回测的值，也是优化期间的基线值。应位于 [low, high] 范围内。

类型:: int | float

low

优化期间搜索的候选值下限（含）。

类型:: int | float

high

优化期间搜索的候选值上限（含）。候选值依次为 low、 low + step…，直到不超过 high 的最大值。

类型:: int | float

step

候选值之间的间隔。必须为正数。整数超参数使用整数步长；浮点数超参数使用浮点数步长，其值会被四舍五入以匹配 Optuna 的分步建议值。

类型:: int | float

示例

指标周期从 5 到 50，步长为 5:

period = hyperparam("period", default=14, low=5, high=50, step=5)

class ObjectiveBundle(objective: Callable[[Trial], float], search_space: SearchSpace, score_overrides: Callable[[dict[str, Any]], float])[源代码]

基类：object

make_objective() 的返回值。

objective: Callable[[Trial], float]

score_overrides: Callable[[dict[str, Any]], float]

search_space: SearchSpace

class OptimizeMixin[源代码]

基类：object

实现超参数优化的 Mixin。

optimize(score_fn: Callable[[TestResult], float], *, sampler: str | BaseSampler = 'grid', n_trials: int | None = None, direction: str = 'maximize', seed: int | None = None, windows: int | None = None, study: optuna.Study | None = None, pruner: optuna.pruners.BasePruner | None = None, train_size: float = 0.5, lookahead: int = 1, start_date: str | datetime | None = None, end_date: str | datetime | None = None, timeframe: str = '', between_time: tuple[str, str] | None = None, days: Any | None = None, warmup: int | None = None, parallel_indicators: bool = False, adjust: Any | None = None, calc_bootstrap: bool = False, verbose: bool = False) → OptimizeResult[源代码]

在训练窗口上搜索 pybroker.optimize.hyperparam() 的值，然后在保留的测试窗口上评估最佳值。

由 pybroker.data.DataSource 提供的数据会按照 train_size 指定的比例划分为训练集和测试集。每次试验都会使用一种超参数值组合在训练窗口上进行回测，并用 score_fn 为其打分。获胜的组合随后会在测试窗口上重放，而 score_fn 永远不会看到测试窗口。

预训练模型（model(..., pretrained=True)）会按训练窗口加载，并在该窗口的所有试验中复用。不支持可训练模型；请在 train_fn 内部使用验证集划分来调优，或改用 pybroker.strategy.Strategy.walkforward()。

参数:

score_fn -- 为单次试验的训练窗口回测打分的 Callable[[TestResult], float]。默认按最大化处理；参见 direction。
sampler -- 候选值的选取方式。 "grid" （默认）会穷举所有组合， "tpe" 使用 optuna.samplers.TPESampler， "random" 使用 optuna.samplers.RandomSampler。也可以传入 optuna.samplers.BaseSampler 实例；它会按窗口进行深拷贝并重新播种，而多窗口运行会将副本分发给工作进程，因此该实例必须是可 pickle 的。网格采样器和随机采样器会在已配置的工作进程上并行评估试验。任何其他采样器 —— 即 "tpe"，或不是 GridSampler 也不是 RandomSampler 的实例 —— 都是自适应的，分批评估其试验会改变其提出的值，并使结果与工作进程数量绑定；因此这类采样器的试验会按顺序运行，并会有一条 info 级别的日志消息说明并行已被禁用。
n_trials -- 要运行的试验数量。除 "grid" 外，其他所有采样器都必须设置该参数；对于 "grid"，默认值为完整网格大小，设置较小的值则会随机抽取相应数量的组合。
direction -- 对 score_fn 进行 "maximize" （默认）或 "minimize"。
seed -- 用于采样器和自助法指标的随机种子。默认为 None，表示不可复现。
windows -- 当大于 1 时，超参数会在 windows 个向前分析窗口中分别独立优化，各测试窗口会被拼接为一个连续的结果。默认为 None，即单次训练/测试划分。
study -- 用于记录试验的现有 optuna.study.Study，例如一个由持久化存储支持的 study。会使用该 study 自身的采样器和剪枝器，且其优化方向必须与 direction 一致。当 windows 大于 1 时不支持此参数。
pruner -- 附加到所创建 study 的 optuna.pruners.BasePruner。每次试验都是一次完整的回测，没有可报告的中间值，因此剪枝实际上永远不会触发。
train_size -- 每个窗口中用于训练的比例，不含 0 和 1。默认为 0.5。
lookahead -- 预测目标未来的 K 线数量。会在训练集和测试集之间保留出来，以防止训练数据跨边界泄漏，其单位是每个模型所拟合的时间框架的 K 线：通过 pybroker.model.ModelSource.intervals() 绑定到某个区间的模型，会保留该区间的 lookahead 根 K 线，而不是基础时间框架的 K 线。默认为 1。
start_date -- 优化的起始日期（含）。必须在传递给 pybroker.strategy.Strategy 构造函数的范围之内。
end_date -- 优化的结束日期（含）。必须在传递给 pybroker.strategy.Strategy 构造函数的范围之内。
timeframe -- 指定数据时间框架分辨率的格式化字符串，用法与 pybroker.strategy.Strategy.walkforward() 中相同。
between_time -- 用于过滤数据的时间段（含边界），为 tuple[str, str]，例如 ('9:30', '16:00')。
days -- 用于过滤数据的星期（例如 "mon"、 "tues"）。
warmup -- 运行执行函数之前需要经过的 K 线数量。设置时必须大于 0。
parallel_indicators -- 如果为 True，pybroker.indicator.Indicator 数据会使用多个进程并行计算。默认为 False。
adjust -- 对 pybroker.data.DataSource 应用的复权调整类型。
calc_bootstrap -- 是否为测试结果计算随机自助法评估指标。默认为 False。
verbose -- 是否记录每次试验回测的日志 —— 包括指标计算、测试集划分进度条以及 Optuna 自身的试验日志。默认为 False，此时仅记录优化摘要和最终测试窗口的评估结果。

返回:

包含获胜超参数值、其在训练窗口获得的得分，以及其产生的测试窗口 pybroker.strategy.TestResult 的 OptimizeResult。当 windows 大于 1 时， best_params、 best_score 和 study 描述的是 最后一个 窗口，而 result 则是所有窗口拼接后的结果；各窗口的结果参见 OptimizeResult.windows。

抛出:

ValueError -- 如果未添加任何执行函数；如果任何模型源是可训练的；如果 train_size 不在 0 和 1 之间（不含）；如果 warmup 不大于 0；如果 windows 不大于 0；如果 study 与大于 1 的 windows 同时使用，或其优化方向存在冲突；或者如果日期超出了传递给 pybroker.strategy.Strategy 构造函数的范围。

class OptimizeResult(best_params: dict[str, Any], best_score: float, result: TestResult, study: optuna.Study, windows: tuple[WindowOptimizeResult, ...] | None = None)[源代码]

基类：object

Strategy.optimize() 的结果。

best_params

获胜的超参数值，包括那些被固定而非搜索的值。当设置了 windows 时，这些是 最后一个 窗口的值，因为每个窗口都是分别调优的。

类型:: dict[str, Any]

best_score

best_params 在训练窗口上获得的 score_fn 值。当设置了 windows 时，为最后一个窗口的得分。

类型:: float

result

测试窗口的 pybroker.strategy.TestResult。当设置了 windows 时，这是一个由每个窗口的测试数据拼接而成的单一连续结果，仓位和现金会跨窗口边界延续。

类型:: TestResult

study

保存试验的 optuna.study.Study。当设置了 windows 时，为最后一个窗口的 study；其余窗口参见 windows。

类型:: optuna.Study

windows

各窗口的调优结果；对于单次训练/测试划分则为 None。

类型:: Optional[tuple[WindowOptimizeResult, ...]]

to_json(*, include: frozenset[str] | None = None, max_rows: int | None = 100, symbols: frozenset[str] | None = None) → dict[str, Any][源代码]: 返回可 JSON 序列化的优化结果。

to_json_str(*, include: frozenset[str] | None = None, max_rows: int | None = 100, symbols: frozenset[str] | None = None) → str[源代码]: 从 to_json() 返回严格的 JSON 文本。

class SearchSpace(hyperparams: frozenset[str], specs: Mapping[str, Hyperparam])[源代码]

基类：object

从策略中收集到的可搜索超参数。

仅包含满足 low < high 且会在 Strategy.optimize() 期间传递给 Optuna 的超参数。

hyperparams

优化期间被搜索的超参数名称。

类型:: frozenset[str]

specs

超参数名称到 Hyperparam 规格的映射。

类型:: Mapping[str, pybroker.optimize.Hyperparam]

grid_size() → int[源代码]: 网格组合的总数。

class WindowOptimizeResult(params: dict[str, Any], study: Study, train_score: float, train_start_date: datetime | None = None, train_end_date: datetime | None = None, test_start_date: datetime | None = None, test_end_date: datetime | None = None, execution_symbols: dict[int, frozenset[str]] | None = None)[源代码]

基类：object

各窗口的向前分析优化结果。

保存某个窗口被调优到的值，本身并不是一次独立的回测：唯一的样本外 pybroker.strategy.TestResult 存放在 OptimizeResult.result 中，由所有窗口拼接而成。

params

该窗口获胜的超参数值，包括那些被固定而非搜索的值。

类型:: dict[str, Any]

study

保存该窗口试验的 optuna.study.Study。

类型:: optuna.study.study.Study

train_score

params 在该窗口训练数据上获得的 score_fn 值。

类型:: float

train_start_date

该窗口训练数据的起始日期；当训练集划分为空时为 None。

类型:: datetime.datetime | None

train_end_date

该窗口训练数据的结束日期；当训练集划分为空时为 None。

类型:: datetime.datetime | None

test_start_date

该窗口测试数据的起始日期 —— 即其调优后的 params 在拼接后的 OptimizeResult.result 中进行交易的区间 —— 当测试集划分为空时为 None。

类型:: datetime.datetime | None

test_end_date

该窗口测试数据的结束日期；当测试集划分为空时为 None。

类型:: datetime.datetime | None

execution_symbols

当 pybroker.common.SymbolSelector 为该窗口选择品种时，每个执行 id 所解析到的品种；如果没有任何执行使用选择器，则为 None。

类型:: dict[int, frozenset[str]] | None

to_json() → dict[str, Any][源代码]: 返回可 JSON 序列化的向前分析窗口优化结果。

to_json_str() → str[源代码]: 从 to_json() 返回严格的 JSON 文本。

build_run_hyperparams(specs: Mapping[str, Hyperparam], overrides: dict[str, Any] | None = None) → dict[str, Any][源代码]

为单次回测或单次试验构建超参数字典。

参数:

specs -- 从策略可达的全部超参数。
overrides -- 要合并到默认值之上的试验值或用户提供的值。

返回:

specs 中每个超参数的 name -> value 字典。

collect_hyperparams(strategy: _ExecutionsHost) → dict[str, Hyperparam][源代码]: 收集从 strategy 可达的全部超参数。

collect_search_space(strategy: _ExecutionsHost) → SearchSpace[源代码]: 收集从 strategy 可达的可搜索超参数。

hyperparam(name: str, *, default: int | float, low: int | float, high: int | float, step: int | float) → Hyperparam[源代码]

创建并注册一个 Hyperparam。

参数:

name -- 该超参数的唯一标识符。在指标关键字参数、 add_execution(..., hyperparams=[...]) 以及 ctx.hyperparam(name) 中被引用。
default -- 用于回测的值。
low -- 优化期间搜索的候选值下限（含）。
high -- 优化期间搜索的候选值上限（含）。
step -- 候选值之间的间隔。必须为正数。

返回:

已注册的 Hyperparam 实例。

make_objective(strategy: _OptimizeTrialHost, score_fn: Callable[[TestResult], float], *, train_rows: np.ndarray, df: pd.DataFrame, hyperparams: Mapping[str, Hyperparam], search_space: SearchSpace, invariant_indicator_data: dict[IndicatorSymbol, pd.Series], window_executions: set[Execution], master_store: Any, interval_data: Any, parallel_indicators: bool, warmup: int | None, pretrained_models: Mapping[ModelSymbol, TrainedModel], exit_dates: Mapping[str, np.datetime64], verbose: bool = False) → ObjectiveBundle[源代码]

构建一个用于训练窗口打分的 Optuna 目标函数。

当 verbose 为 False （默认值）时，每次试验的回测都会在抑制日志的情况下运行，以避免每个搜索到的组合都重复显示逐次试验的进度条。