tests/test_feature_pipeline.py

"""
Executable checks for features/feature_pipeline.py -- in particular an
actual, run-it-yourself proof of the "no lookahead" property, not just a
docstring claim.

Run with:  python -m unittest tests.test_feature_pipeline -v
(from the project root; only needs pandas/numpy, same as indicators.py).
"""
import unittest

import numpy as np
import pandas as pd

from features.feature_pipeline import (
    FEATURE_COLUMNS,
    carry_forward_future,
    compute_feature_frame,
    compute_feature_frame_for_period,
    slice_features,
    validate_features,
)


def _synthetic_ohlcv(n: int, seed: int = 0) -> pd.DataFrame:
    """A plausible-looking random-walk OHLCV frame, positive prices,
    strictly increasing DatetimeIndex -- enough to exercise every
    indicator without needing real market data."""
    rng = np.random.default_rng(seed)
    steps = rng.normal(loc=0.0, scale=0.5, size=n)
    close = 100 + np.cumsum(steps)
    close = np.abs(close) + 1.0  # keep strictly positive
    open_ = close + rng.normal(0, 0.1, n)
    high = np.maximum(open_, close) + np.abs(rng.normal(0, 0.2, n))
    low = np.minimum(open_, close) - np.abs(rng.normal(0, 0.2, n))
    volume = rng.integers(100, 10000, n).astype(float)
    idx = pd.date_range("2024-01-01", periods=n, freq="min")
    return pd.DataFrame(
        {"Open": open_, "High": high, "Low": low, "Close": close, "Volume": volume},
        index=idx,
    )


class TestNoLookahead(unittest.TestCase):
    """The property the whole no-leakage requirement rests on: a feature
    value at row t must be identical no matter how much (or little) data
    after row t exists when it's computed."""

    def test_truncation_invariance(self):
        df = _synthetic_ohlcv(300, seed=1)
        full = compute_feature_frame(df)

        # Recompute using only data up to several different cutoffs, and
        # check every row that exists in both versions matches EXACTLY.
        for cutoff in (50, 100, 150, 250, 299):
            truncated = compute_feature_frame(df.iloc[: cutoff + 1])
            overlap = min(len(full), len(truncated))
            pd.testing.assert_frame_equal(
                full.iloc[:overlap].reset_index(drop=True),
                truncated.iloc[:overlap].reset_index(drop=True),
                check_exact=False,
                atol=1e-10,
                obj=f"feature frame truncated at row {cutoff}",
            )

    def test_single_row_extension_does_not_rewrite_history(self):
        """Appending ONE new candle must never change any prior row --
        this is what the walk-forward backtest loop depends on candle to
        candle."""
        df = _synthetic_ohlcv(120, seed=2)
        shorter = compute_feature_frame(df.iloc[:100])
        longer = compute_feature_frame(df.iloc[:101])
        pd.testing.assert_frame_equal(
            shorter.reset_index(drop=True),
            longer.iloc[:100].reset_index(drop=True),
            check_exact=False,
            atol=1e-10,
        )


class TestShapeAndColumns(unittest.TestCase):
    def test_columns_are_exactly_the_expected_13(self):
        df = _synthetic_ohlcv(200, seed=3)
        feats = compute_feature_frame(df)
        self.assertEqual(list(feats.columns), FEATURE_COLUMNS)
        # 4 raw OHLCV (Open/High/Low/Volume) + 1 RSI + 3 MACD + 2 Bollinger
        # + 2 SMA/EMA + 2 Stochastic, from 9 used categories (Close excluded).
        self.assertEqual(len(FEATURE_COLUMNS), 14)

    def test_close_is_not_a_feature_column(self):
        self.assertNotIn("Close", FEATURE_COLUMNS)

    def test_bb_mid_not_duplicated_as_sma20(self):
        # BB_Mid == SMA_20 by construction (see module docstring) -- make
        # sure we didn't accidentally keep both.
        self.assertIn("SMA_20", FEATURE_COLUMNS)
        self.assertNotIn("BB_Mid", FEATURE_COLUMNS)


