pybroker.strategy 模块

包含用于回测交易策略的实现。

class BacktestMixin[源代码]

基类：object

实现回测功能的 Mixin。

backtest_executions(config: StrategyConfig, executions: set[Execution], before_exec_fn: Callable[[Mapping[str, ExecContext]], None] | None, after_exec_fn: Callable[[Mapping[str, ExecContext]], None] | None, sessions: Mapping[str, MutableMapping], models: Mapping[ModelSymbol, TrainedModel], indicator_data: Mapping[IndicatorSymbol, Series], test_data: DataFrame, portfolio: Portfolio, exit_dates: Mapping[str, datetime64], backtest_settings: BacktestSettings = BacktestSettings(max_long_positions=None, max_short_positions=None, worst_rank_held=None), rotation_sizer: Callable[[RotationContext], None] | None = None, train_only: bool = False, slippage_model: SlippageModel | None = None, enable_fractional_shares: bool = False, round_fill_price: bool = True, warmup: int | None = None, interval_data: IntervalData = IntervalData(compressed={}), history_col_scope: ColumnScope | None = None, test_col_scope: ColumnScope | None = None, run_hyperparams: dict[str, Any] | None = None, pending_order_scope: PendingOrderScope | None = None, master_col_scope: ColumnScope | None = None) → dict[str, DataFrame][源代码]

对实现交易逻辑的一组 Execution（set）进行回测。

参数:
返回:

当 pybroker.config.StrategyConfig.return_signals 为 True 时，返回一个字典，其中包含每个品种的 K 线数据、指标数据和模型预测结果的 pandas.DataFrame。信号仅包含基础时间框架的值；仅当绑定中包含 'base' 时，绑定到某个区间的指标或模型才会出现在信号中。

class BacktestSettings(max_long_positions: int | None = None, max_short_positions: int | None = None, worst_rank_held: int | None = None)[源代码]

基类：object

class Execution(id: int, symbols: frozenset[str] | Callable[[DataFrame], Sequence[str]], fn: Callable[[ExecContext], None] | None, model_names: frozenset[str], indicator_names: frozenset[str], intervals: frozenset[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str] = frozenset({}), hyperparam_names: frozenset[str] = frozenset({}), args: tuple[Any, ...] = (), kwargs: tuple[tuple[str, Any], ...] = ())[源代码]

基类：NamedTuple

表示 Strategy 的一次执行。持有一个实现交易逻辑的 Callable 的引用。

id

唯一 ID。

类型:

int

symbols

用于执行 fn 的股票代码。

类型:

frozenset[str] | Callable[[pandas.DataFrame], Sequence[str]]

fn

实现交易逻辑。

类型:

Callable[[pybroker.context.ExecContext], None] | None

model_names

用于执行 fn 的 pybroker.model.ModelSource名称，包括绑定到某个区间的模型的带后缀区间名称。

类型:

frozenset[str]

indicator_names

用于执行 fn 的 pybroker.indicator.Indicator名称，包括绑定到某个区间的指标的带后缀区间名称。

类型:

frozenset[str]

intervals

fn 可通过 pybroker.context.ExecContext.interval() 使用的压缩区间：即在 pybroker.strategy.Strategy.add_execution() 上声明的 intervals，与绑定到该执行函数的模型和指标的区间的并集。

类型:

frozenset[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str]

args

fn 的额外位置参数。

类型:

tuple[Any, ...]

kwargs

fn 的额外关键字参数。

类型:

tuple[tuple[str, Any], ...]

class Strategy(data_source: DataSource | DataFrame, start_date: str | datetime, end_date: str | datetime, config: StrategyConfig | None = None)[源代码]

基类：BacktestMixin, EvaluateMixin, IndicatorsMixin, ModelsMixin, WalkforwardMixin, OptimizeMixin

表示要回测的交易策略的类。

