Quant Buffet API
实验室契约
必需符号、函数签名、ready 语义，以及 make_on_day → on_day 模式。
沙箱加载器（backtest/sandbox_runner.py）要求一份固定契约。缺少或改名这些部件的策略会在加载任何价格之前就失败。
ASSETS（必需）
必须定义在模块作用域 —— 不能放在函数里，也不能放在 if __name__ == "__main__" 中：
python
ASSETS = ["SPY", "QQQ", "TLT", "GLD", "BIL"]| 规则 | 细节 | 失败信息 |
|---|---|---|
| 必须存在于模块级 | 从执行后的命名空间读取 | *Define ASSETS = ['SPY', ...] at module level…* |
| 必须是非空 list 或 tuple | 字符串会被 strip，空串被丢弃 | *ASSETS is empty after validation.* |
| 每个标的都在白名单内 | WHITELIST 加上 BTC-USD / ETH-USD | *Symbol 'X' is not in the Quant Buffet whitelist.* |
| 最多 15 个标的 | 按去重后的数量计算 | *Too many symbols (max 15).* |
make_on_day(prices)
python
def make_on_day(prices: pd.DataFrame):
# prices: 行 = 交易日, 列 = 成功加载的 ASSETS 标的（复权收盘价）
...
return on_day, ready| 返回值 | 类型 | 说明 |
|---|---|---|
on_day | Callable[[PortfolioEngine, pd.Timestamp], None] | 自 ready 起，每个交易日调用一次。 |
ready | pd.Timestamp | None | 信号首次有效的日期，会直接传入 engine.run(..., start=ready)。 |
make_on_day 只被调用一次，因此所有昂贵计算都应放在这里：滚动均值、排名、z-score、波动率。放进 on_day 的计算会被重复数千次，可能导致运行超时。
正确计算 `ready`
| 写法 | 适用场景 |
|---|---|
indicator.dropna(how="all").index.min() | 标准：任一标的首次有信号的日期 |
indicator.dropna().index.min() | 更严格：所有标的都有信号的日期 |
series.first_valid_index() | 单序列信号 |
prices.index[0] | 完全无需预热，例如静态等权 |
on_day(engine, dt)
- 接收当前的 `PortfolioEngine` 实例与当前 `pd.Timestamp`。
- 返回
None。它唯一的职责是调用 `engine.set_target_weights(dt, weights)`，或提前 return 以维持现有持仓。 - 权重仅做多且应加总为 ≤ 1.0，剩余部分留作现金。
- 提前 return 是完全合法且零成本的选择：不调用即当天不交易。
- 用
on_day闭包捕获的状态字典实现调仓节奏与持仓记忆。
调仓节奏写法
python
def make_on_day(prices: pd.DataFrame):
state = {"month": None, "week": None, "count": 0}
def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
# A) Monthly — fires on the first trading day of each new month
key = (dt.year, dt.month)
if state["month"] == key:
return
state["month"] = key
# B) Weekly — swap the guard above for the ISO week number
# key = dt.isocalendar()[:2]
# if state["week"] == key: return
# state["week"] = key
# C) Quarterly — month boundary, but only Jan / Apr / Jul / Oct
# if dt.month not in (1, 4, 7, 10): return
# D) Every N trading days
# state["count"] += 1
# if state["count"] % 21 != 0: return
engine.set_target_weights(dt, {"SPY": 1.0})
return on_day, prices.index[0]防御性保护
python
def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
# 1. Skip dates where the whole indicator row is NaN (warmup, holidays)
row = sma.loc[dt]
if row.isna().all():
return
# 2. Skip individual symbols that are not ready yet
live = [s for s in cols if pd.notna(row[s]) and pd.notna(prices.at[dt, s])]
if not live:
return
# 3. Never divide by a zero or NaN denominator
v = vol.at[dt, "SPY"]
if pd.isna(v) or v <= 1e-8:
return
# 4. any() / all() are NOT injected — use pandas or a comprehension
if row.gt(0).any(): # pandas method: fine
pass
engine.set_target_weights(dt, {s: 1.0 / len(live) for s in live})推荐的 import
python
from __future__ import annotations
import numpy as np
import pandas as pd
from backtest.data import load_daily_prices
from backtest.engine import EngineConfig, PortfolioEngine
from backtest.metrics import compute_metrics反模式
| 反模式 | 为何出错 |
|---|---|
把 ASSETS 写在 make_on_day 内部 | 加载器只读取模块级 ASSETS |
从 make_on_day 返回权重 | 它必须返回 (on_day, ready) 这一对值 |
只 return on_day 而缺少 ready | TypeError: cannot unpack non-sequence |
在 on_day 内部计算滚动均值 | 每日重复计算；长历史会超时 |
使用全样本统计量（close.mean()） | 前视偏差 — 会大幅虚高 Sharpe |
使用个股或 UUP / FXE | 不在白名单内，加载阶段即被拒绝 |
import os / requests / sklearn | 被 AST 校验拦截 |
any(...) / all(...) | 未被注入为 builtins — 请用 pandas 的 .any() / .all() |