tests/test_backtester_alignment.py

import unittest
import json
import numpy as np
import pandas as pd
from backtest.backtester import backtest_model, run_all_models_backtest, run_comparison_backtest
from features.feature_pipeline import FEATURE_COLUMNS, compute_feature_frame
from models.base_model import BaseForecastModel

def _synthetic_ohlcv(n: int, seed: int=7) -> pd.DataFrame:
    rng = np.random.default_rng(seed)
    steps = rng.normal(loc=0.0, scale=0.4, size=n)
    close = 100 + np.cumsum(steps)
    close = np.abs(close) + 5.0
    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 _AlignmentCheckingModel(BaseForecastModel):
    name = 'AlignmentChecker'
    RAW_COLUMNS = ['Open', 'High', 'Low', 'Volume']

    def __init__(self, df: pd.DataFrame):
        self.full_close = df['Close'].reset_index(drop=True)
        self.raw = df[self.RAW_COLUMNS].reset_index(drop=True)

    def predict(self, history, horizon=1, features=None):
        history = pd.Series(history).reset_index(drop=True)
        if features is not None:
            assert len(features) == len(history), f'length mismatch: history={len(history)} features={len(features)}'
            assert list(features.columns) == FEATURE_COLUMNS
        n = len(self.full_close)
        w = len(history)
        match_start = None
        for start in range(0, n - w + 1):
            if np.allclose(self.full_close.iloc[start:start + w].to_numpy(), history.to_numpy()):
                match_start = start
                break
        assert match_start is not None, 'history window not found in full series -- corrupted slice'
        if features is not None:
            expected = self.raw.iloc[match_start:match_start + w].to_numpy()
            got = features[self.RAW_COLUMNS].to_numpy()
            assert np.allclose(got, expected), f'features window is shifted relative to history at position {match_start}'
        return [float(history.iloc[-1])] * horizon

class TestBacktesterAlignment(unittest.TestCase):

    def setUp(self):
        self.df = _synthetic_ohlcv(200, seed=11)
        self.features_df = compute_feature_frame(self.df)

    def test_no_failed_rows_with_features(self):
        model = _AlignmentCheckingModel(self.df)
        results = backtest_model(self.df, model, window=40, horizon=1, step=1, features_df=self.features_df)
        failed = results['error'].notna().sum()
        self.assertEqual(failed, 0, f"{failed} rows failed alignment checks:\n{results[results['error'].notna()]}")
        self.assertGreater(len(results), 0)

    def test_features_window_matches_close_window_exactly(self):
        close = self.df['Close'].reset_index(drop=True)
        window, horizon, step = (35, 2, 3)
        last_i = len(close) - horizon
        for i in range(window, last_i + 1, step):
            expected_close_window = close.iloc[i - window:i].reset_index(drop=True)
            expected_feat_window = self.features_df.iloc[i - window:i].reset_index(drop=True)
            source_volume = self.df['Volume'].reset_index(drop=True).iloc[i - window:i].reset_index(drop=True)
            pd.testing.assert_series_equal(expected_feat_window['Volume'], source_volume, check_names=False)
            self.assertEqual(len(expected_close_window), len(expected_feat_window))

    def test_baseline_mode_never_touches_features_module(self):

        class RecordsFeatureArg(BaseForecastModel):
            name = 'Recorder'
            saw_features = []

            def predict(self, history, horizon=1, features=None):
                RecordsFeatureArg.saw_features.append(features)
                return [float(pd.Series(history).iloc[-1])] * horizon
        RecordsFeatureArg.saw_features.clear()
        _ = backtest_model(self.df, RecordsFeatureArg(), window=40, horizon=1, step=5)
        self.assertTrue(len(RecordsFeatureArg.saw_features) > 0)
        self.assertTrue(all((f is None for f in RecordsFeatureArg.saw_features)))

