pybroker.scope 模块

包含用于存储执行 pybroker.strategy.Strategy 所需数据和对象引用的作用域。

class ColumnScope(store: SymbolArrayStore | DataFrame)

基类：object

从 SymbolArrayStore 中缓存并获取列数据。

参数:

store -- 预先构建好的 numpy 列存储，或 MultiIndex pandas.DataFrame （为兼容旧版而保留）。

bar_data_from_data_columns(symbol: str, end_index: int) → BarData

返回一个新的 pybroker.common.BarData 实例，其中包含通过 StaticScope 注册的默认和自定义数据列的列数据。

参数:

• symbol -- 要查询的股票代码。

• end_index -- 截断列值（不含该索引）。如果为 None，则不截断列值。

fetch(symbol: str, name: str, end_index: int | None = None) → NDArray | None

获取 symbol 的列数据 numpy.ndarray。

参数:

• symbol -- 要查询的股票代码。

• name -- 要查询的列名称。

• end_index -- 截断列值（不含该索引）。如果为 None，则不截断列值。

返回:

截至 end_index （如指定）为止每根 K 线的列数据 numpy.ndarray。

fetch_dict(symbol: str, names: Iterable[str], end_index: int | None = None) → dict[str, NDArray | None]

获取 symbol 的列数据 dict。

参数:

• symbol -- 要查询的股票代码。

• names -- 要查询的列名称。

• end_index -- 截断列值（不含该索引）。如果为 None，则不截断列值。

返回:

将列名称映射到列值 numpy.ndarray的 dict。

fetch_value(symbol: str, name: str, end_index: int) → float | None

返回 end_index - 1 处的标量值，不进行切片。

property store: SymbolArrayStore

property symbols: frozenset[str]

底层存储中持有的品种。

unique_dates() → NDArray[datetime64]

返回该存储中所有品种去重排序后的日期。

class IndicatorScope(indicator_data: Mapping[IndicatorSymbol, Series], filter_dates: Sequence[datetime64])

基类：object

缓存并获取 pybroker.indicator.Indicator 数据。

参数:

• indicator_data -- 将 pybroker.common.IndicatorSymbol 组合映射到 pybroker.indicator.Indicator 值 pandas.Series 的 Mapping。

• filter_dates -- 根据日期 Sequence 过滤 pybroker.indicator.Indicator 数据。

fetch(symbol: str, name: str, end_index: int | None = None) → NDArray[float64]

获取 pybroker.indicator.Indicator 数据。

参数:

• symbol -- 要查询的股票代码。

• name -- 要查询的 pybroker.indicator.Indicator 名称。

• end_index -- 截断返回的 pybroker.indicator.Indicator 数据数组（不含该索引）。如果为 None，则不截断指标数据。

返回:

截至 end_index （如指定）为止每根 K 线的 pybroker.indicator.Indicator 数据 numpy.ndarray。

fetch_full(symbol: str, name: str) → NDArray[float64]

获取未经截断的完整指标数组。

fetch_history(symbol: str, name: str, dates: NDArray[Any]) → NDArray[float64] | None

将完整历史指标值对齐到 dates。

fetch() 会将基础时间框架的指标遮罩到 filter_dates，因此无法提供当前窗口之前的数据。滞后特征需要能够回溯到训练窗口的历史数据，本方法正是从未过滤的序列中读取这些数据。

返回:

对齐到 dates 的值 numpy.ndarray；当 symbol 未注册该指标时为 None。

fetch_value(symbol: str, name: str, end_index: int) → float

返回 end_index - 1 处的标量值，不进行切片。

has_indicator(symbol: str, name: str) → bool

symbol 是否已注册 pybroker.indicator.Indicator 数据。

class IntervalScope(interval_data: IntervalData, ind_scope: IndicatorScope, models: Mapping[ModelSymbol, TrainedModel] | None = None, test_dates: Sequence[datetime64] | None = None)

基类：object

通过对齐映射提供压缩 K 线和指标数据。

clear_cache()

丢弃所有已缓存的数组。

