multilabel-learn/mlearn/models/probabilistic_classifier_chains.py at master · yangarbiter/multilabel-learn · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
"""
Probabilistic Classifier Chain
Contributed by Kuan-Hao Huang
"""
import copy

import numpy as np

from ..utils import seed_random_state
from .model_wrapper import ModelWrapper
from mlearn.criteria import pairwise_rank_loss

class ProbabilisticClassifierChains():
    """
    References
    ----------
    .. [1] Cheng, Weiwei, Eyke Hullermeier, and Krzysztof J. Dembczynski. "Bayes
           optimal multilabel classification via probabilistic classifier
           chains." Proceedings of the 27th international conference on machine
           learning (ICML-10). 2010.
    """
    def __init__(self, base_model, cost, n_samples=100, random_state=None):
        self.base_model = base_model
        self.model = PCCModel(base_model=self.base_model, cost=cost,
            n_samples=n_samples, random_state=random_state)

    def train(self, X, Y):
        self.model.train(X, Y)

    def predict(self, X):
        return self.model.predict(X)

class PCCModel():
    def __init__(self, base_model, cost, n_samples, random_state=None):
        if cost not in ['f1', 'hamming', 'rankloss', 'acc']:
            raise NotImplementedError('cost {} is not implemented'.format(cost))
        self.base_model = base_model
        self.cost = cost
        self.n_samples = n_samples
        self.random_state_ = seed_random_state(random_state)

        self.clfs = None
        self.K = None

    def init(self, data_y):
        self.K = data_y.shape[1]

    def new_clfs(self, K):
        return [ModelWrapper(copy.deepcopy(self.base_model)) for i in range(K)]

    def predict_prob(self, data_x):
        pred = np.zeros((data_x.shape[0], len(self.clfs)))
        for i in range(len(self.clfs)):
            pred[:, i] = 1.0 - self.clfs[i].predict_proba(np.concatenate((data_x, (pred[:, :i]>0.5).astype(int)), axis=1))[:, 0]
        return pred

    def predict_one(self, x, pb):
        prob = np.repeat(pb, self.n_samples).reshape((pb.shape[0], self.n_samples)).T
        y_sample = (self.random_state_.rand(self.n_samples, self.K)<prob).astype(int)
        if self.cost == "rankloss":
            thr = 0.0
            pred = (pb>thr).astype(int)
            p_sample = np.repeat(pred, self.n_samples).reshape((pred.shape[0], self.n_samples)).T
            score = pairwise_rank_loss(y_sample, p_sample).mean()
            for p in pb:
                pred = (pb>p).astype(int)
                p_sample = np.repeat(pred, self.n_samples).reshape((pred.shape[0], self.n_samples)).T
                score_t = pairwise_rank_loss(y_sample, p_sample).mean()
                if score_t < score:
                    score = score_t
                    thr = p
            return (pb>thr).astype(int)

        elif self.cost == "hamming":
            return (pb>0.5).astype(int)

        elif self.cost == "f1" or self.cost == "acc":
            s_idxs = y_sample.sum(axis=1)
            P = np.zeros((self.K, self.K))
            for i in range(self.K):
                P[i, :] = y_sample[s_idxs==(i+1), :].sum(axis=0)*1.0/self.n_samples

            W = 1.0 / (np.cumsum(np.ones((self.K, self.K)), axis=1) + np.cumsum(np.ones((self.K, self.K)), axis=0))
            F = P*W
            idxs = (-F).argsort(axis=1)
            H = np.zeros((self.K, self.K), dtype=int)
            for i in range(self.K):
                H[i, idxs[i, :i+1]] = 1
            scores = (F*H).sum(axis=1)
            pred = H[scores.argmax(), :]
            # if (s_idxs==0).mean() > 2*scores.max():
            #   pred = np.zeros((self.K, ), dtype=int)
            return pred

    def predict(self, data_x):
        prob = self.predict_prob(data_x)
        pred = np.zeros((data_x.shape[0], self.K), dtype=int)
        for i in range(data_x.shape[0]):
            pred[i, :] = self.predict_one(data_x[i, :], prob[i, :])
        return pred

    def train(self, X, Y):
        self.init(Y)

        self.clfs = self.new_clfs(self.K)
        for i in range(self.K):
            self.clfs[i].fit(np.concatenate((X, Y[:, :i]), axis=1), Y[:, i])
            try:
                self.clfs[i].set_params(n_jobs=1)
            except:
                pass