dededdew

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}")

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}")

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]}<br>信号:%{customdata[1]}<br>价格:%{y:.2f}<br>时间:%{x}<extra></extra>"
+                ),
+                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请用浏览器打开进行交互分析。")
+        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如需进一步优化，请取消注释参数优化代码段。")