| """ |
| 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 |
|
|
