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
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
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 ¶
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
conf_int ¶
Confidence intervals for each parameter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
alpha
|
float
|
|
0.05
|
Returns:
| Type | Description |
|---|---|
DataFrame
|
Indexed by parameter name with |
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 ¶
Orthogonalised impulse responses (Cholesky decomposition).
credible_interval ¶
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
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:
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: |
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