股票，债券，商品 · 美国 · 期货 · 每周 Free

基于成交量、持仓量和过去收益的美国期货周度反转策略

登录后收藏

学术论文

全球股票短期均值回归指标比较

作者Raymond Micaletti

机构

Relative Sentiment Technologies, LLC

论文链接https://papers.ssrn.com/sol3/papers.cfm?abstract_id=4339128

导读

投资标的：GLD、SLV、DBC、GSG、USO、UNG、DBA。再平衡：每周。Universe: GLD, SLV, DBC, GSG, USO, UNG, DBA. Parameters: rebalance=weekly.

策略概要

该策略投资于24类美国期货（4种货币、5种金融、8种农产品、7种商品）。使用周三至周三的周度区间。采用最接近到期的合约，但在交割月改用次近月，并在交割月开始时移仓。若合约在t-1至t相对t-2至t-1的成交量变化高于（低于）全部合约成交量变化的中位数，则定义为高（低）成交量合约（周成交量除以样本均值去趋势，以便跨市场比较）。合约还按滞后持仓量变化分为高持仓量（变化最高50%）或低持仓量（最低50%）组。投资者做多（做空）高成交量、低持仓量组中上周收益最低（最高）的期货。各合约权重正比于其过去一周收益与该组N份合约等权平均收益之差。

经济逻辑

短窗口收益可预测性与过度反应假说一致：交易者对消息的后验信念调整超过基本面所保证的幅度。过度自信与过度反应本身意味着大量交易，从而与价格反转幅度正相关。因此，非理性导致的市场无效会在成交量与预期收益之间产生负向关系。持仓量代表套保者或不知情交易的套保活动，因而也是市场状态的重要决定因素。

入场与出场

入场

每周按 21 日收益从低到高排序 GLD, SLV, DBC, GSG, USO, UNG, DBA（反转），等权买入最弱的 1 只，博弈超跌回归。

出场

下一再平衡日不再位于最弱 1 的标的卖出，换入新的输家。

所需行情数据

需要 GLD, SLV, DBC, GSG, USO, UNG (+1 more) 至少 21 根复权收盘价，才能形成第一个有效信号。ETF 仅交易日有行情。实验室通过 backtest.data.load_daily_prices 读取 Yahoo Finance。

标的

GLD, SLV, DBC, GSG, USO, UNG, DBA

频率

每周 · 日频复权收盘价；信号按月再平衡（用当月最后一个有效交易日）

所需历史

每个标的至少 21 根交易日 K 线（约 1 个交易月），外加数根缓冲才能发出第一个信号。

字段

复权收盘价（Adj Close / auto_adjust Close）
交易日日期索引

凭证

回测与实验室：无需密钥
实盘下单才需要券商 / 交易所 API

API 与数据源

Quant Buffet 数据 API

实验室用 load_daily_prices 拉取 ASSETS 的复权收盘价面板，磁盘缓存后回测可复现。

提供方: backtest.data.load_daily_prices
鉴权: 回测无需密钥
接口: load_daily_prices(ASSETS, start="2000-01-01")

API / docs → /zh/docs/data-api

Yahoo Finance（yfinance）

美股 / ETF 复权 OHLCV，免费研究接口，有速率限制。

提供方: yfinance → Yahoo Finance
鉴权: 公开历史无需密钥
接口: Ticker.history / yf.download(auto_adjust=True)

API / docs → github.com/ranaroussi/yfinance

Onsite backtest IDE

Quant Buffet 原生回测 IDE

Edit and run Quant Buffet Python for 基于成交量、持仓量和过去收益的美国期货周度反转策略 in the browser. Results update live with equity, drawdown, and metrics charts. Allowed: backtest.data, backtest.engine, backtest.metrics, numpy, pandas. Define ASSETS and make_on_day(prices). Shortcut: Ctrl+Enter. API docs →

IDE · 42 lines

"""
Quant Buffet interactive lab — 基于成交量、持仓量和过去收益的美国期货周度反转策略
Allowed: backtest.data / backtest.engine / backtest.metrics, numpy, pandas.
Required: a module-level ASSETS list, and def make_on_day(prices) returning (on_day, ready)
"""
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 = ['GLD', 'SLV', 'DBC', 'GSG', 'USO', 'UNG', 'DBA']
LOOKBACK = 21
TOP_N = 1
INVERT = True  # True = short-term reversal (rank ascending)

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

    def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        row = rets.loc[dt]
        if row.isna().all():
            return
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key
        ranked = row.dropna().sort_values(ascending=INVERT)
        picks = list(ranked.head(TOP_N).index)
        if not INVERT:
            picks = [s for s in picks if ranked[s] > 0] or picks[:1]
        w = {} if not picks else {s: 1.0 / len(picks) for s in picks}
        engine.set_target_weights(dt, w)

    ready = rets.dropna(how="all").index.min() if rets.notna().any().any() else None
    return on_day, ready