参数:
add_execution(fn: ~typing.Callable[[~typing.Concatenate[~pybroker.context.ExecContext, ~P]], None] | None, symbols: str | ~typing.Iterable[str] | ~typing.Callable[[~pandas.DataFrame], ~typing.Sequence[str]], models: ~pybroker.model.ModelSource | ~pybroker.model.IntervalBoundModel | ~typing.Iterable[~pybroker.model.ModelSource | ~pybroker.model.IntervalBoundModel] | None = None, indicators: ~pybroker.indicator.Indicator | ~pybroker.indicator.IntervalBoundIndicator | ~typing.Iterable[~pybroker.indicator.Indicator | ~pybroker.indicator.IntervalBoundIndicator] | None = None, hyperparams: ~typing.Iterable[~pybroker.optimize.Hyperparam] | None = None, intervals: int | ~typing.Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | ~typing.Iterable[int | ~typing.Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str] | None = None, *args: ~typing.~P, **kwargs: ~typing.~P)[源代码]

添加一个要回测的执行函数。

传递给 intervals 的 TimeframeInterval 为以下之一：

  • 每 n 根 K 线 （int）：将每 n 根基础 K 线压缩为一根 K 线，其中 n > 1。在 1 分钟数据上， 5 会生成 5 根 K 线一组的分箱（约为 5 分钟 K 线）。

  • 时长 （str）：以数字加单个单位字母表示的固定时间跨度 —— "1m"、 "5m"、 "1h"、 "30s" 或 "1d"。

  • 日历 （str）：日历分桶 —— "daily"、 "weekly"、 "monthly"、 "quarterly" 或 "yearly"。周从周一开始，月从每月 1 日开始，季度从 1、4、7、10 月开始，年从 1 月 1 日开始。

例如，要在 1 分钟数据源上读取 5 根 K 线分箱和 1 小时时长 K 线，并额外在周线 K 线上计算 sma:

strategy.add_execution(
    fn,
    "SPY",
    indicators=sma.intervals("weekly"),
    intervals=[5, "1h"],
)
strategy.walkforward(windows=1, timeframe="1m")
参数:

  • fn -- 在回测期间的每根数据 K 线上被调用，并为 symbols 中的每个股票代码传入一个 pybroker.context.ExecContext 的 Callable。

  • symbols -- 用于运行 fn 的股票代码， fn 会为每个品种单独调用。也可以是 pybroker.common.SymbolSelector —— 一个 (df) -> Sequence[str] 形式的 Callable，为每个向前分析窗口选择一次要交易的品种，因此品种池会在回测过程中变化。它接收该窗口的 训练 数据，绝不会接收测试数据，因此需要一个训练窗口：backtest() 与 train_size=0 会引发 ValueError。候选品种池必须以 pandas.DataFrame 形式提供，而非 pybroker.data.DataSource，因为在窗口划分之前，待查询的品种是未知的。某个品种在后续窗口中被丢弃时，其仓位会在该窗口的第一根 K 线上平仓；如果该品种已没有剩余 K 线，则会在其最后一根 K 线上平仓。请注意， shuffle=True 会打乱训练数据帧的行顺序，因此应避免将其与依赖 K 线顺序的选择器搭配使用。

  • models -- 要为回测训练/加载的 pybroker.model.ModelSource的 Iterable。直接传入的模型会在基础时间框架上训练。使用 pybroker.model.ModelSource.intervals() 将模型绑定到压缩区间后，该模型会改为在所列区间的压缩 K 线上训练，连同注册给它的指标一起；在绑定中包含字面量 'base' 即可同时在基础时间框架上训练。逐区间的预测结果通过 pybroker.context.IntervalContext.preds() 读取。绑定到某个区间的模型会以该区间的单位遵循传递给 backtest()、walkforward() 或 optimize() 的 lookahead：其训练行和测试行之间会保留 lookahead 根压缩 K 线。

  • indicators -- 要为回测计算的 pybroker.indicator.Indicator的 Iterable。直接传入的指标会在基础时间框架上计算。使用 pybroker.indicator.Indicator.intervals() 将指标绑定到压缩区间后，该指标会改为在所列区间的压缩 K 线上计算，通过 pybroker.context.IntervalContext.indicator() 读取；在绑定中包含字面量 'base' 即可同时在基础时间框架上计算。

  • hyperparams -- fn 可通过 pybroker.context.ExecContext.hyperparam() 读取的 pybroker.optimize.Hyperparam的 Iterable。

  • intervals -- 一个或多个压缩区间，其 K 线会通过 pybroker.context.ExecContext.interval() 提供给 fn。在此处声明一个区间仅提供压缩 K 线，不会 在该区间上计算指标或训练模型。请改为通过 pybroker.indicator.Indicator.intervals() 和 pybroker.model.ModelSource.intervals() 按来源分别绑定；已绑定的区间会自动通过 interval() 提供，无需在此处重复声明。每个区间都必须严格粗于传递给 backtest() 或 walkforward() 的 timeframe 所指定的基础 K 线间隔；无效的组合会在回测运行时引发 ValueError。区间的作用域限定在本执行函数内，因此另一个执行函数的 pybroker.context.ExecContext 无法读取它们 —— 即使在 set_before_exec() 或 set_after_exec() 回调中也是如此，尽管该回调会接收所有执行函数的上下文。

  • args -- 传递给 fn 的位置参数。

  • kwargs -- 传递给 fn 的关键字参数。

