sam
67d016fe25
`snapshot` writes each tracked model's forecast quantiles at the seven backtest horizons from the latest price to data/forecasts/<origin>.csv. It refuses stale data (older than two days) and duplicate dates, so snapshots can't be reconstructed after the fact; committing them dates them. `forward` scores every recorded forecast whose target date has passed, reusing the backtest's scoring (now factored out as evaluate.score). Tracked: random_walk, drift_rw, cycle, powerlaw, plus powerlaw_ou, powerlaw_ou_param and cycle_on_powerlaw, which development data couldn't settle. `just weekly` runs update, snapshot and forward. First snapshot: 2026-09-23 (BTC $84.4K). The first outcomes are due 2026-10-23.
58 lines
2.1 KiB
Python
58 lines
2.1 KiB
Python
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from btcmodel import forward
|
|
from btcmodel.evaluate import HORIZONS, score
|
|
from btcmodel.models import MODELS
|
|
|
|
from .test_models import synthetic_prices
|
|
|
|
MODELS_UNDER_TEST = [MODELS["random_walk"], MODELS["powerlaw"]]
|
|
|
|
|
|
@pytest.fixture
|
|
def prices():
|
|
return synthetic_prices(lambda day: 0.001 + 0 * day, end="2025-03-31", noise=0.03)
|
|
|
|
|
|
def test_snapshot_records_every_model_and_horizon(prices, tmp_path):
|
|
history = prices.loc[:"2024-11-26"]
|
|
path = forward.snapshot(history, MODELS_UNDER_TEST, tmp_path, pd.Timestamp("2024-11-27"))
|
|
recorded = pd.read_csv(path)
|
|
assert path.name == "2024-11-26.csv"
|
|
assert len(recorded) == len(MODELS_UNDER_TEST) * len(HORIZONS)
|
|
assert recorded.target.iloc[0] == "2024-12-26"
|
|
|
|
|
|
def test_snapshot_refuses_stale_data_and_duplicates(prices, tmp_path):
|
|
history = prices.loc[:"2024-11-26"]
|
|
with pytest.raises(ValueError, match="real time"):
|
|
forward.snapshot(history, MODELS_UNDER_TEST, tmp_path, pd.Timestamp("2024-12-10"))
|
|
forward.snapshot(history, MODELS_UNDER_TEST, tmp_path, pd.Timestamp("2024-11-27"))
|
|
with pytest.raises(FileExistsError):
|
|
forward.snapshot(history, MODELS_UNDER_TEST, tmp_path, pd.Timestamp("2024-11-27"))
|
|
|
|
|
|
def test_scores_only_due_horizons_and_match_the_backtest(prices, tmp_path):
|
|
history = prices.loc[:"2024-11-26"]
|
|
forward.snapshot(history, MODELS_UNDER_TEST, tmp_path, pd.Timestamp("2024-11-27"))
|
|
|
|
assert forward.score_snapshots(prices.loc[:"2024-12-20"], tmp_path).empty
|
|
assert forward.next_due(prices.loc[:"2024-12-20"], tmp_path) == pd.Timestamp("2024-12-26")
|
|
|
|
scores = forward.score_snapshots(prices, tmp_path)
|
|
assert sorted(scores.horizon.unique()) == [30, 91]
|
|
|
|
model = MODELS["powerlaw"]
|
|
horizons = np.array([30, 91])
|
|
direct = score(
|
|
"powerlaw",
|
|
model.forecast(history, horizons),
|
|
np.log(
|
|
prices.close.loc[history.index[-1] + pd.to_timedelta(horizons, unit="D")]
|
|
).to_numpy(),
|
|
)
|
|
recorded = scores[scores.model == "powerlaw"].sort_values("horizon")
|
|
np.testing.assert_allclose(recorded.crps, direct.crps, atol=1e-5)
|