"""
Quant Buffet interactive lab — 基于成交量、持仓量和过去收益的美国期货周度反转策略
Allowed: backtest.data / backtest.engine / backtest.metrics, numpy, pandas.
Required: a module-level ASSETS list, and def make_on_day(prices) returning (on_day, ready)
"""
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 = ['GLD', 'SLV', 'DBC', 'GSG', 'USO', 'UNG', 'DBA']
LOOKBACK = 21
TOP_N = 1
INVERT = True  # True = short-term reversal (rank ascending)

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

def on_day(engine: PortfolioEngine, dt: pd.Timestamp) -> None:
        row = rets.loc[dt]
        if row.isna().all():
            return
        key = (dt.year, dt.month)
        if state["last"] == key:
            return
        state["last"] = key
        ranked = row.dropna().sort_values(ascending=INVERT)
        picks = list(ranked.head(TOP_N).index)
        if not INVERT:
            picks = [s for s in picks if ranked[s] > 0] or picks[:1]
        w = {} if not picks else {s: 1.0 / len(picks) for s in picks}
        engine.set_target_weights(dt, w)

ready = rets.dropna(how="all").index.min() if rets.notna().any().any() else None
    return on_day, ready

Quant Buffet syntax cheat sheet (copy / insert)

Paste these fragments into the editor. The sandbox rejects QuantConnect, os, and network libraries.

Required imports

Only these libraries are allowed in the sandbox.

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 list (whitelisted ETFs)

Module-level list. Tickers must be in the Quant Buffet whitelist.

ASSETS = ["SPY", "QQQ", "TLT", "GLD", "BIL"]

make_on_day contract

Must return (on_day, ready). on_day calls engine.set_target_weights.

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 pd.notna(prices.at[dt, s]) and pd.notna(sma.at[dt, s])
            and prices.at[dt, s] > sma.at[dt, s]
        ]
        weights = {} if not long else {s: 1.0 / len(long) for s in long}
        engine.set_target_weights(dt, weights)

    ready = sma.dropna(how="all").index.min() if sma.notna().any().any() else None
    return on_day, ready

Set target weights

Weights should sum to about 1.0. Empty dict = 100% cash.

engine.set_target_weights(dt, {"SPY": 0.60, "BIL": 0.40})

Live backtest performance

CAGR

-13.34%

Sharpe

-0.15

Max DD

-99.15%

Vol

40.75%

Sortino

-0.23

Beta

0.44

Up days

47%

Run the backtest to populate charts.

Order records

Run the backtest to load every order fill alongside performance.

Export to your platform

Transform Quant Buffet lab code (ASSETS + make_on_day / PortfolioEngine) into native classes for a third-party IDE — then copy and paste.

Target platform

Run in: QuantConnect Cloud or LEAN CLI · QCAlgorithm with Equity securities and monthly rebalance.

Detected pattern: Momentum rotationAssets: GLD, SLV, DBC, GSG, USO, UNG, DBA

# Generated from Quant Buffet → QuantConnect LEAN
# Strategy: 基于成交量、持仓量和过去收益的美国期货周度反转策略
# Detected pattern: Momentum rotation
# Source uses Quant Buffet lab APIs (ASSETS + make_on_day / PortfolioEngine).
# Review fees, data, and risk before live trading — educational export only.

from AlgorithmImports import *


class QuantBuffetExport(QCAlgorithm):
    def Initialize(self):
        self.SetStartDate(2010, 1, 1)
        self.SetCash(100000)
        tickers = ["GLD", "SLV", "DBC", "GSG", "USO", "UNG", "DBA"]
        self.symbols = []
        for t in tickers:
            if "-" in t:  # crypto proxy e.g. BTC-USD
                self.symbols.append(self.AddCrypto(t.replace("-USD", ""), Resolution.Daily).Symbol)
            else:
                self.symbols.append(self.AddEquity(t, Resolution.Daily).Symbol)
        self.Schedule.On(
            self.DateRules.MonthStart(self.symbols[0]),
            self.TimeRules.AfterMarketOpen(self.symbols[0], 30),
            self.Rebalance,
        )
        # Logic: Hold top 1 by 21-day return; monthly.

    def Rebalance(self):
        scores = {}
        for symbol in self.symbols:
            hist = self.History(symbol, 21 + 5, Resolution.Daily)
            if hist.empty: continue
            close = hist["close"]
            if hasattr(close, "unstack"):
                close = close.unstack(level=0).iloc[:, 0]
            if len(close) < 21 + 1: continue
            scores[symbol] = float(close.iloc[-1] / close.iloc[-21 - 1] - 1)
        ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)[:1]
        for symbol in self.symbols:
            self.SetHoldings(symbol, 0)
        if ranked:
            w = 1.0 / len(ranked)
            for symbol, _ in ranked:
                self.SetHoldings(symbol, w)

导出代码使用目标平台的原生类与库。请在第三方 IDE 中安装依赖后运行；实盘前请自行验证。