From 26bc1ed7154630898b4bd3b625d87ad34c2799fb Mon Sep 17 00:00:00 2001 From: 27942 <2794236280@qq.com4> Date: Wed, 8 Oct 2025 13:57:44 +0800 Subject: [PATCH] fsefesf --- 回测数据/推测策略，回测.py | 106 +++++++++++++++++++++++++------------ 1 file changed, 71 insertions(+), 35 deletions(-) diff --git a/回测数据/推测策略，回测.py b/回测数据/推测策略，回测.py index 8bb20d7..250ee51 100644 --- a/回测数据/推测策略，回测.py +++ b/回测数据/推测策略，回测.py @@ -24,7 +24,7 @@ headers = { def fetch_kline(day: int): """获取某一天的分钟级 K线数据""" # 构造该日的起止时间戳 - time_ser = datetime.datetime(2025, 9, day) # 修正为2024年9月 + time_ser = datetime.datetime(2025, 10, day) # 修正为2024年9月 start_of_day = time_ser.replace(hour=0, minute=0, second=0, microsecond=0) end_of_day = time_ser.replace(hour=23, minute=59, second=59, microsecond=0) params = { @@ -43,38 +43,50 @@ def fetch_kline(day: int): # ================= 辅助函数 ================= def is_bullish(candle): - """判断是否是阳线（收盘价 > 开盘价）""" - return float(candle['close']) > float(candle['open']) + """判断是否是阳线（开盘价 < 收盘价，即涨）""" + return float(candle['open']) < float(candle['close']) def is_bearish(candle): - """判断是否是阴线（收盘价 < 开盘价）""" - return float(candle['close']) < float(candle['open']) + """判断是否是阴线（开盘价 > 收盘价，即跌）""" + return float(candle['open']) > float(candle['close']) def check_signal(prev, curr): """ - 判断是否出现反包形态： - 1. 前一根阴线，后一根阳线，并且阳线包住前一根阴线 -> 做多信号 - 2. 前一根阳线，后一根阴线，并且阴线包住前一根阳线 -> 做空信号 + 判断是否出现包住形态，返回信号类型和方向： + 1. 前跌后涨包住 -> 做多信号 (bear_bull_engulf) + 2. 前涨后跌包住 -> 做空信号 (bull_bear_engulf) + 3. 前涨后涨包住 -> 做多信号 (bull_bull_engulf) + 4. 前跌后跌包住 -> 做空信号 (bear_bear_engulf) """ p_open, p_close = float(prev['open']), float(prev['close']) c_open, c_close = float(curr['open']), float(curr['close']) - # 情况1：前阴后阳，且阳线包住前阴 + # 情况1：前跌后涨，且涨线包住前跌线 -> 做多信号 if is_bullish(curr) and is_bearish(prev): if c_open < p_close and c_close > p_open: - return "long" + return "long", "bear_bull_engulf" - # 情况2：前阳后阴，且阴线包住前阳 + # 情况2：前涨后跌，且跌线包住前涨线 -> 做空信号 if is_bearish(curr) and is_bullish(prev): if c_open > p_close and c_close < p_open: - return "short" + return "short", "bull_bear_engulf" - return None + # 情况3：前涨后涨，且后涨线包住前涨线 -> 做多信号 + if is_bullish(curr) and is_bullish(prev): + if c_open < p_open and c_close > p_close: + return "long", "bull_bull_engulf" + + # 情况4：前跌后跌，且后跌线包住前跌线 -> 做空信号 + if is_bearish(curr) and is_bearish(prev): + if c_open > p_open and c_close < p_close: + return "short", "bear_bear_engulf" + + return None, None -def simulate_trade(direction, entry_price, future_candles, take_profit_diff=5, stop_loss_diff=-2): +def simulate_trade(direction, entry_price, future_candles, take_profit_diff=10, stop_loss_diff=-2): """ 模拟交易（逐根K线回测） 使用价差来控制止盈止损： @@ -114,12 +126,21 @@ if __name__ == '__main__': zh_project = 0 # 累计盈亏 all_trades = [] # 保存所有交易明细 + # 四种信号类型的统计 + signal_stats = { + "bear_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包涨"}, + "bull_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包跌"}, + "bull_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包涨"}, + "bear_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包跌"} + } + # 遍历 9月1日 ~ 9月27日 - for i in range(1, 30): + for i in range(1, 9): sorted_data = fetch_kline(i) # 获取数据 - signals = wins = 0 - lig_price = low_price = 0 # 分别记录多/空的净收益 + daily_signals = 0 + daily_wins = 0 + daily_profit = 0 # 遍历每根K线，寻找信号 for idx in range(1, len(sorted_data) - 2): # 留出未来K线 @@ -128,47 +149,61 @@ if __name__ == '__main__': future_candles = sorted_data[idx + 2:] # 未来行情 entry_open = float(entry_candle['open']) # 开仓价格 - direction = check_signal(prev, curr) # 判断开仓方向 + direction, signal_type = check_signal(prev, curr) # 判断开仓方向和信号类型 - if direction: - signals += 1 # 总共信号数 + if direction and signal_type: + daily_signals += 1 exit_price, diff, exit_time = simulate_trade(direction, entry_open, future_candles) - # 统计多单/空单盈亏情况 - if direction == "long": - lig_price += diff - if diff > 0: wins += 1 - else: - low_price += diff - if diff > 0: wins += 1 + # 统计该信号类型的表现 + signal_stats[signal_type]["count"] += 1 + signal_stats[signal_type]["total_profit"] += diff + if diff > 0: + signal_stats[signal_type]["wins"] += 1 + daily_wins += 1 + + daily_profit += diff # 保存交易详情 all_trades.append((f"{i}号", entry_candle["id"], "做多" if direction == "long" else "做空", - entry_open, exit_price, diff, exit_time)) + signal_stats[signal_type]["name"], entry_open, exit_price, diff, exit_time)) # 输出每日统计结果 - if signals > 0: + if daily_signals > 0: + win_rate = daily_wins / daily_signals * 100 logger.info( - f"日期：{i}号，信号数={signals}, 胜率={wins / signals * 100:.2f}%，" - f"上涨方向={lig_price:.2f}，下跌方向={low_price:.2f}，综合={lig_price + low_price:.2f}" + f"日期：{i}号，信号数={daily_signals}, 胜率={win_rate:.2f}%，" + f"当日盈亏={daily_profit:.2f}" ) else: logger.info(f"日期：{i}号，没有信号") # 累计盈亏 - zh_project += (lig_price + low_price) + zh_project += daily_profit - logger.success(f"综合价格：{zh_project:.2f}") + logger.success(f"综合盈亏：{zh_project:.2f}") + + # 输出四种信号类型的统计结果 + logger.info("===== 四种信号类型统计 =====") + for signal_type, stats in signal_stats.items(): + if stats["count"] > 0: + win_rate = stats["wins"] / stats["count"] * 100 + logger.info( + f"{stats['name']}: 信号数={stats['count']}, 胜率={win_rate:.2f}%, " + f"总盈亏={stats['total_profit']:.2f}, 平均盈亏={stats['total_profit'] / stats['count']:.2f}" + ) + else: + logger.info(f"{stats['name']}: 无信号") # ===== 输出每笔交易详情 ===== logger.info("===== 每笔交易详情 =====") n = n1 = 0 # n = 总盈利，n1 = 总手续费 - for date, time_str, direction, entry, exit, diff, end_time in all_trades: + for date, time_str, direction, signal_name, entry, exit, diff, end_time in all_trades: profit_amount = diff / entry * 10000 # 计算盈利金额 close_fee = 10000 / entry * exit * 0.0005 # 平仓手续费 logger.info( - f"{date} {time_str} {direction} 入场={entry:.2f} 出场={exit:.2f} 出场时间={end_time} " + f"{date} {time_str} {direction}({signal_name}) 入场={entry:.2f} 出场={exit:.2f} 出场时间={end_time} " f"差价={diff:.2f} 盈利={profit_amount:.2f} " f"开仓手续费=5u 平仓手续费={close_fee:.2f}" ) @@ -178,3 +213,4 @@ if __name__ == '__main__': print(f'一共笔数：{len(all_trades)}') print(f"一共盈利：{n:.2f}") print(f'一共手续费：{n1:.2f}') +