Quant Buffet API

示例

面向实验室与本地脚本的完整可运行策略范式。

以下每个示例都是一个完整文件。把其中之一粘贴到任意策略页的 IDE 中点击运行，然后改动一个参数并对比指标。

1. 从零编写的 SMA 趋势策略

按月调仓，等权持有价格位于自身 200 日均线之上的 ETF。state 字典是节奏门控；ready 等待首个有效 SMA。

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 = ["SPY", "QQQ", "TLT", "GLD", "BIL"]


def make_on_day(prices: pd.DataFrame):
    cols = [c for c in ASSETS if c in prices.columns]
    sma = prices[cols].rolling(200, min_periods=200).mean()
    state = {"last": None}

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        if sma.loc[dt].isna().all():
            return
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key
        long = [s for s in cols if prices.at[dt, s] > sma.at[dt, s]]
        w = 1.0 / len(long) if long else 0.0
        engine.set_target_weights(dt, {s: w for s in long})

    ready = sma.dropna(how="all").index.min()
    return on_day, ready

2. 同样的想法，用模板实现

三行代替二十行，机制完全一致。当你信任这个模式后，优先使用它。

python

from backtest.templates import make_sma_trend

ASSETS = ["SPY", "QQQ", "IWM"]

def make_on_day(prices: pd.DataFrame):
    return make_sma_trend(prices, ASSETS, {"sma_days": 200})

3. 从零编写的双动量策略

相对强弱挑出领先者；绝对强弱决定是否持有它。这是策略库中最常见的形态。

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 = ["SPY", "EFA", "EEM", "TLT", "GLD", "BIL"]
LOOKBACK = 252
CASH = "BIL"


def make_on_day(prices: pd.DataFrame):
    cols = [c for c in ASSETS if c in prices.columns]
    risky = [c for c in cols if c != CASH]
    mom = prices[cols].pct_change(LOOKBACK)
    state = {"last": None}

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key

        scores = {s: float(mom.at[dt, s]) for s in risky
                  if pd.notna(mom.at[dt, s])}
        if not scores:
            engine.set_target_weights(dt, {CASH: 1.0})
            return

        best = max(scores, key=scores.get)          # relative momentum
        if scores[best] > 0:                        # absolute momentum
            engine.set_target_weights(dt, {best: 1.0})
        else:
            engine.set_target_weights(dt, {CASH: 1.0})

    ready = mom.dropna(how="all").index.min()
    return on_day, ready

4. 横截面均值回归

买入近期收益向下偏离过大的标的。注意 sigma.replace(0, np.nan) 这个保护，以及把换手率控制在可承受范围内的月度门控。

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 = ["SPY", "QQQ", "IWM", "TLT"]
LOOKBACK = 20
ENTRY_Z = 1.0


def make_on_day(prices: pd.DataFrame):
    cols = [c for c in ASSETS if c in prices.columns]
    rets = prices[cols].pct_change()
    mu = rets.rolling(LOOKBACK, min_periods=LOOKBACK).mean()
    sigma = rets.rolling(LOOKBACK, min_periods=LOOKBACK).std()
    z = (rets - mu) / sigma.replace(0, np.nan)
    state = {"last": None}

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        if z.loc[dt].isna().all():
            return
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key
        # Buy recent losers (negative z), equal weight
        picks = [s for s in cols if z.at[dt, s] < -ENTRY_Z]
        w = 1.0 / len(picks) if picks else 0.0
        engine.set_target_weights(dt, {s: w for s in picks})

    ready = z.dropna(how="all").index.min()
    return on_day, ready

5. 带波动率预算的风险平价

先计算波动率倒数权重，再做第二步缩放，为整个组合的估计波动率设上限，并把剩余部分留作现金。

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 = ["SPY", "TLT", "GLD", "DBC"]
VOL_LOOKBACK = 63
TARGET_VOL = 0.10


def make_on_day(prices: pd.DataFrame):
    cols = [c for c in ASSETS if c in prices.columns]
    rets = prices[cols].pct_change()
    vol = rets.rolling(VOL_LOOKBACK, min_periods=VOL_LOOKBACK).std() * np.sqrt(252)
    state = {"last": None}

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key

        inv = {s: 1.0 / float(vol.at[dt, s]) for s in cols
               if pd.notna(vol.at[dt, s]) and vol.at[dt, s] > 1e-8}
        total = sum(inv.values())
        if total <= 0:
            return
        weights = {s: v / total for s, v in inv.items()}

        # Scale the whole book to a volatility budget, keep rest in cash
        book_vol = sum(weights[s] * float(vol.at[dt, s]) for s in weights)
        scale = min(1.0, TARGET_VOL / book_vol) if book_vol > 1e-8 else 0.0
        engine.set_target_weights(dt, {s: w * scale for s, w in weights.items()})

    ready = vol.dropna(how="all").index.min()
    return on_day, ready

6. 模板 + 市场过滤

把核心逻辑交给工厂函数，再在机制过滤器判定为风险规避时覆盖它的输出。

python

from backtest.templates import make_momentum_rotation

ASSETS = ["XLK", "XLF", "XLE", "XLV", "XLI", "XLP", "XLU", "XLY", "BIL"]


def make_on_day(prices: pd.DataFrame):
    base_on_day, ready = make_momentum_rotation(
        prices, ASSETS, {"lookback": 126, "top_n": 3}
    )

    # Add your own risk overlay on top of a template
    spy_sma = prices["XLK"].rolling(200, min_periods=200).mean()

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        m = spy_sma.at[dt] if dt in spy_sma.index else np.nan
        if pd.notna(m) and prices.at[dt, "XLK"] < m:
            engine.set_target_weights(dt, {"BIL": 1.0})   # market filter: risk off
            return
        base_on_day(engine, dt)

    return on_day, ready

7. 本地脚本：完整流水线

同一份策略主体接上加载器、引擎与指标模块，即可在沙箱之外运行。

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 = ["SPY", "TLT"]


def make_on_day(prices: pd.DataFrame):
    sma = prices[ASSETS].rolling(200, min_periods=200).mean()

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        risk_on = prices.at[dt, "SPY"] > sma.at[dt, "SPY"]
        engine.set_target_weights(dt, {"SPY": 1.0} if risk_on else {"TLT": 1.0})

    return on_day, sma.dropna(how="all").index.min()


if __name__ == "__main__":
    prices = load_daily_prices(ASSETS, start="2010-01-01")
    on_day, ready = make_on_day(prices)
    engine = PortfolioEngine(prices, EngineConfig())
    result = engine.run(on_day, start=ready)
    print(compute_metrics(result.equity, trades_count=len(result.trades)))

Note. 请在仓库根目录运行本地脚本，以便 backtest 包可被解析 —— 设置 PYTHONPATH=. 后执行 python my_strategy.py，或用 python -m backtest.sandbox_runner 走一遍沙箱路径。

接下来去哪里

语法手册 — 更多指标、排名与权重表达式。
模板 — 全部九个工厂函数的精确参数名。
错误 — 你第一次改动会遇到的问题的修复方法。
课程 — 这些范式背后的概念，配合真实市场历史。