pair_trading_run_V1_J_price_0811

# --- Import Freqtrade Libraries ---
from typing import Optional, Union

import numpy as np
from freqtrade.strategy.interface import IStrategy
from datetime import datetime, timedelta, timezone
from freqtrade.persistence import Trade
from datetime import datetime, timedelta
from freqtrade.exchange import timeframe_to_prev_date

import pandas as pd
import logging

logger = logging.getLogger(__name__)


#  Discord real-time push for this strategy, welcome to join: https://discord.gg/3EABfUPxbQ
class pair_trading_run_V1_J_price_0811(IStrategy):
    can_short = True
    timeframe = "5m"
    startup_candle_count = 3
    process_only_new_candles = True
    # Trading parameters
    stoploss = -200
    trailing_stop = False
    use_custom_stoploss = False  # Enable dynamic stop loss
    Zscore_entry = 1  # Z-score entry threshold
    Zscore_entry1 = 1.1 * Zscore_entry  # Z-score entry threshold 1.1x
    Zscore_exit = 0.8  # Z-score exit threshold
    Zscore_stop = 3  # Z-score stop loss threshold (loss close)
    can_trade_usdt = False
    min_consistency_score = 0.0002
    zscore_mean_window = 10  # Z-score mean calculation window
    zscore_abs_max_window = 100  # Z-score absolute maximum calculation window
    leverage1 = 10

    # order_types = {
    #    'entry': 'limit',
    #    'exit': 'market',
    #    'emergency_exit': 'market',
    #    'force_entry': 'market',
    #    'force_exit': "market",
    #    'stoploss': 'market',
    #    'stoploss_on_exchange': False,
    #    'stoploss_on_exchange_interval': 60,
    #    'stoploss_on_exchange_market_ratio': 0.99
    # }

    def __init__(self, config):
        super().__init__(config)
        # self.free_usdt = 0
        # self.free_usdt = 0.5 * self.free_usdt  # Assume 60% of USDT is available
        self.whitelist = config.get("exchange", {}).get("pair_whitelist", [])
        self.pvalue_dict = config.get("pvalue_dict", {})
        self.gamma_dict = config.get("gamma_dict", {})
        self.c_dict = config.get("c_dict", {})
        self.z_mean_dict = config.get("z_mean_dict", {})
        self.z_std_dict = config.get("z_std_dict", {})
        self.adfstat_dict = config.get("adfstat_dict", {})
        self.consistency_score_dict = config.get("consistency_score_dict", {})
        self.half_life_dict = config.get("half_life_dict", {})
        self.z_cross_zero_count_dict = config.get("z_cross_zero_count_dict", {})
        # Pair dictionary related
        self.all_candidate_pairs = {}
        self.tradable_pairs = {}
        self.stake_allocations = {}
        self.pair_states = {}
        ### Update: Add a timer to update the pair pool every 4 hours
        self.last_update_time = datetime(1945, 8, 15, tzinfo=timezone.utc)
        self.update_interval = timedelta(days=30)

    def _initialize_all_candidates(self) -> list:

        pvalue_filtered = {k: v for k, v in self.pvalue_dict.items() if v <= 0.0001}
        logger.info(f"Remaining pairs after p-value filtering: {len(pvalue_filtered)}")
        logger.info(
            f"First 5 pvalue_filtered pairs: {list(pvalue_filtered.items())[:5]}"
        )

        # Filter pairs based on zmean from pvalue_filtered
        zmean_filtered = {}
        for p_key, pvalue in pvalue_filtered.items():
            try:
                a_key, b_key = p_key.replace("_pvalue", "").split("_")
                zmean_key = f"{a_key}_{b_key}_z_mean"
                z_mean_value = self.z_mean_dict.get(zmean_key)

                # Filter based on z_mean value
                if z_mean_value is not None and abs(z_mean_value) < 0.00001:
                    zmean_filtered[p_key] = pvalue

            except Exception as e:
                logger.error(f"Error during zmean filtering: {e} for key {p_key}")
                continue
        logger.info(f"Remaining pairs after zmean filtering: {len(zmean_filtered)}")
        logger.info(f"First 5 zmean_filtered pairs: {list(zmean_filtered.items())[:5]}")

        # Filter pairs based on z_cross_zero_count from zstd_filtered
        z_cross_zero_count_filtered = {}
        for p_key, pvalue in zmean_filtered.items():
            try:
                a_key, b_key = p_key.replace("_pvalue", "").split("_")
                z_cross_zero_count_key = f"{a_key}_{b_key}_z_cross_zero_count"
                z_cross_zero_count_value = self.z_cross_zero_count_dict.get(
                    z_cross_zero_count_key
                )

                # Filter based on z_cross_zero_count value
                if (
                    z_cross_zero_count_value is not None
                    and z_cross_zero_count_value >= 1000
                ):
                    z_cross_zero_count_filtered[p_key] = pvalue

            except Exception as e:
                logger.error(
                    f"Error during z_cross_zero_count filtering: {e} for key {p_key}"
                )
                continue
        logger.info(
            f"Remaining pairs after z_cross_zero_count filtering: {len(z_cross_zero_count_filtered)}"
        )
        logger.info(
            f"First 5 z_cross_zero_count_filtered pairs: {list(z_cross_zero_count_filtered.items())[:5]}"
        )

        # Filter pairs based on adf from z_cross_zero_count_filtered
        adfstat_filtered = {}
        for p_key, pvalue in z_cross_zero_count_filtered.items():
            try:
                a_key, b_key = p_key.replace("_pvalue", "").split("_")
                adfstat_key = f"{a_key}_{b_key}_adfstat"
                adfstat_value = self.adfstat_dict.get(adfstat_key)

                # ADF statistic needs to be less than critical value (more negative) to consider series stationary.
                # -3.0 is an empirical strict threshold.
                if adfstat_value is not None and adfstat_value < -3.0:
                    adfstat_filtered[p_key] = pvalue

            except Exception as e:
                logger.error(f"Error during adf filtering: {e} for key {p_key}")
                continue
        logger.info(f"Remaining pairs after adf filtering: {len(adfstat_filtered)}")
        logger.info(
            f"First 5 adfstat_filtered pairs: {list(adfstat_filtered.items())[:5]}"
        )

        # Filter pairs based on half_life from adfstat_filtered
        half_life_filtered = {}
        for p_key, pvalue in adfstat_filtered.items():
            try:
                a_key, b_key = p_key.replace("_pvalue", "").split("_")
                half_life_key = f"{a_key}_{b_key}_half_life"
                half_life_value = self.half_life_dict.get(half_life_key)

                # Filter based on half_life value
                # if half_life_value is not None and half_life_value < 2000:
                if half_life_value is not None and half_life_value < 700:
                    half_life_filtered[p_key] = pvalue

            except Exception as e:
                logger.error(f"Error during half_life filtering: {e} for key {p_key}")
                continue
        logger.info(
            f"Remaining pairs after half_life filtering: {len(half_life_filtered)}"
        )
        logger.info(
            f"First 5 half_life_filtered pairs: {list(half_life_filtered.items())[:5]}"
        )

        candidates = []
        balance = self.wallets.get_total("USDT")
        for p_key, pvalue in half_life_filtered.items():
            try:
                a_key, b_key = p_key.replace("_pvalue", "").split("_")
                adf_key = f"{a_key}_{b_key}_adfstat"
                adfstat = self.adfstat_dict.get(adf_key)
                a_df = self.dp.get_pair_dataframe(pair=a_key, timeframe=self.timeframe)
                if self.dp.runmode.value in ("live", "dry_run"):
                    a_price = a_df["close"].iloc[-1]
                else:
                    a_price = a_df["close"].iloc[0]

                b_df = self.dp.get_pair_dataframe(pair=b_key, timeframe=self.timeframe)
                if self.dp.runmode.value in ("live", "dry_run"):
                    b_price = b_df["close"].iloc[-1]
                else:
                    b_price = b_df["close"].iloc[0]

                gamma = self.gamma_dict.get(f"{a_key}_{b_key}_gamma")
                if a_price is None or b_price is None or gamma is None:
                    continue

                # 1. Define financial constants
                MIN_STAKE_USDT = 5.0  # Minimum order nominal value required by exchange

                # 2. Use dollar neutrality method to calculate target position value
                #    This method is the correct algorithm that matches the log price regression model
                #    公式: Stake_B = Budget / (gamma + 1), Stake_A = Budget - Stake_B
                if (gamma + 1) <= 0:
                    continue  # Avoid division by zero or negative numbers

                # target_stake_B = self.free_usdt / (gamma + 1)
                # target_stake_A = self.free_usdt - target_stake_B
                target_stake_B = balance / (gamma + 1)
                target_stake_A = balance - target_stake_B

                # 3. Check if position value meets minimum requirements (leverage independent!)
                if target_stake_A < MIN_STAKE_USDT or target_stake_B < MIN_STAKE_USDT:
                    continue

                half_key = f"{a_key}_{b_key}_half_life"
                half_life = self.half_life_dict.get(half_key)

                ### 修改
                candidates.append(
                    {
                        "raw_pair_key": f"{a_key}_{b_key}",
                        "pair_A": a_key,
                        "pair_B": b_key,
                        # 'consistency_score': consistency_score,
                        "half_life": half_life,
                        "adfstat": adfstat,
                    }
                )
            except Exception as e:
                logger.error(f"Error during data preloading: {e} for key {p_key}")

        # 按照半衰期升序排序
        logger.info(f"Number of pairs before price filtering: {len(candidates)}")
        candidates = sorted(candidates, key=lambda x: x["half_life"], reverse=False)
        # 前面已经出现过的币种，后面直接删除掉改配对，例如出现过A-B，后面出现的B-C就不需要了
        seen_currencies = set()
        filtered_candidates = []
        for candidate in candidates:
            pair_A_base = candidate["pair_A"].split("/")[0]
            pair_B_base = candidate["pair_B"].split("/")[0]

            if pair_A_base in seen_currencies or pair_B_base in seen_currencies:
                continue
            seen_currencies.add(pair_A_base)
            seen_currencies.add(pair_B_base)
            filtered_candidates.append(candidate)
        # logger.info(f"Number of pairs after filtering: {len(filtered_candidates)}")
        candidates = filtered_candidates
        logger.info(
            f"Number of pairs after removing duplicate currencies: {len(candidates)}"
        )
        return candidates
        # return candidates

    def _update_tradable_pairs(self):
        logger.warning(f"Triggering dynamic pair pool update...")
        open_trades = Trade.get_trades_proxy(is_open=True)
        new_tradable_pairs = {}
        locked_currencies = set()

        # 1. 无条件保留并锁定已有持仓的配对
        for trade in open_trades:
            if not trade.enter_tag:
                continue
            try:
                trade_pair_key = "_".join(trade.enter_tag.split("_")[2:])
                if (
                    trade_pair_key in self.tradable_pairs
                    and trade_pair_key not in new_tradable_pairs
                ):
                    new_tradable_pairs[trade_pair_key] = self.tradable_pairs[
                        trade_pair_key
                    ]

                    pair_A_base = new_tradable_pairs[trade_pair_key]["pair_A"].split(
                        "/"
                    )[0]
                    pair_B_base = new_tradable_pairs[trade_pair_key]["pair_B"].split(
                        "/"
                    )[0]
                    locked_currencies.add(pair_A_base)
                    locked_currencies.add(pair_B_base)
                    # logger.info(f"Keeping existing position pair {trade_pair_key}, locking currencies: {pair_A_base}, {pair_B_base}")
            except Exception as e:
                logger.error(f"Error processing existing positions: {e}")
        logger.info(f"Number of existing position pairs: {len(new_tradable_pairs)}")
        # 2. 贪婪选择，填充剩余名额
        max_allowed_pairs = (self.config.get("max_open_trades", 1)) / 2
        logger.info(
            f"Maximum tradable pairs allowed by maxopentrade: {max_allowed_pairs}"
        )
        logger.info(f"Number of initially filtered pairs: {len(new_tradable_pairs)}")
        logger.info(
            f"Number of pairs after filter function screening: {len(self.all_candidate_pairs)}"
        )
        for candidate in self.all_candidate_pairs:
            # logger.info(f"Checking candidate pair: {candidate['raw_pair_key']} ")
            if len(new_tradable_pairs) >= max_allowed_pairs:
                logger.info(
                    f"Reached maximum tradable pairs limit: {max_allowed_pairs}"
                )
                break

            pair_A_base = candidate["pair_A"].split("/")[0]
            pair_B_base = candidate["pair_B"].split("/")[0]

            if pair_A_base in locked_currencies or pair_B_base in locked_currencies:
                continue

            pair_key_sorted = "_".join((candidate["pair_A"], candidate["pair_B"]))
            if pair_key_sorted not in new_tradable_pairs:
                new_tradable_pairs[pair_key_sorted] = {
                    "pair_A": candidate["pair_A"],
                    "pair_B": candidate["pair_B"],
                    "gamma": self.gamma_dict.get(f"{candidate['raw_pair_key']}_gamma"),
                    "c": self.c_dict.get(f"{candidate['raw_pair_key']}_c"),
                    "z_mean": self.z_mean_dict.get(
                        f"{candidate['raw_pair_key']}_z_mean"
                    ),
                    "z_std": self.z_std_dict.get(f"{candidate['raw_pair_key']}_z_std"),
                    "half_life": candidate["half_life"],
                    "adfstat": candidate["adfstat"],
                    " z_cross_zero_count": self.z_cross_zero_count_dict.get(
                        f"{candidate['raw_pair_key']}_z_cross_zero_count"
                    ),
                }
                locked_currencies.add(pair_A_base)
                locked_currencies.add(pair_B_base)
                # logger.info(f"Dynamically selected new pair {pair_key_sorted} (consistency score: {candidate['consistency_score']:.2f})")

        # self.tradable_pairs = new_tradable_pairs
        # self._allocate_stake_amounts()
        # logger.info(f"Updated tradable pairs count: {len(self.tradable_pairs)}")

        # 3. 原子化地更新 tradable_pairs 和资金分配
        self.tradable_pairs = new_tradable_pairs
        logger.info(f"更新后的可交易配对数量: {len(self.tradable_pairs)}")

        if len(self.tradable_pairs) > 0:
            # capital_per_pair = self.free_usdt / len(self.tradable_pairs)
            # capital_per_pair = self.free_usdt / 6
            capital_per_pair = self.wallets.get_total("USDT") / len(self.tradable_pairs)
            self.stake_allocations = {
                key: capital_per_pair for key in self.tradable_pairs
            }
        else:
            self.stake_allocations = {}
        # logger.info(f"Stake allocation: {self.stake_allocations}")
        logger.warning(
            f"Dynamic pair pool update completed. Current tradable pairs: {list(self.tradable_pairs.keys())}"
        )
        logger.warning(f"Current tradable pairs: {self.tradable_pairs}")

    def populate_indicators(
        self, dataframe: pd.DataFrame, metadata: dict
    ) -> pd.DataFrame:
        current_time = dataframe.iloc[-1]["date"].to_pydatetime()
        if current_time >= self.last_update_time + self.update_interval:
            # self.free_usdt = 0.6 * self.wallets.get_total("USDT")
            # self.free_usdt = 0.6 * 2000
            self.all_candidate_pairs = self._initialize_all_candidates()
            self._update_tradable_pairs()
            self.last_update_time = current_time

        current_pair = metadata["pair"]
        dataframe["zero"] = 0
        dataframe["2"] = 2
        dataframe["-2"] = -2
        dataframe["-1"] = -1
        dataframe["1"] = 1
        dataframe["-0.8"] = -0.8
        dataframe["0.8"] = 0.8

        # 收集所有新列
        new_columns = {}

        for pair in self.whitelist:
            pair_df = self.dp.get_pair_dataframe(pair=pair, timeframe=self.timeframe)
            new_columns[f"{pair}_log_close"] = pair_df["close"]

        # 为每个配对计算 Z-score 列
        for pair_key_str, params in self.tradable_pairs.items():
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]

            if (
                f"{pair_A}_log_close" in new_columns
                and f"{pair_B}_log_close" in new_columns
            ):
                gamma = params.get("gamma")
                c = params.get("c")
                z_mean = params.get("z_mean")
                z_std = params.get("z_std")

                if all(p is not None for p in [gamma, c, z_mean, z_std]) and z_std != 0:
                    y_series = new_columns[f"{pair_A}_log_close"]
                    x_series = new_columns[f"{pair_B}_log_close"]

                    z_value = self.zvalue(y_series, x_series, gamma, c)
                    z_score = (z_value - z_mean) / z_std

                    new_columns[f"{pair_A}_{pair_B}_Zscore"] = z_score
                    new_columns[f"{pair_A}_{pair_B}_Zscore_mean"] = z_score.rolling(
                        window=self.zscore_mean_window
                    ).mean()

        # 一次性添加所有新列
        if new_columns:
            new_df = pd.DataFrame(new_columns, index=dataframe.index)
            dataframe = pd.concat([dataframe, new_df], axis=1)

        return dataframe

    def populate_entry_trend(
        self, dataframe: pd.DataFrame, metadata: dict
    ) -> pd.DataFrame:
        current_pair = metadata["pair"]
        candle_time = dataframe.iloc[-1]["date"]
        # pair_0, pair_1 = self.pair_0, self.pair_1

        # logger.info(f"--- [populate_entry_trend] for {current_pair} at {dataframe.iloc[-1]['date']} ---")

        for pair_key, params in self.tradable_pairs.items():
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]
            zscore_col_name = f"{pair_A}_{pair_B}_Zscore"
            zscore_col_name_mean = f"{pair_A}_{pair_B}_Zscore_mean"

            if zscore_col_name in dataframe.columns and (
                current_pair == pair_A or current_pair == pair_B
            ):
                zscore = dataframe[zscore_col_name]
                zscore_mean = dataframe[zscore_col_name_mean]
                # zscore_diff = zscore - zscore.shift(1)
                # short_cond = (zscore > self.Zscore_entry) & (zscore_diff < 0)
                # long_cond = (zscore < -self.Zscore_entry) & (zscore_diff > 0)
                # short_cond = (zscore > self.Zscore_entry) & (zscore <= 1.5)
                # long_cond = (zscore < -self.Zscore_entry) & (zscore >= -1.5)
                # short_cond = (zscore > self.Zscore_entry) & (zscore.diff() < 0)
                # long_cond = (zscore < -self.Zscore_entry) & (zscore.diff() > 0)
                # short_cond = (zscore > self.Zscore_entry) &  (zscore < self.Zscore_entry1)
                # long_cond = (zscore < -self.Zscore_entry) &  (zscore > -self.Zscore_entry1)
                short_cond = zscore > self.Zscore_entry
                long_cond = zscore < -self.Zscore_entry
                # short_cond = (zscore > (1.00 * zscore_mean))  &  (zscore > 0.5)
                # long_cond = (zscore < (-1.00 * zscore_mean)) &  (zscore < -0.5)
                short_tag = f"entry_short_{pair_A}_{pair_B}"
                long_tag = f"entry_long_{pair_A}_{pair_B}"
                if current_pair == pair_A:
                    dataframe.loc[short_cond, ["enter_short", "enter_tag"]] = (
                        1,
                        short_tag,
                    )
                    dataframe.loc[long_cond, ["enter_long", "enter_tag"]] = (
                        1,
                        long_tag,
                    )
                if current_pair == pair_B:
                    dataframe.loc[short_cond, ["enter_long", "enter_tag"]] = (
                        1,
                        long_tag,
                    )
                    dataframe.loc[long_cond, ["enter_short", "enter_tag"]] = (
                        1,
                        short_tag,
                    )

        return dataframe

    def populate_exit_trend(
        self, dataframe: pd.DataFrame, metadata: dict
    ) -> pd.DataFrame:

        current_pair = metadata["pair"]
        for pair_key, params in self.tradable_pairs.items():
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]
            zscore_col_name = f"{pair_A}_{pair_B}_Zscore"
            if zscore_col_name in dataframe.columns and (
                current_pair == pair_A or current_pair == pair_B
            ):

                zscore = dataframe[zscore_col_name]
                open_trades = Trade.get_trades_proxy(is_open=True)
                # 计算配对A和B的收益
                trade_A = next((t for t in open_trades if t.pair == pair_A), None)
                trade_B = next((t for t in open_trades if t.pair == pair_B), None)

                profit_A = self.get_trade_profit(trade_A) if trade_A else 0.0
                profit_B = self.get_trade_profit(trade_B) if trade_B else 0.0

                # 判断收益是否大于0
                pair_profit_ratio = (profit_A + profit_B) / (
                    (self.wallets.get_total("USDT")) / len(self.tradable_pairs)
                )
                # pair_profit_ratio = (profit_A + profit_B) / (0.6 * 2000 / 6)
                # profit_condition = pair_profit_ratio > 0

                # exit_cond = (abs(zscore) < self.Zscore_exit) & profit_condition
                exit_cond = abs(zscore) < self.Zscore_exit
                exit_tag = f"populate_exit_trend_{pair_A}_{pair_B}_get_profit"
                dataframe.loc[exit_cond, ["exit_short", "exit_tag"]] = (1, exit_tag)
                dataframe.loc[exit_cond, ["exit_long", "exit_tag"]] = (1, exit_tag)

        return dataframe

    def custom_exit(
        self,
        pair: str,
        trade: "Trade",
        current_time: "datetime",
        current_rate: float,
        current_profit: float,
        **kwargs,
    ):
        # logger.info(f"--- [custom_exit] for {pair} at {current_time} ---")
        if current_time >= self.last_update_time + self.update_interval:
            # self.free_usdt = 0.6 * self.wallets.get_total('USDT')
            # self.free_usdt = 0.6 * 2000
            # self.all_candidate_pairs = self._initialize_all_candidates()
            # self._update_tradable_pairs()
            # self.last_update_time = current_time
            # logger.info(
            #    f"+++++++++++++++++++++++Updated tradable pairs at custom_exit time {current_time},last_update_time: {self.last_update_time}+++++++++++++++++++++++ ")
            pass

        open_trades = Trade.get_trades_proxy(is_open=True)
        all_profits = {}
        total_profit_amount = 0.0
        for open_trade in open_trades:
            profit_ratio = 0.0
            if open_trade.pair == pair:
                profit_ratio = current_profit
            else:
                other_pair = open_trade.pair
                (dataframe, _) = self.dp.get_analyzed_dataframe(
                    pair=other_pair, timeframe=self.timeframe
                )
                if not dataframe.empty:
                    last_candle = dataframe.iloc[-1]
                    rate_for_other_pair = last_candle["close"]
                    if open_trade.is_short:
                        # Short position profit ratio
                        profit_ratio = 1 - (rate_for_other_pair / open_trade.open_rate)
                    else:
                        # Long position profit ratio
                        profit_ratio = (rate_for_other_pair / open_trade.open_rate) - 1
                    profit_ratio *= open_trade.leverage
            profit_pct = profit_ratio * 100
            profit_amount = open_trade.stake_amount * profit_ratio
            total_profit_amount += profit_amount
            all_profits[open_trade.pair] = (
                f"{profit_pct:.2f}% ({profit_amount:.2f} {open_trade.stake_currency})"
            )
        ratio = total_profit_amount / self.wallets.get_total("USDT")
        # ratio = total_profit_amount / 2000
        # logger.info(f"log_____total_profit_amount:{total_profit_amount},ratio:{ratio}")
        if ratio <= -1:
            logger.info(
                "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
            )
            logger.info(
                f"Liquidity ratio at {current_time}: {ratio:.2%},strategy={self.__class__.__name__} failed "
            )
            logger.info(
                "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
            )
        self.liqutation_ratio = ratio

        dataframe, _ = self.dp.get_analyzed_dataframe(
            pair=pair, timeframe=self.timeframe
        )
        trade_date = timeframe_to_prev_date(self.timeframe, trade.open_date_utc)
        trade_candle = dataframe.loc[dataframe["date"] == trade_date]
        if not trade_candle.empty:
            trade_candle = trade_candle.iloc[0]
        else:
            # 没有找到对应K线，可以选择continue或return None
            return None
        current_candle = dataframe.iloc[-1].squeeze()
        current_pair = pair
        for pair_key, params in self.tradable_pairs.items():
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]
            zscore_col_name = f"{pair_A}_{pair_B}_Zscore"
            # 1. z分数退出
            if zscore_col_name in dataframe.columns and (
                current_pair == pair_A or current_pair == pair_B
            ):
                if abs(current_candle[zscore_col_name]) < 0.30 * abs(
                    trade_candle[zscore_col_name]
                ):
                    return f"Zscore_get_profit_{pair_A}_{pair_B}"

            trade_A = next((t for t in open_trades if t.pair == pair_A), None)
            trade_B = next((t for t in open_trades if t.pair == pair_B), None)
            # 更新浮动利润
            profit_A = self.get_trade_profit(trade_A) if trade_A else 0.0
            profit_B = self.get_trade_profit(trade_B) if trade_B else 0.0
            # pair_profit_ratio = (profit_A + profit_B) / ((0.6 * self.wallets.get_total('USDT')) / len(self.tradable_pairs))
            pair_profit_ratio = (profit_A + profit_B) / (
                self.wallets.get_total("USDT") / len(self.tradable_pairs)
            )
            state = self.pair_states.get(pair_key)

            # logger.info(f"Pair {pair_key} profit ratio: {pair_profit_ratio:.2%} (A: {profit_A:.2f}, B: {profit_B:.2f})")
            if (
                pair_profit_ratio >= (0.01 * self.leverage1)
                and (current_pair == pair_A or current_pair == pair_B)
                and state.get("one_pair_is_already_exit") == 0
            ):
                return f"Pair_profit_get_profit_{pair_A}_{pair_B}"

            if (
                state
                and state.get("one_pair_is_already_exit") == 1
                and (current_pair == pair_A or current_pair == pair_B)
            ):
                return f"Pair_already_exit_{pair_A}_{pair_B}"

        return None

    def _calculate_precise_stakes(
        self, pair_key: str, budget: float, pair_A_price: float, pair_B_price: float
    ) -> dict:
        logger.info(
            f"_calculate_precise_stakes called with pair_key={pair_key}, budget={budget}"
        )
        params = self.tradable_pairs.get(pair_key)
        if not params:
            logger.info(f"log----_calculate_precise_stakes----params={params}")
            return {"A_amount": 0, "B_amount": 0}

        gamma = params["gamma"]
        pair_A = params["pair_A"]
        pair_B = params["pair_B"]

        if gamma is None:
            logger.info(f"log----_calculate_precise_stakes----gamma={gamma}")
            return {"A_amount": 0, "B_amount": 0}

        # fee = 1.0015
        fee = 1
        price_A_with_fee = pair_A_price * fee
        price_B_with_fee = pair_B_price * fee
        min_usdt = 1.0 * len(self.tradable_pairs) / self.leverage1

        market_A = self.dp.market(pair_A)
        market_B = self.dp.market(pair_B)

        base_A_precision = market_A.get("precision", {}).get("amount", {})
        base_B_precision = market_B.get("precision", {}).get("amount", {})

        # logger.info(f"A_precision: {A_precision}  B_precision: {B_precision}")
        best_qty_A = 0
        best_qty_B = 0
        best_err = float("inf")

        if (price_A_with_fee + price_B_with_fee * gamma) < 0:
            logger.info(
                f"log----_calculate_precise_stakes----price_A_with_fee + price_B_with_fee * gamma={price_A_with_fee + price_B_with_fee * gamma}"
            )
            return {"A_amount": 0, "B_amount": 0}

        max_qty_A = budget / (price_A_with_fee + price_B_with_fee * gamma)

        qty_A_iterator = base_A_precision

        while qty_A_iterator <= max_qty_A + 1e-8:
            qty_A = round(qty_A_iterator, 8)
            cost_A = qty_A * price_A_with_fee
            if cost_A < min_usdt:
                qty_A_iterator += base_A_precision
                continue

            qty_B_theoretical = qty_A * gamma
            qty_B = round(qty_B_theoretical / base_B_precision) * base_B_precision
            qty_B = round(qty_B, 8)

            cost_B = qty_B * price_B_with_fee
            # logger.info(f"Checking qty_A={qty_A}, cost_A={cost_A}, cost_B={cost_B}")

            if cost_B < min_usdt:
                qty_A_iterator += base_A_precision
                continue

            total_cost = cost_A + cost_B
            if total_cost > budget:
                break

            ratio = qty_B / qty_A if qty_A > 0 else 0
            err = abs(ratio - gamma)
            # logger.info(f"err {err} ")

            if err < best_err:
                best_qty_A = qty_A
                best_qty_B = qty_B
                best_err = err
            # logger.info(f"best_err={best_err}")
            qty_A_iterator += base_A_precision

        final_stake_A = best_qty_A * pair_A_price
        final_stake_B = best_qty_B * pair_B_price

        logger.info(
            f"pair_key={pair_key}, price_A_with_fee={price_A_with_fee}, price_B_with_fee={price_B_with_fee}, gamma={gamma}"
        )
        logger.info(
            f"base_A_precision={base_A_precision}, base_B_precision={base_B_precision}"
        )

        err_pct = best_err / gamma if gamma > 0 else 0
        if best_qty_A > 0:
            logger.info(
                f"Precise position calculation for pair {pair_key}: "
                f"Budget=${budget}, "
                f"{pair_A} quantity={best_qty_A}, "
                f"{pair_B} quantity={best_qty_B}, "
                f"Final position A=${final_stake_A}, B=${final_stake_B}, "
                f"Gamma error={err_pct}"
            )

        result = {"A_amount": best_qty_A, "B_amount": best_qty_B}

        return result

    def custom_stake_amount(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_stake: float,
        min_stake: float,
        max_stake: float,
        leverage: float,
        entry_tag: str,
        side: str,
        **kwargs,
    ) -> float:
        logger.info(
            f"----------------------------------------Buy Log Start---------------------------------------"
        )
        logger.info(
            f"Calculating position for pair {pair}, current time: {current_time}, current_rate: {current_rate}"
        )
        if not entry_tag:
            return 0.0
        try:
            pair_key = "_".join(entry_tag.split("_")[2:])
        except IndexError:
            return 0.0

        if (
            pair_key not in self.tradable_pairs
            or pair_key not in self.stake_allocations
        ):
            return 0.0

        # 获取配对状态，如果不存在则为空字典
        state = self.pair_states.get(pair_key)

        if not state:
            budget = self.stake_allocations[pair_key]
            # logger.info(f"计算配对 {pair_key} 的仓位，预算: ${budget:.2f}")
            params = self.tradable_pairs[pair_key]
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]

            dataframe_A = self.dp.get_pair_dataframe(
                pair=pair_A, timeframe=self.timeframe
            )
            dataframe_B = self.dp.get_pair_dataframe(
                pair=pair_B, timeframe=self.timeframe
            )

            if dataframe_A.empty or dataframe_B.empty:
                return 0.0

            pair_A_price = dataframe_A.iloc[-1]["close"]
            pair_B_price = dataframe_B.iloc[-1]["close"]
            logger.info(
                f"A:dataframe open{dataframe_A.iloc[-1]['open']} close:{dataframe_A.iloc[-1]['close']}"
            )
            logger.info(
                f"B:dataframe open{dataframe_B.iloc[-1]['open']} close:{dataframe_B.iloc[-1]['close']}"
            )
            logger.info(f"pair_A_price {pair_A_price}, pair_B_price{pair_B_price}")

            stakes = self._calculate_precise_stakes(
                pair_key, budget, pair_A_price, pair_B_price
            )
            if stakes["A_amount"] <= 0 or stakes["B_amount"] <= 0:
                logger.warning(
                    f"Invalid calculated position for pair {pair_key}: A leg: {stakes['A_amount']}, B leg: {stakes['B_amount']}"
                )
                return 0.0

            self.pair_states[pair_key] = {
                "stakes_calculated": True,
                "A_amount": stakes["A_amount"],
                "B_amount": stakes["B_amount"],
                "leg_A_opened": False,
                "leg_B_opened": False,
                "one_pair_is_already_exit": 0,
                "entry_tag": entry_tag,
            }

            state = self.pair_states.get(pair_key)

        pair_A = self.tradable_pairs[pair_key]["pair_A"]
        stake_to_return = (
            (state["A_amount"] * current_rate)
            if pair == pair_A
            else (state["B_amount"] * current_rate)
        )
        # logger.info(f"Current pair status: {pair_key}: {self.pair_states[pair_key]}")
        logger.info(f"Buy amount: {stake_to_return}")
        return stake_to_return

    def custom_entry_price(
        self,
        pair: str,
        current_time: datetime,
        proposed_rate: float,
        entry_tag: Optional[str],
        **kwargs,
    ) -> float:
        dataframe = self.dp.get_pair_dataframe(pair, timeframe=self.timeframe)
        pair_price = dataframe.iloc[-1]["close"]
        last_time = dataframe.iloc[-1]["date"].to_pydatetime()

        logger.info(
            f"Custom entry price function calculating buy price for pair {pair}, current time: {current_time}, last time: {last_time}, proposed_rate: {proposed_rate}, pair_price: {pair_price}"
        )
        return pair_price

    def custom_exit_price(
        self,
        pair: str,
        trade: Trade,
        current_time: datetime,
        proposed_rate: float,
        current_profit: float,
        exit_tag: str | None,
        **kwargs,
    ) -> float:
        dataframe = self.dp.get_pair_dataframe(pair, timeframe=self.timeframe)
        pair_price = dataframe.iloc[-1]["close"]
        last_time = dataframe.iloc[-1]["date"].to_pydatetime()

        logger.info(
            f"Custom exit price function calculating sell price for pair {pair}, current time: {current_time}, last time: {last_time}, proposed_rate: {proposed_rate}, pair_price: {pair_price}"
        )
        return pair_price

    def confirm_trade_entry(
        self,
        pair: str,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        side: str,
        **kwargs,
    ) -> bool:
        logger.info(
            f"Confirming buy {pair} ({side}), order type: {order_type}, quantity: {amount}, price: {rate}, time in force: {time_in_force}"
        )
        dataframe, _ = self.dp.get_analyzed_dataframe(
            pair=pair, timeframe=self.timeframe
        )
        if dataframe.empty:
            return False

        enter_tag = dataframe["enter_tag"].iloc[-1]
        # logger.info(f"Confirming buy {pair} ({side})")
        try:
            pair_key = "_".join(enter_tag.split("_")[2:])
        except IndexError:
            # logger.warning(f"Unable to parse pair key from entry_tag '{enter_tag}'.")
            return False

        state = self.pair_states.get(pair_key)
        if not state:
            return False
        params = self.tradable_pairs[pair_key]
        pair_A = params["pair_A"]
        pair_B = params["pair_B"]

        if pair == pair_A and state["leg_A_opened"] == False:
            state["leg_A_opened"] = True
            logger.info(
                f"Pair {pair_key} A leg opened successfully, status updated to: {state}"
            )
        if pair == pair_B and state["leg_B_opened"] == False:
            state["leg_B_opened"] = True
            logger.info(
                f"Pair {pair_key} B leg opened successfully, status updated to: {state}"
            )

        logger.info(
            f"Trade {pair} ({side}) opened successfully, current pair status: {state}"
        )
        logger.info(
            f"----------------------------------------Buy Log End---------------------------------------"
        )
        return True

    def confirm_trade_exit(
        self,
        pair: str,
        trade: Trade,
        order_type: str,
        amount: float,
        rate: float,
        time_in_force: str,
        exit_reason: str,
        exit_tag: str = None,
        **kwargs,
    ) -> bool:
        logger.info(
            f"----------------------------------------Sell Log Start---------------------------------------"
        )
        enter_tag = trade.enter_tag
        # logger.info(f"Confirming buy {pair} ({side})")
        try:
            pair_key = "_".join(enter_tag.split("_")[2:])
        except IndexError:
            # logger.warning(f"Unable to parse pair key from entry_tag '{enter_tag}'.")
            return False

        state = self.pair_states.get(pair_key)
        if not state:
            return False
        params = self.tradable_pairs[pair_key]
        pair_A = params["pair_A"]
        pair_B = params["pair_B"]

        if pair == pair_A:
            if (
                state["leg_A_opened"] == True
                and state["leg_B_opened"] == True
                and exit_reason == f"Pair_profit_get_profit_{pair_A}_{pair_B}"
            ):
                state["one_pair_is_already_exit"] = 1
                logger.info(
                    f"-----------------------Triggering one_pair_is_already_exit set to 1---------------------"
                )
            if (
                state["leg_A_opened"] == True
                and state["leg_B_opened"] == False
                and state["one_pair_is_already_exit"] == 1
                and exit_reason == f"Pair_already_exit_{pair_A}_{pair_B}"
            ):
                logger.info(
                    f"-----------------------Triggering one_pair_is_already_exit set to 0---------------------"
                )
                state["one_pair_is_already_exit"] = 0
            state["leg_A_opened"] = False
        if pair == pair_B:
            if (
                state["leg_A_opened"] == True
                and state["leg_B_opened"] == True
                and exit_reason == f"Pair_profit_get_profit_{pair_A}_{pair_B}"
            ):
                state["one_pair_is_already_exit"] = 1
                logger.info(
                    f"-----------------------Triggering one_pair_is_already_exit set to 1---------------------"
                )
            if (
                state["leg_A_opened"] == False
                and state["leg_B_opened"] == True
                and state["one_pair_is_already_exit"] == 1
                and exit_reason == f"Pair_already_exit_{pair_A}_{pair_B}"
            ):
                logger.info(
                    f"-----------------------Triggering one_pair_is_already_exit set to 0---------------------"
                )
                state["one_pair_is_already_exit"] = 0
            state["leg_B_opened"] = False
        logger.info(
            f"Trade {trade.id} ({pair}) closed, exit reason: {exit_reason}, close completed, current pair status: {state}, close quantity: {amount}"
        )
        logger.info(
            f"------------- ---------------------------Sell Log End---------------------------------------"
        )

        if not state["leg_A_opened"] and not state["leg_B_opened"]:
            self.pair_states.pop(pair_key, None)

        return True

    position_adjustment_enable = True

    def adjust_trade_position(
        self,
        trade: Trade,
        current_time: datetime,
        current_rate: float,
        current_profit: float,
        min_stake: float | None,
        max_stake: float,
        current_entry_rate: float,
        current_exit_rate: float,
        current_entry_profit: float,
        current_exit_profit: float,
        **kwargs,
    ) -> float | None | tuple[float | None, str | None]:
        current_pair = trade.pair
        dataframe, _ = self.dp.get_analyzed_dataframe(
            pair=current_pair, timeframe=self.timeframe
        )
        current_candle = dataframe.iloc[-1].squeeze()
        for pair_key, params in self.tradable_pairs.items():
            pair_A = params["pair_A"]
            pair_B = params["pair_B"]
            zscore_col_name = f"{pair_A}_{pair_B}_Zscore"
            # 1. z分数加仓
            last_time = trade.date_last_filled_utc + timedelta(days=3)
            if zscore_col_name in dataframe.columns and (
                current_pair == pair_A or current_pair == pair_B
            ):
                if (
                    abs(current_candle[zscore_col_name])
                    > ((1.5 * trade.nr_of_successful_entries) * self.Zscore_entry)
                    and trade.nr_of_successful_entries < 2
                    and current_time > last_time
                ):
                    entry_tag = current_candle["enter_tag"]
                    pair = current_pair
                    if not entry_tag:
                        return 0.0
                    try:
                        pair_key = "_".join(entry_tag.split("_")[2:])
                    except IndexError:
                        return 0.0

                    if (
                        pair_key not in self.tradable_pairs
                        or pair_key not in self.stake_allocations
                    ):
                        return 0.0

                    # 获取配对状态，如果不存在则为空字典
                    state = self.pair_states.get(pair_key)

                    if not state:
                        A_amount = B_amount = 0.0
                        leg_A_opened = leg_B_opened = False
                        for ot in Trade.get_trades_proxy(is_open=True):
                            if ot.enter_tag == entry_tag:
                                if ot.pair == params["pair_A"]:
                                    A_amount, leg_A_opened = ot.stake_amount, True
                                elif ot.pair == pair_B:
                                    B_amount, leg_B_opened = ot.stake_amount, True

                        if not (leg_A_opened or leg_B_opened):
                            budget = self.stake_allocations[pair_key]
                            # logger.info(f"Calculating position for pair {pair_key}, budget: ${budget:.2f}")
                            params = self.tradable_pairs[pair_key]
                            pair_A = params["pair_A"]
                            pair_B = params["pair_B"]

                            dataframe_A = self.dp.get_pair_dataframe(
                                pair=pair_A, timeframe=self.timeframe
                            )
                            dataframe_B = self.dp.get_pair_dataframe(
                                pair=pair_B, timeframe=self.timeframe
                            )

                            if dataframe_A.empty or dataframe_B.empty:
                                return 0.0

                            pair_A_price = dataframe_A.iloc[-1]["open"]
                            pair_B_price = dataframe_B.iloc[-1]["open"]

                            stakes = self._calculate_precise_stakes(
                                pair_key, budget, pair_A_price, pair_B_price
                            )
                            if stakes["A_amount"] <= 0 or stakes["B_amount"] <= 0:
                                logger.warning(
                                    f"adjust_trade_position Invalid calculated position for pair {pair_key}: A leg: {stakes['A_amount']}, B leg: {stakes['B_amount']}"
                                )
                                return 0.0

                        self.pair_states[pair_key] = {
                            "stakes_calculated": True,
                            "A_amount": A_amount,
                            "B_amount": B_amount,
                            "leg_A_opened": leg_A_opened,
                            "leg_B_opened": leg_B_opened,
                            "one_pair_is_already_exit": 0,
                            "enter_tag": entry_tag,
                        }

                        state = self.pair_states.get(pair_key)

                    pair_A = self.tradable_pairs[pair_key]["pair_A"]
                    stake_to_return = (
                        (state["A_amount"] * current_rate)
                        if pair == pair_A
                        else (state["B_amount"] * current_rate)
                    )
                    # logger.info(f"Current pair status: {pair_key}: {self.pair_states[pair_key]}")
                    logger.info(
                        f"Current currency {pair} pair {pair_key} add position amount: {stake_to_return:.2f} USDT"
                    )
                    return stake_to_return

        return None

    def leverage(
        self,
        pair: str,
        current_time: datetime,
        current_rate: float,
        proposed_leverage: float,
        max_leverage: float,
        entry_tag: str,
        side: str,
        **kwargs,
    ) -> float:
        return self.leverage1

    def zvalue(self, y, x, gamma, c):
        if gamma is None or c is None:
            logger.error("Gamma or c value not set, unable to calculate Z value.")
            return None

        Z = y - (gamma * x + c)
        return Z

    def Yvalue(self, y, x, gamma, c):
        if gamma is None or c is None:
            logger.error("Gamma or c value not set, unable to calculate Y value.")
            return None

        Y = gamma * x + c
        return Y

    def change_y(self, y, x, gamma, c):
        """
        Calculate the percentage of the difference relative to y value
        """
        if gamma is None or c is None:
            logger.error("Gamma or c value not set, unable to calculate Z value.")
            return None

        Y = gamma * x + c
        y_change = (
            (Y - y) / y * 100
        )  # Calculate the percentage of the difference relative to Y
        return y_change

    def get_trade_profit(self, trade):
        if not trade:
            return 0.0
        (dataframe, _) = self.dp.get_analyzed_dataframe(
            pair=trade.pair, timeframe=self.timeframe
        )
        if dataframe.empty:
            return 0.0
        last_candle = dataframe.iloc[-1]
        current_rate = last_candle["close"]
        if trade.is_short:
            profit_ratio = (1 - (current_rate / trade.open_rate)) * trade.leverage
        else:
            profit_ratio = ((current_rate / trade.open_rate) - 1) * trade.leverage
        profit_amount = trade.stake_amount * profit_ratio
        return profit_amount