fewfef

weex/调试信号检测.py

@@ -0,0 +1,90 @@
+"""
+调试信号检测逻辑，查看为什么没有检测到交易信号
+"""
+import datetime
+from models.weex import Weex30
+
+def get_data_by_date(model, date_str: str):
+    """按天获取指定表的数据"""
+    try:
+        target_date = datetime.datetime.strptime(date_str, '%Y-%m-%d')
+    except ValueError:
+        print(f"日期格式不正确: {date_str}")
+        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 is_bullish(c):
+    return float(c['close']) > float(c['open'])
+
+def is_bearish(c):
+    return float(c['close']) < float(c['open'])
+
+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
+
+# 获取数据
+dates = []
+for i in range(2, 32):
+    dates.append(f"2025-12-{i:02d}")
+for i in range(1, 5):
+    dates.append(f"2026-01-{i:02d}")
+
+all_data = []
+for d in dates:
+    day_data = get_data_by_date(Weex30, d)
+    all_data.extend(day_data)
+    if day_data:
+        print(f"{d}: 读取到 {len(day_data)} 条数据")
+
+print(f"\n总共读取到 {len(all_data)} 条数据")
+
+if len(all_data) < 2:
+    print("数据不足，无法检测信号（至少需要2条K线）")
+    exit()
+
+# 排序
+all_data.sort(key=lambda x: x['id'])
+
+# 检查信号
+signal_count = 0
+for idx in range(1, len(all_data)):
+    prev = all_data[idx - 1]
+    curr = all_data[idx]
+    direction, signal_key = check_signal(prev, curr)
+    
+    if direction:
+        signal_count += 1
+        prev_time = datetime.datetime.fromtimestamp(prev['id'] / 1000)
+        curr_time = datetime.datetime.fromtimestamp(curr['id'] / 1000)
+        print(f"\n信号 #{signal_count}: {signal_key} ({direction})")
+        print(f"  前一根K线 ({prev_time}): O={prev['open']:.2f} H={prev['high']:.2f} L={prev['low']:.2f} C={prev['close']:.2f} {'阳' if is_bullish(prev) else '阴'}")
+        print(f"  当前K线 ({curr_time}): O={curr['open']:.2f} H={curr['high']:.2f} L={curr['low']:.2f} C={curr['close']:.2f} {'阳' if is_bullish(curr) else '阴'}")
+        print(f"  包住条件检查:")
+        print(f"    - 前跌后涨: {is_bearish(prev)} and {is_bullish(curr)}")
+        print(f"    - 开盘<=前收盘: {float(curr['open'])} <= {float(prev['close'])} = {float(curr['open']) <= float(prev['close'])}")
+        print(f"    - 收盘>=前开盘: {float(curr['close'])} >= {float(prev['open'])} = {float(curr['close']) >= float(prev['open'])}")
+
+if signal_count == 0:
+    print("\n未检测到任何信号！")
+    print("\n检查前10根K线的情况：")
+    for idx in range(min(10, len(all_data))):
+        k = all_data[idx]
+        k_time = datetime.datetime.fromtimestamp(k['id'] / 1000)
+        print(f"  {idx}: {k_time} O={k['open']:.2f} C={k['close']:.2f} {'阳' if is_bullish(k) else '阴'}")