压缩数据在一个作用域的生命周期内是不可变的（每个向前分析窗口都会构建一个新的作用域），且每个缓存的键都与当前 K 线无关，因此本方法仅用于销毁作用域 —— 若按每根 K 线调用，会导致每根 K 线都重新构建模型输入并重新运行 predict。

completed_index(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, end_index: int) → int

fetch_bar(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, col: str, end_index: int) → NDArray[Any]

fetch_indicator(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_name: str, end_index: int) → NDArray[float64]

fetch_input(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_model_name: str, end_index: int) → DataFrame

fetch_preds(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_model_name: str, end_index: int) → NDArray

window_len(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → int

返回当前窗口中可见的压缩 K 线数量。

completed 会由 pybroker.interval.IntervalData.slice_for_test() 重新对齐到向前分析测试窗口，因此其最后一个条目是在该窗口内完成的最新压缩 K 线。模型输入和预测在此处被限定，以确保用户回调永远不会看到属于未来窗口的压缩 K 线。

class ModelInputScope(col_scope: ColumnScope, ind_scope: IndicatorScope, models: Mapping[ModelSymbol, TrainedModel], history_col_scope: ColumnScope | None = None, test_dates: Sequence[datetime64] | None = None)

基类：object

缓存并获取模型输入数据。

参数:

• col_scope -- ColumnScope。

• ind_scope -- IndicatorScope。

• models -- 将 pybroker.common.ModelSymbol 组合映射到 pybroker.common.TrainedModel的 Mapping。

fetch(symbol: str, name: str, end_index: int | None = None) → DataFrame

获取模型输入数据。

参数:

• symbol -- 要查询的股票代码。

• name -- 要查询输入数据的 pybroker.model.ModelSource 名称。

• end_index -- 截断返回的模型输入数据数组（不含该索引）。如果为 None，则不截断模型输入数据。

返回:

截至 end_index （如指定）为止每根 K 线的模型输入数据 pandas.DataFrame。

fetch_model_input(symbol: str, name: str, end_index: int | None = None) → ModelInput

以内部 pybroker.model.ModelInput 形式获取模型输入（不使用 DataFrame）。

参数:

• symbol -- 要查询的股票代码。

• name -- 要查询输入数据的 pybroker.model.ModelSource 名称。

• end_index -- 截断返回的模型输入数据数组（不含该索引）。如果为 None，则不截断模型输入数据。

返回:

截至 end_index （如指定）为止每根 K 线的 pybroker.model.ModelInput。

class PendingOrder(id: int, type: Literal['buy', 'sell'], symbol: str, created: np.datetime64, exec_date: np.datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | np.floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal], exec_bar: int, timeout_bars: int | None, stops: frozenset['Stop'] | None, exit_pos_type: Literal['long', 'short'] | None = None)

基类：NamedTuple

保存一笔待处理订单的数据。

id

唯一 ID。

类型:

int

type

订单类型，为 buy 或 sell。

类型:

Literal['buy', 'sell']

symbol

该订单的股票代码。

类型:

str

created

订单创建的日期。

类型:

np.datetime64

exec_date

订单将被执行的日期。

类型:

np.datetime64

shares

待买入或卖出的股数。

类型:

Decimal

limit_price

该订单使用的限价。

类型:

Optional[Decimal]

fill_price

该订单将以之成交的价格。

类型:

Union[int, float, np.floating, Decimal, PriceType, Callable[[str, BarData], Union[int, float, Decimal]]]

exec_bar

订单首次尝试成交时对应的品种 K 线索引。

类型:

int

timeout_bars

首次尝试之后重试的 K 线数量。 None 表示仅尝试一次， -1 表示无限期保留，正整数表示有限的重试 K 线数量。

类型:

Optional[int]

stops

订单成交时要附加的止损。

类型:

Optional[frozenset['Stop']]

exit_pos_type

该订单所平掉的 pybroker.portfolio.Position 类型，为 long 或 short；当该订单不是离场订单时为 None。离场订单在成交时会被限定在仍持有的股数以内，因此只能平仓，永远不会将仓位翻转到相反方向。

类型:

Optional[Literal['long', 'short']]

class PendingOrderScope

