diff --git a/core/trade/__pycache__/orb_trade.cpython-312.pyc b/core/trade/__pycache__/orb_trade.cpython-312.pyc index 59f0225..be7de17 100644 Binary files a/core/trade/__pycache__/orb_trade.cpython-312.pyc and b/core/trade/__pycache__/orb_trade.cpython-312.pyc differ diff --git a/core/trade/orb_trade.py b/core/trade/orb_trade.py index 1ecf580..733e458 100644 --- a/core/trade/orb_trade.py +++ b/core/trade/orb_trade.py @@ -4,6 +4,12 @@ import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns +from openpyxl import Workbook +from openpyxl.drawing.image import Image +import openpyxl +from openpyxl.styles import Font +from PIL import Image as PILImage + import core.logger as logging from config import OKX_MONITOR_CONFIG, MYSQL_CONFIG, WINDOW_SIZE from core.db.db_market_data import DBMarketData @@ -19,11 +25,19 @@ logger = logging.logger class ORBStrategy: def __init__( self, + symbol: str, + bar: str, + start_date: str, + end_date: str, initial_capital=25000, max_leverage=4, risk_per_trade=0.01, commission_per_share=0.0005, + profit_target_multiple=10, is_us_stock=False, + direction=None, + by_sar=False, + symbol_bar_data=None, ): """ 初始化ORB策略参数 @@ -37,18 +51,31 @@ class ORBStrategy: 6. 止损/止盈：根据$R计算，$R=|entry_price-stop_price| 7. 盈利目标：10R，即10*$R 8. 账户净值曲线：账户价值与市场价格 + :param symbol: 股票代码 + :param bar: K线周期 + :param start_date: 开始日期 + :param end_date: 结束日期 :param initial_capital: 初始账户资金（美元） :param max_leverage: 最大杠杆倍数（默认4倍，符合FINRA规定） :param risk_per_trade: 单次交易风险比例（默认1%） :param commission_per_share: 每股交易佣金（美元，默认0.0005） + :param profit_target_multiple: 盈利目标倍数（默认10倍$R，即10R） + :param is_us_stock: 是否是美股 + :param direction: 方向，None=自动，Long=多头，Short=空头 + :param by_sar: 是否根据SAR指标生成信号，True=是，False=否 """ logger.info( - f"初始化ORB策略参数：初始账户资金={initial_capital}，最大杠杆倍数={max_leverage}，单次交易风险比例={risk_per_trade}，每股交易佣金={commission_per_share}" + f"初始化ORB策略参数：股票代码={symbol}，K线周期={bar}，开始日期={start_date}，结束日期={end_date}，初始账户资金={initial_capital}，最大杠杆倍数={max_leverage}，单次交易风险比例={risk_per_trade}，每股交易佣金={commission_per_share}" ) + self.symbol = symbol + self.bar = bar + self.start_date = start_date + self.end_date = end_date self.initial_capital = initial_capital self.max_leverage = max_leverage self.risk_per_trade = risk_per_trade self.commission_per_share = commission_per_share + self.profit_target_multiple = profit_target_multiple self.data = None # 存储K线数据 self.trades = [] # 存储交易记录 self.equity_curve = None # 存储账户净值曲线 @@ -64,9 +91,35 @@ class ORBStrategy: self.db_market_data = DBMarketData(self.db_url) self.is_us_stock = is_us_stock self.output_chart_folder = r"./output/trade_sandbox/orb_strategy/chart/" + self.output_excel_folder = r"./output/trade_sandbox/orb_strategy/excel/" os.makedirs(self.output_chart_folder, exist_ok=True) + os.makedirs(self.output_excel_folder, exist_ok=True) + self.direction = direction + self.by_sar = by_sar + self.direction_desc = "既做多又做空" + if self.direction == "Long": + self.direction_desc = "做多" + elif self.direction == "Short": + self.direction_desc = "做空" + + self.sar_desc = "不考虑SAR" + if self.by_sar: + self.sar_desc = "考虑SAR" + self.symbol_bar_data = symbol_bar_data + + def run(self): + """ + 运行ORB策略 + """ + self.fetch_intraday_data() + self.generate_orb_signals() + self.backtest() + if len(self.trades) > 0: + self.plot_equity_curve() + self.output_trade_summary() + return self.symbol_bar_data, self.trades_df, self.trades_summary_df - def fetch_intraday_data(self, symbol, start_date, end_date, interval="5m"): + def fetch_intraday_data(self): """ 获取日内5分钟K线数据（需yfinance支持，部分数据可能有延迟） :param ticker: 股票代码（如QQQ、TQQQ） @@ -74,36 +127,69 @@ class ORBStrategy: :param end_date: 结束日期（格式：YYYY-MM-DD） :param interval: K线周期（默认5分钟） """ - logger.info(f"开始获取{symbol}数据：{start_date}至{end_date}，间隔{interval}") - # data = yf.download( - # symbol, start=start_date, end=end_date, interval=interval, progress=False - # ) - data = self.db_market_data.query_market_data_by_symbol_bar( - symbol, interval, start=start_date, end=end_date - ) - data = pd.DataFrame(data) - data.sort_values(by="date_time", inplace=True) - # 保留核心列：开盘价、最高价、最低价、收盘价、成交量 - data["Open"] = data["open"] - data["High"] = data["high"] - data["Low"] = data["low"] - data["Close"] = data["close"] - data["Volume"] = data["volume"] - if self.is_us_stock: - date_time_field = "date_time_us" - else: - date_time_field = "date_time" - data[date_time_field] = pd.to_datetime(data[date_time_field]) - # data["Date"]为日期，不包括时分秒，即date_time如果是2025-01-01 10:00:00，则Date为2025-01-01 - data["Date"] = data[date_time_field].dt.date - # 将Date转换为datetime64[ns]类型以确保类型一致 - data["Date"] = pd.to_datetime(data["Date"]) + logger.info(f"开始获取{self.symbol}数据：{self.start_date}至{self.end_date}，间隔{self.bar}") + if self.symbol_bar_data is None or len(self.symbol_bar_data) == 0: + data = self.db_market_data.query_market_data_by_symbol_bar( + self.symbol, self.bar, start=self.start_date, end=self.end_date + ) + data = pd.DataFrame(data) + data.sort_values(by="date_time", inplace=True) + # 保留核心列：开盘价、最高价、最低价、收盘价、成交量 + data["Open"] = data["open"] + data["High"] = data["high"] + data["Low"] = data["low"] + data["Close"] = data["close"] + data["Volume"] = data["volume"] + if self.is_us_stock: + date_time_field = "date_time_us" + else: + date_time_field = "date_time" + data[date_time_field] = pd.to_datetime(data[date_time_field]) + # data["Date"]为日期，不包括时分秒，即date_time如果是2025-01-01 10:00:00，则Date为2025-01-01 + data["Date"] = data[date_time_field].dt.date + # 将Date转换为datetime64[ns]类型以确保类型一致 + data["Date"] = pd.to_datetime(data["Date"]) + # 最小data["Date"] + self.start_date = data["Date"].min().strftime("%Y-%m-%d") + # 最大data["Date"] + self.end_date = data["Date"].max().strftime("%Y-%m-%d") - self.data = data[ - ["symbol", "bar", "Date", date_time_field, "Open", "High", "Low", "Close", "Volume"] - ].copy() - self.data.rename(columns={date_time_field: "date_time"}, inplace=True) - logger.info(f"成功获取{symbol}数据：{len(self.data)}根{interval}K线") + self.data = data[ + [ + "symbol", + "bar", + "Date", + date_time_field, + "Open", + "High", + "Low", + "Close", + "Volume", + "sar_signal", + ] + ].copy() + self.data.rename(columns={date_time_field: "date_time"}, inplace=True) + self.symbol_bar_data = self.data.copy() + else: + self.data = self.symbol_bar_data.copy() + + # 获取Close的mean + self.close_mean = self.data["Close"].mean() + if self.close_mean > 10000: + self.initial_capital = self.initial_capital * 10000 + elif self.close_mean > 5000: + self.initial_capital = self.initial_capital * 5000 + elif self.close_mean > 1000: + self.initial_capital = self.initial_capital * 1000 + elif self.close_mean > 500: + self.initial_capital = self.initial_capital * 500 + elif self.close_mean > 100: + self.initial_capital = self.initial_capital * 100 + else: + pass + logger.info(f"收盘价均值：{self.close_mean}") + logger.info(f"初始资金调整为：{self.initial_capital}") + logger.info(f"成功获取{self.symbol}数据：{len(self.data)}根{self.bar}K线,开始日期={self.start_date}，结束日期={self.end_date}") def calculate_shares(self, account_value, entry_price, stop_price): """ @@ -134,20 +220,15 @@ class ORBStrategy: return int(max_shares) # 股数取整 - def generate_orb_signals(self, direction: str = None, by_sar: bool = False): + def generate_orb_signals(self): """ 生成ORB策略信号（每日仅1次交易机会） - 第一根5分钟K线：确定开盘区间（High1, Low1） - 第二根5分钟K线：根据第一根K线方向生成多空信号 - :param direction: 方向，None=自动，Long=多头，Short=空头 - :param by_sar: 是否根据SAR指标生成信号，True=是，False=否 """ - direction_desc = "既做多又做空" - if direction == "Long": - direction_desc = "做多" - elif direction == "Short": - direction_desc = "做空" - logger.info(f"开始生成ORB策略信号：{direction_desc}，根据SAR指标：{by_sar}") + logger.info( + f"开始生成ORB策略信号：{self.direction_desc}，根据SAR指标：{self.by_sar}" + ) if self.data is None: raise ValueError("请先调用fetch_intraday_data获取数据") @@ -164,6 +245,7 @@ class ORBStrategy: low1 = first_candle["Low"] open1 = first_candle["Open"] close1 = first_candle["Close"] + sar_signal = first_candle["sar_signal"] # 第二根5分钟K线（entry信号） second_candle = daily_data.iloc[1] @@ -171,11 +253,19 @@ class ORBStrategy: entry_time = second_candle.date_time # entry时间 # 生成信号：第一根K线方向决定多空（排除十字星：open1 == close1） - if open1 < close1 and (direction == "Long" or direction is None): + if ( + open1 < close1 + and (self.direction == "Long" or self.direction is None) + and ((self.by_sar and sar_signal == "SAR多头") or not self.by_sar) + ): # 第一根K线收涨→多头信号 signal = "Long" stop_price = low1 # 多头止损=第一根K线最低价 - elif open1 > close1 and (direction == "Short" or direction is None): + elif ( + open1 > close1 + and (self.direction == "Short" or self.direction is None) + and ((self.by_sar and sar_signal == "SAR空头") or not self.by_sar) + ): # 第一根K线收跌→空头信号 signal = "Short" stop_price = high1 # 空头止损=第一根K线最高价 @@ -213,12 +303,12 @@ class ORBStrategy: f"生成信号完成：共{len(signals_df)}个交易日，其中多头{sum(signals_df['Signal']=='Long')}次，空头{sum(signals_df['Signal']=='Short')}次" ) - def backtest(self, profit_target_multiple=10): + def backtest(self): """ 回测ORB策略 :param profit_target_multiple: 盈利目标倍数（默认10倍$R，即10R） """ - logger.info(f"开始回测ORB策略：盈利目标倍数={profit_target_multiple}") + logger.info(f"开始回测ORB策略：盈利目标倍数={self.profit_target_multiple}") if "Signal" not in self.data.columns: raise ValueError("请先调用generate_orb_signals生成策略信号") @@ -255,9 +345,9 @@ class ORBStrategy: low1 = signal_row["Low1"] risk_assumed = abs(entry_price - stop_price) # 计算$R profit_target = ( - entry_price + (risk_assumed * profit_target_multiple) + entry_price + (risk_assumed * self.profit_target_multiple) if signal == "Long" - else entry_price - (risk_assumed * profit_target_multiple) + else entry_price - (risk_assumed * self.profit_target_multiple) ) # 计算交易股数 @@ -293,7 +383,7 @@ class ORBStrategy: break elif high >= profit_target: exit_price = profit_target - exit_reason = "Profit Target (10R)" + exit_reason = f"Profit Target ({self.profit_target_multiple}R)" exit_time = row["date_time"] break elif signal == "Short": @@ -305,7 +395,7 @@ class ORBStrategy: break elif low <= profit_target: exit_price = profit_target - exit_reason = "Profit Target (10R)" + exit_reason = f"Profit Target ({self.profit_target_multiple}R)" exit_time = row["date_time"] break @@ -333,6 +423,10 @@ class ORBStrategy: self.trades.append( { "TradeID": trade_id, + "Direction": self.direction_desc, + "BySar": self.sar_desc, + "Symbol": self.symbol, + "Bar": self.bar, "Date": date, "Signal": signal, "EntryTime": signal_row.date_time.strftime("%Y-%m-%d %H:%M:%S"), @@ -353,40 +447,91 @@ class ORBStrategy: equity_history.append(account_value) trade_id += 1 + if len(self.trades) == 0: + logger.info("没有交易") + self.trades_df = pd.DataFrame() + self.initial_trade_summary() + return # 生成净值曲线 self.create_equity_curve() # 输出回测结果 - trades_df = pd.DataFrame(self.trades) + self.trades_df = pd.DataFrame(self.trades) + self.trades_df.sort_values(by="ExitTime", inplace=True) total_return = ( (account_value - self.initial_capital) / self.initial_capital * 100 ) win_rate = ( - (trades_df["ProfitLoss"] > 0).sum() / len(trades_df) * 100 - if len(trades_df) > 0 + (self.trades_df["ProfitLoss"] > 0).sum() / len(self.trades_df) * 100 + if len(self.trades_df) > 0 else 0 ) # 计算盈亏比 - profit_sum = trades_df[trades_df["ProfitLoss"] > 0]["ProfitLoss"].sum() - loss_sum = abs(trades_df[trades_df["ProfitLoss"] < 0]["ProfitLoss"].sum()) + profit_sum = self.trades_df[self.trades_df["ProfitLoss"] > 0]["ProfitLoss"].sum() + loss_sum = abs(self.trades_df[self.trades_df["ProfitLoss"] < 0]["ProfitLoss"].sum()) if loss_sum == 0: - profit_loss_ratio = float('inf') + profit_loss_ratio = float("inf") else: profit_loss_ratio = (profit_sum / loss_sum) * 100 + first_entry_price = self.trades_df.iloc[0]["EntryPrice"] + last_exit_price = self.trades_df.iloc[-1]["ExitPrice"] + natural_return = (last_exit_price - first_entry_price) / first_entry_price * 100 + self.initial_trade_summary() + if len(self.trades_df) > 0: + logger.info("\n" + "=" * 50) + logger.info("ORB策略回测结果") + logger.info("=" * 50) + logger.info(f"股票代码：{self.symbol}") + logger.info(f"K线周期：{self.bar}") + logger.info(f"开始日期：{self.start_date}") + logger.info(f"结束日期：{self.end_date}") + logger.info(f"盈利目标倍数：{self.profit_target_multiple}") + logger.info(f"初始资金：${self.initial_capital:,.2f}") + logger.info(f"最终资金：${account_value:,.2f}") + self.trades_summary["最终资金$"] = account_value + logger.info(f"总收益率：{total_return:.2f}%") + self.trades_summary["总收益率%"] = total_return + logger.info(f"自然收益率：{natural_return:.2f}%") + self.trades_summary["自然收益率%"] = natural_return + logger.info(f"总交易次数：{len(self.trades_df)}") + self.trades_summary["总交易次数"] = len(self.trades_df) + logger.info(f"盈亏比：{profit_loss_ratio:.2f}%") + self.trades_summary["盈亏比%"] = profit_loss_ratio + logger.info(f"胜率：{win_rate:.2f}%") + self.trades_summary["胜率%"] = win_rate + logger.info(f"平均每笔盈亏：${self.trades_df['ProfitLoss'].mean():.2f}") + self.trades_summary["平均每笔盈亏$"] = self.trades_df["ProfitLoss"].mean() + logger.info(f"最大单笔盈利：${self.trades_df['ProfitLoss'].max():.2f}") + self.trades_summary["最大单笔盈利$"] = self.trades_df["ProfitLoss"].max() + logger.info(f"最大单笔亏损：${abs(self.trades_df['ProfitLoss'].min()):.2f}") + self.trades_summary["最大单笔亏损$"] = abs(self.trades_df["ProfitLoss"].min()) + else: + logger.info("没有交易") + self.trades_summary_df = pd.DataFrame([self.trades_summary]) + + def initial_trade_summary(self): + """ + 初始化交易总结 + """ + self.trades_summary = {} + self.trades_summary["方向"] = self.direction_desc + self.trades_summary["根据SAR"] = self.sar_desc + self.trades_summary["股票代码"] = self.symbol + self.trades_summary["K线周期"] = self.bar + self.trades_summary["开始日期"] = self.start_date + self.trades_summary["结束日期"] = self.end_date + self.trades_summary["盈利目标倍数"] = self.profit_target_multiple + self.trades_summary["初始资金$"] = self.initial_capital + self.trades_summary["最终资金$"] = self.initial_capital + self.trades_summary["总收益率%"] = 0 + self.trades_summary["自然收益率%"] = 0 + self.trades_summary["总交易次数"] = 0 + self.trades_summary["盈亏比%"] = 0 + self.trades_summary["胜率%"] = 0 + self.trades_summary["平均每笔盈亏$"] = 0 + self.trades_summary["最大单笔盈利$"] = 0 + self.trades_summary["最大单笔亏损$"] = 0 - logger.info("\n" + "=" * 50) - logger.info("ORB策略回测结果") - logger.info("=" * 50) - logger.info(f"初始资金：${self.initial_capital:,.2f}") - logger.info(f"最终资金：${account_value:,.2f}") - logger.info(f"总收益率：{total_return:.2f}%") - logger.info(f"总交易次数：{len(trades_df)}") - logger.info(f"盈亏比：{profit_loss_ratio:.2f}%") - logger.info(f"胜率：{win_rate:.2f}%") - if len(trades_df) > 0: - logger.info(f"平均每笔盈亏：${trades_df['ProfitLoss'].mean():.2f}") - logger.info(f"最大单笔盈利：${trades_df['ProfitLoss'].max():.2f}") - logger.info(f"最大单笔亏损：${trades_df['ProfitLoss'].min():.2f}") def create_equity_curve(self): """ @@ -437,9 +582,25 @@ class ORBStrategy: account_value_to_1 = self.equity_curve["AccountValue"] / first_account_value market_price_to_1 = self.equity_curve["MarketPrice"] / first_market_price plt.figure(figsize=(12, 6)) - plt.plot(self.equity_curve["DateTime"], account_value_to_1, label="账户价值", color='blue', linewidth=2, marker='o', markersize=4) - plt.plot(self.equity_curve["DateTime"], market_price_to_1, label="市场价格", color='green', linewidth=2, marker='s', markersize=4) - plt.title(f"ORB策略账户净值曲线 {symbol} {bar}", fontsize=14, fontweight='bold') + plt.plot( + self.equity_curve["DateTime"], + account_value_to_1, + label="账户价值", + color="blue", + linewidth=2, + marker="o", + markersize=4, + ) + plt.plot( + self.equity_curve["DateTime"], + market_price_to_1, + label="市场价格", + color="green", + linewidth=2, + marker="s", + markersize=4, + ) + plt.title(f"ORB曲线 {symbol} {bar} {self.direction_desc} {self.sar_desc}", fontsize=14, fontweight="bold") plt.xlabel("时间", fontsize=12) plt.ylabel("涨跌变化", fontsize=12) plt.legend(fontsize=11) @@ -450,54 +611,108 @@ class ORBStrategy: if len(self.equity_curve) > 30: # 如果数据点较多，选择间隔显示，但确保第一条和最后一条始终显示 step = max(1, len(self.equity_curve) // 30) - + # 创建标签索引列表，确保包含首尾数据 label_indices = [0] # 第一条 - + # 添加中间间隔的标签 for i in range(step, len(self.equity_curve) - 1, step): label_indices.append(i) - + # 添加最后一条（如果还没有包含的话） if len(self.equity_curve) - 1 not in label_indices: label_indices.append(len(self.equity_curve) - 1) - + # 设置x轴标签 - plt.xticks(self.equity_curve["DateTime"].iloc[label_indices], - self.equity_curve["DateTime"].iloc[label_indices], - rotation=45, ha='right', fontsize=10) + plt.xticks( + self.equity_curve["DateTime"].iloc[label_indices], + self.equity_curve["DateTime"].iloc[label_indices], + rotation=45, + ha="right", + fontsize=10, + ) else: # 如果数据点较少，全部显示 - plt.xticks(self.equity_curve["DateTime"], - self.equity_curve["DateTime"], - rotation=45, ha='right', fontsize=10) + plt.xticks( + self.equity_curve["DateTime"], + self.equity_curve["DateTime"], + rotation=45, + ha="right", + fontsize=10, + ) plt.tight_layout() - save_path = f"{self.output_chart_folder}/{symbol}_{bar}_orb_strategy_equity_curve.png" - plt.savefig(save_path, dpi=150, bbox_inches='tight') + self.chart_save_path = ( + f"{self.output_chart_folder}/{symbol}_{bar}_{self.direction_desc}_{self.sar_desc}_orb_strategy_equity_curve.png" + ) + plt.savefig(self.chart_save_path, dpi=150, bbox_inches="tight") plt.close() + + def output_trade_summary(self): + """ + 输出交易明细，交易总结与Chart图片到Excel + """ + start_date = self.start_date.replace("-", "") + end_date = self.end_date.replace("-", "") + output_file_name = f"orb_{self.symbol}_{self.bar}_{start_date}_{end_date}_{self.direction_desc}_{self.sar_desc}.xlsx" + output_file_path = os.path.join(self.output_excel_folder, output_file_name) + logger.info(f"导出{output_file_path}") + with pd.ExcelWriter(output_file_path) as writer: + self.trades_df.to_excel(writer, sheet_name="交易明细", index=False) + self.trades_summary_df.to_excel(writer, sheet_name="交易总结", index=False) + if os.path.exists(self.chart_save_path): + charts_dict = { + "账户净值曲线": self.chart_save_path + } + self.output_chart_to_excel(output_file_path, charts_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) + logger.info(f"图表已输出到{excel_file_path}") + # ------------------- 策略示例：回测QQQ的ORB策略（2016-2023） ------------------- if __name__ == "__main__": # 初始化ORB策略 orb_strategy = ORBStrategy( + symbol="ETH-USDT", + bar="5m", + start_date="2025-05-15", + end_date="2025-08-20", initial_capital=25000, max_leverage=4, risk_per_trade=0.01, commission_per_share=0.0005, + profit_target_multiple=10, + is_us_stock=False, + direction=None, + by_sar=False, ) - # 1. 获取QQQ的5分钟日内数据（2024-2025，注意：yfinance免费版可能限制历史日内数据，建议用专业数据源） - orb_strategy.fetch_intraday_data( - symbol="ETH-USDT", start_date="2025-05-15", end_date="2025-08-20", interval="5m" - ) - - # 2. 生成ORB策略信号 - orb_strategy.generate_orb_signals() - - # 3. 回测策略（盈利目标10R） - orb_strategy.backtest(profit_target_multiple=10) - - # 4. 绘制净值曲线 - orb_strategy.plot_equity_curve() + orb_strategy.run() diff --git a/orb_trade_main.py b/orb_trade_main.py index 5421938..0f53e19 100644 --- a/orb_trade_main.py +++ b/orb_trade_main.py @@ -1,35 +1,343 @@ from core.trade.orb_trade import ORBStrategy -from config import US_STOCK_MONITOR_CONFIG +from config import US_STOCK_MONITOR_CONFIG, OKX_MONITOR_CONFIG import core.logger as logging +from datetime import datetime +from openpyxl import Workbook +from openpyxl.drawing.image import Image +import openpyxl +import pandas as pd +import os logger = logging.logger + def main(): - symbols = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get("symbols", ["QQQ"]) + is_us_stock_list = [True, False] + bar = "5m" + direction_list = [None, "Long", "Short"] + by_sar_list = [False, True] + start_date = "2024-01-01" + end_date = datetime.now().strftime("%Y-%m-%d") + profit_target_multiple = 10 + initial_capital = 25000 + max_leverage = 4 + risk_per_trade = 0.01 + commission_per_share = 0.0005 + + trades_df_list = [] + trades_summary_df_list = [] + symbol_data_cache = [] + for is_us_stock in is_us_stock_list: + for direction in direction_list: + for by_sar in by_sar_list: + if is_us_stock: + symbols = US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get( + "symbols", ["QQQ"] + ) + else: + symbols = OKX_MONITOR_CONFIG.get("volume_monitor", {}).get( + "symbols", ["BTC-USDT"] + ) + for symbol in symbols: + logger.info( + f"开始回测 {symbol}, 交易周期:{bar}, 开始日期:{start_date}, 结束日期:{end_date}, 是否是美股:{is_us_stock}, 交易方向:{direction}, 是否使用SAR:{by_sar}" + ) + symbol_bar_data = None + found_symbol_bar_data = False + for symbol_data_dict in symbol_data_cache: + if ( + symbol_data_dict["symbol"] == symbol + and symbol_data_dict["bar"] == bar + ): + symbol_bar_data = symbol_data_dict["data"] + found_symbol_bar_data = True + break + + orb_strategy = ORBStrategy( + symbol=symbol, + bar=bar, + start_date=start_date, + end_date=end_date, + is_us_stock=is_us_stock, + direction=direction, + by_sar=by_sar, + profit_target_multiple=profit_target_multiple, + initial_capital=initial_capital, + max_leverage=max_leverage, + risk_per_trade=risk_per_trade, + commission_per_share=commission_per_share, + symbol_bar_data=symbol_bar_data, + ) + symbol_bar_data, trades_df, trades_summary_df = orb_strategy.run() + if symbol_bar_data is None or len(symbol_bar_data) == 0: + continue + if not found_symbol_bar_data: + symbol_data_cache.append( + {"symbol": symbol, "bar": bar, "data": symbol_bar_data} + ) + if trades_summary_df is None or len(trades_summary_df) == 0: + continue + trades_summary_df_list.append(trades_summary_df) + trades_df_list.append(trades_df) + total_trades_df = pd.concat(trades_df_list) + total_trades_summary_df = pd.concat(trades_summary_df_list) + statitics_dict = statistics_summary(total_trades_summary_df) + output_excel_folder = r"./output/trade_sandbox/orb_strategy/excel/summary/" + os.makedirs(output_excel_folder, exist_ok=True) + now_str = datetime.now().strftime("%Y%m%d%H%M%S") + excel_file_name = f"orb_strategy_summary_{now_str}.xlsx" + output_file_path = os.path.join(output_excel_folder, excel_file_name) + with pd.ExcelWriter(output_file_path) as writer: + total_trades_df.to_excel(writer, sheet_name="交易详情", index=False) + total_trades_summary_df.to_excel(writer, sheet_name="交易总结", index=False) + statitics_dict["statistics_summary_df"].to_excel( + writer, sheet_name="统计总结", index=False + ) + statitics_dict["max_total_return_record_df"].to_excel( + writer, sheet_name="最大总收益率记录", index=False + ) + statitics_dict["max_total_return_record_df_grouped_count"].to_excel( + writer, sheet_name="最大总收益率记录_方向和根据SAR的组合", index=False + ) + statitics_dict["max_total_return_record_df_direction_count"].to_excel( + writer, sheet_name="最大总收益率记录_方向", index=False + ) + statitics_dict["max_total_return_record_df_sar_count"].to_excel( + writer, sheet_name="最大总收益率记录_根据SAR", index=False + ) + chart_path = r"./output/trade_sandbox/orb_strategy/chart/" + os.makedirs(chart_path, exist_ok=True) + copy_chart_to_excel(chart_path, output_file_path) + logger.info(f"交易总结已输出到{output_file_path}") + + +def statistics_summary(trades_summary_df: pd.DataFrame): + statistics_summary_list = [] + summary = {} + # 1. 统计总收益率% > 0 的占比 + total_return_gt_0 = trades_summary_df[trades_summary_df["总收益率%"] > 0].shape[0] + total_return_gt_0_ratio = round((total_return_gt_0 / trades_summary_df.shape[0]) * 100, 2) + summary["总收益率%>0占比"] = total_return_gt_0_ratio + logger.info(f"总收益率% > 0 的占比：{total_return_gt_0_ratio:.2f}%") + # 2. 统计总收益率% > 自然收益率% 的占比 + total_return_gt_natural_return = trades_summary_df[ + trades_summary_df["总收益率%"] > trades_summary_df["自然收益率%"] + ].shape[0] + total_return_gt_natural_return_ratio = ( + round((total_return_gt_natural_return / trades_summary_df.shape[0]) * 100, 2) + ) + summary["总收益率%>自然收益率%占比"] = total_return_gt_natural_return_ratio + logger.info( + f"总收益率% > 自然收益率% 的占比：{total_return_gt_natural_return_ratio:.2f}%" + ) + statistics_summary_list.append(summary) + statistics_summary_df = pd.DataFrame(statistics_summary_list) + + symbol_list = trades_summary_df["股票代码"].unique() + max_total_return_record_list = [] + for symbol in symbol_list: + trades_summary_df_copy = trades_summary_df.copy() + symbol_trades_summary_df = trades_summary_df_copy[ + trades_summary_df_copy["股票代码"] == symbol + ] + symbol_trades_summary_df.reset_index(drop=True, inplace=True) + if symbol_trades_summary_df.empty: + continue + # 过滤掉NaN，避免idxmax报错 + valid_df = symbol_trades_summary_df[ + symbol_trades_summary_df["总收益率%"].notna() + ] + if valid_df.empty: + continue + # 获得总收益率%最大的记录 + max_idx = valid_df["总收益率%"].idxmax() + max_total_return_record = symbol_trades_summary_df.loc[max_idx] + summary = {} + summary["股票代码"] = symbol + summary["方向"] = max_total_return_record["方向"] + summary["根据SAR"] = max_total_return_record["根据SAR"] + summary["总收益率%"] = max_total_return_record["总收益率%"] + summary["自然收益率%"] = max_total_return_record["自然收益率%"] + max_total_return_record_list.append(summary) + max_total_return_record_df = pd.DataFrame(max_total_return_record_list) + # 统计max_total_return_record_df中方向和根据SAR的组合（使用size更稳健，支持空分组与缺失值） + # 强制将分组键转为可哈希的标量类型，避免单元格为Series/列表导致的unhashable错误 + if len(max_total_return_record_df) > 0: + + def _to_hashable_scalar(v): + # 标量或None直接返回 + if isinstance(v, (str, int, float, bool)) or v is None: + return v + try: + import numpy as _np + + if _np.isscalar(v): + return v + except Exception: + pass + # 其它（如Series、list、dict、ndarray等）转字符串 + return str(v) + + for key_col in ["方向", "根据SAR"]: + if key_col in max_total_return_record_df.columns: + max_total_return_record_df[key_col] = max_total_return_record_df[ + key_col + ].apply(_to_hashable_scalar) + # 分组统计 + max_total_return_record_df_grouped_count = ( + max_total_return_record_df.groupby(["方向", "根据SAR"], dropna=False) + .size() + .reset_index(name="数量") + ) + max_total_return_record_df_grouped_count.sort_values( + by="数量", ascending=False, inplace=True + ) + max_total_return_record_df_grouped_count.reset_index(drop=True, inplace=True) + + # 统计方向的记录数目 + max_total_return_record_df_direction_count = ( + max_total_return_record_df.groupby(["方向"], dropna=False) + .size() + .reset_index(name="数量") + ) + max_total_return_record_df_direction_count.sort_values( + by="数量", ascending=False, inplace=True + ) + max_total_return_record_df_direction_count.reset_index(drop=True, inplace=True) + + # 统计根据SAR的记录数目 + max_total_return_record_df_sar_count = ( + max_total_return_record_df.groupby(["根据SAR"], dropna=False) + .size() + .reset_index(name="数量") + ) + max_total_return_record_df_sar_count.sort_values( + by="数量", ascending=False, inplace=True + ) + max_total_return_record_df_sar_count.reset_index(drop=True, inplace=True) + else: + # 构造空结果，保证下游写入Excel不报错 + max_total_return_record_df_grouped_count = pd.DataFrame( + columns=["方向", "根据SAR", "数量"] + ) + max_total_return_record_df_direction_count = pd.DataFrame( + columns=["方向", "数量"] + ) + max_total_return_record_df_sar_count = pd.DataFrame(columns=["根据SAR", "数量"]) + + result = { + "statistics_summary_df": statistics_summary_df, + "max_total_return_record_df": max_total_return_record_df, + "max_total_return_record_df_grouped_count": max_total_return_record_df_grouped_count, + "max_total_return_record_df_direction_count": max_total_return_record_df_direction_count, + "max_total_return_record_df_sar_count": max_total_return_record_df_sar_count, + } + return result + + +def copy_chart_to_excel(chart_path: str, excel_file_path: str): + f""" + 将chart图片复制到excel中 + 算法： + 1. 读取chart_path + 2. chart文件名开头是symbol，结尾是.png + 3. 每个symbol创建一个Excel Sheet，Sheet名称为symbol_chart + 4. 将chart图片插入到Sheet中 + 5. 要求每张图片大小为800x400 + 6. 要求两张图片左右并列显示 + 7. 要求上下图片间距为20px + """ + # 收集所有图片 + if not os.path.isdir(chart_path): + return + chart_files = [f for f in os.listdir(chart_path) if f.lower().endswith(".png")] + if len(chart_files) == 0: + return + + # 汇总需要处理的symbol列表（去重） + symbols = set(US_STOCK_MONITOR_CONFIG.get("volume_monitor", {}).get("symbols", ["QQQ"])) + symbols.update(OKX_MONITOR_CONFIG.get("volume_monitor", {}).get("symbols", ["BTC-USDT"])) + symbols = list(symbols) + symbols.sort() + # 每个symbol创建一个sheet并插图 for symbol in symbols: - logger.info(f"开始回测 {symbol}") - # 初始化ORB策略 - orb_strategy = ORBStrategy( - initial_capital=25000, - max_leverage=4, - risk_per_trade=0.01, - commission_per_share=0.0005, - is_us_stock=True, - ) - # 1. 获取QQQ的5分钟日内数据（2024-2025，注意：yfinance免费版可能限制历史日内数据，建议用专业数据源） - orb_strategy.fetch_intraday_data( - symbol=symbol, start_date="2024-11-30", end_date="2025-08-30", interval="5m" - ) + logger.info(f"开始保存{symbol}的图表") + symbol_files = [f for f in chart_files if f.startswith(symbol)] + if len(symbol_files) == 0: + continue + # 排序以稳定显示顺序 + symbol_files.sort() + copy_chart_to_excel_sheet(chart_path, symbol_files, excel_file_path, symbol) - # 2. 生成ORB策略信号 - orb_strategy.generate_orb_signals() - # 3. 回测策略（盈利目标10R） - orb_strategy.backtest(profit_target_multiple=10) +def copy_chart_to_excel_sheet( + chart_path: str, chart_files: list, excel_file_path: str, symbol: str +): + """ + 将chart图片复制到excel中 + 算法： + 1. 读取chart_files + 2. 创建一个Excel Sheet，Sheet名称为{symbol}_chart + 3. 将chart_files中的图片插入到Sheet中 + 4. 要求每张图片大小为800x400 + 5. 要求两张图片左右并列显示, 如6张图片则图片行数为3，列数为2 + 6. 要求上下图片间距为20px + """ + # 打开已经存在的Excel文件 + wb = openpyxl.load_workbook(excel_file_path) + # 如果sheet已存在，先删除，避免重复插入 + sheet_name = f"{symbol}_chart" + if sheet_name in wb.sheetnames: + del wb[sheet_name] + ws = wb.create_sheet(title=sheet_name) - # 4. 绘制净值曲线 - orb_strategy.plot_equity_curve() + # 两列布局：左列A，右列L；行间距通过起始行步进控制 + left_col = "A" + right_col = "L" + row_step = 26 # 行步进，控制上下间距 + + for idx, chart_file in enumerate(chart_files): + try: + img_path = os.path.join(chart_path, chart_file) + img = Image(img_path) + # 设置图片尺寸 800x400 像素 + img.width = 800 + img.height = 400 + + row_block = idx // 2 + col_block = idx % 2 + anchor_col = left_col if col_block == 0 else right_col + anchor_cell = f"{anchor_col}{1 + row_block * row_step}" + ws.add_image(img, anchor_cell) + except Exception: + continue + + wb.save(excel_file_path) + logger.info(f"{symbol}的图表已输出到{excel_file_path}") + + +def test(): + orb_strategy = ORBStrategy( + symbol="BTC-USDT", + bar="5m", + start_date="2024-01-01", + end_date="2025-09-02", + is_us_stock=False, + direction=None, + by_sar=True, + profit_target_multiple=10, + initial_capital=25000, + max_leverage=4, + risk_per_trade=0.01, + commission_per_share=0.0005, + ) + orb_strategy.run() if __name__ == "__main__": - main() \ No newline at end of file + # main() + + chart_path = r"./output/trade_sandbox/orb_strategy/chart/" + excel_file_path = r"./output/trade_sandbox/orb_strategy/excel/summary/orb_strategy_summary_20250902174203.xlsx" + copy_chart_to_excel(chart_path, excel_file_path) + # test()