liaozhaorun/NewQuant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1、研究多空不对称策略

Browse Source
This commit is contained in:
liaozhaorun
2025-09-20 00:04:51 +08:00
parent 9a58fec9ca
commit bc93a547f0
15 changed files with 14240 additions and 3432 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
data/ analysis/OI.ipynb
View File

File diff suppressed because one or more lines are too long

327
data/ analysis/Trend.ipynb
View File

File diff suppressed because one or more lines are too long

6
src/execution_simulator.py
View File

@@ -228,7 +228,7 @@ class ExecutionSimulator:
else:
pnl_per_share = current_average_cost - fill_price
realized_pnl = pnl_per_share * min(abs(current_position), volume)
temp_cash += realized_pnl - trade_value
temp_cash -= trade_value
temp_positions[symbol] += volume
if temp_positions[symbol] == 0:
del temp_average_costs[symbol]
@@ -255,7 +255,8 @@ class ExecutionSimulator:
else:
pnl_per_share = fill_price - current_average_cost
realized_pnl = pnl_per_share * min(current_position, volume)
temp_cash += realized_pnl + trade_value
temp_cash += trade_value
# temp_cash += trade_value
temp_positions[symbol] -= volume
if temp_positions[symbol] == 0:
del temp_average_costs[symbol]
@@ -328,6 +329,7 @@ class ExecutionSimulator:
def get_portfolio_value(self, current_bar: Bar) -> float:
total_value = self.cash
for symbol, quantity in self.positions.items():
if symbol == current_bar.symbol:
total_value += quantity * current_bar.open

8
src/indicators/indicator_list.py
View File