class TestNoNaN(unittest.TestCase):
    def test_no_nan_once_past_warmup(self):
        df = _synthetic_ohlcv(200, seed=4)
        feats = compute_feature_frame(df)
        # Row 60 has 60 real candles of history before it -- comfortably
        # past every indicator's warm-up (the longest is Bollinger/SMA_20
        # at 20 candles).
        self.assertFalse(feats.iloc[60:].isna().any().any())

    def test_first_19_rows_ARE_nan_without_padding(self):
        """The failure mode this project actually hit: compute_feature_frame
        alone always leaves its own first ~19 rows NaN, no matter how large
        the input is -- window size doesn't matter, row 0 never has warm-up
        history *within this fetch*. This is the bug compute_feature_frame_
        for_period (tested below) exists to work around."""
        df = _synthetic_ohlcv(500, seed=99)
        feats = compute_feature_frame(df)
        self.assertTrue(feats.iloc[:19].isna().any().any())
        self.assertFalse(feats.iloc[19:].isna().any().any())

    def test_missing_ohlcv_column_raises(self):
        bad = pd.DataFrame({"Open": [1, 2], "High": [1, 2], "Low": [1, 2]})
        with self.assertRaises(ValueError):
            compute_feature_frame(bad)


class TestComputeFeatureFrameForPeriod(unittest.TestCase):
    """The actual fix for the observed bug: a Backtest/Live run whose
    requested period starts right at the edge of the fetched data always
    had NaN in its first ~19 rows (see test_first_19_rows_ARE_nan_without_
    padding above) -- this is what makes that go away."""

    def test_no_nan_anywhere_in_the_trimmed_result(self):
        # Simulate "fetch padding_days extra, then only want the last
        # keep_days" -- e.g. window=60 on 1-minute BTC-USD candles, which
        # is the exact case that surfaced this bug.
        df_padded = _synthetic_ohlcv(60 * 24 * 3, seed=21)  # 3 days of 1-min candles
        df_trimmed, feats_trimmed = compute_feature_frame_for_period(df_padded, keep_days=2)
        self.assertFalse(feats_trimmed.isna().any().any(),
                          "compute_feature_frame_for_period must never leave "
                          "leading NaN in the period actually requested")

    def test_trimmed_period_is_the_requested_length_not_the_padded_one(self):
        df_padded = _synthetic_ohlcv(60 * 24 * 3, seed=22)
        keep_days = 1
        df_trimmed, feats_trimmed = compute_feature_frame_for_period(df_padded, keep_days=keep_days)
        expected_span = df_padded.index[-1] - df_trimmed.index[0]
        self.assertLessEqual(expected_span, pd.Timedelta(days=keep_days) + pd.Timedelta(minutes=1))
        self.assertEqual(len(df_trimmed), len(feats_trimmed))

    def test_trimmed_values_match_computing_on_the_full_padded_series(self):
        """The trim must only DROP rows, never change the surviving ones --
        same truncation-invariance property as compute_feature_frame itself,
        just checked the other direction (padding in front, not truncating
        the end)."""
        df_padded = _synthetic_ohlcv(60 * 24 * 2, seed=23)
        full_features = compute_feature_frame(df_padded)
        df_trimmed, feats_trimmed = compute_feature_frame_for_period(df_padded, keep_days=1)

        keep_mask = df_padded.index >= df_padded.index[-1] - pd.Timedelta(days=1)
        expected = full_features.iloc[keep_mask].reset_index(drop=True)
        pd.testing.assert_frame_equal(feats_trimmed, expected, atol=1e-10)

    def test_reproduces_and_fixes_the_reported_btc_1m_window60_case(self):
        """Direct reproduction of the actual failure: BTC-USD, 1-minute
        candles, 1 day lookback, window=60 -- the first backtest window is
        rows [0:60] of whatever got fetched, which used to include the
        always-NaN rows [0:19]. With padding, it no longer does."""
        from utils.helpers import feature_warmup_days
        keep_days = 1
        padding_days = feature_warmup_days("1m", minimum_days=0)
        df_padded = _synthetic_ohlcv(60 * 24 * (keep_days + padding_days) + 100, seed=24)

        df_trimmed, feats_trimmed = compute_feature_frame_for_period(df_padded, keep_days=keep_days)
        window = 60
        first_window = slice_features(feats_trimmed, 0, window)
        self.assertFalse(first_window.isna().any().any(),
                          "the FIRST walk-forward window must be NaN-free after padding")


