欢迎加入专注于财经数据与量化投研的【数据科学实战】知识星球！在这里，您将获取持续更新的《财经数据宝典》和《量化投研宝典》，这两部宝典相辅相成，为您在量化投研道路上提供明确指引。 《量化投研宝典》精选了业内持续维护且实用性强的开源工具（Backtrader、Qlib、VeighNa等），配合详细教程与代码示例，帮助您快速构建量化策略；《财经数据宝典》则汇集了多年财经数据维护经验，全面介绍从 AKShare、Tushare 到 Wind、iFind 等国内外数据源，并附有丰富的使用技巧。 无论您是量化投资新手还是经验丰富的研究者，星球社区都能帮您少走弯路，事半功倍，共同探索数据驱动的投资世界！

引言

在量化交易的世界里，如何避免假突破、减少无效交易一直是困扰交易者的难题。今天给大家介绍一个结合了趋势强度、机构资金流向和波动率自适应出场的策略框架。这个策略通过 ADX 趋势强度指标和 VWAP 成交量加权均价的双重过滤，只在市场趋势明确且机构资金配合的时候才入场，大大提高了交易的成功率。

策略核心思想

这个策略的精髓在于"三重验证"机制，只有当以下三个条件同时满足时才会触发交易：

1. 趋势强度验证：ADX 指标必须超过设定阈值（默认 25），确保市场处于强趋势中
2. 市场波动验证：布林带宽度必须显示足够的波动性扩张，避免在横盘震荡中交易
3. 资金流向验证：价格必须与 VWAP 保持正确的相对位置，确保与机构资金流向一致

信号构建详解

1. ADX 和方向指标

ADX（Average Directional Index）用于衡量市场是否处于强趋势状态：

• ADX 值高于 25 表示趋势强劲
• +DI 和 -DI 识别买方或卖方主导
• 只有当 ADX > 阈值时，系统才考虑交易

2. 布林带宽度过滤

布林带的作用是识别市场波动性：

• 窄带表示压缩的低能量状态
• 策略要求最小相对带宽才能入场
• 这是一个波动率门槛：只在有"空间"进行价格扩张时才交易

3. VWAP 资金流过滤器

VWAP 代表机构资金的平均成交价位：

• 做多交易只在价格高于 VWAP 时发生
• 做空交易只在价格低于 VWAP 时发生
• 这确保交易不会逆着大资金的方向操作

完整策略代码实现

import backtrader as bt
import datetime
import yfinance as yf
import pandas as pd
import matplotlib.pyplot as plt