基类：object

存储 PendingOrder

add(type: Literal['buy', 'sell'], symbol: str, created: np.datetime64, exec_date: np.datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | np.floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal], exec_bar: int, timeout_bars: int | None, stops: frozenset['Stop'] | None = None, exit_pos_type: Literal['long', 'short'] | None = None) → int

创建一个 PendingOrder。

参数:

• type -- 订单类型，为 buy 或 sell。

• symbol -- 该订单的股票代码。

• created -- 订单创建的日期。

• exec_date -- 订单将被执行的日期。

• shares -- 待买入或卖出的股数。

• limit_price -- 该订单使用的限价。

• fill_price -- 该订单将以之成交的价格。

• exec_bar -- 订单首次尝试成交时对应的品种 K 线索引。

• timeout_bars -- 首次尝试之后重试的 K 线数量。

• stops -- 订单成交时要附加的止损。

• exit_pos_type -- 该订单所平掉的仓位类型；当该订单不是离场订单时为 None。

返回:

PendingOrder 的 ID。

advance_retry_bars(order_id: int) → None

记录 order_id 已在某根 K 线上尝试过成交。

contains(order_id: int) → bool

返回是否存在 ID 为 order_id 的 PendingOrder。

get(order_id: int) → PendingOrder | None

返回 ID 为 order_id 的 PendingOrder。

has_orders() → bool

返回是否存在任何待处理订单。

mark_attempted(order_id: int) → None

记录 order_id 已经过首次成交尝试。

orders(symbol: str | None = None, order_id: int | None = None) → Iterable[PendingOrder]

返回 PendingOrder的 Iterable。

参数:

• symbol -- 按股票代码过滤。

• order_id -- 按订单 ID 过滤。

remove(order_id: int) → bool

移除 ID 为 order_id 的 PendingOrder。

remove_all(symbol: str | None = None)

移除所有 PendingOrder。

retry_bars(order_id: int) → int

返回 order_id 已被重试的 K 线数量。

在其首次尝试的那根 K 线上为 0。

was_attempted(order_id: int) → bool

返回 order_id 是否已经过首次成交尝试。

class PredictionScope(models: Mapping[ModelSymbol, TrainedModel], input_scope: ModelInputScope)

基类：object

缓存并获取模型预测结果。

参数:

• models -- 将 pybroker.common.ModelSymbol 组合映射到 pybroker.common.TrainedModel的 Mapping。

• input_scope -- ModelInputScope。

fetch(symbol: str, name: str, end_index: int | None = None) → NDArray

获取模型预测结果。

参数:

• symbol -- 要查询的股票代码。

• name -- 做出该预测的 pybroker.model.ModelSource 名称。

• end_index -- 截断返回的预测结果数组（不含该索引）。如果为 None，则不截断预测结果。

返回:

截至 end_index （如指定）为止每根 K 线的模型预测结果 numpy.ndarray。

class PriceScope(col_scope: ColumnScope, sym_end_index: Mapping[str, int], round_fill_price: bool)

基类：object

获取最新价格。

fetch(symbol: str, price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal]) → Decimal

fetch_bar_ohlc(symbol: str, date: datetime64) → tuple[float | None, float | None, float | None]

返回 symbol 在 date 上的 (close, low, high)，若不可用则为 None。

按 K 线记忆化： check_stops 循环和 capture_bar 都会在每根 K 线为每个品种读取该值，每次未命中都会重新获取列字典。与 has_bar_on() 一样，键同时包含日期和品种，因此过期的条目不会为之后的 K 线给出错误答案。

fetch_float(symbol: str, price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal]) → float

尽可能使用按 K 线缓存，以 float 形式返回某根 K 线的价格。

has_bar(symbol: str) → bool

返回 symbol 是否存在可用于定价的 K 线。

对于当前测试窗口中不存在的品种 —— 例如已停止交易，或被 pybroker.common.SymbolSelector 剔除的品种 —— 为 False；否则获取其价格会引发异常。

has_bar_on(symbol: str, date: datetime64) → bool

返回 symbol 的当前 K 线是否落在 date 上。

