crypto_quant/core/trade/mean_reversion_sandbox.py

import json
import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
import re
from openpyxl import Workbook
from openpyxl.drawing.image import Image
import openpyxl
from openpyxl.styles import Font
from PIL import Image as PILImage
from config import OKX_MONITOR_CONFIG, MYSQL_CONFIG, WINDOW_SIZE
import core.logger as logging
from core.db.db_merge_market_huge_volume import DBMergeMarketHugeVolume
from core.utils import timestamp_to_datetime, transform_date_time_to_timestamp

# seaborn支持中文
plt.rcParams["font.family"] = ["SimHei"]

logger = logging.logger


class MeanReversionSandbox:
    def __init__(self, solution: str):
        mysql_user = MYSQL_CONFIG.get("user", "xch")
        mysql_password = MYSQL_CONFIG.get("password", "")
        if not mysql_password:
            raise ValueError("MySQL password is not set")
        mysql_host = MYSQL_CONFIG.get("host", "localhost")
        mysql_port = MYSQL_CONFIG.get("port", 3306)
        mysql_database = MYSQL_CONFIG.get("database", "okx")

        self.db_url = f"mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}"
        self.db_merge_market_huge_volume = DBMergeMarketHugeVolume(self.db_url)
        self.peak_valley_data = self.get_peak_valley_data()

        self.solution = solution
        self.save_path = f"./output/trade_sandbox/mean_reversion/{self.solution}/"
        os.makedirs(self.save_path, exist_ok=True)
        self.strategy_description = self.get_startegy_description()

    def get_startegy_description(self):
        desc_dict = {
            "买入": [
                "1. 窗口周期为100, 即100个K线",
                "2. 满足close_10_low为1, 即当前收盘价在窗口周期的10分位以下",
                "3. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量",
                "4. 当前K线为阳线, 即close > open或者K线为一字, 长倒T线, 倒T线, 长十字星, 十字星",
                # "5. 相同symbol的1H当前周期, ma5大于ma10",
                # "5. KDJ, RSI, BOLL任意一个指标出现超卖",
            ],
            "止损": ["跌幅超过下跌周期跌幅中位数, 即down_median后卖出"],
            "止盈": {
                "solution_1": [
                    "高位放量止盈 - 简易版",
                    "1. 当前close_80_high为1或者close_90_high为1",
                    "2. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量",
                ],
                "solution_2": [
                    "高位放量止盈 - 复杂版",
                    "前提条件" "1. 当前close_80_high为1或者close_90_high为1",
                    "2. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量",
                    "以下两个条件, 任一满足即可",
                    "1. K线为阴线, 即close < open",
                    "2. K线为阳线, 即close >= open, 且k_shape满足:",
                    "一字, 长吊锤线, 吊锤线, 长倒T线, 倒T线, 长十字星, 十字星, 长上影线纺锤体, 长下影线纺锤体",
                ],
                "solution_3": [
                    "上涨波段盈利中位数止盈法",
                    "1. 超过波段中位数涨幅, 即up_median/ 到达价位90分位/ 任意技术指标出现中度超卖, 记录当前价格, 继续持仓",
                    "2. 之后一个周期, 如果价格上涨, 则记录该价格继续持仓",
                    "3. 之后一个周期, 如果价格跌到记录价格之下, 则卖出",
                    "4. 如果买入时ma5小于ma10, 过程中ma5大于ma10, 进行记录。之后出现ma5小于ma10, 则卖出",
                ],
            },
        }
        buy_list = desc_dict.get("买入", [])
        stop_loss_list = desc_dict.get("止损", [])
        take_profit_list = desc_dict.get("止盈", {}).get(self.solution, [])
        if len(take_profit_list) == 0:
            self.solution = "solution_1"
            take_profit_list = desc_dict.get("止盈", {}).get(self.solution, [])
        desc = f"策略名称: {self.solution}\n\n"
        buy_desc = "\n".join(buy_list)
        stop_loss_desc = "\n".join(stop_loss_list)
        take_profit_desc = "\n".join(take_profit_list)
        desc += f"买入策略\n {buy_desc}\n\n"
        desc += f"止损策略\n {stop_loss_desc}\n\n"
        desc += f"止盈策略\n {take_profit_desc}\n\n"
        with open(f"{self.save_path}/策略描述.txt", "w", encoding="utf-8") as f:
            f.write(desc)
        return desc

    def get_peak_valley_data(self):
        os.makedirs("./json/", exist_ok=True)
        json_file_path = "./json/peak_valley_data.json"
        if not os.path.exists(json_file_path):
            excel_file_path = "./output/statistics/excel/price_volume_stats_window_size_100_from_20250515000000_to_20250822145000.xlsx"
            if not os.path.exists(excel_file_path):
                raise FileNotFoundError(f"Excel file not found: {excel_file_path}")
            sheet_name = "波峰波谷统计"
            df = pd.read_excel(excel_file_path, sheet_name=sheet_name)
            if df is None or len(df) == 0:
                raise ValueError("Excel file is empty")
            data_list = []
            for index, row in df.iterrows():
                data_list.append(
                    {
                        "symbol": row["symbol"],
                        "bar": row["bar"],
                        "up_mean": row["up_mean"],
                        "up_median": row["up_median"],
                        "down_mean": row["down_mean"],
                        "down_median": row["down_median"],
                    }
                )
            with open(json_file_path, "w", encoding="utf-8") as f:
                json.dump(data_list, f, ensure_ascii=False, indent=4)
            peak_valley_data = pd.DataFrame(data_list)
        else:
            with open(json_file_path, "r", encoding="utf-8") as f:
                peak_valley_data = json.load(f)
        return pd.DataFrame(peak_valley_data)

    def trade_sandbox(
        self, symbol: str, bar: str, window_size: int, start: str, end: str
    ):
        logger.info(f"策略描述: {self.strategy_description}")
        logger.info(
            f"开始获取{symbol} {bar} 的{window_size}分钟窗口大小的数据, 开始时间: {start}, 结束时间: {end}"
        )
        market_data = self.db_merge_market_huge_volume.merge_market_huge_volume(
            symbol, bar, window_size, start, end
        )
        if market_data is None:
            return None
        logger.info(f"数据条数: {len(market_data)}")

        trade_list = []
        trade_pair_dict = {}
        for index, row in market_data.iterrows():
            # check buy condition
            if trade_pair_dict.get("buy_timestamp", None) is None:
                buy_condition = self.check_buy_condition(market_data, row, index, window_size)
            else:
                buy_condition = False
            if buy_condition:
                trade_pair_dict = {}
                trade_pair_dict["solution"] = self.solution
                trade_pair_dict["symbol"] = symbol
                trade_pair_dict["bar"] = bar
                trade_pair_dict["window_size"] = window_size
                trade_pair_dict["buy_timestamp"] = row["timestamp"]
                trade_pair_dict["buy_date_time"] = timestamp_to_datetime(
                    row["timestamp"]
                )
                trade_pair_dict["buy_close"] = row["close"]
                trade_pair_dict["buy_pct_chg"] = row["pct_chg"]
                trade_pair_dict["buy_volume"] = row["volume"]
                trade_pair_dict["buy_huge_volume"] = row["huge_volume"]
                trade_pair_dict["buy_volume_ratio"] = row["volume_ratio"]
                trade_pair_dict["buy_k_shape"] = row["k_shape"]
                trade_pair_dict["buy_close_10_low"] = row["close_10_low"]
                trade_pair_dict["buy_ma5_lt_ma10"] = row["ma5"] < row["ma10"]
                continue

            if trade_pair_dict.get("buy_timestamp", None) is not None:
                sell_condition = False
                # check stop loss condition
                sell_condition = self.check_stop_loss_condition(trade_pair_dict, row)
                sell = sell_condition["sell"]
                if sell:
                    trade_pair_dict["sell_type"] = "止损"
                else:
                    # check take profit condition
                    sell_condition = self.check_take_profit_condition(
                        trade_pair_dict, market_data, row, index
                    )
                    sell = sell_condition["sell"]
                    if sell:
                        trade_pair_dict["sell_type"] = "止盈"

                if sell:
                    trade_pair_dict["sell_reason"] = sell_condition["reason"]
                    trade_pair_dict["sell_timestamp"] = row["timestamp"]
                    trade_pair_dict["sell_date_time"] = timestamp_to_datetime(
                        row["timestamp"]
                    )
                    trade_pair_dict["sell_close"] = row["close"]
                    trade_pair_dict["sell_pct_chg"] = row["pct_chg"]
                    trade_pair_dict["sell_volume"] = row["volume"]
                    trade_pair_dict["sell_huge_volume"] = row["huge_volume"]
                    trade_pair_dict["sell_volume_ratio"] = row["volume_ratio"]
                    trade_pair_dict["sell_k_shape"] = row["k_shape"]
                    trade_pair_dict["sell_close_80_high"] = row["close_80_high"]
                    trade_pair_dict["sell_close_90_high"] = row["close_90_high"]
                    trade_pair_dict["sell_close_10_low"] = row["close_10_low"]
                    trade_pair_dict["sell_close_20_low"] = row["close_20_low"]
                    trade_pair_dict["profit_pct"] = round(
                        (trade_pair_dict["sell_close"] - trade_pair_dict["buy_close"])
                        / trade_pair_dict["buy_close"]
                        * 100,
                        4,
                    )
                    if trade_pair_dict["profit_pct"] <= 0:
                        trade_pair_dict["sell_type"] = "止损"
                    else:
                        trade_pair_dict["sell_type"] = "止盈"
                    if trade_pair_dict.get("last_max_close", None) is not None:
                        # remove last_max_close
                        trade_pair_dict.pop("last_max_close")

                    if trade_pair_dict.get("process_ma5_gt_ma10", None) is not None:
                        trade_pair_dict.pop("process_ma5_gt_ma10")

                    trade_list.append(trade_pair_dict)
                    trade_pair_dict = {}

        if len(trade_list) == 0:
            return None
        trade_data = pd.DataFrame(trade_list)
        trade_data = trade_data[
            [
                "solution",
                "symbol",
                "bar",
                "window_size",
                "sell_type",
                "sell_reason",
                "profit_pct",
                "buy_timestamp",
                "buy_date_time",
                "sell_timestamp",
                "sell_date_time",
                "buy_close",
                "buy_pct_chg",
                "sell_close",
                "sell_pct_chg",
                "buy_volume",
                "buy_huge_volume",
                "buy_volume_ratio",
                "buy_k_shape",
                "buy_close_10_low",
                "buy_ma5_lt_ma10",
                "sell_volume",
                "sell_huge_volume",
                "sell_volume_ratio",
                "sell_k_shape",
                "sell_close_80_high",
                "sell_close_90_high",
                "sell_close_10_low",
                "sell_close_20_low",
            ]
        ]
        trade_data.sort_values(by="buy_timestamp", inplace=True)
        trade_data.reset_index(drop=True, inplace=True)
        return trade_data

    def check_buy_condition(
        self, market_data: pd.DataFrame, row: pd.Series, index: int, window_size: int
    ):
        """
        买入条件
        1. 窗口周期为100, 即100个K线,
        2. 满足close_10_low为1, 即当前收盘价在窗口周期的10分位以下
        3. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量,
        4. (当前K线为阳线, 即close > open)或者K线为一字, 长倒T线, 倒T线, 长十字星, 十字星,
        """
        if index < 2:
            return False
        if row["close"] <= row["open"] and row["k_shape"] not in [
            "一字",
            "长倒T线",
            "倒T线",
            "长十字星",
            "十字星",
        ]:
            return False


        # 满足close_10_low为1, 即当前收盘价在窗口周期的10分位以下
        if row["close_10_low"] != 1:
            return False

        # 如果当前与前两个K线, huge_volume都不为1, 则返回False
        if (
            row["huge_volume"] != 1
            and market_data.loc[index - 1, "huge_volume"] != 1
            and market_data.loc[index - 2, "huge_volume"] != 1
        ):
            return False

        # if not self.check_metrics_over_sell(row):
        #     return False
        # latest_1h_data = self.get_latest_1h_data(row["symbol"], row["date_time"])
        # if latest_1h_data is None or len(latest_1h_data) == 0:
        #     logger.info(f"符合买入条件")
        #     return True
        # # 当前小时周期的ma5小于ma10, 表明空头趋势, 则返回False
        # elif (
        #     not pd.isna(latest_1h_data["ma5"])
        #     and not pd.isna(latest_1h_data["ma10"])
        #     and latest_1h_data["ma5"] < latest_1h_data["ma10"]
        # ):
        #     # logger.info(f"当前小时周期的ma5小于ma10, 空头趋势, 不符合买入条件")
        #     return False
        # else:
        #     logger.info(f"符合买入条件")
        #     return True

        return True

    def get_latest_1h_data(self, symbol: str, current_date_time: str):
        bar = "1H"
        # 根据current_date_time, 获取当前时间往前推1H的日期时间,
        # 如当前时间为2025-08-20 10:20:05, 则获取2025-08-20 09:00:00
        before_date_time = datetime.strptime(current_date_time, "%Y-%m-%d %H:%M:%S")
        before_date_time = before_date_time - timedelta(hours=1)
        # current_date_time取整数小时，如2025-08-20 10:20:05, 取2025-08-20 10:00:00
        before_date_time = before_date_time.replace(minute=0, second=0, microsecond=0)
        before_date_time = before_date_time.strftime("%Y-%m-%d %H:%M:%S")
        end_date_time = datetime.strptime(current_date_time, "%Y-%m-%d %H:%M:%S")
        end_date_time = end_date_time.replace(minute=0, second=0, microsecond=0)
        end_date_time = end_date_time - timedelta(seconds=1)
        end_date_time = end_date_time.strftime("%Y-%m-%d %H:%M:%S")
        latest_1h_data = self.db_merge_market_huge_volume.merge_market_huge_volume(
            symbol, bar, 100, before_date_time, end_date_time
        )
        if latest_1h_data is None or len(latest_1h_data) == 0:
            return None
        # 只获取第一行数据
        latest_1h_data = latest_1h_data.iloc[0]
        return latest_1h_data

    def check_stop_loss_condition(self, trade_pair_dict: dict, row: pd.Series):
        symbol = trade_pair_dict["symbol"]
        bar = trade_pair_dict["bar"]
        # 获取下跌周期跌幅中位数, 为百分比
        down_median = (
            self.peak_valley_data.loc[
                (self.peak_valley_data["symbol"] == symbol)
                & (self.peak_valley_data["bar"] == bar),
                "down_median",
            ].values[0]
            / 100
        )
        buy_close = trade_pair_dict["buy_close"]
        current_close = row["close"]
        result = {"sell": False, "reason": ""}
        if (
            current_close < buy_close
            and (current_close - buy_close) / buy_close < down_median
        ):
            logger.info(f"符合止损条件")
            result["sell"] = True
            result["reason"] = f"亏损超过下跌波段跌幅中位数"
            return result
        result["sell"] = False
        result["reason"] = "未达到止损条件"
        return result

    def check_take_profit_condition(
        self,
        trade_pair_dict: dict,
        market_data: pd.DataFrame,
        row: pd.Series,
        index: int,
    ):
        try:
            if self.solution == "solution_1":
                return self.check_take_profit_condition_solution_1(
                    market_data, row, index
                )
            elif self.solution == "solution_2":
                return self.check_take_profit_condition_solution_2(
                    market_data, row, index
                )
            elif self.solution == "solution_3":
                return self.check_take_profit_condition_solution_3(trade_pair_dict, row)
            else:
                raise ValueError(f"Invalid strategy name: {self.solution}")
        except Exception as e:
            logger.error(f"检查止盈条件时发生错误: {e}")
            return False

    def check_take_profit_condition_solution_1(
        self,
        market_data: pd.DataFrame,
        row: pd.Series,
        index: int,
    ):
        """
        高位放量止盈 - 简易版
        1. 当前close_80_high为1或者close_90_high为1
        2. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量
        """
        result = {"sell": False, "reason": ""}
        if row["close_80_high"] != 1 and row["close_90_high"] != 1:
            result["sell"] = False
            result["reason"] = "未达到止盈条件"
            return result
        if (
            row["huge_volume"] != 1
            and market_data.loc[index - 1, "huge_volume"] != 1
            and market_data.loc[index - 2, "huge_volume"] != 1
        ):
            result["sell"] = False
            result["reason"] = "未达到止盈条件"
            return result
        logger.info(f"符合高位放量止盈 - 简易版条件")
        result["sell"] = True
        result["reason"] = "符合高位放量止盈 - 简易版条件"
        return result

    def check_take_profit_condition_solution_2(
        self,
        market_data: pd.DataFrame,
        row: pd.Series,
        index: int,
    ):
        """
        高位放量止盈 - 复杂版
        前提条件
        1. 当前close_80_high为1或者close_90_high为1
        2. 之前2个K线与当前K线, 存在任意一个K线huge_volume为1, 即存在一个K线是巨量
        以下两个条件, 任一满足即可
        1. K线为阴线, 即close < open
        2. K线为阳线, 即close >= open, 且k_shape满足:
        一字, 长吊锤线, 吊锤线, 长倒T线, 倒T线, 长十字星, 十字星, 长上影线纺锤体, 长下影线纺锤体
        """
        result = {"sell": False, "reason": ""}
        if not self.check_take_profit_condition_solution_1(market_data, row, index):
            result["sell"] = False
            result["reason"] = "未达到止盈条件"
            return result
        if row["close"] < row["open"]:
            logger.info(f"符合高位放量止盈 - 复杂版条件")
            result["sell"] = True
            result["reason"] = "符合高位放量止盈 - 复杂版条件"
            return result
        elif row["k_shape"] in [
            "一字",
            "长吊锤线",
            "吊锤线",
            "长倒T线",
            "倒T线",
            "长十字星",
            "十字星",
            "长上影线纺锤体",
            "长下影线纺锤体",
        ]:
            logger.info(f"符合高位放量止盈 - 复杂版条件")
            result["sell"] = True
            result["reason"] = "符合高位放量止盈 - 复杂版条件"
            return result
        else:
            result["sell"] = False
            result["reason"] = "未达到止盈条件"
            return result

    def check_take_profit_condition_solution_3(
        self, trade_pair_dict: dict, row: pd.Series
    ):
        """
        上涨波段盈利阶段止盈法
        1. 超过波段中位数涨幅, 即up_median/ 到达价位90分位/ 任意技术指标出现中度超卖, 记录当前价格, 继续持仓
        2. 之后一个周期, 如果价格上涨, 则记录该价格继续持仓
        3. 之后一个周期, 如果价格跌到记录价格之下, 则卖出
        4. 如果买入时ma5小于ma10, 过程中ma5大于ma10, 进行记录。之后出现ma5小于ma10, 则卖出
        """
        current_close = row["close"]
        last_max_close = trade_pair_dict.get("last_max_close", None)
        result = {"sell": False, "reason": ""}
        if trade_pair_dict["buy_ma5_lt_ma10"]:
            if trade_pair_dict.get("process_ma5_gt_ma10", None):
                if row["ma5"] < row["ma10"]:
                    logger.info(f"MA5小于MA10发生转势, 卖出")
                    result["sell"] = True
                    result["reason"] = "MA5小于MA10发生转势"
                    return result

            if row["ma5"] > row["ma10"]:
                trade_pair_dict["process_ma5_gt_ma10"] = True
            else:
                trade_pair_dict["process_ma5_gt_ma10"] = False

        if last_max_close is not None:
            if current_close >= last_max_close:
                logger.info(f"价格上涨, 继续持仓")
                trade_pair_dict["last_max_close"] = current_close
                result["sell"] = False
                result["reason"] = "价格上涨, 继续持仓"
                return result
            else:
                logger.info(f"符合上涨波段盈利中位数止盈法条件")
                result["sell"] = True
                result["reason"] = "符合上涨波段盈利中位数止盈条件"
                return result
        else:
            symbol = trade_pair_dict["symbol"]
            bar = trade_pair_dict["bar"]
            up_median = (
                self.peak_valley_data.loc[
                    (self.peak_valley_data["symbol"] == symbol)
                    & (self.peak_valley_data["bar"] == bar),
                    "up_median",
                ].values[0]
                / 100
            )

            need_record = False
            buy_close = trade_pair_dict["buy_close"]
            price_chg = (current_close - buy_close) / buy_close
            if price_chg > up_median:
                logger.info(f"当前价格上涨超过波段中位数涨幅, 记录当前价格")
                need_record = True
            elif self.check_metrics_over_buy(row):
                logger.info(f"技术指标超买, 记录当前价格")
                need_record = True
            elif row["close_90_high"] == 1:
                logger.info(f"到达价位90分位, 记录当前价格")
                need_record = True
            else:
                need_record = False
            if need_record:
                trade_pair_dict["last_max_close"] = current_close
            result["sell"] = False
            result["reason"] = "未达到止盈条件"
            return result

    def check_metrics_over_buy(self, row: pd.Series):
        """
        检查技术指标是否出现中度超买
        KDJ
        K：85.00
        D：80.00
        J：100.00
        说明：K 和 D 进一步上升, J 显著高于100, 表示超买加剧, 回调概率增加, 但可能仍需确认。
        RSI 14
        RSI：80.00
        说明：RSI 进一步上升, 超买程度加深, 市场可能接近短期顶部, 回调概率增加。
        BOLL
        价格位置：价格突破上轨, 偏离上轨约 +2.00%（即价格 = 上轨 × 1.02）
        说明：价格显著突破上轨, 超买程度加深, 可能预示短期回调或反转
        """
        if row["kdj_k"] > 85 and row["kdj_d"] > 80 and row["kdj_j"] > 100:
            logger.info(f"KDJ超买")
            return True
        if row["rsi_14"] > 80:
            logger.info(f"RSI超买")
            return True
        if row["boll_upper"] * 1.02 < row["close"]:
            logger.info(f"BOLL超买")
            return True
        return False

    def check_metrics_over_sell(self, row: pd.Series):
        """
        检查技术指标是否出现超卖
        KDJ
        K: 25.00
        D: 30.00
        J: 20.00

        RSI 14
        RSI:30.00
        说明: RSI 进一步下降, 超卖程度加深, 市场可能接近短期底部, 反弹概率增加。

        BOLL
        价格位置: 价格接近下轨
        """
        if row["kdj_k"] < 25 and row["kdj_d"] < 30 and row["kdj_j"] < 20:
            logger.info(f"KDJ超卖")
            return True
        if row["rsi_14"] < 30:
            logger.info(f"RSI超卖")
            return True
        if row["boll_lower"] >= row["close"]:
            logger.info(f"BOLL超卖")
            return True
        return False