60 lines
2.7 KiB
Python
60 lines
2.7 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""Integration tests: chronological walk-forward + hybrid ablation + serialization."""
|
||
|
|
import numpy as np
|
||
|
|
|
||
|
|
from src.forecasting.target import ForecastContext
|
||
|
|
from src.forecasting.interface import Series
|
||
|
|
from src.baselines import NaiveBaseline
|
||
|
|
from src.sax import SaxConfig, SaxAnalogForecaster
|
||
|
|
from src.arima import ArimaConfig, ArimaModel
|
||
|
|
from src.validation import WalkForwardConfig, run_walk_forward
|
||
|
|
from src.hybrid import make_hybrid_record
|
||
|
|
|
||
|
|
|
||
|
|
def _trend_series(n=400):
|
||
|
|
close = np.cumsum(np.random.default_rng(7).normal(0.0, 1.0, n)) + 100.0
|
||
|
|
return Series(symbol="XAUUSD", timeframe="H1",
|
||
|
|
timestamp=[f"ts-{i:06d}" for i in range(n)],
|
||
|
|
open=list(close), high=list(close + 0.5), low=list(close - 0.5),
|
||
|
|
close=list(close))
|
||
|
|
|
||
|
|
|
||
|
|
def test_walkforward_is_chronological_and_reveals_outcome():
|
||
|
|
s = _trend_series(300)
|
||
|
|
ctx = ForecastContext(symbol="XAUUSD", timeframe="H1", horizon=10, atr_period=20)
|
||
|
|
cfg = WalkForwardConfig(min_origin=120, n_forecast_points=25)
|
||
|
|
recs = run_walk_forward(NaiveBaseline(), s, ctx, cfg)
|
||
|
|
assert len(recs) == 25
|
||
|
|
for r in recs:
|
||
|
|
assert r.outcome_boundary == r.forecast_origin + 10
|
||
|
|
assert r.actual_forward_return is not None
|
||
|
|
assert r.actual_forward_return_ATR is not None
|
||
|
|
assert r.prediction_timestamp < r.actual_outcome_timestamp # strict order
|
||
|
|
|
||
|
|
|
||
|
|
def test_sax_can_run_walkforward_producing_records():
|
||
|
|
s = _trend_series(400)
|
||
|
|
ctx = ForecastContext(symbol="XAUUSD", timeframe="H1", horizon=6, atr_period=20)
|
||
|
|
model = SaxAnalogForecaster(SaxConfig(window_length=16, min_analogs=5,
|
||
|
|
top_k_analogs=8))
|
||
|
|
recs = run_walk_forward(model, s, ctx, WalkForwardConfig(min_origin=100, n_forecast_points=15))
|
||
|
|
assert len(recs) == 15
|
||
|
|
assert all(r.actual_forward_return_ATR is not None for r in recs)
|
||
|
|
|
||
|
|
|
||
|
|
def test_arima_and_sax_records_serialize():
|
||
|
|
s = _trend_series(300)
|
||
|
|
ctx = ForecastContext(symbol="XAUUSD", timeframe="H1", horizon=3, atr_period=20)
|
||
|
|
arima = run_walk_forward(ArimaModel(ArimaConfig(p=1, d=0, q=0, fit_window=150)),
|
||
|
|
s, ctx, WalkForwardConfig(min_origin=180, n_forecast_points=8))
|
||
|
|
sax = run_walk_forward(SaxAnalogForecaster(SaxConfig(window_length=16, min_analogs=5)),
|
||
|
|
s, ctx, WalkForwardConfig(min_origin=180, n_forecast_points=8))
|
||
|
|
assert len(arima) == len(sax) == 8
|
||
|
|
merged = make_hybrid_record(arima[0], sax[0])
|
||
|
|
d = merged.to_dict()
|
||
|
|
assert d["hybrid_state"] in {
|
||
|
|
"strong_agreement", "disagreement", "partial_evidence",
|
||
|
|
"no_edge", "insufficient_evidence",
|
||
|
|
}
|
||
|
|
import json
|
||
|
|
json.dumps(d)
|