# src/strategies/ReversalVolatilityStrategy.py import numpy as np import talib from typing import Optional, Dict, Any, List, Tuple from src.core_data import Bar, Order from src.indicators.indicators import StochasticOscillator, RSI from src.strategies.base_strategy import Strategy # def calculate_atr(bars: List[Bar], period: int) -> float: # """ # 从Bar对象列表中计算平均真实波幅(ATR)。 # """ # if len(bars) < period + 1: # return 0.0 # true_ranges = [] # for i in range(1, len(bars)): # high_low = bars[i].high - bars[i].low # high_prev_close = abs(bars[i].high - bars[i - 1].close) # low_prev_close = abs(bars[i].low - bars[i - 1].close) # tr = max(high_low, high_prev_close, low_prev_close) # true_ranges.append(tr) # return np.mean(true_ranges[-(period):]) def find_fvg(bars: List[Bar]) -> Optional[Tuple[str, float, float]]: """ 使用最近的三根K线识别公允价值缺口(FVG)。 返回: 一个元组 (方向, 上沿价格, 下沿价格) 或 None (如果没有找到FVG)。 'bullish': K线1的最高价和K线3的最低价之间的缺口。 'bearish': K线1的最低价和K线3的最高价之间的缺口。 """ if len(bars) < 3: return None bar1, bar2, bar3 = bars[-3], bars[-2], bars[-1] # 检查看涨FVG (向上的失衡) if bar1.high < bar3.low: # 缺口本身就意味着中间的K线是强劲的 return ('bullish', bar3.low, bar1.high) # 检查看跌FVG (向下的失衡) if bar1.low > bar3.high: return ('bearish', bar1.low, bar3.high) return None class ReversalVolatilityStrategy(Strategy): """ 一个反转波动率策略，该策略在价格回调至公允价值缺口(FVG)时入场， 并使用基于ATR的两阶段止盈系统，其中第二阶段为ATR回撤止盈。 """ def __init__( self, context: Any, main_symbol: str, enable_log: bool, trade_volume: int, atr_period: int = 14, stop_loss_atr_multiplier: float = 1.0, partial_profit_atr_multiplier: float | List[float] = 2.0, # full_profit_atr_multiplier: float = 4.0, # 移除固定止盈乘数 trailing_stop_atr_multiplier: float = 1.0, # 新增：回撤止盈ATR乘数 open_atr_multiplier: float | List[float] = 1.0, order_direction=None, indicators=[None, None], ): """ Args: context: 回测上下文。 symbol (str): 主要交易的合约代码。 trade_volume (int): 每笔交易的总手数。必须是偶数。 atr_period (int): 计算ATR的回看周期。 stop_loss_atr_multiplier (float): ATR的乘数，用于设置止损。 partial_profit_atr_multiplier (float): ATR的乘数，用于设置第一止盈目标。 trailing_stop_atr_multiplier (float): ATR的乘数，用于设置最大回撤止盈。 """ super().__init__(context, main_symbol, enable_log) if order_direction is None: order_direction = ['BUY', 'SELL', ] # if trade_volume % 2 != 0: # raise ValueError("为了实现部分止盈, trade_volume必须是偶数。") self.trade_volume = trade_volume self.atr_period = atr_period self.stop_loss_atr_multiplier = stop_loss_atr_multiplier if isinstance(partial_profit_atr_multiplier, float) or isinstance(partial_profit_atr_multiplier, int): self.partial_profit_atr_multiplier = [partial_profit_atr_multiplier, partial_profit_atr_multiplier] elif isinstance(partial_profit_atr_multiplier, list): self.partial_profit_atr_multiplier = partial_profit_atr_multiplier self.trailing_stop_atr_multiplier = trailing_stop_atr_multiplier if isinstance(open_atr_multiplier, float) or isinstance(open_atr_multiplier, int): self.open_atr_multiplier = [open_atr_multiplier, open_atr_multiplier] elif isinstance(open_atr_multiplier, list): self.open_atr_multiplier = open_atr_multiplier # 初始化新增参数 self.order_direction = order_direction self.indicator_long = indicators[0] self.indicator_short = indicators[1] self.main_symbol = main_symbol self.order_id_counter = 0 self._last_order_id: Optional[str] = None # 字典，用于存储交易的元数据，如入场价、入场时的ATR和最高/最低价 self.position_meta: Dict[str, Any] = {} self.log( "ReversalVolatilityStrategy 初始化参数: \n" f"交易量={self.trade_volume}, ATR周期={self.atr_period}, \n" f"止损ATR乘数={self.stop_loss_atr_multiplier}, 部分止盈ATR乘数={self.partial_profit_atr_multiplier}, \n" f"回撤止盈ATR乘数={self.trailing_stop_atr_multiplier}" # 更新日志信息 ) def on_init(self): super().on_init() self.cancel_all_pending_orders(self.main_symbol) def on_open_bar(self, open_price: float, symbol: str): self.symbol = symbol current_time = self.get_current_time() # --- 1. 取消上一根K线未成交的限价单 --- if self._last_order_id and self._last_order_id in self.get_pending_orders(): self.cancel_order(self._last_order_id) self.log(f"已取消上一根K线的挂单: {self._last_order_id}") self._last_order_id = None bar_history = self.get_bar_history() if len(bar_history) < self.atr_period + 3: return # 数据不足，无法计算指标 # --- 2. 计算指标 --- # current_atr = calculate_atr(bar_history, self.atr_period) current_atr = talib.ATR(np.array([bar.high for bar in bar_history]), np.array([bar.low for bar in bar_history]), np.array([bar.close for bar in bar_history]), timeperiod=self.atr_period)[-1] if current_atr == 0: return # 避免除以零或在不活跃的市场中交易 fvg_signal = find_fvg(bar_history) # --- 3. 管理现有持仓 (平仓逻辑) --- current_positions = self.get_current_positions() position_volume = current_positions.get(self.symbol, 0) avg_entry_price = self.get_average_position_price(self.symbol) if position_volume != 0: # 传递当前Bar的最高价和最低价用于更新最优价格 current_bar_high = bar_history[-1].high current_bar_low = bar_history[-1].low self.manage_open_position(position_volume, open_price, current_bar_high, current_bar_low, avg_entry_price, current_atr) return # 如果正在管理平仓，则不评估进场信号 # --- 4. 评估新机会 (开仓逻辑) --- if position_volume == 0 and fvg_signal: self.evaluate_entry_signal(fvg_signal, open_price, current_atr) def manage_open_position(self, volume: int, current_price: float, current_bar_high: float, current_bar_low: float, entry_price: float, current_atr: float): # current_atr = self.position_meta[self.symbol]['entry_atr'] # 定义基于入场时ATR的止损目标价格 # stop_loss_price = entry_price - entry_atr * self.stop_loss_atr_multiplier if volume > 0 else entry_price + entry_atr * self.stop_loss_atr_multiplier stop_loss_price = entry_price - 5 if volume > 0 else entry_price + 5 partial_profit_atr_multiplier = self.partial_profit_atr_multiplier[0] if volume > 0 else \ self.partial_profit_atr_multiplier[1] partial_tp_price = entry_price + current_atr * partial_profit_atr_multiplier if volume > 0 else entry_price - current_atr * partial_profit_atr_multiplier self.log(f'持仓:{volume}, pos:{current_price - entry_price}, current_price:{current_price}, stop_loss_price:{stop_loss_price}, partial_profit_atr_multiplier:{partial_profit_atr_multiplier}, partial_tp_price:{partial_tp_price}') # 检查多头持仓 if volume > 0: # # 最大回撤止盈 (只在部分止盈后激活) # if partially_exited: # trailing_stop_level = optimal_price - self.trailing_stop_atr_multiplier * current_atr # if current_price <= trailing_stop_level: # self.log(f"多头回撤止盈到达 {current_price:.2f}. 关闭剩余仓位。") # self.close_position("CLOSE_LONG", abs(volume)) # 部分止盈 if current_price >= partial_tp_price: self.log(f"多头部分止盈到达 {current_price:.2f}. 关闭仓位。") # self.send_market_order("CLOSE_LONG", self.trade_volume) self.close_position("CLOSE_LONG", abs(volume)) # 首次触发部分止盈时，将当前价格设置为最优价格，作为回撤止盈的起点 # 止损 elif current_price <= stop_loss_price: self.log(f"多头止损到达 {current_price:.2f}. 关闭全部仓位。") self.close_position("CLOSE_LONG", abs(volume)) # 检查空头持仓 elif volume < 0: # # 最大回撤止盈 (只在部分止盈后激活) # if partially_exited: # trailing_stop_level = optimal_price + self.trailing_stop_atr_multiplier * current_atr # if current_price >= trailing_stop_level: # self.log(f"空头回撤止盈到达 {current_price:.2f}. 关闭剩余仓位。") # self.close_position("CLOSE_SHORT", abs(volume)) # 部分止盈 if current_price <= partial_tp_price: self.log(f"空头部分止盈到达 {current_price:.2f}. 关闭一半仓位。") # self.send_market_order("CLOSE_SHORT", self.trade_volume) self.close_position("CLOSE_SHORT", abs(volume)) # 止损 elif current_price >= stop_loss_price: self.log(f"空头止损到达 {current_price:.2f}. 关闭全部仓位。") self.close_position("CLOSE_SHORT", abs(volume)) def evaluate_entry_signal(self, fvg_signal, open_price, current_atr): direction, fvg_high, fvg_low = fvg_signal open_atr_multiplier = self.open_atr_multiplier[0] if 'bullish' == direction else self.open_atr_multiplier[1] # 多头入场信号 if 'bullish' == direction: # 在FVG上沿挂限价单，等待价格回调 if self.indicator_long is None or self.indicator_long.is_condition_met(*self.get_indicator_tuple()): limit_price = fvg_low + open_atr_multiplier * current_atr # 避免追高 if open_price > limit_price: self.log(f"检测到看涨FVG。在 {limit_price:.2f} 挂限价买单。fvg_high：{fvg_low}，current_atr：{current_atr}") new_order = self.send_limit_order("BUY", limit_price, self.trade_volume, {'entry_atr': current_atr}) if new_order: self._last_order_id = new_order.id # 空头入场信号 elif 'bearish' == direction: # 在FVG下沿挂限价单 if self.indicator_short is None or self.indicator_short.is_condition_met(*self.get_indicator_tuple()): limit_price = fvg_high - open_atr_multiplier * current_atr # 避免追低 if open_price < limit_price: self.log(f"检测到看跌FVG。在 {limit_price:.2f} 挂限价卖单。fvg_high：{fvg_high}，current_atr：{current_atr}") new_order = self.send_limit_order("SELL", limit_price, self.trade_volume, {'entry_atr': current_atr}) if new_order: self._last_order_id = new_order.id def close_position(self, direction: str, volume: int): self.send_market_order(direction, volume) if self.symbol in self.position_meta: del self.position_meta[self.symbol] # 完全平仓后清理元数据 def send_limit_order(self, direction: str, limit_price: float, volume: int, meta: Dict) -> Optional[Order]: if direction not in self.order_direction: return None order_id = f"{self.symbol}_{direction}_{self.get_current_time().strftime('%Y%m%d%H%M%S')}_{self.order_id_counter}" self.order_id_counter += 1 order = Order( id=order_id, symbol=self.symbol, direction=direction, volume=volume, price_type="LIMIT", limit_price=limit_price, submitted_time=self.get_current_time(), offset="OPEN", ) # 存储开仓时的元数据，包括入场价和入场时的ATR self.position_meta[self.symbol] = { "entry_price": limit_price, "entry_atr": meta.get('entry_atr', 0), "partially_exited": False, "optimal_price": limit_price # 初始最优价格设置为入场价格 } entry_price = limit_price entry_atr = meta.get('entry_atr', 0) stop_loss_price = entry_price - 5 if direction == "BUY" else entry_price + 5 partial_profit_atr_multiplier = self.partial_profit_atr_multiplier[0] if volume > 0 else \ self.partial_profit_atr_multiplier[1] partial_tp_price = entry_price + entry_atr * partial_profit_atr_multiplier if volume > 0 else entry_price - entry_atr * partial_profit_atr_multiplier self.log( f'atr: {meta.get("entry_atr", 0)}, stop_loss_price: {stop_loss_price}, partial_tp_price: {partial_tp_price}') return self.send_order(order) def send_market_order(self, direction: str, volume: int): order_id = f"{self.symbol}_{direction}_{self.get_current_time().strftime('%Y%m%d%H%M%S')}_{self.order_id_counter}" self.order_id_counter += 1 order = Order( id=order_id, symbol=self.symbol, direction=direction, volume=volume, price_type="MARKET", submitted_time=self.get_current_time(), offset="CLOSE" # 在本策略中，所有市价单都是为了平仓 ) self.send_order(order) def on_rollover(self, old_symbol: str, new_symbol: str): super().on_rollover(old_symbol, new_symbol) self._last_order_id = None self.position_meta = {} # 换月时清空所有元数据 self.log("检测到换月。已清空挂单和持仓元数据。") def on_close_bar(self, bar: Bar, next_bar_open: Optional[float] = None): self.cancel_all_pending_orders(self.main_symbol)