1、新增傅里叶策略

2、新增策略管理、策略重启功能
This commit is contained in:
2025-11-20 16:15:45 +08:00
parent 2c917a467a
commit 711b86d33f
46 changed files with 12136 additions and 0 deletions

View File

@@ -0,0 +1,214 @@
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
# =============================================================================
# 策略实现 (V9 - 盘整突破版)
# =============================================================================
class ConsolidationBreakoutStrategy(Strategy):
"""
一个基于“盘整即突破之母”原则的终极波段策略。(V9)
本策略旨在通过量化的方法,识别市场能量的“压缩-释放”周期,
以捕捉趋势的主干行情,并严格规避无序的震荡。
1. 【战略层】: 使用长期均线定义宏观牛熊环境。
2. 【战术层】: 使用布林带宽度(BBW)的极度收缩,来识别“盘整准备”状态。
3. 【执行层】: 在准备状态下,等待价格“突破”凯尔特纳通道作为“大幅移动”的确认信号,
立即以市价入场并使用ATR动态追踪止盈来截取波段利润。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【战略层】环境定义 ---
regime_ma_period: int = 200, # 定义牛熊市的长期均线
# --- 【战术层】盘整收缩识别 (Squeeze) ---
bollinger_period: int = 20, # 布林带周期
bollinger_std: float = 2.0, # 布林带标准差
bbw_lookback: int = 252, # BBW历史分位数回看窗口
bbw_squeeze_percentile: float = 15.0, # 定义“极度收缩”的百分位
# --- 【执行层】突破与风控 ---
keltner_period: int = 20, # 凯尔特纳通道周期
keltner_atr_multiplier: float = 1.5, # 凯尔特纳通道ATR乘数
atr_period: int = 20,
trailing_stop_atr_multiplier: float = 3.0,
initial_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']
self.trade_volume = trade_volume
self.regime_ma_period = regime_ma_period
self.bollinger_period = bollinger_period
self.bollinger_std = bollinger_std
self.bbw_lookback = bbw_lookback
self.bbw_squeeze_percentile = bbw_squeeze_percentile
self.keltner_period = keltner_period
self.keltner_atr_multiplier = keltner_atr_multiplier
self.atr_period = atr_period
self.trailing_stop_atr_multiplier = trailing_stop_atr_multiplier
self.initial_stop_atr_multiplier = initial_stop_atr_multiplier
self.order_direction = order_direction
# 状态变量
self._last_order_id: Optional[str] = None
self.position_meta: Dict[str, Any] = {}
self._bbw_history: deque = deque(maxlen=self.bbw_lookback)
self._in_squeeze_ready_state: bool = False
self.main_symbol = main_symbol
self.order_id_counter = 0
if indicators is None:
indicators = [Empty(), Empty()]
self.indicators = indicators
self.log("ConsolidationBreakoutStrategy (V9) Initialized.")
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
bar_history = self.get_bar_history()
required_bars = max(self.regime_ma_period, self.bbw_lookback) + 1
if len(bar_history) < required_bars: 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)
# 战略指标
regime_ma = talib.MA(closes, timeperiod=self.regime_ma_period)[-1]
# 战术指标 (Squeeze)
bb_upper, bb_mid, bb_lower = talib.BBANDS(closes, timeperiod=self.bollinger_period, nbdevup=self.bollinger_std,
nbdevdn=self.bollinger_std)
bb_width = (bb_upper[-1] - bb_lower[-1]) / bb_mid[-1] if bb_mid[-1] > 0 else 0
self._bbw_history.append(bb_width)
# 执行指标 (Breakout)
current_atr = talib.ATR(highs, lows, closes, self.atr_period)[-1]
keltner_ma = talib.MA(closes, timeperiod=self.keltner_period)
keltner_upper = keltner_ma[-1] + self.keltner_atr_multiplier * current_atr
keltner_lower = keltner_ma[-1] - self.keltner_atr_multiplier * current_atr
# --- 管理现有持仓 ---
position_volume = self.get_current_positions().get(self.symbol, 0)
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1], current_atr)
return
# --- 评估新机会 ---
self.evaluate_entry_signal(bar_history[-1], regime_ma, bb_width, keltner_upper, keltner_lower)
def manage_open_position(self, volume: int, current_bar: Bar, current_atr: float):
"""使用ATR追踪止盈。"""
# (此部分与V6/V7版本代码逻辑相同成熟且有效)
meta = self.position_meta.get(self.symbol);
if not meta: return
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"Initial Stop Loss hit at {initial_stop_price:.2f}");
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume));
return
trailing_stop_level = meta.get('trailing_stop_level', None)
if volume > 0:
meta['high_water_mark'] = max(meta.get('high_water_mark', meta['entry_price']), current_bar.high)
new_trailing_stop = meta['high_water_mark'] - self.trailing_stop_atr_multiplier * current_atr
if trailing_stop_level is None or new_trailing_stop > trailing_stop_level: trailing_stop_level = new_trailing_stop
trailing_stop_level = max(trailing_stop_level, initial_stop_price)
if current_bar.low <= trailing_stop_level: self.log(
f"ATR Trailing Stop hit for LONG at {trailing_stop_level:.2f}"); self.close_position("CLOSE_LONG",
abs(volume))
elif volume < 0:
meta['low_water_mark'] = min(meta.get('low_water_mark', meta['entry_price']), current_bar.low)
new_trailing_stop = meta['low_water_mark'] + self.trailing_stop_atr_multiplier * current_atr
if trailing_stop_level is None or new_trailing_stop < trailing_stop_level: trailing_stop_level = new_trailing_stop
trailing_stop_level = min(trailing_stop_level, initial_stop_price)
if current_bar.high >= trailing_stop_level: self.log(
f"ATR Trailing Stop hit for SHORT at {trailing_stop_level:.2f}"); self.close_position("CLOSE_SHORT",
abs(volume))
meta['trailing_stop_level'] = trailing_stop_level
def evaluate_entry_signal(self, current_bar: Bar, regime_ma: float, bb_width: float, keltner_upper: float,
keltner_lower: float):
"""执行“盘整-突破”的入场逻辑。"""
if len(self._bbw_history) < self.bbw_lookback: return
# 【战术层】检查是否进入“盘整准备”状态
bbw_threshold = np.percentile(list(self._bbw_history), self.bbw_squeeze_percentile)
if bb_width < bbw_threshold:
self._in_squeeze_ready_state = True
self.log(f"Squeeze Detected! BBW {bb_width:.4f} < Threshold {bbw_threshold:.4f}. Ready for breakout.")
return # 进入准备状态,等待突破
# 【执行层】在准备状态下,检查突破信号
if self._in_squeeze_ready_state:
direction = None
# 战略过滤 + 突破确认
if "BUY" in self.order_direction and current_bar.close > regime_ma and current_bar.close > keltner_upper and \
self.indicators[0].is_condition_met(*self.get_indicator_tuple()):
direction = "BUY"
elif "SELL" in self.order_direction and current_bar.close < regime_ma and current_bar.close < keltner_lower and \
self.indicators[1].is_condition_met(*self.get_indicator_tuple()):
direction = "SELL"
if direction:
self.log(f"Squeeze Fired! Direction: {direction}. Price broke Keltner Channel. Entering at market.")
# 市价单入场
entry_price = current_bar.close
current_atr = talib.ATR(
np.array([b.high for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
np.array([b.low for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
np.array([b.close for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
self.atr_period
)[-1]
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 = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
# 重置状态机
self._in_squeeze_ready_state = False
# --- 订单发送与仓位管理辅助函数 ---
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE");
if self.symbol in self.position_meta: del self.position_meta[self.symbol]
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 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._bbw_history.clear();
self._in_squeeze_ready_state = False;
self.log("Rollover detected. All strategy states have been reset.")

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,63 @@
# 卡尔曼策略Kalman Strategy
本策略文件夹包含基于卡尔曼滤波器技术的交易策略实现,主要包括以下 Python 文件:
## 📌 策略文件概述
### 1. `KalmanStrategy.py`
#### 策略概述
- **核心思想**: 解决动能策略在“回调中吐出利润、震荡中持续亏损”的问题,通过动态识别市场状态,在有利的环境下交易,并使用结构化的方法进行风险管理。
- **主要组成部分**:
- **状态过滤**: 使用ATR历史分位数构建“波动率状态机”在低波动率的“震荡区”主动休眠只在高波动率的“趋势区”寻找机会。
- **动能催化**: 沿用卡尔曼滤波器估算内在趋势当价格以ATR标准化的力量“逃逸”出内在趋势时视为入场信号。
- **结构化持仓**: 使用基于卡尔曼滤波线本身的“结构化止损”,给趋势以充分的“呼吸空间”,旨在持有完整的波段,避免在健康回调中被过早洗出。
#### 关键参数
- `kalman_process_noise`: 卡尔曼滤波器过程噪声。
- `kalman_measurement_noise`: 卡尔曼滤波器测量噪声。
- `atr_period`: ATR计算周期。
- `atr_lookback`: 用于计算ATR分位数的历史窗口。
- `atr_percentile_threshold`: ATR必须高于其历史的哪个百分位才认为是“趋势区”。
- `entry_threshold_atr`: 入场阈值。
- `initial_stop_atr_multiplier`: 初始止损的ATR乘数。
- `structural_stop_atr_multiplier`: 结构化止损的ATR乘数。
- `order_direction`: 交易方向(买入/卖出)。
- `indicators`: 使用的指标列表。
### 2. `KalmanStrategy2.py`
#### 策略概述
- **核心思想**: 通过一个核心参数 `strategy_mode`,在两种市场范式间切换,以适应不同品种的内在“性格”。
- **主要组成部分**:
- **人格1: 'TREND' (趋势模式)**:
- 哲学: 价格的强力突破会引发自我强化的趋势。
- 入场: 价格向上/下“逃逸”卡尔曼线 -> 做多/做空。
- 出场: 使用结构化卡尔曼止损,让利润奔跑。
- **人格2: 'REVERSION' (均值回归模式)**:
- 哲学: 价格的极端偏离是不可持续的,终将回归均值。
- 入场: 价格向上/下极端偏离(超买/超卖) -> 做空/做多。
- 出场: 当价格回归至卡尔曼线时获利了结。
#### 关键参数
- `strategy_mode`: 核心参数,可选 'TREND' 或 'REVERSION'。
- `kalman_process_noise`: 卡尔曼滤波器过程噪声。
- `kalman_measurement_noise`: 卡尔曼滤波器测量噪声。
- `atr_period`: ATR计算周期。
- `atr_lookback`: 用于计算ATR分位数的历史窗口。
- `atr_percentile_threshold`: ATR必须高于其历史的哪个百分位才认为是“趋势区”。
- `entry_threshold_atr`: 入场阈值。
- `initial_stop_atr_multiplier`: 初始止损的ATR乘数。
- `structural_stop_atr_multiplier`: 结构化止损的ATR乘数。
- `order_direction`: 交易方向(买入/卖出)。
- `indicators`: 使用的指标列表。
## 📌 文件区别
### `KalmanStrategy.py`
- 实现了一个自适应波段策略 (`AdaptiveKalmanStrategy`),主要用于解决动能策略在回调和震荡市场中的问题。
- 使用卡尔曼滤波器来识别市场状态和内在趋势并结合ATR指标进行风险管理。
### `KalmanStrategy2.py`
- 实现了一个双模式卡尔曼策略 (`DualModeKalmanStrategy`),可以在两种不同的市场模式之间切换:趋势模式和均值回归模式。
- 通过一个核心参数 `strategy_mode` 来决定使用哪种模式,并相应地调整入场和出场逻辑。

View File

@@ -0,0 +1,200 @@
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
# =============================================================================
# 策略实现 (Adaptive Kalman Strategy)
# =============================================================================
class AdaptiveKalmanStrategy(Strategy):
"""
一个基于市场状态识别与结构化风控的自适应波段策略。
本策略旨在解决动能策略在“回调中吐出利润、震荡中持续亏损”的核心痛点。
它通过动态识别市场状态,在有利的环境下交易,并使用结构化的方法持有仓位。
1. 【状态过滤】: 使用ATR历史分位数构建“波动率状态机”在低波动率的
“震荡区”主动休眠,只在高波动率的“趋势区”寻找机会。
2. 【动能催化】: 沿用卡尔曼滤波器估算内在趋势当价格以ATR标准化的
力量“逃逸”出内在趋势时,视为入场信号。
3. 【结构化持仓】: 抛弃紧跟价格峰值的传统追踪止损,改用基于卡尔曼滤波线
本身的“结构化止损”,给趋势以充分的“呼吸空间”,旨在持有
一个完整的波段,避免在健康回调中被过早洗出。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【信号层】卡尔曼滤波器参数 ---
kalman_process_noise: float = 0.01,
kalman_measurement_noise: float = 0.5,
# --- 【状态过滤】波动率状态机 ---
atr_period: int = 20,
atr_lookback: int = 100, # 用于计算ATR分位数的历史窗口
atr_percentile_threshold: float = 25.0, # ATR必须高于其历史的哪个百分位才认为是“趋势区”
# --- 【执行与风控】 ---
entry_threshold_atr: float = 2.5,
initial_stop_atr_multiplier: float = 2.0,
structural_stop_atr_multiplier: float = 2.5, # 结构化止损的ATR乘数
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']
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.main_symbol = main_symbol
self.order_id_counter = 0
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
self.log("AdaptiveKalmanStrategy Initialized.")
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
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_close = closes[-1]
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 = current_close
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 * (current_close - x_hat_minus)
self.P = (1 - K) * P_minus
kalman_price = self.x_hat
# --- 管理现有持仓 ---
position_volume = self.get_current_positions().get(self.symbol, 0)
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:
# self.log(f"Market in Chop Zone. ATR {current_atr:.4f} < Threshold {atr_threshold:.4f}. Standing by.")
return # 市场处于震荡区,休眠
# --- 评估新机会 ---
self.evaluate_entry_signal(bar_history[-1], kalman_price, current_atr)
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
# 阶段一:始终检查初始硬止损
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"Phase 1: Initial Stop Loss hit at {initial_stop_price:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
return
# 阶段二:检查结构化卡尔曼止损
structural_stop_price = 0
if volume > 0: # 多头持仓
structural_stop_price = kalman_price - self.structural_stop_atr_multiplier * current_atr
# 确保结构止损不会比初始止损更差
structural_stop_price = max(structural_stop_price, initial_stop_price)
if current_bar.low <= structural_stop_price:
self.log(f"Phase 2: Structural Kalman Stop hit for LONG at {structural_stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
elif volume < 0: # 空头持仓
structural_stop_price = kalman_price + self.structural_stop_atr_multiplier * current_atr
# 确保结构止损不会比初始止损更差
structural_stop_price = min(structural_stop_price, initial_stop_price)
if current_bar.high >= structural_stop_price:
self.log(f"Phase 2: Structural Kalman Stop hit for SHORT at {structural_stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
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 "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 = "SELL"
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 = "BUY"
if direction:
self.log(
f"Trend Zone Active! Momentum 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 = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
# --- 订单发送与仓位管理辅助函数 ---
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta: del self.position_meta[self.symbol]
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 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.log("Rollover detected. All strategy states have been reset.")

View File

@@ -0,0 +1,260 @@
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' (趋势模式)
- 哲学: 价格的强力突破会引发自我强化的趋势。
- 入场: 价格向上/下“逃逸”卡尔曼线 -> 做多/做空。
- 出场: 使用结构化卡尔曼止损,让利润奔跑。
人格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.log(f'manage_open_position')
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.log(f'evaluate_entry_signal')
self.evaluate_entry_signal(bar_history[-1], kalman_price, current_atr)
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
# --- 通用逻辑:首先检查初始硬止损 ---
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"Initial Stop Loss hit at {initial_stop_price:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
return
# --- 分模式出场逻辑 ---
if self.strategy_mode == 'TREND':
# 【趋势模式】: 使用结构化卡尔曼止损
stop_price = 0
if volume > 0:
stop_price = max(kalman_price - self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.low <= stop_price:
self.log(f"TREND Mode: Structural Stop hit for LONG at {stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
else: # volume < 0
stop_price = min(kalman_price + self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.high >= stop_price:
self.log(f"TREND Mode: Structural Stop hit for SHORT at {stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
elif self.strategy_mode == 'REVERSION':
# 【回归模式】: 检查是否触及均值作为止盈
if volume > 0:
stop_price = max(kalman_price - self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.low <= stop_price:
self.log(f"TREND Mode: Structural Stop hit for LONG at {stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
else: # volume < 0
stop_price = min(kalman_price + self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.high >= stop_price:
self.log(f"TREND Mode: Structural Stop hit for SHORT at {stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
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 + (1 if direction == "BUY" else -1)
stop_loss_price = entry_price - self.initial_stop_atr_multiplier * current_atr if direction in ["BUY",
"CLOSE_SHORT"] else entry_price + self.initial_stop_atr_multiplier * current_atr
# stop_loss_price = min(stop_loss_price, entry_price - 30) if direction == 'BUY' else max(stop_loss_price, entry_price + 30)
meta = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_limit_order(entry_price, direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
# ... (辅助函数 close_position, send_market_order, on_rollover 保持不变) ...
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta:
self.position_meta = {}
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}_MARKET_{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.log("Rollover detected. All strategy states have been reset.")

View File

@@ -0,0 +1,177 @@
import numpy as np
import talib
from typing import Optional, Any, List, Dict
from src.core_data import Bar, Order
from src.strategies.base_strategy import Strategy
class TVDZScoreStrategy(Strategy):
"""
内嵌 TVD (Condat 算法) + Z-Score ATR 的趋势突破策略。
无任何外部依赖(如 pytv纯 NumPy 实现。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
tvd_lam: float = 50.0,
atr_window: int = 14,
z_window: int = 100,
vol_threshold: float = -0.5,
entry_threshold_atr: float = 3.0,
stop_atr_multiplier: float = 3.0,
order_direction: Optional[List[str]] = None,
):
super().__init__(context, main_symbol, enable_log)
self.trade_volume = trade_volume
self.order_direction = order_direction or ["BUY", "SELL"]
self.tvd_lam = tvd_lam
self.atr_window = atr_window
self.z_window = z_window
self.vol_threshold = vol_threshold
self.entry_threshold_atr = entry_threshold_atr
self.stop_atr_multiplier = stop_atr_multiplier
self.position_meta: Dict[str, Any] = self.context.load_state()
self.main_symbol = main_symbol
self.order_id_counter = 0
self.log(f"TVDZScoreStrategy Initialized | λ={tvd_lam}, VolThresh={vol_threshold}")
@staticmethod
def _tvd_condat(y, lam):
"""Condat's O(N) TVD algorithm."""
n = y.size
if n == 0:
return y.copy()
x = y.astype(np.float64)
k = 0
k0 = 0
vmin = x[0] - lam
vmax = x[0] + lam
for i in range(1, n):
if x[i] < vmin:
while k < i:
x[k] = vmin
k += 1
k0 = i
vmin = x[i] - lam
vmax = x[i] + lam
elif x[i] > vmax:
while k < i:
x[k] = vmax
k += 1
k0 = i
vmin = x[i] - lam
vmax = x[i] + lam
else:
vmin = max(vmin, x[i] - lam)
vmax = min(vmax, x[i] + lam)
if vmin > vmax:
k = k0
s = np.sum(x[k0:i+1])
s /= (i - k0 + 1)
x[k0:i+1] = s
k = i + 1
k0 = k
if k0 < n:
vmin = x[k0] - lam
vmax = x[k0] + lam
while k < n:
x[k] = vmin
k += 1
return x
def _compute_zscore_atr_last(self, high, low, close) -> float:
n = len(close)
min_req = self.atr_window + self.z_window - 1
if n < min_req:
return np.nan
start = max(0, n - (self.z_window + self.atr_window))
seg_h, seg_l, seg_c = high[start:], low[start:], close[start:]
atr_full = talib.ATR(seg_h, seg_l, seg_c, timeperiod=self.atr_window)
atr_valid = atr_full[self.atr_window - 1:]
if len(atr_valid) < self.z_window:
return np.nan
window_atr = atr_valid[-self.z_window:]
mu = np.mean(window_atr)
sigma = np.std(window_atr)
last_atr = window_atr[-1]
return (last_atr - mu) / sigma if sigma > 1e-12 else 0.0
def on_open_bar(self, open_price: float, symbol: str):
self.symbol = symbol
bar_history = self.get_bar_history()
if len(bar_history) < max(100, self.atr_window + self.z_window):
return
closes = np.array([b.close for b in bar_history], dtype=np.float64)
highs = np.array([b.high for b in bar_history], dtype=np.float64)
lows = np.array([b.low for b in bar_history], dtype=np.float64)
# === TVD 平滑 ===
tvd_prices = self._tvd_condat(closes, self.tvd_lam)
tvd_price = tvd_prices[-1]
# === Z-Score ATR ===
current_atr = talib.ATR(highs, lows, closes, timeperiod=self.atr_window)[-1]
if current_atr <= 0:
return
deviation = closes[-1] - tvd_price
deviation_in_atr = deviation / current_atr
position_volume = self.get_current_positions().get(self.symbol, 0)
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1], current_atr, tvd_price)
return
direction = None
if "BUY" in self.order_direction and deviation_in_atr > self.entry_threshold_atr:
direction = "BUY"
elif "SELL" in self.order_direction and deviation_in_atr < -self.entry_threshold_atr:
direction = "SELL"
if direction:
self.log(f"Signal Fired | Dir: {direction}, Dev: {deviation_in_atr:.2f} ATR")
entry_price = closes[-1]
stop_loss = (
entry_price - self.stop_atr_multiplier * current_atr
if direction == "BUY"
else entry_price + self.stop_atr_multiplier * current_atr
)
meta = {"entry_price": entry_price, "stop_loss": stop_loss}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
def manage_open_position(self, volume: int, current_bar: Bar, current_atr: float, tvd_price: float):
meta = self.position_meta.get(self.symbol)
if not meta:
return
stop_loss = meta["stop_loss"]
if (volume > 0 and current_bar.low <= stop_loss) or (volume < 0 and current_bar.high >= stop_loss):
self.log(f"Stop Loss Hit at {stop_loss:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta:
del self.position_meta[self.symbol]
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 on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
self.position_meta = {}
self.log("Rollover: Strategy state reset.")

View File

@@ -0,0 +1,246 @@
import numpy as np
import pandas as pd
import talib
from collections import deque
from typing import Optional, Any, List, Dict, Tuple
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
# =============================================================================
# 策略实现 (ImbalanceZFlowStrategy)
# =============================================================================
class ImbalanceZFlowStrategy(Strategy):
"""
一个基于稳态核心指标“Z-Flow”的纯粹动量策略。
核心哲学:
1. 根本性解决稳态问题先将原始imbalance序列通过Z-Score转化为
稳态序列Z_I(t),所有后续分析都基于此坚实基础。
2. 核心驱动 Z_Flow: 对稳态的Z_I(t)序列计算MACD以度量“统计
显著性”的动量,形成一个高质量的稳态振荡器。
3. 真实的回测逻辑所有开仓和风险计算都严格基于当前bar的open_price
杜绝前视偏差。
4. 极致简洁:策略由单一指标、两个对称阈值驱动,逻辑纯粹,鲁棒性强。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【霍克斯过程参数】 ---
hawkes_lookback: int = 60,
hawkes_alpha: float = 0.8,
hawkes_beta: float = 0.2,
# --- 【Z-Score 标准化参数】 ---
z_score_period: int = 200,
# --- 【Z-Flow 指标参数】 ---
z_flow_fast_ema_period: int = 12,
z_flow_slow_ema_period: int = 26,
# --- 【交易阈值】 ---
entry_threshold: float = 0.5,
exit_threshold: float = 0.1,
# --- 【风险管理】 ---
atr_period: int = 20,
stop_loss_atr_multiplier: float = 3.0,
# --- 其他 ---
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']
# --- 参数赋值 ---
self.trade_volume = trade_volume
self.hawkes_lookback = hawkes_lookback
self.hawkes_alpha = hawkes_alpha
self.hawkes_beta = hawkes_beta
self.z_score_period = z_score_period
self.z_flow_fast_ema_period = z_flow_fast_ema_period
self.z_flow_slow_ema_period = z_flow_slow_ema_period
self.entry_threshold = entry_threshold
self.exit_threshold = exit_threshold
self.atr_period = atr_period
self.stop_loss_atr_multiplier = stop_loss_atr_multiplier
self.order_direction = order_direction
# --- 内部状态变量 ---
self._imbalance_history: deque = deque(maxlen=self.z_score_period + self.z_flow_slow_ema_period)
# 核心指标
self.z_flow = 0.0
self.prev_z_flow = 0.0
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"ImbalanceZFlowStrategy Initialized")
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()
required_bars = self._imbalance_history.maxlen + self.hawkes_lookback + 5
if len(bar_history) < required_bars:
return
self.prev_z_flow = self.z_flow
self._update_indicators_and_state(bar_history)
position_volume = self.get_current_positions().get(self.symbol, 0)
self._sync_position_state(position_volume, symbol)
if not self.trading: return
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1])
else:
self.evaluate_entry_signal(open_price, bar_history) # 传入open_price
def _update_indicators_and_state(self, bar_history: List[Bar]):
"""核心计算函数I(t) -> Z_I(t) -> Z_Flow(t)"""
# --- 1. 计算瞬时不均衡力量 I(t) ---
long_events, short_events = [], []
lookback_bars = bar_history[-self.hawkes_lookback:]
for i, bar in enumerate(lookback_bars):
event_age = len(lookback_bars) - 1 - i
mark = (bar.high - bar.low) * bar.volume
if mark > 0:
if bar.close > bar.open:
long_events.append({'age': event_age, 'mark': mark})
elif bar.close < bar.open:
short_events.append({'age': event_age, 'mark': mark})
lambda_long = sum(self.hawkes_alpha * e['mark'] * np.exp(-self.hawkes_beta * e['age']) for e in long_events)
lambda_short = sum(self.hawkes_alpha * e['mark'] * np.exp(-self.hawkes_beta * e['age']) for e in short_events)
imbalance = lambda_long - lambda_short
self._imbalance_history.append(imbalance)
if len(self._imbalance_history) < self.z_score_period:
self.z_flow = 0.0
return
imbalance_series = pd.Series(list(self._imbalance_history))
# --- 2. Z-Score 标准化,得到稳态序列 Z_I(t) ---
rolling_mean = imbalance_series.rolling(window=self.z_score_period).mean()
rolling_std = imbalance_series.rolling(window=self.z_score_period).std()
# 避免除以零并处理初始NaN值
if rolling_std.iloc[-1] > 1e-9:
z_score_series = (imbalance_series - rolling_mean) / rolling_std
else:
z_score_series = pd.Series(0.0, index=imbalance_series.index)
z_score_series.fillna(0.0, inplace=True)
# --- 3. 对稳态序列 Z_I(t) 计算MACD得到 Z_Flow ---
fast_ema = z_score_series.ewm(span=self.z_flow_fast_ema_period, adjust=False).mean()
slow_ema = z_score_series.ewm(span=self.z_flow_slow_ema_period, adjust=False).mean()
self.z_flow = fast_ema.iloc[-1] - slow_ema.iloc[-1]
def manage_open_position(self, volume: int, current_bar: Bar):
"""由Z-Flow驱动的统一平仓逻辑并辅以价格止损。"""
meta = self.position_meta.get(self.symbol)
if not meta: return
is_long = volume > 0
# 1. 价格硬止损
stop_loss_price = meta['stop_loss_price']
if (is_long and current_bar.low <= stop_loss_price) or \
(not is_long and current_bar.high >= stop_loss_price):
self.log(f"ATR Stop Loss Hit at {stop_loss_price:.4f}")
self.close_position("CLOSE_LONG" if is_long else "CLOSE_SHORT", abs(volume))
return
# 2. Z_Flow 驱动的平仓 (动能耗散)
exit_triggered = False
if is_long and self.z_flow < self.exit_threshold:
exit_triggered = True
elif not is_long and self.z_flow > -self.exit_threshold:
exit_triggered = True
if exit_triggered:
self.log(f"Z-Flow Dissipation Exit. Direction: {'LONG' if is_long else 'SHORT'}. "
f"Z-Flow ({self.z_flow:.2f}) returned to neutral zone.")
self.close_position("CLOSE_LONG" if is_long else "CLOSE_SHORT", abs(volume))
def evaluate_entry_signal(self, open_price: float, bar_history: List[Bar]):
"""当Z-Flow穿越入场阈值时在open_price开仓。"""
direction = None
if "BUY" in self.order_direction and self.prev_z_flow < self.entry_threshold <= self.z_flow:
direction = "BUY"
elif "SELL" in self.order_direction and self.prev_z_flow > -self.entry_threshold >= self.z_flow:
direction = "SELL"
if direction:
self.log(
f"Z-Flow Signal: {direction}. Z-Flow: {self.z_flow:.2f} crossed threshold. Entry on Open: {open_price}")
# 使用完整的bar_history计算ATR
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]
if current_atr <= 0: return
# ** 关键修正:基于 open_price 计算止损 **
stop_loss_price = open_price - self.stop_loss_atr_multiplier * current_atr if direction == "BUY" \
else open_price + self.stop_loss_atr_multiplier * current_atr
# ** 关键修正:记录 open_price 为入场价 **
meta = {'entry_price': open_price, 'stop_loss_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
# ... (辅助函数保持不变) ...
def _sync_position_state(self, position_volume, symbol):
meta = self.position_meta.get(symbol)
if position_volume != 0 and not meta:
self.log(f"警告:持仓({position_volume})与策略状态不一致!将强制平仓。", level='WARNING')
self.close_position("CLOSE_LONG" if position_volume > 0 else "CLOSE_SHORT", abs(position_volume))
return
if position_volume == 0 and meta:
self.log(f"信息:清理过时的策略状态。", level='INFO')
self.position_meta.pop(symbol, None)
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:
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 on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
self.position_meta = {}
self._imbalance_history.clear()
self.z_flow = 0.0
self.prev_z_flow = 0.0
self.log("Rollover detected. All strategy states have been reset.")

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,113 @@
import multiprocessing
from typing import Tuple, Dict, Any, Optional
from src.analysis.result_analyzer import ResultAnalyzer
from src.backtest_engine import BacktestEngine
from src.data_manager import DataManager
# --- 单个回测任务函数 ---
# 这个函数将在每个独立的进程中运行,因此它必须是自包含的
def run_single_backtest(
combination: Tuple[float, float], # 传入当前参数组合
common_config: Dict[str, Any] # 传入公共配置 (如数据路径, 初始资金等)
) -> Optional[Dict[str, Any]]:
"""
运行单个参数组合的回测任务。
此函数将在一个独立的进程中执行。
"""
p1_value, p2_value = combination
# 从 common_config 中获取必要的配置
symbol = common_config['symbol']
data_path = common_config['data_path']
initial_capital = common_config['initial_capital']
slippage_rate = common_config['slippage_rate']
commission_rate = common_config['commission_rate']
start_time = common_config['start_time']
end_time = common_config['end_time']
roll_over_mode = common_config['roll_over_mode']
# bar_duration_seconds = common_config['bar_duration_seconds'] # 如果DataManager需要可以再传
param1_name = common_config['param1_name']
param2_name = common_config['param2_name']
# 每个进程内部独立初始化 DataManager 和 BacktestEngine
# 确保每个进程有自己的数据副本和模拟状态
data_manager = DataManager(
file_path=data_path,
symbol=symbol,
# bar_duration_seconds=bar_duration_seconds, # 如果DataManager需要根据数据文件路径推断或者额外参数传入
# start_date=start_time.date(), # DataManager 现在通过 file_path 和 symbol 处理数据
# end_date=end_time.date(),
)
# data_manager.load_data() # DataManager 内部加载数据
strategy_parameters = {
'main_symbol': common_config['main_symbol'],
'trade_volume': 1,
param1_name: p1_value, # 15分钟扫荡K线下影线占其总范围的最小比例。
param2_name: p2_value, # 15分钟限价单的入场点位于扫荡K线低点到收盘价的斐波那契回撤比例。
'order_direction': common_config['order_direction'],
'enable_log': False, # 建议在调试和测试时开启日志
}
if 'strategy_mode' in common_config:
strategy_parameters['strategy_mode'] = common_config['strategy_mode']
if 'kalman_measurement_noise' in common_config:
strategy_parameters['kalman_measurement_noise'] = common_config['kalman_measurement_noise']
if 'entry_threshold_atr' in common_config:
strategy_parameters['entry_threshold_atr'] = common_config['entry_threshold_atr']
# 打印当前进程正在处理的组合信息
# 注意:多进程打印会交错显示
# print(f"--- 正在运行组合: {strategy_parameters} (PID: {multiprocessing.current_process().pid}) ---")
try:
# 初始化回测引擎
engine = BacktestEngine(
data_manager=data_manager,
strategy_class=common_config['strategy'],
strategy_params=strategy_parameters,
initial_capital=initial_capital,
slippage_rate=slippage_rate,
commission_rate=commission_rate,
roll_over_mode=True, # 保持换月模式
start_time=common_config['start_time'],
end_time=common_config['end_time']
)
# 运行回测,传入时间范围
engine.run_backtest()
# 获取回测结果并分析
results = engine.get_backtest_results()
portfolio_snapshots = results["portfolio_snapshots"]
trade_history = results["trade_history"]
bars = results["all_bars"]
initial_capital_result = results["initial_capital"]
if portfolio_snapshots:
analyzer = ResultAnalyzer(portfolio_snapshots, trade_history, bars, initial_capital_result)
# analyzer.generate_report()
# analyzer.plot_performance()
metrics = analyzer.calculate_all_metrics()
# 将当前组合的参数和性能指标存储起来
result_entry = {**strategy_parameters, **metrics}
return result_entry
else:
print(
f" 组合 {strategy_parameters} 没有生成投资组合快照,无法进行结果分析。(PID: {multiprocessing.current_process().pid})")
# 返回一个包含参数和默认0值的结果以便追踪失败组合
return {**strategy_parameters, "total_return": 0.0, "annualized_return": 0.0, "sharpe_ratio": 0.0,
"max_drawdown": 0.0, "error": "No portfolio snapshots"}
except Exception as e:
import traceback
error_trace = traceback.format_exc()
print(
f" 组合 {strategy_parameters} 运行失败: {e}\n{error_trace} (PID: {multiprocessing.current_process().pid})")
# 返回错误信息,以便后续处理
return {**strategy_parameters, "error": str(e), "traceback": error_trace}

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,284 @@
import numpy as np
from scipy.signal import stft
from datetime import datetime, timedelta
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, NormalizedATR, AtrVolatility, ZScoreATR
from src.strategies.base_strategy import Strategy
# =============================================================================
# 策略实现 (SpectralTrendStrategy)
# =============================================================================
class SpectralTrendStrategy(Strategy):
"""
频域能量相变策略 - 捕获肥尾趋势
核心哲学:
1. 显式傅里叶变换: 直接分离低频(趋势)、高频(噪音)能量
2. 相变临界点: 仅当低频能量占比 > 阈值时入场
3. 低频交易: 每月仅2-5次信号持仓数日捕获肥尾
4. 完全参数化: 无硬编码,适配任何市场时间结构
参数说明:
- bars_per_day: 市场每日K线数量 (e.g., 23 for 15min US markets)
- low_freq_days: 低频定义下限 (天), 默认2.0
- high_freq_days: 高频定义上限 (天), 默认1.0
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【市场结构参数】 ---
bars_per_day: int = 23, # 关键: 适配23根/天的市场
# --- 【频域核心参数】 ---
spectral_window_days: float = 2.0, # STFT窗口大小(天)
low_freq_days: float = 2.0, # 低频下限(天)
high_freq_days: float = 1.0, # 高频上限(天)
trend_strength_threshold: float = 0.1, # 相变临界值
exit_threshold: float = 0.4, # 退出阈值
# --- 【持仓管理】 ---
max_hold_days: int = 10, # 最大持仓天数
# --- 其他 ---
order_direction: Optional[List[str]] = None,
indicators: Optional[List[Indicator]] = None,
model_indicator: Indicator = None,
):
super().__init__(context, main_symbol, enable_log)
if order_direction is None:
order_direction = ['BUY', 'SELL']
if indicators is None:
indicators = [Empty(), Empty()] # 保持兼容性
# --- 参数赋值 (完全参数化) ---
self.trade_volume = trade_volume
self.bars_per_day = bars_per_day
self.spectral_window_days = spectral_window_days
self.low_freq_days = low_freq_days
self.high_freq_days = high_freq_days
self.trend_strength_threshold = trend_strength_threshold
self.exit_threshold = exit_threshold
self.max_hold_days = max_hold_days
self.order_direction = order_direction
if model_indicator is None:
model_indicator = Empty()
self.model_indicator = model_indicator
# --- 动态计算参数 ---
self.spectral_window = int(self.spectral_window_days * self.bars_per_day)
# 确保窗口大小为偶数 (STFT要求)
self.spectral_window = self.spectral_window if self.spectral_window % 2 == 0 else self.spectral_window + 1
# 频率边界 (cycles/day)
self.low_freq_bound = 1.0 / self.low_freq_days if self.low_freq_days > 0 else float('inf')
self.high_freq_bound = 1.0 / self.high_freq_days if self.high_freq_days > 0 else 0.0
# --- 内部状态变量 ---
self.main_symbol = main_symbol
self.order_id_counter = 0
self.indicators = indicators
self.entry_time = None # 入场时间
self.position_direction = None # 'LONG' or 'SHORT'
self.last_trend_strength = 0.0
self.last_dominant_freq = 0.0 # 主导周期(天)
self.log(f"SpectralTrendStrategy Initialized (bars/day={bars_per_day}, window={self.spectral_window} bars)")
def on_open_bar(self, open_price: float, symbol: str):
"""每根K线开盘时被调用"""
self.symbol = symbol
bar_history = self.get_bar_history()
current_time = self.get_current_time()
self.cancel_all_pending_orders(self.main_symbol)
# 需要足够的数据 (STFT窗口 + 缓冲)
if len(bar_history) < self.spectral_window + 10:
if self.enable_log and len(bar_history) % 50 == 0:
self.log(f"Waiting for {len(bar_history)}/{self.spectral_window + 10} bars")
return
position_volume = self.get_current_positions().get(self.symbol, 0)
# 获取历史价格 (使用完整历史)
closes = np.array([b.close for b in bar_history], dtype=float)
# 【核心】计算频域趋势强度 (显式傅里叶)
trend_strength, dominant_freq = self.calculate_trend_strength(closes)
self.last_trend_strength = trend_strength
self.last_dominant_freq = dominant_freq
# 检查最大持仓时间 (防止极端事件)
if self.entry_time and (current_time - self.entry_time) >= timedelta(days=self.max_hold_days):
self.log(f"Max hold time reached ({self.max_hold_days} days). Forcing exit.")
self.close_all_positions()
self.entry_time = None
self.position_direction = None
return
# 核心逻辑:相变入场/退出
if position_volume == 0:
self.evaluate_entry_signal(open_price, trend_strength, dominant_freq)
else:
self.manage_open_position(position_volume, trend_strength, dominant_freq)
def calculate_trend_strength(self, prices: np.array) -> (float, float):
"""
【显式傅里叶】计算低频能量占比 (完全参数化)
步骤:
1. 价格归一化 (窗口内)
2. 短时傅里叶变换 (STFT) - 采样率=bars_per_day
3. 动态计算频段边界 (基于bars_per_day)
4. 趋势强度 = 低频能量 / (低频+高频能量)
"""
# 1. 验证数据长度
if len(prices) < self.spectral_window:
return 0.0, 0.0
# 2. 价格归一化 (仅使用窗口内数据)
window_data = prices[-self.spectral_window:]
normalized = (window_data - np.mean(window_data)) / (np.std(window_data) + 1e-8)
# 3. STFT (采样率=bars_per_day)
try:
# fs: 每天的样本数 (bars_per_day)
f, t, Zxx = stft(
normalized,
fs=self.bars_per_day, # 关键: 适配市场结构
nperseg=self.spectral_window,
noverlap=max(0, self.spectral_window // 2),
boundary=None,
padded=False
)
except Exception as e:
self.log(f"STFT calculation error: {str(e)}")
return 0.0, 0.0
# 4. 过滤无效频率 (STFT返回频率范围: 0 到 fs/2)
valid_mask = (f >= 0) & (f <= self.bars_per_day / 2)
f = f[valid_mask]
Zxx = Zxx[valid_mask, :]
if Zxx.size == 0 or Zxx.shape[1] == 0:
return 0.0, 0.0
# 5. 计算最新时间点的能量
current_energy = np.abs(Zxx[:, -1]) ** 2
# 6. 动态频段定义 (cycles/day)
# 低频: 周期 > low_freq_days → 频率 < 1/low_freq_days
low_freq_mask = f < self.low_freq_bound
# 高频: 周期 < high_freq_days → 频率 > 1/high_freq_days
high_freq_mask = f > self.high_freq_bound
# 7. 能量计算
low_energy = np.sum(current_energy[low_freq_mask]) if np.any(low_freq_mask) else 0.0
high_energy = np.sum(current_energy[high_freq_mask]) if np.any(high_freq_mask) else 0.0
total_energy = low_energy + high_energy + 1e-8 # 防除零
# 8. 趋势强度 = 低频能量占比
trend_strength = low_energy / total_energy
# 9. 计算主导趋势周期 (天)
dominant_freq = 0.0
if np.any(low_freq_mask) and low_energy > 0:
# 找到低频段最大能量对应的频率
low_energies = current_energy[low_freq_mask]
max_idx = np.argmax(low_energies)
dominant_freq = 1.0 / (f[low_freq_mask][max_idx] + 1e-8) # 转换为周期(天)
return trend_strength, dominant_freq
def evaluate_entry_signal(self, open_price: float, trend_strength: float, dominant_freq: float):
"""评估相变入场信号"""
# 仅当趋势强度跨越临界点且有明确周期时入场
self.log(
f"Strength={trend_strength:.2f}")
if (trend_strength > self.trend_strength_threshold
and self.model_indicator.is_condition_met(*self.get_indicator_tuple())):
direction = None
indicator = self.model_indicator
# 做多信号: 价格在窗口均值上方
closes = np.array([b.close for b in self.get_bar_history()[-self.spectral_window:]], dtype=float)
if "BUY" in self.order_direction and np.mean(closes[-5:]) > np.mean(closes):
direction = "BUY" if indicator.is_condition_met(*self.get_indicator_tuple()) else "SELL"
# 做空信号: 价格在窗口均值下方
elif "SELL" in self.order_direction and np.mean(closes[-5:]) < np.mean(closes):
direction = "SELL" if indicator.is_condition_met(*self.get_indicator_tuple()) else "BUY"
if direction:
self.send_limit_order(direction, open_price, self.trade_volume, "OPEN")
self.entry_time = self.get_current_time()
self.position_direction = "LONG" if direction == "BUY" else "SHORT"
def manage_open_position(self, volume: int, trend_strength: float, dominant_freq: float):
"""管理持仓:仅当相变逆转时退出"""
# 相变逆转条件: 趋势强度 < 退出阈值
if trend_strength < self.exit_threshold:
direction = "CLOSE_LONG" if volume > 0 else "CLOSE_SHORT"
self.log(f"Phase Transition Exit: {direction} | Strength={trend_strength:.2f} < {self.exit_threshold}")
self.close_position(direction, abs(volume))
self.entry_time = None
self.position_direction = None
# --- 辅助函数区 ---
def close_all_positions(self):
"""强制平仓所有头寸"""
positions = self.get_current_positions()
if self.symbol in positions and positions[self.symbol] != 0:
direction = "CLOSE_LONG" if positions[self.symbol] > 0 else "CLOSE_SHORT"
self.close_position(direction, abs(positions[self.symbol]))
self.log(f"Forced exit of {abs(positions[self.symbol])} contracts")
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
def send_market_order(self, direction: str, volume: int, offset: str):
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, direction: str, limit_price: float, volume: int, offset: str):
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="LIMIT",
submitted_time=self.get_current_time(),
offset=offset,
limit_price=limit_price
)
self.send_order(order)
def on_init(self):
super().on_init()
self.cancel_all_pending_orders(self.main_symbol)
self.log("Strategy initialized. Waiting for phase transition signals...")
def on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
self.log(f"Rollover from {old_symbol} to {new_symbol}. Resetting position state.")
self.entry_time = None
self.position_direction = None
self.last_trend_strength = 0.0

View File

@@ -0,0 +1,377 @@
import numpy as np
from scipy.signal import stft
from datetime import datetime, timedelta
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
# =============================================================================
# 策略实现 (VolatilityAdaptiveSpectralStrategy)
# =============================================================================
class SpectralTrendStrategy(Strategy):
"""
波动率自适应频域趋势策略
核心哲学:
1. 显式傅里叶变换: 分离低频(趋势)、高频(噪音)能量
2. 波动率条件信号: 根据波动率环境动态调整交易方向
- 低波动环境: 趋势策略 (高趋势强度 → 延续)
- 高波动环境: 反转策略 (高趋势强度 → 反转)
3. 无硬编码参数: 所有阈值通过配置参数设定
4. 严格无未来函数: 所有计算使用历史数据
参数说明:
- bars_per_day: 市场每日K线数量
- volatility_lookback: 波动率计算窗口(天)
- low_vol_threshold: 低波动环境阈值(0-1)
- high_vol_threshold: 高波动环境阈值(0-1)
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【市场结构参数】 ---
bars_per_day: int = 23, # 适配23根/天的市场
# --- 【频域核心参数】 ---
spectral_window_days: float = 2.0, # STFT窗口大小(天)
low_freq_days: float = 2.0, # 低频下限(天)
high_freq_days: float = 1.0, # 高频上限(天)
trend_strength_threshold: float = 0.8, # 趋势强度阈值
exit_threshold: float = 0.5, # 退出阈值
# --- 【波动率参数】 ---
volatility_lookback_days: float = 5.0, # 波动率计算窗口(天)
low_vol_threshold: float = 0.3, # 低波动环境阈值(0-1)
high_vol_threshold: float = 0.7, # 高波动环境阈值(0-1)
# --- 【持仓管理】 ---
max_hold_days: int = 10, # 最大持仓天数
# --- 其他 ---
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 indicators is None:
indicators = [Empty(), Empty()] # 保持兼容性
# --- 参数赋值 (完全参数化) ---
self.trade_volume = trade_volume
self.bars_per_day = bars_per_day
self.spectral_window_days = spectral_window_days
self.low_freq_days = low_freq_days
self.high_freq_days = high_freq_days
self.trend_strength_threshold = trend_strength_threshold
self.exit_threshold = exit_threshold
self.volatility_lookback_days = volatility_lookback_days
self.low_vol_threshold = low_vol_threshold
self.high_vol_threshold = high_vol_threshold
self.max_hold_days = max_hold_days
self.order_direction = order_direction
# --- 动态计算参数 ---
self.spectral_window = int(self.spectral_window_days * self.bars_per_day)
self.spectral_window = self.spectral_window if self.spectral_window % 2 == 0 else self.spectral_window + 1
self.volatility_window = int(self.volatility_lookback_days * self.bars_per_day)
# 频率边界 (cycles/day)
self.low_freq_bound = 1.0 / self.low_freq_days if self.low_freq_days > 0 else float('inf')
self.high_freq_bound = 1.0 / self.high_freq_days if self.high_freq_days > 0 else 0.0
# --- 内部状态变量 ---
self.main_symbol = main_symbol
self.order_id_counter = 0
self.indicators = indicators
self.entry_time = None # 入场时间
self.position_direction = None # 'LONG' or 'SHORT'
self.last_trend_strength = 0.0
self.last_dominant_freq = 0.0 # 主导周期(天)
self.last_volatility = 0.0 # 标准化波动率(0-1)
self.volatility_history = [] # 存储历史波动率
self.log(f"VolatilityAdaptiveSpectralStrategy Initialized (bars/day={bars_per_day}, "
f"window={self.spectral_window} bars, vol_window={self.volatility_window} bars)")
def on_open_bar(self, open_price: float, symbol: str):
"""每根K线开盘时被调用"""
self.symbol = symbol
bar_history = self.get_bar_history()
current_time = self.get_current_time()
# 需要足够的数据 (最大窗口 + 缓冲)
min_required = max(self.spectral_window, self.volatility_window) + 10
if len(bar_history) < min_required:
if self.enable_log and len(bar_history) % 50 == 0:
self.log(f"Waiting for {len(bar_history)}/{min_required} bars")
return
position_volume = self.get_current_positions().get(self.symbol, 0)
# 获取必要历史价格 (仅取所需部分)
recent_bars = bar_history[-(max(self.spectral_window, self.volatility_window) + 5):]
closes = np.array([b.close for b in recent_bars], dtype=np.float32)
highs = np.array([b.high for b in recent_bars], dtype=np.float32)
lows = np.array([b.low for b in recent_bars], dtype=np.float32)
# 【核心】计算频域趋势强度 (显式傅里叶)
trend_strength, dominant_freq = self.calculate_trend_strength(closes)
self.last_trend_strength = trend_strength
self.last_dominant_freq = dominant_freq
# 【核心】计算标准化波动率 (0-1范围)
volatility = self.calculate_normalized_volatility(highs, lows, closes)
self.last_volatility = volatility
# 检查最大持仓时间 (防止极端事件)
if self.entry_time and (current_time - self.entry_time) >= timedelta(days=self.max_hold_days):
self.log(f"Max hold time reached ({self.max_hold_days} days). Forcing exit.")
self.close_all_positions()
self.entry_time = None
self.position_direction = None
return
# 核心逻辑:相变入场/退出
if position_volume == 0:
self.evaluate_entry_signal(open_price, trend_strength, dominant_freq, volatility, recent_bars)
else:
self.manage_open_position(position_volume, trend_strength, volatility)
def calculate_trend_strength(self, closes: np.array) -> (float, float):
"""
【显式傅里叶】计算低频能量占比 (完全参数化)
"""
if len(closes) < self.spectral_window:
return 0.0, 0.0
# 仅使用窗口内数据
window_data = closes[-self.spectral_window:]
window_mean = np.mean(window_data)
window_std = np.std(window_data)
if window_std < 1e-8:
return 0.0, 0.0
normalized = (window_data - window_mean) / window_std
try:
f, t, Zxx = stft(
normalized,
fs=self.bars_per_day,
nperseg=self.spectral_window,
noverlap=max(0, self.spectral_window // 2),
boundary=None,
padded=False
)
except Exception as e:
self.log(f"STFT calculation error: {str(e)}")
return 0.0, 0.0
# 过滤无效频率
max_freq = self.bars_per_day / 2
valid_mask = (f >= 0) & (f <= max_freq)
if not np.any(valid_mask):
return 0.0, 0.0
f = f[valid_mask]
Zxx = Zxx[valid_mask, :]
if Zxx.size == 0 or Zxx.shape[1] == 0:
return 0.0, 0.0
# 计算最新时间点的能量
current_energy = np.abs(Zxx[:, -1]) ** 2
# 动态频段定义
low_freq_mask = f < self.low_freq_bound
high_freq_mask = f > self.high_freq_bound
# 能量计算
low_energy = np.sum(current_energy[low_freq_mask]) if np.any(low_freq_mask) else 0.0
high_energy = np.sum(current_energy[high_freq_mask]) if np.any(high_freq_mask) else 0.0
total_energy = low_energy + high_energy + 1e-8
# 趋势强度 = 低频能量占比
trend_strength = low_energy / total_energy
# 计算主导趋势周期 (天)
dominant_freq = 0.0
if np.any(low_freq_mask) and low_energy > 0:
low_energies = current_energy[low_freq_mask]
max_idx = np.argmax(low_energies)
dominant_freq = 1.0 / (f[low_freq_mask][max_idx] + 1e-8)
return float(trend_strength), float(dominant_freq)
def calculate_normalized_volatility(self, highs: np.array, lows: np.array, closes: np.array) -> float:
"""
计算标准化波动率 (0-1范围)
步骤:
1. 计算ATR (真实波幅)
2. 标准化ATR (除以价格)
3. 归一化到0-1范围 (基于历史波动率)
"""
if len(closes) < self.volatility_window + 1:
return 0.5 # 默认中性值
# 1. 计算真实波幅 (TR)
tr1 = highs[-self.volatility_window - 1:] - lows[-self.volatility_window - 1:]
tr2 = np.abs(highs[-self.volatility_window - 1:] - np.roll(closes, 1)[-self.volatility_window - 1:])
tr3 = np.abs(lows[-self.volatility_window - 1:] - np.roll(closes, 1)[-self.volatility_window - 1:])
tr = np.maximum(tr1, np.maximum(tr2, tr3))
# 2. 计算ATR
atr = np.mean(tr[-self.volatility_window:])
# 3. 标准化ATR (除以当前价格)
current_price = closes[-1]
normalized_atr = atr / current_price if current_price > 0 else 0.0
# 4. 归一化到0-1范围 (基于历史波动率)
self.volatility_history.append(normalized_atr)
if len(self.volatility_history) > 1000: # 保留1000个历史值
self.volatility_history.pop(0)
if len(self.volatility_history) < 50: # 需要足够历史数据
return 0.5
# 使用历史50-95百分位进行归一化
low_percentile = np.percentile(self.volatility_history, 50)
high_percentile = np.percentile(self.volatility_history, 95)
if high_percentile - low_percentile < 1e-8:
return 0.5
# 归一化到0-1范围
normalized_vol = (normalized_atr - low_percentile) / (high_percentile - low_percentile + 1e-8)
normalized_vol = max(0.0, min(1.0, normalized_vol)) # 限制在0-1范围内
return normalized_vol
def evaluate_entry_signal(self, open_price: float, trend_strength: float, dominant_freq: float,
volatility: float, recent_bars: List[Bar]):
"""评估波动率条件入场信号"""
# 仅当趋势强度跨越临界点且有明确周期时入场
if trend_strength > self.trend_strength_threshold:
direction = None
trade_type = ""
# 计算价格位置 (短期vs长期均值)
window_closes = np.array([b.close for b in recent_bars[-self.spectral_window:]], dtype=np.float32)
short_avg = np.mean(window_closes[-5:])
long_avg = np.mean(window_closes)
# 添加统计显著性过滤
if abs(short_avg - long_avg) < 0.0005 * long_avg:
return
# 【核心】根据波动率环境决定交易逻辑
if volatility < self.low_vol_threshold:
# 低波动环境: 趋势策略
trade_type = "TREND"
if "BUY" in self.order_direction and short_avg > long_avg:
direction = "BUY"
elif "SELL" in self.order_direction and short_avg < long_avg:
direction = "SELL"
elif volatility > self.high_vol_threshold:
# 高波动环境: 反转策略
trade_type = "REVERSAL"
if "BUY" in self.order_direction and short_avg < long_avg:
direction = "BUY" # 价格低于均值,预期回归
elif "SELL" in self.order_direction and short_avg > long_avg:
direction = "SELL" # 价格高于均值,预期反转
else:
# 中波动环境: 谨慎策略 (需要更强信号)
trade_type = "CAUTIOUS"
if trend_strength > 0.9 and "BUY" in self.order_direction and short_avg > long_avg:
direction = "BUY"
elif trend_strength > 0.9 and "SELL" in self.order_direction and short_avg < long_avg:
direction = "SELL"
if direction:
self.log(
f"Entry: {direction} | Type={trade_type} | Strength={trend_strength:.2f} | "
f"Volatility={volatility:.2f} | ShortAvg={short_avg:.4f} vs LongAvg={long_avg:.4f}"
)
self.send_market_order(direction, self.trade_volume, "OPEN")
self.entry_time = self.get_current_time()
self.position_direction = "LONG" if direction == "BUY" else "SHORT"
def manage_open_position(self, volume: int, trend_strength: float, volatility: float):
"""管理持仓:波动率条件退出"""
# 退出条件1: 趋势强度 < 退出阈值
if trend_strength < self.exit_threshold:
direction = "CLOSE_LONG" if volume > 0 else "CLOSE_SHORT"
self.log(f"Exit (Strength): {direction} | Strength={trend_strength:.2f} < {self.exit_threshold}")
self.close_position(direction, abs(volume))
self.entry_time = None
self.position_direction = None
return
# 退出条件2: 波动率环境突变 (从低波动变为高波动,或反之)
if self.position_direction == "LONG" and volatility > self.high_vol_threshold * 1.2:
# 多头仓位在波动率突增时退出
self.log(
f"Exit (Volatility Spike): CLOSE_LONG | Volatility={volatility:.2f} > {self.high_vol_threshold * 1.2:.2f}")
self.close_position("CLOSE_LONG", abs(volume))
self.entry_time = None
self.position_direction = None
elif self.position_direction == "SHORT" and volatility > self.high_vol_threshold * 1.2:
# 空头仓位在波动率突增时退出
self.log(
f"Exit (Volatility Spike): CLOSE_SHORT | Volatility={volatility:.2f} > {self.high_vol_threshold * 1.2:.2f}")
self.close_position("CLOSE_SHORT", abs(volume))
self.entry_time = None
self.position_direction = None
# --- 辅助函数区 ---
def close_all_positions(self):
"""强制平仓所有头寸"""
positions = self.get_current_positions()
if not positions or self.symbol not in positions or positions[self.symbol] == 0:
return
direction = "CLOSE_LONG" if positions[self.symbol] > 0 else "CLOSE_SHORT"
self.close_position(direction, abs(positions[self.symbol]))
if self.enable_log:
self.log(f"Closed {abs(positions[self.symbol])} contracts")
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
def send_market_order(self, direction: str, volume: int, offset: str):
order_id = f"{self.symbol}_{direction[-6:]}_{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 on_init(self):
super().on_init()
self.cancel_all_pending_orders(self.main_symbol)
if self.enable_log:
self.log("Strategy initialized. Waiting for volatility-adaptive signals...")
def on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
if self.enable_log:
self.log(f"Rollover: {old_symbol} -> {new_symbol}. Resetting state.")
self.entry_time = None
self.position_direction = None
self.last_trend_strength = 0.0
self.volatility_history = [] # 重置波动率历史

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,103 @@
import multiprocessing
from typing import Tuple, Dict, Any, Optional
from src.analysis.result_analyzer import ResultAnalyzer
from src.backtest_engine import BacktestEngine
from src.data_manager import DataManager
# --- 单个回测任务函数 ---
# 这个函数将在每个独立的进程中运行,因此它必须是自包含的
def run_single_backtest(
combination: Tuple[float, float], # 传入当前参数组合
common_config: Dict[str, Any] # 传入公共配置 (如数据路径, 初始资金等)
) -> Optional[Dict[str, Any]]:
"""
运行单个参数组合的回测任务。
此函数将在一个独立的进程中执行。
"""
p1_value, p2_value = combination
# 从 common_config 中获取必要的配置
symbol = common_config['symbol']
data_path = common_config['data_path']
initial_capital = common_config['initial_capital']
slippage_rate = common_config['slippage_rate']
commission_rate = common_config['commission_rate']
start_time = common_config['start_time']
end_time = common_config['end_time']
roll_over_mode = common_config['roll_over_mode']
# bar_duration_seconds = common_config['bar_duration_seconds'] # 如果DataManager需要可以再传
param1_name = common_config['param1_name']
param2_name = common_config['param2_name']
# 每个进程内部独立初始化 DataManager 和 BacktestEngine
# 确保每个进程有自己的数据副本和模拟状态
data_manager = DataManager(
file_path=data_path,
symbol=symbol,
# bar_duration_seconds=bar_duration_seconds, # 如果DataManager需要根据数据文件路径推断或者额外参数传入
# start_date=start_time.date(), # DataManager 现在通过 file_path 和 symbol 处理数据
# end_date=end_time.date(),
)
# data_manager.load_data() # DataManager 内部加载数据
strategy_parameters = {
'main_symbol': common_config['main_symbol'],
'trade_volume': 1,
param1_name: p1_value, # 15分钟扫荡K线下影线占其总范围的最小比例。
param2_name: p2_value, # 15分钟限价单的入场点位于扫荡K线低点到收盘价的斐波那契回撤比例。
'order_direction': common_config['order_direction'],
'enable_log': False, # 建议在调试和测试时开启日志
}
# 打印当前进程正在处理的组合信息
# 注意:多进程打印会交错显示
print(f"--- 正在运行组合: {strategy_parameters} (PID: {multiprocessing.current_process().pid}) ---")
try:
# 初始化回测引擎
engine = BacktestEngine(
data_manager=data_manager,
strategy_class=common_config['strategy'],
strategy_params=strategy_parameters,
initial_capital=initial_capital,
slippage_rate=slippage_rate,
commission_rate=commission_rate,
roll_over_mode=True, # 保持换月模式
start_time=common_config['start_time'],
end_time=common_config['end_time']
)
# 运行回测,传入时间范围
engine.run_backtest()
# 获取回测结果并分析
results = engine.get_backtest_results()
portfolio_snapshots = results["portfolio_snapshots"]
trade_history = results["trade_history"]
bars = results["all_bars"]
initial_capital_result = results["initial_capital"]
if portfolio_snapshots:
analyzer = ResultAnalyzer(portfolio_snapshots, trade_history, bars, initial_capital_result)
# analyzer.generate_report()
# analyzer.plot_performance()
metrics = analyzer.calculate_all_metrics()
# 将当前组合的参数和性能指标存储起来
result_entry = {**strategy_parameters, **metrics}
return result_entry
else:
print(
f" 组合 {strategy_parameters} 没有生成投资组合快照,无法进行结果分析。(PID: {multiprocessing.current_process().pid})")
# 返回一个包含参数和默认0值的结果以便追踪失败组合
return {**strategy_parameters, "total_return": 0.0, "annualized_return": 0.0, "sharpe_ratio": 0.0,
"max_drawdown": 0.0, "error": "No portfolio snapshots"}
except Exception as e:
import traceback
error_trace = traceback.format_exc()
print(
f" 组合 {strategy_parameters} 运行失败: {e}\n{error_trace} (PID: {multiprocessing.current_process().pid})")
# 返回错误信息,以便后续处理
return {**strategy_parameters, "error": str(e), "traceback": error_trace}

View File

@@ -0,0 +1,214 @@
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
# =============================================================================
# 策略实现 (V9 - 盘整突破版)
# =============================================================================
class ConsolidationBreakoutStrategy(Strategy):
"""
一个基于“盘整即突破之母”原则的终极波段策略。(V9)
本策略旨在通过量化的方法,识别市场能量的“压缩-释放”周期,
以捕捉趋势的主干行情,并严格规避无序的震荡。
1. 【战略层】: 使用长期均线定义宏观牛熊环境。
2. 【战术层】: 使用布林带宽度(BBW)的极度收缩,来识别“盘整准备”状态。
3. 【执行层】: 在准备状态下,等待价格“突破”凯尔特纳通道作为“大幅移动”的确认信号,
立即以市价入场并使用ATR动态追踪止盈来截取波段利润。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【战略层】环境定义 ---
regime_ma_period: int = 200, # 定义牛熊市的长期均线
# --- 【战术层】盘整收缩识别 (Squeeze) ---
bollinger_period: int = 20, # 布林带周期
bollinger_std: float = 2.0, # 布林带标准差
bbw_lookback: int = 252, # BBW历史分位数回看窗口
bbw_squeeze_percentile: float = 15.0, # 定义“极度收缩”的百分位
# --- 【执行层】突破与风控 ---
keltner_period: int = 20, # 凯尔特纳通道周期
keltner_atr_multiplier: float = 1.5, # 凯尔特纳通道ATR乘数
atr_period: int = 20,
trailing_stop_atr_multiplier: float = 3.0,
initial_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']
self.trade_volume = trade_volume
self.regime_ma_period = regime_ma_period
self.bollinger_period = bollinger_period
self.bollinger_std = bollinger_std
self.bbw_lookback = bbw_lookback
self.bbw_squeeze_percentile = bbw_squeeze_percentile
self.keltner_period = keltner_period
self.keltner_atr_multiplier = keltner_atr_multiplier
self.atr_period = atr_period
self.trailing_stop_atr_multiplier = trailing_stop_atr_multiplier
self.initial_stop_atr_multiplier = initial_stop_atr_multiplier
self.order_direction = order_direction
# 状态变量
self._last_order_id: Optional[str] = None
self.position_meta: Dict[str, Any] = {}
self._bbw_history: deque = deque(maxlen=self.bbw_lookback)
self._in_squeeze_ready_state: bool = False
self.main_symbol = main_symbol
self.order_id_counter = 0
if indicators is None:
indicators = [Empty(), Empty()]
self.indicators = indicators
self.log("ConsolidationBreakoutStrategy (V9) Initialized.")
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
bar_history = self.get_bar_history()
required_bars = max(self.regime_ma_period, self.bbw_lookback) + 1
if len(bar_history) < required_bars: 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)
# 战略指标
regime_ma = talib.MA(closes, timeperiod=self.regime_ma_period)[-1]
# 战术指标 (Squeeze)
bb_upper, bb_mid, bb_lower = talib.BBANDS(closes, timeperiod=self.bollinger_period, nbdevup=self.bollinger_std,
nbdevdn=self.bollinger_std)
bb_width = (bb_upper[-1] - bb_lower[-1]) / bb_mid[-1] if bb_mid[-1] > 0 else 0
self._bbw_history.append(bb_width)
# 执行指标 (Breakout)
current_atr = talib.ATR(highs, lows, closes, self.atr_period)[-1]
keltner_ma = talib.MA(closes, timeperiod=self.keltner_period)
keltner_upper = keltner_ma[-1] + self.keltner_atr_multiplier * current_atr
keltner_lower = keltner_ma[-1] - self.keltner_atr_multiplier * current_atr
# --- 管理现有持仓 ---
position_volume = self.get_current_positions().get(self.symbol, 0)
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1], current_atr)
return
# --- 评估新机会 ---
self.evaluate_entry_signal(bar_history[-1], regime_ma, bb_width, keltner_upper, keltner_lower)
def manage_open_position(self, volume: int, current_bar: Bar, current_atr: float):
"""使用ATR追踪止盈。"""
# (此部分与V6/V7版本代码逻辑相同成熟且有效)
meta = self.position_meta.get(self.symbol);
if not meta: return
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"Initial Stop Loss hit at {initial_stop_price:.2f}");
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume));
return
trailing_stop_level = meta.get('trailing_stop_level', None)
if volume > 0:
meta['high_water_mark'] = max(meta.get('high_water_mark', meta['entry_price']), current_bar.high)
new_trailing_stop = meta['high_water_mark'] - self.trailing_stop_atr_multiplier * current_atr
if trailing_stop_level is None or new_trailing_stop > trailing_stop_level: trailing_stop_level = new_trailing_stop
trailing_stop_level = max(trailing_stop_level, initial_stop_price)
if current_bar.low <= trailing_stop_level: self.log(
f"ATR Trailing Stop hit for LONG at {trailing_stop_level:.2f}"); self.close_position("CLOSE_LONG",
abs(volume))
elif volume < 0:
meta['low_water_mark'] = min(meta.get('low_water_mark', meta['entry_price']), current_bar.low)
new_trailing_stop = meta['low_water_mark'] + self.trailing_stop_atr_multiplier * current_atr
if trailing_stop_level is None or new_trailing_stop < trailing_stop_level: trailing_stop_level = new_trailing_stop
trailing_stop_level = min(trailing_stop_level, initial_stop_price)
if current_bar.high >= trailing_stop_level: self.log(
f"ATR Trailing Stop hit for SHORT at {trailing_stop_level:.2f}"); self.close_position("CLOSE_SHORT",
abs(volume))
meta['trailing_stop_level'] = trailing_stop_level
def evaluate_entry_signal(self, current_bar: Bar, regime_ma: float, bb_width: float, keltner_upper: float,
keltner_lower: float):
"""执行“盘整-突破”的入场逻辑。"""
if len(self._bbw_history) < self.bbw_lookback: return
# 【战术层】检查是否进入“盘整准备”状态
bbw_threshold = np.percentile(list(self._bbw_history), self.bbw_squeeze_percentile)
if bb_width < bbw_threshold:
self._in_squeeze_ready_state = True
self.log(f"Squeeze Detected! BBW {bb_width:.4f} < Threshold {bbw_threshold:.4f}. Ready for breakout.")
return # 进入准备状态,等待突破
# 【执行层】在准备状态下,检查突破信号
if self._in_squeeze_ready_state:
direction = None
# 战略过滤 + 突破确认
if "BUY" in self.order_direction and current_bar.close > regime_ma and current_bar.close > keltner_upper and \
self.indicators[0].is_condition_met(*self.get_indicator_tuple()):
direction = "BUY"
elif "SELL" in self.order_direction and current_bar.close < regime_ma and current_bar.close < keltner_lower and \
self.indicators[1].is_condition_met(*self.get_indicator_tuple()):
direction = "SELL"
if direction:
self.log(f"Squeeze Fired! Direction: {direction}. Price broke Keltner Channel. Entering at market.")
# 市价单入场
entry_price = current_bar.close
current_atr = talib.ATR(
np.array([b.high for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
np.array([b.low for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
np.array([b.close for b in self.get_bar_history()[-self.atr_period:]], dtype=float),
self.atr_period
)[-1]
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 = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
# 重置状态机
self._in_squeeze_ready_state = False
# --- 订单发送与仓位管理辅助函数 ---
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE");
if self.symbol in self.position_meta: del self.position_meta[self.symbol]
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 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._bbw_history.clear();
self._in_squeeze_ready_state = False;
self.log("Rollover detected. All strategy states have been reset.")

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,63 @@
# 卡尔曼策略Kalman Strategy
本策略文件夹包含基于卡尔曼滤波器技术的交易策略实现,主要包括以下 Python 文件:
## 📌 策略文件概述
### 1. `KalmanStrategy.py`
#### 策略概述
- **核心思想**: 解决动能策略在“回调中吐出利润、震荡中持续亏损”的问题,通过动态识别市场状态,在有利的环境下交易,并使用结构化的方法进行风险管理。
- **主要组成部分**:
- **状态过滤**: 使用ATR历史分位数构建“波动率状态机”在低波动率的“震荡区”主动休眠只在高波动率的“趋势区”寻找机会。
- **动能催化**: 沿用卡尔曼滤波器估算内在趋势当价格以ATR标准化的力量“逃逸”出内在趋势时视为入场信号。
- **结构化持仓**: 使用基于卡尔曼滤波线本身的“结构化止损”,给趋势以充分的“呼吸空间”,旨在持有完整的波段,避免在健康回调中被过早洗出。
#### 关键参数
- `kalman_process_noise`: 卡尔曼滤波器过程噪声。
- `kalman_measurement_noise`: 卡尔曼滤波器测量噪声。
- `atr_period`: ATR计算周期。
- `atr_lookback`: 用于计算ATR分位数的历史窗口。
- `atr_percentile_threshold`: ATR必须高于其历史的哪个百分位才认为是“趋势区”。
- `entry_threshold_atr`: 入场阈值。
- `initial_stop_atr_multiplier`: 初始止损的ATR乘数。
- `structural_stop_atr_multiplier`: 结构化止损的ATR乘数。
- `order_direction`: 交易方向(买入/卖出)。
- `indicators`: 使用的指标列表。
### 2. `KalmanStrategy2.py`
#### 策略概述
- **核心思想**: 通过一个核心参数 `strategy_mode`,在两种市场范式间切换,以适应不同品种的内在“性格”。
- **主要组成部分**:
- **人格1: 'TREND' (趋势模式)**:
- 哲学: 价格的强力突破会引发自我强化的趋势。
- 入场: 价格向上/下“逃逸”卡尔曼线 -> 做多/做空。
- 出场: 使用结构化卡尔曼止损,让利润奔跑。
- **人格2: 'REVERSION' (均值回归模式)**:
- 哲学: 价格的极端偏离是不可持续的,终将回归均值。
- 入场: 价格向上/下极端偏离(超买/超卖) -> 做空/做多。
- 出场: 当价格回归至卡尔曼线时获利了结。
#### 关键参数
- `strategy_mode`: 核心参数,可选 'TREND' 或 'REVERSION'。
- `kalman_process_noise`: 卡尔曼滤波器过程噪声。
- `kalman_measurement_noise`: 卡尔曼滤波器测量噪声。
- `atr_period`: ATR计算周期。
- `atr_lookback`: 用于计算ATR分位数的历史窗口。
- `atr_percentile_threshold`: ATR必须高于其历史的哪个百分位才认为是“趋势区”。
- `entry_threshold_atr`: 入场阈值。
- `initial_stop_atr_multiplier`: 初始止损的ATR乘数。
- `structural_stop_atr_multiplier`: 结构化止损的ATR乘数。
- `order_direction`: 交易方向(买入/卖出)。
- `indicators`: 使用的指标列表。
## 📌 文件区别
### `KalmanStrategy.py`
- 实现了一个自适应波段策略 (`AdaptiveKalmanStrategy`),主要用于解决动能策略在回调和震荡市场中的问题。
- 使用卡尔曼滤波器来识别市场状态和内在趋势并结合ATR指标进行风险管理。
### `KalmanStrategy2.py`
- 实现了一个双模式卡尔曼策略 (`DualModeKalmanStrategy`),可以在两种不同的市场模式之间切换:趋势模式和均值回归模式。
- 通过一个核心参数 `strategy_mode` 来决定使用哪种模式,并相应地调整入场和出场逻辑。

View File

@@ -0,0 +1,200 @@
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
# =============================================================================
# 策略实现 (Adaptive Kalman Strategy)
# =============================================================================
class AdaptiveKalmanStrategy(Strategy):
"""
一个基于市场状态识别与结构化风控的自适应波段策略。
本策略旨在解决动能策略在“回调中吐出利润、震荡中持续亏损”的核心痛点。
它通过动态识别市场状态,在有利的环境下交易,并使用结构化的方法持有仓位。
1. 【状态过滤】: 使用ATR历史分位数构建“波动率状态机”在低波动率的
“震荡区”主动休眠,只在高波动率的“趋势区”寻找机会。
2. 【动能催化】: 沿用卡尔曼滤波器估算内在趋势当价格以ATR标准化的
力量“逃逸”出内在趋势时,视为入场信号。
3. 【结构化持仓】: 抛弃紧跟价格峰值的传统追踪止损,改用基于卡尔曼滤波线
本身的“结构化止损”,给趋势以充分的“呼吸空间”,旨在持有
一个完整的波段,避免在健康回调中被过早洗出。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【信号层】卡尔曼滤波器参数 ---
kalman_process_noise: float = 0.01,
kalman_measurement_noise: float = 0.5,
# --- 【状态过滤】波动率状态机 ---
atr_period: int = 20,
atr_lookback: int = 100, # 用于计算ATR分位数的历史窗口
atr_percentile_threshold: float = 25.0, # ATR必须高于其历史的哪个百分位才认为是“趋势区”
# --- 【执行与风控】 ---
entry_threshold_atr: float = 2.5,
initial_stop_atr_multiplier: float = 2.0,
structural_stop_atr_multiplier: float = 2.5, # 结构化止损的ATR乘数
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']
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.main_symbol = main_symbol
self.order_id_counter = 0
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
self.log("AdaptiveKalmanStrategy Initialized.")
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
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_close = closes[-1]
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 = current_close
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 * (current_close - x_hat_minus)
self.P = (1 - K) * P_minus
kalman_price = self.x_hat
# --- 管理现有持仓 ---
position_volume = self.get_current_positions().get(self.symbol, 0)
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:
# self.log(f"Market in Chop Zone. ATR {current_atr:.4f} < Threshold {atr_threshold:.4f}. Standing by.")
return # 市场处于震荡区,休眠
# --- 评估新机会 ---
self.evaluate_entry_signal(bar_history[-1], kalman_price, current_atr)
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
# 阶段一:始终检查初始硬止损
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"Phase 1: Initial Stop Loss hit at {initial_stop_price:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
return
# 阶段二:检查结构化卡尔曼止损
structural_stop_price = 0
if volume > 0: # 多头持仓
structural_stop_price = kalman_price - self.structural_stop_atr_multiplier * current_atr
# 确保结构止损不会比初始止损更差
structural_stop_price = max(structural_stop_price, initial_stop_price)
if current_bar.low <= structural_stop_price:
self.log(f"Phase 2: Structural Kalman Stop hit for LONG at {structural_stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
elif volume < 0: # 空头持仓
structural_stop_price = kalman_price + self.structural_stop_atr_multiplier * current_atr
# 确保结构止损不会比初始止损更差
structural_stop_price = min(structural_stop_price, initial_stop_price)
if current_bar.high >= structural_stop_price:
self.log(f"Phase 2: Structural Kalman Stop hit for SHORT at {structural_stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
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 "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 = "SELL"
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 = "BUY"
if direction:
self.log(
f"Trend Zone Active! Momentum 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 = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
# --- 订单发送与仓位管理辅助函数 ---
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta: del self.position_meta[self.symbol]
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 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.log("Rollover detected. All strategy states have been reset.")

View File

@@ -0,0 +1,259 @@
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' (趋势模式)
- 哲学: 价格的强力突破会引发自我强化的趋势。
- 入场: 价格向上/下“逃逸”卡尔曼线 -> 做多/做空。
- 出场: 使用结构化卡尔曼止损,让利润奔跑。
人格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.log(f'manage_open_position')
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.log(f'evaluate_entry_signal')
self.evaluate_entry_signal(bar_history[-1], kalman_price, current_atr)
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
# --- 通用逻辑:首先检查初始硬止损 ---
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"Initial Stop Loss hit at {initial_stop_price:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
return
# --- 分模式出场逻辑 ---
if self.strategy_mode == 'TREND':
# 【趋势模式】: 使用结构化卡尔曼止损
stop_price = 0
if volume > 0:
stop_price = max(kalman_price - self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.low <= stop_price:
self.log(f"TREND Mode: Structural Stop hit for LONG at {stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
else: # volume < 0
stop_price = min(kalman_price + self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.high >= stop_price:
self.log(f"TREND Mode: Structural Stop hit for SHORT at {stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
elif self.strategy_mode == 'REVERSION':
# 【回归模式】: 检查是否触及均值作为止盈
if volume > 0:
stop_price = max(kalman_price - self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.low <= stop_price:
self.log(f"TREND Mode: Structural Stop hit for LONG at {stop_price:.4f}")
self.close_position("CLOSE_LONG", abs(volume))
else: # volume < 0
stop_price = min(kalman_price + self.structural_stop_atr_multiplier * current_atr, initial_stop_price)
self.log(f'stop_price: {stop_price:.4f}, kalman_price: {kalman_price:.4f}')
if current_bar.high >= stop_price:
self.log(f"TREND Mode: Structural Stop hit for SHORT at {stop_price:.4f}")
self.close_position("CLOSE_SHORT", abs(volume))
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 + (1 if direction == "BUY" else -1)
stop_loss_price = entry_price - self.initial_stop_atr_multiplier * current_atr if direction in ["BUY",
"CLOSE_SHORT"] else entry_price + self.initial_stop_atr_multiplier * current_atr
meta = {'entry_price': entry_price, 'initial_stop_price': stop_loss_price}
self.send_limit_order(entry_price, direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
# ... (辅助函数 close_position, send_market_order, on_rollover 保持不变) ...
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta:
self.position_meta = {}
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}_MARKET_{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.log("Rollover detected. All strategy states have been reset.")

View File

@@ -0,0 +1,177 @@
import numpy as np
import talib
from typing import Optional, Any, List, Dict
from src.core_data import Bar, Order
from src.strategies.base_strategy import Strategy
class TVDZScoreStrategy(Strategy):
"""
内嵌 TVD (Condat 算法) + Z-Score ATR 的趋势突破策略。
无任何外部依赖(如 pytv纯 NumPy 实现。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
tvd_lam: float = 50.0,
atr_window: int = 14,
z_window: int = 100,
vol_threshold: float = -0.5,
entry_threshold_atr: float = 3.0,
stop_atr_multiplier: float = 3.0,
order_direction: Optional[List[str]] = None,
):
super().__init__(context, main_symbol, enable_log)
self.trade_volume = trade_volume
self.order_direction = order_direction or ["BUY", "SELL"]
self.tvd_lam = tvd_lam
self.atr_window = atr_window
self.z_window = z_window
self.vol_threshold = vol_threshold
self.entry_threshold_atr = entry_threshold_atr
self.stop_atr_multiplier = stop_atr_multiplier
self.position_meta: Dict[str, Any] = self.context.load_state()
self.main_symbol = main_symbol
self.order_id_counter = 0
self.log(f"TVDZScoreStrategy Initialized | λ={tvd_lam}, VolThresh={vol_threshold}")
@staticmethod
def _tvd_condat(y, lam):
"""Condat's O(N) TVD algorithm."""
n = y.size
if n == 0:
return y.copy()
x = y.astype(np.float64)
k = 0
k0 = 0
vmin = x[0] - lam
vmax = x[0] + lam
for i in range(1, n):
if x[i] < vmin:
while k < i:
x[k] = vmin
k += 1
k0 = i
vmin = x[i] - lam
vmax = x[i] + lam
elif x[i] > vmax:
while k < i:
x[k] = vmax
k += 1
k0 = i
vmin = x[i] - lam
vmax = x[i] + lam
else:
vmin = max(vmin, x[i] - lam)
vmax = min(vmax, x[i] + lam)
if vmin > vmax:
k = k0
s = np.sum(x[k0:i+1])
s /= (i - k0 + 1)
x[k0:i+1] = s
k = i + 1
k0 = k
if k0 < n:
vmin = x[k0] - lam
vmax = x[k0] + lam
while k < n:
x[k] = vmin
k += 1
return x
def _compute_zscore_atr_last(self, high, low, close) -> float:
n = len(close)
min_req = self.atr_window + self.z_window - 1
if n < min_req:
return np.nan
start = max(0, n - (self.z_window + self.atr_window))
seg_h, seg_l, seg_c = high[start:], low[start:], close[start:]
atr_full = talib.ATR(seg_h, seg_l, seg_c, timeperiod=self.atr_window)
atr_valid = atr_full[self.atr_window - 1:]
if len(atr_valid) < self.z_window:
return np.nan
window_atr = atr_valid[-self.z_window:]
mu = np.mean(window_atr)
sigma = np.std(window_atr)
last_atr = window_atr[-1]
return (last_atr - mu) / sigma if sigma > 1e-12 else 0.0
def on_open_bar(self, open_price: float, symbol: str):
self.symbol = symbol
bar_history = self.get_bar_history()
if len(bar_history) < max(100, self.atr_window + self.z_window):
return
closes = np.array([b.close for b in bar_history], dtype=np.float64)
highs = np.array([b.high for b in bar_history], dtype=np.float64)
lows = np.array([b.low for b in bar_history], dtype=np.float64)
# === TVD 平滑 ===
tvd_prices = self._tvd_condat(closes, self.tvd_lam)
tvd_price = tvd_prices[-1]
# === Z-Score ATR ===
current_atr = talib.ATR(highs, lows, closes, timeperiod=self.atr_window)[-1]
if current_atr <= 0:
return
deviation = closes[-1] - tvd_price
deviation_in_atr = deviation / current_atr
position_volume = self.get_current_positions().get(self.symbol, 0)
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1], current_atr, tvd_price)
return
direction = None
if "BUY" in self.order_direction and deviation_in_atr > self.entry_threshold_atr:
direction = "BUY"
elif "SELL" in self.order_direction and deviation_in_atr < -self.entry_threshold_atr:
direction = "SELL"
if direction:
self.log(f"Signal Fired | Dir: {direction}, Dev: {deviation_in_atr:.2f} ATR")
entry_price = closes[-1]
stop_loss = (
entry_price - self.stop_atr_multiplier * current_atr
if direction == "BUY"
else entry_price + self.stop_atr_multiplier * current_atr
)
meta = {"entry_price": entry_price, "stop_loss": stop_loss}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
def manage_open_position(self, volume: int, current_bar: Bar, current_atr: float, tvd_price: float):
meta = self.position_meta.get(self.symbol)
if not meta:
return
stop_loss = meta["stop_loss"]
if (volume > 0 and current_bar.low <= stop_loss) or (volume < 0 and current_bar.high >= stop_loss):
self.log(f"Stop Loss Hit at {stop_loss:.4f}")
self.close_position("CLOSE_LONG" if volume > 0 else "CLOSE_SHORT", abs(volume))
def close_position(self, direction: str, volume: int):
self.send_market_order(direction, volume, offset="CLOSE")
if self.symbol in self.position_meta:
del self.position_meta[self.symbol]
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 on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
self.position_meta = {}
self.log("Rollover: Strategy state reset.")

View File

@@ -0,0 +1,246 @@
import numpy as np
import pandas as pd
import talib
from collections import deque
from typing import Optional, Any, List, Dict, Tuple
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
# =============================================================================
# 策略实现 (ImbalanceZFlowStrategy)
# =============================================================================
class ImbalanceZFlowStrategy(Strategy):
"""
一个基于稳态核心指标“Z-Flow”的纯粹动量策略。
核心哲学:
1. 根本性解决稳态问题先将原始imbalance序列通过Z-Score转化为
稳态序列Z_I(t),所有后续分析都基于此坚实基础。
2. 核心驱动 Z_Flow: 对稳态的Z_I(t)序列计算MACD以度量“统计
显著性”的动量,形成一个高质量的稳态振荡器。
3. 真实的回测逻辑所有开仓和风险计算都严格基于当前bar的open_price
杜绝前视偏差。
4. 极致简洁:策略由单一指标、两个对称阈值驱动,逻辑纯粹,鲁棒性强。
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
trade_volume: int,
# --- 【霍克斯过程参数】 ---
hawkes_lookback: int = 60,
hawkes_alpha: float = 0.8,
hawkes_beta: float = 0.2,
# --- 【Z-Score 标准化参数】 ---
z_score_period: int = 200,
# --- 【Z-Flow 指标参数】 ---
z_flow_fast_ema_period: int = 12,
z_flow_slow_ema_period: int = 26,
# --- 【交易阈值】 ---
entry_threshold: float = 0.5,
exit_threshold: float = 0.1,
# --- 【风险管理】 ---
atr_period: int = 20,
stop_loss_atr_multiplier: float = 3.0,
# --- 其他 ---
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']
# --- 参数赋值 ---
self.trade_volume = trade_volume
self.hawkes_lookback = hawkes_lookback
self.hawkes_alpha = hawkes_alpha
self.hawkes_beta = hawkes_beta
self.z_score_period = z_score_period
self.z_flow_fast_ema_period = z_flow_fast_ema_period
self.z_flow_slow_ema_period = z_flow_slow_ema_period
self.entry_threshold = entry_threshold
self.exit_threshold = exit_threshold
self.atr_period = atr_period
self.stop_loss_atr_multiplier = stop_loss_atr_multiplier
self.order_direction = order_direction
# --- 内部状态变量 ---
self._imbalance_history: deque = deque(maxlen=self.z_score_period + self.z_flow_slow_ema_period)
# 核心指标
self.z_flow = 0.0
self.prev_z_flow = 0.0
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"ImbalanceZFlowStrategy Initialized")
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()
required_bars = self._imbalance_history.maxlen + self.hawkes_lookback + 5
if len(bar_history) < required_bars:
return
self.prev_z_flow = self.z_flow
self._update_indicators_and_state(bar_history)
position_volume = self.get_current_positions().get(self.symbol, 0)
self._sync_position_state(position_volume, symbol)
if not self.trading: return
if position_volume != 0:
self.manage_open_position(position_volume, bar_history[-1])
else:
self.evaluate_entry_signal(open_price, bar_history) # 传入open_price
def _update_indicators_and_state(self, bar_history: List[Bar]):
"""核心计算函数I(t) -> Z_I(t) -> Z_Flow(t)"""
# --- 1. 计算瞬时不均衡力量 I(t) ---
long_events, short_events = [], []
lookback_bars = bar_history[-self.hawkes_lookback:]
for i, bar in enumerate(lookback_bars):
event_age = len(lookback_bars) - 1 - i
mark = (bar.high - bar.low) * bar.volume
if mark > 0:
if bar.close > bar.open:
long_events.append({'age': event_age, 'mark': mark})
elif bar.close < bar.open:
short_events.append({'age': event_age, 'mark': mark})
lambda_long = sum(self.hawkes_alpha * e['mark'] * np.exp(-self.hawkes_beta * e['age']) for e in long_events)
lambda_short = sum(self.hawkes_alpha * e['mark'] * np.exp(-self.hawkes_beta * e['age']) for e in short_events)
imbalance = lambda_long - lambda_short
self._imbalance_history.append(imbalance)
if len(self._imbalance_history) < self.z_score_period:
self.z_flow = 0.0
return
imbalance_series = pd.Series(list(self._imbalance_history))
# --- 2. Z-Score 标准化,得到稳态序列 Z_I(t) ---
rolling_mean = imbalance_series.rolling(window=self.z_score_period).mean()
rolling_std = imbalance_series.rolling(window=self.z_score_period).std()
# 避免除以零并处理初始NaN值
if rolling_std.iloc[-1] > 1e-9:
z_score_series = (imbalance_series - rolling_mean) / rolling_std
else:
z_score_series = pd.Series(0.0, index=imbalance_series.index)
z_score_series.fillna(0.0, inplace=True)
# --- 3. 对稳态序列 Z_I(t) 计算MACD得到 Z_Flow ---
fast_ema = z_score_series.ewm(span=self.z_flow_fast_ema_period, adjust=False).mean()
slow_ema = z_score_series.ewm(span=self.z_flow_slow_ema_period, adjust=False).mean()
self.z_flow = fast_ema.iloc[-1] - slow_ema.iloc[-1]
def manage_open_position(self, volume: int, current_bar: Bar):
"""由Z-Flow驱动的统一平仓逻辑并辅以价格止损。"""
meta = self.position_meta.get(self.symbol)
if not meta: return
is_long = volume > 0
# 1. 价格硬止损
stop_loss_price = meta['stop_loss_price']
if (is_long and current_bar.low <= stop_loss_price) or \
(not is_long and current_bar.high >= stop_loss_price):
self.log(f"ATR Stop Loss Hit at {stop_loss_price:.4f}")
self.close_position("CLOSE_LONG" if is_long else "CLOSE_SHORT", abs(volume))
return
# 2. Z_Flow 驱动的平仓 (动能耗散)
exit_triggered = False
if is_long and self.z_flow < self.exit_threshold:
exit_triggered = True
elif not is_long and self.z_flow > -self.exit_threshold:
exit_triggered = True
if exit_triggered:
self.log(f"Z-Flow Dissipation Exit. Direction: {'LONG' if is_long else 'SHORT'}. "
f"Z-Flow ({self.z_flow:.2f}) returned to neutral zone.")
self.close_position("CLOSE_LONG" if is_long else "CLOSE_SHORT", abs(volume))
def evaluate_entry_signal(self, open_price: float, bar_history: List[Bar]):
"""当Z-Flow穿越入场阈值时在open_price开仓。"""
direction = None
if "BUY" in self.order_direction and self.prev_z_flow < self.entry_threshold <= self.z_flow:
direction = "BUY"
elif "SELL" in self.order_direction and self.prev_z_flow > -self.entry_threshold >= self.z_flow:
direction = "SELL"
if direction:
self.log(
f"Z-Flow Signal: {direction}. Z-Flow: {self.z_flow:.2f} crossed threshold. Entry on Open: {open_price}")
# 使用完整的bar_history计算ATR
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]
if current_atr <= 0: return
# ** 关键修正:基于 open_price 计算止损 **
stop_loss_price = open_price - self.stop_loss_atr_multiplier * current_atr if direction == "BUY" \
else open_price + self.stop_loss_atr_multiplier * current_atr
# ** 关键修正:记录 open_price 为入场价 **
meta = {'entry_price': open_price, 'stop_loss_price': stop_loss_price}
self.send_market_order(direction, self.trade_volume, "OPEN", meta)
self.save_state(self.position_meta)
# ... (辅助函数保持不变) ...
def _sync_position_state(self, position_volume, symbol):
meta = self.position_meta.get(symbol)
if position_volume != 0 and not meta:
self.log(f"警告:持仓({position_volume})与策略状态不一致!将强制平仓。", level='WARNING')
self.close_position("CLOSE_LONG" if position_volume > 0 else "CLOSE_SHORT", abs(position_volume))
return
if position_volume == 0 and meta:
self.log(f"信息:清理过时的策略状态。", level='INFO')
self.position_meta.pop(symbol, None)
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:
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 on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)
self.position_meta = {}
self._imbalance_history.clear()
self.z_flow = 0.0
self.prev_z_flow = 0.0
self.log("Rollover detected. All strategy states have been reset.")

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,107 @@
import multiprocessing
from typing import Tuple, Dict, Any, Optional
from src.analysis.result_analyzer import ResultAnalyzer
from src.backtest_engine import BacktestEngine
from src.data_manager import DataManager
# --- 单个回测任务函数 ---
# 这个函数将在每个独立的进程中运行,因此它必须是自包含的
def run_single_backtest(
combination: Tuple[float, float], # 传入当前参数组合
common_config: Dict[str, Any] # 传入公共配置 (如数据路径, 初始资金等)
) -> Optional[Dict[str, Any]]:
"""
运行单个参数组合的回测任务。
此函数将在一个独立的进程中执行。
"""
p1_value, p2_value = combination
# 从 common_config 中获取必要的配置
symbol = common_config['symbol']
data_path = common_config['data_path']
initial_capital = common_config['initial_capital']
slippage_rate = common_config['slippage_rate']
commission_rate = common_config['commission_rate']
start_time = common_config['start_time']
end_time = common_config['end_time']
roll_over_mode = common_config['roll_over_mode']
# bar_duration_seconds = common_config['bar_duration_seconds'] # 如果DataManager需要可以再传
param1_name = common_config['param1_name']
param2_name = common_config['param2_name']
# 每个进程内部独立初始化 DataManager 和 BacktestEngine
# 确保每个进程有自己的数据副本和模拟状态
data_manager = DataManager(
file_path=data_path,
symbol=symbol,
# bar_duration_seconds=bar_duration_seconds, # 如果DataManager需要根据数据文件路径推断或者额外参数传入
# start_date=start_time.date(), # DataManager 现在通过 file_path 和 symbol 处理数据
# end_date=end_time.date(),
)
# data_manager.load_data() # DataManager 内部加载数据
strategy_parameters = {
'main_symbol': common_config['main_symbol'],
'trade_volume': 1,
param1_name: p1_value, # 15分钟扫荡K线下影线占其总范围的最小比例。
param2_name: p2_value, # 15分钟限价单的入场点位于扫荡K线低点到收盘价的斐波那契回撤比例。
'order_direction': common_config['order_direction'],
'enable_log': False, # 建议在调试和测试时开启日志
}
if 'strategy_mode' in common_config:
strategy_parameters['strategy_mode'] = common_config['strategy_mode']
# 打印当前进程正在处理的组合信息
# 注意:多进程打印会交错显示
# print(f"--- 正在运行组合: {strategy_parameters} (PID: {multiprocessing.current_process().pid}) ---")
try:
# 初始化回测引擎
engine = BacktestEngine(
data_manager=data_manager,
strategy_class=common_config['strategy'],
strategy_params=strategy_parameters,
initial_capital=initial_capital,
slippage_rate=slippage_rate,
commission_rate=commission_rate,
roll_over_mode=True, # 保持换月模式
start_time=common_config['start_time'],
end_time=common_config['end_time']
)
# 运行回测,传入时间范围
engine.run_backtest()
# 获取回测结果并分析
results = engine.get_backtest_results()
portfolio_snapshots = results["portfolio_snapshots"]
trade_history = results["trade_history"]
bars = results["all_bars"]
initial_capital_result = results["initial_capital"]
if portfolio_snapshots:
analyzer = ResultAnalyzer(portfolio_snapshots, trade_history, bars, initial_capital_result)
# analyzer.generate_report()
# analyzer.plot_performance()
metrics = analyzer.calculate_all_metrics()
# 将当前组合的参数和性能指标存储起来
result_entry = {**strategy_parameters, **metrics}
return result_entry
else:
print(
f" 组合 {strategy_parameters} 没有生成投资组合快照,无法进行结果分析。(PID: {multiprocessing.current_process().pid})")
# 返回一个包含参数和默认0值的结果以便追踪失败组合
return {**strategy_parameters, "total_return": 0.0, "annualized_return": 0.0, "sharpe_ratio": 0.0,
"max_drawdown": 0.0, "error": "No portfolio snapshots"}
except Exception as e:
import traceback
error_trace = traceback.format_exc()
print(
f" 组合 {strategy_parameters} 运行失败: {e}\n{error_trace} (PID: {multiprocessing.current_process().pid})")
# 返回错误信息,以便后续处理
return {**strategy_parameters, "error": str(e), "traceback": error_trace}

View File

@@ -0,0 +1,51 @@
import numpy as np
from scipy.signal import stft
from datetime import datetime, timedelta
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, NormalizedATR, AtrVolatility
from src.strategies.base_strategy import Strategy
# =============================================================================
# 策略实现 (SpectralTrendStrategy)
# =============================================================================
class TestConnectionStrategy(Strategy):
"""
频域能量相变策略 - 捕获肥尾趋势
核心哲学:
1. 显式傅里叶变换: 直接分离低频(趋势)、高频(噪音)能量
2. 相变临界点: 仅当低频能量占比 > 阈值时入场
3. 低频交易: 每月仅2-5次信号持仓数日捕获肥尾
4. 完全参数化: 无硬编码,适配任何市场时间结构
参数说明:
- bars_per_day: 市场每日K线数量 (e.g., 23 for 15min US markets)
- low_freq_days: 低频定义下限 (天), 默认2.0
- high_freq_days: 高频定义上限 (天), 默认1.0
"""
def __init__(
self,
context: Any,
main_symbol: str,
enable_log: bool,
):
super().__init__(context, main_symbol, enable_log)
# --- 内部状态变量 ---
self.main_symbol = main_symbol
def on_open_bar(self, open_price: float, symbol: str):
self.log(f'on open bar: {symbol}')
def on_init(self):
super().on_init()
def on_rollover(self, old_symbol: str, new_symbol: str):
super().on_rollover(old_symbol, new_symbol)