Skip to content

statspai.timeseries

timeseries

Time series methods for causal inference contexts.

Provides VAR (vector autoregression), structural break tests, Granger causality, and cointegration analysis.

VARResult

Bases: ResultProtocolMixin

Results from VAR estimation.

Returned by :func:var; carries coefficient tables, the residual covariance, information criteria, and helper methods .irf(), .granger_test() and .plot_irf().

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(3)
>>> T = 200
>>> x = np.zeros(T); y = np.zeros(T)
>>> for t in range(1, T):
...     x[t] = 0.5 * x[t - 1] + rng.normal()
...     y[t] = 0.3 * y[t - 1] + 0.4 * x[t - 1] + rng.normal()
>>> df = pd.DataFrame({"y": y, "x": x})
>>> vr = sp.var(df, variables=["y", "x"], lags=2)
>>> type(vr).__name__
'VARResult'
>>> vr.var_names
['y', 'x']
>>> vr.lags
2
>>> print(vr.summary().splitlines()[0])
Vector Autoregression (VAR)
References

lutkepohl2005new

irf

irf(periods: int = 20, impulse: Optional[str] = None, response: Optional[str] = None, orthogonal: bool = True) -> Dict[str, Any]

Compute impulse response functions.

granger_test

granger_test(caused: str, causing: str) -> Dict[str, Any]

Test Granger causality.

plot_irf

plot_irf(periods: int = 20, orthogonal: bool = True, **kwargs: Any) -> Any

Plot impulse response functions.

StructuralBreakResult

Bases: ResultProtocolMixin

Results from structural break tests.

Returned by :func:structural_break. Holds the detected break point(s), the sup-F statistic(s) with their Andrews (1993) asymptotic p-value(s), and segment goodness-of-fit (RSS, BIC).

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> # mean shift halfway through the sample
>>> y = np.concatenate([rng.normal(0.0, 1.0, 100),
...                     rng.normal(3.0, 1.0, 100)])
>>> df = pd.DataFrame({"y": y})
>>> res = sp.structural_break(df, y="y", method="sup-f")
>>> isinstance(res, sp.StructuralBreakResult)
True
>>> res.test_type
'Sup-F'
>>> res.n_obs
200
>>> res.n_breaks >= 1
True

plot

plot(ax: Any = None, **kwargs: Any) -> Any

Plot with break dates marked.

CointegrationResult

Bases: ResultProtocolMixin

Results from cointegration test.

Produced by :func:engle_granger (and the Johansen routine). Exposes the test statistic(s), critical values, estimated cointegration rank and a formatted .summary().

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> T = 200
>>> x = np.cumsum(rng.normal(size=T))      # random walk
>>> y = 2.0 * x + rng.normal(size=T)       # cointegrated with x
>>> df = pd.DataFrame({"y": y, "x": x})
>>> res = sp.engle_granger(df, variables=["y", "x"])
>>> type(res).__name__
'CointegrationResult'
>>> res.test_type
'Engle-Granger'
>>> res.n_vars
2
>>> isinstance(res.summary(), str)
True
References

[@engle1987integration]

LocalProjectionsResult dataclass

Bases: ResultProtocolMixin

Impulse-response container returned by :func:local_projections.

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(3)
>>> T = 200
>>> shock = rng.normal(size=T)
>>> y = np.zeros(T)
>>> for t in range(1, T):
...     y[t] = 0.5 * y[t - 1] + 0.8 * shock[t] + rng.normal(0, 0.5)
>>> df = pd.DataFrame({"y": y, "shock": shock})
>>> res = sp.local_projections(df, outcome="y", shock="shock", horizons=8)
>>> type(res).__name__
'LocalProjectionsResult'
>>> len(res.horizons) == 9
True
>>> bool(res.irf[0] > 0)
True

GARCHResult dataclass

Bases: ResultProtocolMixin

Fitted GARCH(p,q) model returned by :func:garch.

Holds the conditional-variance parameters, the volatility path, and standardised residuals, plus :meth:forecast for multi-step variance.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> rng = np.random.default_rng(0)
>>> T = 400
>>> eps = np.zeros(T)
>>> s2 = np.ones(T)
>>> omega, a1, b1 = 0.05, 0.1, 0.85
>>> for t in range(1, T):
...     s2[t] = omega + a1 * eps[t - 1] ** 2 + b1 * s2[t - 1]
...     eps[t] = np.sqrt(s2[t]) * rng.standard_normal()
>>> res = sp.garch(eps, p=1, q=1)
>>> bool(res.persistence < 1.0)   # alpha + beta < 1 => stationary
True
>>> res.forecast(horizon=3).shape
(3,)

forecast

forecast(horizon: int = 1) -> ndarray

Multi-step ahead variance forecast (analytic recursion).

ARIMAResult dataclass

Bases: ResultProtocolMixin

Fitted ARIMA(p,d,q) / SARIMAX model returned by :func:statspai.arima.

