Files

290 lines
14 KiB
Python
Raw Permalink Normal View History

2025-11-07 16:37:16 +08:00
import numpy as np
import pandas as pd
import talib
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
# =============================================================================
# 策略实现 (Dual-Mode Kalman Strategy) - 优化版
# =============================================================================
class DualModeKalmanStrategy(Strategy):
"""
一个内置两种相反交易逻辑趋势跟踪 vs. 均值回归的自适应策略
本策略旨在通过一个核心参数 `strategy_mode`在两种市场范式间切换
以适应不同品种的内在性格
人格1: 'TREND' (趋势模式)
- 哲学: 价格的强力突破会引发自我强化的趋势
- 入场: 价格向上/逃逸卡尔曼线 -> 做多/做空
- 出场: 使用真正的单向追踪止盈Trailing Stop锁定利润
人格2: 'REVERSION' (均值回归模式)
- 哲学: 价格的极端偏离是不可持续的终将回归均值
- 入场: 价格向上/下极端偏离超买/超卖 -> 做空/做多
- 出场: 当价格回归至卡尔曼线时获利了结
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【核心人格切换】 ---
strategy_mode: str = 'TREND', # 可选: 'TREND' 或 'REVERSION'
# --- 【通用参数】 ---
kalman_process_noise: float = 0.01,
kalman_measurement_noise: float = 0.5,
atr_period: int = 20,
atr_lookback: int = 100,
atr_percentile_threshold: float = 25.0,
entry_threshold_atr: float = 2.5,
initial_stop_atr_multiplier: float = 2.0,
# --- 【趋势模式专用】 ---
structural_stop_atr_multiplier: float = 2.5,
order_direction: Optional[List[str]] = None,
indicators: Optional[List[Indicator]] = None,
):
super().__init__(context, main_symbol, enable_log)
if order_direction is None: order_direction = ['BUY', 'SELL']
# --- 参数验证与赋值 ---
if strategy_mode.upper() not in ['TREND', 'REVERSION']:
raise ValueError(f"strategy_mode must be 'TREND' or 'REVERSION', but got {strategy_mode}")
self.strategy_mode = strategy_mode.upper()
self.trade_volume = trade_volume
self.atr_period = atr_period
self.atr_lookback = atr_lookback
self.atr_percentile_threshold = atr_percentile_threshold
self.entry_threshold_atr = entry_threshold_atr
self.initial_stop_atr_multiplier = initial_stop_atr_multiplier
self.structural_stop_atr_multiplier = structural_stop_atr_multiplier
self.order_direction = order_direction
# 卡尔曼滤波器状态
self.Q = kalman_process_noise
self.R = kalman_measurement_noise
self.P = 1.0;
self.x_hat = 0.0;
self.kalman_initialized = False
self._atr_history: deque = deque(maxlen=self.atr_lookback)
self.position_meta: Dict[str, Any] = self.context.load_state()
self.main_symbol = main_symbol
self.order_id_counter = 0
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
self.log(f"DualModeKalmanStrategy Initialized with Personality: [{self.strategy_mode}]")
def on_init(self):
super().on_init()
self.cancel_all_pending_orders(self.main_symbol)
self.position_meta = self.context.load_state()
def on_open_bar(self, open_price: float, symbol: str):
self.symbol = symbol
bar_history = self.get_bar_history()
if len(bar_history) < max(self.atr_period, self.atr_lookback) + 2: return
# --- 通用数据计算 ---
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)
current_atr = talib.ATR(highs, lows, closes, self.atr_period)[-1]
self._atr_history.append(current_atr)
if current_atr <= 0 or len(self._atr_history) < self.atr_lookback: return
if not self.kalman_initialized: self.x_hat = closes[-1]; self.kalman_initialized = True
x_hat_minus = self.x_hat
P_minus = self.P + self.Q
K = P_minus / (P_minus + self.R)
self.x_hat = x_hat_minus + K * (closes[-1] - x_hat_minus)
self.P = (1 - K) * P_minus
kalman_price = self.x_hat
# --- 分模式管理持仓 ---
position_volume = self.get_current_positions().get(self.symbol, 0)
meta = self.position_meta.get(symbol)
if position_volume != 0 and not meta:
self.log(f"警告:检测到实际持仓({position_volume})与策略状态(无记录)不一致!"
f"可能由状态加载失败导致。将强制平仓以同步状态。", level='WARNING')
direction_to_close = "CLOSE_LONG" if position_volume > 0 else "CLOSE_SHORT"
self.send_market_order(direction_to_close, abs(position_volume), 'CLOSE')
return
if position_volume == 0 and meta:
self.log(f"信息:检测到策略状态({meta.get('direction')})与实际持仓(0)不一致meta:{meta}"
f"可能是外部平仓导致。正在清理过时状态。", level='INFO')
new_pos_meta = {k: v for k, v in self.position_meta.items() if k != symbol}
self.position_meta = new_pos_meta
self.save_state(new_pos_meta)
if not self.trading:
return
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1], current_atr, kalman_price)
return
# --- 状态过滤 (对两种模式都适用) ---
atr_threshold = np.percentile(list(self._atr_history), self.atr_percentile_threshold)
if current_atr < atr_threshold: return
# --- 分模式评估新机会 ---
self.evaluate_entry_signal(bar_history[-1], kalman_price, current_atr)
# --- [OPTIMIZED] ---
# 重写 manage_open_position 方法,以实现真正的追踪止盈和修正回归模式逻辑
def manage_open_position(self, volume: int, current_bar: Bar, current_atr: float, kalman_price: float):
meta = self.position_meta.get(self.symbol)
if not meta: return
# --- 分模式出场逻辑 ---
if self.strategy_mode == 'TREND':
# 【趋势模式】: 实现真正的单向追踪止盈 (Trailing Stop)
# 1. 从 meta 中获取当前的追踪止损位
# (使用 .get() 是为了兼容可能没有此字段的旧版状态文件)
current_trailing_stop = meta.get('trailing_stop_price', meta['initial_stop_price'])
new_potential_stop = 0
if volume > 0: # 持有多单
# 2. 计算基于当前卡尔曼线和ATR的新“潜在”止损位
new_potential_stop = kalman_price - self.structural_stop_atr_multiplier * current_atr
# 3. 核心逻辑: 只让止损位朝有利方向(向上)移动,绝不回撤
updated_trailing_stop = max(current_trailing_stop, new_potential_stop)
self.log(
f"TREND LONG: Trailing Stop Updated to {updated_trailing_stop:.4f} (from {current_trailing_stop:.4f}). Potential new stop was {new_potential_stop:.4f}.")
# 4. 检查价格是否触及更新后的追踪止损位
if current_bar.low <= updated_trailing_stop:
self.log(f"TREND Mode: Trailing Stop hit for LONG at {updated_trailing_stop:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
return
else: # 持有空单
# 2. 计算新“潜在”止损位
new_potential_stop = kalman_price + self.structural_stop_atr_multiplier * current_atr
# 3. 核心逻辑: 只让止损位朝有利方向(向下)移动,绝不回撤
updated_trailing_stop = min(current_trailing_stop, new_potential_stop)
self.log(
f"TREND SHORT: Trailing Stop Updated to {updated_trailing_stop:.4f} (from {current_trailing_stop:.4f}). Potential new stop was {new_potential_stop:.4f}.")
# 4. 检查价格是否触及更新后的追踪止损位
if current_bar.high >= updated_trailing_stop:
self.log(f"TREND Mode: Trailing Stop hit for SHORT at {updated_trailing_stop:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
return
# 5. 【状态持久化】将更新后的追踪止损位存回 meta用于下一根K线
if updated_trailing_stop != current_trailing_stop:
self.position_meta[self.symbol]['trailing_stop_price'] = updated_trailing_stop
self.save_state(self.position_meta)
elif self.strategy_mode == 'REVERSION':
# 【回归模式】: 目标是回归均值,因此止盈逻辑是触及卡尔曼线
# 首先,检查初始硬止损,作为最后的安全防线
initial_stop_price = meta['initial_stop_price']
if (volume > 0 and current_bar.low <= initial_stop_price) or \
(volume < 0 and current_bar.high >= initial_stop_price):
self.log(f"REVERSION Mode: Initial Stop Loss hit at {initial_stop_price:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
return
# 其次,检查止盈条件:价格是否已回归至卡尔曼线
if volume > 0: # 持有从下方回归的多单
if current_bar.high >= kalman_price:
self.log(f"REVERSION Mode: Take Profit for LONG as price hit Kalman line at {kalman_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
else: # 持有从上方回归的空单
if current_bar.low <= kalman_price:
self.log(f"REVERSION Mode: Take Profit for SHORT as price hit Kalman line at {kalman_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
# --- [OPTIMIZED] ---
# 修改 evaluate_entry_signal 方法,以在开仓时初始化追踪止盈状态
def evaluate_entry_signal(self, current_bar: Bar, kalman_price: float, current_atr: float):
deviation = current_bar.close - kalman_price
deviation_in_atr = deviation / current_atr
direction = None
if self.strategy_mode == 'TREND':
if "BUY" in self.order_direction and deviation_in_atr > self.entry_threshold_atr and self.indicators[
0].is_condition_met(*self.get_indicator_tuple()):
direction = "BUY"
elif "SELL" in self.order_direction and deviation_in_atr < -self.entry_threshold_atr and self.indicators[
1].is_condition_met(*self.get_indicator_tuple()):
direction = "SELL"
elif self.strategy_mode == 'REVERSION':
if "SELL" in self.order_direction and deviation_in_atr > self.entry_threshold_atr and self.indicators[
1].is_condition_met(*self.get_indicator_tuple()):
direction = "SELL"
elif "BUY" in self.order_direction and deviation_in_atr < -self.entry_threshold_atr and self.indicators[
0].is_condition_met(*self.get_indicator_tuple()):
direction = "BUY"
if direction:
self.log(
f"{self.strategy_mode} Mode: Catalyst Fired. Direction: {direction}. Deviation: {deviation_in_atr:.2f} ATRs.")
entry_price = current_bar.close
stop_loss_price = entry_price - self.initial_stop_atr_multiplier * current_atr if direction == "BUY" else entry_price + self.initial_stop_atr_multiplier * current_atr
# --- 【状态持久化】---
# 在 meta 中增加 trailing_stop_price 字段,并用初始止损位为其赋值
# 这样,追踪止盈就从一个固定的“硬止损”开始了
meta = {
'entry_price': entry_price,
'initial_stop_price': stop_loss_price,
'trailing_stop_price': stop_loss_price # 关键:初始化追踪止损
}
# 使用市价单确保入场
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta:
# 使用 pop 来安全地删除键,并返回其值(虽然这里我们不需要)
self.position_meta.pop(self.symbol, None)
self.save_state(self.position_meta)
def send_market_order(self, direction: str, volume: int, offset: str, meta: Optional[Dict] = None):
if offset == "OPEN" and meta: self.position_meta[self.symbol] = meta
order_id = f"{self.symbol}_{direction}_MARKET_{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,
meta: Optional[Dict] = None):
if offset == "OPEN" and meta: self.position_meta[self.symbol] = meta
order_id = f"{self.symbol}_{direction}_LIMIT_{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)
def on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
# 重置所有与特定合约相关的状态
self.position_meta = {}
self.kalman_initialized = False
self._atr_history.clear()
self.save_state({}) # 清空持久化状态
self.log("Rollover detected. All strategy states have been reset.")