File size: 3,696 Bytes
ef20ebe | 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 | """
Unlike test_moirai_model.py / test_chronos_model.py, these tests
exercise AutoArimaForecaster's predict() for REAL -- statsforecast has
no torch/network dependency, so there's no reason to stub it out.
"""
import numpy as np
import pandas as pd
from arima_model import AutoArimaForecaster
from fixtures import make_synthetic_ohlcv
from forecast_types import ModelInputError, QuantileForecast
def test_predict_returns_a_real_quantile_forecast():
df = make_synthetic_ohlcv(n=150, seed=60, freq="1h")
model = AutoArimaForecaster()
forecast = model.predict(context_df=df, target_col="close", past_covariate_cols=[],
prediction_length=5, freq="1h")
assert isinstance(forecast, QuantileForecast)
assert forecast.quantiles.shape == (5, 5)
assert forecast.quantile_levels == [0.1, 0.25, 0.5, 0.75, 0.9]
assert forecast.context_length_used == 150
def test_quantiles_are_monotonically_non_decreasing_per_step():
"""P10 <= P25 <= P50 <= P75 <= P90 must hold at every horizon step --
a real property of valid prediction intervals, not just "some
numbers came back"."""
df = make_synthetic_ohlcv(n=150, seed=61, freq="1h")
model = AutoArimaForecaster()
forecast = model.predict(context_df=df, target_col="close", past_covariate_cols=[],
prediction_length=5, freq="1h")
for step in range(5):
row = forecast.quantiles[step]
assert np.all(np.diff(row) >= -1e-9), f"step {step} quantiles not monotonic: {row}"
def test_rejects_too_short_context():
df = make_synthetic_ohlcv(n=5, seed=62, freq="1h")
model = AutoArimaForecaster(min_context=10)
try:
model.predict(context_df=df, target_col="close", past_covariate_cols=[],
prediction_length=5, freq="1h")
raise AssertionError("expected ModelInputError")
except ModelInputError as e:
assert "too short" in str(e)
def test_rejects_nan_in_context():
df = make_synthetic_ohlcv(n=150, seed=63, freq="1h")
df = df.copy()
df.iloc[10, df.columns.get_loc("close")] = np.nan
model = AutoArimaForecaster()
try:
model.predict(context_df=df, target_col="close", past_covariate_cols=[],
prediction_length=5, freq="1h")
raise AssertionError("expected ModelInputError")
except ModelInputError as e:
assert "NaN" in str(e)
def test_predict_is_deterministic_for_identical_input():
"""ARIMA fitting is deterministic (no sampling) -- two calls on the
exact same context must return the exact same forecast, unlike
Moirai-MoE before its seeding fix."""
df = make_synthetic_ohlcv(n=150, seed=64, freq="1h")
model = AutoArimaForecaster()
f1 = model.predict(context_df=df, target_col="close", past_covariate_cols=[], prediction_length=5, freq="1h")
f2 = model.predict(context_df=df, target_col="close", past_covariate_cols=[], prediction_length=5, freq="1h")
assert np.allclose(f1.quantiles, f2.quantiles)
def test_works_through_the_real_backtest_engine():
"""End-to-end: AutoArimaForecaster must be usable as a drop-in
WalkForwardBacktester forecaster with no special-casing, proving
the shared interface actually holds."""
from backtest_engine import WalkForwardBacktester
df = make_synthetic_ohlcv(n=150, seed=65, freq="1h")
from features import compute_indicators
feat = compute_indicators(df)
bt = WalkForwardBacktester(AutoArimaForecaster(min_context=10), timeframe="1h",
horizon=5, context_length=32)
result = bt.run(feat)
assert result["num_origins"] > 0
assert len(result["errors"]) == 0
|