Carries the estimated params / se (indexed by parameter name), information criteria (aic / bic / aicc), the log-likelihood, residuals and fitted values, plus inference accessors (:attr:tvalues, :attr:pvalues, :meth:conf_int) and a .summary() / .forecast() interface.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> rng = np.random.default_rng(0)
>>> n = 120
>>> y = np.zeros(n)
>>> for t in range(1, n):
...     y[t] = 0.6 * y[t - 1] + rng.normal(0, 1)
>>> res = sp.arima(y, order=(1, 0, 0))
>>> type(res).__name__
'ARIMAResult'
>>> bool(np.isfinite(res.aic))
True
>>> len(res.params)
2

std_errors property

std_errors: Series

Alias for :attr:se (regression-style naming).

tvalues property

tvalues: Series

z-statistics params / se (SARIMAX uses a normal reference).

pvalues property

pvalues: Series

Two-sided p-values from the normal reference distribution.

conf_int

conf_int(alpha: float = 0.05) -> DataFrame

Confidence intervals for each parameter.

Parameters:

Name Type Description Default
alpha float

1 - alpha is the coverage (0.05 → 95% CI).

0.05

Returns:

Type Description
DataFrame

Indexed by parameter name with lower / upper columns.

BVARResult dataclass

Bases: ResultProtocolMixin

Posterior summary of a Bayesian VAR with Minnesota prior.

Produced by :func:bvar. Holds the posterior-mean coefficient matrix and residual covariance, and exposes .forecast(), .irf() and .summary().

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> T = 120
>>> y1 = np.zeros(T); y2 = np.zeros(T)
>>> for t in range(1, T):
...     y1[t] = 0.5 * y1[t - 1] + 0.2 * y2[t - 1] + rng.normal()
...     y2[t] = 0.3 * y2[t - 1] + 0.1 * y1[t - 1] + rng.normal()
>>> df = pd.DataFrame({"gdp": y1, "inflation": y2})
>>> res = sp.bvar(df, lags=2)
>>> type(res).__name__
'BVARResult'
>>> res.coef.shape
(5, 2)
>>> res.lags
2
>>> isinstance(res.summary(), str)
True

irf

irf(shock_var: int = 0, horizon: int = 20) -> ndarray

Orthogonalised impulse responses (Cholesky decomposition).

credible_interval

credible_interval(level: float = 0.9) -> tuple[ndarray, ndarray]

Posterior credible interval for every coefficient.

