Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
mu = (self.mu0 - self.mu1) * (-1)**E
var = self.var0 + self.var1 - 2*self.cov
pdf = dist().pdf(-mu/var**(1/2))
cdf = dist().cdf(-mu/var**(1/2))
D[:,0] = -pdf / cdf / var**(1/2) * (-1)**E
D[:,1] = -pdf / cdf / var**(1/2) * (-1)**(1-E)
D[:,2] = pdf / cdf * mu / (2*var**(3/2)) * (2*self.var0 - 2*self.cov)
D[:,3] = pdf / cdf * mu / (2*var**(3/2)) * (2*self.var1 - 2*self.cov)
D[:,4] = -2*pdf / cdf * mu / (2*var**(3/2)) * self.scale0 * self.scale1 * (1./(1. + self.s**2))
return D
class BivariateLogNormal(RegressionDistn):
n_params = 5
censored_scores = [BivariateLogNormalCensoredScore]
# Event = 1 means event
# Event = 0 means censored
# mu1 = Event
# mu0 = Censored
def __init__(self, params):
self._params = params
self.n_params, self.N = params.shape
self.mu0, self.mu1 = params[0], params[1]
self.scale0, self.scale1 = np.exp(params[2]), np.exp(params[3])
self.var0, self.var1 = self.scale0**2 + eps, self.scale1**2 + eps
self.s = params[4]
self.cor = np.tanh(self.s)
D[:, 1] = self.score(Y) + (Y - self.loc) * D[:, 0]
return D
def metric(self):
I = np.c_[
2 * np.ones_like(self.var),
np.zeros_like(self.var),
np.zeros_like(self.var),
self.var,
]
I = I.reshape((self.var.shape[0], 2, 2))
I = 1 / (2 * np.sqrt(np.pi)) * I
return I
class Normal(RegressionDistn):
"""
Implements the normal distribution for NGBoost.
The normal distribution has two parameters, loc and scale, which are the mean and standard deviation, respectively.
This distribution has both LogScore and CRPScore implemented for it.
"""
n_params = 2
scores = [NormalLogScore, NormalCRPScore]
def __init__(self, params):
super().__init__(params)
self.loc = params[0]
self.scale = np.exp(params[1])
self.var = self.scale ** 2
self.dist = dist(loc=self.loc, scale=self.scale)
self,
Dist=Normal,
Score=LogScore,
Base=default_tree_learner,
natural_gradient=True,
n_estimators=500,
learning_rate=0.01,
minibatch_frac=1.0,
col_sample=1.0,
verbose=True,
verbose_eval=100,
tol=1e-4,
random_state=None,
):
assert issubclass(
Dist, RegressionDistn
), f"{Dist.__name__} is not useable for regression."
if not hasattr(
Dist, "scores"
): # user is trying to use a dist that only has censored scores implemented
Dist = Dist.uncensor(Score)
super().__init__(
Dist,
Score,
Base,
natural_gradient,
n_estimators,
learning_rate,
minibatch_frac,
col_sample,
return score
def d_score(self, Y):
E, T = Y["Event"], Y["Time"]
deriv = 2 * np.exp(-T / self.scale) * (self.scale + T) - 1.5 * self.scale
deriv[E == 1] -= np.exp(-2 * T[E == 1] / self.scale[E == 1]) * (
0.5 * self.scale[E == 1] - T[E == 1]
)
return deriv.reshape((-1, 1))
def metric(self):
M = 0.5 * self.scale[:, np.newaxis, np.newaxis]
return M
class Exponential(RegressionDistn):
"""
Implements the exponential distribution for NGBoost.
The exponential distribution has one parameters, scale. See scipy.stats.expon for details.
This distribution has both LogScore and CRPScore implemented for it and both work with right-censored data
"""
n_params = 1
censored_scores = [ExponentialLogScore, ExponentialCRPScore]
def __init__(self, params):
self._params = params
self.scale = np.exp(params[0])
self.dist = dist(scale=self.scale)
def __getattr__(self, name):
+ 2 * sp.stats.norm.pdf(Z) ** 2
- 2 * sp.stats.norm.cdf(Z) * sp.stats.norm.pdf(Z) ** 2
- np.sqrt(2 / np.pi) * sp.stats.norm.pdf(np.sqrt(2) * Z)
)
D[:, 1] = self.score(Y) + (lT - self.loc) * D[:, 0]
return D
def metric(self):
I = np.zeros((self.loc.shape[0], 2, 2))
I[:, 0, 0] = 2
I[:, 1, 1] = self.scale ** 2
I /= 2 * np.sqrt(np.pi)
return I
class LogNormal(RegressionDistn):
"""
Implements the log-normal distribution for NGBoost.
The normal distribution has two parameters, s and scale (see scipy.stats.lognorm)
This distribution has both LogScore and CRPScore implemented for it and both work for right-censored data.
"""
n_params = 2
censored_scores = [LogNormalLogScoreCensored, LogNormalCRPScoreCensored]
def __init__(self, params):
self._params = params
self.loc = params[0]
self.scale = np.exp(params[1])
self.dist = dist(s=self.scale, scale=np.exp(self.loc))
mu = (self.mu0 - self.mu1) * (-1)**E
var = self.var0 + self.var1 - 2*self.cov
pdf = dist().pdf(-mu/var**(1/2))
cdf = dist().cdf(-mu/var**(1/2))
D[:,0] = -pdf / cdf / var**(1/2) * (-1)**E
D[:,1] = -pdf / cdf / var**(1/2) * (-1)**(1-E)
D[:,2] = pdf / cdf * mu / (2*var**(3/2)) * (2*self.var0 - 2*self.cov)
D[:,3] = pdf / cdf * mu / (2*var**(3/2)) * (2*self.var1 - 2*self.cov)
D[:,4] = -2*pdf / cdf * mu / (2*var**(3/2)) * self.scale0 * self.scale1 * (1./(1. + self.s**2))
return D
class BivariateNormal(RegressionDistn):
n_params = 5
censored_scores = [BivariateNormalCensoredScore]
# Event = 1 means event
# Event = 0 means censored
# mu1 = Event
# mu0 = Censored
def __init__(self, params):
self._params = params
self.n_params, self.N = params.shape
self.mu0, self.mu1 = params[0], params[1]
self.scale0, self.scale1 = np.exp(params[2]), np.exp(params[3])
self.var0, self.var1 = self.scale0**2, self.scale1**2
self.s = params[4]
self.cor = np.tanh(self.s)