用 Python 实现趋势过滤的多资产投资策略

欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。我们提供了精选的国内外量化投研的 180+ 篇高质量文章，并每日更新最新研究成果，涵盖策略开发、因子分析、风险管理等核心领域。无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

在金融市场中，有一个众所周知的现象：许多资产的走势与市场基准高度相关。当基准资产处于上涨趋势时，相关资产往往涨得更猛；而在下跌趋势中，它们的跌幅也更为惨烈。

这个现象给我们提供了一个简单而有效的交易思路：用基准资产的趋势来判断市场环境，然后通过资产组合来表达我们的观点。

本文将带你使用 Python 从零实现这个策略，包括数据获取、信号生成、回测分析以及策略优化。无论你是量化交易新手还是有一定基础的开发者，都能从中获得启发。

策略核心思路

整个策略的逻辑非常清晰：

第一步，使用比特币的 30 日和 90 日指数移动平均线（EMA）来定义市场的牛熊状态。当短期均线在长期均线之上时，我们认为市场处于牛市；反之则为熊市。

第二步，构建一个等权重的山寨币投资组合，包含 ETH、BNB、SOL、XRP、ADA 等主流币种。

第三步，在牛市环境下做多山寨币组合，在熊市环境下做空（或者空仓）。

环境准备与数据获取

首先，我们需要安装必要的 Python 库并设置交易所连接：

import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime

# 初始化交易所连接
exchange = ccxt.binanceus({
    "enableRateLimit": True,  


    
# 启用请求频率限制，避免被封禁
})

# 设置数据获取参数
START_ISO = "2020-01-01T00:00:00Z"  # 数据起始日期
TIMEFRAME = "1d"  # 日线级别
LIMIT = 1000  # 每次请求获取的 K 线数量

接下来，我们编写一个通用的数据下载函数：

def fetch_daily_ohlcv(symbol: str,
                      start_iso: str = START_ISO,
                      timeframe: str = TIMEFRAME,
                      limit: int = LIMIT) -> pd.DataFrame:
    """
    获取指定交易对的日线 OHLCV 数据
    
    参数：
        symbol: 交易对名称，如 "BTC/USDT"
        start_iso: 起始时间（ISO 格式）
        timeframe: K 线周期
        limit: 每次请求的 K 线数量
    
    返回：
        包含 OHLCV 数据的 DataFrame
    """
    since = exchange.parse8601(start_iso)
    now = exchange.milliseconds()
    all_ohlcv = []
    
    # 分页获取历史数据
    while since < now:
        ohlcv = exchange.fetch_ohlcv(
            symbol=symbol,
            timeframe=timeframe,
            since=since,
            limit=limit
        )
        if not ohlcv:
            break
        all_ohlcv.extend(ohlcv)
        last_ts = ohlcv[-1][0]
        since = last_ts + 1  # 移动到下一个时间点，避免重复
        
        # 如果已经获取到最新数据，则停止
        if last_ts >= now - 24*60*60*1000:
            break
    
    if not all_ohlcv:
        raise ValueError(f"无法获取 {symbol} 的数据")
    
    # 转换为 DataFrame 格式
    df = pd.DataFrame(
        all_ohlcv,
        columns=["timestamp", "open", "high", "low", "close", "volume"]
    )
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    df = df.set_index("datetime").drop(columns=["timestamp"])
    return df

现在，定义我们的投资标的并下载数据：

# 定义交易标的
btc_symbol = "BTC/USDT"  # 比特币作为趋势参考
alt_symbols = [
    "ETH/USDT",   # 以太坊
    


    
"BNB/USDT",   # 币安币
    "SOL/USDT",   # Solana
    "XRP/USDT",   # 瑞波币
    "ADA/USDT",   # 艾达币
]

# 获取比特币数据
btc_df = fetch_daily_ohlcv(btc_symbol)

# 获取山寨币数据
alt_data = {}
for sym in alt_symbols:
    try:
        print(f"正在获取 {sym} 数据...")
        alt_data[sym] = fetch_daily_ohlcv(sym)
    except Exception as e:
        print(f"获取 {sym} 失败：{e}")

# 提取收盘价并对齐日期
btc_close = btc_df["close"].rename("BTC")
alt_closes = pd.DataFrame(
    {sym: df["close"] for sym, df in alt_data.items()}
)
btc_close, alt_closes = btc_close.align(alt_closes, join="inner")

构建比特币趋势指标

