import numpy as np import math from collections import deque from typing import Optional, Any, List, Dict from src.core_data import Bar, Order from src.indicators.base_indicators import Indicator from src.indicators.indicators import Empty from src.strategies.base_strategy import Strategy class ITrendStrategy(Strategy): """ 【Ehlers 瞬时趋势线策略 (期货实战版)】 针对期货日内交易优化： 1. 【跳数止损】：使用固定的 min_tick 跳数作为止损，替代百分比。 2. 【价格对齐】：所有计算出的挂单价格，强制对齐到 min_tick 的整数倍。 3. 【无反手】：止损或信号反转仅平仓。 参数含义变更： - min_tick: 合约最小变动价位 (如 IF为0.2, RB为1) - stop_loss_ticks: 止损跳数 (如 50跳) """ def __init__( self, context: Any, main_symbol: str, enable_log: bool, trade_volume: int, # --- 【合约规格参数】 --- min_tick: float = 1.0, # 核心：必须根据品种设置 (例如 0.2, 1.0, 5.0) # --- 【策略参数】 --- length: int = 20, # 趋势线周期 range_fraction: float = 0.35, # 入场回调系数 (Range的比例) stop_loss_ticks: int = 30, # 【新】硬止损跳数 (替代百分比) # --- 【其他】 --- order_direction: Optional[List[str]] = None, indicator: Indicator = None, ): super().__init__(context, main_symbol, enable_log) if order_direction is None: order_direction = ['BUY', 'SELL'] self.trade_volume = trade_volume self.min_tick = min_tick self.rng_frac = range_fraction self.stop_ticks = stop_loss_ticks # 止损跳数 self.order_direction = order_direction self.alpha = 2.0 / (length + 1.0) self.indicator = indicator # 历史数据缓存 self._mid_price_history = deque(maxlen=50) self._itrend_history = deque(maxlen=50) self._trigger_history = deque(maxlen=50) self.bar_count = 0 self.order_id_counter = 0 def round_to_tick(self, price: float) -> float: """辅助函数：将价格对齐到最小变动价位""" if self.min_tick <= 0: return price return round(price / self.min_tick) * self.min_tick def on_init(self): super().on_init() def on_rollover(self, old_symbol: str, new_symbol: str): self.log(f"合约换月: {old_symbol} -> {new_symbol}，重置状态。") self._mid_price_history.clear() self._itrend_history.clear() self._trigger_history.clear() self.bar_count = 0 self.symbol = new_symbol self.cancel_all_pending_orders(old_symbol) def on_open_bar(self, open_price: float, symbol: str): self.symbol = symbol bars = self.get_bar_history() if len(bars) < 2: return self.cancel_all_pending_orders(symbol) prev_bar = bars[-1] # --- 1. 指标计算 --- mid_price = (prev_bar.high + prev_bar.low) / 2.0 self._mid_price_history.append(mid_price) self.bar_count += 1 if len(self._mid_price_history) < 3: self._itrend_history.append(mid_price) self._trigger_history.append(mid_price) return price = list(self._mid_price_history) itrend_prev = list(self._itrend_history) current_itrend = 0.0 if self.bar_count < 7: current_itrend = (price[-1] + 2 * price[-2] + price[-3]) / 4.0 else: alpha = self.alpha a2 = alpha * alpha current_itrend = (alpha - a2 / 4) * price[-1] + (a2 / 2) * price[-2] - (alpha - 0.75 * a2) * price[-3] + \ 2 * (1 - alpha) * itrend_prev[-1] - (1 - alpha) ** 2 * itrend_prev[-2] self._itrend_history.append(current_itrend) if len(self._itrend_history) < 3: current_trigger = current_itrend else: current_trigger = 2.0 * current_itrend - self._itrend_history[-3] self._trigger_history.append(current_trigger) # --- 2. 交易决策 --- if len(self._trigger_history) < 2: return curr_trig = self._trigger_history[-1] prev_trig = self._trigger_history[-2] curr_itrend = self._itrend_history[-1] prev_itrend = self._itrend_history[-2] is_bullish = (prev_trig <= prev_itrend) and (curr_trig > curr_itrend) is_bearish = (prev_trig >= prev_itrend) and (curr_trig < curr_itrend) position_volume = self.get_current_positions().get(self.symbol, 0) prev_range = prev_bar.high - prev_bar.low # === 分支 A: 持仓管理 (止损/平仓) === if position_volume != 0: entry_price = self.get_average_position_price(self.symbol) # --- A1. 强制跳数止损 (Tick Stop) --- # 计算止损价差绝对值 stop_diff = self.stop_ticks * self.min_tick self.log(f'Holding {position_volume} position volume' f'is_bullish={is_bullish}, is_bearish={is_bearish}' f'entry_price={entry_price}, stop_diff={stop_diff}') if position_volume > 0: # 多头 # 止损价 = 开仓价 - 止损点数 stop_price = entry_price - stop_diff # 对齐一下(虽然entry_price理论上已对齐，但为了保险) stop_price = self.round_to_tick(stop_price) if prev_bar.close < stop_price: self.log( f"LONG STOP TRIGGERED. Close:{prev_bar.close} < Stop:{stop_price} (-{self.stop_ticks} ticks)") self.send_market_order("CLOSE_LONG", abs(position_volume), "CLOSE") return elif position_volume < 0: # 空头 # 止损价 = 开仓价 + 止损点数 stop_price = entry_price + stop_diff stop_price = self.round_to_tick(stop_price) if prev_bar.close > stop_price: self.log( f"SHORT STOP TRIGGERED. Close:{prev_bar.close} > Stop:{stop_price} (+{self.stop_ticks} ticks)") self.send_market_order("CLOSE_SHORT", abs(position_volume), "CLOSE") return # --- A2. 信号平仓 --- if position_volume > 0 and is_bearish: self.send_market_order("CLOSE_LONG", abs(position_volume), "CLOSE") return if position_volume < 0 and is_bullish: self.send_market_order("CLOSE_SHORT", abs(position_volume), "CLOSE") return # === 分支 B: 开仓管理 === else: # position_volume == 0 # 计算回调距离 (Range Fraction) # 例如 Range=10, Frac=0.35 -> 3.5 -> 对齐到tick raw_offset = self.rng_frac offset_price = self.round_to_tick(raw_offset) self.log(f'RANDOM OFFSET: {offset_price}, is_bullish={is_bullish}, is_bearish={is_bearish}') # 开多 if is_bullish and "BUY" in self.order_direction and (self.indicator is None or self.indicator.is_condition_met(*self.get_indicator_tuple())): # 挂单价 = Open - 回调点数 limit_price = prev_bar.close - offset_price limit_price = self.round_to_tick(limit_price) # 再次确保对齐 self.send_limit_order(limit_price, "BUY", self.trade_volume, "OPEN") self.log(f"Signal BUY. Open:{open_price} - Offset:{offset_price} = Limit:{limit_price}") # 开空 elif is_bearish and "SELL" in self.order_direction and (self.indicator is None or self.indicator.is_condition_met(*self.get_indicator_tuple())): # 挂单价 = Open + 回调点数 limit_price = prev_bar.close + offset_price limit_price = self.round_to_tick(limit_price) self.send_limit_order(limit_price, "SELL", self.trade_volume, "OPEN") self.log(f"Signal SELL. Open:{open_price} + Offset:{offset_price} = Limit:{limit_price}") # --- 交易辅助函数 --- def send_market_order(self, direction: str, volume: int, offset: str): order_id = f"{self.symbol}_{direction}_MKT_{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=offset) self.send_order(order) def send_limit_order(self, limit_price: float, direction: str, volume: int, offset: str): order_id = f"{self.symbol}_{direction}_LMT_{self.order_id_counter}" self.order_id_counter += 1 order = Order(id=order_id, symbol=self.symbol, direction=direction, volume=volume, price_type="LIMIT", submitted_time=self.get_current_time(), offset=offset, limit_price=limit_price) self.send_order(order)