dededdew

weex/优化开仓方向版本.py

@@ -0,0 +1,398 @@
+"""
+量化交易回测系统 - 优化版
+功能：基于包住形态的交易信号识别和回测分析
+作者：量化交易团队
+版本：2.0
+"""
+
+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-7-1"
+    end_date: str = "2025-7-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):
+    """按天获取指定表的数据"""
+    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):
+    """判断是否出现包住形态"""
+    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 "long", "bear_bull_engulf"
+
+    # 前涨后跌包住 -> 做空
+    if is_bearish(curr) and is_bullish(prev) and c_open >= p_close and c_close <= p_open:
+        return "short", "bull_bear_engulf"
+
+    return None, None
+
+
+# ===============================================================
+# 💹 回测模拟模块（使用 1 分钟数据）
+# ===============================================================
+
+# ---------- 替换后的 simulate_trade ----------
+def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl=-1):
+    """
+    返回 (exit_price, profit_diff, exit_time, exit_reason)
+    exit_reason: 'tp' (触及止盈并以tp价平仓),
+                 'sl' (触及止损并以sl价平仓),
+                 'open_tp' (开盘跳空并以开盘价止盈),
+                 'open_sl' (开盘跳空并以开盘价止损),
+                 'timeout' (到分析窗口末尾，用最后一根收盘价平仓)
+    注意：sl 参数为负数（你的设定），但在计算止损价时已处理方向
+    """
+    future_candles = get_future_data_1min(entry_time, next_15min_time)
+    if not future_candles:
+        return None, 0.0, None, None
+
+    # 计算目标价位（数值）
+    tp_price = entry_price + tp if direction == "long" else entry_price - tp
+    sl_price = entry_price + sl if direction == "long" else entry_price - sl  # sl 为负数，long 情况为 entry + (负数) -> 小于 entry
+
+    for candle in future_candles:
+        open_p = float(candle['open'])
+        high = float(candle['high'])
+        low = float(candle['low'])
+        ts = candle['id']
+
+        if direction == "long":
+            # 开盘跳空（以开盘价直接触及止盈/止损）
+            if open_p >= tp_price:
+                return open_p, open_p - entry_price, ts, 'open_tp'
+            if open_p <= sl_price:
+                return open_p, open_p - entry_price, ts, 'open_sl'
+            # 盘中触及
+            if high >= tp_price:
+                return tp_price, tp, ts, 'tp'
+            if low <= sl_price:
+                return sl_price, sl, ts, 'sl'
+        else:  # short
+            if open_p <= tp_price:
+                return open_p, entry_price - open_p, ts, 'open_tp'
+            if open_p >= sl_price:
+                return open_p, entry_price - open_p, ts, 'open_sl'
+            if low <= tp_price:
+                return tp_price, tp, ts, 'tp'
+            if high >= sl_price:
+                return sl_price, sl, ts, 'sl'
+
+    # 未触发止盈止损，用最后一根收盘价平仓（视为 timeout）
+    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'], 'timeout'
+
+
+# ---------- 替换后的 backtest_single_position ----------
+def backtest_single_position(dates, tp, sl):
+    """
+    单笔持仓回测（增强版），加入连续3次止损触发反向开仓（转向单）逻辑
+    - 只有当 exit_reason 属于 'sl' 或 'open_sl' 时才算“真实止损”
+    - 当连续真实止损计数达到 3 时，**下一笔**信号反向开仓（且该转向单不计入连续止损统计）
+    """
+    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": "涨包跌"},
+        "bull_bear_engulf": {"count": 0, "wins": 0, "total_profit": 0, "name": "跌包涨"},
+    }
+
+    trades = []
+    current_position = None  # 当前持仓信息 dict or None
+
+    consec_sl_count = 0          # 连续真实止损计数（只有 exit_reason 为 'sl' 或 'open_sl' 时 +1）
+    reverse_next_signal = False  # 是否需要将下一笔信号取反（由连续3次止损触发）
+    ignore_next_result = False   # 标记下一笔成交的结果是否要被忽略（用于转向单：转向单不计入连续止损计数，也不影响计数）
+
+    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
+
+        # 如果需要反向下一笔信号（由之前三次止损触发），则翻转 direction 并标记为转向单（只对这一次有效）
+        is_reversal_trade = False
+        if reverse_next_signal:
+            direction = 'long' if direction == 'short' else 'short'
+            is_reversal_trade = True
+            reverse_next_signal = False
+            ignore_next_result = True  # 这笔成交的平仓结果不影响 consec_sl_count
+
+        # 下一个 15 分钟K线的时间范围（用 idx+50 作为 15min 窗口近似）
+        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
+            # 反向信号 -> 先按当前位置止盈止损平仓，再根据规则决定是否开新仓
+            else:
+                exit_price, diff, exit_time, exit_reason = 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,
+                        "exit_reason": exit_reason,
+                        "is_reversal_trade": current_position.get('is_reversal_trade', False)
+                    })
+                    # 更新统计（只有非转向单会计入统计？按你原逻辑计入）
+                    stats_key = 'bear_bull_engulf' if current_position['signal'] == '涨包跌' else 'bull_bear_engulf'
+                    stats[stats_key]['count'] += 1
+                    stats[stats_key]['total_profit'] += diff
+                    if diff > 0:
+                        stats[stats_key]['wins'] += 1
+
+                    # 根据 exit_reason 更新 consec_sl_count（但如果当时该仓为被标记为 ignore_result，则不变）
+                    if not current_position.get('ignore_result', False):
+                        if exit_reason in ('sl', 'open_sl'):
+                            consec_sl_count += 1
+                        else:
+                            consec_sl_count = 0
+
+                        # 如果累计达到 3 次真实止损 -> 标记下一笔反向
+                        if consec_sl_count >= 3:
+                            reverse_next_signal = True
+                            consec_sl_count = 0  # 触发后清零（下一笔为转向单）
+                    else:
+                        # 如果这笔被标记为 ignore（通常是前面是转向单），则不影响计数
+                        pass
+
+                current_position = None  # 清空持仓
+
+        # 开新仓（注意：如果这笔是转向单，我们之前已经取反了 direction，并设置了 is_reversal_trade 和 ignore_next_result）
+        current_position = {
+            "direction": direction,
+            "signal": stats[signal]['name'],
+            "entry_price": entry_price,
+            "entry_time": entry_candle['id'],
+            "is_reversal_trade": is_reversal_trade,
+            "ignore_result": ignore_next_result  # 如果为 True，则本仓平仓结果不计入 consec_sl_count
+        }
+
+        # 如果我们标记了 ignore_next_result（说明这是转向单），只在刚开仓后清除标记，确保仅忽略这笔的**平仓**结果
+        if ignore_next_result:
+            # 清掉，只有这笔仓的平仓结果需要被忽略（记录在 current_position 中），
+            # 后续在处理平仓时会读取 current_position['ignore_result'] 并决定是否影响计数
+            ignore_next_result = False
+
+    # 最后一笔持仓如果未平仓，用最后收盘价平掉
+    if current_position:
+        exit_price, diff, exit_time, exit_reason = 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,
+                "exit_reason": exit_reason,
+                "is_reversal_trade": current_position.get('is_reversal_trade', False)
+            })
+            stats_key = 'bear_bull_engulf' if current_position['signal'] == '涨包跌' else 'bull_bear_engulf'
+            stats[stats_key]['count'] += 1
+            stats[stats_key]['total_profit'] += diff
+            if diff > 0:
+                stats[stats_key]['wins'] += 1
+
+            # 最后一笔是否计入连续止损计数
+            if not current_position.get('ignore_result', False):
+                if exit_reason in ('sl', 'open_sl'):
+                    consec_sl_count += 1
+                else:
+                    consec_sl_count = 0
+                if consec_sl_count >= 3:
+                    # 可选择记录或告警，这里仅重置计数
+                    consec_sl_count = 0
+
+    return trades, stats
+
+
+
+# ===============================================================
+# 🚀 启动主流程
+# ===============================================================
+
+if __name__ == '__main__':
+    dates = [f"2025-9-{i}" for i in range(1, 31)]
+
+    trades, stats = backtest_single_position(dates, tp=50, sl=-10)
+
+    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)
+    total_fee = sum(5 + 10000 / t['entry'] * t['exit'] * 0.0005 for t in trades)
+
+    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 i in range(1, 16):
+    #     for i1 in range(1, 51):
+    #         trades, stats = backtest_single_position(dates, tp=i1, sl=-i)
+    #
+    #         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)
+    #
+    #         if total_profit > total_fee * 0.1:
+    #             print("\n===== 信号统计 =====")
+    #             print(f"止盈：{i1}, 止损：{i}")
+    #             print(f"\n一共交易笔数：{len(trades)}")
+    #             print(f"一共盈利：{total_profit:.2f}")
+    #             print(f"一共手续费：{total_fee:.2f}")
+    #             print(f"净利润：{total_profit - total_fee * 0.1}")
+
+# 需要优化，目前有两种情况，第一种，同向，不如说上一单开单是涨，上一单还没有结束，当前信号来了，就不开单，等上一单到了上一单的止损位或者止盈位在平仓
+# 第二种，方向，上一单是涨，上一单还没有结束，当前信号来了，是跌，然后就按照现在这个信号要开仓的位置，平掉上一单，然后开一单方向的，
+# 一笔中可能有好几次信号，都按照上面的规则去判断，要保证同一时间，只会有一笔持仓，
+# 打印每笔的交易详细，如果一笔中同向，输入为一条交易记录