dededdew

weex/读取数据库分析数据.py

@@ -0,0 +1,227 @@
+import datetime
+
+from loguru import logger
+from models.weex import Weex
+
+
+def is_bullish(candle):
+    """判断是否是阳线（开盘价 < 收盘价，即涨）"""
+    return float(candle['open']) < float(candle['close'])
+
+
+def is_bearish(candle):
+    """判断是否是阴线（开盘价 > 收盘价，即跌）"""
+    return float(candle['open']) > float(candle['close'])
+
+
+def check_signal(prev, curr):
+    """
+    判断是否出现包住形态，返回信号类型和方向：
+    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：前跌后涨，且涨线包住前跌线 -> 做多信号
+    if is_bullish(curr) and is_bearish(prev):
+        if c_open <= p_close and c_close >= p_open:
+            return "long", "bear_bull_engulf"
+
+    # 情况2：前涨后跌，且跌线包住前涨线 -> 做空信号
+    if is_bearish(curr) and is_bullish(prev):
+        if c_open >= p_close and c_close <= p_open:
+            return "short", "bull_bear_engulf"
+
+    # # 情况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=30, stop_loss_diff=-10):
+    """
+    模拟交易（逐根K线回测）
+    改进版：考虑开盘跳空触发止盈/止损的情况
+    """
+    for candle in future_candles:
+        open_p = float(candle['open'])
+        high = float(candle['high'])
+        low = float(candle['low'])
+        close = float(candle['close'])
+
+        if direction == "long":
+            tp_price = entry_price + take_profit_diff
+            sl_price = entry_price + stop_loss_diff
+
+            # 🧩 开盘就跳空止盈
+            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, take_profit_diff, candle['id']
+            if low <= sl_price:
+                return sl_price, stop_loss_diff, candle['id']
+
+        elif direction == "short":
+            tp_price = entry_price - take_profit_diff
+            sl_price = entry_price - stop_loss_diff
+
+            # 🧩 开盘就跳空止盈
+            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, take_profit_diff, candle['id']
+            if high >= sl_price:
+                return sl_price, stop_loss_diff, candle['id']
+
+    # 未触发止盈止损，按最后收盘价平仓
+    final_price = float(future_candles[-1]['close'])
+    if direction == "long":
+        diff_money = final_price - entry_price
+    else:
+        diff_money = entry_price - final_price
+
+    return final_price, diff_money, future_candles[-1]['id']
+
+
+def get_data_by_date(date_str):
+    # 将日期字符串转换为 datetime 对象
+    try:
+        target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+    except ValueError:
+        print("日期格式不正确，请使用 YYYY-MM-DD 格式。")
+        return []
+
+    # 计算该天的开始时间戳（毫秒级）
+    start_timestamp = int(target_date.timestamp() * 1000)
+    # 计算该天的结束时间戳（毫秒级），即下一天 00:00:00 的时间戳减去 1 毫秒
+    end_timestamp = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1
+
+    # 查询该天的数据，并按照 id 字段从小到大排序
+    query = Weex.select().where(Weex.id.between(start_timestamp, end_timestamp)).order_by(Weex.id.asc())
+    results = list(query)
+
+    # 将结果转换为列表嵌套字典的形式
+    data_list = []
+    for item in results:
+        item_dict = {
+            'id': item.id,
+            'open': item.open,
+            'high': item.high,
+            'low': item.low,
+            'close': item.close
+        }
+        data_list.append(item_dict)
+
+    return data_list
+
+
+if __name__ == '__main__':
+
+    # 示例调用
+    datas = []
+    for i in range(1, 31):
+        date_str = f'2025-9-{i}'
+        data = get_data_by_date(date_str)
+
+        datas.extend(data)
+
+    datas = sorted(datas, key=lambda x: x["id"])
+
+    zh_project = 0  # 累计盈亏
+    all_trades = []  # 保存所有交易明细
+    daily_signals = 0  # 信号总数
+    daily_wins = 0
+    daily_profit = 0  # 价差总和
+
+    # 四种信号类型的统计
+    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": "跌包跌"}
+    }
+
+    # 遍历每根K线，寻找信号
+    for idx in range(1, len(datas) - 1):  # 留出未来K线
+        prev, curr = datas[idx - 1], datas[idx]  # 前一笔，当前一笔
+        entry_candle = datas[idx + 1]  # 下一根开盘价作为入场价
+        future_candles = datas[idx + 1:]  # 未来行情
+
+        entry_open = float(entry_candle['open'])  # 开仓价格
+        direction, signal_type = check_signal(prev, curr)  # 判断开仓方向和信号类型
+
+        if direction and signal_type:
+            daily_signals += 1
+
+            exit_price, diff, exit_time = simulate_trade(direction, entry_open, future_candles, take_profit_diff=6,
+                                                         stop_loss_diff=-2)
+
+            # 统计该信号类型的表现
+            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
+
+            # 将时间戳转换为本地时间
+            local_time = datetime.datetime.fromtimestamp(entry_candle['id'] / 1000)
+            formatted_time = local_time.strftime("%Y-%m-%d %H:%M:%S")
+
+            exit_time = datetime.datetime.fromtimestamp(exit_time / 1000)
+            exit_time1 = exit_time.strftime("%Y-%m-%d %H:%M:%S")
+
+            # 保存交易详情
+            all_trades.append(
+                (
+                    f"{formatted_time}号",
+                    "做多" if direction == "long" else "做空",
+                    signal_stats[signal_type]["name"],
+                    entry_open,
+                    exit_price,
+                    diff,
+                    exit_time1
+                )
+            )
+
+    # ===== 输出每笔交易详情 =====
+    logger.info("===== 每笔交易详情 =====")
+    n = n1 = 0  # n = 总盈利，n1 = 总手续费
+    for date, 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} {direction}({signal_name}) 入场={entry:.2f} 出场={exit:.2f} 出场时间={end_time} "
+            f"差价={diff:.2f} 盈利={profit_amount:.2f} "
+            f"开仓手续费=5u 平仓手续费={close_fee:.2f}"
+        )
+        n1 += 5 + close_fee
+        n += profit_amount
+
+    print(f'一共笔数：{len(all_trades)}')
+    print(f"一共盈利：{n:.2f}")
+    print(f'一共手续费：{n1:.2f}')

weex/读取数据库分析数据2.0.py

@@ -0,0 +1,163 @@
+import datetime
+from loguru import logger
+from models.weex import Weex
+
+
+def get_data_by_date(date_str):
+    """按日期获取K线数据"""
+    try:
+        target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+    except ValueError:
+        print("日期格式不正确，请使用 YYYY-MM-DD 格式。")
+        return []
+
+    start_timestamp = int(target_date.timestamp() * 1000)
+    end_timestamp = int((target_date + datetime.timedelta(days=1)).timestamp() * 1000) - 1
+
+    query = Weex.select().where(Weex.id.between(start_timestamp, end_timestamp)).order_by(Weex.id.asc())
+    results = list(query)
+
+    return [
+        {'id': item.id, 'open': item.open, 'high': item.high, 'low': item.low, 'close': item.close}
+        for item in results
+    ]
+
+
+# ========== 信号逻辑 ==========
+def is_bullish(candle):
+    return float(candle['open']) < float(candle['close'])
+
+
+def is_bearish(candle):
+    return float(candle['open']) > float(candle['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):
+        if c_open <= p_close and c_close >= p_open:
+            return "long", "bear_bull_engulf"
+
+    # 前涨后跌包住 -> 做空
+    if is_bearish(curr) and is_bullish(prev):
+        if c_open >= p_close and c_close <= p_open:
+            return "short", "bull_bear_engulf"
+
+    return None, None
+
+
+# ========== 止盈止损模拟 ==========
+def simulate_trade(direction, entry_price, entry_candle, future_candles, take_profit_diff=30, stop_loss_diff=-10):
+    """
+    模拟一笔交易
+    direction: long / short
+    entry_price: 入场价格（下一根开盘）
+    entry_candle: 开仓K线（判断是否立刻止损）
+    future_candles: 开仓后未来K线
+    """
+    entry_high = float(entry_candle['high'])
+    entry_low = float(entry_candle['low'])
+
+    if direction == "long":
+        take_profit_price = entry_price + take_profit_diff
+        stop_loss_price = entry_price + stop_loss_diff
+    else:
+        take_profit_price = entry_price - take_profit_diff
+        stop_loss_price = entry_price - stop_loss_diff
+
+    # ✅【优化点1】：如果开仓当根已经触及止损价，则仍按开仓价计算，不立即止损
+    if direction == "long" and entry_low <= stop_loss_price:
+        logger.debug(f"多单开仓价{entry_price}，但当根最低价{entry_low}已触及止损{stop_loss_price}，按开仓价计算。")
+        stop_loss_price = entry_price
+
+    if direction == "short" and entry_high >= stop_loss_price:
+        logger.debug(f"空单开仓价{entry_price}，但当根最高价{entry_high}已触及止损{stop_loss_price}，按开仓价计算。")
+        stop_loss_price = entry_price
+
+    # ✅【优化点2】：从 entry_candle 下一根开始计算止盈止损
+    for candle in future_candles:
+        high, low, close = float(candle['high']), float(candle['low']), float(candle['close'])
+
+        if direction == "long":
+            if high >= take_profit_price:
+                return take_profit_price, take_profit_diff, candle['id']
+            if low <= stop_loss_price:
+                return stop_loss_price, stop_loss_price - entry_price, candle['id']
+
+        elif direction == "short":
+            if low <= take_profit_price:
+                return take_profit_price, entry_price - take_profit_price, candle['id']
+            if high >= stop_loss_price:
+                return stop_loss_price, entry_price - stop_loss_price, candle['id']
+
+    # ✅【未触发止盈止损，最后一根收盘平仓】
+    final_candle = future_candles[-1]
+    final_price = float(final_candle['close'])
+    diff = final_price - entry_price if direction == "long" else entry_price - final_price
+
+    return final_price, diff, final_candle['id']
+
+
+# ========== 主逻辑 ==========
+def run_backtest(date_str='2025-10-08'):
+    data = get_data_by_date(date_str)
+    if not data:
+        print("无数据")
+        return
+
+    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": "跌包涨"},
+    }
+
+    total_profit = 0
+    total_fee = 0
+
+    for idx in range(1, len(data) - 2):
+        prev, curr = data[idx - 1], data[idx]
+        entry_candle = data[idx + 1]
+        future_candles = data[idx + 2:]
+        entry_open = float(entry_candle['open'])
+
+        direction, signal_type = check_signal(prev, curr)
+        if not direction:
+            continue
+
+        exit_price, diff, exit_time = simulate_trade(direction, entry_open, entry_candle, future_candles,
+                                                     take_profit_diff=30, stop_loss_diff=-2)
+
+        signal_stats[signal_type]["count"] += 1
+        signal_stats[signal_type]["total_profit"] += diff
+        if diff > 0:
+            signal_stats[signal_type]["wins"] += 1
+
+        local_time = datetime.datetime.fromtimestamp(entry_candle['id'] / 1000).strftime("%Y-%m-%d %H:%M:%S")
+
+        all_trades.append(
+            (local_time, direction, signal_stats[signal_type]["name"], entry_open, exit_price, diff, exit_time))
+
+    # ===== 输出结果 =====
+    logger.info("===== 每笔交易详情 =====")
+    for date, direction, name, entry, exit, diff, end_time in all_trades:
+        profit_amount = diff / entry * 10000
+        close_fee = 10000 / entry * exit * 0.0005
+        total_profit += profit_amount
+        total_fee += 5 + close_fee
+
+        logger.info(f"{date} {('做多' if direction == 'long' else '做空')}({name}) 入场={entry:.2f} 出场={exit:.2f} "
+                    f"出场时间={end_time} 差价={diff:.2f} 盈利={profit_amount:.2f} 手续费={close_fee:.2f}")
+
+    print(f"总交易数：{len(all_trades)}")
+    print(f"总盈利：{total_profit:.2f}")
+    print(f"总手续费：{total_fee:.2f}")
+
+
+if __name__ == "__main__":
+    run_backtest("2025-10-08")