backtest(start_date: str | datetime | None = None, end_date: str | datetime | None = None, timeframe: str = '', between_time: tuple[str, str] | None = None, days: str | Day | Iterable[str | Day] | None = None, lookahead: int = 1, train_size: float = 0, shuffle: bool = False, calc_bootstrap: bool = False, parallel_indicators: bool = False, parallel_models: bool = False, warmup: int | None = None, portfolio: Portfolio | None = None, adjust: Any | None = None, seed: int | None = 42, params: dict[str, Any] | None = None) → TestResult[源代码]

通过运行使用 add_execution() 添加的执行函数，对交易策略进行回测。

参数:

  • start_date -- 回测的起始日期（含）。必须在传递给 Strategy 的 start_date 和 end_date 范围之内。

  • end_date -- 回测的结束日期（含）。必须在传递给 Strategy 的 start_date 和 end_date 范围之内。

  • timeframe -- 指定回测数据时间框架分辨率的格式化字符串。该时间框架字符串支持以下单位： - "s"/"sec"：秒 - "m"/"min"：分钟 - "h"/"hour"：小时 - "d"/"day"：天 - "w"/"week"：周 时间框架字符串示例： 1h 30m。当任何执行函数声明了 intervals 时为必需，因为它定义了压缩区间据以验证和对齐的基础 K 线间隔。

  • between_time -- 用于过滤回测数据的时间段（含边界），为 tuple[str, str]，例如 ('9:30', '16:00')。

  • days -- 用于过滤回测数据的星期（例如 "mon"、 "tues" 等）。

  • lookahead -- 预测目标未来的 K 线数量。例如，预测下一根 K 线的收益率对应的 lookahead 为 1。需要这个量来防止训练数据泄漏到测试边界。它以每个模型所拟合的时间框架的 K 线为单位：通过 pybroker.model.ModelSource.intervals() 绑定到某个区间的模型，会保留该区间的 lookahead 根 K 线，而不是基础时间框架的 K 线。某个窗口测试集中的任何 K 线都绝不会被用于拟合该窗口的模型。

  • train_size -- 用于训练的 pybroker.data.DataSource 数据量，其中 train_size 的最大值为 1。例如， train_size 为 0.9 表示 90% 的数据用于训练，剩余 10% 的数据用于测试。

  • shuffle -- 是否随机打乱用于训练的数据。默认为 False。当通过 pybroker.cache.enable_model_cache() 启用模型缓存时，该选项会被禁用。

  • calc_bootstrap -- 是否计算随机自助法评估指标。默认为 False。

  • parallel_indicators -- 如果为 True，pybroker.indicator.Indicator 数据会使用多个进程并行计算。默认为 False。

  • parallel_models -- 如果为 True，pybroker.model.ModelTrainer 模型会使用多个进程并行训练。默认为 False。

  • warmup -- 运行执行函数之前需要经过的 K 线数量。

  • portfolio -- 用于回测的自定义 pybroker.portfolio.Portfolio。

  • adjust -- 对 pybroker.data.DataSource 应用的复权调整类型。

  • seed -- 用于结果可复现的随机种子。默认为 42。

