外汇 · 全球 · 期货，远期 · 每月 Free

全球货币动量策略：做多前3、做空后3

登录后收藏

学术论文

论文链接http://globalmarkets.db.com/new/docs/dbCurrencyReturns_March2009.pdf

原文论文截图

导读

投资标的：SPY、EFA、EWJ、GLD、TLT。再平衡：每月。Universe: SPY, EFA, EWJ, GLD, TLT. Parameters: rebalance=monthly.

策略概要

该策略投资于10–20种货币。做多过去12个月相对美元动量最强的3种货币，做空动量最弱的3种；未配置现金投放隔夜利率，并按月再平衡以跟踪表现。

经济逻辑

货币动量来自各国货币政策与风险溢价的持续分化。投资者对利率与宏观信息的反应往往先滞后、后过度修正，使相对美元的汇率趋势在数月内延续。按月再平衡可动态跟踪动量轮动，隔夜利率则消化闲置资金，降低汇率波动下的现金拖累。

入场与出场

入场

每月按 126 日收益排序 SPY, EFA, EWJ, GLD, TLT，等权持有最强的 1 只。

出场

跌出前 1 即卖出，换入新的领涨标的。

所需行情数据

需要 SPY, EFA, EWJ, GLD, TLT 至少 126 根复权收盘价，才能形成第一个有效信号。ETF 仅交易日有行情。实验室通过 backtest.data.load_daily_prices 读取 Yahoo Finance。

标的

SPY, EFA, EWJ, GLD, TLT

频率

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

所需历史

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

字段

复权收盘价（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 全球货币动量策略：做多前3、做空后3 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 — 全球货币动量策略：做多前3、做空后3
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 = ['SPY', 'EFA', 'EWJ', 'GLD', 'TLT']
LOOKBACK = 126
TOP_N = 1
INVERT = False  # 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 — 全球货币动量策略：做多前3、做空后3
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 = ['SPY', 'EFA', 'EWJ', 'GLD', 'TLT']
LOOKBACK = 126
TOP_N = 1
INVERT = False  # 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

3.30%

Sharpe

0.27

Max DD

-51.95%

Vol

18.24%

Sortino

0.42

Beta

0.26

Up days

55%

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: SPY, EFA, EWJ, GLD, TLT

# Generated from Quant Buffet → QuantConnect LEAN
# Strategy: 全球货币动量策略：做多前3、做空后3
# 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 = ["SPY", "EFA", "EWJ", "GLD", "TLT"]
        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 126-day return; monthly.

    def Rebalance(self):
        scores = {}
        for symbol in self.symbols:
            hist = self.History(symbol, 126 + 5, Resolution.Daily)
            if hist.empty: continue
            close = hist["close"]
            if hasattr(close, "unstack"):
                close = close.unstack(level=0).iloc[:, 0]
            if len(close) < 126 + 1: continue
            scores[symbol] = float(close.iloc[-1] / close.iloc[-126 - 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 中安装依赖后运行；实盘前请自行验证。