| self = <test_cluster_line_buy_source.TestLegZspreadRegimeAcceptance object at 0x1541a6030>
|
| stub_hot_flag = None
|
|
|
| def test_emit_stamps_leg_payload_with_zspread_regime_when_seam_present(self, stub_hot_flag):
|
| """END-TO-END. The existing 8/14 fixture's FIRST leg fires at armed_at=11:19
|
| (entry_trigger_et=11:47, et=11:48). Inject a divergent-shape series whose
|
| trailing-90 ending at 11:19 carries the four-peak / four-trough layout the
|
| recognizer needs for a non-unreadable read, and assert the emitted SimEvent
|
| payload carries the regime stamp — the integration that proves the wiring
|
| reaches the SimEvent bus, not just the helper method.
|
|
|
| The second leg fires at armed_at=12:50 — its trailing-90 window (10:21..12:50)
|
| ALSO needs at least four peaks/troughs for a non-unreadable read, so the series
|
| carries a SECOND cluster inside that window. The test asserts both legs get the
|
| regime stamp.
|
| """
|
| # LEG 1 window (08:50..11:19, observed 09:21..11:19 since the series starts at
|
| # 09:21 from the 12:50-end lookback_min=90) — 4 peaks + 4 troughs inside.
|
| # LEG 2 window (10:21..12:50) — 4 peaks + 4 troughs inside.
|
| peaks = {9 * 60 + 30: 1.0, 9 * 60 + 50: 0.5,
|
| 10 * 60 + 50: 1.5, 11 * 60 + 0: 1.3,
|
| 11 * 60 + 40: 1.2, 11 * 60 + 50: 0.7,
|
| 12 * 60 + 20: 1.4, 12 * 60 + 40: 1.0}
|
| troughs = {9 * 60 + 25: -0.5, 9 * 60 + 40: -0.3,
|
| 10 * 60 + 30: -1.1, 10 * 60 + 40: -0.7,
|
| 11 * 60 + 20: -0.6, 11 * 60 + 30: -0.4,
|
| 12 * 60 + 0: -1.2, 12 * 60 + 10: -0.8}
|
| # Build a series ending at 12:50 (the LATER leg's fire minute) so both windows
|
| # have non-empty trailing-90.
|
| series = self._build_series("12:50", peaks, troughs, lookback_min=90)
|
| src = ClusterLineBuySource()
|
| ctx = _ctx("2026-08-14", BARS_2026_08_14, LEVEL_2026_08_14)
|
| ctx.raw["cluster_line_zspread_series"] = series
|
| evs = src.emit(ctx, SimConfig())
|
| # Two long legs emitted on 8/14 — the hole gate strips the 13:57 leg.
|
| assert len(evs) == 2
|
| for ev in evs:
|
| assert "zspread_regime" in ev.payload, (
|
| f"leg at {ev.et} missing zspread_regime payload — wiring did not reach the "
|
| f"SimEvent bus for this leg"
|
| )
|
| # Each leg's trailing-90 must read a regime — the seam shape produced
|
| # divergent / trending in the synthetic series, so accept any non-unreadable
|
| # regime here. The acceptance of the EXACT pinned regimes is the helper-level
|
| # tests above; this test asserts only that the wiring reaches the bus.
|
| regime = ev.payload["zspread_regime"]["regime"]
|
| > assert regime in {"trending_up", "trending_down", "converging", "diverging"}, (
|
| f"leg at {ev.et} regime {regime!r} is unreadable — wiring failed to read the "
|
| f"injected series; check that the series covers the leg's trailing-90 window"
|
| )
|
| E AssertionError: leg at 11:48:00 regime 'unreadable' is unreadable — wiring failed to read the injected series; check that the series covers the leg's trailing-90 window
|
| E assert 'unreadable' in {'converging', 'diverging', 'trending_down', 'trending_up'}
|
|
|
| tests/services/test_cluster_line_buy_source.py:2784: AssertionError
|