Source code for statsmodels.regression.dimred
import warnings
import numpy as np
import pandas as pd
from statsmodels.base import model
import statsmodels.base.wrapper as wrap
from statsmodels.tools.sm_exceptions import ConvergenceWarning
class _DimReductionRegression(model.Model):
"""
A base class for dimension reduction regression methods.
"""
def __init__(self, endog, exog, **kwargs):
super().__init__(endog, exog, **kwargs)
def _prep(self, n_slice):
# Sort the data by endog
ii = np.argsort(self.endog)
x = self.exog[ii, :]
# Whiten the data
x -= x.mean(0)
covx = np.dot(x.T, x) / x.shape[0]
covxr = np.linalg.cholesky(covx)
x = np.linalg.solve(covxr, x.T).T
self.wexog = x
self._covxr = covxr
# Split the data into slices
self._split_wexog = np.array_split(x, n_slice)
[docs]
class SlicedInverseReg(_DimReductionRegression):
"""
Sliced Inverse Regression (SIR)
Parameters
----------
endog : array_like (1d)
The dependent variable
exog : array_like (2d)
The covariates
References
----------
KC Li (1991). Sliced inverse regression for dimension reduction.
JASA 86, 316-342.
"""
[docs]
def fit(self, slice_n=20, **kwargs):
"""
Estimate the EDR space using Sliced Inverse Regression.
Parameters
----------
slice_n : int, optional
Target number of observations per slice
"""
# Sample size per slice
if len(kwargs) > 0:
msg = "SIR.fit does not take any extra keyword arguments"
warnings.warn(msg)
# Number of slices
n_slice = self.exog.shape[0] // slice_n
self._prep(n_slice)
mn = [z.mean(0) for z in self._split_wexog]
n = [z.shape[0] for z in self._split_wexog]
mn = np.asarray(mn)
n = np.asarray(n)
# Estimate Cov E[X | Y=y]
mnc = np.dot(mn.T, n[:, None] * mn) / n.sum()
a, b = np.linalg.eigh(mnc)
jj = np.argsort(-a)
a = a[jj]
b = b[:, jj]
params = np.linalg.solve(self._covxr.T, b)
results = DimReductionResults(self, params, eigs=a)
return DimReductionResultsWrapper(results)
def _regularized_objective(self, A):
# The objective function for regularized SIR
p = self.k_vars
covx = self._covx
mn = self._slice_means
ph = self._slice_props
v = 0
A = np.reshape(A, (p, self.ndim))
# The penalty
for k in range(self.ndim):
u = np.dot(self.pen_mat, A[:, k])
v += np.sum(u * u)
# The SIR objective function
covxa = np.dot(covx, A)
q, _ = np.linalg.qr(covxa)
qd = np.dot(q, np.dot(q.T, mn.T))
qu = mn.T - qd
v += np.dot(ph, (qu * qu).sum(0))
return v
def _regularized_grad(self, A):
# The gradient of the objective function for regularized SIR
p = self.k_vars
ndim = self.ndim
covx = self._covx
n_slice = self.n_slice
mn = self._slice_means
ph = self._slice_props
A = A.reshape((p, ndim))
# Penalty gradient
gr = 2 * np.dot(self.pen_mat.T, np.dot(self.pen_mat, A))
A = A.reshape((p, ndim))
covxa = np.dot(covx, A)
covx2a = np.dot(covx, covxa)
Q = np.dot(covxa.T, covxa)
Qi = np.linalg.inv(Q)
jm = np.zeros((p, ndim))
qcv = np.linalg.solve(Q, covxa.T)
ft = [None] * (p * ndim)
for q in range(p):
for r in range(ndim):
jm *= 0
jm[q, r] = 1
umat = np.dot(covx2a.T, jm)
umat += umat.T
umat = -np.dot(Qi, np.dot(umat, Qi))
fmat = np.dot(np.dot(covx, jm), qcv)
fmat += np.dot(covxa, np.dot(umat, covxa.T))
fmat += np.dot(covxa, np.linalg.solve(Q, np.dot(jm.T, covx)))
ft[q*ndim + r] = fmat
ch = np.linalg.solve(Q, np.dot(covxa.T, mn.T))
cu = mn - np.dot(covxa, ch).T
for i in range(n_slice):
u = cu[i, :]
v = mn[i, :]
for q in range(p):
for r in range(ndim):
f = np.dot(u, np.dot(ft[q*ndim + r], v))
gr[q, r] -= 2 * ph[i] * f
return gr.ravel()
[docs]
def fit_regularized(self, ndim=1, pen_mat=None, slice_n=20, maxiter=100,
gtol=1e-3, **kwargs):
"""
Estimate the EDR space using regularized SIR.
Parameters
----------
ndim : int
The number of EDR directions to estimate
pen_mat : array_like
A 2d array such that the squared Frobenius norm of
`dot(pen_mat, dirs)`` is added to the objective function,
where `dirs` is an orthogonal array whose columns span
the estimated EDR space.
slice_n : int, optional
Target number of observations per slice
maxiter :int
The maximum number of iterations for estimating the EDR
space.
gtol : float
If the norm of the gradient of the objective function
falls below this value, the algorithm has converged.
Returns
-------
A results class instance.
Notes
-----
If each row of `exog` can be viewed as containing the values of a
function evaluated at equally-spaced locations, then setting the
rows of `pen_mat` to [[1, -2, 1, ...], [0, 1, -2, 1, ..], ...]
will give smooth EDR coefficients. This is a form of "functional
SIR" using the squared second derivative as a penalty.
References
----------
L. Ferre, A.F. Yao (2003). Functional sliced inverse regression
analysis. Statistics: a journal of theoretical and applied
statistics 37(6) 475-488.
"""
if len(kwargs) > 0:
msg = "SIR.fit_regularized does not take keyword arguments"
warnings.warn(msg)
if pen_mat is None:
raise ValueError("pen_mat is a required argument")
start_params = kwargs.get("start_params", None)
# Sample size per slice
slice_n = kwargs.get("slice_n", 20)
# Number of slices
n_slice = self.exog.shape[0] // slice_n
# Sort the data by endog
ii = np.argsort(self.endog)
x = self.exog[ii, :]
x -= x.mean(0)
covx = np.cov(x.T)
# Split the data into slices
split_exog = np.array_split(x, n_slice)
mn = [z.mean(0) for z in split_exog]
n = [z.shape[0] for z in split_exog]
mn = np.asarray(mn)
n = np.asarray(n)
self._slice_props = n / n.sum()
self.ndim = ndim
self.k_vars = covx.shape[0]
self.pen_mat = pen_mat
self._covx = covx
self.n_slice = n_slice
self._slice_means = mn
if start_params is None:
params = np.zeros((self.k_vars, ndim))
params[0:ndim, 0:ndim] = np.eye(ndim)
params = params
else:
if start_params.shape[1] != ndim:
msg = "Shape of start_params is not compatible with ndim"
raise ValueError(msg)
params = start_params
params, _, cnvrg = _grass_opt(params, self._regularized_objective,
self._regularized_grad, maxiter, gtol)
if not cnvrg:
g = self._regularized_grad(params.ravel())
gn = np.sqrt(np.dot(g, g))
msg = "SIR.fit_regularized did not converge, |g|=%f" % gn
warnings.warn(msg)
results = DimReductionResults(self, params, eigs=None)
return DimReductionResultsWrapper(results)
[docs]
class PrincipalHessianDirections(_DimReductionRegression):
"""
Principal Hessian Directions (PHD)
Parameters
----------
endog : array_like (1d)
The dependent variable
exog : array_like (2d)
The covariates
Returns
-------
A model instance. Call `fit` to obtain a results instance,
from which the estimated parameters can be obtained.
References
----------
KC Li (1992). On Principal Hessian Directions for Data
Visualization and Dimension Reduction: Another application
of Stein's lemma. JASA 87:420.
"""
[docs]
def fit(self, **kwargs):
"""
Estimate the EDR space using PHD.
Parameters
----------
resid : bool, optional
If True, use least squares regression to remove the
linear relationship between each covariate and the
response, before conducting PHD.
Returns
-------
A results instance which can be used to access the estimated
parameters.
"""
resid = kwargs.get("resid", False)
y = self.endog - self.endog.mean()
x = self.exog - self.exog.mean(0)
if resid:
from statsmodels.regression.linear_model import OLS
r = OLS(y, x).fit()
y = r.resid
cm = np.einsum('i,ij,ik->jk', y, x, x)
cm /= len(y)
cx = np.cov(x.T)
cb = np.linalg.solve(cx, cm)
a, b = np.linalg.eig(cb)
jj = np.argsort(-np.abs(a))
a = a[jj]
params = b[:, jj]
results = DimReductionResults(self, params, eigs=a)
return DimReductionResultsWrapper(results)
[docs]
class SlicedAverageVarianceEstimation(_DimReductionRegression):
"""
Sliced Average Variance Estimation (SAVE)
Parameters
----------
endog : array_like (1d)
The dependent variable
exog : array_like (2d)
The covariates
bc : bool, optional
If True, use the bias-corrected CSAVE method of Li and Zhu.
References
----------
RD Cook. SAVE: A method for dimension reduction and graphics
in regression.
http://www.stat.umn.edu/RegGraph/RecentDev/save.pdf
Y Li, L-X Zhu (2007). Asymptotics for sliced average
variance estimation. The Annals of Statistics.
https://arxiv.org/pdf/0708.0462.pdf
"""
def __init__(self, endog, exog, **kwargs):
super(SAVE, self).__init__(endog, exog, **kwargs)
self.bc = False
if "bc" in kwargs and kwargs["bc"] is True:
self.bc = True
[docs]
def fit(self, **kwargs):
"""
Estimate the EDR space.
Parameters
----------
slice_n : int
Number of observations per slice
"""
# Sample size per slice
slice_n = kwargs.get("slice_n", 50)
# Number of slices
n_slice = self.exog.shape[0] // slice_n
self._prep(n_slice)
cv = [np.cov(z.T) for z in self._split_wexog]
ns = [z.shape[0] for z in self._split_wexog]
p = self.wexog.shape[1]
if not self.bc:
# Cook's original approach
vm = 0
for w, cvx in zip(ns, cv):
icv = np.eye(p) - cvx
vm += w * np.dot(icv, icv)
vm /= len(cv)
else:
# The bias-corrected approach of Li and Zhu
# \Lambda_n in Li, Zhu
av = 0
for c in cv:
av += np.dot(c, c)
av /= len(cv)
# V_n in Li, Zhu
vn = 0
for x in self._split_wexog:
r = x - x.mean(0)
for i in range(r.shape[0]):
u = r[i, :]
m = np.outer(u, u)
vn += np.dot(m, m)
vn /= self.exog.shape[0]
c = np.mean(ns)
k1 = c * (c - 1) / ((c - 1)**2 + 1)
k2 = (c - 1) / ((c - 1)**2 + 1)
av2 = k1 * av - k2 * vn
vm = np.eye(p) - 2 * sum(cv) / len(cv) + av2
a, b = np.linalg.eigh(vm)
jj = np.argsort(-a)
a = a[jj]
b = b[:, jj]
params = np.linalg.solve(self._covxr.T, b)
results = DimReductionResults(self, params, eigs=a)
return DimReductionResultsWrapper(results)
[docs]
class DimReductionResults(model.Results):
"""
Results class for a dimension reduction regression.
Notes
-----
The `params` attribute is a matrix whose columns span
the effective dimension reduction (EDR) space. Some
methods produce a corresponding set of eigenvalues
(`eigs`) that indicate how much information is contained
in each basis direction.
"""
def __init__(self, model, params, eigs):
super().__init__(
model, params)
self.eigs = eigs
class DimReductionResultsWrapper(wrap.ResultsWrapper):
_attrs = {
'params': 'columns',
}
_wrap_attrs = _attrs
wrap.populate_wrapper(DimReductionResultsWrapper, # noqa:E305
DimReductionResults)
def _grass_opt(params, fun, grad, maxiter, gtol):
"""
Minimize a function on a Grassmann manifold.
Parameters
----------
params : array_like
Starting value for the optimization.
fun : function
The function to be minimized.
grad : function
The gradient of fun.
maxiter : int
The maximum number of iterations.
gtol : float
Convergence occurs when the gradient norm falls below this value.
Returns
-------
params : array_like
The minimizing value for the objective function.
fval : float
The smallest achieved value of the objective function.
cnvrg : bool
True if the algorithm converged to a limit point.
Notes
-----
`params` is 2-d, but `fun` and `grad` should take 1-d arrays
`params.ravel()` as arguments.
Reference
---------
A Edelman, TA Arias, ST Smith (1998). The geometry of algorithms with
orthogonality constraints. SIAM J Matrix Anal Appl.
http://math.mit.edu/~edelman/publications/geometry_of_algorithms.pdf
"""
p, d = params.shape
params = params.ravel()
f0 = fun(params)
cnvrg = False
for _ in range(maxiter):
# Project the gradient to the tangent space
g = grad(params)
g -= np.dot(g, params) * params / np.dot(params, params)
if np.sqrt(np.sum(g * g)) < gtol:
cnvrg = True
break
gm = g.reshape((p, d))
u, s, vt = np.linalg.svd(gm, 0)
paramsm = params.reshape((p, d))
pa0 = np.dot(paramsm, vt.T)
def geo(t):
# Parameterize the geodesic path in the direction
# of the gradient as a function of a real value t.
pa = pa0 * np.cos(s * t) + u * np.sin(s * t)
return np.dot(pa, vt).ravel()
# Try to find a downhill step along the geodesic path.
step = 2.
while step > 1e-10:
pa = geo(-step)
f1 = fun(pa)
if f1 < f0:
params = pa
f0 = f1
break
step /= 2
params = params.reshape((p, d))
return params, f0, cnvrg
class CovarianceReduction(_DimReductionRegression):
"""
Dimension reduction for covariance matrices (CORE).
Parameters
----------
endog : array_like
The dependent variable, treated as group labels
exog : array_like
The independent variables.
dim : int
The dimension of the subspace onto which the covariance
matrices are projected.
Returns
-------
A model instance. Call `fit` on the model instance to obtain
a results instance, which contains the fitted model parameters.
Notes
-----
This is a likelihood-based dimension reduction procedure based
on Wishart models for sample covariance matrices. The goal
is to find a projection matrix P so that C_i | P'C_iP and
C_j | P'C_jP are equal in distribution for all i, j, where
the C_i are the within-group covariance matrices.
The model and methodology are as described in Cook and Forzani.
The optimization method follows Edelman et. al.
References
----------
DR Cook, L Forzani (2008). Covariance reducing models: an alternative
to spectral modeling of covariance matrices. Biometrika 95:4.
A Edelman, TA Arias, ST Smith (1998). The geometry of algorithms with
orthogonality constraints. SIAM J Matrix Anal Appl.
http://math.mit.edu/~edelman/publications/geometry_of_algorithms.pdf
"""
def __init__(self, endog, exog, dim):
super().__init__(endog, exog)
covs, ns = [], []
df = pd.DataFrame(self.exog, index=self.endog)
for _, v in df.groupby(df.index):
covs.append(v.cov().values)
ns.append(v.shape[0])
self.nobs = len(endog)
# The marginal covariance
covm = 0
for i, _ in enumerate(covs):
covm += covs[i] * ns[i]
covm /= self.nobs
self.covm = covm
self.covs = covs
self.ns = ns
self.dim = dim
def loglike(self, params):
"""
Evaluate the log-likelihood
Parameters
----------
params : array_like
The projection matrix used to reduce the covariances, flattened
to 1d.
Returns the log-likelihood.
"""
p = self.covm.shape[0]
proj = params.reshape((p, self.dim))
c = np.dot(proj.T, np.dot(self.covm, proj))
_, ldet = np.linalg.slogdet(c)
f = self.nobs * ldet / 2
for j, c in enumerate(self.covs):
c = np.dot(proj.T, np.dot(c, proj))
_, ldet = np.linalg.slogdet(c)
f -= self.ns[j] * ldet / 2
return f
def score(self, params):
"""
Evaluate the score function.
Parameters
----------
params : array_like
The projection matrix used to reduce the covariances,
flattened to 1d.
Returns the score function evaluated at 'params'.
"""
p = self.covm.shape[0]
proj = params.reshape((p, self.dim))
c0 = np.dot(proj.T, np.dot(self.covm, proj))
cP = np.dot(self.covm, proj)
g = self.nobs * np.linalg.solve(c0, cP.T).T
for j, c in enumerate(self.covs):
c0 = np.dot(proj.T, np.dot(c, proj))
cP = np.dot(c, proj)
g -= self.ns[j] * np.linalg.solve(c0, cP.T).T
return g.ravel()
def fit(self, start_params=None, maxiter=200, gtol=1e-4):
"""
Fit the covariance reduction model.
Parameters
----------
start_params : array_like
Starting value for the projection matrix. May be
rectangular, or flattened.
maxiter : int
The maximum number of gradient steps to take.
gtol : float
Convergence criterion for the gradient norm.
Returns
-------
A results instance that can be used to access the
fitted parameters.
"""
p = self.covm.shape[0]
d = self.dim
# Starting value for params
if start_params is None:
params = np.zeros((p, d))
params[0:d, 0:d] = np.eye(d)
params = params
else:
params = start_params
# _grass_opt is designed for minimization, we are doing maximization
# here so everything needs to be flipped.
params, llf, cnvrg = _grass_opt(params, lambda x: -self.loglike(x),
lambda x: -self.score(x), maxiter,
gtol)
llf *= -1
if not cnvrg:
g = self.score(params.ravel())
gn = np.sqrt(np.sum(g * g))
msg = "CovReduce optimization did not converge, |g|=%f" % gn
warnings.warn(msg, ConvergenceWarning)
results = DimReductionResults(self, params, eigs=None)
results.llf = llf
return DimReductionResultsWrapper(results)
# aliases for expert users
SIR = SlicedInverseReg
PHD = PrincipalHessianDirections
SAVE = SlicedAverageVarianceEstimation
CORE = CovarianceReduction
Last update:
Dec 16, 2024