@@ -23,12 +23,18 @@ INDICATOR_LIST = [
# DifferencedVolumeIndicator(shift_window=13),
# DifferencedVolumeIndicator(shift_window=20),
StochasticOscillator(fastk_period=14, slowd_period=3, slowk_period=3),
StochasticOscillator(fastk_period=5, slowd_period=3, slowk_period=3),
StochasticOscillator(5, 3, 3),
StochasticOscillator(fastk_period=21, slowd_period=5, slowk_period=5),
RateOfChange(5),
RateOfChange(10),
RateOfChange(15),
RateOfChange(20),
ROC_MA(5, 5),
ROC_MA(5, 10),
ROC_MA(10, 10),
ROC_MA(10, 20),
ROC_MA(20, 20),
ROC_MA(20, 40),
NormalizedATR(5),
NormalizedATR(14),
NormalizedATR(21),

70
src/indicators/indicators.py
View File

@@ -521,4 +521,72 @@ class RelativeVolumeInWindow(Indicator):
return relative_volume
def get_name(self) -> str:
return f"relative_volume_sma{self.n_period}_idx{self.impulse_index_from_end}"
return f"relative_volume_sma{self.n_period}_idx{self.impulse_index_from_end}"
class ROC_MA(Indicator):
"""
变动率的移动平均 (ROC_MA) 指标实现。
该指标首先计算ROC，然后对其结果应用移动平均，以获得更平滑的动量曲线。
"""
def __init__(
self,
roc_window: int = 60,
ma_window: int = 20,
down_bound: float = None,
up_bound: float = None,
shift_window: int = 0,
):
"""
初始化 ROC_MA 指标。
Args:
roc_window (int): 计算ROC所需的回看周期。
ma_window (int): 对ROC值进行平滑的移动平均周期。
down_bound (float): (可选) 用于条件判断的下轨。
up_bound (float): (可选) 用于条件判断的上轨。
shift_window (int): (可选) 指标值的时间偏移。
"""
# 【关键】调用父类的初始化方法
super().__init__(down_bound, up_bound)
self.roc_window = roc_window
self.ma_window = ma_window
self.shift_window = shift_window
def get_values(
self,
close: np.array,
open: np.array,
high: np.array,
low: np.array,
volume: np.array,
) -> np.array:
"""
根据收盘价列表计算 ROC_MA 值。
Args:
close (np.array): 收盘价列表。
其他 OHLCV 参数在此指标中不使用。
Returns:
np.array: ROC_MA 值列表。如果数据不足，则列表开头为NaN。
"""
# 步骤 1: 使用 talib.ROC 计算原始的ROC值
# TA-Lib 会在数据不足时自动填充 NaN
roc_values = talib.ROC(close, timeperiod=self.roc_window)
# 步骤 2: 对 roc_values 计算移动平均 (SMA)
# 注意：在计算MA之前，ROC已经产生了一些NaN，TA-Lib的MA函数会处理这些NaN
# 并产生更多的NaN，这是正常的。
roc_ma_values = talib.SMA(roc_values, timeperiod=self.ma_window)
# 返回最终的 numpy 数组
return roc_ma_values
def get_name(self) -> str:
"""
返回指标的唯一名称，用于标识和调试。
"""
return f"roc_ma_{self.roc_window}_{self.ma_window}"

1701
src/strategies/InformedTraderDPStrategy/InformedTraderDPStrategy2.ipynb
View File

File diff suppressed because one or more lines are too long

8872
src/strategies/InformedTraderDPStrategy/InformedTraderDPStrategy2_2.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

241
src/strategies/InformedTraderDPStrategy/InformedTraderDPStrategy2_2.py Normal file
View File

@@ -0,0 +1,241 @@
from collections import deque
from typing import List, Union, Optional
import numpy as np
from src.core_data import Bar, Order
from src.indicators.base_indicators import Indicator
from src.strategies.base_strategy import Strategy
from src.indicators.indicators import Empty
class AdaptiveInformedTraderDPStrategy(Strategy):
"""
机构级思路 · 自适应DP策略 v5.0 (集成ATR动态止损)
1. 核心信号: 动态规划计算的多空累积优势差
2. 核心优化: a) 动态阈值 b) 趋势增强
3. 风险管理: 【新增】ATR动态追踪止损，保护利润，控制回撤
4. 性能优化: 采用 deque 实现 O(1) 状态更新
"""
def __init__(self, context, main_symbol: str,
trade_volume: int = 1,
lookback_period: int = 120,
decay_factor: Union[float, List[float]] = 0.7,
dominance_atr_multiplier: Union[float, List[float]] = 1.5,
dominance_std_limit: Union[float, List[float]] = 0.01,
activity_atr_multiplier: Union[float, List[float]] = 20,
exit_atr_multiplier: Union[float, List[float]] = 0.5,
atr_period: int = 20,
# --- 【新增】ATR止损参数 ---
atr_stop_multiplier: float = 2.0,
order_direction: Optional[list] = None,
enable_log: bool = False,
indicators: List[Union[Indicator, List[Indicator]]] = None, ):
super().__init__(context, main_symbol, enable_log=enable_log)
def _to_two(v):
if isinstance(v, (int, float)): return [v, v]
if len(v) == 2: return [v[0], v[1]]
raise ValueError(f"{v} 需为单值或长度为 2 的列表")
self.decay_factor = _to_two(decay_factor)
self.dominance_atr_multiplier = _to_two(dominance_atr_multiplier)
self.dominance_std_limit = _to_two(dominance_std_limit)
self.activity_atr_multiplier = _to_two(activity_atr_multiplier)
self.exit_atr_multiplier = _to_two(exit_atr_multiplier)
self.lookback_period = lookback_period
self.atr_period = atr_period
# --- 保存ATR止损乘数 ---
self.atr_stop_multiplier = atr_stop_multiplier
self.symbol = main_symbol
self.volume = trade_volume
self.order_direction = order_direction or ['BUY', 'SELL']
self.min_bars_required = max(self.lookback_period, self.atr_period) + 2
# --- 高性能状态变量 ---
self.dp_long = 0.0
self.dp_short = 0.0
self.dominance_history = deque(maxlen=self.atr_period)
self.previous_bar: Optional[Bar] = None
self.volume_history = deque(maxlen=self.lookback_period)
self.price_history = deque(maxlen=self.lookback_period)
self.tr_history = deque(maxlen=self.atr_period)
self.atr = 0.0
# --- 【新增】ATR止损状态变量 ---
self.trailing_stop_price: Optional[float] = None
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
# on_open_bar 保持不变
def on_open_bar(self, open_price: float, symbol: str):
bars = self.get_bar_history()
if len(bars) < self.min_bars_required:
return
latest_bar = bars[-1] # 获取最新的K线
self.cancel_all_pending_orders(symbol)
self._update_dp_state(latest_bar)
pos = self.get_current_positions().get(symbol, 0)
if pos != 0:
# 【修改】将最新K线传入出场逻辑，用于止损判断
self._exit_logic(pos, latest_bar)
return
for side_idx, direction in enumerate([1, -1]):
if (direction > 0 and 'BUY' not in self.order_direction) or \
(direction < 0 and 'SELL' not in self.order_direction):
continue
# 【修改】将最新K线传入入场逻辑，用于设置初始止损
if self._entry_logic(side_idx, latest_bar):
break
# _update_dp_state 保持不变
def _update_dp_state(self, latest_bar: Bar):
current_volume = latest_bar.volume
if len(self.volume_history) == self.lookback_period:
recent_avg_volume = (sum(self.volume_history) - self.volume_history[
0] + current_volume) / self.lookback_period
else:
recent_avg_volume = np.mean(list(self.volume_history) + [current_volume])
self.volume_history.append(current_volume)
current_price = (latest_bar.close + latest_bar.high + latest_bar.low) / 3
if len(self.price_history) == self.lookback_period:
recent_avg_price = (sum(self.price_history) - self.price_history[
0] + current_price) / self.lookback_period
else:
recent_avg_price = np.mean(list(self.price_history) + [current_price])
self.price_history.append(current_price)
if self.previous_bar is None:
self.previous_bar = latest_bar
return
price_change = latest_bar.close - self.previous_bar.close
oi_change = getattr(latest_bar, 'close_oi', 0) - getattr(self.previous_bar, 'close_oi', 0)
epsilon = 1e-9
volume_factor = current_volume / (recent_avg_volume + epsilon)
volume_factor = np.clip(volume_factor, 0, 3)
price_factor = current_price / (recent_avg_price + epsilon)
price_factor = np.clip(price_factor, 0, 3)
bullish_base_score, bearish_base_score = 0, 0
if price_change > 0:
if oi_change > 0:
bullish_base_score = 2
elif oi_change < 0:
bullish_base_score = 1
elif price_change < 0:
if oi_change > 0:
bearish_base_score = 2
elif oi_change < 0:
bearish_base_score = 1
final_bullish_impulse = bullish_base_score * volume_factor
final_bearish_impulse = bearish_base_score * volume_factor
self.dp_long = self.dp_long * self.decay_factor[0] + final_bullish_impulse
self.dp_short = self.dp_short * self.decay_factor[1] + final_bearish_impulse
true_range = max(latest_bar.high - latest_bar.low, abs(latest_bar.high - self.previous_bar.close),
abs(latest_bar.low - self.previous_bar.close))
self.tr_history.append(true_range)
self.atr = np.mean(list(self.tr_history))
dominance = self.dp_long - self.dp_short
self.dominance_history.append(dominance)
self.log(
f"DP更新: Dominance={dominance:.2f}, ATR={self.atr:.2f}, dp_long={self.dp_long:.2f}, dp_short={self.dp_short:.2f}")
self.previous_bar = latest_bar
# --- 【核心修改】入场逻辑：成功下单后设置初始止损 ---
def _entry_logic(self, side_idx: int, latest_bar: Bar) -> bool:
# (前面的逻辑保持不变)
dominance = self.dp_long - self.dp_short
activity = self.dp_long + self.dp_short
if len(self.dominance_history) < self.atr_period: return False
dominance_std = np.std(list(self.dominance_history))
dynamic_dom_thresh = dominance_std * self.dominance_atr_multiplier[side_idx]
dynamic_act_thresh = dominance_std * self.activity_atr_multiplier[side_idx]
if activity < dynamic_act_thresh: return False
signal, side_str = False, ''
if side_idx == 0:
if dominance > dynamic_dom_thresh: signal, side_str = True, 'BUY'
else:
if dominance < -dynamic_dom_thresh: signal, side_str = True, 'SELL'
if not signal: return False
indicator_condition_met = self.indicators[side_idx].is_condition_met(*self.get_indicator_tuple())
if not indicator_condition_met: return False
# --- 执行下单 ---
self.log(
f"触发{side_str}信号: Dominance={dominance:.2f} > {dynamic_dom_thresh:.2f}, Activity={activity:.2f} > {dynamic_act_thresh:.2f}")
self._send_market_order(side_str, self.volume)
# --- 【新增】设置初始ATR止损价 ---
if side_str == 'BUY':
self.trailing_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
self.log(f"多头开仓，设置初始ATR止损价: {self.trailing_stop_price:.2f}")
elif side_str == 'SELL':
self.trailing_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
self.log(f"空头开仓，设置初始ATR止损价: {self.trailing_stop_price:.2f}")
return True
# --- 【核心修改】出场逻辑：整合ATR止损和DP信号 ---
def _exit_logic(self, pos: int, latest_bar: Bar):
"""
出场逻辑现在包含两种情况：
1. ATR追踪止损（优先，用于风险控制）
2. DP优势差回落（策略性离场）
"""
# --- 1. ATR追踪止损逻辑 ---
if pos > 0: # 持有多仓
# 更新追踪止损价：只上移，不下移
new_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
self.trailing_stop_price = max(self.trailing_stop_price, new_stop_price)
# 检查是否触发止损
if latest_bar.low <= self.trailing_stop_price:
self.log(f"触发多头ATR追踪止损: 最新价 {latest_bar.low:.2f} <= 止损价 {self.trailing_stop_price:.2f}")
self._send_market_order('CLOSE_LONG', abs(pos))
self.trailing_stop_price = None # 清空止损价
return # 止损优先，直接返回
elif pos < 0: # 持有空仓
# 更新追踪止损价：只下移，不上移
new_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
self.trailing_stop_price = min(self.trailing_stop_price, new_stop_price)
# 检查是否触发止损
if latest_bar.high >= self.trailing_stop_price:
self.log(f"触发空头ATR追踪止损: 最新价 {latest_bar.high:.2f} >= 止损价 {self.trailing_stop_price:.2f}")
self._send_market_order('CLOSE_SHORT', abs(pos))
self.trailing_stop_price = None # 清空止损价
return # 止损优先，直接返回
# --- 2. DP优势差回落逻辑 (如果ATR止损未触发) ---
dominance = self.dp_long - self.dp_short
side_idx = 0 if pos > 0 else 1
dominance_std = np.std(list(self.dominance_history))
dynamic_exit_thresh = dominance_std * self.exit_atr_multiplier[side_idx]
exit_signal, side_str = False, ''
if pos > 0 and dominance < dynamic_exit_thresh:
exit_signal, side_str = True, 'CLOSE_LONG'
elif pos < 0 and dominance > -dynamic_exit_thresh:
exit_signal, side_str = True, 'CLOSE_SHORT'
if exit_signal:
self.log(f"DP优势差回落，策略性离场: Dominance={dominance:.2f} < 阈值 {dynamic_exit_thresh:.2f}")
self._send_market_order(side_str, abs(pos))
self.trailing_stop_price = None # 清空止损价
def _send_market_order(self, direction: str, vol: int):
offset = 'OPEN' if direction in ('BUY', 'SELL') else 'CLOSE'
oid = f"{self.symbol}_{direction}_{self.get_current_time():%Y%m%d%H%M%S}"
self.send_order(
Order(id=oid, symbol=self.symbol, direction=direction, volume=vol, price_type='MARKET', offset=offset))

4162
src/strategies/InformedTraderDPStrategy/InformedTraderDPStrategy3.ipynb
View File

File diff suppressed because one or more lines are too long

206
src/strategies/InformedTraderDPStrategy/InformedTraderDPStrategy3.py
View File

@@ -1,5 +1,3 @@
# ... 省略头部和 __init__ 等未修改部分 ...
# ...
from collections import deque
from typing import List, Union, Optional
import numpy as np
@@ -9,56 +7,71 @@ from src.strategies.base_strategy import Strategy
from src.indicators.indicators import Empty
class AdaptiveInformedTraderDPStrategy(Strategy):
# ... 省略 import 和基类定义 ...
class ContextualDPStrategy(Strategy):
"""
机构级思路 · 自适应DP策略 v7.0 (最终版 · 连续加权)
1. 核心信号: 动态规划计算的多空累积优势差
2. 核心优化: a) 动态阈值 b) 连续加权的K线形态分析，无硬编码阈值
3. 风险管理: ATR动态追踪止损
4. 性能优化: 采用 deque 实现 O(1) 状态更新
机构级思路 · 上下文感知的DP策略 v9.0 (最终版)
将流动性真空作为上下文，非对称地融入DP状态计算，
以模拟知情交易者的最优操作。
"""
# ... __init__ 函数与 v8.0 基本相同，但参数名可以更清晰 ...
def __init__(self, context, main_symbol: str,
trade_volume: int = 1,
lookback_period: int = 120,
decay_factor: Union[float, List[float]] = 0.95,
dominance_atr_multiplier: Union[float, List[float]] = 1.5,
activity_atr_multiplier: Union[float, List[float]] = 2.0,
exit_atr_multiplier: Union[float, List[float]] = 0.5,
dominance_multiplier: Union[float, List[float]] = 1.1,
activity_multiplier: Union[float, List[float]] = [2.0, 1.8],
exit_multiplier: Union[float, List[float]] = [0.5, 0.6],
atr_period: int = 20,
atr_stop_multiplier: float = 2.0,
lvi_period: int = 50, # LVI现在是DP计算的一部分
order_direction: Optional[list] = None,
enable_log: bool = False,
indicators: List[Union[Indicator, List[Indicator]]] = None, ):
indicators: List[Union[Indicator, List[Indicator]]] = None,
):
super().__init__(context, main_symbol, enable_log=enable_log)
def _to_two(v):
def _to_two(v, default_v):
if isinstance(v, (int, float)): return [v, v]
if len(v) == 2: return [v[0], v[1]]
if v is None: return [default_v, default_v]
if len(v) == 2: return v
raise ValueError(f"{v} 需为单值或长度为 2 的列表")
self.decay_factor = _to_two(decay_factor)
self.dominance_atr_multiplier = _to_two(dominance_atr_multiplier)
self.activity_atr_multiplier = _to_two(activity_atr_multiplier)
self.exit_atr_multiplier = _to_two(exit_atr_multiplier)
self.long_params = {
"decay_factor": _to_two(decay_factor, 0.95)[0],
"dominance_multiplier": _to_two(dominance_multiplier, 1.5)[0],
"activity_multiplier": _to_two(activity_multiplier, 2.0)[0],
"exit_multiplier": _to_two(exit_multiplier, 0.5)[0],
}
self.short_params = {
"decay_factor": _to_two(decay_factor, 0.95)[1],
"dominance_multiplier": _to_two(dominance_multiplier, 1.2)[1],
"activity_multiplier": _to_two(activity_multiplier, 1.8)[1],
"exit_multiplier": _to_two(exit_multiplier, 0.6)[1],
}
self.lookback_period = lookback_period
self.atr_period = atr_period
self.atr_stop_multiplier = atr_stop_multiplier
self.lvi_period = lvi_period
self.symbol = main_symbol
self.volume = trade_volume
self.order_direction = order_direction or ['BUY', 'SELL']
self.min_bars_required = max(self.lookback_period, self.atr_period) + 2
self.min_bars_required = max(self.lookback_period, self.atr_period, self.lvi_period) + 2
self.dp_long = 0.0
self.dp_short = 0.0
self.dominance_history = deque(maxlen=self.atr_period)
self.previous_bar: Optional[Bar] = None
self.volume_history = deque(maxlen=self.lookback_period)
self.tr_history = deque(maxlen=self.atr_period)
self.lvi_history = deque(maxlen=self.lvi_period)
self.atr = 0.0
self.trailing_stop_price: Optional[float] = None
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
# ... on_open_bar 保持不变 ...
def on_open_bar(self, open_price: float, symbol: str):
bars = self.get_bar_history()
if len(bars) < self.min_bars_required:
@@ -77,133 +90,138 @@ class AdaptiveInformedTraderDPStrategy(Strategy):
if self._entry_logic(side_idx, latest_bar):
break
# --- 【核心修改】_update_dp_state 函数的最终版本 ---
# --- 【核心】最终版的DP状态更新函数 ---
def _update_dp_state(self, latest_bar: Bar):
# 1. 更新历史数据 (不变)
current_volume = latest_bar.volume
if len(self.volume_history) == self.lookback_period:
recent_avg_volume = (sum(self.volume_history) - self.volume_history[
0] + current_volume) / self.lookback_period
else:
recent_avg_volume = np.mean(list(self.volume_history) + [current_volume])
self.volume_history.append(current_volume)
# 1. 更新历史数据
if self.previous_bar is None:
self.previous_bar = latest_bar
return
# 2. 计算宏观指标 (不变)
current_volume = latest_bar.volume
self.volume_history.append(current_volume)
recent_avg_volume = np.mean(list(self.volume_history)) if self.volume_history else 0
# 2. 计算宏观、微观和流动性指标
oi_change = getattr(latest_bar, 'close_oi', 0) - getattr(self.previous_bar, 'close_oi', 0)
epsilon = 1e-9
volume_factor = current_volume / (recent_avg_volume + epsilon)
volume_factor = np.clip(volume_factor, 0, 3)
# 3. K线形态分析 (不变)
o, h, l, c = latest_bar.open, latest_bar.high, latest_bar.low, latest_bar.close
range_size = h - l if h > l else epsilon
candlestick_factor = ((c - l) - (h - c)) / range_size # 重新定义为 (多头力量 - 空头力量) / 总波幅
candlestick_factor = ((c - l) - (h - c)) / range_size
# 4. 【核心重构】无阈值的基础分数计算
OPEN_POSITION_WEIGHT = 2.0
CLOSE_POSITION_WEIGHT = 1.0
current_lvi = range_size / (current_volume + epsilon)
self.lvi_history.append(current_lvi)
avg_lvi = np.mean(list(self.lvi_history)) if self.lvi_history else 0
# 分离出单向的K线压力 (0到1之间)
# 3. 【新增】计算流动性上下文因子 (Liquidity Context Factor)
# lvi_ratio > 1: 流动性差; lvi_ratio < 1: 流动性好
lvi_ratio = current_lvi / (avg_lvi + epsilon)
# 因子经过处理，使其效果更平滑和稳健
# 当流动性差时(lvi_ratio>1)，空头增强，多头削弱
# 当流动性好时(lvi_ratio<1)，多头增强，空头削弱
bearish_lvi_booster = np.log1p(max(0, lvi_ratio - 1)) + 1 # 对大于1的部分取对数，保证平滑增长
bullish_lvi_booster = np.log1p(max(0, 1 / lvi_ratio - 1)) + 1 if lvi_ratio > 0 else 1 # 对小于1的部分取倒数再处理
# 4. 计算无阈值的基础分数 (与v8.0相同)
OPEN_POSITION_WEIGHT, CLOSE_POSITION_WEIGHT = 2.0, 1.0
bullish_candle_pressure = max(0, candlestick_factor)
bearish_candle_pressure = max(0, -candlestick_factor)
bullish_base_score = 0.0
bearish_base_score = 0.0
if oi_change > 0: # 市场在增仓 (开仓行为)
bullish_base_score, bearish_base_score = 0.0, 0.0
if oi_change > 0:
bullish_base_score = OPEN_POSITION_WEIGHT * bullish_candle_pressure
bearish_base_score = OPEN_POSITION_WEIGHT * bearish_candle_pressure
elif oi_change < 0: # 市场在减仓 (平仓行为)
elif oi_change < 0:
bullish_base_score = CLOSE_POSITION_WEIGHT * bullish_candle_pressure
bearish_base_score = CLOSE_POSITION_WEIGHT * bearish_candle_pressure
# 5. 应用状态转移方程 (不变)
final_bullish_impulse = bullish_base_score * volume_factor
final_bearish_impulse = bearish_base_score * volume_factor
# 5. 【核心】非对称地应用LVI因子，并更新DP状态
final_bullish_impulse = (bullish_base_score * volume_factor) * bullish_lvi_booster
final_bearish_impulse = (bearish_base_score * volume_factor) * bearish_lvi_booster
self.dp_long = self.dp_long * self.decay_factor[0] + final_bullish_impulse
self.dp_short = self.dp_short * self.decay_factor[1] + final_bearish_impulse
self.dp_long = self.dp_long * self.long_params['decay_factor'] + final_bullish_impulse
self.dp_short = self.dp_short * self.short_params['decay_factor'] + final_bearish_impulse
# 6. 更新ATR和Dominance历史 (不变)
# 6. 更新其他指标和日志
true_range = max(h - l, abs(h - self.previous_bar.close), abs(l - self.previous_bar.close))
self.tr_history.append(true_range)
self.atr = np.mean(list(self.tr_history))
dominance = self.dp_long - self.dp_short
self.dominance_history.append(dominance)
self.log(
f"DP更新: Dominance={dominance:.2f}, ATR={self.atr:.2f}, "
f"dp_long={self.dp_long:.2f}, dp_short={self.dp_short:.2f}, "
f"CandleFactor={candlestick_factor:.2f}"
)
f"DP更新: Dom={dominance:.2f}, BullBoost={bullish_lvi_booster:.2f}, BearBoost={bearish_lvi_booster:.2f}")
self.previous_bar = latest_bar
# ... _entry_logic, _exit_logic, _send_market_order 函数保持不变 ...
# --- 入场逻辑现在回归统一，因为非对称性已在DP计算中体现 ---
def _entry_logic(self, side_idx: int, latest_bar: Bar) -> bool:
dominance = self.dp_long - self.dp_short
activity = self.dp_long + self.dp_short
if len(self.dominance_history) < self.atr_period: return False
dominance_std = np.std(list(self.dominance_history))
dynamic_dom_thresh = dominance_std * self.dominance_atr_multiplier[side_idx]
dynamic_act_thresh = dominance_std * self.activity_atr_multiplier[side_idx]
if activity < dynamic_act_thresh: return False
signal, side_str = False, ''
if side_idx == 0:
if dominance > dynamic_dom_thresh: signal, side_str = True, 'BUY'
else:
if dominance < -dynamic_dom_thresh: signal, side_str = True, 'SELL'
if not signal: return False
indicator_condition_met = self.indicators[side_idx].is_condition_met(*self.get_indicator_tuple())
if not indicator_condition_met: return False
self.log(
f"触发{side_str}信号: Dominance={dominance:.2f} > {dynamic_dom_thresh:.2f}, Activity={activity:.2f} > {dynamic_act_thresh:.2f}")
self._send_market_order(side_str, self.volume)
if side_str == 'BUY':
self.trailing_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
self.log(f"多头开仓，设置初始ATR止损价: {self.trailing_stop_price:.2f}")
elif side_str == 'SELL':
self.trailing_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
self.log(f"空头开仓，设置初始ATR止损价: {self.trailing_stop_price:.2f}")
return True
if len(self.dominance_history) < self.atr_period: return False
dominance_std = np.std(list(self.dominance_history))
params = self.long_params if side_idx == 0 else self.short_params
dynamic_dom_thresh = dominance_std * params['dominance_multiplier']
dynamic_act_thresh = dominance_std * params['activity_multiplier']
if activity < dynamic_act_thresh: return False
if side_idx == 0: # 多头
if dominance > dynamic_dom_thresh:
self.log(f"触发BUY信号: Dominance={dominance:.2f} > {dynamic_dom_thresh:.2f}")
self._send_market_order('BUY', self.volume, latest_bar)
return True
else: # 空头
if dominance < -dynamic_dom_thresh and self.indicators[side_idx].is_condition_met(*self.get_indicator_tuple()):
self.log(f"触发SELL信号: Dominance={dominance:.2f} < {-dynamic_dom_thresh:.2f}")
self._send_market_order('SELL', self.volume, latest_bar)
return True
return False
# ... _exit_logic 和 _send_market_order 保持不变，它们已经是独立的了 ...
def _exit_logic(self, pos: int, latest_bar: Bar):
params = self.long_params if pos > 0 else self.short_params
if pos > 0:
new_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
if self.trailing_stop_price is None: self.trailing_stop_price = new_stop_price
self.trailing_stop_price = max(self.trailing_stop_price, new_stop_price)
if latest_bar.low <= self.trailing_stop_price:
self.log(f"触发多头ATR追踪止损: 最新价 {latest_bar.low:.2f} <= 止损价 {self.trailing_stop_price:.2f}")
self._send_market_order('CLOSE_LONG', abs(pos))
self.trailing_stop_price = None
self.log(f"触发多头ATR追踪止损")
self._send_market_order('CLOSE_LONG', abs(pos), latest_bar)
return
elif pos < 0:
else: # pos < 0
new_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
if self.trailing_stop_price is None: self.trailing_stop_price = new_stop_price
self.trailing_stop_price = min(self.trailing_stop_price, new_stop_price)
if latest_bar.high >= self.trailing_stop_price:
self.log(f"触发空头ATR追踪止损: 最新价 {latest_bar.high:.2f} >= 止损价 {self.trailing_stop_price:.2f}")
self._send_market_order('CLOSE_SHORT', abs(pos))
self.trailing_stop_price = None
self.log(f"触发空头ATR追踪止损")
self._send_market_order('CLOSE_SHORT', abs(pos), latest_bar)
return
dominance = self.dp_long - self.dp_short
side_idx = 0 if pos > 0 else 1
dominance_std = np.std(list(self.dominance_history))
dynamic_exit_thresh = dominance_std * self.exit_atr_multiplier[side_idx]
exit_signal, side_str = False, ''
dynamic_exit_thresh = dominance_std * params['exit_multiplier']
if pos > 0 and dominance < dynamic_exit_thresh:
exit_signal, side_str = True, 'CLOSE_LONG'
self.log(f"多头DP优势差回落，策略性离场")
self._send_market_order('CLOSE_LONG', abs(pos), latest_bar)
elif pos < 0 and dominance > -dynamic_exit_thresh:
exit_signal, side_str = True, 'CLOSE_SHORT'
if exit_signal:
self.log(f"DP优势差回落，策略性离场: Dominance={dominance:.2f} < 阈值 {dynamic_exit_thresh:.2f}")
self._send_market_order(side_str, abs(pos))
self.trailing_stop_price = None
self.log(f"空头DP优势差回落，策略性离场")
self._send_market_order('CLOSE_SHORT', abs(pos), latest_bar)
def _send_market_order(self, direction: str, vol: int):
offset = 'OPEN' if direction in ('BUY', 'SELL') else 'CLOSE'
def _send_market_order(self, direction: str, vol: int, latest_bar: Bar):
if direction in ('BUY', 'SELL'):
offset = 'OPEN'
else:
offset = 'CLOSE'
if offset == 'OPEN':
if direction == 'BUY':
self.trailing_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
else: # SELL
self.trailing_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
else:
self.trailing_stop_price = None
oid = f"{self.symbol}_{direction}_{self.get_current_time():%Y%m%d%H%M%S}"
self.send_order(
Order(id=oid, symbol=self.symbol, direction=direction, volume=vol, price_type='MARKET', offset=offset))