class TestSliceFeatures(unittest.TestCase):
    def test_slice_matches_close_slice_positionally(self):
        df = _synthetic_ohlcv(200, seed=5)
        feats = compute_feature_frame(df)
        close = df["Close"].reset_index(drop=True)

        window, i = 60, 130
        close_window = close.iloc[i - window : i]
        feat_window = slice_features(feats, i - window, i)

        self.assertEqual(len(close_window), len(feat_window))
        # Same positional slice, just re-indexed -- last row of the feature
        # window must describe the SAME candle as the last row of the
        # close window (spot check via Close reconstructed from OHLCV: the
        # feature frame doesn't carry Close, so we check against the
        # original df's own Close at that position instead).
        self.assertAlmostEqual(
            df["Close"].iloc[i - 1], close_window.iloc[-1], places=8
        )


class TestValidateFeatures(unittest.TestCase):
    def test_length_mismatch_raises(self):
        feats = pd.DataFrame({c: [1.0, 2.0] for c in FEATURE_COLUMNS})
        with self.assertRaises(ValueError):
            validate_features(feats, expected_len=3, model_name="Test")

    def test_nan_raises(self):
        feats = pd.DataFrame({c: [1.0, np.nan] for c in FEATURE_COLUMNS})
        with self.assertRaises(ValueError):
            validate_features(feats, expected_len=2, model_name="Test")

    def test_check_nan_false_skips_nan_but_still_checks_length(self):
        """The Moirai/TimesFM pattern: a caller that only ever uses the
        TAIL of `features` calls this once with check_nan=False against the
        FULL history length (to still catch a genuinely misaligned caller
        early), then again with the default check_nan=True against just
        the truncated tail it actually uses."""
        feats_with_nan = pd.DataFrame({c: [np.nan, 1.0, 2.0] for c in FEATURE_COLUMNS})
        validate_features(feats_with_nan, expected_len=3, model_name="Test", check_nan=False)  # no raise

        with self.assertRaises(ValueError):
            validate_features(feats_with_nan, expected_len=2, model_name="Test", check_nan=False)

    def test_check_nan_false_then_true_on_truncated_tail_mirrors_model_usage(self):
        # row 0 has NaN, rows 1-2 don't -- a model that only uses the last
        # 2 rows should pass; one that (mistakenly) got handed the NaN row
        # inside its own truncation should still fail.
        feats = pd.DataFrame({c: [np.nan, 1.0, 2.0] for c in FEATURE_COLUMNS})
        validate_features(feats, expected_len=3, model_name="Test", check_nan=False)

        clean_tail = feats.iloc[-2:].reset_index(drop=True)
        validate_features(clean_tail, expected_len=2, model_name="Test")  # no raise -- no NaN in the tail

        dirty_tail = feats.iloc[-3:].reset_index(drop=True)
        with self.assertRaises(ValueError):
            validate_features(dirty_tail, expected_len=3, model_name="Test")  # NaN still inside this one

    def test_clean_window_passes(self):
        feats = pd.DataFrame({c: [1.0, 2.0, 3.0] for c in FEATURE_COLUMNS})
        validate_features(feats, expected_len=3, model_name="Test")  # no raise


class TestCarryForwardFuture(unittest.TestCase):
    def test_repeats_last_row(self):
        feats = pd.DataFrame({"a": [1.0, 2.0, 3.0], "b": [10.0, 20.0, 30.0]})
        future = carry_forward_future(feats, horizon=4)
        self.assertEqual(len(future), 4)
        self.assertTrue((future["a"] == 3.0).all())
        self.assertTrue((future["b"] == 30.0).all())

    def test_uses_zero_future_information(self):
        """The defining property: carry-forward for a window ending at
        cutoff `c` must be identical whether or not candles AFTER `c`
        exist in the source data -- otherwise it would be leaking them."""
        df = _synthetic_ohlcv(150, seed=6)
        feats_short = compute_feature_frame(df.iloc[:80])
        feats_long = compute_feature_frame(df)  # 150 rows, i.e. real future data exists

        future_from_short = carry_forward_future(feats_short.iloc[:80], horizon=5)
        future_from_long = carry_forward_future(feats_long.iloc[:80], horizon=5)
        pd.testing.assert_frame_equal(future_from_short, future_from_long, atol=1e-10)

    def test_rejects_bad_input(self):
        feats = pd.DataFrame({"a": [1.0]})
        with self.assertRaises(ValueError):
            carry_forward_future(feats, horizon=0)
        with self.assertRaises(ValueError):
            carry_forward_future(pd.DataFrame({"a": []}), horizon=1)


if __name__ == "__main__":
    unittest.main()