pybroker.common 模块

包含通用类和工具。

class BarData(date: NDArray[datetime64], open: NDArray[float64], high: NDArray[float64], low: NDArray[float64], close: NDArray[float64], volume: NDArray[float64] | None, vwap: NDArray[float64] | None, **kwargs)

基类：object

包含一系列 K 线的数据。每个字段都是一个 numpy.ndarray，存放该序列中各 K 线的值。这些值按时间先后升序排列。

参数:

• date -- 每根 K 线的时间戳。

• open -- 开盘价。

• high -- 最高价。

• low -- 最低价。

• close -- 收盘价。

• volume -- 成交量。

• vwap -- 成交量加权平均价（VWAP）。

• **kwargs -- 自定义数据字段。

class DataCol(*values)

基类：Enum

默认数据列名称。

DATE = 'date'

SYMBOL = 'symbol'

VOLUME = 'volume'

VWAP = 'vwap'

class Day(*values)

基类：Enum

星期枚举。

FRI = 4

MON = 0

SAT = 5

SUN = 6

THURS = 3

TUES = 1

WEDS = 2

class FeeInfo(symbol: str, shares: Decimal, fill_price: Decimal, order_type: Literal['buy', 'sell'])

基类：NamedTuple

包含用于自定义手续费计算的信息。

symbol

交易品种代码。

类型:

str

shares

订单中的股数。

类型:

decimal.Decimal

fill_price

订单的成交价格。

类型:

decimal.Decimal

order_type

订单类型，为 "buy" 或 "sell"。

类型:

Literal['buy', 'sell']

class FeeMode(*values)

基类：Enum

回测中使用的经纪商手续费模式。

ORDER_PERCENT

手续费为订单金额的百分比，其中订单金额为 fill_price * shares。

PER_ORDER

手续费为每笔订单的固定金额。

PER_SHARE

手续费为订单中每股的固定金额。

class IndicatorSymbol(ind_name: str, symbol: str)

基类：NamedTuple

pybroker.indicator.Indicator/品种标识符。

ind_name

指标名称。

类型:

str

symbol

股票代码。

类型:

str

class ModelSymbol(model_name: str, symbol: str)

基类：NamedTuple

pybroker.model.ModelSource/品种标识符。

model_name

模型名称。

类型:

str

symbol

股票代码。

类型:

str

class OrderType(*values)

基类：Enum

订单类型分类。

MARKET

市价单。

LIMIT

限价单。

STOP_BAR

由 K 线止损触发的订单。

STOP_LOSS

由止损触发的订单。

STOP_PROFIT

由止盈触发的订单。

STOP_TRAILING

由移动止损触发的订单。

class PositionIntent(*values)

基类：Enum

订单的仓位意图。

BUY_TO_OPEN

买入以开多头仓位。

BUY_TO_CLOSE

买入以平空头仓位。

SELL_TO_OPEN

卖出以开空头仓位。

SELL_TO_CLOSE

卖出以平多头仓位。

class PositionMode(*values)

基类：Enum

回测中使用的仓位模式。

DEFAULT

多头和空头仓位。

LONG_ONLY

仅多头仓位。

SHORT_ONLY

仅空头仓位。

class PriceType(*values)

基类：Enum

用于通过 pybroker.context.ExecContext 指定成交价格的价格类型枚举。

OPEN

当前 K 线的开盘价。

LOW

当前 K 线的最低价。

HIGH

当前 K 线的最高价。

CLOSE

当前 K 线的收盘价。

MIDDLE

当前 K 线最低价与最高价的中点。

AVERAGE

当前 K 线开盘价、最低价、最高价和收盘价的平均值。

class StopType(*values)

基类：Enum

止损类型。

BAR

在 n 根 K 线后触发的止损。

LOSS

止损。

PROFIT

止盈。

TRAILING

移动止损。

SymbolSelector

为每个向前分析窗口选择该执行函数交易的股票代码。

传递给 pybroker.strategy.Strategy.add_execution() 以替代固定的品种列表。每个向前分析窗口都会调用一次，传入该窗口的 训练 pandas.DataFrame —— 绝不会传入测试数据 —— 并且必须返回一个非空的、由该数据帧中存在的唯一品种组成的序列。可接受任意 str 序列，包括 list、tuple、pandas.Index 或 numpy.ndarray，因此 ranked.nlargest(10).index 可以直接使用。