class TestRunAllModelsBacktestFeaturesDf(unittest.TestCase):

    def setUp(self):
        self.df = _synthetic_ohlcv(150, seed=13)
        self.features_df = compute_feature_frame(self.df)

    def test_features_df_none_passes_none_through(self):

        class Recorder(BaseForecastModel):
            name = 'Recorder'

            def __init__(self):
                self.seen = []

            def predict(self, history, horizon=1, features=None):
                self.seen.append(features is None)
                return [float(pd.Series(history).iloc[-1])] * horizon
        m = Recorder()
        run_all_models_backtest(self.df, {'Recorder': m}, window=40, horizon=1, step=10, features_df=None)
        self.assertTrue(all(m.seen))

    def test_features_df_given_passes_dataframe_through(self):

        class Recorder(BaseForecastModel):
            name = 'Recorder'

            def __init__(self):
                self.seen = []

            def predict(self, history, horizon=1, features=None):
                self.seen.append(features is not None and len(features) == len(history))
                return [float(pd.Series(history).iloc[-1])] * horizon
        m = Recorder()
        run_all_models_backtest(self.df, {'Recorder': m}, window=40, horizon=1, step=10, features_df=self.features_df)
        self.assertTrue(len(m.seen) > 0)
        self.assertTrue(all(m.seen))

class TestRunComparisonBacktest(unittest.TestCase):

    def setUp(self):
        self.df = _synthetic_ohlcv(150, seed=17)
        self.features_df = compute_feature_frame(self.df)

    def test_both_modes_run_and_are_labeled_separately(self):

        class Recorder(BaseForecastModel):
            name = 'Recorder'

            def __init__(self):
                self.seen_features = []

            def predict(self, history, horizon=1, features=None):
                self.seen_features.append(features is not None)
                return [float(pd.Series(history).iloc[-1])] * horizon
        baseline_model, featured_model = (Recorder(), Recorder())
        results, summaries = run_comparison_backtest(self.df, {'Recorder': (baseline_model, featured_model)}, self.features_df, window=40, horizon=1, step=10)
        self.assertTrue(all((f is False for f in baseline_model.seen_features)))
        self.assertTrue(all((f is True for f in featured_model.seen_features)))
        modes_present = set(results['Recorder']['mode'].unique())
        self.assertEqual(modes_present, {'Close-only', '+14 Features'})
        self.assertEqual(len(summaries['Recorder']), 2)
        labels = {s['mode'] for s in summaries['Recorder']}
        self.assertEqual(labels, {'Close-only', '+14 Features'})

class TestUiSerializableTimestamps(unittest.TestCase):

    def test_backtest_results_json_serializable(self):
        model = _ConstantModel()
        results = backtest_model(self._ohlcv(), model, window=20, horizon=1, step=5)
        self.assertEqual(results['datetime'].map(type).eq(str).all(), True)
        json.dumps(results.to_dict(orient='records'))

    def test_live_log_new_row_json_serializable(self):
        row = {'predicted_at': pd.Timestamp('2026-01-01 00:00:00+00:00').isoformat(), 'mode': 'Close-only', 'target_time': pd.Timestamp('2026-01-01 00:01:00+00:00').isoformat(), 'current_close': 1.0, 'predicted_close': 1.0, 'predicted_direction': 'up', 'actual_close': np.nan, 'actual_direction': None, 'correct': None}
        json.dumps(row)

    @staticmethod
    def _ohlcv(n: int=60, seed: int=3) -> pd.DataFrame:
        rng = np.random.default_rng(seed)
        close = 100 + np.cumsum(rng.normal(0, 0.5, n))
        idx = pd.date_range('2026-01-01', periods=n, freq='15min', tz='UTC')
        return pd.DataFrame({'Open': close, 'High': close + 1, 'Low': close - 1, 'Close': close, 'Volume': 1000.0}, index=idx)

class _ConstantModel(BaseForecastModel):

    def predict(self, history, horizon=1, features=None):
        return np.array([float(history.iloc[-1])] * horizon)
    name = 'ConstantModel'
if __name__ == '__main__':
    unittest.main()