1"""Leakage gate: nothing in the inputs may depend on bars after the origin; the deliberate-leak fixture must be rejected."""2import numpy as np3import pytest45from chronos_x.data.synthetic import synthetic_bars6from chronos_x.data.features import build_grid, N_DAY7from chronos_x.data.windows import SymbolGrid, WindowConfig, make_example, eligible_origins8910import polars as pl111213def _grid(seed=3, mutate_future=None):14 """mutate_future = (s, m): every bar strictly after origin (s, m) gets its prices multiplied by 1.05 and its volume x3."""15 df, actions = synthetic_bars(start="2024-01-02", end="2024-03-15", seed=seed, p_missing=0.03)16 if mutate_future is not None:17 s, m = mutate_future18 from chronos_x.data.calendar import sessions19 sess = sessions("2024-01-02", "2024-03-15")20 cutoff = sess[s].open_utc + np.timedelta64(m, "m") # bar-open instant of slot m21 fut = pl.col("timestamp_utc") > pl.lit(cutoff.to_pydatetime())22 df = df.with_columns([pl.when(fut).then(pl.col(c) * 1.05).otherwise(pl.col(c)).alias(c) for c in ("open", "high", "low", "close")]23 + [pl.when(fut).then(pl.col("volume") * 3).otherwise(pl.col("volume")).alias("volume")])24 arrays, meta = build_grid(df, actions, "2024-01-02", "2024-03-15", "SYN", "stock")25 return SymbolGrid("SYN", arrays, meta.__dict__)262728def test_inputs_invariant_to_future_bars():29 cfg = WindowConfig()30 g0 = _grid()31 s, m = 30, 20032 ex0 = make_example(g0, s, m, cfg)3334 g1 = _grid(mutate_future=(s, m))35 ex1 = make_example(g1, s, m, cfg)36 for k in ("xA", "xB", "xC", "glob", "sigma", "fut_cal", "mA", "mB", "mC"):37 assert np.array_equal(ex0[k], ex1[k]), f"input {k} depends on the future"38 assert not np.array_equal(ex0["y"], ex1["y"])394041def test_targets_identity_with_log_close():42 cfg = WindowConfig()43 g = _grid()44 for s, m in eligible_origins(g, cfg, np.array([25, 40]), stride=37)[:20]:45 ex = make_example(g, int(s), int(m), cfg)46 y, ym = ex["y"], ex["y_mask"] > 047 for H in (5, 15, 30, 60):48 fut = np.arange(1, H + 1)49 present = ym[:H]50 if not present.any():51 continue52 last = s * N_DAY + m + fut[present][-1]53 exact = g.flat_lc[last] - g.flat_lc[s * N_DAY + m]54 assert np.isclose(y[:H][present].sum(), exact, atol=1e-6)555657def test_bucket_completion_rule():58 """At origin m, Branch B's last bucket must end at a slot <= m and Branch C likewise."""59 cfg = WindowConfig()60 g = _grid()61 s = 3062 for m in (30, 33, 34, 35, 59, 60, 200):63 ex = make_example(g, s, m, cfg)64 jB = (m + 1) // 5 - 165 assert 5 * jB + 4 <= m and 5 * (jB + 1) + 4 > m66 jC = (m + 1) // 30 - 167 assert 30 * jC + 29 <= m and 30 * (jC + 1) + 29 > m68 # last Branch-B row equals bucket jB of session s (close feature consistent)69 assert ex["mB"][-1] == g.flat_m5[s * 78 + jB]707172def test_deliberate_future_feature_fixture_is_rejected():73 """Quarantined fixture: a feature built from the next bar must be detected by the invariance check."""74 cfg = WindowConfig()75 g0 = _grid(); s, m = 30, 20076 ex0 = make_example(g0, s, m, cfg)7778 def leaky_example(g):79 ex = make_example(g, s, m, cfg)80 ex["xA"] = ex["xA"].copy()81 ex["xA"][-1, 0] = g.flat_lc[s * N_DAY + m + 1] - g.flat_lc[s * N_DAY + m] # uses t+1 close82 return ex8384 g1 = _grid(mutate_future=(s, m))85 leak0, leak1 = leaky_example(g0), leaky_example(g1)86 with pytest.raises(AssertionError):87 assert np.array_equal(leak0["xA"], leak1["xA"])88 assert np.array_equal(ex0["xA"], make_example(g1, s, m, cfg)["xA"])899091def test_origin_eligibility_respects_close_and_early_close():92 cfg = WindowConfig()93 df, actions = synthetic_bars(start="2024-11-20", end="2024-12-05", seed=5, p_missing=0.0)94 arrays, meta = build_grid(df, actions, "2024-11-20", "2024-12-05", "SYN", "stock")95 g = SymbolGrid("SYN", arrays, meta.__dict__)96 i = meta.session_dates.index("2024-11-29")97 o = eligible_origins(g, cfg, np.array([i, i + 1]))98 early = o[o[:, 0] == i][:, 1]; normal = o[o[:, 0] == i + 1][:, 1]99 assert early.max() + cfg.H <= 209 and normal.max() + cfg.H <= 389100 assert early.min() == cfg.m_min and normal.max() == 389 - cfg.H101