Module fri.model.ordinal_regression
View Source
import cvxpy as cvx
import numpy as np
from sklearn.metrics import make_scorer
from sklearn.utils import check_X_y
from .base_cvxproblem import Relevance_CVXProblem
from .base_initmodel import InitModel
from .base_type import ProblemType
class OrdinalRegression(ProblemType):
@classmethod
def parameters(cls):
return ["C"]
@property
def get_initmodel_template(cls):
return OrdinalRegression_SVM
@property
def get_cvxproblem_template(cls):
return OrdinalRegression_Relevance_Bound
def relax_factors(cls):
return ["loss_slack", "w_l1_slack"]
def preprocessing(self, data, **kwargs):
X, y = data
# Check that X and y have correct shape
X, y = check_X_y(X, y)
if np.min(y) > 0:
print("First ordinal class has index > 0. Shifting index...")
y = y - np.min(y)
return X, y
class OrdinalRegression_SVM(InitModel):
HYPERPARAMETER = ["C"]
def __init__(self, C=1):
super().__init__()
self.C = C
def fit(self, X, y, **kwargs):
(n, d) = X.shape
C = self.get_params()["C"]
self.classes_ = np.unique(y)
original_bins = sorted(self.classes_)
n_bins = len(original_bins)
bins = np.arange(n_bins)
get_old_bin = dict(zip(bins, original_bins))
w = cvx.Variable(shape=(d), name="w")
# For ordinal regression we use two slack variables, we observe the slack in both directions
slack_left = cvx.Variable(shape=(n), name="slack_left")
slack_right = cvx.Variable(shape=(n), name="slack_right")
# We have an offset for every bin boundary
b_s = cvx.Variable(shape=(n_bins - 1), name="bias")
objective = cvx.Minimize(cvx.norm(w, 1) + C * cvx.sum(slack_left + slack_right))
constraints = [slack_left >= 0, slack_right >= 0]
# Add constraints for slack into left neighboring bins
for i in range(n_bins - 1):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w - slack_left[indices] <= b_s[i] - 1)
# Add constraints for slack into right neighboring bins
for i in range(1, n_bins):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w + slack_right[indices] >= b_s[i - 1] + 1)
# Add explicit constraint, that all bins are ascending
for i in range(n_bins - 2):
constraints.append(b_s[i] <= b_s[i + 1])
# Solve problem.
problem = cvx.Problem(objective, constraints)
problem.solve(**self.SOLVER_PARAMS)
w = w.value
b_s = b_s.value
slack_left = np.asarray(slack_left.value).flatten()
slack_right = np.asarray(slack_right.value).flatten()
self.model_state = {"w": w, "b_s": b_s, "slack": (slack_left, slack_right)}
loss = np.sum(slack_left + slack_right)
w_l1 = np.linalg.norm(w, ord=1)
self.constraints = {"loss": loss, "w_l1": w_l1}
return self
def predict(self, X):
w = self.model_state["w"]
b_s = self.model_state["b_s"]
scores = np.dot(X, w.T)[np.newaxis]
bin_thresholds = np.append(b_s, np.inf)
# If thresholds are smaller than score the value belongs to the bigger bin
# after subtracting we check for positive elements
indices = np.sum(scores.T - bin_thresholds >= 0, -1)
return self.classes_[indices]
def score(self, X, y, error_type="mmae", return_error=False, **kwargs):
X, y = check_X_y(X, y)
prediction = self.predict(X)
score = ordinal_scores(y, prediction, error_type, return_error=return_error)
return score
def make_scorer(self):
# Use multiple scores for ordinal regression
mze = make_scorer(ordinal_scores, error_type="mze")
mae = make_scorer(ordinal_scores, error_type="mae")
mmae = make_scorer(ordinal_scores, error_type="mmae")
scorer = {"mze": mze, "mae": mae, "mmae": mmae}
return scorer, "mmae"
def ordinal_scores(y, prediction, error_type, return_error=False):
"""Score function for ordinal problems.
Parameters
----------
y : target class vector
Truth vector
prediction : prediction class vector
Predicted classes
error_type : str
Error type "mze","mae","mmae"
return_error : bool, optional
Return error (lower is better) or score (inverted, higher is better)
Returns
-------
float
Error or score depending on 'return_error'
Raises
------
ValueError
When using wrong error_type
"""
n = len(y)
classes = np.unique(y)
n_bins = len(classes)
max_dist = n_bins - 1
# If only one class available, we dont need to average
if max_dist == 0:
error_type = "mze"
def mze(prediction, y):
return np.sum(prediction != y)
def mae(prediction, y):
return np.sum(np.abs(prediction - y))
# Score based on mean zero-one error
if error_type == "mze":
error = mze(prediction, y) / n
score = 1 - error
# Score based on mean absolute error
elif error_type == "mae":
error = mae(prediction, y) / n
score = (max_dist - error) / max_dist
# Score based on macro-averaged mean absolute error
elif error_type == "mmae":
sum = 0
for i in range(n_bins):
samples = y == i
n_samples = np.sum(samples)
if n_samples > 0:
bin_error = mae(prediction[samples], y[samples]) / n_samples
sum += bin_error
error = sum / n_bins
score = (max_dist - error) / max_dist
else:
raise ValueError("error_type {} not available!'".format(error_type))
if return_error:
return error
else:
return score
class OrdinalRegression_Relevance_Bound(Relevance_CVXProblem):
def init_objective_UB(self, sign=None, **kwargs):
self.add_constraint(
self.feature_relevance <= sign * self.w[self.current_feature]
)
self._objective = cvx.Maximize(self.feature_relevance)
def init_objective_LB(self, **kwargs):
self.add_constraint(
cvx.abs(self.w[self.current_feature]) <= self.feature_relevance
)
self._objective = cvx.Minimize(self.feature_relevance)
def _init_constraints(self, parameters, init_model_constraints):
n_bins = len(np.unique(self.y))
# Upper constraints from initial model
l1_w = init_model_constraints["w_l1"]
init_loss = init_model_constraints["loss"]
C = parameters["C"]
# New Variables
self.w = cvx.Variable(shape=(self.d), name="w")
# For ordinal regression we use two slack variables, we observe the slack in both directions
self.slack_left = cvx.Variable(shape=(self.n), name="slack_left", nonneg=True)
self.slack_right = cvx.Variable(shape=(self.n), name="slack_right", nonneg=True)
# We have an offset for every bin boundary
self.b_s = cvx.Variable(shape=(n_bins - 1), name="bias")
# New Constraints
self.loss = cvx.sum(self.slack_left + self.slack_right)
self.weight_norm = cvx.norm(self.w, 1)
for i in range(n_bins - 1):
indices = np.where(self.y == i)
self.add_constraint(
self.X[indices] @ self.w - self.slack_left[indices] <= self.b_s[i] - 1
)
for i in range(1, n_bins):
indices = np.where(self.y == i)
self.add_constraint(
self.X[indices] @ self.w + self.slack_right[indices]
>= self.b_s[i - 1] + 1
)
for i in range(n_bins - 2):
self.add_constraint(self.b_s[i] <= self.b_s[i + 1])
self.add_constraint(self.weight_norm <= l1_w)
self.add_constraint(C * self.loss <= C * init_loss)
self.feature_relevance = cvx.Variable(nonneg=True, name="Feature Relevance")
Functions
ordinal_scores
def ordinal_scores(
y,
prediction,
error_type,
return_error=False
)
Score function for ordinal problems.
Parameters
y : target class vector Truth vector prediction : prediction class vector Predicted classes error_type : str Error type "mze","mae","mmae" return_error : bool, optional Return error (lower is better) or score (inverted, higher is better)
Returns
float Error or score depending on 'return_error'
Raises
ValueError When using wrong error_type
View Source
def ordinal_scores(y, prediction, error_type, return_error=False):
"""Score function for ordinal problems.
Parameters
----------
y : target class vector
Truth vector
prediction : prediction class vector
Predicted classes
error_type : str
Error type "mze","mae","mmae"
return_error : bool, optional
Return error (lower is better) or score (inverted, higher is better)
Returns
-------
float
Error or score depending on 'return_error'
Raises
------
ValueError
When using wrong error_type
"""
n = len(y)
classes = np.unique(y)
n_bins = len(classes)
max_dist = n_bins - 1
# If only one class available, we dont need to average
if max_dist == 0:
error_type = "mze"
def mze(prediction, y):
return np.sum(prediction != y)
def mae(prediction, y):
return np.sum(np.abs(prediction - y))
# Score based on mean zero-one error
if error_type == "mze":
error = mze(prediction, y) / n
score = 1 - error
# Score based on mean absolute error
elif error_type == "mae":
error = mae(prediction, y) / n
score = (max_dist - error) / max_dist
# Score based on macro-averaged mean absolute error
elif error_type == "mmae":
sum = 0
for i in range(n_bins):
samples = y == i
n_samples = np.sum(samples)
if n_samples > 0:
bin_error = mae(prediction[samples], y[samples]) / n_samples
sum += bin_error
error = sum / n_bins
score = (max_dist - error) / max_dist
else:
raise ValueError("error_type {} not available!'".format(error_type))
if return_error:
return error
else:
return score
Classes
OrdinalRegression
class OrdinalRegression(
**kwargs
)
Helper class that provides a standard way to create an ABC using inheritance.
View Source
class OrdinalRegression(ProblemType):
@classmethod
def parameters(cls):
return ["C"]
@property
def get_initmodel_template(cls):
return OrdinalRegression_SVM
@property
def get_cvxproblem_template(cls):
return OrdinalRegression_Relevance_Bound
def relax_factors(cls):
return ["loss_slack", "w_l1_slack"]
def preprocessing(self, data, **kwargs):
X, y = data
# Check that X and y have correct shape
X, y = check_X_y(X, y)
if np.min(y) > 0:
print("First ordinal class has index > 0. Shifting index...")
y = y - np.min(y)
return X, y
Ancestors (in MRO)
- fri.model.base_type.ProblemType
- abc.ABC
Static methods
parameters
def parameters(
)
View Source
@classmethod
def parameters(cls):
return ["C"]
Instance variables
get_cvxproblem_template
get_initmodel_template
Methods
get_all_parameters
def get_all_parameters(
self
)
View Source
def get_all_parameters(self):
return {p: self.get_chosen_parameter(p) for p in self.parameters()}
get_all_relax_factors
def get_all_relax_factors(
self
)
View Source
def get_all_relax_factors(self):
return {p: self.get_chosen_relax_factors(p) for p in self.relax_factors()}
get_chosen_parameter
def get_chosen_parameter(
self,
p
)
View Source
def get_chosen_parameter(self, p):
try:
return [
self.chosen_parameters_[p]
] # We return list for param search function
except:
# # TODO: rewrite the parameter logic
# # TODO: move this to subclass
if p == "scaling_lupi_w":
# return [0.1, 1, 10, 100, 1000]
return scipy.stats.reciprocal(a=1e-15, b=1e10)
# if p == "scaling_lupi_loss":
# # value 0>p<1 causes standard svm solution
# # p>1 encourages usage of lupi function
# return scipy.stats.reciprocal(a=1e-15, b=1e15)
if p == "C":
return scipy.stats.reciprocal(a=1e-5, b=1e5)
if p == "epsilon":
return [0, 0.001, 0.01, 0.1, 1, 10, 100]
else:
return scipy.stats.reciprocal(a=1e-10, b=1e10)
get_chosen_relax_factors
def get_chosen_relax_factors(
self,
p
)
View Source
def get_chosen_relax_factors(self, p):
try:
factor = self.relax_factors_[p]
except KeyError:
try:
factor = self.relax_factors_[p + "_slack"]
except KeyError:
factor = 0.1
if factor < 0:
raise ValueError("Slack Factor multiplier is positive!")
return factor
get_relaxed_constraints
def get_relaxed_constraints(
self,
constraints
)
View Source
def get_relaxed_constraints(self, constraints):
return {c: self.relax_constraint(c, v) for c, v in constraints.items()}
postprocessing
def postprocessing(
self,
bounds
)
View Source
def postprocessing(self, bounds):
return bounds
preprocessing
def preprocessing(
self,
data,
**kwargs
)
View Source
def preprocessing(self, data, **kwargs):
X, y = data
# Check that X and y have correct shape
X, y = check_X_y(X, y)
if np.min(y) > 0:
print("First ordinal class has index > 0. Shifting index...")
y = y - np.min(y)
return X, y
relax_constraint
def relax_constraint(
self,
key,
value
)
View Source
def relax_constraint(self, key, value):
return value * (1 + self.get_chosen_relax_factors(key))
relax_factors
def relax_factors(
cls
)
View Source
def relax_factors(cls):
return ["loss_slack", "w_l1_slack"]
OrdinalRegression_Relevance_Bound
class OrdinalRegression_Relevance_Bound(
current_feature: int,
data: tuple,
hyperparameters,
best_model_constraints,
preset_model=None,
best_model_state=None,
probeID=-1,
**kwargs
)
Helper class that provides a standard way to create an ABC using inheritance.
View Source
class OrdinalRegression_Relevance_Bound(Relevance_CVXProblem):
def init_objective_UB(self, sign=None, **kwargs):
self.add_constraint(
self.feature_relevance <= sign * self.w[self.current_feature]
)
self._objective = cvx.Maximize(self.feature_relevance)
def init_objective_LB(self, **kwargs):
self.add_constraint(
cvx.abs(self.w[self.current_feature]) <= self.feature_relevance
)
self._objective = cvx.Minimize(self.feature_relevance)
def _init_constraints(self, parameters, init_model_constraints):
n_bins = len(np.unique(self.y))
# Upper constraints from initial model
l1_w = init_model_constraints["w_l1"]
init_loss = init_model_constraints["loss"]
C = parameters["C"]
# New Variables
self.w = cvx.Variable(shape=(self.d), name="w")
# For ordinal regression we use two slack variables, we observe the slack in both directions
self.slack_left = cvx.Variable(shape=(self.n), name="slack_left", nonneg=True)
self.slack_right = cvx.Variable(shape=(self.n), name="slack_right", nonneg=True)
# We have an offset for every bin boundary
self.b_s = cvx.Variable(shape=(n_bins - 1), name="bias")
# New Constraints
self.loss = cvx.sum(self.slack_left + self.slack_right)
self.weight_norm = cvx.norm(self.w, 1)
for i in range(n_bins - 1):
indices = np.where(self.y == i)
self.add_constraint(
self.X[indices] @ self.w - self.slack_left[indices] <= self.b_s[i] - 1
)
for i in range(1, n_bins):
indices = np.where(self.y == i)
self.add_constraint(
self.X[indices] @ self.w + self.slack_right[indices]
>= self.b_s[i - 1] + 1
)
for i in range(n_bins - 2):
self.add_constraint(self.b_s[i] <= self.b_s[i + 1])
self.add_constraint(self.weight_norm <= l1_w)
self.add_constraint(C * self.loss <= C * init_loss)
self.feature_relevance = cvx.Variable(nonneg=True, name="Feature Relevance")
Ancestors (in MRO)
- fri.model.base_cvxproblem.Relevance_CVXProblem
- abc.ABC
Descendants
- fri.model.lupi_ordinal_regression.LUPI_OrdinalRegression_Relevance_Bound
Static methods
aggregate_max_candidates
def aggregate_max_candidates(
max_problems_candidates
)
View Source
@classmethod
def aggregate_max_candidates(cls, max_problems_candidates):
vals = [candidate.solved_relevance for candidate in max_problems_candidates]
max_value = max(vals)
return max_value
aggregate_min_candidates
def aggregate_min_candidates(
min_problems_candidates
)
View Source
@classmethod
def aggregate_min_candidates(cls, min_problems_candidates):
vals = [candidate.solved_relevance for candidate in min_problems_candidates]
min_value = min(vals)
return min_value
generate_lower_bound_problem
def generate_lower_bound_problem(
best_hyperparameters,
init_constraints,
best_model_state,
data,
di,
preset_model,
probeID=-1
)
View Source
@classmethod
def generate_lower_bound_problem(
cls,
best_hyperparameters,
init_constraints,
best_model_state,
data,
di,
preset_model,
probeID=-1,
):
problem = cls(
di,
data,
best_hyperparameters,
init_constraints,
preset_model=preset_model,
best_model_state=best_model_state,
probeID=probeID,
)
problem.init_objective_LB()
problem.isLowerBound = True
yield problem
generate_upper_bound_problem
def generate_upper_bound_problem(
best_hyperparameters,
init_constraints,
best_model_state,
data,
di,
preset_model,
probeID=-1
)
View Source
@classmethod
def generate_upper_bound_problem(
cls,
best_hyperparameters,
init_constraints,
best_model_state,
data,
di,
preset_model,
probeID=-1,
):
for sign in [-1, 1]:
problem = cls(
di,
data,
best_hyperparameters,
init_constraints,
preset_model=preset_model,
best_model_state=best_model_state,
probeID=probeID,
)
problem.init_objective_UB(sign=sign)
problem.isLowerBound = False
yield problem
Instance variables
accepted_status
constraints
cvx_problem
isProbe
is_solved
objective
probeID
solved_relevance
solver_kwargs
Methods
add_constraint
def add_constraint(
self,
new
)
View Source
def add_constraint(self, new):
self._constraints.append(new)
init_objective_LB
def init_objective_LB(
self,
**kwargs
)
View Source
def init_objective_LB(self, **kwargs):
self.add_constraint(
cvx.abs(self.w[self.current_feature]) <= self.feature_relevance
)
self._objective = cvx.Minimize(self.feature_relevance)
init_objective_UB
def init_objective_UB(
self,
sign=None,
**kwargs
)
View Source
def init_objective_UB(self, sign=None, **kwargs):
self.add_constraint(
self.feature_relevance <= sign * self.w[self.current_feature]
)
self._objective = cvx.Maximize(self.feature_relevance)
preprocessing_data
def preprocessing_data(
self,
data,
best_model_state
)
View Source
def preprocessing_data(self, data, best_model_state):
X, y = data
self.n = X.shape[0]
self.d = X.shape[1]
self.X = X
self.y = np.array(y)
solve
def solve(
self
) -> object
View Source
def solve(self) -> object:
# We init cvx problem here because pickling LP solver objects is problematic
# by deferring it to here, worker threads do the problem building themselves and we spare the serialization
self._cvx_problem = cvx.Problem(
objective=self.objective, constraints=self.constraints
)
try:
# print("Solve", self)
self._cvx_problem.solve(**self.solver_kwargs)
except SolverError:
# We ignore Solver Errors, which are common with our framework:
# We solve multiple problems per bound and choose a feasible solution later (see '_create_interval')
pass
self._solver_status = self._cvx_problem.status
# self._cvx_problem = None
return self
OrdinalRegression_SVM
class OrdinalRegression_SVM(
C=1
)
Helper class that provides a standard way to create an ABC using inheritance.
View Source
class OrdinalRegression_SVM(InitModel):
HYPERPARAMETER = ["C"]
def __init__(self, C=1):
super().__init__()
self.C = C
def fit(self, X, y, **kwargs):
(n, d) = X.shape
C = self.get_params()["C"]
self.classes_ = np.unique(y)
original_bins = sorted(self.classes_)
n_bins = len(original_bins)
bins = np.arange(n_bins)
get_old_bin = dict(zip(bins, original_bins))
w = cvx.Variable(shape=(d), name="w")
# For ordinal regression we use two slack variables, we observe the slack in both directions
slack_left = cvx.Variable(shape=(n), name="slack_left")
slack_right = cvx.Variable(shape=(n), name="slack_right")
# We have an offset for every bin boundary
b_s = cvx.Variable(shape=(n_bins - 1), name="bias")
objective = cvx.Minimize(cvx.norm(w, 1) + C * cvx.sum(slack_left + slack_right))
constraints = [slack_left >= 0, slack_right >= 0]
# Add constraints for slack into left neighboring bins
for i in range(n_bins - 1):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w - slack_left[indices] <= b_s[i] - 1)
# Add constraints for slack into right neighboring bins
for i in range(1, n_bins):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w + slack_right[indices] >= b_s[i - 1] + 1)
# Add explicit constraint, that all bins are ascending
for i in range(n_bins - 2):
constraints.append(b_s[i] <= b_s[i + 1])
# Solve problem.
problem = cvx.Problem(objective, constraints)
problem.solve(**self.SOLVER_PARAMS)
w = w.value
b_s = b_s.value
slack_left = np.asarray(slack_left.value).flatten()
slack_right = np.asarray(slack_right.value).flatten()
self.model_state = {"w": w, "b_s": b_s, "slack": (slack_left, slack_right)}
loss = np.sum(slack_left + slack_right)
w_l1 = np.linalg.norm(w, ord=1)
self.constraints = {"loss": loss, "w_l1": w_l1}
return self
def predict(self, X):
w = self.model_state["w"]
b_s = self.model_state["b_s"]
scores = np.dot(X, w.T)[np.newaxis]
bin_thresholds = np.append(b_s, np.inf)
# If thresholds are smaller than score the value belongs to the bigger bin
# after subtracting we check for positive elements
indices = np.sum(scores.T - bin_thresholds >= 0, -1)
return self.classes_[indices]
def score(self, X, y, error_type="mmae", return_error=False, **kwargs):
X, y = check_X_y(X, y)
prediction = self.predict(X)
score = ordinal_scores(y, prediction, error_type, return_error=return_error)
return score
def make_scorer(self):
# Use multiple scores for ordinal regression
mze = make_scorer(ordinal_scores, error_type="mze")
mae = make_scorer(ordinal_scores, error_type="mae")
mmae = make_scorer(ordinal_scores, error_type="mmae")
scorer = {"mze": mze, "mae": mae, "mmae": mmae}
return scorer, "mmae"
Ancestors (in MRO)
- fri.model.base_initmodel.InitModel
- abc.ABC
- sklearn.base.BaseEstimator
Class variables
HYPERPARAMETER
SOLVER_PARAMS
Instance variables
L1_factor
Methods
fit
def fit(
self,
X,
y,
**kwargs
)
View Source
def fit(self, X, y, **kwargs):
(n, d) = X.shape
C = self.get_params()["C"]
self.classes_ = np.unique(y)
original_bins = sorted(self.classes_)
n_bins = len(original_bins)
bins = np.arange(n_bins)
get_old_bin = dict(zip(bins, original_bins))
w = cvx.Variable(shape=(d), name="w")
# For ordinal regression we use two slack variables, we observe the slack in both directions
slack_left = cvx.Variable(shape=(n), name="slack_left")
slack_right = cvx.Variable(shape=(n), name="slack_right")
# We have an offset for every bin boundary
b_s = cvx.Variable(shape=(n_bins - 1), name="bias")
objective = cvx.Minimize(cvx.norm(w, 1) + C * cvx.sum(slack_left + slack_right))
constraints = [slack_left >= 0, slack_right >= 0]
# Add constraints for slack into left neighboring bins
for i in range(n_bins - 1):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w - slack_left[indices] <= b_s[i] - 1)
# Add constraints for slack into right neighboring bins
for i in range(1, n_bins):
indices = np.where(y == get_old_bin[i])
constraints.append(X[indices] @ w + slack_right[indices] >= b_s[i - 1] + 1)
# Add explicit constraint, that all bins are ascending
for i in range(n_bins - 2):
constraints.append(b_s[i] <= b_s[i + 1])
# Solve problem.
problem = cvx.Problem(objective, constraints)
problem.solve(**self.SOLVER_PARAMS)
w = w.value
b_s = b_s.value
slack_left = np.asarray(slack_left.value).flatten()
slack_right = np.asarray(slack_right.value).flatten()
self.model_state = {"w": w, "b_s": b_s, "slack": (slack_left, slack_right)}
loss = np.sum(slack_left + slack_right)
w_l1 = np.linalg.norm(w, ord=1)
self.constraints = {"loss": loss, "w_l1": w_l1}
return self
get_params
def get_params(
self,
deep=True
)
Get parameters for this estimator.
Parameters
deep : bool, default=True If True, will return the parameters for this estimator and contained subobjects that are estimators.
Returns
params : dict Parameter names mapped to their values.
View Source
def get_params(self, deep=True):
"""
Get parameters for this estimator.
Parameters
----------
deep : bool, default=True
If True, will return the parameters for this estimator and
contained subobjects that are estimators.
Returns
-------
params : dict
Parameter names mapped to their values.
"""
out = dict()
for key in self._get_param_names():
value = getattr(self, key)
if deep and hasattr(value, 'get_params'):
deep_items = value.get_params().items()
out.update((key + '__' + k, val) for k, val in deep_items)
out[key] = value
return out
make_scorer
def make_scorer(
self
)
View Source
def make_scorer(self):
# Use multiple scores for ordinal regression
mze = make_scorer(ordinal_scores, error_type="mze")
mae = make_scorer(ordinal_scores, error_type="mae")
mmae = make_scorer(ordinal_scores, error_type="mmae")
scorer = {"mze": mze, "mae": mae, "mmae": mmae}
return scorer, "mmae"
predict
def predict(
self,
X
)
View Source
def predict(self, X):
w = self.model_state["w"]
b_s = self.model_state["b_s"]
scores = np.dot(X, w.T)[np.newaxis]
bin_thresholds = np.append(b_s, np.inf)
# If thresholds are smaller than score the value belongs to the bigger bin
# after subtracting we check for positive elements
indices = np.sum(scores.T - bin_thresholds >= 0, -1)
return self.classes_[indices]
score
def score(
self,
X,
y,
error_type='mmae',
return_error=False,
**kwargs
)
View Source
def score(self, X, y, error_type="mmae", return_error=False, **kwargs):
X, y = check_X_y(X, y)
prediction = self.predict(X)
score = ordinal_scores(y, prediction, error_type, return_error=return_error)
return score
set_params
def set_params(
self,
**params
)
Set the parameters of this estimator.
The method works on simple estimators as well as on nested objects
(such as :class:~sklearn.pipeline.Pipeline). The latter have
parameters of the form <component>__<parameter> so that it's
possible to update each component of a nested object.
Parameters
**params : dict Estimator parameters.
Returns
self : estimator instance Estimator instance.
View Source
def set_params(self, **params):
"""
Set the parameters of this estimator.
The method works on simple estimators as well as on nested objects
(such as :class:`~sklearn.pipeline.Pipeline`). The latter have
parameters of the form ``<component>__<parameter>`` so that it's
possible to update each component of a nested object.
Parameters
----------
**params : dict
Estimator parameters.
Returns
-------
self : estimator instance
Estimator instance.
"""
if not params:
# Simple optimization to gain speed (inspect is slow)
return self
valid_params = self.get_params(deep=True)
nested_params = defaultdict(dict) # grouped by prefix
for key, value in params.items():
key, delim, sub_key = key.partition('__')
if key not in valid_params:
raise ValueError('Invalid parameter %s for estimator %s. '
'Check the list of available parameters '
'with `estimator.get_params().keys()`.' %
(key, self))
if delim:
nested_params[key][sub_key] = value
else:
setattr(self, key, value)
valid_params[key] = value
for key, sub_params in nested_params.items():
valid_params[key].set_params(**sub_params)
return self