Quant Buffet API
引擎 API
PortfolioEngine、EngineConfig、精确的成交算法、Trade 与 BacktestResult。
`backtest.engine` 实现了一个带现金、佣金与滑点的日频、仅做多模拟器。策略与它的交互几乎全部通过 `set_target_weights` 完成。
EngineConfig
python
@dataclass
class EngineConfig:
initial_cash: float = 100_000.0
commission_bps: float = 5.0 # 每笔成交按成交额收取的基点数
slippage_bps: float = 2.0 # 每个方向的价格冲击基点数实验室始终使用 EngineConfig(initial_cash=100_000, commission_bps=5.0, slippage_bps=2.0)。本地脚本可以覆盖其中任意一项。
PortfolioEngine
python
config = EngineConfig(
initial_cash=100_000.0,
commission_bps=5.0, # 5 bps of notional per fill
slippage_bps=2.0, # 2 bps price impact per side
)
engine = PortfolioEngine(prices, config)
def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
engine.set_target_weights(dt, {"SPY": 0.6, "TLT": 0.4})
result = engine.run(on_day, start=pd.Timestamp("2015-01-01"), end=None)
# result.equity, result.holdings, result.trades, result.cash| 成员 | 类型 | 说明 |
|---|---|---|
prices | pd.DataFrame | 已按日期排序的价格面板。 |
symbols | list[str] | 面板列 —— 唯一可交易的标的集合。 |
cash | float | 未投资现金,每笔成交后更新。 |
positions | dict[str, float] | 每个标的的股数(允许小数)。 |
trades | list[Trade] | 按时间顺序记录的每一笔成交。 |
config | EngineConfig | 当前生效的成本假设。 |
set_target_weights(dt, weights) | 方法 | 调仓入口。 |
run(on_day, *, start, end) | 方法 | 遍历交易日并返回 BacktestResult。 |
python
def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
engine.cash # float: uninvested cash right now
engine.positions # dict[str, float]: share count per symbol
engine.symbols # list[str]: columns of the price panel
engine.config.commission_bps # cost assumptions in force
len(engine.trades) # fills so far
# There is no current_weights() helper — derive it when you need it
px = engine.prices.loc[dt]
mv = {s: engine.positions[s] * float(px[s])
for s in engine.symbols if pd.notna(px[s])}
equity = engine.cash + sum(mv.values())
weights = {s: v / equity for s, v in mv.items()} if equity > 0 else {}set_target_weights 的精确规则
- 不在面板中的标的会被静默丢弃。写错代码意味着那一腿根本不会交易。
- 每个权重都经过 `max(0.0, w)` 截断,负值变成 0 —— 引擎仅做多。
- 若权重之和大于 1.0,所有权重会除以总和,把组合归一化为恰好 100% 投资。杠杆不可能出现。
- 若新目标与上一次目标在 1e-6 以内相同,调用会立即返回 —— 不交易、不产生成本。
- 当前净值按
cash + Σ 股数 × 价格估算,价格为NaN的标的会被跳过。 - 目标股数为
净值 × 权重 / 价格;价格为NaN或非正时,目标股数为 0。 - 先执行全部卖出以释放现金,再执行全部买入。
- 若买入超出可用现金,会按扣除佣金后的可用现金等比缩减。
成交定价
python
# 滑点在买卖两侧都对你不利
buy_price = close * (1 + slippage_bps / 10_000) # 买入付得更多
sell_price = close * (1 - slippage_bps / 10_000) # 卖出收得更少
notional = shares * fill_price
commission = notional * (commission_bps / 10_000)
# 买入: cash -= notional + commission
# 卖出: cash += notional - commission按实验室默认值,一次完整往返约花费 14 bps:卖出侧 5 bps 佣金加 2 bps 滑点,买入侧再来一次。小于 1e-8 股的仓位会被直接归零,以防积累灰尘头寸。
run(on_day, *, start=None, end=None)
start与end是仅关键字参数且为闭区间;None表示使用面板边界。- 每个日期上,先调用
on_day(engine, dt),再按当日收盘价对组合估值。 - 由于交易与估值使用同一收盘价,净值曲线会立即反映新权重。
- 返回的
BacktestResult中benchmark为None—— 需要基准时由调用方自行附加。
BacktestResult
| 字段 | 类型 | 说明 |
|---|---|---|
equity | pd.Series | 每个日期的组合总价值。 |
holdings | pd.DataFrame | 按日期索引的各标的股数。 |
trades | list[Trade] | 每笔成交,含方向、股数、价格与佣金。 |
cash | pd.Series | 每个日期的现金余额。 |
benchmark | pd.Series | None | run() 返回时始终为 None,需自行设置。 |
meta | dict | 自由格式的元数据容器。 |
Trade
python
@dataclass
class Trade:
date: str # "YYYY-MM-DD" 字符串,而非 Timestamp
symbol: str
side: str # "buy" | "sell"
shares: float # 保留 6 位小数
price: float # 含滑点后价格,保留 6 位小数
value: float # 成交额,保留 2 位小数
commission: float # 保留 4 位小数python
# 本地脚本中可做的事后分析
import pandas as pd
tr = pd.DataFrame([t.__dict__ for t in result.trades])
tr.groupby("symbol")["commission"].sum() # 按标的统计成本
tr.groupby("side")["value"].sum() # 买卖各自的成交额
tr["value"].sum() / result.equity.mean() # 粗略的换手率度量