Skip to content

Commit

Permalink
TST: Add initial tests for append
Browse files Browse the repository at this point in the history
  • Loading branch information
bashtage committed Jul 15, 2023
1 parent 478d0f6 commit 4cd013d
Show file tree
Hide file tree
Showing 3 changed files with 182 additions and 6 deletions.
167 changes: 167 additions & 0 deletions arch/tests/univariate/test_append.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
import datetime as dt
from functools import partial
from itertools import product

import numpy as np
from numpy.random import RandomState
from numpy.testing import assert_allclose
import pandas as pd
import pytest

from arch.data import sp500
from arch.univariate import (
APARCH,
ARX,
EGARCH,
FIGARCH,
GARCH,
HARCH,
HARX,
LS,
ConstantMean,
ConstantVariance,
EWMAVariance,
MIDASHyperbolic,
RiskMetrics2006,
ZeroMean,
arch_model,
)

SP500 = 100 * sp500.load()["Adj Close"].pct_change().dropna()
N = SP500.shape[0]
SP500_initial = SP500.iloc[: N // 2]
SP500_append = SP500.iloc[N // 2 :]


class HARXWrapper(HARX):
def __init__(self, y, x=None, volatility=None):
super().__init__(y, lags=[1, 5], x=x, volatility=volatility)


class ARXWrapper(ARX):
def __init__(self, y, x=None, volatility=None):
super().__init__(y, lags=2, x=x, volatility=volatility)


MEAN_MODELS = [
HARXWrapper,
ARXWrapper,
ConstantMean,
ZeroMean,
]

VOLATILITIES = [
ConstantVariance(),
GARCH(),
FIGARCH(),
EWMAVariance(lam=0.94),
MIDASHyperbolic(),
HARCH(lags=[1, 5, 22]),
RiskMetrics2006(),
APARCH(),
EGARCH(),
]

X_MEAN_MODELS = [HARXWrapper, ARXWrapper, LS]

MODEL_SPECS = list(product(MEAN_MODELS, VOLATILITIES))

IDS = [f"{mean.__name__}-{str(vol).split('(')[0]}" for mean, vol in MODEL_SPECS]


@pytest.fixture(params=MODEL_SPECS, ids=IDS)
def mean_volatility(request):
mean, vol = request.param
return mean, vol


def test_append():
mod = arch_model(SP500_initial)
mod.append(SP500_append)
res = mod.fit(disp="off")

direct = arch_model(SP500)
res_direct = direct.fit(disp="off")
assert_allclose(res.params, res_direct.params, rtol=1e-5)
assert_allclose(res.conditional_volatility, res_direct.conditional_volatility)
assert_allclose(res.resid, res_direct.resid)
assert_allclose(mod._backcast, direct._backcast)


def test_alt_means(mean_volatility):
mean, vol = mean_volatility
mod = mean(SP500_initial, volatility=vol)
mod.append(SP500_append)
res = mod.fit(disp="off")

direct = mean(SP500, volatility=vol)
res_direct = direct.fit(disp="off")
assert_allclose(res.conditional_volatility, res_direct.conditional_volatility)
assert_allclose(res.resid, res_direct.resid)
if mod._backcast is not None:
assert_allclose(mod._backcast, direct._backcast)
else:
assert direct._backcast is None


def test_append_scalar_no_reestiamtion(mean_volatility):
mean, vol = mean_volatility
mod = mean(np.asarray(SP500_initial), volatility=vol)
for val in np.asarray(SP500_append):
mod.append(val)


def test_append_scalar_bad_value():
mod = HARX(SP500_initial, lags=[1, 5], volatility=GARCH())
with pytest.raises(TypeError):
mod.append(SP500_append.iloc[0])


def test_append_type_mismatch(mean_volatility):
mean, vol = mean_volatility
mod = mean(SP500_initial, volatility=vol)
with pytest.raises(TypeError, match="Input data must be the same"):
mod.append(np.asarray(SP500_append))
with pytest.raises(TypeError, match="Input data must be the same"):
mod.append(SP500_append.tolist())

mod_arr = mean(np.asarray(SP500_initial), volatility=vol)
with pytest.raises(TypeError, match="Input data must be the same"):
mod_arr.append(SP500_append)
with pytest.raises(TypeError, match="Input data must be the same"):
mod_arr.append(SP500_append.tolist())

mod_list = mean(SP500_initial.tolist(), volatility=vol)
with pytest.raises(TypeError, match="Input data must be the same"):
mod_list.append(SP500_append)
with pytest.raises(TypeError, match="Input data must be the same"):
mod_list.append(np.asarray(SP500_append))


def test_append_x_type_mismatch():
pass


@pytest.mark.parametrize("mean", X_MEAN_MODELS)
def test_bad_append_model_with_exog(mean):
mod = mean(SP500_initial, volatility=GARCH())
x = pd.DataFrame(
np.random.randn(SP500_append.shape[0], 2),
columns=["a", "b"],
index=SP500_append.index,
)
with pytest.raises(ValueError, match=""):
mod.append(SP500_append, x=x)

x_initial = pd.DataFrame(
np.random.randn(SP500_initial.shape[0], 2),
columns=["a", "b"],
index=SP500_initial.index,
)
mod = mean(SP500_initial, x=x_initial, volatility=GARCH())
with pytest.raises(ValueError, match=""):
mod.append(SP500_append)


def test_bad_append_ls():
pass
2 changes: 1 addition & 1 deletion arch/univariate/mean.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,7 @@ def append(self, y: ArrayLike, x: ArrayLike2D | None = None) -> None:
if self._x.shape[0] != self._y.shape[0]:
raise ValueError("x must have the same number of observations as y")

self._init_model()
self._init_model()

def _scale_changed(self):
"""
Expand Down
19 changes: 14 additions & 5 deletions arch/utility/array.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from typing import Any, Literal, overload

import numpy as np
import pandas as pd
from pandas import (
DataFrame,
DatetimeIndex,
Expand Down Expand Up @@ -322,17 +323,25 @@ def find_index(s: AnyPandas, index: int | DateLike) -> int:


def append_same_type(original, new):
if not isinstance(new, type(original)):
append_ok = isinstance(original, (list, np.ndarray)) and isinstance(
new, (float, np.floating)
)
append_ok = append_ok or isinstance(new, type(original))
if not append_ok:
raise TypeError(
"Input data must be the same type as the original data. "
f"Got {type(new)}, expected {type(original)}."
"Input data must be the same type as the original data, unless the "
"original was an NDArray or a list, in which case the input data can "
f"be a scalar float. Got {type(new)}, expected {type(original)}."
)
if isinstance(original, (Series, DataFrame)):
extended = concat([original, new], axis=0)
elif isinstance(original, np.ndarray):
extended = np.concatenate([original, new])
extended = np.concatenate([original, np.atleast_1d(new)])
elif isinstance(original, list):
extended = original + new
if isinstance(new, list):
extended = original + new
else:
extended = original + [new]
else:
raise TypeError(
"Input data must be a pandas Series, DataFrame, numpy ndarray, or "
Expand Down

0 comments on commit 4cd013d

Please sign in to comment.