在波诡云谲的期货市场，将交易灵感转化为可执行的量化策略是每个量化交易者的追求。今天，我们要一起探讨如何将B站Z哥在A股市场分享的"3/4阴量线"交易思想，通过代码实现转化为商品期货量化交易策略。这是一次从交易灵感到量化实践的完整尝试，展示了如何用程序化的方式验证和执行交易思想！
灵感来源：B站Z哥的交易智慧
这个交易策略的灵感来源于B站（哔哩哔哩）上颇有名气的"Z哥"。作为一位备受关注的交易分享者，Z哥通过其视频内容向广大交易爱好者分享了这一独特的交易方法。他独到的市场洞察力和系统化的交易思路，为无数交易者提供了宝贵的实战经验和技术分析方法。Z哥不仅详细讲解了3/4阴量线战法的操作细节，还深入阐释了其背后的资金逻辑和市场心理学原理。正是他的启发，让我们得以将这一策略应用到商品期货量化交易的领域，进行更为系统化和自动化的尝试。
神奇的3/4阴量线是什么？
简单来说，3/4阴量线战法是一种专注于判断突破真伪的技术分析方法。其核心理念可以概括为：
当市场出现放量中大阳线突破前期平台（也可以是趋势通道、颈位线）后，如果第二天收盘是一根阴线，且成交量约为前一天的3/4左右，这很可能是一个假突破信号！
这种假突破信号的回测正确率居然高达75%~90%！是不是听起来就很诱人？但别急，先让我们深入了解它的运作机制。
战法的核心思路与使用方法
当我们在日K线图上发现第一天出现了放量中长阳突破形态时，需要特别关注第二天的走势：
🔍 观察第二天是否出现3/4阴量线特征
- 如果没有出现，可以将止损设置在第二天K线最低价下方3~5个价位入场
- 注意：一定不要在突破当天就急着入场！否则可能遇到第二天低开直接被闷杀的尴尬局面
"如果是真突破，一定不会只有一两根阳线，所以不用怕踏空"——这句话道出了交易中的一个重要心态调整，宁可错过，不要错入。
使用注意事项：打好擦边球的关键
在实战中，有几个特别需要注意的点：
-
适用范围有限：3/4阴量线仅适用于判断第一根放量中长阳K线突破的真假，不能用于已形成上涨趋势后的突破判断，也不能作为持股或预测未来走势的依据。
-
阴线判断标准：判断依据是第二天收盘价是否比第一天收盘价低，而非简单看K线图的颜色。
-
成交量比例灵活：成交量比例不必严格是3/4，可在1/2到4/5之间波动，理想情况是1/2以下的极度缩量。
-
仅适用于A股：在港股和美股回测中并无显著优势。
-
设置止损是王道：这个战法不是100%成功，所以一定要设好止损并严格执行！
在优宽量化平台上的代码实现
将这个战法搬到优宽量化平台上，我们编写了一个名为ThreeQuarterReverseStrategy的策略类。让我们来看看它的几个核心部分：
策略初始化
策略通过设置一系列参数来捕捉3/4阴量线形态：
- 成交量比例范围设置在0.5到0.8之间（即1/2到4/5）
- 突破日成交量至少要比之前10天平均成交量增长20%
- K线实体长度至少是近期平均实体长度的0.6倍（定义中长阳）
状态机设计
策略使用状态机来管理不同的交易阶段：
SCANNING：扫描新的突破形态FOUND_BREAKOUT：找到突破形态，等待确认CONFIRMED_FALSE：确认为假突破，准备执行空头交易CONFIRMED_TRUE：确认为真突破，准备执行多头交易IN_TRADE：已入场交易，管理止盈止损
这种状态机设计使策略逻辑清晰，便于跟踪和管理交易过程。
动态止盈止损机制
策略采用基于ATR（真实波动幅度均值）的动态止盈止损机制：
- 止损设置为ATR的2倍
- 止盈设置为ATR的3倍
- 当盈利达到ATR的1.5倍时，启动移动止损机制
- 移动止损步进为ATR的0.5倍
这种动态机制能够很好地适应市场波动，既保护利润又给予价格足够的呼吸空间。
实战体验：商品期货市场的表现
将这个起源于股票市场的策略应用到商品期货市场，结果颇为有趣。我们以螺纹钢(rb888)为例进行了回测：
在大宗商品的波动中，3/4阴量线战法展现出了不同于股市的特点。由于期货市场的杠杆特性和流动性差异，我们发现：
-
反应更为迅速：假突破信号出现后，价格往往在当天就有明显反应，不像股市可能需要几天时间。
-
波动幅度更大：期货市场的波动幅度往往更大，同样的策略在设置止损时需要留出更多空间。
-
日内交易与隔夜风险：由于期货市场存在隔夜风险，策略需要更加谨慎地处理仓位管理。
实际测试结果
经过实践测试，我们发现针对螺纹钢主力合约，以天为周期时，策略的信号较难满足条件。这可能是因为螺纹钢的走势相对平稳，不像股票市场那样容易出现明显的突破形态。
当我们将时间框架调整为小时级别后，情况有所改善，但在一年的测试时间内，仍然只有少数几次开仓机会。尽管如此，这些有限的交易机会仍然为我们带来了一定的收益。这也符合量化交易的一个重要原则：宁可少交易，也要确保交易质量。
值得注意的是，不同的期货品种具有不同的波动特性，建议大家可以根据自己关注和熟悉的品种进行更多尝试，可能会发现更适合此策略的交易标的。
底层逻辑解析：为什么它有效？
为什么3/4阴量线能够有效地判断假突破？这与主力资金的运作逻辑密切相关：
真正的突破需要主力已经完成高度控盘，并且要一鼓作气地涨上去。如果出现了3/4阴量线特征，说明主力可能有出货嫌疑，因为在高度控盘的情况下，不会出现这么大的成交量，散户是交易不出来这么大量的。
想象一下，主力花了大价钱解放前期所有套牢盘，已经拿到大量筹码，如果真想继续往上做，绝不可能在突破位置停留，让其他人来抢筹码。正如一句经典的交易格言："真突破必定气贯长虹，假突破总是踌躇不前。"
实践中的难点与解决方案
在实际应用中，3/4阴量线战法最大的难点在于确定前一天的突破阳线。市场中并非所有的形态都像教科书一样清晰可辨：
- 箱体震荡、楔形整理、旗形整理等比较规则的走势容易判断
- 波段图形不规则的情况下，很难找到明确的前期高点
解决方案：只做自己看得懂的图形！这是交易中一个朴素但极为重要的原则。面对不确定的形态，宁可不交易，也不要强行套用。
优化与展望
针对商品期货市场的特性，我们对原始策略进行了一些优化：
-
引入ATR动态止损：考虑到期货市场的波动特性，使用ATR来动态调整止损位。
-
移动止损机制：当盈利达到特定阈值后，启动移动止损，锁定部分利润。
-
时间过滤器：增加了时间过滤条件，如果超过一定时间未确认突破真伪，则重置状态。
进一步优化方向
后续策略还有多个可能的优化方向：
-
双向突破策略：目前策略主要针对做多突破进行反向操作，未来可以从做空突破出发，同样尝试反向操作的可行性。这将使策略更加全面，能够适应更多的市场环境。
-
周期适配：根据实践结果，可以尝试更多的时间周期组合，找到最适合各个品种的时间框架。
-
多品种测试：对更多期货品种进行测试，找出哪些品种更适合使用此策略。波动性更强的品种可能会产生更多的交易信号。
-
参数优化：对成交量比例、突破确认时间等关键参数进行更精细的优化，提高策略的适应性。
-
结合其他指标：可以考虑结合其他技术指标，如RSI、MACD等，增加信号的可靠性。
未来，我们计划进一步优化策略，考虑引入更多的市场情绪指标和波动率过滤器，以提高策略在不同市场环境下的适应性。
结语：从交易灵感到量化实践的价值
3/4阴量线战法从B站Z哥的交易灵感到优宽量化平台上的代码实现，正是体现了量化交易的核心价值——将模糊的交易经验转化为精确的计算机指令，用代码实现交易智慧的继承与发展。
这种量化实践的过程远比寻找所谓的"交易明珠"更有价值。通过编程，我们将交易思想系统化、规则化，不仅降低了情绪干扰，还能通过回测验证策略的有效性，让交易决策建立在数据和逻辑的基础上，而非直觉和猜测。
风险控制和纪律执行在量化系统中被程序化，代码中反复强调的止损设置，正是将风险管理从理念落实到实践的体现。交易成功的关键不是预测市场走向，而是如何通过代码将风险管理系统化、自动化。
在这个数据主导的时代，将交易灵感转化为量化实践，是每一个交易者值得思考和尝试的方向。愿你在量化交易的道路上，不断探索，不断优化，将更多交易智慧转化为可执行、可验证的代码实践！
'''backtest
start: 2024-04-25 18:40:00
end: 2025-04-08 00:00:00
period: 1m
basePeriod: 1m
exchanges: [{"eid":"Futures_CTP","currency":"FUTURES"}]
'''
import numpy as np
class ThreeQuarterReverseStrategy:
def __init__(self):
# 策略参数
self.instrument = "rb888" # 交易品种，可根据实际需要修改
self.volume_ratio_min = 0.5 # 成交量比例下限
self.volume_ratio_max = 0.8 # 成交量比例上限，理想值0.75(即3/4)
self.volume_increase_ratio = 1.2 # 突破日成交量相比之前N日平均成交量的增长比例
self.previous_days = 10 # 计算平均成交量的天数
self.min_candle_length_ratio = 0.6 # 至少是近期平均实体长度的倍数，定义中长阳
self.position_ratio = 0.3 # 仓位比例
# 动态止盈止损参数
self.atr_period = 14 # ATR计算周期
self.stop_loss_atr_multiple = 2.0 # 止损为ATR的倍数
self.take_profit_atr_multiple = 3.0 # 止盈为ATR的倍数
self.trailing_stop_activation = 1.5 # 移动止损激活阈值（盈利达到ATR的倍数）
self.trailing_stop_step = 0.5 # 移动止损步进（ATR的倍数）
# 状态变量
self.state = "SCANNING" # 策略状态: SCANNING, FOUND_BREAKOUT, CONFIRMED_FALSE, CONFIRMED_TRUE, IN_TRADE
self.days_since_signal = 0 # 自信号产生后的天数计数
self.breakout_price = 0 # 突破日收盘价
self.breakout_volume = 0 # 突破日成交量
self.breakout_time = 0 # 突破日时间
self.entry_price = 0 # 记录入场价格
self.stop_loss = 0 # 记录止损价格
self.take_profit = 0 # 记录止盈价格
self.position_type = "" # 持仓类型："long" 或 "short"
self.max_favorable_excursion = 0 # 最大有利偏移（用于跟踪止损）
self.atr_value = 0 # 当前ATR值
# 日志前缀
self.log_prefix = "[3/4阴量线反向]"
def OnBar(self):
"""每个K线周期事件"""
# 获取K线数据
bars = exchange.GetRecords(self.instrument, PERIOD_H1)
if len(bars) < 30: # 确保有足够的历史数据
return
# 提取数据
close = np.array([bar.Close for bar in bars])
high = np.array([bar.High for bar in bars])
low = np.array([bar.Low for bar in bars])
open_price = np.array([bar.Open for bar in bars])
volume = np.array([bar.Volume for bar in bars])
time = np.array([bar.Time for bar in bars])
# 计算ATR
self.calculate_atr(high, low, close)
# 获取当前持仓
position = exchange.GetPositions(self.instrument)
long_position = 0
short_position = 0
if position:
for pos in position:
if pos.Type == 0: # 多头持仓
long_position += pos.Amount
elif pos.Type == 1: # 空头持仓
short_position += pos.Amount
# 获取当前价格
current_price = bars[-1].Close
# 状态机处理
if self.state == "IN_TRADE":
# 已经入场交易，检查止盈止损
self.manage_open_position(current_price, long_position, short_position)
elif self.state == "FOUND_BREAKOUT":
# 找到突破形态，检查第二天是否是假突破
self.days_since_signal += 1
# 检查时间条件：当突破日是time[-3]时，检查time[-2]是否出现假突破
if len(time) >= 3 and self.breakout_time == time[-3]:
# 检查是否为假突破
is_false = self.check_false_breakout(close[-2], open_price[-2], volume[-2])
if is_false:
# 确认为假突破，准备执行空头交易
Log(f"{self.log_prefix} 确认为假突破，准备执行空头交易")
else:
# 不是假突破，准备执行多头交易
Log(f"{self.log_prefix} 确认为真突破，准备执行多头交易")
self.state = "CONFIRMED_TRUE" # 更新状态为确认真突破
elif self.days_since_signal > 5:
# 超过5天未确认假突破，重置状态
self.reset_state()
elif self.state == "CONFIRMED_FALSE":
# 已确认为假突破，执行反向交易
self.execute_reverse_trade(current_price, short_position)
elif self.state == "CONFIRMED_TRUE":
# 已确认为真突破，执行多头交易
self.execute_long_trade(current_price, long_position)
elif self.state == "SCANNING":
# 扫描新的突破形态
self.scan_breakout(bars, close, high, open_price, volume, time)
def calculate_atr(self, high, low, close):
"""计算ATR值"""
if len(close) >= self.atr_period:
tr1 = np.max(np.vstack((high[1:] - low[1:],
np.abs(high[1:] - close[:-1]),
np.abs(low[1:] - close[:-1]))), axis=0)
self.atr_value = np.mean(tr1[-self.atr_period:])
else:
self.atr_value = (high[-1] - low[-1]) * 0.1 # 默认值
def manage_open_position(self, current_price, long_position, short_position):
"""管理已开仓位的止盈止损"""
# 更新动态止损位
self.update_dynamic_stops(current_price)
# 检查止盈止损条件
if self.position_type == "long" and long_position > 0:
# 多头止损检查
if current_price <= self.stop_loss:
exchange.CreateOrder(self.instrument, "closebuy", -1, long_position)
Log(f"{self.log_prefix} 多头止损：当前价格：{current_price}，止损价格：{self.stop_loss}")
self.reset_state()
# 多头止盈检查
elif current_price >= self.take_profit:
exchange.CreateOrder(self.instrument, "closebuy", -1, long_position)
Log(f"{self.log_prefix} 多头止盈：当前价格：{current_price}，止盈价格：{self.take_profit}")
self.reset_state()
elif self.position_type == "short" and short_position > 0:
# 空头止损检查
if current_price >= self.stop_loss:
exchange.CreateOrder(self.instrument, "closesell", current_price, short_position)
Log(f"{self.log_prefix} 空头止损：当前价格：{current_price}，止损价格：{self.stop_loss}")
self.reset_state()
# 空头止盈检查
elif current_price <= self.take_profit:
exchange.CreateOrder(self.instrument, "closesell", current_price, short_position)
Log(f"{self.log_prefix} 空头止盈：当前价格：{current_price}，止盈价格：{self.take_profit}")
self.reset_state()
def update_dynamic_stops(self, current_price):
"""更新动态止盈止损"""
if self.state != "IN_TRADE":
return
# 多头仓位的处理
if self.position_type == "long":
# 更新最大有利偏移
if current_price > self.entry_price:
profit_points = current_price - self.entry_price
if profit_points > self.max_favorable_excursion:
self.max_favorable_excursion = profit_points
# 如果盈利超过激活阈值，启动移动止损
if profit_points >= self.trailing_stop_activation * self.atr_value:
new_stop = current_price - self.trailing_stop_step * self.atr_value
# 确保移动止损只会上移，不会下移
if new_stop > self.stop_loss:
self.stop_loss = new_stop
Log(f"{self.log_prefix} 更新多头移动止损：{self.stop_loss}")
# 空头仓位的处理
elif self.position_type == "short":
# 更新最大有利偏移
if current_price < self.entry_price:
profit_points = self.entry_price - current_price
if profit_points > self.max_favorable_excursion:
self.max_favorable_excursion = profit_points
# 如果盈利超过激活阈值，启动移动止损
if profit_points >= self.trailing_stop_activation * self.atr_value:
new_stop = current_price + self.trailing_stop_step * self.atr_value
# 确保移动止损只会下移，不会上移
if new_stop < self.stop_loss or self.stop_loss == 0:
self.stop_loss = new_stop
Log(f"{self.log_prefix} 更新空头移动止损：{self.stop_loss}")
def reset_state(self):
"""重置策略状态"""
self.state = "SCANNING"
self.days_since_signal = 0
self.breakout_price = 0
self.breakout_volume = 0
self.breakout_time = 0
self.entry_price = 0
self.stop_loss = 0
self.take_profit = 0
self.position_type = ""
self.max_favorable_excursion = 0
def scan_breakout(self, bars, close, high, open_price, volume, time):
"""扫描突破形态"""
if len(bars) < 20:
return False
# 检查最新完成的K线（即倒数第二根，因为最后一根可能还在形成中）
yesterday_close = close[-2]
yesterday_open = open_price[-2]
yesterday_volume = volume[-2]
# 计算过去N日的平均成交量和平均K线实体长度
avg_volume = np.mean(volume[-self.previous_days-2:-2])
body_lengths = abs(close[-self.previous_days-2:-2] - open_price[-self.previous_days-2:-2])
avg_body_length = np.mean(body_lengths)
# 判断是否为放量中长阳
is_long_bullish = yesterday_close > yesterday_open and \
(yesterday_close - yesterday_open) > self.min_candle_length_ratio * avg_body_length
is_volume_increase = yesterday_volume > self.volume_increase_ratio * avg_volume
# 检查是否突破前期平台
# 使用前20天的最高价作为参考点
previous_high = np.max(high[-22:-2])
is_breakout = yesterday_close > previous_high
# 如果满足突破条件，记录这个突破日
if is_long_bullish and is_volume_increase and is_breakout:
self.state = "FOUND_BREAKOUT"
self.breakout_time = time[-2] # 记录突破日时间
self.days_since_signal = 0
self.breakout_price = yesterday_close
self.breakout_volume = yesterday_volume
Log(f"{self.log_prefix} 发现突破形态：突破价格：{yesterday_close}，前期高点：{previous_high}，突破时间：{self.breakout_time}")
return True
return False
def check_false_breakout(self, need_close, need_open, need_volume):
"""检查是否是假突破"""
if self.state != "FOUND_BREAKOUT":
return False
# 检查当前K线是否收阴（收盘价低于突破日收盘价）
is_bearish = need_close < need_open
# 检查当前K线成交量是否在突破日的50%-80%之间
volume_ratio = need_volume / self.breakout_volume
is_volume_reduced = self.volume_ratio_min <= volume_ratio <= self.volume_ratio_max
# 如果确认为假突破，转入反向交易准备状态
if is_bearish and is_volume_reduced:
self.state = "CONFIRMED_FALSE"
Log(f"{self.log_prefix} 确认假突破：成交量比例：{volume_ratio:.2f}，准备执行反向（空头）交易")
return True
return False
def execute_long_trade(self, current_price, current_long_position):
"""执行多头交易"""
if self.state != "CONFIRMED_TRUE":
return False
# 设置动态止盈止损
self.stop_loss = current_price - self.stop_loss_atr_multiple * self.atr_value
self.take_profit = current_price + self.take_profit_atr_multiple * self.atr_value
# 计算可买入的手数（这里简化为1手，实际应根据资金比例计算）
lots_to_buy = 1
if lots_to_buy > 0 and current_long_position == 0:
# 执行多头开仓
exchange.CreateOrder(self.instrument, "buy", -1, lots_to_buy)
Log(f"{self.log_prefix} 多头开仓：数量：{lots_to_buy}手，价格：{current_price}，止损：{self.stop_loss}，止盈：{self.take_profit}")
# 记录交易状态
self.entry_price = current_price
self.position_type = "long"
self.state = "IN_TRADE"
self.max_favorable_excursion = 0
return True
return False
def execute_reverse_trade(self, current_price, current_short_position):
"""执行反向交易"""
if self.state != "CONFIRMED_FALSE":
return False
# 设置动态止盈止损
self.stop_loss = current_price + self.stop_loss_atr_multiple * self.atr_value
self.take_profit = current_price - self.take_profit_atr_multiple * self.atr_value
# 计算可卖出的手数（这里简化为1手，实际应根据资金比例计算）
lots_to_sell = 1
if lots_to_sell > 0 and current_short_position == 0:
# 执行空头开仓
exchange.CreateOrder(self.instrument, "sell", -1, lots_to_sell)
Log(f"{self.log_prefix} 空头开仓：数量：{lots_to_sell}手，价格：{current_price}，止损：{self.stop_loss}，止盈：{self.take_profit}")
# 记录交易状态
self.entry_price = current_price
self.position_type = "short"
self.state = "IN_TRADE"
self.max_favorable_excursion = 0
return True
return False
# 初始化策略
strategy = ThreeQuarterReverseStrategy()
def OnBar():
strategy.OnBar()
# 用于直接运行测试
def main():
while True:
strategy.OnBar()
Sleep(1000 * 60 * 5) # 模拟时间间隔