比 has_bar() 更严格，后者只报告该品种在某个时间点是否有过交易。当各品种交易日历参差不齐时，某品种在没有 K 线的日期上其索引不会前进，因此其“当前” K 线实际上是更早的一根，按其定价会使用过期的价格。

按 K 线记忆化： check_stops 会在每根 K 线为每个持有止损的品种调用一次本方法，每次未命中都会获取该品种的完整日期数组。键同时包含日期和品种，因此即使调用方未调用 reset_bar()，仍能读到正确的结果。

reset_bar() → None

清除按 K 线缓存的 OHLC 数据。应在每根 K 线开始时调用一次。

class StaticScope

基类：object

数据和对象引用的静态注册表。

logger

pybroker.log.Logger

data_source_cache

存储从 pybroker.data.DataSource 获取的数据的 diskcache.Cache。

data_source_cache_ns

为 data_source_cache 设置的命名空间。

indicator_cache

存储 pybroker.indicator.Indicator 数据的 diskcache.Cache。

indicator_cache_ns

为 indicator_cache 设置的命名空间。

model_cache

存储已训练模型的 diskcache.Cache。

model_cache_ns

为 model_cache 设置的命名空间。

default_data_cols

从 pybroker.data.DataSource 获取的 pandas.DataFrame 中的默认数据列。

custom_data_cols

从 pybroker.data.DataSource 获取的 pandas.DataFrame 中的用户自定义数据列。

property all_data_cols: frozenset[str]

所有已注册的数据列名称。无序；当迭代顺序有意义时，请使用 ordered_data_cols。

clear_params()

清除所有全局参数。

freeze_data_cols()

阻止注册额外的数据列。

get_hyperparam(name: str) → Any

从静态作用域中获取一个超参数。

get_indicator(name: str)

从静态作用域中获取一个 pybroker.indicator.Indicator。

get_indicator_names(model_name: str) → tuple[str]

返回一个 tuple[str]，包含所有注册给名为 model_name 的 pybroker.model.ModelSource 的 pybroker.indicator.Indicator 名称。

get_model_source(name: str)

从静态作用域中获取一个 pybroker.model.ModelSource。

has_hyperparam(name: str) → bool

静态作用域中是否存储了某个超参数。

has_indicator(name: str) → bool

静态作用域中是否存储了 pybroker.indicator.Indicator。

has_model_source(name: str) → bool

静态作用域中是否存储了 pybroker.model.ModelSource。

classmethod instance() → StaticScope

返回单例实例。

iter_hyperparams() → Iterable[Any]

遍历已注册的超参数。

property ordered_data_cols: tuple[str, ...]

所有已注册数据列名称的确定性顺序列表。若改为迭代 all_data_cols，得到的顺序会依赖于具体进程，这会使对列顺序敏感的输出（例如模型输入数据）在不同运行之间无法复现。

param(name: str, value: Any | None = <object object>) → Any | None

获取或设置一个全局参数。

register_custom_cols(names: str | Iterable[str], *args)

注册用户自定义列名称。

set_hyperparam(hyperparam: Any) → None

将一个 pybroker.optimize.Hyperparam 存储到静态作用域中。

set_indicator(indicator)

将 pybroker.indicator.Indicator 存储到静态作用域中。

classmethod set_instance(scope: StaticScope | None) → None

替换单例实例；当 scope 为 None 时则清除它。

用于安装一个从另一个进程 pickle 而来的作用域，使工作任务能够看到调用方已注册的指标、模型源和参数，而不是一个空的作用域。整体替换（而非合并）还能避免过期的注册在跨多次运行复用的工作进程中残留。

set_model_source(source)

将 pybroker.model.ModelSource 存储到静态作用域中。

unfreeze_data_cols()

如果之前调用过 pybroker.scope.StaticScope.freeze_data_cols()，则重新允许注册额外的数据列。

unregister_custom_cols(names: str | Iterable[str], *args)

取消注册用户自定义列名称。

validate_registered_names(indicators: Iterable[str] | None = None, models: Iterable[str] | None = None)

