pybroker.interval 模块

多区间 K 线压缩工具。

Copyright (C) 2023 Edward West. All rights reserved.

This code is licensed under Apache 2.0 with Commons Clause license (see LICENSE for details).

BASE_INTERVAL: Final = 'base'

pybroker.indicator.Indicator.intervals() 和 pybroker.model.ModelSource.intervals() 所接受的哨兵值，用于在列出的压缩区间之外，额外请求基础时间框架的版本。

class CompressedBars(open: ~numpy._typing._array_like.NDArray[~numpy.float64], high: ~numpy._typing._array_like.NDArray[~numpy.float64], low: ~numpy._typing._array_like.NDArray[~numpy.float64], close: ~numpy._typing._array_like.NDArray[~numpy.float64], volume: ~numpy._typing._array_like.NDArray[~numpy.float64], dates: ~numpy._typing._array_like.NDArray[~numpy.datetime64], custom: ~typing.Mapping[str, ~numpy._typing._array_like.NDArray[~numpy.float64]] = <factory>, vwap: ~numpy._typing._array_like.NDArray[~numpy.float64] | None = None)[源代码]

基类：object

聚合为压缩 K 线的 OHLCV 及自定义列。

close: NDArray[float64]
custom: Mapping[str, NDArray[float64]]
dates: NDArray[datetime64]
high: NDArray[float64]
low: NDArray[float64]
open: NDArray[float64]
slice_by_dates(dates: Iterable[datetime64]) → CompressedBars[源代码]

返回限定在 dates 范围内的压缩 K 线。

volume: NDArray[float64]
vwap: NDArray[float64] | None = None
class CompressedSymbolData(bars: CompressedBars, completed: NDArray[int64], base_dates: NDArray[datetime64])[源代码]

基类：object

单个品种的压缩 K 线数据及对齐映射。

bars: CompressedBars
base_dates: NDArray[datetime64]
completed: NDArray[int64]
INTERVAL_NAME_SEPARATOR = '@'

指标和模型名称中为区间绑定保留的分隔符。

class IntervalData(compressed: dict[tuple[str, int | ~typing.Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], ~pybroker.interval.CompressedSymbolData] = <factory>)[源代码]

基类：object

以 (symbol, interval) 为键的压缩数据。

compressed: dict[tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], CompressedSymbolData]
slice_for_test(test_symbol_dates: Mapping[str, NDArray[datetime64]]) → IntervalData[源代码]

返回一个 completed 数组已与测试日期对齐的副本。

TimeframeInterval

用于多区间数据的压缩区间。

  • int``（``n > 1）：每 n 根基础 K 线（例如 5）。

  • str 时长：数字加单个单位字母 —— "5m"、"1h"、"30s" 或 "1d"``（字母：``s、m、h、d）。

  • str 日历："daily"、"weekly"、"monthly"、"quarterly" 或 "yearly"，对齐到日历边界：周从周一开始，月从每月 1 日开始，季度从 1、4、7、10 月开始，年从 1 月 1 日开始。

int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str 的别名

base_timeframe_to_seconds(base_timeframe: str) → float[源代码]

将基础时间框架字符串转换为秒数。

build_compressed_symbol_arrays(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, compressed: CompressedSymbolData, indicator_data: Mapping[IndicatorSymbol, Series], indicator_names: Iterable[str], custom_cols: Iterable[str]) → tuple[tuple[str, ...], dict[str, NDArray], NDArray[datetime64]][源代码]

使用基础指标名称构建压缩 K 线的列数组。

build_compressed_symbol_df(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, compressed: CompressedSymbolData, indicator_data: Mapping[IndicatorSymbol, Series], indicator_names: Iterable[str], custom_cols: Iterable[str]) → DataFrame[源代码]

使用基础指标列名构建压缩 K 线的 DataFrame。

不用于回测热路径；请优先使用 build_compressed_symbol_arrays()。

compress(dates: NDArray[datetime64], open_: NDArray[float64], high: NDArray[float64], low: NDArray[float64], close: NDArray[float64], volume: NDArray[float64], interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Mapping[str, NDArray[float64]] | None = None, vwap: NDArray[float64] | None = None) → tuple[CompressedBars, NDArray[int64]][源代码]

将基础 K 线压缩为更粗粒度的区间 K 线。

返回压缩 K 线以及一个 completed 对齐映射，其中 completed[t] 是基础 K 线 t 处最后一根 已完成 压缩 K 线的索引，在预热期内为 -1。

compress_bars(data: BarData | DataFrame, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, *, base_timeframe: str) → BarData[源代码]

将基础 OHLCV K 线压缩为更粗粒度的 interval。

参数:

  • data -- 单品种的 BarData，或 OHLCV pandas.DataFrame。

  • interval -- 目标压缩区间。

  • base_timeframe -- 声明的基础 K 线间隔（例如 "1m"、"1d"）。

返回:

压缩后的 BarData。

compress_intervals_from_frame(df: DataFrame, symbol_intervals: Mapping[str, Iterable[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str]], custom_cols: Iterable[str], base_bar_seconds: float) → IntervalData[源代码]

