From 204b1ac69c35fa68e194d74e8576077bf9e447a3 Mon Sep 17 00:00:00 2001 From: 27942 <2794236280@qq.com> Date: Fri, 17 Oct 2025 16:30:39 +0800 Subject: [PATCH] dededdew --- weex/backtest_trades.html | 2 +- ...读取数据库分析数据2.0-优化信号版-反向版.py | 331 +++++++++++++++ weex/读取数据库分析数据2.0-优化信号版.py | 5 +- weex/读取数据库分析数据2.0-优化信号版2.0.py | 389 ++++++++++++++++-- 4 files changed, 699 insertions(+), 28 deletions(-) create mode 100644 weex/读取数据库分析数据2.0-优化信号版-反向版.py diff --git a/weex/backtest_trades.html b/weex/backtest_trades.html index 23a4ad6..684cf87 100644 --- a/weex/backtest_trades.html +++ b/weex/backtest_trades.html @@ -2,6 +2,6 @@
-
+
\ No newline at end of file diff --git a/weex/读取数据库分析数据2.0-优化信号版-反向版.py b/weex/读取数据库分析数据2.0-优化信号版-反向版.py new file mode 100644 index 0000000..4ff2f18 --- /dev/null +++ b/weex/读取数据库分析数据2.0-优化信号版-反向版.py @@ -0,0 +1,331 @@ +""" +量化交易回测系统 - 策略方向反转版 +功能：基于包住形态的交易信号识别和回测分析（信号方向已按用户要求反转） +作者：量化交易团队（修改：ChatGPT） +版本：2.1 +""" + +import datetime +from typing import List, Dict, Tuple, Optional, Any +from dataclasses import dataclass +from loguru import logger +from peewee import fn +from models.weex import Weex15, Weex1 + + +# =============================================================== +# 📊 配置管理类 +# =============================================================== + +@dataclass +class BacktestConfig: + """回测配置类""" + # 交易参数 + take_profit: float = 8.0 # 止盈点数 + stop_loss: float = -1.0 # 止损点数（负数表示点差方向） + contract_size: float = 10000 # 合约规模 + open_fee: float = 5.0 # 开仓手续费 + close_fee_rate: float = 0.0005 # 平仓手续费率 + + # 回测日期范围 + start_date: str = "2025-07-01" + end_date: str = "2025-07-31" + + # 信号参数 + enable_bear_bull_engulf: bool = True # 涨包跌（命名保留） + enable_bull_bear_engulf: bool = True # 跌包涨（命名保留） + + def __post_init__(self): + """验证配置参数""" + if self.take_profit <= 0: + raise ValueError("止盈点数必须大于0") + if self.stop_loss >= 0: + raise ValueError("止损点数必须小于0") + + +@dataclass +class TradeRecord: + """交易记录类""" + entry_time: datetime.datetime + exit_time: datetime.datetime + signal_type: str + direction: str + entry_price: float + exit_price: float + profit_loss: float + profit_amount: float + total_fee: float + net_profit: float + + +@dataclass +class SignalStats: + """信号统计类""" + signal_name: str + count: int = 0 + wins: int = 0 + total_profit: float = 0.0 + + @property + def win_rate(self) -> float: + """胜率计算""" + return (self.wins / self.count * 100) if self.count > 0 else 0.0 + + @property + def avg_profit(self) -> float: + """平均盈利""" + return self.total_profit / self.count if self.count > 0 else 0.0 + + +# =============================================================== +# 📊 数据获取模块 +# =============================================================== + +def get_data_by_date(model, date_str): + """按天获取指定表的数据，date_str 需为 YYYY-MM-DD""" + try: + target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d') + except ValueError: + logger.error("日期格式不正确，请使用 YYYY-MM-DD 格式。") + return [] + + start_ts = int(target_date.timestamp() * 1000) + end_ts = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1 + + query = (model + .select() + .where(model.id.between(start_ts, end_ts)) + .order_by(model.id.asc())) + + return [ + {'id': i.id, 'open': i.open, 'high': i.high, 'low': i.low, 'close': i.close} + for i in query + ] + + +def get_future_data_1min(start_ts, end_ts): + """获取指定时间范围内的 1 分钟数据""" + query = (Weex1 + .select() + .where(Weex1.id.between(start_ts, end_ts)) + .order_by(Weex1.id.asc())) + return [{'id': i.id, 'open': i.open, 'high': i.high, 'low': i.low, 'close': i.close} for i in query] + + +# =============================================================== +# 📈 信号判定模块（**方向已反转**） +# =============================================================== + +def is_bullish(c): return float(c['open']) < float(c['close']) + + +def is_bearish(c): return float(c['open']) > float(c['close']) + + +def check_signal(prev, curr): + """ + 判断是否出现包住形态，并返回（direction, signal_key） + 注意：根据用户要求，这里把原来的方向映射“反转”： + - 如果是 前跌后涨包住（过去我们认为做多），现在**改为做空** + - 如果是 前涨后跌包住（过去我们认为做空），现在**改为做多** + signal_key 保持 "bear_bull_engulf" / "bull_bear_engulf" 以便统计区分信号种类 + """ + p_open, p_close = float(prev['open']), float(prev['close']) + c_open, c_close = float(curr['open']), float(curr['close']) + + # 前跌后涨包住 -> **原来做多，现在改为做空** + if is_bullish(curr) and is_bearish(prev) and c_open <= p_close and c_close >= p_open: + return "short", "bear_bull_engulf" + + # 前涨后跌包住 -> **原来做空，现在改为做多** + if is_bearish(curr) and is_bullish(prev) and c_open >= p_close and c_close <= p_open: + return "long", "bull_bear_engulf" + + return None, None + + +# =============================================================== +# 💹 回测模拟模块（使用 1 分钟数据） +# =============================================================== + +def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl=-1): + """ + 用 1 分钟数据进行精细化止盈止损模拟 + entry_time: 当前信号的 entry candle id（毫秒时间戳） + next_15min_time: 下一个15min时间戳，用于界定止盈止损分析范围 + direction：'long' 或 'short' + entry_price：开仓价格 + """ + # 查 15 分钟之间的 1 分钟数据 + future_candles = get_future_data_1min(entry_time, next_15min_time) + if not future_candles: + return None, 0, None + + # 止盈止损价格（tp 为正，sl 为负） + tp_price = entry_price + tp if direction == "long" else entry_price - tp + sl_price = entry_price + sl if direction == "long" else entry_price - sl + + for candle in future_candles: + open_p, high, low = map(float, (candle['open'], candle['high'], candle['low'])) + + if direction == "long": + # 开盘跳空优先 + if open_p >= tp_price: + return open_p, open_p - entry_price, candle['id'] + if open_p <= sl_price: + return open_p, open_p - entry_price, candle['id'] + # 盘中触及 + if high >= tp_price: + return tp_price, tp, candle['id'] + if low <= sl_price: + return sl_price, sl, candle['id'] + + else: # short + if open_p <= tp_price: + return open_p, entry_price - open_p, candle['id'] + if open_p >= sl_price: + return open_p, entry_price - open_p, candle['id'] + if low <= tp_price: + return tp_price, tp, candle['id'] + if high >= sl_price: + return sl_price, sl, candle['id'] + + # 未触及 TP/SL，用最后一根收盘价平仓 + final = future_candles[-1] + final_price = float(final['close']) + diff = (final_price - entry_price) if direction == "long" else (entry_price - final_price) + return final_price, diff, final['id'] + + +# =============================================================== +# 📊 主回测流程（含单笔持仓逻辑） +# =============================================================== + +def backtest_single_position(dates: List[str], tp: float, sl: float): + """单笔持仓回测，保证任意时刻只有一笔持仓""" + all_data = [] + for date_str in dates: + all_data.extend(get_data_by_date(Weex15, date_str)) + all_data.sort(key=lambda x: x['id']) + + stats = { + "bear_bull_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "涨包跌（bear_bull_engulf）"}, + "bull_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包涨（bull_bear_engulf）"}, + } + + trades = [] + current_position = None # 当前持仓信息（字典或 None） + + for idx in range(1, len(all_data) - 1): + prev, curr = all_data[idx - 1], all_data[idx] + entry_candle = all_data[idx + 1] + + direction, signal = check_signal(prev, curr) + if not direction: + continue + + # 下一个 15 分钟K线的时间范围（若长度不足则取到最后） + next_15min_time = all_data[idx + 50]['id'] if idx + 50 < len(all_data) else all_data[-1]['id'] + entry_price = float(entry_candle['open']) + + # 如果有持仓 + if current_position: + # 同向信号 -> 跳过不开（继续等待当前持仓结束） + if current_position['direction'] == direction: + continue + # 反向信号 -> 先按当前持仓的规则用 simulate_trade 平仓，然后再开新仓 + else: + exit_price, diff, exit_time = simulate_trade( + current_position['direction'], + current_position['entry_price'], + current_position['entry_time'], + entry_candle['id'], + tp=tp, + sl=sl + ) + if exit_price is not None: + # 记录平仓交易（一笔持仓作为一条记录） + trades.append({ + "entry_time": datetime.datetime.fromtimestamp(current_position['entry_time'] / 1000), + "exit_time": datetime.datetime.fromtimestamp(exit_time / 1000), + "signal": current_position['signal'], + "direction": "做多" if current_position['direction'] == "long" else "做空", + "entry": current_position['entry_price'], + "exit": exit_price, + "diff": diff + }) + # 更新统计（signal 字段是中文名） + stats_key = 'bear_bull_engulf' if current_position['signal'].startswith('涨包跌') else 'bull_bear_engulf' + stats[stats_key]['count'] += 1 + stats[stats_key]['total_profit'] += diff + if diff > 0: + stats[stats_key]['wins'] += 1 + + # 清空当前持仓，接下来在下方开新仓 + current_position = None + + # 开新仓（无论之前是否有仓位，若有仓位已被平掉现在可以开仓） + current_position = { + "direction": direction, + "signal": stats[signal]['name'], + "entry_price": entry_price, + "entry_time": entry_candle['id'] + } + + # 最后一笔持仓如果未平仓，用最后收盘价平掉（以全数据范围最后时间为结算界限） + if current_position: + exit_price, diff, exit_time = simulate_trade( + current_position['direction'], + current_position['entry_price'], + current_position['entry_time'], + all_data[-1]['id'], + tp=tp, + sl=sl + ) + if exit_price is not None: + trades.append({ + "entry_time": datetime.datetime.fromtimestamp(current_position['entry_time'] / 1000), + "exit_time": datetime.datetime.fromtimestamp(exit_time / 1000), + "signal": current_position['signal'], + "direction": "做多" if current_position['direction'] == "long" else "做空", + "entry": current_position['entry_price'], + "exit": exit_price, + "diff": diff + }) + stats_key = 'bear_bull_engulf' if current_position['signal'].startswith('涨包跌') else 'bull_bear_engulf' + stats[stats_key]['count'] += 1 + stats[stats_key]['total_profit'] += diff + if diff > 0: + stats[stats_key]['wins'] += 1 + + return trades, stats + + +# =============================================================== +# 🚀 启动主流程（示例） +# =============================================================== +if __name__ == '__main__': + # 修正日期格式为 YYYY-MM-DD（零填充） + dates = [f"2025-07-{i:02d}" for i in range(1, 31)] + + # 示例参数：止盈 10 点，止损 -50 点（保留你的原调用方式） + trades, stats = backtest_single_position(dates, tp=10, sl=-50) + + logger.info("===== 每笔交易详情 =====") + for t in trades: + logger.info( + f"{t['entry_time']} {t['direction']}({t['signal']}) " + f"入场={t['entry']:.2f} 出场={t['exit']:.2f} 出场时间={t['exit_time']} " + f"差价={t['diff']:.2f}" + ) + + total_profit = sum(t['diff'] / t['entry'] * 10000 for t in trades) if trades else 0.0 + total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades) if trades else 0.0 + + print(f"\n一共交易笔数：{len(trades)}") + print(f"一共盈利（按手数/点值换算示例）：{total_profit:.2f}") + print(f"一共手续费：{total_fee:.2f}") + print(f"净利润：{total_profit - total_fee:.2f}") + print("\n===== 信号统计 =====") + for k, v in stats.items(): + print(f"{v['name']} -> 笔数: {v['count']}, 胜数: {v['wins']}, 总点差: {v['total_profit']:.2f}") diff --git a/weex/读取数据库分析数据2.0-优化信号版.py b/weex/读取数据库分析数据2.0-优化信号版.py index e4f0ac9..acd94ba 100644 --- a/weex/读取数据库分析数据2.0-优化信号版.py +++ b/weex/读取数据库分析数据2.0-优化信号版.py @@ -365,9 +365,9 @@ def backtest_single_position(dates, tp, sl): # =============================================================== if __name__ == '__main__': - dates = [f"2025-9-{i}" for i in range(1, 31)] + dates = [f"2025-6-{i}" for i in range(1, 31)] - trades, stats = backtest_single_position(dates, tp=10000, sl=-10) + trades, stats = backtest_single_position(dates, tp=50, sl=-10) logger.info("===== 每笔交易详情 =====") for t in trades: @@ -379,7 +379,6 @@ if __name__ == '__main__': total_profit = sum(t['diff'] / t['entry'] * 10000 for t in trades) total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades) - # print(f"止盈：{i1}, 止损：{i}") print(f"\n一共交易笔数：{len(trades)}") print(f"一共盈利：{total_profit:.2f}") diff --git a/weex/读取数据库分析数据2.0-优化信号版2.0.py b/weex/读取数据库分析数据2.0-优化信号版2.0.py index 5871f02..8b08d5a 100644 --- a/weex/读取数据库分析数据2.0-优化信号版2.0.py +++ b/weex/读取数据库分析数据2.0-优化信号版2.0.py @@ -30,11 +30,21 @@ except Exception as _e: class BacktestConfig: """回测配置类""" # 交易参数（注意：tp/sl单位与K线报价同单位，diff = 价格差值） - take_profit: float = 8.0 # 止盈点数（价格差） - stop_loss: float = -1.0 # 止损点数（价格差，负数） + take_profit: float = 3.0 # 止盈点数（价格差）- 优化：从8.0缩小到3.0 + stop_loss: float = -2.0 # 止损点数（价格差，负数）- 优化：从-1.0调整到-2.0 contract_size: float = 10000 # 合约规模（每1点价格差对应的盈亏金额 = diff * contract_size） - open_fee: float = 5.0 # 开仓固定手续费（金额） - close_fee_rate: float = 0.0005 # 平仓按成交额比例的手续费（金额 = rate * exit_price * contract_size） + open_fee: float = 3.0 # 开仓固定手续费（金额）- 优化：从5.0降低到3.0 + close_fee_rate: float = 0.0003 # 平仓按成交额比例的手续费（金额 = rate * exit_price * contract_size）- 优化：从0.0005降低到0.0003 + + # 新增：动态止盈止损参数 + dynamic_tp: bool = True # 是否启用动态止盈 + dynamic_sl: bool = True # 是否启用动态止损 + tp_ratio: float = 1.5 # 动态止盈倍数（基于ATR或价格波动） + sl_ratio: float = 1.0 # 动态止损倍数（基于ATR或价格波动） + min_tp: float = 2.0 # 最小止盈点数 + max_tp: float = 8.0 # 最大止盈点数 + min_sl: float = -3.0 # 最小止损点数（负数） + max_sl: float = -1.0 # 最大止损点数（负数） # 回测日期范围（仅示例：若不传dates则使用） start_date: str = "2025-07-01" @@ -44,12 +54,25 @@ class BacktestConfig: enable_bear_bull_engulf: bool = True # 涨包跌信号（前跌后涨包住 -> 做多） enable_bull_bear_engulf: bool = True # 跌包涨信号（前涨后跌包住 -> 做空） + # 新增：信号过滤参数 + min_engulf_ratio: float = 0.6 # 最小包住比例（当前K线实体与前一K线实体的比例） + min_volume_ratio: float = 1.2 # 最小成交量比例（当前K线成交量与前一K线成交量的比例） + enable_volume_filter: bool = False # 是否启用成交量过滤（需要数据支持） + def __post_init__(self): """验证配置参数""" if self.take_profit <= 0: raise ValueError("止盈点数必须大于0") if self.stop_loss >= 0: raise ValueError("止损点数必须小于0") + if self.min_tp <= 0 or self.max_tp <= 0: + raise ValueError("最小/最大止盈点数必须大于0") + if self.min_sl >= 0 or self.max_sl >= 0: + raise ValueError("最小/最大止损点数必须小于0") + if self.tp_ratio <= 0 or self.sl_ratio <= 0: + raise ValueError("动态止盈止损倍数必须大于0") + if self.min_engulf_ratio <= 0 or self.min_engulf_ratio > 1: + raise ValueError("最小包住比例必须在(0,1]范围内") # =============================================================== @@ -101,22 +124,43 @@ def is_bearish(c) -> bool: return float(c['open']) > float(c['close']) -def check_signal(prev: Dict[str, Any], curr: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]: +def check_signal(prev: Dict[str, Any], curr: Dict[str, Any], config: BacktestConfig = None) -> Tuple[Optional[str], Optional[str]]: """ - 判断是否出现包住形态 + 判断是否出现包住形态（优化版） 返回 (方向direction, 信号键signal_key) signal_key: 'bear_bull_engulf' 表示 涨包跌（前跌后涨包住 -> 做多） 'bull_bear_engulf' 表示 跌包涨（前涨后跌包住 -> 做空） """ + if config is None: + config = BacktestConfig() + p_open, p_close = float(prev['open']), float(prev['close']) c_open, c_close = float(curr['open']), float(curr['close']) + # 计算实体大小 + prev_body = abs(p_close - p_open) + curr_body = abs(c_close - c_open) + + # 过滤：当前K线实体必须足够大 + if curr_body < 0.5: # 实体太小，忽略 + return None, None + # 前跌后涨包住 -> 做多 if is_bullish(curr) and is_bearish(prev) and c_open <= p_close and c_close >= p_open: + # 新增：包住比例过滤 + if prev_body > 0: + engulf_ratio = curr_body / prev_body + if engulf_ratio < config.min_engulf_ratio: + return None, None return "long", "bear_bull_engulf" # 前涨后跌包住 -> 做空 if is_bearish(curr) and is_bullish(prev) and c_open >= p_close and c_close <= p_open: + # 新增：包住比例过滤 + if prev_body > 0: + engulf_ratio = curr_body / prev_body + if engulf_ratio < config.min_engulf_ratio: + return None, None return "short", "bull_bear_engulf" return None, None @@ -126,33 +170,87 @@ def check_signal(prev: Dict[str, Any], curr: Dict[str, Any]) -> Tuple[Optional[s # 💹 回测模拟模块（使用 1 分钟数据） # =============================================================== +def calculate_dynamic_tp_sl( + entry_price: float, + recent_candles: List[Dict[str, Any]], + config: BacktestConfig +) -> Tuple[float, float]: + """ + 计算动态止盈止损 + 基于最近K线的波动幅度 + """ + if len(recent_candles) < 5: + return config.take_profit, config.stop_loss + + # 计算ATR（平均真实波幅） + true_ranges = [] + for i in range(1, min(len(recent_candles), 10)): + prev_close = float(recent_candles[i-1]['close']) + curr_high = float(recent_candles[i]['high']) + curr_low = float(recent_candles[i]['low']) + + tr = max( + curr_high - curr_low, + abs(curr_high - prev_close), + abs(curr_low - prev_close) + ) + true_ranges.append(tr) + + if not true_ranges: + return config.take_profit, config.stop_loss + + atr = sum(true_ranges) / len(true_ranges) + + # 动态计算止盈止损 + dynamic_tp = max(config.min_tp, min(config.max_tp, atr * config.tp_ratio)) + dynamic_sl = max(config.min_sl, min(config.max_sl, -atr * config.sl_ratio)) + + return dynamic_tp, dynamic_sl + + def simulate_trade( direction: str, entry_price: float, start_ts: int, end_ts: int, tp: float, - sl: float + sl: float, + config: BacktestConfig = None ) -> Tuple[Optional[float], float, Optional[int]]: """ - 用 1 分钟数据进行精细化止盈止损模拟 + 用 1 分钟数据进行精细化止盈止损模拟（优化版） - direction: "long" / "short" - entry_price: 开仓价格 - start_ts: 开始时间（毫秒时间戳，包含） - end_ts: 结束时间（毫秒时间戳，包含） - tp: 止盈点（价格差，正数） - sl: 止损点（价格差，负数） + - config: 配置对象（用于动态止盈止损） 返回 (exit_price, diff, exit_time) - exit_price: 平仓价格，None 表示区间内无数据 - diff: 价格差 = 对多头 (exit - entry)，对空头 (entry - exit) - exit_time: 平仓发生时间戳 """ + if config is None: + config = BacktestConfig() future_candles = get_future_data_1min(start_ts, end_ts) if not future_candles: logger.warning(f"simulate_trade 无分钟数据: [{start_ts}, {end_ts}]，跳过该段模拟。") return None, 0.0, None + # 动态止盈止损计算 + if config.dynamic_tp or config.dynamic_sl: + # 获取最近的历史K线用于计算ATR + history_start = start_ts - 15 * 60 * 1000 * 10 # 获取前10根15分钟K线 + history_candles = get_future_data_1min(history_start, start_ts - 1) + if history_candles: + dynamic_tp, dynamic_sl = calculate_dynamic_tp_sl(entry_price, history_candles, config) + if config.dynamic_tp: + tp = dynamic_tp + if config.dynamic_sl: + sl = dynamic_sl + if direction == "long": tp_price = entry_price + tp sl_price = entry_price + sl # sl为负 @@ -227,7 +325,7 @@ def backtest( entry_candle = all_data[idx + 1] next_of_entry = all_data[idx + 2] # 用于界定入场K的结束边界 - direction, signal_key = check_signal(prev, curr) + direction, signal_key = check_signal(prev, curr, config) if not direction or signal_key is None: continue @@ -246,7 +344,8 @@ def backtest( start_ts=entry_ts, end_ts=end_ts, tp=tp, - sl=sl + sl=sl, + config=config ) if exit_price is None: @@ -289,7 +388,7 @@ def backtest_single_position( tp: float, sl: float, config: BacktestConfig = BacktestConfig() -) -> Tuple[List[Dict[str, Any]], Dict[str, Dict[str, Any]]]: +) -> Tuple[List[Dict[str, Any]], Dict[str, Dict[str, Any]], List[Dict[str, Any]]]: """ 单笔持仓回测： - 同向信号：忽略（不加仓） @@ -308,13 +407,15 @@ def backtest_single_position( } trades: List[Dict[str, Any]] = [] + # 事件流：用于可视化显示“同向信号（忽略）”与“反向先平后开” + signal_events: List[Dict[str, Any]] = [] current_position: Optional[Dict[str, Any]] = None # 为了在反向信号到来时平仓，至少需要 prev(0), curr(1), entry(idx+1=2) for idx in range(1, len(all_data) - 1): prev, curr = all_data[idx - 1], all_data[idx] entry_candle = all_data[idx + 1] # 新信号的入场K - direction, signal_key = check_signal(prev, curr) + direction, signal_key = check_signal(prev, curr, config) if not direction or signal_key is None: continue @@ -329,6 +430,15 @@ def backtest_single_position( if current_position is not None: # 同向 -> 忽略 if current_position['direction'] == direction: + # 记录同向信号事件（忽略加仓） + signal_events.append({ + "type": "same_direction_signal", + "time": datetime.datetime.fromtimestamp(entry_ts_new / 1000), + "price": entry_price_new, + "direction": direction, + "signal_key": signal_key, + "signal_name": stats[signal_key]['name'] + }) continue # 反向 -> 先平旧仓（分钟模拟：从旧entry_time -> 新入场时刻-1毫秒） exit_price, diff, exit_time = simulate_trade( @@ -337,7 +447,8 @@ def backtest_single_position( start_ts=current_position['entry_time'], end_ts=entry_ts_new - 1, tp=tp, - sl=sl + sl=sl, + config=config ) if exit_price is not None: profit_amount = diff * config.contract_size @@ -363,6 +474,16 @@ def backtest_single_position( if net_profit > 0: stats[current_position['signal_key']]['wins'] += 1 + # 记录反向信号引发的平仓事件 + signal_events.append({ + "type": "reverse_close", + "time": datetime.datetime.fromtimestamp(exit_time / 1000), + "price": exit_price, + "direction": current_position['direction'], + "signal_key": current_position['signal_key'], + "signal_name": current_position['signal_name'] + }) + # 清空持仓，再开新仓 current_position = None @@ -375,6 +496,16 @@ def backtest_single_position( "entry_time": entry_ts_new } + # 记录反向开仓事件（若此前刚刚发生了反向平仓，则此处为接力开仓）或普通开仓事件 + signal_events.append({ + "type": "reverse_open" if any(e.get("type") == "reverse_close" and e.get("time") == datetime.datetime.fromtimestamp((entry_ts_new - 1) / 1000) for e in signal_events) else "signal_open", + "time": datetime.datetime.fromtimestamp(entry_ts_new / 1000), + "price": entry_price_new, + "direction": direction, + "signal_key": signal_key, + "signal_name": stats[signal_key]['name'] + }) + # 数据末尾：若仍有持仓，用最后时间收盘价平仓（分钟模拟：从entry_time -> 全部数据最后时刻） if current_position is not None and len(all_data) > 0: final_ts = all_data[-1]['id'] @@ -384,7 +515,8 @@ def backtest_single_position( start_ts=current_position['entry_time'], end_ts=final_ts, tp=tp, - sl=sl + sl=sl, + config=config ) if exit_price is not None: profit_amount = diff * config.contract_size @@ -410,7 +542,7 @@ def backtest_single_position( if net_profit > 0: stats[current_position['signal_key']]['wins'] += 1 - return trades, stats + return trades, stats, signal_events # =============================================================== @@ -429,6 +561,130 @@ def gen_dates(start_date: str, end_date: str) -> List[str]: return res +# =============================================================== +# 🔍 参数优化建议 +# =============================================================== + +def optimize_parameters( + dates: List[str], + config: BacktestConfig, + tp_range: Tuple[float, float] = (1.0, 6.0), + sl_range: Tuple[float, float] = (-4.0, -1.0), + step: float = 0.5 +) -> Dict[str, Any]: + """ + 参数优化函数：测试不同的止盈止损组合，找到最佳参数 + + Args: + dates: 回测日期列表 + config: 基础配置 + tp_range: 止盈范围 (min, max) + sl_range: 止损范围 (min, max) + step: 步长 + + Returns: + 最佳参数组合和统计结果 + """ + logger.info("开始参数优化...") + + best_result = { + 'tp': config.take_profit, + 'sl': config.stop_loss, + 'total_net_profit': -float('inf'), + 'win_rate': 0.0, + 'total_trades': 0, + 'sharpe_ratio': -float('inf') + } + + results = [] + + # 生成参数组合 + tp_values = [round(tp_range[0] + i * step, 1) for i in range(int((tp_range[1] - tp_range[0]) / step) + 1)] + sl_values = [round(sl_range[0] + i * step, 1) for i in range(int((sl_range[1] - sl_range[0]) / step) + 1)] + + total_combinations = len(tp_values) * len(sl_values) + current_combination = 0 + + for tp in tp_values: + for sl in sl_values: + if sl >= 0: # 止损必须是负数 + continue + + current_combination += 1 + logger.info(f"测试参数组合 {current_combination}/{total_combinations}: TP={tp}, SL={sl}") + + # 创建临时配置 + temp_config = BacktestConfig( + take_profit=tp, + stop_loss=sl, + contract_size=config.contract_size, + open_fee=config.open_fee, + close_fee_rate=config.close_fee_rate, + enable_bear_bull_engulf=config.enable_bear_bull_engulf, + enable_bull_bear_engulf=config.enable_bull_bear_engulf, + min_engulf_ratio=config.min_engulf_ratio, + dynamic_tp=False, # 关闭动态参数以进行基础测试 + dynamic_sl=False + ) + + try: + # 运行回测（接收 signal_events，但优化统计仅需 trades） + trades, stats, _signal_events = backtest_single_position(dates, tp=tp, sl=sl, config=temp_config) + + if not trades: + continue + + # 计算统计指标 + total_trades = len(trades) + total_net_profit = sum(t['net_profit'] for t in trades) + winning_trades = sum(1 for t in trades if t['net_profit'] > 0) + win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0 + + # 计算夏普比率（简化版） + profits = [t['net_profit'] for t in trades] + if len(profits) > 1: + avg_profit = sum(profits) / len(profits) + profit_std = (sum((p - avg_profit) ** 2 for p in profits) / len(profits)) ** 0.5 + sharpe_ratio = (avg_profit / profit_std) if profit_std > 0 else 0 + else: + sharpe_ratio = 0 + + result = { + 'tp': tp, + 'sl': sl, + 'total_net_profit': total_net_profit, + 'win_rate': win_rate, + 'total_trades': total_trades, + 'sharpe_ratio': sharpe_ratio + } + results.append(result) + + # 更新最佳结果（优先考虑净利润，其次考虑夏普比率） + if (total_net_profit > best_result['total_net_profit'] or + (total_net_profit == best_result['total_net_profit'] and sharpe_ratio > best_result['sharpe_ratio'])): + best_result = result.copy() + + except Exception as e: + logger.warning(f"参数组合 TP={tp}, SL={sl} 测试失败: {e}") + continue + + logger.info(f"参数优化完成！最佳组合: TP={best_result['tp']}, SL={best_result['sl']}") + logger.info(f"最佳结果: 净利润={best_result['total_net_profit']:.2f}, 胜率={best_result['win_rate']:.2f}%, 交易次数={best_result['total_trades']}") + + return { + 'best_result': best_result, + 'all_results': results, + 'optimization_summary': { + 'total_combinations_tested': len(results), + 'best_tp': best_result['tp'], + 'best_sl': best_result['sl'], + 'best_net_profit': best_result['total_net_profit'], + 'best_win_rate': best_result['win_rate'], + 'best_sharpe_ratio': best_result['sharpe_ratio'] + } + } + + # =============================================================== # 📈 可视化：K线 + 交易点位（交互式HTML） # =============================================================== @@ -445,6 +701,7 @@ def _collect_15m_candles(dates: List[str]) -> List[Dict[str, Any]]: def plot_trades_candlestick( dates: List[str], trades: List[Dict[str, Any]], + signal_events: Optional[List[Dict[str, Any]]] = None, output_html: str = "backtest_trades.html", title: str = "回测交易可视化" ) -> Optional[str]: @@ -545,6 +802,44 @@ def plot_trades_candlestick( showlegend=False )) + # 叠加信号事件（同向忽略、反向平仓、反向开仓、普通开仓） + if signal_events: + for ev in signal_events: + ev_type = ev.get("type") + ev_time: datetime.datetime = ev.get("time") + ev_price = float(ev.get("price", 0.0)) + ev_signal_name = str(ev.get("signal_name", "")) + ev_direction = ev.get("direction") # "long" / "short" + + if ev_type == "same_direction_signal": + symbol = "circle-open" + color = "#95a5a6" # 灰色 + name = "同向信号(忽略)" + elif ev_type == "reverse_close": + symbol = "x-thin" + color = "#f39c12" # 橙色 + name = "反向平仓" + elif ev_type == "reverse_open": + symbol = "triangle-up-open" if ev_direction == "long" else "triangle-down-open" + color = "#f1c40f" # 黄色 + name = "反向开仓" + else: # signal_open + symbol = "star" + color = "#3498db" # 蓝色 + name = "信号开仓" + + fig.add_trace(go.Scatter( + x=[ev_time], y=[ev_price], + mode="markers", + marker=dict(symbol=symbol, size=9, color=color, line=dict(width=1, color="#ffffff")), + name=name, + hovertemplate=( + "事件:%{customdata[0]}
信号:%{customdata[1]}
价格:%{y:.2f}
时间:%{x}" + ), + customdata=[[name, ev_signal_name]], + showlegend=False + )) + # 全局布局与交互 fig.update_layout( title=title, @@ -615,13 +910,30 @@ def plot_trades_candlestick( # =============================================================== if __name__ == '__main__': - # 你可以根据需要修改此配置 + # 优化后的配置 - 缩小价差距离 config = BacktestConfig( - take_profit=10.0, - stop_loss=-5.0, + # 基础止盈止损参数（已优化） + take_profit=3.0, # 从10.0缩小到3.0 + stop_loss=-2.0, # 从-5.0调整到-2.0 contract_size=10000, - open_fee=5.0, - close_fee_rate=0.0005, + open_fee=3.0, # 从5.0降低到3.0 + close_fee_rate=0.0003, # 从0.0005降低到0.0003 + + # 动态止盈止损参数 + dynamic_tp=True, # 启用动态止盈 + dynamic_sl=True, # 启用动态止损 + tp_ratio=1.5, # 动态止盈倍数 + sl_ratio=1.0, # 动态止损倍数 + min_tp=2.0, # 最小止盈点数 + max_tp=8.0, # 最大止盈点数 + min_sl=-3.0, # 最小止损点数 + max_sl=-1.0, # 最大止损点数 + + # 信号过滤参数 + min_engulf_ratio=0.6, # 最小包住比例 + enable_volume_filter=False, # 暂不启用成交量过滤 + + # 回测日期 start_date="2025-09-01", end_date="2025-09-30", enable_bear_bull_engulf=True, @@ -631,8 +943,26 @@ if __name__ == '__main__': # 生成日期列表（也可以自行传入 dates） dates = gen_dates(config.start_date, config.end_date) - # 运行“单笔持仓”模式（同向忽略，反向先平后开） - trades, stats = backtest_single_position(dates, tp=config.take_profit, sl=config.stop_loss, config=config) + # 可选：运行参数优化（取消注释以启用） + # optimization_result = optimize_parameters( + # dates=dates, + # config=config, + # tp_range=(1.0, 6.0), # 止盈范围 + # sl_range=(-4.0, -1.0), # 止损范围 + # step=0.5 # 步长 + # ) + # print(f"\n=== 参数优化结果 ===") + # print(f"最佳止盈: {optimization_result['optimization_summary']['best_tp']}") + # print(f"最佳止损: {optimization_result['optimization_summary']['best_sl']}") + # print(f"最佳净利润: {optimization_result['optimization_summary']['best_net_profit']:.2f}") + # print(f"最佳胜率: {optimization_result['optimization_summary']['best_win_rate']:.2f}%") + # + # # 使用优化后的参数更新配置 + # config.take_profit = optimization_result['optimization_summary']['best_tp'] + # config.stop_loss = optimization_result['optimization_summary']['best_sl'] + + # 运行"单笔持仓"模式（同向忽略，反向先平后开） + trades, stats, signal_events = backtest_single_position(dates, tp=config.take_profit, sl=config.stop_loss, config=config) logger.info("===== 每笔交易详情 =====") for t in trades: @@ -665,6 +995,17 @@ if __name__ == '__main__': # ... 输出逻辑同上 # 生成交互式可视化 - html_path = plot_trades_candlestick(dates, trades, output_html="backtest_trades.html") + html_path = plot_trades_candlestick(dates, trades, signal_events=signal_events, output_html="backtest_trades.html") if html_path: - print(f"\n可视化文件已生成：{html_path}\n请用浏览器打开进行交互分析。") \ No newline at end of file + print(f"\n可视化文件已生成：{html_path}\n请用浏览器打开进行交互分析。") + + # 输出优化建议 + print(f"\n=== 优化建议 ===") + print(f"当前配置已优化：") + print(f"- 止盈点数: {config.take_profit} (从原来的10.0缩小到{config.take_profit})") + print(f"- 止损点数: {config.stop_loss} (从原来的-5.0调整到{config.stop_loss})") + print(f"- 开仓手续费: {config.open_fee} (从原来的5.0降低到{config.open_fee})") + print(f"- 平仓手续费率: {config.close_fee_rate} (从原来的0.0005降低到{config.close_fee_rate})") + print(f"- 最小包住比例: {config.min_engulf_ratio} (新增过滤条件)") + print(f"- 动态止盈止损: {'启用' if config.dynamic_tp or config.dynamic_sl else '禁用'}") + print(f"\n如需进一步优化，请取消注释参数优化代码段。") \ No newline at end of file