216 lines
9.1 KiB
Python
216 lines
9.1 KiB
Python
import numpy as np
|
||
import math
|
||
from collections import deque
|
||
from typing import Optional, Any, List, Dict
|
||
|
||
from src.core_data import Bar, Order
|
||
from src.indicators.base_indicators import Indicator
|
||
from src.strategies.base_strategy import Strategy
|
||
|
||
|
||
class PragmaticCyberneticStrategy(Strategy):
|
||
"""
|
||
【务实优化版·控制论策略】
|
||
|
||
优化核心：
|
||
1. 保持“钝感”：拒绝灵敏的趋势反转平仓，保留“死拿趋势”的盈利特性。
|
||
2. 修复系数：将 Stoch 映射系数从 0.66 提升至 1.98，使 Fisher 值域恢复到 [-3, 3] 可调范围。
|
||
- 现在你可以设置 exit_level = 2.0 来捕捉极端利润，而不是被迫等到换月。
|
||
3. 纯粹逻辑：开仓是开仓，平仓是平仓。互不干扰，消除中间地带的内耗。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
context: Any,
|
||
main_symbol: str,
|
||
enable_log: bool,
|
||
trade_volume: int,
|
||
min_tick: float = 1.0,
|
||
|
||
# --- 核心参数 (保持原策略风格) ---
|
||
trend_period: int = 26, # T: 趋势中轴
|
||
fisher_period: int = 20, # FB: 动能周期 (原策略 46 可能太慢，建议 20-30)
|
||
atr_period: int = 23,
|
||
|
||
# --- 阈值参数 (关键调整) ---
|
||
# 1. 止盈阈值：因为系数修复了，现在 FB 能跑到 2.5 以上。
|
||
# 建议设为 2.0 - 2.5。如果设得很高(如 5.0)，效果就等于你原来的“死拿”。
|
||
fisher_exit_level: float = 2.2,
|
||
|
||
# 2. 入场阈值：保持在 0.5 左右，只接深回调
|
||
fb_entry_threshold: float = 0.,
|
||
|
||
stop_mult: float = 2, # 稍微放宽止损，适应趋势震荡
|
||
limit_offset_mult: float = 0.2, # FV 挂单偏移
|
||
|
||
order_direction: Optional[List[str]] = None,
|
||
indicator: Indicator = None,
|
||
):
|
||
super().__init__(context, main_symbol, enable_log)
|
||
self.trade_volume = trade_volume
|
||
self.min_tick = min_tick
|
||
|
||
self.t_len = trend_period
|
||
self.f_len = fisher_period
|
||
self.atr_len = atr_period
|
||
|
||
self.fisher_exit_level = fisher_exit_level
|
||
self.fb_entry_threshold = fb_entry_threshold
|
||
self.stop_mult = stop_mult
|
||
self.limit_offset_mult = limit_offset_mult
|
||
|
||
self.order_direction = order_direction or ['BUY', 'SELL']
|
||
self.indicator = indicator
|
||
|
||
# 缓存
|
||
self._buf_len = max(self.t_len, self.f_len, self.atr_len) + 5
|
||
self._highs = deque(maxlen=self._buf_len)
|
||
self._lows = deque(maxlen=self._buf_len)
|
||
self._closes = deque(maxlen=self._buf_len)
|
||
|
||
# 记录上一根 Bar 的 FB，仅用于判断拐点
|
||
self._prev_fb = 0.0
|
||
self.order_id_counter = 0
|
||
|
||
def round_to_tick(self, price: float) -> float:
|
||
if self.min_tick <= 0: return price
|
||
return round(price / self.min_tick) * self.min_tick
|
||
|
||
def _calculate_indicators(self):
|
||
"""
|
||
计算逻辑：保留原策略的简洁性，仅修复 Scaling 系数
|
||
"""
|
||
if len(self._closes) < self._buf_len:
|
||
return None, None, None, None
|
||
|
||
# 1. 趋势中轴 T (物理中轴)
|
||
h_trend = list(self._highs)[-self.t_len:]
|
||
l_trend = list(self._lows)[-self.t_len:]
|
||
T = (max(h_trend) + min(l_trend)) / 2.0
|
||
|
||
# 2. 公平价 FV (FIR 滤波)
|
||
FV = (self._closes[-1] + 2 * self._closes[-2] + 2 * self._closes[-3] + self._closes[-4]) / 6.0
|
||
|
||
# 3. Fisher Transform (非递归版，响应更快)
|
||
h_fisher = list(self._highs)[-self.f_len:]
|
||
l_fisher = list(self._lows)[-self.f_len:]
|
||
max_h, min_l = max(h_fisher), min(l_fisher)
|
||
denom = max_h - min_l if max_h != min_l else self.min_tick
|
||
|
||
# --- 关键修正点 ---
|
||
# 你的原代码是 0.66 * (stoc - 0.5)，这导致最大值被锁死。
|
||
# 改为 1.98，使得输入范围从 [-0.33, 0.33] 扩大到 [-0.99, 0.99]。
|
||
# 这样 math.log 就能计算出 -3 到 +3 的值，让止盈逻辑“复活”且可控。
|
||
stoc = (self._closes[-1] - min_l) / denom
|
||
value = max(-0.999, min(0.999, 1.98 * (stoc - 0.5)))
|
||
FB = 0.5 * math.log((1.0 + value) / (1.0 - value))
|
||
|
||
# 4. ATR
|
||
tr_list = []
|
||
for i in range(1, self.atr_len + 1):
|
||
h, l, pc = self._highs[-i], self._lows[-i], self._closes[-i - 1]
|
||
tr = max(h - l, abs(h - pc), abs(l - pc))
|
||
tr_list.append(tr)
|
||
ATR = sum(tr_list) / self.atr_len
|
||
|
||
return T, FV, FB, ATR
|
||
|
||
def on_open_bar(self, open_price: float, symbol: str):
|
||
self.symbol = symbol
|
||
# 每次 Bar 重置挂单，防止挂单“长在”图表上
|
||
self.cancel_all_pending_orders(self.symbol)
|
||
|
||
bars = self.get_bar_history()
|
||
if len(bars) < 1: return
|
||
prev_bar = bars[-1]
|
||
|
||
self._highs.append(prev_bar.high)
|
||
self._lows.append(prev_bar.low)
|
||
self._closes.append(prev_bar.close)
|
||
|
||
T, FV, FB, ATR = self._calculate_indicators()
|
||
if T is None: return
|
||
|
||
pos = self.get_current_positions().get(self.symbol, 0)
|
||
entry_price = self.get_average_position_price(self.symbol)
|
||
|
||
# 状态定义
|
||
# 这里我们不去定义 "short_signal" 这种会诱发反向平仓的变量
|
||
# 而是只关注眼下的：大势(T) 和 动能(FB)
|
||
trend_up = prev_bar.close > T
|
||
trend_down = prev_bar.close < T
|
||
|
||
# ==========================================
|
||
# 1. 持仓逻辑：简单、迟钝、粘性强
|
||
# ==========================================
|
||
if pos != 0:
|
||
stop_dist = max(self.stop_mult * ATR, self.min_tick * 20)
|
||
|
||
if pos > 0:
|
||
# A. 动能止盈 (Mean Reversion Exit)
|
||
# 只有当行情极其疯狂(FB > 2.2) 且开始回头时才止盈。
|
||
# 正常波动绝不下车。
|
||
if FB > self.fisher_exit_level and FB < self._prev_fb:
|
||
self.log(f"TAKE PROFIT (Long): FB {FB:.2f} Peak Reached")
|
||
self.send_market_order("CLOSE_LONG", abs(pos), "CLOSE")
|
||
return
|
||
|
||
# B. 硬止损 (ATR Stop) - 最后的防线
|
||
if prev_bar.close < entry_price - stop_dist:
|
||
self.send_market_order("CLOSE_LONG", abs(pos), "CLOSE")
|
||
return
|
||
|
||
# C. (可选) 极端趋势反转保护
|
||
# 如果你希望策略更像原来的“死拿”，这部分可以注释掉，或者把判定条件设严
|
||
# if prev_bar.close < T - ATR: ...
|
||
|
||
elif pos < 0:
|
||
# A. 动能止盈
|
||
if FB < -self.fisher_exit_level and FB > self._prev_fb:
|
||
self.log(f"TAKE PROFIT (Short): FB {FB:.2f} Bottom Reached")
|
||
self.send_market_order("CLOSE_SHORT", abs(pos), "CLOSE")
|
||
return
|
||
|
||
# B. 硬止损
|
||
if prev_bar.close > entry_price + stop_dist:
|
||
self.send_market_order("CLOSE_SHORT", abs(pos), "CLOSE")
|
||
return
|
||
|
||
# ==========================================
|
||
# 2. 开仓逻辑：严苛、左侧、顺大势
|
||
# ==========================================
|
||
if pos == 0:
|
||
is_met = self.indicator is None or self.indicator.is_condition_met(*self.get_indicator_tuple())
|
||
|
||
# 挂单价格优化：
|
||
# 在 FV 基础上再便宜一点点，增加胜率
|
||
long_limit = self.round_to_tick(FV - (self.limit_offset_mult * ATR))
|
||
short_limit = self.round_to_tick(FV + (self.limit_offset_mult * ATR))
|
||
|
||
# 开多：趋势向上 + 动能深跌 (FB < -0.5)
|
||
if trend_up and FB < -self.fb_entry_threshold and is_met:
|
||
if "BUY" in self.order_direction:
|
||
self.send_limit_order(long_limit, "BUY", self.trade_volume, "OPEN")
|
||
|
||
# 开空：趋势向下 + 动能冲高 (FB > 0.5)
|
||
elif trend_down and FB > self.fb_entry_threshold and is_met:
|
||
if "SELL" in self.order_direction:
|
||
self.send_limit_order(short_limit, "SELL", self.trade_volume, "OPEN")
|
||
|
||
# 记录上一根FB，仅用于止盈时的拐点比较
|
||
self._prev_fb = FB
|
||
|
||
# --- 辅助 ---
|
||
def send_market_order(self, direction, volume, offset):
|
||
ts = self.get_current_time().strftime('%H%M%S')
|
||
oid = f"{self.symbol}_{direction}_MKT_{ts}_{self.order_id_counter}"
|
||
self.order_id_counter += 1
|
||
self.send_order(Order(id=oid, symbol=self.symbol, direction=direction, volume=volume, price_type="MARKET",
|
||
submitted_time=self.get_current_time(), offset=offset))
|
||
|
||
def send_limit_order(self, price, direction, volume, offset):
|
||
ts = self.get_current_time().strftime('%H%M%S')
|
||
oid = f"{self.symbol}_{direction}_LMT_{ts}_{self.order_id_counter}"
|
||
self.order_id_counter += 1
|
||
self.send_order(Order(id=oid, symbol=self.symbol, direction=direction, volume=volume, price_type="LIMIT",
|
||
submitted_time=self.get_current_time(), offset=offset, limit_price=price)) |