用 Python 打造智能投资组合：Butterworth 滤波器实战指南

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

引言

在金融市场中，价格波动剧烈、噪音繁多，如何从杂乱的数据中提取有效的趋势信号，是每个量化交易者面临的核心挑战。今天，我们来介绍一种结合 Butterworth 滤波器、动量因子和市场状态识别的量化投资策略，帮助你构建一个能够动态调仓的投资组合。

这篇文章将带你了解这套策略的核心思路，并附带完整的 Python 代码实现。无论你是量化交易的初学者，还是希望拓展技术栈的 Python 开发者，都能从中获得启发。

策略核心思路

这套策略的核心可以概括为三个关键模块：

趋势检测：使用 Butterworth 低通滤波器平滑价格曲线，过滤掉短期噪音，保留真实的趋势方向。通过快慢两条滤波线的差值来判断短期与长期趋势的关系。

市场状态识别：以 BTC 作为市场风向标，结合 EMA（指数移动平均线）和 ADX（平均趋向指数）来划分牛市、熊市和震荡市。不同市场状态下采用不同的交易策略。

动态资产配置：根据各资产的动量得分进行排名，在牛市做多表现最好的资产，在熊市做空表现最差的资产，同时引入波动率加权和风控机制。

核心代码实现

环境配置与导入依赖

import backtrader as bt  # 回测框架
import ccxt  # 加密货币交易所 API
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from scipy import signal  # 信号处理库，用于 Butterworth 滤波
from collections import deque
from datetime import datetime, timedelta, timezone

# 定义交易标的：10 种主流加密货币
SYMBOLS = [
    'BTC/USDT', 'ETH/USDT', 'BNB/USDT', 'ADA/USDT', 'XRP/USDT',
    'DOGE/USDT', 'SOL/USDT', 'LTC/USDT', 'LINK/USDT', 'MATIC/USDT'
]

TIMEFRAME = '1d'  # 日线级别
YEARS = 5  # 回测 5 年数据
TRADING_DAYS = 365  # 加密货币全年交易

Butterworth 滤波器指标

Butterworth 滤波器是一种信号处理中常用的低通滤波器，它的特点是在通带内频率响应非常平坦，能够有效去除高频噪音，同时保留趋势信息。

class ButterworthFilter(bt.Indicator):
    """
    Butterworth 低通滤波器指标
    用于平滑价格数据，提取趋势信号
    """
    lines = ('filtered',)  # 定义输出线：滤波后的价格
    params = (
        ('cutoff_freq', 0.1),  # 截止频率，控制平滑程度
        ('order', 2),  # 滤波器阶数
        ('lookback', 100)  # 回看周期
    )
    
    def __init__(self):
        self.addminperiod(self.params.lookback)
        # 初始化 Butterworth 滤波器系数
        self.b, self.a = signal.butter(
            N=self.p.order, 
            Wn=self.p.cutoff_freq
        )
        self.data_buffer = deque(maxlen=self.p.lookback)
        self.zi = signal.lfilter_zi(self.b, self.a)  # 滤波器初始状态
    



    
    def next(self):
        price = self.data.close[0]  # 获取当前收盘价
        self.data_buffer.append(price)
        
        # 数据不足时直接返回原始价格
        if len(self.data_buffer) < self.p.lookback:
            self.lines.filtered[0] = price
            return
        
        # 应用滤波器
        filtered, self.zi = signal.lfilter(
            self.b, self.a, [price], zi=self.zi * price
        )
        self.lines.filtered[0] = filtered[0]

市场状态识别

策略以 BTC 为基准判断整体市场处于何种状态，这是因为 BTC 作为加密货币市场的龙头，其走势往往能反映整个市场的情绪。

