Merge PR #1873: emit swing signals only after confirmation

agent/src/skills/elliott-wave/SKILL.md

@@ -54,3 +54,4 @@ pip install pandas numpy requests
 ## Signal Convention
 
 - `1` = long, `-1` = short, `0` = stand aside
+- A swing point is only known `swing_window` bars after it forms, so the signal is stamped on that confirmation bar, not on the swing bar itself

agent/src/skills/elliott-wave/example_signal_engine.py

@@ -53,15 +53,19 @@ class SignalEngine:
     def _find_swings(
         self, high: pd.Series, low: pd.Series
     ) -> List[Dict]:
-        """用滚动窗口找局部高低点，生成交替的 Zigzag 序列。
+        """用滚动窗口找局部高低点（原始 Swing 点，未做 Zigzag 过滤）。
+
+        中心窗口的极值要等右侧 ``swing_window`` 根 K 线收盘后才能确认，
+        因此每个点带 ``confirmed``：该点实际可被观察到的那根 K 线。
 
         Args:
             high: 最高价序列。
             low: 最低价序列。
 
         Returns:
-            交替排列的 Swing 点列表，每个元素为
-            {"index": datetime, "price": float, "type": "H"|"L"}。
+            按时间排序的 Swing 点列表，每个元素为
+            {"index": datetime, "price": float, "type": "H"|"L",
+            "confirmed": datetime}。
         """
         w = self.swing_window
         full_w = w * 2 + 1
@@ -69,33 +73,44 @@ class SignalEngine:
         if len(high) < full_w:
             return []
 
-        # 检测局部极值
         roll_max = high.rolling(full_w, center=True).max()
         roll_min = low.rolling(full_w, center=True).min()
-        swing_high_mask = high == roll_max
-        swing_low_mask = low == roll_min
+        swing_high_mask = (high == roll_max).to_numpy()
+        swing_low_mask = (low == roll_min).to_numpy()
 
-        # 收集所有候选 Swing 点
         raw_points = []
-        for idx in high.index:
-            is_h = bool(swing_high_mask.get(idx, False))
-            is_l = bool(swing_low_mask.get(idx, False))
+        for position, idx in enumerate(high.index):
+            is_h = bool(swing_high_mask[position])
+            is_l = bool(swing_low_mask[position])
             if is_h and is_l:
-                # 同一根K线同时是高低点，取幅度更大的
-                pass
-            elif is_h:
-                raw_points.append({"index": idx, "price": float(high[idx]), "type": "H"})
-            elif is_l:
-                raw_points.append({"index": idx, "price": float(low[idx]), "type": "L"})
+                continue
+            if is_h or is_l:
+                raw_points.append(
+                    {
+                        "index": idx,
+                        "price": float(high.iloc[position] if is_h else low.iloc[position]),
+                        "type": "H" if is_h else "L",
+                        "confirmed": high.index[position + w],
+                    }
+                )
+        return raw_points
 
+    @staticmethod
+    def _zigzag(raw_points: List[Dict]) -> List[Dict]:
+        """把原始 Swing 点过滤为严格交替的 H/L 序列。
+
+        Args:
+            raw_points: ``_find_swings`` 返回的 Swing 点（可以是其前缀）。
+
+        Returns:
+            交替排列的 Swing 点列表；连续同类型点保留更极端的一个。
+        """
         if len(raw_points) < 2:
-            return raw_points
+            return list(raw_points)
 
-        # 过滤为严格交替的 H/L 序列
         zigzag = [raw_points[0]]
         for pt in raw_points[1:]:
             if pt["type"] == zigzag[-1]["type"]:
-                # 连续同类型：保留更极端的
                 if pt["type"] == "H" and pt["price"] > zigzag[-1]["price"]:
                     zigzag[-1] = pt
                 elif pt["type"] == "L" and pt["price"] < zigzag[-1]["price"]:
@@ -369,21 +384,24 @@ class SignalEngine:
         for code, df in data_map.items():
             signal = pd.Series(0, index=df.index, dtype=int)
 
-            swings = self._find_swings(df["high"], df["low"])
-            if len(swings) < 4:
-                result[code] = signal
-                continue
+            raw_points = self._find_swings(df["high"], df["low"])
+            for count in range(4, len(raw_points) + 1):
+                swings = self._zigzag(raw_points[:count])
+                newest = raw_points[count - 1]
+                if len(swings) < 4 or swings[-1]["index"] != newest["index"]:
+                    continue
+                confirmed = newest["confirmed"]
+                if confirmed not in signal.index:
+                    continue
 
-            # 5 浪推动结构
-            if len(swings) >= 6:
-                for idx, direction in self._find_impulse(swings):
-                    if idx in signal.index:
-                        signal[idx] = direction
+                if len(swings) >= 6:
+                    for idx, direction in self._find_impulse(swings):
+                        if idx == newest["index"]:
+                            signal[confirmed] = direction
 
-            # ABC 调整结构
-            for idx, direction in self._find_abc(swings):
-                if idx in signal.index:
-                    signal[idx] = direction
+                for idx, direction in self._find_abc(swings):
+                    if idx == newest["index"]:
+                        signal[confirmed] = direction
 
             result[code] = signal
 

agent/src/skills/harmonic/SKILL.md

@@ -41,3 +41,4 @@ engine falls back to the bundled detector.
 ## Signal Convention
 
 - `1` = long, `-1` = short, `0` = stand aside
+- A swing point is only known `swing_window` bars after it forms, so the signal is stamped on that confirmation bar, not on the swing bar itself (bundled detector only; the optional `pyharmonics` path is unchanged)

agent/src/skills/harmonic/example_signal_engine.py

@@ -192,15 +192,16 @@ def _detect_patterns_fallback(
         检测到的形态列表，每项含 pattern/direction/d_index/d_price 等字段。
     """
     swing_highs, swing_lows = _find_swings(df["high"], df["low"], swing_window)
