Spaces:
Sleeping
Sleeping
File size: 8,333 Bytes
9831ced a9e6751 9831ced a9e6751 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 | 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, origin_start=70)
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()
|