NewQuant/futures_trading_strategies/FG/FisherTrendStrategy/Strategy3.py

import numpy as np
import math
import talib
from collections import deque
from typing import Optional, Any, List

from src.core_data import Bar, Order
from src.indicators.base_indicators import Indicator
from src.strategies.base_strategy import Strategy


class PragmaticCyberneticStrategy(Strategy):
    """
    【务实回归版·控制论策略】(TA-Lib 提速 + 换月支持)

    哲学坚守：
    1. 捍卫原版出场逻辑：坚决使用基于 entry_price 的固定硬止损。脱离成本区后，给予价格无限宽容度，实现长线死拿。
    2. 动能极值止盈：恢复 FB > 1.6 且拐头时的精准落袋为安。
    3. 拒绝画蛇添足：没有任何 _target_state，不堆砌 Trick，底层数据即真理。
    """

    def __init__(
            self,
            context: Any,
            main_symbol: str,
            enable_log: bool,
            trade_volume: int,
            min_tick: float = 1.0,

            # --- 核心周期 ---
            trend_period: int = 26,
            fisher_period: int = 20,
            atr_period: int = 23,

            # --- 经过验证的优秀参数 ---
            fisher_exit_level: float = 1.6,  # 配合 1.98 修复后的 FB，1.6 是一个非常好的极值止盈点
            fb_entry_threshold: float = 0.1,  # 0.1 确保在顺大势的前提下，动能稍有配合即可入场

            # --- 风险控制 ---
            stop_mult: float = 2.0,  # 原始的硬止损乘数
            limit_offset_mult: float = 0.,

            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

        # 为 talib.ATR 预留充足的数据预热空间
        self._buf_len = max(self.t_len, self.f_len, self.atr_len * 2) + 100
        self._highs = deque(maxlen=self._buf_len)
        self._lows = deque(maxlen=self._buf_len)
        self._closes = deque(maxlen=self._buf_len)

        # 使用 deque 记录 FB，相比单独的 _prev_fb 变量更安全，不会因为初始 0 导致误判
        self._fbs = deque(maxlen=2)
        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):
        """完全无状态的纯数学指标计算 (基于 TA-Lib)"""
        if len(self._closes) < self._buf_len:
            return None, None, None, None

        np_highs = np.array(self._highs, dtype=np.float64)
        np_lows = np.array(self._lows, dtype=np.float64)
        np_closes = np.array(self._closes, dtype=np.float64)

        # 1. T轴 (大势过滤)
        t_max_h = talib.MAX(np_highs, timeperiod=self.t_len)[-1]
        t_min_l = talib.MIN(np_lows, timeperiod=self.t_len)[-1]
        T = (t_max_h + t_min_l) / 2.0

        # 2. FV (公允价值挂单基准)
        FV = (np_closes[-1] + 2 * np_closes[-2] + 2 * np_closes[-3] + np_closes[-4]) / 6.0

        # 3. Fisher Transform (1.98 数学修复版)
        f_max_h = talib.MAX(np_highs, timeperiod=self.f_len)[-1]
        f_min_l = talib.MIN(np_lows, timeperiod=self.f_len)[-1]
        denom = f_max_h - f_min_l if f_max_h != f_min_l else self.min_tick

        stoc = (np_closes[-1] - f_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
        atr_array = talib.ATR(np_highs, np_lows, np_closes, timeperiod=self.atr_len)
        ATR = atr_array[-1]

        return T, FV, FB, ATR

    def on_open_bar(self, open_price: float, symbol: str):
        self.symbol = symbol
        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()
        self.log(f'T: {T} FV: {FV} FB: {FB} ATR: {ATR}')
        if T is None or math.isnan(ATR): return

        self._fbs.append(FB)
        # 必须凑齐两根 FB 才能判断拐点，否则跳过
        if len(self._fbs) < 2: return

        current_fb = self._fbs[-1]
        prev_fb = self._fbs[-2]

        # 绝对无状态：从账户物理数据获取真实持仓和开仓均价
        pos = self.get_current_positions().get(self.symbol, 0)
        entry_price = self.get_average_position_price(self.symbol)

        is_met = self.indicator is None or self.indicator.is_condition_met(*self.get_indicator_tuple())

        self.log(f'is_met: {is_met}, pos: {pos}, current_fb: {current_fb}, prev_fb: {prev_fb}')

        # ==========================================
        # 核心一：经典出场逻辑 (你的原版设计)
        # ==========================================
        if pos != 0:
            stop_dist = max(self.stop_mult * ATR, self.min_tick * 20)

            # --- 预判：是否存在反向入场信号？ ---
            # 反向做多信号：大势已多 (C > T)，且动能回调 (FB < -0.1)
            is_long_signal = (prev_bar.close > T) and (current_fb < -self.fb_entry_threshold)
            # 反向做空信号：大势已空 (C < T)，且动能反弹 (FB > 0.1)
            is_short_signal = (prev_bar.close < T) and (current_fb > self.fb_entry_threshold)

            if pos > 0:
                # A. 动能极值止盈
                if current_fb > self.fisher_exit_level and current_fb < prev_fb:
                    self.log(f"TAKE PROFIT (Long): FB {current_fb} Peak Reached")
                    self.send_market_order("CLOSE_LONG", abs(pos), "CLOSE")
                    return

                # B. 原版硬止损 (最后的底线)
                if prev_bar.close < entry_price - stop_dist:
                    self.log(f"HARD STOP (Long): Close {prev_bar.close} < Limit {entry_price - stop_dist}")
                    self.send_market_order("CLOSE_LONG", abs(pos), "CLOSE")
                    return

                # C. 解决冲突：反向信号强制平仓
                # 如果我们持有多单，但盘面竟然发出了“做空”信号，说明大势已去，逻辑矛盾，必须平仓！
                if is_short_signal and is_met:
                    self.log(f"REVERSAL SIGNAL EXIT (Long): Trend Down & Short Signal Triggered.")
                    self.send_market_order("CLOSE_LONG", abs(pos), "CLOSE")
                    return

            elif pos < 0:
                # A. 动能极值止盈
                if current_fb < -self.fisher_exit_level and current_fb > prev_fb:
                    self.log(f"TAKE PROFIT (Short): FB {current_fb} Bottom Reached")
                    self.send_market_order("CLOSE_SHORT", abs(pos), "CLOSE")
                    return

                # B. 原版硬止损
                if prev_bar.close > entry_price + stop_dist:
                    self.log(f"HARD STOP (Short): Close {prev_bar.close} > Limit {entry_price + stop_dist}")
                    self.send_market_order("CLOSE_SHORT", abs(pos), "CLOSE")
                    return

                # C. 解决冲突：反向信号强制平仓
                # 如果我们持有空单，但盘面发出了“做多”信号，平空！
                if is_long_signal and is_met:
                    self.log(f"REVERSAL SIGNAL EXIT (Short): Trend Up & Long Signal Triggered.")
                    self.send_market_order("CLOSE_SHORT", abs(pos), "CLOSE")
                    return

            # ==========================================
            # 核心二：经典入场逻辑 (完全不变)
            # ==========================================
        elif pos == 0 and is_met:
            if prev_bar.close > T and current_fb < -self.fb_entry_threshold:
                if "BUY" in self.order_direction:
                    long_limit = self.round_to_tick(FV - (self.limit_offset_mult * ATR))
                    self.send_limit_order(long_limit, "BUY", self.trade_volume, "OPEN")

            elif prev_bar.close < T and current_fb > self.fb_entry_threshold:
                if "SELL" in self.order_direction:
                    short_limit = self.round_to_tick(FV + (self.limit_offset_mult * ATR))
                    self.send_limit_order(short_limit, "SELL", self.trade_volume, "OPEN")

    # ==========================================
    # 彻底隔离历史噪音：原生换月机制
    # ==========================================
    def on_contract_rollover(self, old_symbol: str, new_symbol: str):
        self.log(f"ROLLOVER TRIGGERED: {old_symbol} -> {new_symbol}")

        # 撤销老合约全部残留动作
        self.cancel_all_pending_orders(old_symbol)

        # 市价清空老合约，因为换月会导致 entry_price 和盘面断层，强制物理归零最安全
        pos = self.get_current_positions().get(old_symbol, 0)
        if pos > 0:
            self.send_order(Order(id=f"ROLLOVER_CLOSE_L_{old_symbol}", symbol=old_symbol, direction="CLOSE_LONG",
                                  volume=abs(pos), price_type="MARKET", submitted_time=self.get_current_time(),
                                  offset="CLOSE"))
        elif pos < 0:
            self.send_order(Order(id=f"ROLLOVER_CLOSE_S_{old_symbol}", symbol=old_symbol, direction="CLOSE_SHORT",
                                  volume=abs(pos), price_type="MARKET", submitted_time=self.get_current_time(),
                                  offset="CLOSE"))

        # 切换主交易标的
        self.main_symbol = new_symbol
        self.symbol = new_symbol

    # --- 下单系统 ---
    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))