Quant Buffet API
语法手册
真正需要的 pandas 与 numpy 语法：面板取值、指标、横截面排名、权重与调仓节奏。
几乎每个 Quant Buffet 策略都由同样的二十多个表达式拼装而成。本页就是它们的参考手册 —— 复制一段、改改变量名，你就得到了一个信号。
取用价格面板
python
# Always intersect with the columns that actually loaded
cols = [c for c in ASSETS if c in prices.columns]
close = prices[cols]
close.index # DatetimeIndex, tz-naive, sorted ascending
close.columns # loaded tickers, in ASSETS order
close.shape # (n_days, n_symbols)
prices.attrs.get("load_errors") # list[str] when some symbols failed
close.loc[dt] # Series: one row, indexed by symbol
close["SPY"] # Series: one column, indexed by date
close.at[dt, "SPY"] # scalar — the fastest single-cell access
close.iloc[-1] # last row (avoid inside on_day: it hides the date)| 取值方式 | 返回 | 可在 on_day 内使用？ |
|---|---|---|
prices.at[dt, sym] | 标量 float | 可以 — 最快 |
prices.loc[dt] | 按标的索引的 Series | 可以 — 每天一行 |
prices[sym] | 按日期索引的 Series | 请在 make_on_day 中构建 |
prices.loc[:dt] | 不断增长的 DataFrame 切片 | 避免 — 这正是超时的元凶 |
prices.iloc[-1] | 整个面板的最后一行 | 不可 — 忽略 dt，泄露未来 |
指标词汇表
python
close = prices[cols]
# --- Trend ---------------------------------------------------------------
sma_200 = close.rolling(200, min_periods=200).mean()
ema_50 = close.ewm(span=50, adjust=False).mean()
above = close > sma_200 # boolean DataFrame
# --- Returns and momentum ----------------------------------------------
daily = close.pct_change() # 1-day simple returns
mom_12m = close.pct_change(252) # trailing 12-month return
mom_6m = close.pct_change(126) # trailing 6-month return
log_ret = np.log(close).diff() # log returns
# --- Volatility and risk ------------------------------------------------
vol_20 = daily.rolling(20).std() * np.sqrt(252) # annualised
downside = daily.clip(upper=0).rolling(60).std()
# --- Mean reversion -----------------------------------------------------
mu = close.rolling(20, min_periods=20).mean()
sd = close.rolling(20, min_periods=20).std(ddof=0)
zscore = (close - mu) / sd.replace(0, np.nan) # guard divide-by-zero
# --- Range and drawdown -------------------------------------------------
high_252 = close.rolling(252).max()
dd_from_high = close / high_252 - 1
rsi_up = daily.clip(lower=0).rolling(14).mean()
rsi_dn = (-daily.clip(upper=0)).rolling(14).mean()
rsi = 100 - 100 / (1 + rsi_up / rsi_dn.replace(0, np.nan))| 表达式 | 含义 | 预热行数 |
|---|---|---|
close.pct_change() | 1 日简单收益率 | 1 |
close.pct_change(252) | 过去 12 个月收益率 | 252 |
close.rolling(n, min_periods=n).mean() | 简单移动平均 | n |
close.ewm(span=n, adjust=False).mean() | 指数移动平均 | 约 n |
daily.rolling(n).std() * np.sqrt(252) | 年化波动率 | n + 1 |
close.rolling(n).max() | n 日最高价（突破逻辑） | n |
(close - mu) / sd | z-score，均值回归 | mu 的窗口长度 |
横截面计算（axis=1）
动量轮动、相对强弱与 top-N 选择，都是在同一日期上把标的互相比较，在 pandas 中即沿 axis=1 操作。
python
# Rank across symbols on each date: axis=1 is the cross-section
mom = close.pct_change(126)
ranks = mom.rank(axis=1, ascending=False) # 1 = strongest that day
top3 = ranks <= 3 # boolean mask
# Cross-sectional z-score (relative strength, market-neutral in spirit)
row_mu = mom.mean(axis=1)
row_sd = mom.std(axis=1).replace(0, np.nan)
rel = mom.sub(row_mu, axis=0).div(row_sd, axis=0)
# Inside on_day: pick names from a single row
def pick_top(dt, n=3):
row = mom.loc[dt].dropna()
if row.empty:
return []
ranked = row.sort_values(ascending=False)
return [s for s in ranked.head(n).index if ranked[s] > 0]把信号转换为权重
python
# Equal weight over the selected names
def equal_weight(picks: list[str]) -> dict[str, float]:
if not picks:
return {} # {} means go to cash
w = 1.0 / len(picks)
return {s: w for s in picks}
# Inverse-volatility weights (risk parity flavour)
def inverse_vol(dt, picks, vol) -> dict[str, float]:
inv = {s: 1.0 / float(vol.at[dt, s]) for s in picks
if pd.notna(vol.at[dt, s]) and vol.at[dt, s] > 1e-8}
total = sum(inv.values())
return {s: v / total for s, v in inv.items()} if total > 0 else {}
# Volatility targeting: scale exposure, leave the remainder in cash
def vol_scaled(dt, symbol, vol, target=0.10) -> dict[str, float]:
v = float(vol.at[dt, symbol])
return {symbol: min(1.0, target / v)} if v > 1e-8 else {}
# Fixed sleeves with a deliberate cash buffer
CORE = {"SPY": 0.45, "TLT": 0.25, "GLD": 0.15} # 85% invested, 15% cash| 返回的权重 | 引擎行为 |
|---|---|
{"SPY": 1.0} | 100% 持有 SPY |
{"SPY": 0.6, "TLT": 0.4} | 满仓 60/40 |
{"SPY": 0.5, "TLT": 0.2} | 70% 投资，30% 现金 |
{"SPY": 0.8, "TLT": 0.8} | 归一化为 50/50 — 无法使用杠杆 |
{"SPY": -0.4} | 截断为 0.0 — 不支持做空 |
{"NVDA": 1.0} | 被忽略 — 该标的不是面板列 |
{} | 清仓为 100% 现金 |
在 DataFrame 上做布尔逻辑
python
# 用 & | ~ 组合条件，并且务必加括号
trend_up = close > sma_200
cheap = zscore < -1.0
signal = trend_up & cheap # 逐元素 AND
risk_off = ~trend_up["SPY"] # 逐元素 NOT
# 按行归约（这些是 pandas 方法，而非缺失的内置函数）
sma_200.loc[dt].isna().all() # 当天还没有任何信号
trend_up.loc[dt].any() # 至少有一个标的处于上升趋势
int(trend_up.loc[dt].sum()) # 有多少个处于上升趋势
# 用 shift 实现「今天决策、下一根 K 线执行」
delayed = signal.shift(1)正确处理 NaN
| 表达式 | 用途 |
|---|---|
pd.notna(x) / pd.isna(x) | 对标量安全的 NaN 判断 — 优于 x == x |
series.dropna() | 排名前先剔除缺失标的 |
frame.dropna(how="all") | 丢弃完全没有信号的日期 |
sd.replace(0, np.nan) | 避免 z-score 中的除零 |
series.reindex(prices.index).ffill() | 把重采样序列对齐回交易日 |
frame.clip(lower=0) | 把负值归零，例如 RSI 或仅做多的打分 |
日期与日历
python
dt.year, dt.month, dt.day # 整数
dt.dayofweek # 周一 = 0 … 周日 = 6
dt.isocalendar()[:2] # (ISO 年, ISO 周) — 周度门控
dt.strftime("%Y-%m-%d") # Trade.date 使用的字符串形式
dt in prices.index # .at 取值前的成员判断
# 按月末重采样，再对齐回日频交易日
monthly = close.resample("ME").last()
monthly_mom = monthly.pct_change(12).reindex(close.index).ffill()