def _triplet_loss()

in aiops/ContraLSP/attribution/gatemasknn.py [0:0]

• 54 lines of code
• 11 McCabe index (conditional complexity)

    def _triplet_loss(self, condition):
        _, ts_dim, num_dim = condition.shape
        points = condition.reshape(-1, ts_dim * num_dim).detach().numpy()
        num_cluster = 2
        kmeans = KMeans(n_clusters=num_cluster)
        kmeans.fit(points)
        cluster_label = kmeans.labels_
        num_cluster_set = Counter(cluster_label)

        # loss of each cluster
        loss_cluster = condition.abs().mean()   # placeholder
        for i in range(num_cluster):
            if num_cluster_set[i] < 2:
                continue
            cluster_i = points[np.where(cluster_label == i)]
            distance_i = kmeans.transform(cluster_i)[:, i]
            dist_positive = th.DoubleTensor([1])
            dist_negative = th.DoubleTensor([1])
            if num_cluster_set[i] >= 250:
                num_positive = 50
            else:
                num_positive = int(num_cluster_set[i] / 5 + 1)

            # select anchor and positive
            anchor_positive = np.argpartition(distance_i, num_positive)[:(num_positive + 1)]
            # torch anchor
            representation_anc = th.from_numpy(points[anchor_positive[0]])
            # transfer 1D to 3D
            representation_anc = th.reshape(representation_anc, (1, 1, np.shape(points)[1]))

            # positive part
            for l in range(1, num_positive + 1):
                # torch positive
                representation_pos = th.from_numpy(points[anchor_positive[l]])
                # transfer 1D to 3D
                representation_pos = th.reshape(representation_pos, (1, 1, np.shape(points)[1]))
                anchor_minus_positive = representation_anc - representation_pos
                dist_positive += th.norm(anchor_minus_positive, p=1)/np.shape(points)[1]
            dist_positive = dist_positive / num_positive

            # negative part
            for k in range(num_cluster):
                dist_cluster_k_negative = th.DoubleTensor([1])
                if k == i:
                    continue
                else:
                    # select negative
                    if num_cluster_set[k] >= 250:
                        num_negative_cluster_k = 50
                    else:
                        num_negative_cluster_k = int(num_cluster_set[k] / 5 + 1)

                    negative_cluster_k = random.sample(range(points[kmeans.labels_ == k][:, 0].size),
                                                       num_negative_cluster_k)
                    for j in range(num_negative_cluster_k):
                        # torch negative
                        representation_neg = th.from_numpy(points[kmeans.labels_ == k][negative_cluster_k[j]])
                        # transfer 1D to 3D
                        representation_neg = th.reshape(representation_neg, (1, 1, np.shape(points)[1]))
                        anchor_minus_negative = representation_anc - representation_neg
                        dist_cluster_k_negative += th.norm(anchor_minus_negative, p=1)/np.shape(points)[1]

                dist_cluster_k_negative = dist_cluster_k_negative / num_negative_cluster_k
                dist_negative += dist_cluster_k_negative
            dist_negative = dist_negative / (num_cluster - 1)
            #  loss =  -(margin + positives - negatives)
            if self.preservation_mode:
                loss_values = th.max((-dist_positive + dist_negative - 1)[0], th.tensor(0.)) / num_cluster
            else:
                loss_values = th.max(-(-dist_positive + dist_negative - 1)[0], th.tensor(0.)) / num_cluster
            loss_cluster += loss_values

        return loss_cluster