MyXGB_gpt_multi

import logging
from functools import reduce

import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib
import numpy as np
from freqtrade.strategy import IStrategy
import pandas as pd

logger = logging.getLogger(__name__)


from numpy.lib.stride_tricks import sliding_window_view


def triple_barrier_labeling(
    close: np.ndarray,
    high: np.ndarray,
    low: np.ndarray,
    up_pct: float,
    down_pct: float,
    horizon: int,
    vol: np.ndarray = None,
    vol_mult: float = 1.0,
):
    """
    完整版 Triple-Barrier（机构级写法）

    参数：
        - close/high/low: ndarray
        - up_pct / down_pct: 固定障碍百分比
        - horizon: 时间障碍（未来 N 根 K）
        - vol: 波动率 (如1h滚动 std)，如果提供，则 up/down 障碍将基于 volatility
        - vol_mult: 波动率倍数（默认 1）

    返回：
        - labels: up / down / NaN
        - t_events: 第几根触发障碍（从 1 开始）
        - hit_prices: 触发时的价格
    """

    n = len(close)
    valid_len = n - horizon

    # ================================
    # 0) future window vectorization
    # ================================
    future_high = sliding_window_view(high, window_shape=horizon)[1:]
    future_low = sliding_window_view(low, window_shape=horizon)[1:]

    c = close[:valid_len]

    # ================================
    # 1) 构建上下障碍（可选波动率）
    # ================================
    if vol is not None:
        up_barrier = c * (1 + vol_mult * vol[:valid_len])
        down_barrier = c * (1 - vol_mult * vol[:valid_len])
    else:
        up_barrier = c * (1 + up_pct)
        down_barrier = c * (1 - down_pct)

    # ================================
    # 2) 判断未来在哪一根触发障碍
    # ================================
    hit_up = future_high >= up_barrier[:, None]
    hit_down = future_low <= down_barrier[:, None]

    # first occurrence index
    up_first_idx = np.argmax(hit_up, axis=1)
    down_first_idx = np.argmax(hit_down, axis=1)

    # 若未触发，argmax 返回0，但 hit_up 全 False → 需要修正
    up_first_idx[~hit_up.any(axis=1)] = -1
    down_first_idx[~hit_down.any(axis=1)] = -1

    # ================================
    # 3) 确定先触发者
    # ================================
    labels = np.full(n, np.nan, dtype=object)
    t_events = np.full(n, np.nan)
    hit_prices = np.full(n, np.nan)

    for i in range(valid_len):
        up_i = up_first_idx[i]
        dn_i = down_first_idx[i]

        if up_i == -1 and dn_i == -1:
            continue  # NaN

        # 谁先触发？
        if up_i != -1 and (dn_i == -1 or up_i < dn_i):
            labels[i] = "up"
            t_events[i] = up_i + 1
            hit_prices[i] = future_high[i, up_i]
        elif dn_i != -1:
            labels[i] = "down"
            t_events[i] = dn_i + 1
            hit_prices[i] = future_low[i, dn_i]
    return labels, t_events, hit_prices

def evaluate_labels(labels):
    """评分指标：越大越好"""
    counts = pd.Series(labels).value_counts()

    up = counts.get('up', 0)
    down = counts.get('down', 0)
    nan = counts.get(np.nan, 0)

    if up + down == 0:
        return -1  # worthless

    # 平衡度
    balance = min(up, down) / max(up, down)

    # 有效标签比例
    valid_ratio = (up + down) / (up + down + nan)

    score = balance * 0.7 + valid_ratio * 0.3
    return score

def grid_search_tb(close, high, low):

    best_score = -1
    best_params = None
    best_labels = None

    up_pcts = [0.003, 0.005, 0.01, 0.015, 0.02]
    horizons = [3, 5, 8, 10]

    for up in up_pcts:
        for h in horizons:

            labels, _, _ = triple_barrier_labeling(
                close, high, low,
                up_pct=up, down_pct=up,  # 对称
                horizon=h
            )

            score = evaluate_labels(labels)

            print(f"up={up:.3%}, horizon={h}, score={score:.4f}")

            if score > best_score:
                best_score = score
                best_params = (up, h)
                best_labels = labels

    return best_params, best_labels