备注

• 必须存在训练窗口。pybroker.strategy.Strategy.backtest() 与 train_size=0 会引发 ValueError，因为从测试数据中选择品种相当于提前查看了即将交易的 K 线。

• 候选品种池必须来自 pandas.DataFrame 数据源；不能使用 pybroker.data.DataSource，因为在窗口划分完成之前，待查询的品种是未知的。

• shuffle=True 会打乱训练数据帧的行顺序，因此依赖 K 线按日期排序的选择器不应与其搭配使用。

• 当某个窗口丢弃了一个仍持有仓位的品种时，该仓位会在新测试窗口的第一根 K 线上被平掉，使用 pybroker.config.StrategyConfig.exit_sell_fill_price 和 pybroker.config.StrategyConfig.exit_cover_fill_price。而完全没有剩余 K 线的品种，则会在其最后一根 K 线上被平仓。

Callable[[DataFrame], Sequence[str]] 的别名

class TrainedModel(name: str, instance: Any, predict_fn: Callable[[Any, DataFrame | NDArray], NDArray] | None, input_cols: tuple[str] | None, per_bar: bool = False, lag_columns: tuple[str, ...] | None = None)

基类：NamedTuple

已训练模型/品种标识符。

name

已训练模型名称。

类型:

str

instance

已训练模型实例。

类型:

Any

predict_fn

用于覆盖调用模型默认 predict 函数的 Callable。对于使用 lags 训练的模型，调用时传入的是滞后特征矩阵（numpy.ndarray），而不是 pandas.DataFrame。

类型:

Callable[[Any, pandas.DataFrame | numpy._typing._array_like.NDArray], numpy._typing._array_like.NDArray] | None

input_cols

进行预测时用作模型输入的列名称。

类型:

tuple[str] | None

per_bar

如果为 True，预测会按 K 线逐根增量进行。

类型:

bool

lag_columns

训练时用于构建滞后特征的列名称，按特征块顺序排列。进行预测时会复用这些名称，以确保滞后特征与模型训练时所用的一致。如果模型未使用 lags 训练，则为 None。

类型:

tuple[str, ...] | None

bars_to_df(bar_data: BarData) → DataFrame

将 BarData 实例转换为 pandas.DataFrame。

参数:

bar_data -- 待转换的 BarData。

返回:

包含 bar_data 中每个字段所对应列的 pandas.DataFrame，包括自定义数据字段。volume 和 vwap 列仅在已设置时才会包含。

get_unique_sorted_dates(col: Series) → Sequence[datetime64]

从 DataFrame 的日期列中返回排序后的唯一值。

get_unique_sorted_dates_array(dates: Series | NDArray[datetime64] | Sequence[datetime64]) → NDArray[datetime64]

从 numpy 日期数组或 Series 中返回排序后的唯一日期。

parse_timeframe(timeframe: str) → list[tuple[int, str]]

解析支持以下单位的时间框架字符串：

• "s"/"sec"：秒

• "m"/"min"：分钟

• "h"/"hour"：小时

• "d"/"day"：天

• "w"/"week"：周

时间框架字符串示例：1h 30m。

返回:

由 tuple[int, str] 组成的 list，其中每个元组包含一个 int 值和一个 str 单位，单位为以下之一：sec、min、hour、day、week。

quantize(df: DataFrame, col: str, round: bool) → Series

通过将数值四舍五入到最接近的美分，对 pandas.DataFrame 的某一列进行量化。

返回:

转换为 float 值的量化后的列。

to_datetime(date: str | datetime | datetime64 | Timestamp) → datetime

将 date 转换为 datetime.datetime。

to_decimal(value: int | float | Decimal) → Decimal

将 value 转换为 decimal.Decimal。

to_seconds(timeframe: str | None) → int

将时间框架字符串转换为秒数，其中 timeframe 支持以下单位：

• "s"/"sec"：秒

• "m"/"min"：分钟

• "h"/"hour"：小时

• "d"/"day"：天

• "w"/"week"：周

时间框架字符串示例：1h 30m。

返回:

转换后的秒数。

verify_data_source_columns(df: DataFrame)

验证 pandas.DataFrame 是否包含 pybroker.data.DataSource 所需的全部列。

verify_date_range(start_date: datetime, end_date: datetime)

验证日期范围边界。