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
成员类型说明
pricespd.DataFrame已按日期排序的价格面板。
symbolslist[str]面板列 —— 唯一可交易的标的集合。
cashfloat未投资现金,每笔成交后更新。
positionsdict[str, float]每个标的的股数(允许小数)。
tradeslist[Trade]按时间顺序记录的每一笔成交。
configEngineConfig当前生效的成本假设。
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 的精确规则

  1. 不在面板中的标的会被静默丢弃。写错代码意味着那一腿根本不会交易。
  2. 每个权重都经过 `max(0.0, w)` 截断,负值变成 0 —— 引擎仅做多。
  3. 若权重之和大于 1.0,所有权重会除以总和,把组合归一化为恰好 100% 投资。杠杆不可能出现。
  4. 若新目标与上一次目标在 1e-6 以内相同,调用会立即返回 —— 不交易、不产生成本。
  5. 当前净值按 cash + Σ 股数 × 价格 估算,价格为 NaN 的标的会被跳过。
  6. 目标股数为 净值 × 权重 / 价格;价格为 NaN 或非正时,目标股数为 0。
  7. 先执行全部卖出以释放现金,再执行全部买入。
  8. 若买入超出可用现金,会按扣除佣金后的可用现金等比缩减

成交定价

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)

  • startend仅关键字参数且为闭区间;None 表示使用面板边界。
  • 每个日期上,调用 on_day(engine, dt),再按当日收盘价对组合估值。
  • 由于交易与估值使用同一收盘价,净值曲线会立即反映新权重。
  • 返回的 BacktestResultbenchmarkNone —— 需要基准时由调用方自行附加。

BacktestResult

字段类型说明
equitypd.Series每个日期的组合总价值。
holdingspd.DataFrame按日期索引的各标的股数。
tradeslist[Trade]每笔成交,含方向、股数、价格与佣金。
cashpd.Series每个日期的现金余额。
benchmarkpd.Series | Nonerun() 返回时始终为 None,需自行设置。
metadict自由格式的元数据容器。

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()      # 粗略的换手率度量