import numpy as np import pandas as pd from typing import Optional, Dict, Any, List, Union import talib 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 from src.algo.TrendLine import calculate_latest_trendline_values class _SignalGenerator: """ 内部帮助类，用于封装单个策略（趋势或回归）的霍克斯过程信号生成所需的所有状态和逻辑。 """ def __init__(self, hawkes_kappa: float, hawkes_lookback: int, volume_norm_n: int): self.hawkes_kappa = hawkes_kappa self.hawkes_lookback = hawkes_lookback self.volume_norm_n = volume_norm_n # 状态变量 self._last_hawkes_unscaled: float = 0.0 self._hawkes_window: np.ndarray = np.array([], dtype=np.float64) self._hawkes_alpha: float = np.exp(-self.hawkes_kappa) self._volume_window: np.ndarray = np.zeros(self.volume_norm_n, dtype=np.float64) self._volume_sum: float = 0.0 self._volume_sum_sq: float = 0.0 self._volume_pointer: int = 0 self._is_volume_window_full: bool = False def reset(self): """重置所有状态""" self._last_hawkes_unscaled = 0.0 self._hawkes_window = np.array([], dtype=np.float64) self._volume_window.fill(0) self._volume_sum = 0.0 self._volume_sum_sq = 0.0 self._volume_pointer = 0 self._is_volume_window_full = False def initialize_state(self, initial_volumes: np.ndarray): """用历史数据批量初始化状态""" normalized_volumes = [] for vol in initial_volumes: self._update_volume_stats_incrementally(vol) mean, std = self._get_current_volume_stats() z_score = 0.0 if std <= 1e-9 else (vol - mean) / std normalized_volumes.append(z_score) temp_hawkes_history = np.zeros_like(normalized_volumes, dtype=np.float64) if len(normalized_volumes) > 0: temp_hawkes_history[0] = normalized_volumes[0] for i in range(1, len(normalized_volumes)): temp_hawkes_history[i] = temp_hawkes_history[i - 1] * self._hawkes_alpha + normalized_volumes[i] self._last_hawkes_unscaled = temp_hawkes_history[-1] if len(temp_hawkes_history) > 0 else 0.0 self._hawkes_window = (temp_hawkes_history * self.hawkes_kappa)[-self.hawkes_lookback:] def update_state_incrementally(self, latest_volume: float): """在每个bar上增量更新状态""" self._update_volume_stats_incrementally(latest_volume) mean, std = self._get_current_volume_stats() normalized_volume = 0.0 if std <= 1e-9 else (latest_volume - mean) / std new_hawkes_unscaled = self._last_hawkes_unscaled * self._hawkes_alpha + normalized_volume self._last_hawkes_unscaled = new_hawkes_unscaled new_hawkes_scaled = new_hawkes_unscaled * self.hawkes_kappa if self._hawkes_window.size < self.hawkes_lookback: self._hawkes_window = np.append(self._hawkes_window, new_hawkes_scaled) else: self._hawkes_window = np.roll(self._hawkes_window, -1) self._hawkes_window[-1] = new_hawkes_scaled def _update_volume_stats_incrementally(self, latest_volume: float): oldest_volume = self._volume_window[self._volume_pointer] self._volume_sum += latest_volume - oldest_volume self._volume_sum_sq += latest_volume ** 2 - oldest_volume ** 2 self._volume_window[self._volume_pointer] = latest_volume self._volume_pointer = (self._volume_pointer + 1) % self.volume_norm_n if not self._is_volume_window_full and self._volume_pointer == 0: self._is_volume_window_full = True def _get_current_volume_stats(self) -> (float, float): n = self.volume_norm_n if self._is_volume_window_full else self._volume_pointer if n == 0: return 0.0, 0.0 mean = self._volume_sum / n variance = max(0, (self._volume_sum_sq / n) - mean ** 2) std = np.sqrt(variance) return mean, std def get_latest_hawkes_value(self) -> Optional[float]: return self._hawkes_window[-1] if self._hawkes_window.size > 0 else None def get_hawkes_quantile(self, percentile: float) -> Optional[float]: return np.quantile(self._hawkes_window, percentile) if self._hawkes_window.size > 0 else None class DualModeTrendlineHawkesStrategy(Strategy): """ 趋势线与霍克斯过程双重确认策略 (V11 - 完全独立信号版): - 为趋势(Trend)和均值回归(Reversion)策略分别维护一套完全独立的信号生成器。 - 每个策略使用各自的 trendline_n, hawkes_kappa, hawkes_lookback, volume_norm_n 参数。 - 信号生成完全分离，确保逻辑独立性。 """ def __init__( self, context: Any, main_symbol: str, trend_enabled: bool = True, reversion_enabled: bool = True, conflict_resolution_mode: str = 'trend_priority', trend_params: Dict[str, Any] = None, reversion_params: Dict[str, Any] = None, enable_log: bool = True, indicators: Union[Indicator, List[Indicator]] = None, ): super().__init__(context, main_symbol, enable_log) self.main_symbol = main_symbol self.trend_enabled = trend_enabled self.reversion_enabled = reversion_enabled if conflict_resolution_mode not in ['trend_priority', 'reversion_priority', 'none']: raise ValueError("conflict_resolution_mode 必须是 'trend_priority', 'reversion_priority', 或 'none'") self.conflict_resolution_mode = conflict_resolution_mode default_params = { "trade_volume": 1, "order_direction": ["BUY", "SELL"], "hawkes_entry_percent": 0.95, "hawkes_exit_percent": 0.50, "enable_atr_stop_loss": True, "atr_period": 14, "atr_multiplier": 2.0, "trendline_n": 50, "hawkes_kappa": 0.1, "hawkes_lookback": 50, "volume_norm_n": 50, } self.trend_params = default_params.copy() if trend_params: self.trend_params.update(trend_params) self.reversion_params = default_params.copy() if reversion_params: self.reversion_params.update(reversion_params) # --- 【核心修改】创建两个独立的信号生成器实例 --- self.trend_generator = _SignalGenerator( hawkes_kappa=self.trend_params['hawkes_kappa'], hawkes_lookback=self.trend_params['hawkes_lookback'], volume_norm_n=self.trend_params['volume_norm_n'] ) self.reversion_generator = _SignalGenerator( hawkes_kappa=self.reversion_params['hawkes_kappa'], hawkes_lookback=self.reversion_params['hawkes_lookback'], volume_norm_n=self.reversion_params['volume_norm_n'] ) self.pos_meta: Dict[str, Dict[str, Any]] = {} self.indicators = indicators or [Empty(), Empty()] self._is_initialized = False def on_init(self): super().on_init() self.pos_meta.clear() self.trend_generator.reset() self.reversion_generator.reset() self._is_initialized = False def on_open_bar(self, open_price: float, symbol: str): bar_history = self.get_bar_history() min_bars_required = max( self.trend_params['trendline_n'] + 2, self.reversion_params['trendline_n'] + 2, self.trend_params['hawkes_lookback'] + 2, self.reversion_params['hawkes_lookback'] + 2, self.trend_params['volume_norm_n'] + 2, self.reversion_params['volume_norm_n'] + 2, self.trend_params['atr_period'] + 2, self.reversion_params['atr_period'] + 2 ) if len(bar_history) < min_bars_required: return latest_volume = float(bar_history[-1].volume) # --- 【核心修改】初始化或更新两个独立的生成器 --- if not self._is_initialized: initial_volumes = np.array([b.volume for b in bar_history[:-1]], dtype=float) self.log("首次运行，正在初始化趋势信号生成器...") self.trend_generator.initialize_state(initial_volumes) self.log("正在初始化回归信号生成器...") self.reversion_generator.initialize_state(initial_volumes) self._is_initialized = True self.trend_generator.update_state_incrementally(latest_volume) self.reversion_generator.update_state_incrementally(latest_volume) self.cancel_all_pending_orders(symbol) pos = self.get_current_positions().get(symbol, 0) # --- 平仓逻辑 --- meta = self.pos_meta.get(symbol) if meta and pos != 0: strategy_type = meta.get('strategy_type', 'trend') # 根据开仓类型选择正确的参数和生成器 params, generator = (self.trend_params, self.trend_generator) if strategy_type == 'trend' \ else (self.reversion_params, self.reversion_generator) latest_hawkes_value = generator.get_latest_hawkes_value() latest_hawkes_lower = generator.get_hawkes_quantile(params['hawkes_exit_percent']) close_reason = None if latest_hawkes_value is not None and latest_hawkes_lower is not None and latest_hawkes_value < latest_hawkes_lower: close_reason = f"[{strategy_type.upper()}] 霍克斯出场信号(强度: {latest_hawkes_value:.4f} < 阈值: {latest_hawkes_lower:.4f})" if params['enable_atr_stop_loss'] and 'stop_loss_price' in meta and meta['stop_loss_price'] is not None: if (meta['direction'] == "BUY" and bar_history[-1].close < meta['stop_loss_price']) or \ (meta['direction'] == "SELL" and bar_history[-1].close > meta['stop_loss_price']): close_reason = f"[{strategy_type.upper()}] ATR止损触发" if close_reason: self.log(close_reason) self.send_market_order("CLOSE_LONG" if meta['direction'] == "BUY" else "CLOSE_SHORT", abs(pos)) if symbol in self.pos_meta: del self.pos_meta[symbol] return # --- 开仓逻辑 --- if pos == 0: trend_signal, reversion_signal = None, None close_prices = np.array([b.close for b in bar_history]) prev_close, last_close = bar_history[-2].close, bar_history[-1].close # 1. 检查趋势策略信号 if self.trend_enabled: prices_for_trendline = close_prices[-self.trend_params['trendline_n'] - 1:-1] trend_upper, trend_lower = calculate_latest_trendline_values(prices_for_trendline) if trend_upper is not None: upper_break = last_close > trend_upper and prev_close < trend_upper lower_break = last_close < trend_lower and prev_close > trend_lower hawkes_val = self.trend_generator.get_latest_hawkes_value() hawkes_thresh = self.trend_generator.get_hawkes_quantile(self.trend_params['hawkes_entry_percent']) if hawkes_val is not None and hawkes_thresh is not None and hawkes_val > hawkes_thresh and ( upper_break or lower_break): direction = "BUY" if upper_break else "SELL" if direction in self.trend_params['order_direction']: trend_signal = direction # 2. 检查均值回归策略信号 if self.reversion_enabled: prices_for_trendline = close_prices[-self.reversion_params['trendline_n'] - 1:-1] trend_upper, trend_lower = calculate_latest_trendline_values(prices_for_trendline) if trend_upper is not None: upper_break = last_close > trend_upper and prev_close < trend_upper lower_break = last_close < trend_lower and prev_close > trend_lower hawkes_val = self.reversion_generator.get_latest_hawkes_value() hawkes_thresh = self.reversion_generator.get_hawkes_quantile( self.reversion_params['hawkes_entry_percent']) if hawkes_val is not None and hawkes_thresh is not None and hawkes_val > hawkes_thresh and ( upper_break or lower_break): direction = "SELL" if upper_break else "BUY" # 方向相反 if direction in self.reversion_params['order_direction']: reversion_signal = direction # 3. 解决信号冲突和下单 final_direction, final_params, strategy_type, generator = None, None, None, None if trend_signal and not reversion_signal: final_direction, final_params, strategy_type, generator = trend_signal, self.trend_params, 'trend', self.trend_generator elif not trend_signal and reversion_signal: final_direction, final_params, strategy_type, generator = reversion_signal, self.reversion_params, 'reversion', self.reversion_generator elif trend_signal and reversion_signal: self.log( f"开仓信号冲突: 趋势={trend_signal}, 回归={reversion_signal}. 模式: {self.conflict_resolution_mode}") if self.conflict_resolution_mode == 'trend_priority': final_direction, final_params, strategy_type, generator = trend_signal, self.trend_params, 'trend', self.trend_generator elif self.conflict_resolution_mode == 'reversion_priority': final_direction, final_params, strategy_type, generator = reversion_signal, self.reversion_params, 'reversion', self.reversion_generator if final_direction: sl_price = None if final_params['enable_atr_stop_loss']: atr_val = self._calculate_atr(bar_history[:-1], final_params['atr_period']) if atr_val is not None: sl_price = open_price - atr_val * final_params[ 'atr_multiplier'] if final_direction == "BUY" else open_price + atr_val * final_params[ 'atr_multiplier'] self.log( f"[{strategy_type.upper()}] 开仓信号确认 (霍克斯强度: {generator.get_latest_hawkes_value():.4f} > 阈值: {generator.get_hawkes_quantile(final_params['hawkes_entry_percent']):.4f})") self.send_open_order(final_direction, open_price, final_params['trade_volume'], sl_price, strategy_type) # --- 辅助函数 (与之前版本相同) --- def _calculate_atr(self, bar_history: List[Bar], period: int) -> Optional[float]: # ... (代码不变) if len(bar_history) < period + 1: return None highs = np.array([b.high for b in bar_history], dtype=float) lows = np.array([b.low for b in bar_history], dtype=float) closes = np.array([b.close for b in bar_history], dtype=float) atr_values = talib.ATR(highs, lows, closes, timeperiod=period) return atr_values[-1] if not np.isnan(atr_values[-1]) else None def send_open_order(self, direction: str, entry_price: float, volume: int, stop_loss_price: Optional[float], strategy_type: str): # ... (代码不变) current_time = self.get_current_time() order_id = f"{self.symbol}_{direction}_{current_time.strftime('%Y%m%d%H%M%S')}" order = Order(id=order_id, symbol=self.symbol, direction="BUY" if direction == "BUY" else "SELL", volume=volume, price_type="MARKET", submitted_time=current_time, offset="OPEN") self.send_order(order) self.pos_meta[self.symbol] = {"direction": direction, "volume": volume, "entry_price": entry_price, "stop_loss_price": stop_loss_price, "strategy_type": strategy_type} self.log(f"[{strategy_type.upper()}] 发送开仓订单: {direction} {volume}手") def send_market_order(self, direction: str, volume: int): # ... (代码不变) current_time = self.get_current_time() order_id = f"{self.symbol}_{direction}_{current_time.strftime('%Y%m%d%H%M%S')}" order = Order(id=order_id, symbol=self.symbol, direction=direction, volume=volume, price_type="MARKET", submitted_time=current_time, offset="CLOSE") self.send_order(order) self.log(f"发送平仓订单: {direction} {volume}手") def on_rollover(self, old_symbol: str, new_symbol: str): # ... (代码不变) super().on_rollover(old_symbol, new_symbol) self.cancel_all_pending_orders(new_symbol) self.pos_meta.clear()