返回:

包含投资组合余额、订单历史和评估指标的 TestResult。

clear_executions()[源代码]

清除通过 add_execution() 添加的执行函数。

enable_rotation(worst_rank_held: int | Hyperparam | None, sizer: Callable[[RotationContext], None] | None = None) → None[源代码]

启用轮动持仓带逻辑，以及可选的自定义规模计算。

每根 K 线，排名劣于 worst_rank_held 的持仓会被平仓，排名靠前的品种会入场以填补剩余的仓位名额。若未提供 sizer，入场会在 set_max_long_positions() 加 set_max_short_positions() 的名额上等权重分配。

轮动是排他性的：交易完全由 pybroker.context.ExecContext.long_score 和 pybroker.context.ExecContext.short_score 驱动，Execution 下达的订单会被忽略。执行函数中设置的成交价格和止损会被保留，并应用于轮动所下达的订单。

排名覆盖整个投资组合，因此即使某个持仓是由另一个 Execution 开立的，只要它没有可排序的得分，也会被平仓。

参数:

  • worst_rank_held -- 持仓被保留所允许的最差得分排名，可以是一个可搜索的 pybroker.optimize.Hyperparam，或 None 以禁用轮动。必须大于或等于多头和空头仓位数量上限。

  • sizer -- 可选的 Callable，接收一个 pybroker.context.RotationContext，用于在轮动决策做出之后覆盖等权重的入场规模计算。不要覆盖轮动设置的卖出或回补信号。

set_after_exec(fn: Callable[[Mapping[str, ExecContext]], None] | None)[源代码]

在所有执行函数之后运行的 Callable[[Mapping[str, ExecContext]], None]。

参数:

fn -- 接收一个将所有股票代码映射到 ExecContext的 Mapping 的 Callable。

set_before_exec(fn: Callable[[Mapping[str, ExecContext]], None] | None)[源代码]

在所有执行函数之前运行的 Callable[[Mapping[str, ExecContext]], None]。

参数:

fn -- 接收一个将所有股票代码映射到 ExecContext的 Mapping 的 Callable。

set_max_long_positions(max_long: int | Hyperparam | None) → None[源代码]

设置任一时刻持有的最大多头仓位数量。

参数:

max_long -- 最大多头仓位数量，可以是一个可搜索的 pybroker.optimize.Hyperparam，或 None 表示不限。

set_max_short_positions(max_short: int | Hyperparam | None) → None[源代码]

设置任一时刻持有的最大空头仓位数量。

参数:

max_short -- 最大空头仓位数量，可以是一个可搜索的 pybroker.optimize.Hyperparam，或 None 表示不限。

set_slippage_model(slippage_model: SlippageModel | None)[源代码]

设置 pybroker.slippage.SlippageModel。

内置模型包括 pybroker.slippage.FixedSlippageModel （固定基点）、pybroker.slippage.VolatilitySlippageModel （基于 ATR 缩放）和 pybroker.slippage.VolumeSlippageModel （参与度上限与平方律价格冲击）。传入 None 可禁用滑点。

成交时滑点适用于已排定的订单、止损离场和头寸离场。止损和头寸离场仅使用调整后的成交价格；这些路径上的股数调整会被忽略，因为它们会全部平掉一笔入场。

walkforward(windows: int, 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: str | Day | Iterable[str | Day] | None = None, train_size: float = 0.5, shuffle: bool = False, calc_bootstrap: bool = False, parallel_indicators: bool = False, parallel_models: bool = False, warmup: int | None = None, portfolio: Portfolio | None = None, adjust: Any | None = None, seed: int | None = 42, params: dict[str, Any] | None = None) → TestResult[源代码]

使用 向前分析（Walkforward Analysis） 对交易策略进行回测。由 pybroker.data.DataSource 提供的回测数据会被划分为 windows 个大小相等的时间窗口，每个窗口再按照 train_size 指定的比例划分为训练数据和测试数据。回测会在时间上依次“向前”经过每个窗口，运行通过 add_execution() 添加的执行函数。