def _get_regime(self):
    """
    判断市场状态
    返回值：1 表示牛市，-1 表示熊市，0 表示震荡市
    """
    # 牛市条件：短期均线在长期均线之上，且趋势强度足够
    if (self.btc_ema_fast[0] > self.btc_ema_slow[0] and 
        self.btc_adx.adx[0] > 25):
        return 1
    # 熊市条件：短期均线在长期均线之下，且趋势强度足够
    elif (self.btc_ema_fast[0] < self.btc_ema_slow[0] and 
          self.btc_adx.adx[0] > 25):
        return -1
    # 其他情况为震荡市
    return 0

资产评分与配置逻辑

策略使用动量因子和趋势差值来计算每个资产的得分，然后根据市场状态决定做多还是做空。

# 计算每个资产的综合得分
def calculate_score(price, past_price, fast_filter, slow_filter, atr):
    """
    计算资产的综合得分
    
    参数：
        price: 当前价格
        past_price: 30 天前的价格
        fast_filter: 快速滤波器值
        slow_filter: 慢速滤波器值
        atr: 平均真实波幅（衡量波动性）
    """
    # 30 天对数动量
    mom = np.log(price / past_price)
    
    # 趋势差值：快慢滤波器之差
    trend_diff = fast_filter - slow_filter
    
    # 波动率代理指标
    vol_proxy = atr / price
    
    # 综合得分：风险调整后的动量
    score = mom * (trend_diff / atr)



    
    
    return score

# 仓位配置规则
# 牛市：做多得分最高的 3 个资产
# 熊市：做空得分最低的 3 个资产
# 单个资产权重上限：30%
# 采用波动率倒数加权，降低高波动资产的仓位

数据获取

通过 ccxt 库从 Binance US 获取历史 K 线数据。

def fetch_ohlcv_binanceus(symbol, timeframe='1d', years=5):
    """
    从 Binance US 获取历史 K 线数据
    
    参数：
        symbol: 交易对，如 'BTC/USDT'
        timeframe: 时间周期
        years: 获取多少年的数据
    """
    exchange = ccxt.binanceus()  # 初始化交易所对象
    
    # 计算起始时间戳
    since = int(
        (datetime.now(timezone.utc) - timedelta(days=365 * years))
        .timestamp() * 1000
    )
    
    all_ohlcv = []
    while True:
        # 分批获取数据，每次最多 1000 条
        batch = exchange.fetch_ohlcv(
            symbol, timeframe, since=since, limit=1000
        )
        if not batch:
            break
        all_ohlcv.extend(batch)
        if len(batch) 1000:
            break
        since = batch[-1][0] + 1  # 更新起始时间
    
    # 转换为 DataFrame
    df = pd.DataFrame(
        all_ohlcv, 
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms', utc=True)
    
    return df.set_index('timestamp').sort_index()

绩效统计

def compute_perf_stats(equity_series, trading_days=365):
    """
    计算策略绩效指标
    
    参数：
        equity_series: 权益曲线序列
        trading_days: 年化交易天数
    """
    # 计算收益率序列
    rets = equity_series.pct_change().dropna()
    
    


    
# 总收益率
    total_return = equity_series.iloc[-1] / equity_series.iloc[0] - 1
    
    # 年化收益率（CAGR）
    cagr = (1 + total_return) ** (trading_days / len(rets)) - 1
    
    # 年化波动率
    vol_ann = rets.std() * np.sqrt(trading_days)
    
    # 夏普比率
    sharpe = cagr / vol_ann if vol_ann > 0 else np.nan
    
    # 最大回撤
    running_max = equity_series.cummax()
    drawdown = equity_series / running_max - 1.0
    max_dd = drawdown.min()
    
    # 卡尔玛比率
    calmar = cagr / abs(max_dd) if max_dd 0 else np.nan
    
    return rets, drawdown, {
        "Total Return": total_return,
        "CAGR": cagr,
        "Annual Vol": vol_ann,
        "Sharpe": sharpe,
        "Max Drawdown": max_dd,
        "Calmar": calmar
    }