Projects/pyblm
Projects
/
pyblm
Archived
5
0
Fork 0
This repository has been archived on 2024-07-04. You can view files and clone it, but cannot push or open issues or pull requests.
pyblm/engines.py

299 lines
8.0 KiB
Python

"""Module contain algorithms for low-rank L-shaped model.
"""
__all__ = ['nipals_lpls']
__docformat__ = "restructuredtext en"
from math import sqrt as msqrt
from numpy import dot,empty,zeros,apply_along_axis,newaxis,finfo
from numpy.linalg import inv
def nipals_lpls(X, Y, Z, a_max, alpha=.7, mean_ctr=[2, 0, 1], scale='scores', verbose=False):
""" L-shaped Partial Least Sqaures Regression by the nipals algorithm.
An L-shaped low rank model aproximates three matrices in a hyploid
structure. That means that the main matrix (X), has one matrix asociated
with its row space and one to its column space. A simultanously low rank estiamte
of these three matrices tries to discover common directions/subspaces.
*Parameters*:
X : {array}
Main data matrix (m, n)
Y : {array}
External row data (m, l)
Z : {array}
External column data (n, o)
a_max : {integer}
Maximum number of components to calculate (0, min(m,n))
alpha : {float}, optional
Parameter to control the amount of influence from Z-matrix.
0 is none, which returns a pls-solution, 1 is max
mean_center : {array-like}, optional
A three element array-like structure with elements in [-1,0,1,2],
that decides the type of centering used.
-1 : nothing
0 : row center
1 : column center
2 : double center
scale : {'scores', 'loads'}, optional
Option to decide on where the scale goes.
verbose : {boolean}, optional
Verbosity of console output. For use in debugging.
*Returns*:
T : {array}
X-scores
W : {array}
X-weights/Z-weights
P : {array}
X-loadings
Q : {array}
Y-loadings
U : {array}
X-Y relation
L : {array}
Z-scores
K : {array}
Z-loads
B : {array}
Regression coefficients X->Y
evx : {array}
X-explained variance
evy : {array}
Y-explained variance
evz : {array}
Z-explained variance
mnx : {array}
X location
mny : {array}
Y location
mnz : {array}
Z location
*References*
Saeboe et al., LPLS-regression: a method for improved prediction and
classification through inclusion of background information on
predictor variables, J. of chemometrics and intell. laboratory syst.
Martens et.al, Regression of a data matrix on descriptors of
both its rows and of its columns via latent variables: L-PLSR,
Computational statistics & data analysis, 2005
"""
m, n = X.shape
k, l = Y.shape
u, o = Z.shape
max_rank = min(m, n)
assert (a_max>0 and a_max<max_rank), "Number of comp error:\
tried:%d, max_rank:%d" %(a_max,max_rank)
if mean_ctr!=None:
xctr, yctr, zctr = mean_ctr
X, mnX = center(X, xctr)
Y, mnY = center(Y, yctr)
Z, mnZ = center(Z, zctr)
varX = (X**2).sum()
varY = (Y**2).sum()
varZ = (Z**2).sum()
# initialize
U = empty((k, a_max))
Q = empty((l, a_max))
T = zeros((m, a_max))
W = empty((n, a_max))
P = empty((n, a_max))
K = empty((o, a_max))
L = empty((u, a_max))
B = empty((a_max, n, l))
E = X.copy()
F = Y.copy()
G = Z.copy()
#b0 = empty((a_max, 1, l))
var_x = empty((a_max,))
var_y = empty((a_max,))
var_z = empty((a_max,))
MAX_ITER = 450
LIM = finfo(X.dtype).resolution
is_rd = False
for a in range(a_max):
if verbose:
print "\nWorking on comp. %s" %a
u = F[:,:1]
diff = 1
niter = 0
while (diff>LIM and niter<MAX_ITER):
niter += 1
u1 = u.copy()
w = dot(E.T, u)
wn = msqrt(dot(w.T, w))
if wn < LIM:
print "Rank exhausted in X! Comp: %d " %a
is_rd = True
break
w = w/wn
#w = w/dot(w.T, w)
l = dot(G, w)
k = dot(G.T, l)
k = k/msqrt(dot(k.T, k))
#k = k/dot(k.T, k)
w = alpha*k + (1-alpha)*w
w = w/msqrt(dot(w.T, w))
t = dot(E, w)
c = dot(F.T, t)
c = c/msqrt(dot(c.T, c))
u = dot(F, c)
diff = dot((u-u1).T, (u-u1))
if verbose:
print "Converged after %s iterations" %niter
print "Error: %.2E" %diff
if is_rd:
print "Hei og haa ... rank deficient, this should really not happen"
break
tt = dot(t.T, t)
p = dot(X.T, t)/tt
q = dot(Y.T, t)/tt
l = dot(Z, w)
#k = dot(Z.T, l)/dot(l.T, l)
U[:,a] = u.ravel()
W[:,a] = w.ravel()
P[:,a] = p.ravel()
T[:,a] = t.ravel()
Q[:,a] = q.ravel()
L[:,a] = l.ravel()
K[:,a] = k.ravel()
E = E - dot(t, p.T)
F = F - dot(t, q.T)
G = (G.T - dot(k, l.T)).T
var_x[a] = pow(E, 2).sum()
var_y[a] = pow(F, 2).sum()
var_z[a] = pow(G, 2).sum()
B[a] = dot(dot(W[:,:a+1], inv(dot(P[:,:a+1].T, W[:,:a+1]))), Q[:,:a+1].T)
#b0[a] = mnY - dot(mnX, B[a])
# variance explained
evx = 100.*(1 - var_x/varX)
evy = 100.*(1 - var_y/varY)
evz = 100.*(1 - var_z/varZ)
if scale=='loads':
tnorm = apply_along_axis(vnorm, 0, T)
T = T/tnorm
W = W*tnorm
Q = Q*tnorm
knorm = apply_along_axis(vnorm, 0, K)
L = L*knorm
K = K/knorm
return {'T':T, 'W':W, 'P':P, 'Q':Q, 'U':U, 'L':L, 'K':K, 'B':B, 'E': E, 'F': F, 'G': G, 'evx':evx, 'evy':evy, 'evz':evz,'mnx': mnX, 'mny': mnY, 'mnz': mnZ}
def vnorm(a):
"""Returns the norm of a vector.
*Parameters*:
a : {array}
Input data, 1-dim, or column vector (m, 1)
*Returns*:
a_norm : {array}
Norm of input vector
"""
return msqrt(dot(a.T,a))
def center(a, axis):
""" Matrix centering.
*Parameters*:
a : {array}
Input data
axis : {integer}
Which centering to perform.
0 = col center, 1 = row center, 2 = double center
-1 = nothing
*Returns*:
a_centered : {array}
Centered data matrix
mn : {array}
Location vector/matrix
"""
# check if we have a vector
is_vec = len(a.shape)==1
if not is_vec:
is_vec = a.shape[0]==1 or a.shape[1]==1
if is_vec:
if axis==2:
warnings.warn("Double centering of vecor ignored, using ordinary centering")
if axis==-1:
mn = 0
else:
mn = a.mean()
return a - mn, mn
# !!!fixme: use broadcasting
if axis==-1:
mn = zeros((1,a.shape[1],))
#mn = tile(mn, (a.shape[0], 1))
elif axis==0:
mn = a.mean(0)[newaxis]
#mn = tile(mn, (a.shape[0], 1))
elif axis==1:
mn = a.mean(1)[:,newaxis]
#mn = tile(mn, (1, a.shape[1]))
elif axis==2:
mn = a.mean(0)[newaxis] + a.mean(1)[:,newaxis] - a.mean()
return a - mn , a.mean(0)[newaxis]
else:
raise IOError("input error: axis must be in [-1,0,1,2]")
return a - mn, mn
def _scale(a, axis):
""" Matrix scaling to unit variance.
*Parameters*:
a : {array}
Input data
axis : {integer}
Which scaling to perform.
0 = column, 1 = row, -1 = nothing
*Returns*:
a_scaled : {array}
Scaled data matrix
mn : {array}
Scaling vector/matrix
"""
if axis==-1:
sc = zeros((a.shape[1],))
elif axis==0:
sc = a.std(0)
elif axis==1:
sc = a.std(1)[:,newaxis]
else:
raise IOError("input error: axis must be in [-1,0,1]")
return a - sc, sc