-    merged = _merge_swings(swing_highs, swing_lows)
-
-    if len(merged) < 5:
-        return []
 
     found = []
-    # 滑动窗口，每次取5个连续摆动点作为 X-A-B-C-D 候选
-    for i in range(len(merged) - 4):
-        pts = merged[i : i + 5]
+    # 摆动点要等右侧 swing_window 根K线收盘才能确认，因此按时间逐点推进，
+    # 只使用截至该点为止已出现的摆动点，形态在确认那根K线上才算出现
+    for d_ts in sorted(set(swing_highs.index) | set(swing_lows.index)):
+        merged = _merge_swings(swing_highs.loc[:d_ts], swing_lows.loc[:d_ts])
+        if len(merged) < 5 or merged[-1][0] != d_ts:
+            continue
+
+        pts = merged[-5:]
         x_ts, x_price, x_type = pts[0]
         a_ts, a_price, a_type = pts[1]
         b_ts, b_price, b_type = pts[2]
@@ -225,6 +226,7 @@ def _detect_patterns_fallback(
         else:
             direction = "bearish"
 
+        confirmed = df.index[df.index.get_loc(d_ts) + swing_window]
         found.append(
             {
                 "pattern": pattern_name,
@@ -234,7 +236,7 @@ def _detect_patterns_fallback(
                 "b": (b_ts, b_price),
                 "c": (c_ts, c_price),
                 "d": (d_ts, d_price),
-                "d_index": d_ts,
+                "d_index": confirmed,
                 "d_price": d_price,
             }
         )

agent/tests/test_skill_swing_engine_causality.py

@@ -0,0 +1,76 @@
+"""Swing-based skill engines must not trade on pivots before they are confirmed.
+
+A centred rolling extreme at bar ``t`` needs the next ``swing_window`` bars to
+exist, so a signal stamped on the pivot bar itself reads the future. Signals
+must land on the confirmation bar, and truncating the history after bar ``t``
+must not change any signal on or before ``t``.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+SKILLS_ROOT = Path(__file__).resolve().parents[1] / "src" / "skills"
+
+
+def _load(skill: str):
+    path = SKILLS_ROOT / skill / "example_signal_engine.py"
+    spec = importlib.util.spec_from_file_location(f"_causality_{skill.replace('-', '_')}", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _frame(seed: int, bars: int = 320) -> pd.DataFrame:
+    rng = np.random.default_rng(seed)
+    close = 100 + np.cumsum(rng.normal(0, 1.5, bars))
+    index = pd.date_range("2024-01-01", periods=bars, freq="D")
+    return pd.DataFrame(
+        {
+            "open": close,
+            "high": close + rng.uniform(0.2, 1.0, bars),
+            "low": close - rng.uniform(0.2, 1.0, bars),
+            "close": close,
+            "volume": 1.0,
+        },
+        index=index,
+    )
+
+
+def _engines():
+    elliott = _load("elliott-wave").SignalEngine(swing_window=3, fib_tolerance=0.6, min_wave_bars=1)
+    harmonic = _load("harmonic").SignalEngine(swing_window=3, tol=0.3)
+    harmonic.use_pyharmonics = False
+    return {"elliott-wave": elliott, "harmonic": harmonic}
+
+
+@pytest.mark.parametrize("skill", ["elliott-wave", "harmonic"])
+def test_truncating_history_never_changes_earlier_signals(skill):
+    engine = _engines()[skill]
+    emitted = 0
+    for seed in range(6):
+        frame = _frame(seed)
+        full = engine.generate({"X": frame})["X"]
+        emitted += int((full != 0).sum())
+        for cut in range(40, len(frame), 17):
+            head = engine.generate({"X": frame.iloc[: cut + 1]})["X"]
+            pd.testing.assert_series_equal(head, full.iloc[: cut + 1])
+    assert emitted > 0
+
+
+def test_harmonic_signal_bar_is_the_pivot_confirmation_bar():
+    module = _load("harmonic")
+    found = 0
+    for seed in range(6):
+        frame = _frame(seed)
+        for pattern in module._detect_patterns_fallback(frame, swing_window=3, tol=0.3):
+            pivot = frame.index.get_loc(pattern["d"][0])
+            assert frame.index.get_loc(pattern["d_index"]) == pivot + 3
+            found += 1
+    assert found > 0
+