class MyXGB_gpt_multi(IStrategy):
    """
    Example strategy showing how the user connects their own
    IFreqaiModel to the strategy.

    Warning! This is a showcase of functionality,
    which means that it is designed to show various functions of FreqAI
    and it runs on all computers. We use this showcase to help users
    understand how to build a strategy, and we use it as a benchmark
    to help debug possible problems.

    This means this is *not* meant to be run live in production.
    """

    minimal_roi = {"0": 0.01, "20":0.008, "40":0.004, "60":0.001, "240": -1}

    plot_config = {
        "main_plot": {},
        "subplots": {
            "&-s_close": {"&-s_close": {"color": "blue"}},
            "do_predict": {
                "do_predict": {"color": "brown"},
            },
        },
    }

    process_only_new_candles = True
    stoploss = -0.01
    use_exit_signal = False
    # this is the maximum period fed to talib (timeframe independent)
    startup_candle_count: int = 40
    can_short = True

    def feature_engineering_expand_all(
        self, dataframe: DataFrame, period: int, metadata: dict, **kwargs
    ) -> DataFrame:
        # === 动量 & 超买超卖 ===
        rsi = ta.RSI(dataframe, timeperiod=period)
        dataframe[f"%-rsi-{period}"] = rsi
        dataframe[f"%-rsi_gt_70-{period}"] = (rsi > 70).astype(int)
        dataframe[f"%-rsi_lt_30-{period}"] = (rsi < 30).astype(int)
        dataframe[f"%-rsi_middle-{period}"] = ((rsi >= 30) & (rsi <= 70)).astype(int)

        mfi = ta.MFI(dataframe, timeperiod=period)
        dataframe[f"%-mfi-{period}"] = mfi
        dataframe[f"%-mfi_gt_80-{period}"] = (mfi > 80).astype(int)
        dataframe[f"%-mfi_lt_20-{period}"] = (mfi < 20).astype(int)

        # === 趋势强度 ===
        adx = ta.ADX(dataframe, timeperiod=period)
        dataframe[f"%-adx-{period}"] = adx
        dataframe[f"%-adx_gt_25-{period}"] = (adx > 25).astype(int)  # 趋势强弱分界

        # === 均线系统 ===
        sma = ta.SMA(dataframe, timeperiod=period)
        ema = ta.EMA(dataframe, timeperiod=period)
        dataframe[f"%-sma-{period}"] = sma
        dataframe[f"%-ema-{period}"] = ema

        # 价格相对于均线的位置（归一化）
        atr = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=period)
        dataframe[f"%-price_above_sma_norm-{period}"] = (dataframe['close'] - sma) / (atr + 1e-8)
        dataframe[f"%-price_above_ema_norm-{period}"] = (dataframe['close'] - ema) / (atr + 1e-8)

        # 布尔状态：是否在均线上方
        dataframe[f"%-close_gt_sma-{period}"] = (dataframe['close'] > sma).astype(int)
        dataframe[f"%-close_gt_ema-{period}"] = (dataframe['close'] > ema).astype(int)

        # === 布林带 ===
        bb = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=period, stds=2.2)
        bb_width = (bb["upper"] - bb["lower"]) / (bb["mid"] + 1e-8)
        dataframe[f"%-bb_width-{period}"] = bb_width
        dataframe[f"%-bb_width_high-{period}"] = (bb_width > bb_width.rolling(50).quantile(0.8)).astype(int)
        dataframe[f"%-bb_width_low-{period}"] = (bb_width < bb_width.rolling(50).quantile(0.2)).astype(int)

        # 价格在布林带中的相对位置（0=下轨, 1=上轨）
        bb_position = (dataframe['close'] - bb["lower"]) / (bb["upper"] - bb["lower"] + 1e-8)
        dataframe[f"%-bb_position-{period}"] = bb_position.clip(0, 1)

        # === 动量变化 ===
        macd = ta.MACD(dataframe, fastperiod=12, slowperiod=26, signalperiod=9)
        dataframe[f"%-macd-{period}"] = macd["macd"]
        dataframe[f"%-macd_signal-{period}"] = macd["macdsignal"]
        dataframe[f"%-macd_diff-{period}"] = macd["macdhist"]

        roc = ta.ROC(dataframe, timeperiod=period)
        dataframe[f"%-roc-{period}"] = roc
        dataframe[f"%-roc_positive-{period}"] = (roc > 0).astype(int)

        # === 成交量异常 ===
        vol_mean = dataframe["volume"].rolling(period).mean()
        vol_std = dataframe["volume"].rolling(period).std()
        dataframe[f"%-relative_volume-{period}"] = dataframe["volume"] / (vol_mean + 1e-8)
        dataframe[f"%-volume_spike-{period}"] = (
            dataframe["volume"] > (vol_mean + 2 * vol_std)
        ).astype(int)

        return dataframe


    def feature_engineering_expand_basic(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        # === 价格变化率（多尺度）===
        dataframe["%-pct-change-1"] = dataframe["close"].pct_change(1)
        dataframe["%-pct-change-3"] = dataframe["close"].pct_change(3)
        dataframe["%-pct-change-6"] = dataframe["close"].pct_change(6)

        # === 长期趋势 ===
        ema_200 = ta.EMA(dataframe, timeperiod=200)
        dataframe["%-ema-200"] = ema_200
        dataframe["%-close_gt_ema_200"] = (dataframe["close"] > ema_200).astype(int)

        # === 波动率（ATR 归一化）===
        atr_14 = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=14)
        dataframe["%-atr_norm"] = atr_14 / dataframe["close"]

        # ⚠️ 不再传递 raw_price / raw_volume（XGBoost 不需要原始尺度）
        # 如果确实需要，建议用 rank 或 z-score

        return dataframe


    def feature_engineering_standard(
        self, dataframe: DataFrame, metadata: dict, **kwargs
    ) -> DataFrame:
        # === 时间特征（周期性编码）===
        # 使用 sin/cos 编码避免 0 和 23 小时距离过大
        hour = dataframe["date"].dt.hour
        day_of_week = dataframe["date"].dt.dayofweek

        dataframe["%-hour_sin"] = np.sin(2 * np.pi * hour / 24)
        dataframe["%-hour_cos"] = np.cos(2 * np.pi * hour / 24)
        dataframe["%-dow_sin"] = np.sin(2 * np.pi * day_of_week / 7)
        dataframe["%-dow_cos"] = np.cos(2 * np.pi * day_of_week / 7)

        # === 价格极值（近期高低点）===
        dataframe["%-high_24h"] = dataframe["high"].rolling(24).max()
        dataframe["%-low_24h"] = dataframe["low"].rolling(24).min()
        dataframe["%-close_to_high_24h"] = (dataframe["close"] - dataframe["%-low_24h"]) / \
                                        (dataframe["%-high_24h"] - dataframe["%-low_24h"] + 1e-8)

        return dataframe
    # ====== set_freqai_targets: 目标改为回归型（future max/min return） ======
    def set_freqai_targets(self, df, metadata: dict, **kwargs):
        """
        生成回归目标：
        &t-max_ret : 未来 horizon 内的最大上涨率 (max(high)/close -1)
        &t-min_ret : 未来 horizon 内的最小下跌率 (min(low)/close -1)  (负数)
        保持末尾 horizon 行为 NaN。
        """
        df = df.copy()
        horizon = 4  # 推荐 8 根 = 24 分钟（可 grid-search）
        n = len(df)
        if n < horizon + 2:
            df["&t-max_ret"] = np.nan
            df["&t-min_ret"] = np.nan
            return df

        # future windows using sliding_window_view
        highs = df["high"].values
        lows = df["low"].values
        closes = df["close"].values

        from numpy.lib.stride_tricks import sliding_window_view
        sw_high = sliding_window_view(highs, window_shape=horizon)  # shape (n-h+1, h)
        sw_low = sliding_window_view(lows, window_shape=horizon)

        # we want future windows starting at i+1, so slice [1:valid_len+1]
        valid_len = len(sw_high) - 1
        future_high = sw_high[1: valid_len + 1]  # shape (valid_len, horizon)
        future_low = sw_low[1: valid_len + 1]

        cur = closes[:valid_len]

        max_ret = (future_high.max(axis=1) - cur) / (cur + 1e-12)
        min_ret = (future_low.min(axis=1) - cur) / (cur + 1e-12)  # 负数 if drop

        # pad to original length
        pad = np.array([np.nan] * horizon)
        df["&t-max_ret"] = np.concatenate([max_ret, pad])
        df["&t-min_ret"] = np.concatenate([min_ret, pad])

        return df


    # ====== populate_indicators: 添加 EMA200 / ATR_norm / ADX ======
    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        """
        计算额外的过滤指标以用于进出场决策
        """

        dataframe = self.freqai.start(dataframe, metadata, self)

        # keep other previous engineered features unchanged
        return dataframe


    # ====== populate_entry_trend: 用预测回归结果并加过滤器 ======
    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """
        进场：只在多头趋势且波动率充足时，且模型预期有边际收益时入场
        逻辑（多头示例）：
        - do_predict == 1
        - close > EMA200
        - ATR_norm > atr_th
        - edge = pred_up + pred_dn > edge_th
        - pred_up > min_up
        类似地可扩展空头（你 can_short=False 当前)
        """
        df = df.copy()
        
        # EMA200 on 3min (200 candles)
        df["%-ema-200"] = ta.EMA(df, timeperiod=20)

        # ATR 14 -> normalized by price
        atr = ta.ATR(df["high"], df["low"], df["close"], timeperiod=7)
        df["%-atr"] = atr
        df["%-atr_norm"] = atr / (df["close"] + 1e-12)

        # ADX for trend strength
        adx = ta.ADX(df, timeperiod=7)
        df["%-adx"] = adx

        # 超参（可 tune）
        atr_th = 0.0015    # ATR_norm threshold (0.15%)
        edge_th = 0.0015   # minimal expected edge (0.15%)
        min_up = 0.003     # minimal predicted upside (0.1%)

        # 预测列（freqai 输出），若不存在则回退到真值（仅离线分析/回测时）
        pred_up = df.get("&t-max_ret_pred")
        pred_dn = df.get("&t-min_ret_pred")

        if pred_up is None or pred_dn is None:
            # 没有 predictor 输出时，使用未来真值作为回测目标（仅用于离线调参）
            pred_up = df.get("&t-max_ret")
            pred_dn = df.get("&t-min_ret")

        # ensure numpy arrays
        pred_up = np.asarray(pred_up, dtype=float)
        pred_dn = np.asarray(pred_dn, dtype=float)

        trend_up_ok = df["close"] > df["%-ema-200"]
        trend_dn_ok = df["close"] < df["%-ema-200"]
        vol_ok = df["%-atr_norm"] > atr_th

        # edge measured as predicted upside plus predicted down (down is negative) => net edge
        edge = pred_up + pred_dn

        long_cond = (
            (df["do_predict"] == 1) &
            (trend_up_ok) &
            (vol_ok) &
            (edge > edge_th) &
            (pred_up > min_up) 
        )

        df.loc[long_cond, ["enter_long", "enter_tag"]] = (1, "long")
        if self.can_short:
            short_cond = (
                (df["do_predict"] == 1) &
                (trend_dn_ok) &
                (vol_ok) &
                (edge < -edge_th) & 
                (pred_dn < -min_up)
            )
            df.loc[short_cond, ["enter_short", "enter_tag"]] = (1, "short")
        return df


    # ====== populate_exit_trend: 动态退出基于预测与价格 ======
    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        """
        退出逻辑：
        - 若模型预测未来下跌风险扩大（pred_min < exit_dn_th） -> 提前退出
        - 若模型预测 future upside 大幅缩小 -> 提前退出
        - 可结合价格突破短期均线 / 反向信号做退出
        """
        df = df.copy()

        exit_dn_th = -0.003   # 若预测未来最小回撤 < -0.3%，提前退出
        min_up_when_keep = 0.0008  # 如果预测上涨低于 0.08%，也退出

        pred_up = df.get("&t-max_ret_pred")
        pred_dn = df.get("&t-min_ret_pred")
        if pred_up is None or pred_dn is None:
            pred_up = df.get("&t-max_ret")
            pred_dn = df.get("&t-min_ret")

        pred_up = np.asarray(pred_up, dtype=float)
        pred_dn = np.asarray(pred_dn, dtype=float)

        # 当模型预测下跌变大或上涨微乎其微 -> 退出多头
        exit_long_cond = (
            (df["do_predict"] == 1) &
            (pred_dn < exit_dn_th) |
            ((pred_up < min_up_when_keep) & (df["enter_tag"] == "long"))
        )

        df.loc[exit_long_cond, "exit_long"] = 1

        # 空头退出（若启用空头，可用相反逻辑）
        exit_short_cond = False
        if self.can_short:
            exit_short_cond = (
                (df["do_predict"] == 1) &
                (pred_up < -exit_dn_th)  # 若上行预测转差则出空
            )
            df.loc[exit_short_cond, "exit_short"] = 1

        return df


    # ====== confirm_trade_entry: 强化成交价格保护（避免滑点小亏） ======
    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        current_time,
        entry_tag,
        side: str,
        **kwargs,
    ) -> bool:
        """
        价格保护：若是多头，拒绝以超过当前 close 的太高价格成交（防止追高）
        这里阈值设置为 0.25%（可调整）
        """
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = df.iloc[-1].squeeze()
        guard = 0.0025  # 0.25%

        if side == "long":
            # 防止以高于 last close * (1 + guard) 的价格成交（追高）
            if rate > (last_candle["close"] * (1 + guard)):
                return False
        else:
            if rate < (last_candle["close"] * (1 - guard)):
                return False

        return True