策略的核心是使用双均线系统来判断比特币的趋势状态：

def ema(series, span):
    """计算指数移动平均线"""
    return series.ewm(span=span, adjust=False).mean()

# 构建比特币趋势指标
btc_df = pd.DataFrame({"close": btc_close})
btc_df["ema_short"] = ema(btc_df["close"], span=30)   # 30 日短期均线
btc_df["ema_long"] = ema(btc_df["close"], span=90)    # 90 日长期均线

# 定义市场状态：短期均线在上为牛市（+1），否则为熊市（-1）
btc_df["regime"] = np.where(
    btc_df["ema_short"] > btc_df["ema_long"],
    1,    # 牛市
    -1    # 熊市
)

计算投资组合收益

接下来，我们计算山寨币组合的每日收益率：

# 计算各山寨币的日收益率
alt_rets = alt_closes.pct_change().dropna()

# 等权重组合的日收益率（所有币种收益率的平均值）



    
alt_port_ret = alt_rets.mean(axis=1).rename("alt_port_ret")

# 计算比特币日收益率（用于对比）
btc_ret = btc_close.pct_change().dropna().rename("btc_ret")

# 合并所有数据
combined = pd.concat(
    [btc_df["regime"], btc_ret, alt_port_ret],
    axis=1
).dropna()

策略回测

现在是最激动人心的部分——实现交易策略并计算累计收益：

# 策略收益计算
# 牛市时做多（收益 = +组合收益），熊市时做空（收益 = -组合收益）
combined["strategy_ret"] = combined["regime"] * combined["alt_port_ret"]

# 计算累计收益曲线
combined["alt_port_cum"] = (1 + combined["alt_port_ret"]).cumprod()  # 山寨币持有
combined["btc_cum"] = (1 + combined["btc_ret"]).cumprod()            # 比特币持有
combined["strat_cum"] = (1 + combined["strategy_ret"]).cumprod()     # 策略收益

# 绘制收益曲线对比图
fig, ax = plt.subplots(figsize=(12, 6))
ax.plot(combined.index, combined["btc_cum"], label="比特币持有")
ax.plot(combined.index, combined["alt_port_cum"], label="山寨币组合持有")
ax.plot(combined.index, combined["strat_cum"], label="趋势过滤策略")
ax.set_title("累计收益对比")
ax.legend()
plt.show()

策略绩效评估

为了更全面地评估策略表现，我们编写一个绩效统计函数：

def performance_stats(returns: pd.Series, freq: int = 365):
    """
    计算策略绩效指标
    
    参数：
        returns: 日收益率序列
        freq: 年化系数（加密货币市场为 365 天）
    
    返回：
        包含各项绩效指标的字典
    """
    rets = returns.dropna()
    if len(rets) == 0:
        return None
    
    # 计算总收益
    total_return = (1 + rets).prod() - 1
    n_days = rets.shape[0]
    years = n_days / freq
    
    # 年化收益率（CAGR）
    cagr = (


    
1 + total_return) ** (1 / years) - 1 if years > 0 else np.nan
    
    # 年化波动率
    vol = rets.std() * np.sqrt(freq)
    
    # 夏普比率
    sharpe = rets.mean() / rets.std() * np.sqrt(freq) if rets.std() != 0 else np.nan
    
    # 最大回撤
    cum = (1 + rets).cumprod()
    peak = cum.cummax()
    drawdown = (cum / peak) - 1
    max_dd = drawdown.min()
    
    return {
        "总收益": f"{total_return:.2%}",
        "年化收益": f"{cagr:.2%}",
        "年化波动率": f"{vol:.2%}",
        "夏普比率": f"{sharpe:.2f}",
        "最大回撤": f"{max_dd:.2%}"
    }

# 对比三种策略的绩效
stats_df = pd.DataFrame({
    "比特币持有": performance_stats(combined["btc_ret"]),
    "山寨币组合持有": performance_stats(combined["alt_port_ret"]),
    "趋势过滤策略": performance_stats(combined["strategy_ret"])
})
print(stats_df)

策略优化方向

基础版本的策略虽然简单有效，但在实际应用中还需要考虑以下改进。

第一个改进是采用纯多头版本。考虑到做空山寨币在实际操作中存在诸多限制，我们可以改为：牛市做多，熊市空仓：