参数:

  • windows -- 向前分析的时间窗口数量。

  • start_date -- 向前分析的起始日期（含）。必须在传递给 Strategy 的 start_date 和 end_date 范围之内。

  • end_date -- 向前分析的结束日期（含）。必须在传递给 Strategy 的 start_date 和 end_date 范围之内。

  • timeframe -- 指定回测数据时间框架分辨率的格式化字符串。该时间框架字符串支持以下单位： - "s"/"sec"：秒 - "m"/"min"：分钟 - "h"/"hour"：小时 - "d"/"day"：天 - "w"/"week"：周 时间框架字符串示例： 1h 30m。当任何执行函数声明了 intervals 时为必需，因为它定义了压缩区间据以验证和对齐的基础 K 线间隔。

  • between_time -- 用于过滤回测数据的时间段（含边界），为 tuple[str, str]，例如 ('9:30', '16:00')。

  • days -- 用于过滤回测数据的星期（例如 "mon"、 "tues" 等）。

  • lookahead -- 预测目标未来的 K 线数量。例如，预测下一根 K 线的收益率对应的 lookahead 为 1。需要这个量来防止训练数据泄漏到测试边界。它以每个模型所拟合的时间框架的 K 线为单位：通过 pybroker.model.ModelSource.intervals() 绑定到某个区间的模型，会保留该区间的 lookahead 根 K 线，而不是基础时间框架的 K 线。某个窗口测试集中的任何 K 线都绝不会被用于拟合该窗口的模型。

  • train_size -- 用于训练的 pybroker.data.DataSource 数据量，其中 train_size 的最大值为 1。例如， train_size 为 0.9 表示 90% 的数据用于训练，剩余 10% 的数据用于测试。

  • shuffle -- 是否随机打乱用于训练的数据。默认为 False。当通过 pybroker.cache.enable_model_cache() 启用模型缓存时，该选项会被禁用。

  • calc_bootstrap -- 是否计算随机自助法评估指标。默认为 False。

  • parallel_indicators -- 如果为 True，pybroker.indicator.Indicator 数据会使用多个进程并行计算。默认为 False。

  • parallel_models -- 如果为 True，pybroker.model.ModelTrainer 模型会使用多个进程并行训练。默认为 False。

  • warmup -- 运行执行函数之前需要经过的 K 线数量。

  • portfolio -- 用于回测的自定义 pybroker.portfolio.Portfolio。

  • adjust -- 对 pybroker.data.DataSource 应用的复权调整类型。

  • seed -- 用于结果可复现的随机种子。默认为 42。

返回:

包含投资组合余额、订单历史和评估指标的 TestResult。

class TestResult(start_date: datetime, end_date: datetime, portfolio: DataFrame, positions: DataFrame, orders: DataFrame, trades: DataFrame, metrics: EvalMetrics, metrics_df: DataFrame, bootstrap: BootstrapResult | None, signals: dict[str, DataFrame] | None, stops: DataFrame | None, symbols: frozenset[str] = frozenset({}))[源代码]

基类：object

包含 Strategy 的回测结果。

start_date

回测的起始日期。

类型:

datetime.datetime

end_date

回测的结束日期。

类型:

datetime.datetime

portfolio

每根 K 线的 pybroker.portfolio.Portfolio 余额 pandas.DataFrame。

类型:

pandas.DataFrame

positions

每根 K 线的 pybroker.portfolio.Position 余额 pandas.DataFrame。

类型:

pandas.DataFrame

orders

所有已下达订单的 pandas.DataFrame。

类型:

pandas.DataFrame

trades

所有已成交交易的 pandas.DataFrame。

类型:

pandas.DataFrame

metrics

评估指标。

类型:

pybroker.eval.EvalMetrics

metrics_df

评估指标的 pandas.DataFrame。

类型:

pandas.DataFrame

bootstrap

随机自助法评估指标。

类型:

pybroker.eval.BootstrapResult | None

