dededdew

weex/读取数据库分析数据2.0-优化信号版.py

@@ -0,0 +1,258 @@
+"""
+量化交易回测系统 - 优化版（v2.1）
+功能：基于包住形态的交易信号识别和回测分析
+作者：量化交易团队
+"""
+
+import datetime
+from dataclasses import dataclass
+from loguru import logger
+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")
+
+
+# ===============================================================
+# 📊 数据模块
+# ===============================================================
+
+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分钟级止盈止损）
+# ===============================================================
+
+def simulate_trade(direction, entry_price, entry_time, next_15min_time, tp=8, sl=-1):
+    """用 1 分钟数据进行精细化止盈止损模拟"""
+    future_candles = get_future_data_1min(entry_time, next_15min_time)
+    if not future_candles:
+        return None, 0, 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
+
+    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']
+
+    # 未触发止盈止损，用最后一根收盘价平仓
+    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(dates, tp, sl):
+    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 = []
+    idx = 1
+    open_position = None
+
+    while idx < 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:
+            idx += 1
+            continue
+
+        # === 当前有持仓 ===
+        if open_position:
+            if direction == open_position['direction']:
+                # 同方向信号：忽略
+                idx += 1
+                continue
+            else:
+                # 反方向信号：立即平仓
+                exit_price = float(entry_candle['open'])
+                diff = (exit_price - open_position['entry_price']) if open_position['direction'] == 'long' else (
+                        open_position['entry_price'] - exit_price)
+                trades.append({
+                    "entry_time": datetime.datetime.fromtimestamp(open_position['entry_time'] / 1000),
+                    "exit_time": datetime.datetime.fromtimestamp(entry_candle['id'] / 1000),
+                    "signal": "反向平仓",
+                    "direction": "平仓",
+                    "entry": open_position['entry_price'],
+                    "exit": exit_price,
+                    "diff": diff
+                })
+                open_position = None  # 平仓后可立即反手
+
+        # === 开新仓 ===
+        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'])
+        exit_price, diff, exit_time = simulate_trade(
+            direction, entry_price, entry_candle['id'], next_15min_time, tp=tp, sl=sl
+        )
+        if exit_price is None:
+            idx += 1
+            continue
+
+        # 记录统计
+        stats[signal]['count'] += 1
+        stats[signal]['total_profit'] += diff
+        if diff > 0:
+            stats[signal]['wins'] += 1
+
+        trades.append({
+            "entry_time": datetime.datetime.fromtimestamp(entry_candle['id'] / 1000),
+            "exit_time": datetime.datetime.fromtimestamp(exit_time / 1000),
+            "signal": stats[signal]['name'],
+            "direction": "做多" if direction == "long" else "做空",
+            "entry": entry_price,
+            "exit": exit_price,
+            "diff": diff
+        })
+
+        # === 跳过到平仓时间点 ===
+        # 找到 exit_time 对应的 candle 索引，防止未平仓时重复触发信号
+        while idx < len(all_data) - 1 and all_data[idx]['id'] < exit_time:
+            idx += 1
+
+        open_position = None  # 已平仓
+        idx += 1
+
+    return trades, stats
+
+
+# ===============================================================
+# 🚀 主入口
+# ===============================================================
+
+if __name__ == '__main__':
+    dates = [f"2025-9-{i}" for i in range(1, 31)]
+
+    trades, stats = backtest(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 k, v in stats.items():
+        win_rate = (v['wins'] / v['count'] * 100) if v['count'] > 0 else 0
+        print(f"{v['name']} ({k}) - 信号数: {v['count']} | 胜率: {win_rate:.2f}% | 总盈利: {v['total_profit']:.2f}")