# 纯多头版本：牛市做多，熊市空仓（收益为 0）
combined["strategy_ret_long_only"] = np.where(
    combined["regime"] == 1,
    combined["alt_port_ret"],  # 牛市：持有组合
    0.0                         # 熊市：空仓
)
combined["strat_long_only_cum"] = (1 + combined["strategy_ret_long_only"]).cumprod()

第二个改进是增加信号过滤。为了减少震荡市中的频繁交易，可以设置一个缓冲区：

# 计算均线差值


    

spread = btc_df["ema_short"] - btc_df["ema_long"]
threshold = 0.005 * btc_df["close"]  # 价格的 0.5% 作为缓冲区

# 三状态信号：+1 牛市，-1 熊市，0 中性
btc_df["regime3"] = np.where(spread > threshold, 1,
                    np.where(spread < -threshold, -1, 0))

第三个改进是采用风险平价权重。等权重配置忽略了各币种波动率的差异，我们可以使用反波动率加权：

window = 30  # 使用 30 日滚动窗口计算波动率
alt_vol = alt_rets.rolling(window).std()

# 反波动率权重：波动率越低，权重越高
inv_vol = 1 / alt_vol.replace(0, np.nan)
weights = inv_vol.div(inv_vol.sum(axis=1), axis=0)

# 风险平价组合收益
alt_port_ret_rp = (weights * alt_rets).sum(axis=1).dropna()

第四个改进是加入交易成本。真实交易中必须考虑手续费的影响：

# 记录仓位变化
combined["position"] = combined["regime"]
combined["position_shifted"] = combined["position"].shift().fillna(0)

# 标记换仓日期
trade_flag = combined["position"] != combined["position_shifted"]
turnover = trade_flag.astype(float)

# 扣除交易成本（假设单边 0.1%）
fee_rate = 0.001
combined["strategy_ret_after_cost"] = (
    combined["strategy_ret"] - fee_rate * turnover
)

策略的优势与风险

这个策略有几个明显的优势。首先，逻辑简单易懂，用比特币趋势过滤山寨币交易符合市场直觉。其次，通过组合多个山寨币降低了单一币种的特殊风险。最后，策略能够自适应市场环境，在牛市积极进攻，在熊市防守或反向操作。

同时也需要警惕以下风险。均线策略在震荡行情中容易产生频繁的错误信号。做空山寨币在很多交易所受到限制，且需要支付资金费用。等权重配置可能让高波动币种主导整个组合的风险。

总结

本文介绍了一个基于比特币趋势过滤的山寨币投资策略。核心思想是利用比特币的趋势状态来指导山寨币组合的多空方向。通过 Python 实现了完整的数据获取、信号生成、回测分析流程，并探讨了多种优化方向。

这个策略框架非常适合作为量化交易的入门项目，你可以在此基础上进一步扩展：尝试不同的均线参数、加入更多的技术指标、测试不同的币种组合等。

参考文章

加入专注于财经数据与量化投研的知识星球【数据科学实战】，获取本文完整研究解析、代码实现细节。

财经数据与量化投研知识社区

核心权益如下：

赠送《财经数据宝典》完整文档，汇集多年财经数据维护经验
赠送《量化投研宝典》完整文档，汇集多年量化投研领域经验
赠送《PyBroker-入门及实战》视频课程，手把手学习量化策略开发
每日分享高质量量化投研文章（已更新 180+篇）、代码和相关资料
定期更新高频财经数据
参与年度不少于 10 次专属直播与录播课程
与核心开发者直接交流，解决实际问题
获取专业微信群交流机会和课程折扣

星球已有丰富内容积累，包括量化投研论文、财经高频数据、 PyBroker 视频教程、定期直播、数据分享和答疑解难。适合对量化投研和财经数据分析有兴趣的学习者及从业者。欢迎加入我们！

2. 用 Python 打造股票预测系统：Transformer 模型教程（二）

3. 用 Python 打造股票预测系统：Transformer 模型教程（三）

4. 用 Python 打造股票预测系统：Transformer 模型教程（完结）

5. 揭秘隐马尔可夫模型：因子投资的制胜武器

6. YOLO 也能预测股市涨跌？计算机视觉在股票市场预测中的应用

7. 金融 AI 助手：FinGPT 让你轻松掌握市场分析

8. 量化交易秘籍：为什么专业交易员都在用对数收益率？

9. Python 量化投资利器：Ridge、Lasso 和 Elastic Net 回归详解

10. 掌握金融波动率模型：完整 Python 实现指南

好书推荐