class ADXTrendStrengthWithVWAPFilter(bt.Strategy):
    params = (
        ('adx_period', 14),           # ADX 计算周期
        ('adx_threshold', 25),         # ADX 阈值，高于此值才考虑交易
        ('boll_period', 20),           # 布林带周期
        ('boll_devfactor', 2),         # 布林带标准差倍数
        ('confirmation_bars', 3),      # 信号确认所需的连续 K 线数
        ('trail_percent', 0.02),       # 跟踪止损百分比（2%）
        ('vwap_period', 50),           # VWAP 计算周期
        ('printlog', True),
    )
    
    def log(self, txt, dt=None, doprint=False):
        """日志输出函数"""
        if self.params.printlog or doprint:
            dt = dt or self.datas[0].datetime.datetime(0)
            print(f"{dt.isoformat()} - {txt}")
    
    def __init__(self):
        self.dataclose = self.datas[0].close
        
        # 初始化 ADX、+DI 和 -DI 指标
        self.adx = bt.indicators.ADX(self.datas[0], period=self.params.adx_period)
        self.plusdi = bt.indicators.PlusDI(self.datas[0], period=self.params.adx_period)
        self.minusdi = bt.indicators.MinusDI(self.datas[0], period=self.params.adx_period)
        
        # 布林带用于衡量市场波动范围
        self.boll = bt.indicators.BollingerBands(
            self.datas[0],
            period=self.params.boll_period,
            devfactor=self.params.boll_devfactor
        )
        
        # 自定义 VWAP 计算：VWAP = Sum(典型价格 * 成交量) / Sum(成交量)
        self.typical_price = (self.datas[0].high + self.datas[0].low + self.datas[0].close) / 3
        self.vwap = bt.indicators.SumN(self.typical_price * self.datas[0].volume, period=self.params.vwap_period) / \
                    bt.indicators.SumN(self.datas[0].volume, period=self.params.vwap_period)
        
        # 反转信号确认计数器
        self.reversal_counter = 0
        
        # 跟踪市场订单和跟踪止损订单
        self.order = None
        self.trail_order = None
    
    def notify_order(self, order):
        """订单状态通知"""
        if order.status in [order.Submitted, order.Accepted]:
            return
            
        if order.status in [order.Completed]:
            if order.isbuy():
                self.log(f"买入执行，价格：{order.executed.price:.2f}")
            elif order.issell():
                self.log(f"卖出执行，价格：{order.executed.price:.2f}")
            self.bar_executed = len(self)
            
        elif order.status in [order.Canceled, order.Margin, order.Rejected]:
            self.log(f"订单取消/保证金不足/拒绝：{order.getstatusname()}")
        
        # 重置订单引用
        if order == self.order:
            self.order = None
        if order == self.trail_order:
            self.trail_order = None
    
    def notify_trade(self, trade):
        """交易结果通知"""
        ifnot trade.isclosed:
            return
        self.log(f"交易利润：毛利 {trade.pnl:.2f}，净利 {trade.pnlcomm:.2f}")
    
    def cancel_trail(self):
        """取消跟踪止损订单"""
        if self.trail_order:
            self.log("取消活跃的跟踪止损订单")
            self.cancel(self.trail_order)
            self.trail_order = None
    
    def next(self):
        """主策略逻辑"""
        # 如果有待处理订单，跳过
        if self.order:
            return
        
        # 如果有持仓，确保跟踪止损订单处于活跃状态
        if self.position:
            ifnot self.trail_order:
                if self.position.size > 0:
                    self.log(f"为多头仓位设置跟踪止损订单")
                    self.trail_order = self.sell(
                        exectype=bt.Order.StopTrail,
                        trailpercent=self.params.trail_percent
                    )
                elif self.position.size < 0:
                    self.log(f"为空头仓位设置跟踪止损订单")
                    self.trail_order = self.buy(
                        exectype=bt.Order.StopTrail,
                        trailpercent=self.params.trail_percent
                    )
            return
        
        # 确保有足够的数据
        if len(self) < max(self.params.adx_period, self.params.boll_period, self.params.vwap_period):
            return
        
        # 检查市场是否通过 ADX 处于强趋势
        if self.adx[0] < self.params.adx_threshold:
            self.log(f"ADX 值较低（{self.adx[0]:.2f}），市场趋势较弱，跳过")
            self.reversal_counter = 0
            return
        
        # 使用布林带宽度检查是否为区间震荡市场
        boll_width = self.boll.top[0] - self.boll.bot[0]
        if boll_width < 0.01 * self.dataclose[0]:  # 阈值示例：价格的 1%
            self.log(f"布林带较窄（{boll_width:.2f}），市场处于震荡区间")
            self.reversal_counter = 0
            return
        
        # 定义方向信号
        long_signal = self.plusdi[0] > self.minusdi[0]
        short_signal = self.minusdi[0] > self.plusdi[0]
        
        # 确认方向信号持续多个 K 线
        if (long_signal and self.position.size <= 0) or (short_signal and self.position.size >= 0):
            self.reversal_counter += 1
        else:
            self.reversal_counter = 0
        
        # 检查确认条件是否满足
        if self.reversal_counter < self.params.confirmation_bars:
            return# 等待更多确认
        
        # VWAP 过滤器：只在机构资金方向交易
        price_above_vwap = self.dataclose[0] > self.vwap[0]
        price_below_vwap = self.dataclose[0] < self.vwap[0]
        
        # 过滤多头信号：只在价格高于 VWAP 时
        if long_signal andnot price_above_vwap:
            self.log(f"多头信号被过滤 - 价格（{self.dataclose[0]:.2f}）低于 VWAP（{self.vwap[0]:.2f}）")
            self.reversal_counter = 0
            return
        
        # 过滤空头信号：只在价格低于 VWAP 时
        if short_signal andnot price_below_vwap:
            self.log(f"空头信号被过滤 - 价格（{self.dataclose[0]:.2f}）高于 VWAP（{self.vwap[0]:.2f}）")
            self.reversal_counter = 0
            return
        
        # 如果满足反转条件，首先取消现有的跟踪止损
        self.cancel_trail()
        
        # 当确认条件满足且 VWAP 过滤通过时执行订单
        if long_signal and price_above_vwap:
            if self.position and self.position.size < 0:
                self.log(f"反转做多，价格：{self.dataclose[0]:.2f}（高于 VWAP）")
                self.order = self.buy()  # 反转仓位（买入平空）
            elifnot self.position:
                self.log(f"开多仓，价格：{self.dataclose[0]:.2f}（高于 VWAP）")
                self.order = self.buy()
                
        elif short_signal and price_below_vwap:
            if self.position and self.position.size > 0:
                self.log(f"反转做空，价格：{self.dataclose[0]:.2f}（低于 VWAP）")
                self.order = self.sell()  # 反转仓位（卖出平多）
            elifnot self.position:
                self.log(f"开空仓，价格：{self.dataclose[0]:.2f}（低于 VWAP）")
                self.order = self.sell()
    
    def stop(self):
        """策略结束时的处理"""
        self.log(f"最终组合价值：{self.broker.getvalue():.2f}", doprint=True)

