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()