MyXGB_gpt

"""
机构级 高频回归策略（强化版）
支持：
- Triple Barrier 自动调参
- 高频微结构特征 (OFI / VPIN / RV)
- Regime-aware 进出场过滤
- XGBoost Regressor 更专业参数
"""

import logging
import numpy as np
import pandas as pd
import talib.abstract as ta
from pandas import DataFrame
from technical import qtpylib
from numpy.lib.stride_tricks import sliding_window_view

from freqtrade.strategy import IStrategy

logger = logging.getLogger(__name__)
# -------------------------------------------------------
# -------------- 高频强化型特征区 ------------------------
# -------------------------------------------------------

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)


# -------------------------------------------------------
# -------------- Triple Barrier 强化版 -------------------
# -------------------------------------------------------

def triple_barrier_labeling(close, high, low, up_pct, down_pct, horizon):
    """
    精简为回归模式使用的标签。
    """
    n = len(close)
    valid_len = n - horizon

    future_high = sliding_window_view(high, horizon)[1:]
    future_low = sliding_window_view(low, horizon)[1:]

    c = close[:valid_len]

    max_ret = (future_high.max(axis=1) - c) / c
    min_ret = (future_low.min(axis=1) - c) / c

    pad = np.array([np.nan] * horizon)
    return (
        np.concatenate([max_ret, pad]),
        np.concatenate([min_ret, pad]),
    )


def auto_tb_params(close, high, low):
    """
    自动调最佳 horizon & pct （网格少量）
    """
    candidates = [(3,0.003),(4,0.005),(6,0.008),(8,0.010)]
    best = None
    best_score = -999
    
    for h,p in candidates:
        maxr, minr = triple_barrier_labeling(close, high, low, p, p, h)
        valid = (~np.isnan(maxr)) & (~np.isnan(minr))
        r = np.corrcoef(maxr[valid], -minr[valid])[0,1]  # balance
        if r > best_score:
            best_score = r
            best = (h,p)
    return best


# -------------------------------------------------------
# -------------- 策略主体 --------------------------------
# -------------------------------------------------------

class MyXGB_gpt(IStrategy):

    timeframe = "5m"
    process_only_new_candles = True
    startup_candle_count = 250
    can_short = True

    minimal_roi = {"0": 0.008, "30": 0.004, "120": -1}
    stoploss = -0.012

    # =====================================================
    #  多尺度 & 高频特征
    # =====================================================
    def feature_engineering_expand_basic(self, df, metadata: dict, **kwargs):

        df = df.copy()

        # --- 高频结构特征 ---
        if "bid_price" in df.columns:
            df["%-ofi"] = calc_ofi(df)
            df["%-microprice"] = micro_price(df)
        else:
            df["%-ofi"] = 0
            df["%-microprice"] = df["close"]

        df["%-vpin"] = vpin(df)
        df["%-realized_vol"] = realized_vol(df)

        # --- 价格变化 ---
        df["%-ret1"] = df["close"].pct_change()
        df["%-ret3"] = df["close"].pct_change(3)
        df["%-ret6"] = df["close"].pct_change(6)

        # --- 波动率 ---
        atr = ta.ATR(df, timeperiod=14)
        df["%-atr_norm"] = atr / df["close"]

        # --- 超级趋势 ADX ---
        adx = ta.ADX(df, timeperiod=14)
        df["%-adx"] = adx

        # --- Regime ---
        df["%-regime_vol"] = df["%-atr_norm"].rolling(40).mean()

        return df

    # =====================================================
    #  回归目标
    # =====================================================
    def set_freqai_targets(self, df, metadata, **kwargs):
        df = df.copy()

        close = df["close"].values
        high = df["high"].values
        low = df["low"].values

        horizon, pct = auto_tb_params(close, high, low)

        max_ret, min_ret = triple_barrier_labeling(
            close, high, low,
            up_pct=pct,
            down_pct=pct,
            horizon=horizon
        )

        df["&t-max_ret"] = max_ret
        df["&t-min_ret"] = min_ret

        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)

        return dataframe

    # =====================================================
    #  进场逻辑（regime-aware）
    # =====================================================
    def populate_entry_trend(self, df, metadata) -> DataFrame:

        df = df.copy()

        pred_up = df["&t-max_ret"]
        pred_dn = df["&t-min_ret"]

        # --- Market Regime ---
        # 高频交易最重要部分之一
        atr = ta.ATR(df, timeperiod=14)
        df["%-atr_norm"] = atr / df["close"]
        df["%-regime_vol"] = df["%-atr_norm"].rolling(40).mean()
        regime = df["%-regime_vol"]
        high_vol = regime > regime.rolling(200).median()

        # --- 动态阈值 ---
        base_up = 0.002
        base_edge = 0.0015
        
        up_th = np.where(high_vol, base_up * 0.6, base_up * 1.4)
        edge_th = np.where(high_vol, base_edge * 0.7, base_edge * 1.3)

        edge = pred_up + pred_dn

        long = (
            (df["do_predict"] == 1) &
            (pred_up > up_th) &
            (edge > edge_th)
        )

        short = (
            (df["do_predict"] == 1) &
            (pred_dn < -up_th) &
            (edge < -edge_th)
        )

        df.loc[long, ["enter_long","enter_tag"]] = (1,"L")
        df.loc[short, ["enter_short","enter_tag"]] = (1,"S")

        return df

    # =====================================================
    # 退出（预测变化 + regime aware）
    # =====================================================
    def populate_exit_trend(self, df, metadata):
        df = df.copy()

        pred_up = df["&t-max_ret"].astype(float)
        pred_dn = df["&t-min_ret"].astype(float)

        # 动态退出
        exit_l = (
            (pred_dn < -0.002) |
            (pred_up < 0.0005)
        )
        exit_s = (
            (pred_up > 0.002) |
            (pred_dn > -0.0005)
        )

        df.loc[exit_l, "exit_long"] = 1
        df.loc[exit_s, "exit_short"] = 1

        return df