将每个品种压缩为其声明的区间。

参数:

  • df -- 多品种 OHLCV 数据帧。

  • symbol_intervals -- 将每个品种映射到其被压缩到的区间。映射中缺失的品种会被跳过，因此策略只会为其执行函数所声明的 (symbol, interval) 组合付出计算成本，而不是完整的品种 x 区间笛卡尔积。

  • custom_cols -- 随同压缩 K 线一起携带的自定义数据列。

  • base_bar_seconds -- 基础数据源的 K 线间隔，以秒为单位。

compress_symbol_df(sym_df: DataFrame, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True) → CompressedSymbolData[源代码]

压缩单品种 DataFrame。

compress_symbol_from_frame(df: DataFrame, symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True) → CompressedSymbolData[源代码]

从多品种数据帧中压缩单个品种，且不复制行。

compress_symbol_intervals_from_frame(df: DataFrame, symbol: str, intervals: Iterable[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True, rows: NDArray[int64] | None = None) → dict[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, CompressedSymbolData][源代码]

通过一次 OHLCV 提取，将单个品种压缩为多个区间。

rows 可选地提供该品种预先计算好的行索引，使得压缩多个品种的调用方只需对数据帧分组一次，而不必对每个品种都扫描一遍品种列。

compressed_bars_to_bar_data(bars: CompressedBars) → BarData[源代码]

将压缩后的 OHLCV 数组转换为 BarData。

format_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[源代码]

返回 interval 的稳定字符串表示。

indicator_interval_name(base: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[源代码]

返回某个区间绑定所对应的带后缀指标名称。

is_valid_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_bar_seconds: float) → bool[源代码]

返回 interval 对于基础数据源的 K 线间隔是否有效。

lookahead_train_dates(bar_dates: NDArray[datetime64], train_dates: Iterable[datetime64], test_dates: Iterable[datetime64], lookahead: int) → tuple[NDArray[datetime64], int][源代码]

裁剪压缩训练 K 线的日期，使训练/测试之间的保留区间恰为 lookahead 根压缩 K 线宽。

向前分析的划分会保留 基础 时间框架的 lookahead 根 K 线，但绑定到某个区间的模型是基于压缩 K 线拟合的，因此保留区间必须以压缩 K 线为单位重新度量：每根被保留的训练 K 线都满足 compressed_index <= first_test_compressed_index - lookahead。

参数:

  • bar_dates -- 单个品种完整压缩 K 线历史的日期。

  • train_dates -- 基础时间框架的训练窗口日期；压缩 K 线根据其收盘日期是否属于该集合来选取。

  • test_dates -- 基础时间框架的测试窗口日期。

  • lookahead -- 要保留的压缩 K 线数量。

返回:

(dates_to_select, n_dropped) —— 要保留的训练 K 线日期，以及被丢弃的训练压缩 K 线数量。当 lookahead <= 1 时，请求的训练日期会原样返回，这与日期归属本身已经产生的单根 K 线间隔一致。

model_interval_name(base: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[源代码]

返回某个区间绑定所对应的带后缀模型名称。

normalize_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str[源代码]

规范化并验证一个压缩区间。

normalize_intervals(intervals: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | Iterable[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], param: str, allow_base: bool = False) → frozenset[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str][源代码]

将一个或多个压缩区间规范化为一个 frozenset，拒绝空输入和重复项。

参数:

  • intervals -- 单个 TimeframeInterval，或由若干个组成的 Iterable。

  • param -- 错误消息中使用的参数名称。

  • allow_base -- 如果为 True，字面量 'base' 会原样通过。否则会像其他任何无效区间一样被拒绝。

parse_indicator_interval_name(name: str) → tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | None][源代码]

将带后缀的指标名称解析为基础名称和区间。

parse_model_interval_name(name: str) → tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | None][源代码]

将带后缀的模型名称解析为基础名称和区间。

slice_arrays_by_dates(columns: tuple[str, ...], arrays: Mapping[str, NDArray], dates: NDArray[datetime64], selected: Iterable[datetime64]) → tuple[tuple[str, ...], dict[str, NDArray], NDArray[datetime64]][源代码]

将列数组过滤为日期属于 selected 的行。

slice_compressed_df_by_dates(df: DataFrame, dates: Iterable[datetime64]) → DataFrame[源代码]

将压缩 DataFrame 过滤为日期属于 dates 的行。

不用于回测热路径；请优先使用 slice_arrays_by_dates()。

symbol_dates_from_frame(df: DataFrame) → dict[str, NDArray[datetime64]][源代码]

从多品种数据帧中提取每个品种的测试日期。

validate_base_timeframe_data(df: DataFrame, base_bar_seconds: float) → None[源代码]

如果 K 线时间戳与 base_bar_seconds 不一致，则引发异常。

validate_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_bar_seconds: float) → None[源代码]

根据基础数据源的 K 线间隔验证某个区间。

validate_source_name(name: str, kind: str) → None[源代码]

如果 name 不能用作指标或模型名称，则引发异常。

参数:

  • name -- 正在注册的名称。

  • kind -- 'indicator' 或 'model'，用于错误消息中。