crypto_quant/core/trade/ma_break_statistics.py

import core.logger as logging
import os
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta, timezone
from core.utils import get_current_date_time
import re
import json
import math
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,
    US_STOCK_MONITOR_CONFIG,
    COIN_MYSQL_CONFIG,
    A_MYSQL_CONFIG,
    WINDOW_SIZE,
    BINANCE_MONITOR_CONFIG,
    A_STOCK_MONITOR_CONFIG,
    A_INDEX_MONITOR_CONFIG,
)
from core.biz.metrics_calculation import MetricsCalculation
from core.db.db_market_data import DBMarketData
from core.db.db_astock import DBAStockData
from core.db.db_binance_data import DBBinanceData
from core.db.db_huge_volume_data import DBHugeVolumeData
from core.utils import timestamp_to_datetime, transform_date_time_to_timestamp

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

logger = logging.logger


class MaBreakStatistics:
    """
    统计MA突破之后的涨跌幅
    MA向上突破的点位周期K线：5 > 10 > 20 > 30
    统计MA向上突破的点位周期K线，突破之后，到：
    下一个MA向下突破的点位周期K线：30 > 20 > 10 > 5
    之间的涨跌幅
    """

    def __init__(
        self,
        is_us_stock: bool = False,
        is_astock: bool = False,
        is_aindex: bool = False,
        is_binance: bool = True,
        buy_by_long_period: dict = {
            "by_week": False,
            "by_month": False,
            "buy_by_10_percentile": False,
        },
        long_period_condition: dict = {
            "ma5>ma10": True,
            "ma10>ma20": False,
            "macd_diff>0": True,
            "macd>0": True,
        },
        cut_loss_by_valleys_median: bool = False,
        commission_per_share: float = 0.0008,
    ):
        if is_astock or is_aindex:
            mysql_user = A_MYSQL_CONFIG.get("user", "root")
            mysql_password = A_MYSQL_CONFIG.get("password", "")
            if not mysql_password:
                raise ValueError("MySQL password is not set")
            mysql_host = A_MYSQL_CONFIG.get("host", "localhost")
            mysql_port = A_MYSQL_CONFIG.get("port", 3306)
            mysql_database = A_MYSQL_CONFIG.get("database", "astock")
        else:
            mysql_user = COIN_MYSQL_CONFIG.get("user", "xch")
            mysql_password = COIN_MYSQL_CONFIG.get("password", "")
            if not mysql_password:
                raise ValueError("MySQL password is not set")
            mysql_host = COIN_MYSQL_CONFIG.get("host", "localhost")
            mysql_port = COIN_MYSQL_CONFIG.get("port", 3306)
            mysql_database = COIN_MYSQL_CONFIG.get("database", "okx")

        self.db_url = f"mysql+pymysql://{mysql_user}:{mysql_password}@{mysql_host}:{mysql_port}/{mysql_database}"

        self.db_huge_volume_data = DBHugeVolumeData(self.db_url)
        self.is_us_stock = is_us_stock
        self.is_astock = is_astock
        self.is_aindex = is_aindex
        if self.is_us_stock:
            self.date_time_field = "date_time_us"
        else:
            self.date_time_field = "date_time"
        self.is_binance = is_binance
        if is_us_stock:
            self.symbols = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["QQQ"]
            )
            self.bars = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["5m"]
            )
            self.initial_date = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2014-11-30 00:00:00"
            )
            self.db_market_data = DBMarketData(self.db_url)
        elif is_astock:
            self.symbols = A_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["000001.SH"]
            )
            self.bars = A_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["5m"]
            )
            self.initial_date = A_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2014-11-30 00:00:00"
            )
            self.db_market_data = DBAStockData(self.db_url)
        elif is_aindex:
            self.symbols = A_INDEX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "symbols", ["000001.SH"]
            )
            self.bars = A_INDEX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "bars", ["5m"]
            )
            self.initial_date = A_INDEX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                "initial_date", "2014-11-30 00:00:00"
            )
            self.db_market_data = DBAStockData(self.db_url)
        else:
            if is_binance:
                self.symbols = BINANCE_MONITOR_CONFIG.get("volume_monitor", {}).get(
                    "symbols", ["BTC-USDT"]
                )
                self.bars = ["30m", "1H"]
                self.initial_date = BINANCE_MONITOR_CONFIG.get(
                    "volume_monitor", {}
                ).get("initial_date", "2017-08-16 00:00:00")
                self.db_market_data = DBBinanceData(self.db_url)
            else:
                self.symbols = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                    "symbols", ["XCH-USDT"]
                )
                self.bars = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                    "bars", ["5m", "15m", "30m", "1H"]
                )
                self.initial_date = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get(
                    "initial_date", "2025-05-15 00:00:00"
                )
                self.db_market_data = DBMarketData(self.db_url)
        if len(self.initial_date) > 10:
            self.initial_date = self.initial_date[:10]
        self.end_date = get_current_date_time(format="%Y-%m-%d")
        self.commission_per_share = commission_per_share
        self.trade_strategy_config = self.get_trade_strategy_config()
        self.main_strategy = self.trade_strategy_config.get("均线系统策略", None)
        self.buy_by_long_period = buy_by_long_period
        self.long_period_condition = long_period_condition
        self.cut_loss_by_valleys_median = cut_loss_by_valleys_median
        self.metrics_calculation = MetricsCalculation()

    def get_trade_strategy_config(self):
        with open("./json/trade_strategy.json", "r", encoding="utf-8") as f:
            trade_strategy_config = json.load(f)
        return trade_strategy_config

    def get_by_long_period_desc(self):
        by_week = self.buy_by_long_period.get("by_week", False)
        by_month = self.buy_by_long_period.get("by_month", False)
        by_long_period = ""
        if by_week:
            by_long_period += "1W"
        if by_month:
            by_long_period += "1M"
        if self.buy_by_long_period.get("buy_by_10_percentile", False):
            by_long_period += "_10percentile"
        if by_long_period == "":
            return "no_long_period_judge"
        by_condition = ""
        if self.long_period_condition.get("ma5>ma10", False):
            by_condition += "ma5gtma10"
        if self.long_period_condition.get("ma10>ma20", False):
            by_condition += "_ma10gtma20"
        if self.long_period_condition.get("macd_diff>0", False):
            by_condition += "_macd_diffgt0"
        if self.long_period_condition.get("macd>0", False):
            by_condition += "_macdgt0"
        return by_long_period + "_" + by_condition

    def batch_statistics(self, strategy_name: str = "全均线策略"):
        if self.is_us_stock:
            main_folder = "./output/trade_sandbox/ma_strategy/us_stock/"
            if self.cut_loss_by_valleys_median:
                main_folder += "cut_loss_by_valleys_median/"
            else:
                main_folder += "no_cut_loss_by_valleys_median/"
            self.stats_output_dir = f"{main_folder}excel/{strategy_name}/"
            self.stats_chart_dir = f"{main_folder}chart/{strategy_name}/"
        elif self.is_binance:
            main_folder = "./output/trade_sandbox/ma_strategy/binance/"
            if self.cut_loss_by_valleys_median:
                main_folder += "cut_loss_by_valleys_median/"
            else:
                main_folder += "no_cut_loss_by_valleys_median/"
            self.stats_output_dir = f"{main_folder}excel/{strategy_name}/"
            self.stats_chart_dir = f"{main_folder}chart/{strategy_name}/"
        elif self.is_astock:
            long_period_desc = self.get_by_long_period_desc()
            main_folder = "./output/trade_sandbox/ma_strategy/astock/"
            if self.cut_loss_by_valleys_median:
                main_folder += "cut_loss_by_valleys_median/"
            else:
                main_folder += "no_cut_loss_by_valleys_median/"
            if len(long_period_desc) > 0:
                self.stats_output_dir = (
                    f"{main_folder}{long_period_desc}/excel/{strategy_name}/"
                )
                self.stats_chart_dir = (
                    f"{main_folder}{long_period_desc}/chart/{strategy_name}/"
                )
            else:
                self.stats_output_dir = f"{main_folder}excel/{strategy_name}/"
                self.stats_chart_dir = f"{main_folder}chart/{strategy_name}/"
        elif self.is_aindex:
            main_folder = "./output/trade_sandbox/ma_strategy/aindex/"
            if self.cut_loss_by_valleys_median:
                main_folder += "cut_loss_by_valleys_median/"
            else:
                main_folder += "no_cut_loss_by_valleys_median/"
            long_period_desc = self.get_by_long_period_desc()
            if len(long_period_desc) > 0:
                self.stats_output_dir = (
                    f"{main_folder}{long_period_desc}/excel/{strategy_name}/"
                )
                self.stats_chart_dir = (
                    f"{main_folder}{long_period_desc}/chart/{strategy_name}/"
                )
            else:
                self.stats_output_dir = f"{main_folder}excel/{strategy_name}/"
                self.stats_chart_dir = f"{main_folder}chart/{strategy_name}/"
        else:
            main_folder = "./output/trade_sandbox/ma_strategy/okx/"
            if self.cut_loss_by_valleys_median:
                main_folder += "cut_loss_by_valleys_median/"
            else:
                main_folder += "no_cut_loss_by_valleys_median/"
            self.stats_output_dir = f"{main_folder}excel/{strategy_name}/"
            self.stats_chart_dir = f"{main_folder}chart/{strategy_name}/"
        os.makedirs(self.stats_output_dir, exist_ok=True)
        os.makedirs(self.stats_chart_dir, exist_ok=True)

        ma_break_market_data_list = []
        market_data_pct_chg_list = []
        if strategy_name not in self.main_strategy.keys() or strategy_name is None:
            strategy_name = "全均线策略"
        for symbol in self.symbols:
            for bar in self.bars:
                logger.info(
                    f"开始计算{symbol} {bar}的MA突破区间涨跌幅统计, 策略: {strategy_name}，交易费率：{self.commission_per_share}"
                )
                ma_break_market_data, market_data_pct_chg = self.trade_simulate(
                    symbol, bar, strategy_name
                )
                if (
                    ma_break_market_data is not None
                    and len(ma_break_market_data) > 0
                    and market_data_pct_chg is not None
                ):
                    ma_break_market_data_list.append(ma_break_market_data)
                    logger.info(
                        f"{symbol} {bar} 的市场价格变化, {market_data_pct_chg.get('pct_chg', 0)}%"
                    )
                    market_data_pct_chg_list.append(market_data_pct_chg)

        if len(ma_break_market_data_list) > 0:
            ma_break_market_data = pd.concat(ma_break_market_data_list)
            market_data_pct_chg_df = pd.DataFrame(market_data_pct_chg_list)
            # 依据symbol和bar分组，统计每个symbol和bar的pct_chg的max, min, mean, std, median, count
            ma_break_market_data.sort_values(
                by="begin_timestamp", ascending=True, inplace=True
            )
            ma_break_market_data.reset_index(drop=True, inplace=True)

            account_value_chg_list = []
            for symbol in market_data_pct_chg_df["symbol"].unique():
                for bar in market_data_pct_chg_df["bar"].unique():
                    symbol_bar_data = ma_break_market_data[
                        (ma_break_market_data["symbol"] == symbol)
                        & (ma_break_market_data["bar"] == bar)
                    ].copy()  # 创建副本避免SettingWithCopyWarning
                    if len(symbol_bar_data) > 0:
                        symbol_bar_data.sort_values(
                            by="end_timestamp", ascending=True, inplace=True
                        )
                        symbol_bar_data.reset_index(drop=True, inplace=True)
                        initial_capital = int(
                            market_data_pct_chg_df.loc[
                                (market_data_pct_chg_df["symbol"] == symbol)
                                & (market_data_pct_chg_df["bar"] == bar),
                                "initial_capital",
                            ].values[0]
                        )
                        final_account_value = float(
                            symbol_bar_data["end_account_value"].iloc[-1]
                        )
                        account_value_chg = (
                            (final_account_value - initial_capital)
                            / initial_capital
                            * 100
                        )
                        account_value_chg = round(account_value_chg, 4)
                        market_pct_chg = market_data_pct_chg_df.loc[
                            (market_data_pct_chg_df["symbol"] == symbol)
                            & (market_data_pct_chg_df["bar"] == bar),
                            "pct_chg",
                        ].values[0]
                        total_buy_commission = float(
                            symbol_bar_data["buy_commission"].sum()
                        )
                        total_sell_commission = float(
                            symbol_bar_data["sell_commission"].sum()
                        )
                        total_commission = total_buy_commission + total_sell_commission
                        total_commission = round(total_commission, 4)
                        total_buy_commission = round(total_buy_commission, 4)
                        total_sell_commission = round(total_sell_commission, 4)
                        symbol_name = str(symbol_bar_data["symbol_name"].iloc[0])
                        account_value_chg_list.append(
                            {
                                "strategy_name": strategy_name,
                                "symbol": symbol,
                                "symbol_name": symbol_name,
                                "bar": bar,
                                "total_buy_commission": total_buy_commission,
                                "total_sell_commission": total_sell_commission,
                                "total_commission": total_commission,
                                "initial_account_value": initial_capital,
                                "final_account_value": final_account_value,
                                "account_value_chg": account_value_chg,
                                "market_pct_chg": market_pct_chg,
                            }
                        )
            account_value_chg_df = pd.DataFrame(account_value_chg_list)
            account_value_chg_df = account_value_chg_df[
                [
                    "strategy_name",
                    "symbol",
                    "symbol_name",
                    "bar",
                    "total_buy_commission",
                    "total_sell_commission",
                    "total_commission",
                    "initial_account_value",
                    "final_account_value",
                    "account_value_chg",
                    "market_pct_chg",
                ]
            ]

            account_value_statistics_df = (
                ma_break_market_data.groupby(["symbol", "symbol_name", "bar"])[
                    "end_account_value"
                ]
                .agg(
                    account_value_max="max",
                    account_value_min="min",
                    account_value_mean="mean",
                    account_value_std="std",
                    account_value_median="median",
                    account_value_count="count",
                )
                .reset_index()
            )
            account_value_statistics_df["strategy_name"] = strategy_name
            account_value_statistics_df = account_value_statistics_df[
                [
                    "strategy_name",
                    "symbol",
                    "symbol_name",
                    "bar",
                    "account_value_max",
                    "account_value_min",
                    "account_value_mean",
                    "account_value_std",
                    "account_value_median",
                    "account_value_count",
                ]
            ]

            # 依据symbol和bar分组，统计每个symbol和bar的interval_minutes的max, min, mean, std, median, count
            interval_minutes_df = (
                ma_break_market_data.groupby(["symbol", "symbol_name", "bar"])[
                    "interval_minutes"
                ]
                .agg(
                    interval_minutes_max="max",
                    interval_minutes_min="min",
                    interval_minutes_mean="mean",
                    interval_minutes_std="std",
                    interval_minutes_median="median",
                    interval_minutes_count="count",
                )
                .reset_index()
            )
            interval_minutes_df["strategy_name"] = strategy_name
            interval_minutes_df = interval_minutes_df[
                [
                    "strategy_name",
                    "symbol",
                    "symbol_name",
                    "bar",
                    "interval_minutes_max",
                    "interval_minutes_min",
                    "interval_minutes_mean",
                    "interval_minutes_std",
                    "interval_minutes_median",
                    "interval_minutes_count",
                ]
            ]

            earliest_market_date_time = ma_break_market_data["begin_date_time"].min()
            earliest_market_date_time = re.sub(
                r"[\:\-\s]", "", str(earliest_market_date_time)
            )
            latest_market_date_time = ma_break_market_data["end_date_time"].max()
            if latest_market_date_time is None:
                latest_market_date_time = get_current_date_time(format="%Y%m%d%H%M%S")
            latest_market_date_time = re.sub(
                r"[\:\-\s]", "", str(latest_market_date_time)
            )
            if self.commission_per_share > 0:
                output_file_name = f"ma_break_stats_from_{earliest_market_date_time}_to_{latest_market_date_time}_{strategy_name}_with_commission.xlsx"
            else:
                output_file_name = f"ma_break_stats_from_{earliest_market_date_time}_to_{latest_market_date_time}_{strategy_name}_without_commission.xlsx"
            output_file_path = os.path.join(self.stats_output_dir, output_file_name)
            logger.info(f"导出{output_file_path}")
            strategy_info_df = self.get_strategy_info(strategy_name)
            with pd.ExcelWriter(output_file_path) as writer:
                strategy_info_df.to_excel(writer, sheet_name="策略信息", index=False)
                ma_break_market_data.to_excel(
                    writer, sheet_name="买卖记录明细", index=False
                )
                account_value_chg_df.to_excel(
                    writer, sheet_name="资产价值变化", index=False
                )
                account_value_statistics_df.to_excel(
                    writer, sheet_name="买卖账户价值统计", index=False
                )
                interval_minutes_df.to_excel(
                    writer, sheet_name="买卖时间间隔统计", index=False
                )

            chart_dict = self.draw_quant_pct_chg_bar_chart(
                account_value_chg_df, strategy_name
            )
            self.output_chart_to_excel(output_file_path, chart_dict)
            chart_dict = self.draw_quant_line_chart(
                ma_break_market_data, market_data_pct_chg_df, strategy_name
            )
            self.output_chart_to_excel(output_file_path, chart_dict)
            return account_value_chg_df
        else:
            return None

    def get_strategy_info(self, strategy_name: str = "全均线策略"):
        strategy_config = self.main_strategy.get(strategy_name, None)
        if strategy_config is None:
            logger.error(f"策略{strategy_name}不存在")
            return None
        strategy_info = {"策略名称": strategy_name, "买入策略": "", "卖出策略": ""}
        buy_dict = strategy_config.get("buy", {})
        buy_and_list = buy_dict.get("and", [])
        buy_or_list = buy_dict.get("or", [])
        buy_and_text = ""
        buy_or_text = ""
        for and_condition in buy_and_list:
            buy_and_text += f"{and_condition}, \n"
        if len(buy_or_list) > 0:
            for or_condition in buy_or_list:
                buy_or_text += f"{or_condition}, \n"
        if len(buy_or_text) > 0:
            strategy_info["买入策略"] = buy_and_text + " 或者 \n" + buy_or_text
        else:
            strategy_info["买入策略"] = buy_and_text

        # 假如根据长周期判断买入，则需要设置长周期策略
        by_week = self.buy_by_long_period.get("by_week", False)
        by_month = self.buy_by_long_period.get("by_month", False)
        if by_week:
            strategy_info["买入策略"] += "根据周线指标，\n"
        if by_month:
            strategy_info["买入策略"] += "根据月线指标，\n"
        if self.long_period_condition.get("ma5>ma10", False):
            strategy_info["买入策略"] += "ma5>ma10, \n"
        if self.long_period_condition.get("ma10>ma20", False):
            strategy_info["买入策略"] += "ma10>ma20, \n"
        if self.long_period_condition.get("macd_diff>0", False):
            strategy_info["买入策略"] += "macd_diff>0, \n"
        if self.long_period_condition.get("macd>0", False):
            strategy_info["买入策略"] += "macd>0, \n"
        strategy_info["买入策略"] = strategy_info["买入策略"].strip()
        sell_dict = strategy_config.get("sell", {})
        sell_and_list = sell_dict.get("and", [])
        sell_or_list = sell_dict.get("or", [])
        sell_and_text = ""
        sell_or_text = ""
        for and_condition in sell_and_list:
            sell_and_text += f"{and_condition}, \n"
        if len(sell_or_list) > 0:
            for or_condition in sell_or_list:
                sell_or_text += f"{or_condition}, \n"
        if len(sell_or_text) > 0:
            strategy_info["卖出策略"] = sell_and_text + " 或者 \n" + sell_or_text
        else:
            strategy_info["卖出策略"] = sell_and_text
        strategy_info["卖出策略"] = strategy_info["卖出策略"].strip()
        # 将strategy_info转换为pd.DataFrame
        strategy_info_df = pd.DataFrame([strategy_info])
        return strategy_info_df

    def trade_simulate(self, symbol: str, bar: str, strategy_name: str = "全均线策略"):
        market_data = self.get_full_data(symbol, bar)
        if market_data is None or len(market_data) == 0:
            logger.warning(f"获取{symbol} {bar} 数据失败")
            return None, None
        else:
            logger.info(f"获取{symbol} {bar} 数据成功，数据条数: {len(market_data)}")
            # 获得ma5, ma10, ma20, ma30不为空的行
            market_data = market_data[
                (market_data["ma5"].notna())
                & (market_data["ma10"].notna())
                & (market_data["ma20"].notna())
                & (market_data["ma30"].notna())
            ]
            logger.info(
                f"ma5, ma10, ma20, ma30不为空的行，数据条数: {len(market_data)}"
            )
            # 计算volume_ma5
            market_data["volume_ma5"] = market_data["volume"].rolling(window=5).mean()

            market_data["volume_pct_chg"] = (
                market_data["volume"] - market_data["volume_ma5"]
            ) / market_data["volume_ma5"]
            market_data["volume_pct_chg"] = market_data["volume_pct_chg"].fillna(0)

            # 按照timestamp排序
            market_data = market_data.sort_values(by="timestamp", ascending=True)
            # 获得ma_break_market_data的close列
            market_data.reset_index(drop=True, inplace=True)
            ma_break_market_data_pair_list = []
            ma_break_market_data_pair = {}
            close_mean = market_data["close"].mean()
            self.update_initial_capital(close_mean)
            logger.info(
                f"成功获取{symbol}数据：{len(market_data)}根{bar}K线,开始日期={market_data[self.date_time_field].min()}，结束日期={market_data[self.date_time_field].max()}"
            )

            account_value = self.initial_capital

            for index, row in market_data.iterrows():
                if self.is_astock:
                    symbol_name = row["symbol_name"]
                elif self.is_aindex:
                    symbol_name = row["symbol_name"]
                else:
                    symbol_name = row["symbol"]
                ma_cross = row["ma_cross"]
                timestamp = row["timestamp"]
                date_time = row[self.date_time_field]
                close = row["close"]
                ma5 = row["ma5"]
                ma10 = row["ma10"]
                ma20 = row["ma20"]
                ma30 = row["ma30"]
                macd_diff = float(row["dif"])
                macd_dea = float(row["dea"])
                macd = float(row["macd"])
                if ma_break_market_data_pair.get("begin_timestamp", None) is None:
                    buy_condition = self.fit_strategy(
                        strategy_name=strategy_name,
                        row=row,
                        behavior="buy",
                        buy_price=None,
                        window_100_valleys_median=None,
                    )

                    if buy_condition:
                        entry_price = close
                        # 计算交易股数
                        shares, account_value = self.calculate_shares(
                            account_value, entry_price
                        )
                        if shares == 0:
                            # 股数为0→不交易
                            continue

                        # 计算佣金
                        buy_commission = shares * close * self.commission_per_share

                        ma_break_market_data_pair = {}
                        ma_break_market_data_pair["symbol"] = symbol
                        ma_break_market_data_pair["symbol_name"] = symbol_name
                        ma_break_market_data_pair["bar"] = bar
                        ma_break_market_data_pair["begin_timestamp"] = timestamp
                        ma_break_market_data_pair["begin_date_time"] = date_time
                        ma_break_market_data_pair["begin_close"] = close
                        ma_break_market_data_pair["begin_ma5"] = ma5
                        ma_break_market_data_pair["begin_ma10"] = ma10
                        ma_break_market_data_pair["begin_ma20"] = ma20
                        ma_break_market_data_pair["begin_ma30"] = ma30
                        ma_break_market_data_pair["begin_macd_diff"] = macd_diff
                        ma_break_market_data_pair["begin_macd_dea"] = macd_dea
                        ma_break_market_data_pair["begin_macd"] = macd
                        ma_break_market_data_pair["shares"] = shares
                        ma_break_market_data_pair["buy_commission"] = buy_commission
                        ma_break_market_data_pair["begin_account_value"] = account_value
                        continue
                else:
                    valleys_median = None
                    if self.cut_loss_by_valleys_median and index >= 100:
                        window_100_records = market_data.iloc[index - 100 : index]
                        peaks_valleys = self.metrics_calculation.get_peaks_valleys_mean(
                            window_100_records
                        )
                        valleys_median = peaks_valleys.get("valleys_median", None)
                        if valleys_median is not None and valleys_median > 0:
                            valleys_median = valleys_median / 100
                    sell_condition = self.fit_strategy(
                        strategy_name=strategy_name,
                        row=row,
                        behavior="sell",
                        buy_price=ma_break_market_data_pair["begin_close"],
                        window_100_valleys_median=valleys_median,
                    )

                    if sell_condition or index == len(market_data) - 1:
                        # 达到卖出条件或者最后一条数据，则卖出
                        shares = ma_break_market_data_pair["shares"]
                        entry_price = ma_break_market_data_pair["begin_close"]
                        exit_price = close
                        sell_commission = (
                            shares * exit_price * self.commission_per_share
                        )
                        profit_loss = (exit_price - entry_price) * shares
                        begin_account_value = ma_break_market_data_pair[
                            "begin_account_value"
                        ]
                        account_value = (
                            begin_account_value + profit_loss - sell_commission
                        )

                        ma_break_market_data_pair["end_timestamp"] = timestamp
                        ma_break_market_data_pair["end_date_time"] = date_time
                        ma_break_market_data_pair["end_close"] = close
                        ma_break_market_data_pair["end_ma5"] = ma5
                        ma_break_market_data_pair["end_ma10"] = ma10
                        ma_break_market_data_pair["end_ma20"] = ma20
                        ma_break_market_data_pair["end_ma30"] = ma30
                        ma_break_market_data_pair["end_macd_diff"] = macd_diff
                        ma_break_market_data_pair["end_macd_dea"] = macd_dea
                        ma_break_market_data_pair["end_macd"] = macd
                        ma_break_market_data_pair["pct_chg"] = (
                            exit_price - entry_price
                        ) / entry_price
                        ma_break_market_data_pair["pct_chg"] = round(
                            ma_break_market_data_pair["pct_chg"] * 100, 4
                        )
                        if valleys_median is not None:
                            ma_break_market_data_pair["valleys_median"] = (
                                valleys_median * 100
                            )
                        else:
                            ma_break_market_data_pair["valleys_median"] = None
                        ma_break_market_data_pair["profit_loss"] = profit_loss
                        ma_break_market_data_pair["sell_commission"] = sell_commission
                        ma_break_market_data_pair["end_account_value"] = account_value
                        ma_break_market_data_pair["interval_seconds"] = (
                            timestamp - ma_break_market_data_pair["begin_timestamp"]
                        ) / 1000
                        # 将interval转换为分钟
                        ma_break_market_data_pair["interval_minutes"] = (
                            ma_break_market_data_pair["interval_seconds"] / 60
                        )
                        ma_break_market_data_pair["interval_hours"] = (
                            ma_break_market_data_pair["interval_seconds"] / 3600
                        )
                        ma_break_market_data_pair["interval_days"] = (
                            ma_break_market_data_pair["interval_seconds"] / 86400
                        )
                        ma_break_market_data_pair_list.append(ma_break_market_data_pair)
                        ma_break_market_data_pair = {}

            if len(ma_break_market_data_pair_list) > 0:
                ma_break_market_data = pd.DataFrame(ma_break_market_data_pair_list)
                # sort by end_timestamp
                ma_break_market_data.sort_values(
                    by="begin_timestamp", ascending=True, inplace=True
                )
                ma_break_market_data.reset_index(drop=True, inplace=True)
                logger.info(
                    f"获取{symbol} {bar} 的买卖记录明细成功, 买卖次数: {len(ma_break_market_data)}"
                )
                # 量化期间，市场的波动率：
                # ma_break_market_data(最后一条数据的end_close - 第一条数据的begin_close) / 第一条数据的begin_close * 100
                pct_chg = (
                    (
                        ma_break_market_data["end_close"].iloc[-1]
                        - ma_break_market_data["begin_close"].iloc[0]
                    )
                    / ma_break_market_data["begin_close"].iloc[0]
                    * 100
                )
                pct_chg = round(pct_chg, 4)
                symbol_name = ma_break_market_data["symbol_name"].iloc[0]
                market_data_pct_chg = {
                    "symbol": symbol,
                    "symbol_name": symbol_name,
                    "bar": bar,
                    "pct_chg": pct_chg,
                    "initial_capital": self.initial_capital,
                }
                return ma_break_market_data, market_data_pct_chg
            else:
                return None, None

    def update_initial_capital(self, close_mean: float):
        self.initial_capital = 25000
        if close_mean > 10000:
            self.initial_capital = self.initial_capital * 10000
        elif close_mean > 5000:
            self.initial_capital = self.initial_capital * 5000
        elif close_mean > 1000:
            self.initial_capital = self.initial_capital * 1000
        elif close_mean > 500:
            self.initial_capital = self.initial_capital * 500
        elif close_mean > 100:
            self.initial_capital = self.initial_capital * 100
        else:
            pass
        logger.info(f"收盘价均值：{close_mean}")
        logger.info(f"初始资金调整为：{self.initial_capital}")

    def calculate_shares(self, account_value, entry_price):
        """
        根据ORB公式计算交易股数
        :param account_value: 当前账户价值（美元）
        :param entry_price: 交易买入价格
        :param commission_per_share: 交易佣金, 默认为0.0008
        :return: 整数股数（Shares）
        """
        logger.info(
            f"开始计算交易股数：账户价值={account_value}，买入价格={entry_price}"
        )
        try:
            # 验证输入参数
            if account_value <= 0 or entry_price <= 0 or self.commission_per_share < 0:
                logger.error("账户价值、买入价格或佣金不能为负或零")
                return 0, account_value  # 返回0股，账户价值不变

            # 计算考虑手续费后的每单位BTC总成本
            total_cost_per_share = entry_price * (1 + self.commission_per_share)

            # 计算可购买的BTC数量（向下取整）
            shares = math.floor(account_value / total_cost_per_share)

            # 计算总成本（含手续费）
            total_cost = shares * total_cost_per_share

            # 计算剩余现金
            remaining_cash = account_value - total_cost

            # 计算总资产价值 = (购买的BTC数量 × 买入价格) + 剩余现金
            remaining_value = (shares * entry_price) + remaining_cash

            # 记录计算结果
            logger.info(
                f"计算结果：可购买股数={shares}，总成本={total_cost:.2f}美元，"
                f"剩余现金={remaining_cash:.2f}美元，总资产价值={remaining_value:.2f}美元"
            )

            return shares, remaining_value

        except Exception as e:
            logger.error(f"计算股数或账户价值时出错：{str(e)}")
            return 0, account_value  # 出错时返回0股，账户价值不变

    def get_full_data(self, symbol: str, bar: str = "5m"):
        """
        分段获取数据，并将数据合并为完整数据
        分段依据：如果end_date与start_date相差超过一年，则每次取一年数据
        """
        data = pd.DataFrame()
        start_date = datetime.strptime(self.initial_date, "%Y-%m-%d")
        end_date = datetime.strptime(self.end_date, "%Y-%m-%d") + timedelta(days=1)
        table_name = ""
        if self.is_astock:
            if bar == "1D":
                table_name = "stock_daily_price_from_2021"
            elif bar == "1W":
                table_name = "stock_weekly_price_from_2020"
            elif bar == "1M":
                table_name = "stock_monthly_price_from_2015"
        elif self.is_aindex:
            if bar == "1D":
                table_name = "index_daily_price_from_2021"
            elif bar == "1W":
                table_name = "index_weekly_price_from_2020"
            elif bar == "1M":
                table_name = "index_monthly_price_from_2015"
        elif self.is_us_stock:
            table_name = "crypto_market_data"
        elif self.is_binance:
            table_name = "crypto_binance_data"
        else:
            table_name = "crypto_binance_data"

        if self.is_astock or self.is_aindex:
            fields = [
                "a.ts_code as symbol",
                "b.name as symbol_name",
                f"'{bar}' as bar",
                "0 as timestamp",
                "trade_date as date_time",
                "open",
                "high",
                "low",
                "close",
                "vol as volume",
                "MA5 as ma5",
                "MA10 as ma10",
                "MA20 as ma20",
                "MA30 as ma30",
                "均线交叉 as ma_cross",
                "DIF as dif",
                "DEA as dea",
                "MACD as macd",
            ]
        else:
            fields = [
                "symbol",
                "bar",
                "timestamp",
                "date_time",
                "date_time_us",
                "open",
                "high",
                "low",
                "close",
                "volume",
                "sar_signal",
                "ma5",
                "ma10",
                "ma20",
                "ma30",
                "ma_cross",
                "dif",
                "dea",
                "macd",
            ]
        while start_date < end_date:
            current_end_date = min(start_date + timedelta(days=180), end_date)
            start_date_str = start_date.strftime("%Y-%m-%d")
            current_end_date_str = current_end_date.strftime("%Y-%m-%d")
            logger.info(f"获取{symbol}数据：{start_date_str}至{current_end_date_str}")
            current_data = self.db_market_data.query_market_data_by_symbol_bar(
                symbol,
                fields,
                start=start_date_str,
                end=current_end_date_str,
                table_name=table_name,
            )
            if current_data is not None and len(current_data) > 0:
                current_data = pd.DataFrame(current_data)
                data = pd.concat([data, current_data])
            start_date = current_end_date
        data.drop_duplicates(inplace=True)
        data.sort_values(by=self.date_time_field, inplace=True)
        data.reset_index(drop=True, inplace=True)
        if self.is_astock or self.is_aindex:
            data = self.update_data(data)
        return data

    def get_long_period_data(self, symbol: str, bar: str, end_date: str):
        """
        获取长周期数据
        :param data: 数据
        :return: 长周期数据
        """
        if not (self.is_astock or self.is_aindex):
            return None
        table_name = ""
        if self.is_astock:
            if bar == "1M":
                table_name = "stock_monthly_price_from_2015"
            elif bar == "1W":
                table_name = "stock_weekly_price_from_2020"
            else:
                pass
        elif self.is_aindex:
            if bar == "1M":
                table_name = "index_monthly_price_from_2015"
            elif bar == "1W":
                table_name = "index_weekly_price_from_2020"
            else:
                pass
        if len(end_date) != 10:
            end_date = self.change_date_format(end_date)
        if bar == "1M":
            # 获取上五年的日期
            last_date = datetime.strptime(end_date, "%Y-%m-%d") - timedelta(
                days=360 * 5
            )
            last_date = last_date.strftime("%Y-%m-%d")
        elif bar == "1W":
            # 获取上两年的日期
            last_date = datetime.strptime(end_date, "%Y-%m-%d") - timedelta(
                days=360 * 2
            )
            last_date = last_date.strftime("%Y-%m-%d")
        else:
            last_date = None

        if len(table_name) == 0 or last_date is None:
            return None
        fields = [
            "a.ts_code as symbol",
            "b.name as symbol_name",
            f"'{bar}' as bar",
            "0 as timestamp",
            "trade_date as date_time",
            "open",
            "high",
            "low",
            "close",
            "vol as volume",
            "MA5 as ma5",
            "MA10 as ma10",
            "MA20 as ma20",
            "MA30 as ma30",
            "均线交叉 as ma_cross",
            "DIF as dif",
            "DEA as dea",
            "MACD as macd",
        ]
        data = self.db_market_data.query_market_data_by_symbol_bar(
            symbol, fields, start=last_date, end=end_date, table_name=table_name
        )
        if data is not None and len(data) > 0:
            data = pd.DataFrame(data)
            data.sort_values(by="date_time", inplace=True)
            data = self.metrics_calculation.calculate_percentile_indicators(
                data=data,
                window_size=50,
                price_column="close",
                percentiles=[(0.1, "10")],
            )
            latest_row = data.iloc[-1]
            if (
                latest_row["ma5"] is None
                or latest_row["ma10"] is None
                or latest_row["ma20"] is None
                or latest_row["dif"] is None
                or latest_row["macd"] is None
            ):
                return None
            return latest_row
        else:
            return None

    def update_data(self, data: pd.DataFrame):
        """
        更新数据
        1. 将date_time列中的20210104这种格式，替换为2021-01-04的格式
        2. 将date_time转换为timestamp，并更新timestamp列
        3. 通过MetricsCalculation的ma5102030方法更新ma_cross列
        :param data: 数据
        :return: 更新后的数据
        """
        data["date_time"] = data["date_time"].apply(
            lambda x: self.change_date_format(x)
        )
        data["timestamp"] = data["date_time"].apply(
            lambda x: transform_date_time_to_timestamp(x)
        )
        data = self.metrics_calculation.ma5102030(data)
        return data

    def change_date_format(self, date_text: str):
        # 将20210104这种格式，替换为2021-01-04的格式
        if len(date_text) == 8:
            return date_text[0:4] + "-" + date_text[4:6] + "-" + date_text[6:8]
        else:
            return date_text

    def fit_strategy(
        self,
        strategy_name: str = "全均线策略",
        row: pd.Series = None,
        behavior: str = "buy",
        buy_price: float = None,
        window_100_valleys_median: float = None,
    ):
        # 如果行为是卖出，则判断是否根据止损价格卖出
        # 止损价格 = 买入价格 * (1 - window_100_valleys_median)
        # window_100_valleys_median为100日下跌波谷幅度中位数
        # 当前价格 < 止损价格，则卖出
        if (
            behavior == "sell"
            and buy_price is not None
            and window_100_valleys_median is not None
        ):
            current_price = float(row["close"])
            if current_price < buy_price:
                loss_ratio = (buy_price - current_price) / buy_price
                if loss_ratio > window_100_valleys_median:
                    return True
        strategy_config = self.main_strategy.get(strategy_name, None)
        if strategy_config is None:
            logger.error(f"策略{strategy_name}不存在")
            return False
        condition_dict = strategy_config.get(behavior, None)
        if condition_dict is None:
            logger.error(f"策略{strategy_name}的{behavior}条件不存在")
            return False

        and_list = condition_dict.get("and", [])

        condition = True
        condition = self.get_judge_result(row, and_list, "and", condition)
        or_list = condition_dict.get("or", [])
        condition = self.get_judge_result(row, or_list, "or", condition)

        if behavior == "buy" and condition:
            # 如果满足条件，则判断是否根据长周期指标买入
            bar = row["bar"]
            if (self.is_astock or self.is_aindex) and bar == "1D":
                date_time = row["date_time"]
                long_period_condition_list = []
                if self.long_period_condition.get("ma5>ma10", False):
                    long_period_condition_list.append("ma5>ma10")
                if self.long_period_condition.get("ma10>ma20", False):
                    long_period_condition_list.append("ma10>ma20")
                if self.long_period_condition.get("macd_diff>0", False):
                    long_period_condition_list.append("macd_diff>0")
                if self.long_period_condition.get("macd>0", False):
                    long_period_condition_list.append("macd>0")
                if len(long_period_condition_list) > 0:
                    if self.buy_by_long_period.get("by_week", False):
                        long_period_data = self.get_long_period_data(
                            row["symbol"], "1W", date_time
                        )
                        if long_period_data is not None:
                            condition = self.get_judge_result(
                                long_period_data,
                                long_period_condition_list,
                                "and",
                                condition,
                            )
                            if not condition:
                                # 如果周线处于空头条件，但收盘价位于50窗口的低点10分位数，则买入
                                if self.buy_by_long_period.get("buy_by_10_percentile", False):
                                    if long_period_data["close_10_low"] == 1:
                                        condition = True
                                if not condition:
                                    logger.info(
                                        f"根据周线指标，{row['symbol']}不满足买入条件"
                                    )
                    if self.buy_by_long_period.get("by_month", False):
                        long_period_data = self.get_long_period_data(
                            row["symbol"], "1M", date_time
                        )
                        if long_period_data is not None:
                            condition = self.get_judge_result(
                                long_period_data,
                                long_period_condition_list,
                                "and",
                                condition,
                            )
                            
                            if not condition:
                                # 如果月线处于空头条件，但收盘价位于50窗口的低点10分位数，则买入
                                if self.buy_by_long_period.get("buy_by_10_percentile", False):
                                    if long_period_data["close_10_low"] == 1:
                                        condition = True
                                if not condition:
                                    logger.info(
                                        f"根据月线指标，{row['symbol']}不满足买入条件"
                                    )

        return condition

    def get_judge_result(
        self,
        row: pd.Series,
        condition_list: list,
        and_or: str = "and",
        raw_condition: bool = True,
    ):
        ma_cross = row["ma_cross"]
        if pd.isna(ma_cross) or ma_cross is None:
            ma_cross = ""
        ma_cross = str(ma_cross)
        ma5 = float(row["ma5"])
        ma10 = float(row["ma10"])
        ma20 = float(row["ma20"])
        ma30 = float(row["ma30"])
        close = float(row["close"])
        if "volume_pct_chg" in list(row.index) and row["volume_pct_chg"] is not None:
            volume_pct_chg = float(row["volume_pct_chg"])
        else:
            volume_pct_chg = None
        macd_diff = float(row["dif"])
        macd_dea = float(row["dea"])
        macd = float(row["macd"])
        if and_or == "and":
            for and_condition in condition_list:
                if and_condition == "5上穿10":
                    raw_condition = raw_condition and ("5上穿10" in ma_cross)
                elif and_condition == "10上穿20":
                    raw_condition = raw_condition and ("10上穿20" in ma_cross)
                elif and_condition == "20上穿30":
                    raw_condition = raw_condition and ("20上穿30" in ma_cross)
                elif and_condition == "ma5>ma10":
                    raw_condition = raw_condition and (ma5 > ma10)
                elif and_condition == "ma10>ma20":
                    raw_condition = raw_condition and (ma10 > ma20)
                elif and_condition == "ma20>ma30":
                    raw_condition = raw_condition and (ma20 > ma30)
                elif and_condition == "close>ma20":
                    raw_condition = raw_condition and (close > ma20)
                elif (
                    and_condition == "volume_pct_chg>0.2" and volume_pct_chg is not None
                ):
                    raw_condition = raw_condition and (volume_pct_chg > 0.2)
                elif and_condition == "macd_diff>0":
                    raw_condition = raw_condition and (macd_diff > 0)
                elif and_condition == "macd_dea>0":
                    raw_condition = raw_condition and (macd_dea > 0)
                elif and_condition == "macd>0":
                    raw_condition = raw_condition and (macd > 0)
                elif and_condition == "10下穿5":
                    raw_condition = raw_condition and ("10下穿5" in ma_cross)
                elif and_condition == "20下穿10":
                    raw_condition = raw_condition and ("20下穿10" in ma_cross)
                elif and_condition == "30下穿20":
                    raw_condition = raw_condition and ("30下穿20" in ma_cross)
                elif and_condition == "ma5<ma10":
                    raw_condition = raw_condition and (ma5 < ma10)
                elif and_condition == "ma10<ma20":
                    raw_condition = raw_condition and (ma10 < ma20)
                elif and_condition == "ma20<ma30":
                    raw_condition = raw_condition and (ma20 < ma30)
                elif and_condition == "close<ma20":
                    raw_condition = raw_condition and (close < ma20)
                elif and_condition == "macd_diff<0":
                    raw_condition = raw_condition and (macd_diff < 0)
                elif and_condition == "macd_dea<0":
                    raw_condition = raw_condition and (macd_dea < 0)
                elif and_condition == "macd<0":
                    raw_condition = raw_condition and (macd < 0)
                else:
                    pass
        elif and_or == "or":
            for or_condition in condition_list:
                if or_condition == "5上穿10":
                    raw_condition = raw_condition or ("5上穿10" in ma_cross)
                elif or_condition == "10上穿20":
                    raw_condition = raw_condition or ("10上穿20" in ma_cross)
                elif or_condition == "20上穿30":
                    raw_condition = raw_condition or ("20上穿30" in ma_cross)
                elif or_condition == "ma5>ma10":
                    raw_condition = raw_condition or (ma5 > ma10)
                elif or_condition == "ma10>ma20":
                    raw_condition = raw_condition or (ma10 > ma20)
                elif or_condition == "ma20>ma30":
                    raw_condition = raw_condition or (ma20 > ma30)
                elif or_condition == "close>ma20":
                    raw_condition = raw_condition or (close > ma20)
                elif (
                    or_condition == "volume_pct_chg>0.2" and volume_pct_chg is not None
                ):
                    raw_condition = raw_condition or (volume_pct_chg > 0.2)
                elif or_condition == "macd_diff>0":
                    raw_condition = raw_condition or (macd_diff > 0)
                elif or_condition == "macd_dea>0":
                    raw_condition = raw_condition or (macd_dea > 0)
                elif or_condition == "macd>0":
                    raw_condition = raw_condition or (macd > 0)
                elif or_condition == "10下穿5":
                    raw_condition = raw_condition or ("10下穿5" in ma_cross)
                elif or_condition == "20下穿10":
                    raw_condition = raw_condition or ("20下穿10" in ma_cross)
                elif or_condition == "30下穿20":
                    raw_condition = raw_condition or ("30下穿20" in ma_cross)
                elif or_condition == "ma5<ma10":
                    raw_condition = raw_condition or (ma5 < ma10)
                elif or_condition == "ma10<ma20":
                    raw_condition = raw_condition or (ma10 < ma20)
                elif or_condition == "ma20<ma30":
                    raw_condition = raw_condition or (ma20 < ma30)
                elif or_condition == "close<ma20":
                    raw_condition = raw_condition or (close < ma20)
                elif or_condition == "macd_diff<0":
                    raw_condition = raw_condition or (macd_diff < 0)
                elif or_condition == "macd_dea<0":
                    raw_condition = raw_condition or (macd_dea < 0)
                elif or_condition == "macd<0":
                    raw_condition = raw_condition or (macd < 0)
                else:
                    pass
        return raw_condition

    def draw_quant_pct_chg_bar_chart(
        self, data: pd.DataFrame, strategy_name: str = "全均线策略"
    ):
        """
        绘制pct_chg mean的柱状图表（美观，保存到self.stats_chart_dir）
        :param data: 波段pct_chg_mean的数据
        :return: None
        """
        if data is None or data.empty:
            return None
        # seaborn风格设置
        sns.set_theme(style="whitegrid")
        plt.rcParams["font.sans-serif"] = ["SimHei"]  # 也可直接用字体名
        plt.rcParams["font.size"] = 11  # 设置字体大小
        plt.rcParams["axes.unicode_minus"] = False  # 解决负号显示问题
        chart_dict = {}
        column_name_dict = {
            "account_value_chg": "量化策略涨跌",
        }
        for column_name, column_name_text in column_name_dict.items():
            for bar in data["bar"].unique():
                bar_data = data[data["bar"] == bar].copy()  # 一次筛选即可
                if bar_data.empty:
                    continue
                bar_data.rename(columns={column_name: column_name_text}, inplace=True)
                bar_data.rename(
                    columns={"market_pct_chg": "市场自然涨跌"}, inplace=True
                )
                # 可选：按均值排序
                bar_data.sort_values(by=column_name_text, ascending=False, inplace=True)
                bar_data.reset_index(drop=True, inplace=True)

                # 如果column_name_text是"量化策略涨跌"，则柱状图，同时绘制量化策略涨跌与市场自然涨跌的柱状图，并绘制在同一个图表中
                plt.figure(figsize=(12, 7))

                # 设置x轴位置，为并列柱状图做准备
                x = np.arange(len(bar_data))
                width = 0.35  # 柱状图宽度

                # 确保symbol列是字符串类型，避免matplotlib警告
                # bar_data["symbol"] = bar_data["symbol"].astype(str)
                bar_data["symbol_name"] = bar_data["symbol_name"].astype(str)
                # 绘制量化策略涨跌柱状图（蓝色渐变色）
                bars1 = plt.bar(
                    x - width / 2,
                    bar_data[column_name_text],
                    width,
                    label=column_name_text,
                    color=plt.cm.Blues(np.linspace(0.6, 0.9, len(bar_data))),
                )

                # 绘制市场自然涨跌柱状图（绿色渐变色）
                bars2 = plt.bar(
                    x + width / 2,
                    bar_data["市场自然涨跌"],
                    width,
                    label="市场自然涨跌",
                    color=plt.cm.Greens(np.linspace(0.6, 0.9, len(bar_data))),
                )

                # 设置图表标题和标签
                plt.title(
                    f"{bar}趋势{column_name_text}与市场自然涨跌对比(%)",
                    fontsize=14,
                    fontweight="bold",
                )
                plt.xlabel("Symbol", fontsize=12)
                plt.ylabel("涨跌幅(%)", fontsize=12)
                plt.xticks(x, bar_data["symbol_name"], rotation=45, ha="right")
                plt.legend()
                plt.grid(True, alpha=0.3)

                # 在量化策略涨跌柱状图上方添加数值标签
                for i, (bar1, value1) in enumerate(
                    zip(bars1, bar_data[column_name_text])
                ):
                    plt.text(
                        bar1.get_x() + bar1.get_width() / 2,
                        value1 + (0.01 if value1 >= 0 else -0.01),
                        f"{value1:.3f}%",
                        ha="center",
                        va="bottom" if value1 >= 0 else "top",
                        fontsize=9,
                        fontweight="bold",
                        color="darkblue",
                    )

                # 在市场自然涨跌柱状图上方添加数值标签
                for i, (bar2, value2) in enumerate(
                    zip(bars2, bar_data["市场自然涨跌"])
                ):
                    plt.text(
                        bar2.get_x() + bar2.get_width() / 2,
                        value2 + (0.01 if value2 >= 0 else -0.01),
                        f"{value2:.3f}%",
                        ha="center",
                        va="bottom" if value2 >= 0 else "top",
                        fontsize=9,
                        fontweight="bold",
                        color="darkgreen",
                    )

                plt.tight_layout()

                if self.commission_per_share > 0:
                    save_path = os.path.join(
                        self.stats_chart_dir,
                        f"{bar}_bar_chart_{column_name}_{strategy_name}_with_commission.png",
                    )
                else:
                    save_path = os.path.join(
                        self.stats_chart_dir,
                        f"{bar}_bar_chart_{column_name}_{strategy_name}_without_commission.png",
                    )

                plt.savefig(save_path, dpi=150)
                plt.close()

                sheet_name = f"{bar}_趋势{column_name_text}柱状图_{strategy_name}"
                chart_dict[sheet_name] = save_path
        return chart_dict

    def draw_quant_line_chart(
        self,
        data: pd.DataFrame,
        market_data_pct_chg_df: pd.DataFrame,
        strategy_name: str = "全均线策略",
    ):
        """
        根据量化策略买卖明细记录，绘制量化策略涨跌与市场自然涨跌的折线图
        :param data: 量化策略买卖明细记录
        :param market_data_pct_chg_df: 市场自然涨跌记录
        :param strategy_name: 策略名称
        :return: None
        """
        symbols = data["symbol_name"].unique()
        bars = data["bar"].unique()
        chart_dict = {}
        for symbol in symbols:
            for bar in bars:
                symbol_bar_data = data[
                    (data["symbol_name"] == symbol) & (data["bar"] == bar)
                ]
                if symbol_bar_data.empty:
                    continue

                # 获取第一行数据作为基准
                first_row = symbol_bar_data.iloc[0].copy()
                initial_capital = int(
                    market_data_pct_chg_df.loc[
                        (market_data_pct_chg_df["symbol_name"] == symbol)
                        & (market_data_pct_chg_df["bar"] == bar),
                        "initial_capital",
                    ].values[0]
                )
                # 创建初始化行，设置基准值
                init_row = first_row.copy()
                init_row.loc["profit_loss"] = 0
                init_row.loc["end_account_value"] = (
                    initial_capital  # 量化策略初始值为初始资金
                )
                init_row.loc["end_timestamp"] = first_row["begin_timestamp"]
                init_row.loc["end_date_time"] = first_row["begin_date_time"]
                init_row.loc["end_close"] = first_row["begin_close"]
                init_row.loc["end_ma5"] = first_row["begin_ma5"]
                init_row.loc["end_ma10"] = first_row["begin_ma10"]
                init_row.loc["end_ma20"] = first_row["begin_ma20"]
                init_row.loc["end_ma30"] = first_row["begin_ma30"]
                init_row.loc["end_macd_diff"] = first_row["begin_macd_diff"]
                init_row.loc["end_macd_dea"] = first_row["begin_macd_dea"]
                init_row.loc["end_macd"] = first_row["begin_macd"]
                init_row.loc["pct_chg"] = 0
                init_row.loc["interval_seconds"] = 0
                init_row.loc["interval_minutes"] = 0
                init_row.loc["interval_hours"] = 0
                init_row.loc["interval_days"] = 0

                # 将初始化行添加到数据开头
                symbol_bar_data = pd.concat([pd.DataFrame([init_row]), symbol_bar_data])
                symbol_bar_data.sort_values(
                    by="end_timestamp", ascending=True, inplace=True
                )
                symbol_bar_data.reset_index(drop=True, inplace=True)

                # 确保时间列是datetime类型，避免matplotlib警告
                symbol_bar_data["end_date_time"] = pd.to_datetime(
                    symbol_bar_data["end_date_time"]
                )

                # 计算持仓价值归一化数据（相对于初始价格）
                symbol_bar_data["end_account_value_to_1"] = (
                    symbol_bar_data["end_account_value"] / initial_capital
                )
                symbol_bar_data["end_account_value_to_1"] = symbol_bar_data[
                    "end_account_value_to_1"
                ].round(4)

                # 计算市场价位归一化数据（相对于初始价格）
                symbol_bar_data["end_close_to_1"] = (
                    symbol_bar_data["end_close"] / init_row["end_close"]
                )
                symbol_bar_data["end_close_to_1"] = symbol_bar_data[
                    "end_close_to_1"
                ].round(4)

                # 绘制折线图
                plt.figure(figsize=(12, 7))

                # 绘制量化策略涨跌线（蓝色）
                plt.plot(
                    symbol_bar_data["end_date_time"],
                    symbol_bar_data["end_account_value_to_1"],
                    label="量化策略涨跌",
                    color="blue",
                    linewidth=2,
                    marker="o",
                    markersize=4,
                )

                # 绘制市场自然涨跌线（绿色）
                plt.plot(
                    symbol_bar_data["end_date_time"],
                    symbol_bar_data["end_close_to_1"],
                    label="市场自然涨跌",
                    color="green",
                    linewidth=2,
                    marker="s",
                    markersize=4,
                )

                plt.title(
                    f"{symbol} {bar} 量化与市场折线图_{strategy_name}",
                    fontsize=14,
                    fontweight="bold",
                )
                plt.xlabel("时间", fontsize=12)
                plt.ylabel("涨跌变化", fontsize=12)
                plt.legend(fontsize=11)
                plt.grid(True, alpha=0.3)

                # 设置x轴标签，避免matplotlib警告
                # 选择合适的时间间隔显示标签，避免过于密集
                if len(symbol_bar_data) > 30:
                    # 如果数据点较多，选择间隔显示，但确保第一条和最后一条始终显示
                    step = max(1, len(symbol_bar_data) // 30)

                    # 创建标签索引列表，确保包含首尾数据
                    label_indices = [0]  # 第一条

                    # 添加中间间隔的标签
                    for i in range(step, len(symbol_bar_data) - 1, step):
                        label_indices.append(i)

                    # 添加最后一条（如果还没有包含的话）
                    if len(symbol_bar_data) - 1 not in label_indices:
                        label_indices.append(len(symbol_bar_data) - 1)

                    # 设置x轴标签
                    plt.xticks(
                        symbol_bar_data["end_date_time"].iloc[label_indices],
                        symbol_bar_data["end_date_time"]
                        .iloc[label_indices]
                        .dt.strftime("%Y%m%d %H:%M"),
                        rotation=45,
                        ha="right",
                    )
                else:
                    # 如果数据点较少，全部显示
                    plt.xticks(
                        symbol_bar_data["end_date_time"],
                        symbol_bar_data["end_date_time"].dt.strftime("%Y%m%d %H:%M"),
                        rotation=45,
                        ha="right",
                    )

                plt.tight_layout()

                if self.commission_per_share > 0:
                    save_path = os.path.join(
                        self.stats_chart_dir,
                        f"{symbol}_{bar}_line_chart_{strategy_name}_with_commission.png",
                    )
                else:
                    save_path = os.path.join(
                        self.stats_chart_dir,
                        f"{symbol}_{bar}_line_chart_{strategy_name}_without_commission.png",
                    )

                plt.savefig(save_path, dpi=150, bbox_inches="tight")
                plt.close()

                sheet_name = f"{symbol}_{bar}_折线图_{strategy_name}"
                chart_dict[sheet_name] = save_path
        return chart_dict

    def output_chart_to_excel(self, excel_file_path: str, charts_dict: dict):
        """
        输出Excel文件，包含所有图表
        charts_dict: 图表数据字典，格式为：
        {
            "sheet_name": {
                "chart_name": "chart_path"
            }
        }
        """
        logger.info(f"将图表输出到{excel_file_path}")

        # 打开已经存在的Excel文件
        wb = openpyxl.load_workbook(excel_file_path)

        for sheet_name, chart_path in charts_dict.items():
            try:
                ws = wb.create_sheet(title=sheet_name)
                row_offset = 1
                # Insert chart image
                img = Image(chart_path)
                ws.add_image(img, f"A{row_offset}")

            except Exception as e:
                logger.error(f"输出Excel Sheet {sheet_name} 失败: {e}")
                continue
        # Save Excel file
        wb.save(excel_file_path)
        print(f"Chart saved as {excel_file_path}")