当某次运行所使用的指标，或其某个模型的预测列，与某个数据列或另一个已注册的来源同名时，引发异常。

同一个冲突的名称会被不同的消费者以不同方式解析：模型训练会读取数据列，而预测会读取指标，信号输出则会用其中一个值悄然覆盖另一个值 —— 因此这种冲突会被直接拒绝。

参数:

• indicators -- 本次运行所使用的指标名称。默认为所有已注册的指标。

• models -- 本次运行所使用的模型名称。默认为所有已注册的模型。

class SymbolArrayStore(symbols: frozenset[str], sym_arrays: Mapping[str, Mapping[str, NDArray]], backing: _StoreBacking | None = None)

基类：object

以品种为键的内部 numpy 支持的 OHLCV/自定义列。

backing: _StoreBacking | None = None

sym_arrays: Mapping[str, Mapping[str, NDArray]]

symbols: frozenset[str]

unique_dates() → NDArray[datetime64]

返回所有品种去重排序后的日期。

clear_params()

清除所有全局参数。

column_scope_from_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date') → ColumnScope

创建一个预先进行 numpy 提取的 ColumnScope。

disable_logging()

禁用事件日志记录。

disable_progress_bar()

禁用进度条日志记录。

enable_logging()

启用事件日志记录。

enable_progress_bar()

启用进度条日志记录。

get_signals(symbols: Iterable[str], col_scope: ColumnScope, ind_scope: IndicatorScope, pred_scope: PredictionScope) → dict[str, DataFrame]

获取一个字典，其中包含每个品种的 K 线数据、指标数据和模型预测结果的 pandas.DataFrame。

merge_symbol_array_stores(left: SymbolArrayStore, right: SymbolArrayStore) → SymbolArrayStore

拼接两个存储中各品种的列数组。

param(name: str, value: Any | None = <object object>) → Any | None

获取或设置一个全局参数。

register_columns(names: str | Iterable[str], *args)

注册用户自定义数据列的 names。

run_with_scope(scope: StaticScope, fn: Callable[[...], Any], *args: Any) → Any

将 scope 安装为本进程的作用域，然后运行 fn。

StaticScope 是每个进程的单例，因此工作进程启动时会得到一个空的单例，无法看到调用方已注册的指标、模型源、参数或自定义列。将分发给 pybroker.parallel.parallel() 的工作用本方法包裹起来，即可随之传递调用方的作用域。在顺序运行时， scope 已经是已安装的实例，此时本方法不产生任何效果。

slice_symbol_array_store_by_dates(store: SymbolArrayStore, selected_dates: Sequence[datetime64] | NDArray[datetime64]) → SymbolArrayStore

将存储过滤为日期属于 selected_dates 的行。

sym_data_from_store(store: SymbolArrayStore, data_cols: Iterable[str]) → dict[str, dict[str, NDArray | None]]

将 SymbolArrayStore 转换为各品种的列数组。

sym_exec_dates_from_store(store: SymbolArrayStore) → dict[str, frozenset[datetime64]]

从列存储中返回各品种的测试日期。

品种会按排序后的顺序遍历。SymbolArrayStore.symbols 是一个 frozenset[str]，若直接对其迭代，会按字符串哈希顺序生成该映射，而当各品种交易日历参差不齐时，这个顺序会决定每根 K 线上哪个品种被优先处理 —— 这会使资金受限的回测依赖于 PYTHONHASHSEED。排序还能与日历对齐路径已经使用的 sorted(test_syms) 顺序保持一致。

symbol_array_store_from_flat_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date', symbols: frozenset[str] | None = None) → SymbolArrayStore

通过 numpy 字典序排序（lex-sort）和分箱切片，从一个扁平数据帧构建存储。

symbol_array_store_from_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date', symbols: frozenset[str] | None = None) → SymbolArrayStore

从扁平或 MultiIndex OHLCV 数据帧构建存储。

symbol_array_store_from_indexed_df(df: DataFrame) → SymbolArrayStore

从已排序的 MultiIndex 数据帧构建 SymbolArrayStore。

unregister_columns(names: str | Iterable[str], *args)

取消注册用户自定义数据列的 names。