signals

当 pybroker.config.StrategyConfig.return_signals 为 True 时，返回一个字典，其中包含每个品种的 K 线数据、指标数据和模型预测结果的 pandas.DataFrame。信号仅包含基础时间框架的值；仅当绑定中包含 'base' 时，绑定到某个区间的指标或模型才会出现在信号中。

类型:

dict[str, pandas.DataFrame] | None

stops

当 pybroker.config.StrategyConfig.return_stops 为 True 时，包含按 K 线记录的止损数据的 pandas.DataFrame。

类型:

pandas.DataFrame | None

symbols

已回测的股票代码。

类型:

frozenset[str]

to_json(*, include: frozenset[str] = frozenset({'bootstrap', 'metrics', 'orders', 'trades'}), max_rows: int | None = 100, symbols: frozenset[str] | None = None) → dict[str, Any][源代码]

返回可 JSON 序列化的回测结果。

默认包含 start_date、 end_date、 metrics、 trades、 orders 以及 bootstrap （如果存在）。诸如 portfolio、 positions、 signals 和 stops 等大型时间序列需要通过 include 主动选择才会包含。日期以不带时区的 UTC 序列化，NaN 序列化为 null，无穷大的指标值序列化为字符串哨兵值 "Infinity"/"-Infinity"。

参数:

  • include -- 要包含的可选结果部分的名称。有效名称为 metrics、 metrics_df、 trades、 orders、 portfolio、 positions、 bootstrap、 signals 和 stops。请注意，只有当 pybroker.config.StrategyConfig.record_position_bars 为 True 时 positions 才会有行， signals/stops 也仅在各自对应的配置标志开启时才会有内容。

  • max_rows -- 每个表格部分的最大行数。 None 表示不限。

  • symbols -- 设置后，会将各品种专属的部分过滤为这些股票代码。必须是 symbols 的一个非空子集。

to_json_str(*, include: frozenset[str] = frozenset({'bootstrap', 'metrics', 'orders', 'trades'}), max_rows: int | None = 100, symbols: frozenset[str] | None = None) → str[源代码]

从 to_json() 返回严格的 JSON 文本。

class WalkforwardMixin[源代码]

基类：object

实现 向前分析（Walkforward Analysis） 逻辑的 Mixin。

walkforward_split(df: DataFrame, windows: int, lookahead: int, train_size: float = 0.9, shuffle: bool = False) → Iterator[WalkforwardWindow][源代码]

将包含多个股票代码数据的 pandas.DataFrame，为 向前分析（Walkforward Analysis） 拆分为训练/测试时间窗口的 Iterator。

参数:

  • df -- 要为向前分析拆分为训练/测试窗口的数据 pandas.DataFrame。

  • windows -- 向前分析的时间窗口数量。

  • lookahead -- 预测目标未来的 K 线数量。例如，预测下一根 K 线的收益率对应的 lookahead 为 1。需要这个量来防止训练数据泄漏到测试边界。它以每个模型所拟合的时间框架的 K 线为单位：通过 pybroker.model.ModelSource.intervals() 绑定到某个区间的模型，会保留该区间的 lookahead 根 K 线，而不是基础时间框架的 K 线。某个窗口测试集中的任何 K 线都绝不会被用于拟合该窗口的模型。

  • train_size -- df 中用于训练的数据量，其中 train_size 的最大值为 1。例如， train_size 为 0.9 表示 df 中 90% 的数据用于训练，剩余 10% 的数据用于测试。

  • shuffle -- 是否随机打乱用于训练的数据。默认为 False。

返回:

包含训练和测试数据的 WalkforwardWindow的 Iterator。

class WalkforwardWindow(train_data: NDArray[int64], test_data: NDArray[int64])[源代码]

基类：NamedTuple

包含某个向前分析窗口的训练/测试行索引。

train_data

指向主数据帧中用于训练的整数行索引。

类型:

numpy._typing._array_like.NDArray[numpy.int64]

test_data

指向主数据帧中用于测试的整数行索引。

类型:

numpy._typing._array_like.NDArray[numpy.int64]