576
src/strategies/InformedTraderDPStrategy/MomentumDPStrategy.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

221
src/strategies/InformedTraderDPStrategy/MomentumDPStrategy.py Normal file
View File

@@ -0,0 +1,221 @@
from collections import deque
from typing import List, Union, Optional
import numpy as np
from src.core_data import Bar, Order
from src.indicators.base_indicators import Indicator
from src.strategies.base_strategy import Strategy
from src.indicators.indicators import Empty
class AsymmetricOriginalDPStrategy(Strategy):
"""
机构级思路 · 最终版 v11.1
回归原始DP计算，引入非对称决策，并增加独立的Activity过滤器。
1. 统一感知: 采用最原始的、基于P/V/OI的DP状态计算。
2. 非对称决策: a) 牛市寻找多头优势 b) 熊市寻找多头反扑失败点 c) 独立的Activity过滤。
3. 独立参数: 多空双方拥有完全独立的参数集。
4. 风险管理: ATR动态追踪止损。
"""
def __init__(self, context, main_symbol: str,
trade_volume: int = 1,
lookback_period: int = 120,
decay_factor: Union[float, List[float]] = [0.95, 0.95],
dominance_multiplier: Union[float, List[float]] = [1.5, 1.5],
# --- 【新增】独立的Activity乘数 ---
activity_multiplier: Union[float, List[float]] = [1.0, 1.0],
long_power_falloff_pct: float = 0.3,
atr_period: int = 20,
macro_ma_period: int = 200,
atr_stop_multiplier: float = 2.0,
order_direction: Optional[list] = None,
enable_log: bool = False,
indicators: List[Union[Indicator, List[Indicator]]] = None, ):
super().__init__(context, main_symbol, enable_log=enable_log)
# --- 非对称参数的初始化 ---
self.long_params = {
"decay_factor": decay_factor[0] if isinstance(decay_factor, list) else decay_factor,
"dominance_multiplier": dominance_multiplier[0] if isinstance(dominance_multiplier,
list) else dominance_multiplier,
"activity_multiplier": activity_multiplier[0] if isinstance(activity_multiplier,
list) else activity_multiplier,
}
self.short_params = {
"decay_factor": decay_factor[1] if isinstance(decay_factor, list) else decay_factor,
"long_power_falloff_pct": long_power_falloff_pct,
"activity_multiplier": activity_multiplier[1] if isinstance(activity_multiplier,
list) else activity_multiplier,
}
# ... 其他参数 ...
self.lookback_period = lookback_period
self.atr_period = atr_period
self.macro_ma_period = macro_ma_period
self.atr_stop_multiplier = atr_stop_multiplier
self.symbol = main_symbol
self.volume = trade_volume
self.order_direction = order_direction or ['BUY', 'SELL']
self.min_bars_required = max(self.lookback_period, self.atr_period, self.macro_ma_period) + 2
# --- DP状态变量 ---
self.dp_long = 0.0
self.dp_short = 0.0
self.dominance_history = deque(maxlen=self.atr_period)
self.activity_history = deque(maxlen=self.atr_period) # <<< 新增
self.dp_long_history = deque(maxlen=self.atr_period)
self.previous_bar: Optional[Bar] = None
self.volume_history = deque(maxlen=self.lookback_period)
self.price_history = deque(maxlen=self.macro_ma_period)
self.tr_history = deque(maxlen=self.atr_period)
self.atr = 0.0
self.macro_ma: Optional[float] = None
self.trailing_stop_price: Optional[float] = None
if indicators is None: indicators = [Empty(), Empty()]
self.indicators = indicators
def on_open_bar(self, open_price: float, symbol: str):
# ... on_open_bar 逻辑不变 ...
bars = self.get_bar_history()
if len(bars) < self.min_bars_required: return
latest_bar = bars[-1]
self.cancel_all_pending_orders(symbol)
self._update_dp_state(latest_bar)
pos = self.get_current_positions().get(symbol, 0)
if pos != 0:
self._exit_logic(pos, latest_bar)
return
self._entry_logic(latest_bar)
def _update_dp_state(self, latest_bar: Bar):
self.price_history.append(latest_bar.close)
self.volume_history.append(latest_bar.volume)
if len(self.price_history) < self.macro_ma_period:
self.previous_bar = latest_bar
return
self.macro_ma = np.mean(list(self.price_history))
recent_avg_volume = np.mean(list(self.volume_history))
# --- 【核心修正】修复 AttributeError ---
oi_change = getattr(latest_bar, 'close_oi', 0) - getattr(self.previous_bar, 'close_oi', 0)
# ... 后续计算不变 ...
price_change = latest_bar.close - self.previous_bar.close
epsilon = 1e-9
volume_factor = latest_bar.volume / (recent_avg_volume + epsilon)
volume_factor = np.clip(volume_factor, 0, 3)
bullish_base_score, bearish_base_score = 0, 0
if price_change > 0:
if oi_change > 0:
bullish_base_score = 2
elif oi_change < 0:
bullish_base_score = 1
elif price_change < 0:
if oi_change > 0:
bearish_base_score = 2
elif oi_change < 0:
bearish_base_score = 1
final_bullish_impulse = bullish_base_score * volume_factor
final_bearish_impulse = bearish_base_score * volume_factor
self.dp_long = self.dp_long * self.long_params['decay_factor'] + final_bullish_impulse
self.dp_short = self.dp_short * self.short_params['decay_factor'] + final_bearish_impulse
# 更新所有历史队列
dominance = self.dp_long - self.dp_short
activity = self.dp_long + self.dp_short # <<< 计算Activity
self.dominance_history.append(dominance)
self.activity_history.append(activity) # <<< 更新Activity历史
self.dp_long_history.append(self.dp_long)
true_range = max(latest_bar.high - latest_bar.low, abs(latest_bar.high - self.previous_bar.close),
abs(latest_bar.low - self.previous_bar.close))
self.tr_history.append(true_range)
self.atr = np.mean(list(self.tr_history))
self.previous_bar = latest_bar
self.log(f"DP更新: Dominance={dominance:.2f}, Activity={activity:.2f}")
def _entry_logic(self, latest_bar: Bar):
if self.macro_ma is None: return
is_bullish_regime = latest_bar.close > self.macro_ma
# 确保历史数据足够
if len(self.dominance_history) < self.atr_period: return
if is_bullish_regime:
if 'BUY' not in self.order_direction: return
params = self.long_params
# --- 【新增】牛市的Activity过滤器 ---
activity = self.dp_long + self.dp_short
activity_std = np.std(list(self.activity_history))
if activity_std == 0: return
dynamic_act_thresh = activity_std * params['activity_multiplier']
if activity < dynamic_act_thresh: return
# 牛市的多头优势逻辑
dominance = self.dp_long - self.dp_short
dominance_std = np.std(list(self.dominance_history))
if dominance_std == 0: return
dynamic_threshold = dominance_std * params['dominance_multiplier']
if dominance > dynamic_threshold:
if not self.indicators[0].is_condition_met(*self.get_indicator_tuple()): return
self.log(f"【牛市】触发多头信号: Dominance {dominance:.2f} > 阈值 {dynamic_threshold:.2f}")
self._send_market_order('BUY', self.volume, latest_bar)
else: # is_bearish_regime
if 'SELL' not in self.order_direction: return
params = self.short_params
# --- 【新增】熊市的Activity过滤器 ---
activity = self.dp_long + self.dp_short
activity_std = np.std(list(self.activity_history))
if activity_std == 0: return
# dynamic_act_thresh = activity_std * params['activity_multiplier']
# if activity < dynamic_act_thresh: return
# 熊市的多头反扑失败逻辑
recent_max_dp_long = max(self.dp_long_history)
if recent_max_dp_long < np.std(list(self.dp_long_history)) * params['activity_multiplier']: return
falloff_threshold = recent_max_dp_long * (1 - params['long_power_falloff_pct'])
if self.dp_long < falloff_threshold:
if not self.indicators[1].is_condition_met(*self.get_indicator_tuple()): return
self.log(f"【熊市】触发空头信号: 多头力量 {self.dp_long:.2f} 从高点 {recent_max_dp_long:.2f} 回落")
self._send_market_order('SELL', self.volume, latest_bar)
def _exit_logic(self, pos: int, latest_bar: Bar):
if pos > 0:
new_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
if self.trailing_stop_price is None: self.trailing_stop_price = new_stop_price
self.trailing_stop_price = max(self.trailing_stop_price, new_stop_price)
if latest_bar.low <= self.trailing_stop_price:
self._send_market_order('CLOSE_LONG', abs(pos), latest_bar)
return
elif pos < 0:
new_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
if self.trailing_stop_price is None: self.trailing_stop_price = new_stop_price
self.trailing_stop_price = min(self.trailing_stop_price, new_stop_price)
if latest_bar.high >= self.trailing_stop_price:
self._send_market_order('CLOSE_SHORT', abs(pos), latest_bar)
return
dominance = self.dp_long - self.dp_short
if pos > 0 and dominance < 0:
self._send_market_order('CLOSE_LONG', abs(pos), latest_bar)
elif pos < 0 and dominance > 0:
self._send_market_order('CLOSE_SHORT', abs(pos), latest_bar)
def _send_market_order(self, direction: str, vol: int, latest_bar: Bar):
offset = 'OPEN' if direction in ('BUY', 'SELL') else 'CLOSE'
if offset == 'OPEN':
if direction == 'BUY':
self.trailing_stop_price = latest_bar.close - self.atr * self.atr_stop_multiplier
else: # SELL
self.trailing_stop_price = latest_bar.close + self.atr * self.atr_stop_multiplier
else: # CLOSE
self.trailing_stop_price = None
oid = f"{self.symbol}_{direction}_{self.get_current_time():%Y%m%d%H%M%S}"
self.send_order(
Order(id=oid, symbol=self.symbol, direction=direction, volume=vol, price_type='MARKET', offset=offset))

122
src/strategies/InformedTraderDPStrategy/grid_search_multi_process2.ipynb
View File

File diff suppressed because one or more lines are too long

542
src/strategies/InformedTraderDPStrategy/grid_search_multi_process3.ipynb Normal file
View File

File diff suppressed because one or more lines are too long

544
src/strategies/InformedTraderDPStrategy/grid_search_multi_process3_MomentumDPS.ipynb Normal file
View File

File diff suppressed because one or more lines are too long