Returns (lower, upper) matrices the same shape as coef, using a Normal approximation to the matrix-t marginal posterior (exact as T grows). coef_sd is the marginal posterior standard deviation sqrt(diag((X'X + V^{-1})^{-1})_i * Sigma_{kk}).

ITSResult dataclass

Bases: ResultProtocolMixin

Result container for :func:its.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> t = np.arange(60)
>>> y = 1 + 0.1 * t + (t >= 30) * 3.0 + rng.normal(scale=0.5, size=60)
>>> df = pd.DataFrame({"y": y, "t": t})
>>> res = sp.its(df, y="y", time="t", intervention=30)
>>> type(res).__name__
'ITSResult'
>>> float(res.level_change) > 0
True

to_dict

to_dict() -> dict

JSON-safe dict of every field (agent-native serialization).

granger_causality

granger_causality(var_result: Optional[VARResult] = None, data: Optional[DataFrame] = None, caused: Optional[str] = None, causing: Optional[str] = None, lags: Optional[int] = None) -> Dict[str, Any]

Granger causality test.

Tests whether causing variable Granger-causes caused variable.

Parameters:

Name Type Description Default
var_result VARResult

Pre-estimated VAR model.

None
data DataFrame

Data (if var_result not provided).

None
caused str

Variable being tested for causation.

None
causing str

Variable hypothesized to cause.

None
lags int

Number of lags (if fitting new VAR).

None

Returns:

Type Description
dict

Keys: 'F_stat', 'p_value', 'df1', 'df2', 'caused', 'causing'.

Examples:

x Granger-causes y by construction, but not vice versa:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(5)
>>> T = 200
>>> x = np.zeros(T)
>>> y = np.zeros(T)
>>> for t in range(1, T):
...     x[t] = 0.5 * x[t - 1] + rng.normal()
...     y[t] = 0.3 * y[t - 1] + 0.4 * x[t - 1] + rng.normal()
>>> df = pd.DataFrame({"y": y, "x": x})
>>> vr = sp.var(df, variables=["y", "x"], lags=2)
>>> gc = sp.granger_causality(vr, caused="y", causing="x")
>>> print(gc["p_value"] < 0.05)  # True — x helps predict y

Or fit the VAR implicitly from data:

>>> gc2 = sp.granger_causality(data=df, caused="x", causing="y", lags=2)
>>> print(gc2["reject"])  # False — y does not Granger-cause x

irf

irf(var_result: VARResult, periods: int = 20, impulse: Optional[str] = None, response: Optional[str] = None, orthogonal: bool = True) -> Dict[str, Any]

Compute impulse response functions from VAR.

Parameters:

Name Type Description Default
var_result VARResult

Estimated VAR model.

required
periods int

Number of periods for IRF.

20
impulse str

Impulse variable (if None, all).

None
response str

Response variable (if None, all).

None
orthogonal bool

Orthogonalized (Cholesky) IRF.

True

Returns:

Type Description
dict

Keys: 'irf' (dict of arrays), 'periods'.

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(3)
>>> T = 200
>>> x = np.zeros(T); y = np.zeros(T)
>>> for t in range(1, T):
...     x[t] = 0.5 * x[t - 1] + rng.normal()
...     y[t] = 0.3 * y[t - 1] + 0.4 * x[t - 1] + rng.normal()
>>> df = pd.DataFrame({"y": y, "x": x})
>>> vr = sp.var(df, variables=["y", "x"], lags=2)
>>> out = sp.irf(vr, periods=10, impulse="x", response="y")
>>> sorted(out.keys())
['irf', 'periods']
>>> len(out["irf"]["x -> y"])  # s = 0, 1, ..., 10
11
>>> out["periods"][0]
0
>>> bool(np.isfinite(out["irf"]["x -> y"]).all())
True
References

lutkepohl2005new

cusum_test

cusum_test(data: DataFrame, y: str, x: Optional[List[str]] = None, alpha: float = 0.05) -> Dict[str, Any]

CUSUM test for parameter stability.

Tests H0: parameters are stable vs H1: parameter shift.

Parameters:

Name Type Description Default
data DataFrame
required
y str

Dependent variable.

required
x list of str

Regressors.

None
alpha float
0.05

Returns:

Type Description
dict

Keys: 'cusum' (standardised CUSUM path of the recursive residuals), 'max_cusum' (its supremum in absolute value), 'critical_value' (the Brown-Durbin-Evans crossing boundary, an array a * [1 + 2 s / (n - k)] that widens from a to 3a across the sample -- not a constant), 'reject' (True if the path crosses that boundary anywhere), and 'n_obs'.

Notes

This is the recursive-residual CUSUM of Brown, Durbin & Evans (1975), as in R's strucchange::efp(type="Rec-CUSUM"). Its boundary is linear in the recursion index with coefficient a (0.948 at the 5% level), not the constant sup|Brownian-bridge| value (1.358) that belongs to the OLS-CUSUM of Ploberger & Kramer (1992).

References

brown1975techniques

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 120
>>> x = rng.normal(size=n)
>>> y = 1.0 + 0.5 * x + rng.normal(scale=0.5, size=n)  # stable relation
>>> df = pd.DataFrame({"y": y, "x": x})
>>> res = sp.cusum_test(df, y="y", x=["x"])
>>> sorted(res.keys())
['critical_value', 'cusum', 'max_cusum', 'n_obs', 'reject']
>>> res["n_obs"]
120
>>> bool(res["reject"])
False

engle_granger

engle_granger(data: DataFrame, variables: Optional[List[str]] = None, lags: Optional[int] = None, trend: str = 'c', alpha: float = 0.05) -> CointegrationResult

Engle-Granger (1987) two-step cointegration test.

Step 1: OLS regression of y on x Step 2: ADF test on residuals

Parameters:

Name Type Description Default
data DataFrame
required
variables list of str

Variables to test (first is dependent).

None
lags int

Lags for ADF test. If None, uses AIC selection.

None
trend str
'c'
alpha float
0.05

Returns:

Type Description
CointegrationResult

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> T = 200
>>> x = np.cumsum(rng.normal(size=T))      # random walk
>>> y = 2.0 * x + rng.normal(size=T)       # cointegrated with x
>>> df = pd.DataFrame({"y": y, "x": x})
>>> res = sp.engle_granger(df, variables=["y", "x"])
>>> res.test_type
'Engle-Granger'
>>> res.n_vars
2
>>> bool(res.rank in (0, 1))
True
>>> isinstance(res.summary(), str)
True
References

[@engle1987integration]

johansen

johansen(data: DataFrame, variables: Optional[List[str]] = None, lags: int = 1, trend: str = 'c', test: str = 'trace', alpha: float = 0.05) -> CointegrationResult

Johansen (1991) cointegration test.

Tests for the cointegration rank using the trace or maximum eigenvalue test statistic.

Equivalent to Stata's vecrank and R's ca.jo().

Parameters:

Name Type Description Default
data DataFrame
required
variables list of str

Variables to test.

None
lags int

Number of lags in the VECM.

1
trend str

'n' (none), 'c' (constant), 'ct' (constant + trend).

'c'
test str

'trace' or 'maxeig' (maximum eigenvalue).

'trace'
alpha float
0.05

Returns:

Type Description
CointegrationResult

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> trend = np.cumsum(rng.normal(size=200))  # common stochastic trend
>>> df = pd.DataFrame({
...     'gdp': trend + rng.normal(scale=0.5, size=200),
...     'consumption': 0.8 * trend + rng.normal(scale=0.5, size=200),
...     'investment': 0.3 * trend + rng.normal(scale=0.5, size=200),
... })
>>> result = sp.johansen(
...     df, variables=['gdp', 'consumption', 'investment'], lags=2)
>>> type(result).__name__
'CointegrationResult'
>>> isinstance(result.summary(), str)
True