MyXGB

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 freqtrade.strategy import IntParameter, IStrategy, merge_informative_pair  # noqa


from numpy.lib.stride_tricks import sliding_window_view

def calc_ofi(df):
    """
    Order Flow Imbalance (高频交易最常用特征之一)
    """
    ofi = (
        (df["bid_size"] - df["bid_size"].shift(1)) -
        (df["ask_size"] - df["ask_size"].shift(1))
    )
    return ofi.fillna(0)

def realized_vol(df, window=20):
    """
    高频 realized volatility
    """
    ret = np.log(df["close"]).diff()
    return (ret**2).rolling(window).sum()

def micro_price(df):
    """
    微价格 microprice = (bid*pask + ask*pbid) / (bid+ask)
    """
    return (
        df["ask_price"] * df["bid_size"] +
        df["bid_price"] * df["ask_size"]
    ) / (df["bid_size"] + df["ask_size"] + 1e-9)

def vpin(df, bucket=50):
    """
    VPIN 高频风险特征（机构大量使用）
    """
    ret = df["close"].pct_change()
    volume = df["volume"]
    dfv = (np.abs(ret) * volume)
    return dfv.rolling(bucket).sum() / (volume.rolling(bucket).sum() + 1e-9)

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.5 + abs(valid_ratio-0.4) * 0.5
    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 = [8, 10, 20, 40]

    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(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.02, "15":0.01,"240": -1}
    minimal_roi = {"0": 0.010, }

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

    process_only_new_candles = True
    stoploss = -0.010
    use_exit_signal = False
    # this is the maximum period fed to talib (timeframe independent)
    startup_candle_count: int = 40
    can_short = True
    buy_rsi = IntParameter(low=1, high=50, default=30, space="buy", optimize=True, load=True)
    sell_rsi = IntParameter(low=50, high=100, default=70, space="sell", optimize=True, load=True)
    short_rsi = IntParameter(low=51, high=100, default=70, space="sell", optimize=True, load=True)
    exit_short_rsi = IntParameter(low=1, high=50, default=30, space="buy", optimize=True, load=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=20)
        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
        # --- 高频结构特征 ---
        if "bid_price" in dataframe.columns:
            dataframe["%-ofi"] = calc_ofi(dataframe)
            dataframe["%-microprice"] = micro_price(dataframe)
        else:
            dataframe["%-ofi"] = 0
            dataframe["%-microprice"] = dataframe["close"]

        dataframe["%-vpin"] = vpin(dataframe)

        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

    def set_freqai_targets(self, df, metadata: dict, **kwargs):

        close = df['close'].values
        high = df['high'].values
        low = df['low'].values
        vol = np.std(df['volume'].values)
        # 自动搜索最佳参数
        # (best_up, best_horizon), labels = grid_search_tb(close, high, low)
        labels, _, _ = triple_barrier_labeling(
            close, high, low, 
            up_pct=0.02, down_pct=0.02, horizon=10
        )
        # print("最佳参数：")
        # print("  阈值（%）:", best_up)
        # print("  时间长度:", best_horizon)

        # ② 保存到 df
        df["&s-up_or_down"] = labels
        df["&s-up_or_down"] = df["&s-up_or_down"].fillna('same')
        print(df["&s-up_or_down"].value_counts())
        return df

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        # All indicators must be populated by feature_engineering_*() functions

        # the model will return all labels created by user in `set_freqai_targets()`
        # (& appended targets), an indication of whether or not the prediction should be accepted,
        # the target mean/std values for each of the labels created by user in
        # `set_freqai_targets()` for each training period.

        dataframe = self.freqai.start(dataframe, metadata, self)
        # Bollinger Bands
        dataframe["rsi"] = ta.RSI(dataframe)
        bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2)
        dataframe["bb_lowerband"] = bollinger["lower"]
        dataframe["bb_middleband"] = bollinger["mid"]
        dataframe["bb_upperband"] = bollinger["upper"]
        dataframe["bb_percent"] = (dataframe["close"] - dataframe["bb_lowerband"]) / (
            dataframe["bb_upperband"] - dataframe["bb_lowerband"]
        )
        dataframe["bb_width"] = (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe[
            "bb_middleband"
        ]

        # TEMA - Triple Exponential Moving Average
        dataframe["tema"] = ta.TEMA(dataframe, timeperiod=9)
        return dataframe

    def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame:

        enter_long_conditions = (
            (df["do_predict"] == 1) &
            (df["rsi"] > self.buy_rsi.value) &
            (df["tema"] <= df["bb_middleband"]) & # Guard: tema below BB middle
            (df["tema"] > df["tema"].shift(1)) &  # Guard: tema is raising
            (df["&s-up_or_down"] == "up")
        )

        df.loc[enter_long_conditions, ["enter_long", "enter_tag"]] = (1, "long")

        enter_short_conditions = (
            (df["do_predict"] == 1) &
            (df["rsi"] < self.short_rsi.value) &
            (df["tema"] > df["bb_middleband"]) &  # Guard: tema above BB middle
            (df["tema"] < df["tema"].shift(1)) &  # Guard: tema is falling
            (df["&s-up_or_down"] == "down")
        )
            # Signal: RSI crosses above 70
            # (qtpylib.crossed_above(df["rsi"], self.short_rsi.value)) &


        df.loc[enter_short_conditions, ["enter_short", "enter_tag"]] = (1, "short")

        return df

    def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame:
        exit_long_conditions = [df["do_predict"] == 1, df["&s-up_or_down"] == "down"]
        if exit_long_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_long_conditions), "exit_long"] = 1

        exit_short_conditions = [df["do_predict"] == 1, df["&s-up_or_down"] == "up"]
        if exit_short_conditions:
            df.loc[reduce(lambda x, y: x & y, exit_short_conditions), "exit_short"] = 1

        return df

    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:
        df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        last_candle = df.iloc[-1].squeeze()

        if side == "long":
            if rate > (last_candle["close"] * (1 + 0.0025)):
                return False
        else:
            if rate < (last_candle["close"] * (1 - 0.0025)):
                return False

        return True