实战案例分析

让我们通过一个具体的例子来理解这个策略的运作：

假设我们在交易比特币（BTC/USDT）：

1. 第 1 步：ADX 指标显示为 28（> 25），表明市场处于强趋势
2. 第 2 步：布林带宽度为价格的 2.5%，表明有足够的波动空间
3. 第 3 步：+DI（30）> -DI（15），显示买方占优
4. 第 4 步：价格为 45000 USDT，VWAP 为 44500 USDT，价格在 VWAP 上方
5. 第 5 步：连续 3 根 K 线满足上述条件，触发买入信号
6. 第 6 步：系统以市价买入，同时设置 2% 的跟踪止损

当价格上涨到 46000 USDT 时，跟踪止损自动上移到 45080 USDT（46000 * 0.98）。如果价格回落触及止损价，系统自动平仓锁定利润。

策略优势与局限

优势

1. 多层确认机制：没有单一指标主导，交易入场需要趋势、波动率和机构偏好的一致认同
2. 自适应保护：跟踪止损根据价格动态调整，而不是依赖固定目标
3. 资金流敏感性：通过尊重 VWAP，策略有效地"与鲸鱼站在一边"
4. 跨市场适用：可扩展到各种资产和时间框架（加密货币、股票、外汇）

局限性

1. 在长期横盘市场中，系统可能长时间处于空闲状态
2. 如果在低成交量条件下 VWAP 失真，可能会过滤掉有效信号
3. 跳空缺口或剧烈的新闻事件仍可能触发意外的止损

使用场景

这个策略特别适合以下类型的交易者：

• 波段交易者：寻找强劲方向性走势的中期交易机会
• 日内交易者：希望通过 VWAP 获得机构资金配合的短线交易
• 组合策略师：需要趋势过滤组件来与均值回归系统混合使用

总结

ADX + VWAP 双重过滤策略是一个纪律严明的突破执行器。它只在趋势、波动率和机构资金流汇聚时才采取行动，然后通过自动跟踪止损锁定利润。这种严格的入场条件和灵活的出场机制，使其非常适合作为多元化系统组合中的核心趋势引擎。

对于 Python 量化交易的学习者来说，这个策略展示了如何将多个技术指标有机结合，形成一个完整的交易系统。通过 Backtrader 框架的实现，我们可以轻松地对策略进行回测和优化，找到最适合自己交易风格的参数组合。

参考文章

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

核心权益如下：

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

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