Repository: dpuenteramirez/Semisupervised-learning-and-instance-selection-methods (language: python)

Issue:
Tri-Training to class

Tri-Training change to use the same classificator

Tri-Training change to use the same classificator



Discussion / hints:
/cib
Branch [issue-61-Tri-Training_to_class](https://github.com/dpr1005/Semisupervised-learning-and-instance-selection-methods/tree/issue-61-Tri-Training_to_class) created!



Fix patch:
diff --git a/SemiSupervised Learning Algorithms/Tri-Training.py b/SemiSupervised Learning Algorithms/Tri-Training.py
deleted file mode 100644
index 8de4722..0000000
--- a/SemiSupervised Learning Algorithms/Tri-Training.py
+++ /dev/null
@@ -1,206 +0,0 @@
-#!/usr/bin/env python
-# -*- coding:utf-8 -*-
-# @Filename:    Tri-Training.py
-# @Author:      Daniel Puente Ramírez
-# @Time:        27/12/21 10:25
-import time
-from math import floor, ceil
-
-import numpy as np
-from sklearn.datasets import load_iris
-from sklearn.ensemble import RandomForestClassifier
-from sklearn.metrics import classification_report
-from sklearn.model_selection import train_test_split
-from sklearn.neighbors import KNeighborsClassifier
-from sklearn.tree import DecisionTreeClassifier
-from sklearn.utils import Bunch
-
-
-def measure_error(classifier_j, classifier_k, labeled_data):
-    pred_j = classifier_j.predict(labeled_data)
-    pred_k = classifier_k.predict(labeled_data)
-    same = len([0 for x, y in zip(pred_j, pred_k) if x == y])
-    return (len(pred_j) - same) / same
-
-
-def subsample(l_t, s):
-    np.random.seed(42)
-    rng = np.random.default_rng()
-    data = np.array(l_t['data'])
-    target = np.array(l_t['target'])
-    samples_index = rng.choice(len(data), size=s, replace=False)
-    samples = data[samples_index]
-    targets = target[samples_index]
-    return Bunch(data=samples, target=targets)
-
-
-def tri_training(L, U):
-    start = time.time()
-    # Initialize values
-    # BootStrap
-    # Classifier
-    # e' and l
-    labeled = L['data']
-    target = L['target']
-    unlabeled = U['data']
-
-    train, _, test, _ = train_test_split(labeled, target, train_size=floor(
-        len(labeled) / 3), stratify=target, random_state=42)
-    h_j = KNeighborsClassifier(n_neighbors=3, n_jobs=-1).fit(train, test)
-    ep_j = 0.5
-    lp_j = 0
-    train, _, test, _ = train_test_split(labeled, target, train_size=floor(
-        len(labeled) / 3), stratify=target, random_state=420)
-    h_k = DecisionTreeClassifier(random_state=420).fit(train, test)
-    ep_k = 0.5
-    lp_k = 0
-    train, _, test, _ = train_test_split(labeled, target, train_size=floor(
-        len(labeled) / 3), stratify=target, random_state=4200)
-    h_i = RandomForestClassifier(random_state=4200).fit(train, test)
-    ep_i = 0.5
-    lp_i = 0
-
-    while True:
-        print(1)
-        hash_i = h_i.__hash__()
-        hash_j = h_j.__hash__()
-        hash_k = h_k.__hash__()
-
-        # Classifier h_j
-        update_j = False
-        L_j = Bunch(data=[], target=[])
-        e_j = measure_error(h_j, h_k, L['data'])
-
-        if e_j < ep_j:
-            for sample in unlabeled:
-                sample_s = sample.reshape(1, -1)
-                if h_j.predict(sample_s) == h_k.predict(sample_s):
-                    pred = h_i.predict(sample_s)
-                    prev_dat = list(L_j['data'])
-                    prev_tar = list(L_j['target'])
-                    prev_dat.append(sample)
-                    L_j['data'] = np.array(prev_dat)
-                    prev_tar.append(pred)
-                    L_j['target'] = np.array(prev_tar)
-
-            if lp_j == 0:
-                lp_j = floor(e_j / (ep_j - e_j) + 1)
-
-            if lp_j < len(L_j['data']):
-                if e_j * len(L_j['data']) < ep_j * lp_j:
-                    update_j = True
-                elif lp_j > e_j / (ep_j - e_j):
-                    L_j = subsample(L_j, ceil(((ep_j * lp_j) / e_j) - 1))
-                    update_j = True
-
-        # Classifier h_k
-        update_k = False
-        L_k = Bunch(data=np.array([]), target=np.array([]))
-        e_k = measure_error(h_j, h_k, L['data'])
-
-        if e_k < ep_k:
-            for sample in unlabeled:
-                sample_s = sample.reshape(1, -1)
-                if h_j.predict(sample_s) == h_k.predict(sample_s):
-                    pred = h_i.predict(sample_s)
-                    prev_dat = list(L_k['data'])
-                    prev_tar = list(L_k['target'])
-                    prev_dat.append(sample)
-                    L_k['data'] = np.array(prev_dat)
-                    prev_tar.append(pred)
-                    L_k['target'] = np.array(prev_tar)
-
-            if lp_k == 0:
-                lp_k = floor(e_k / (ep_k - e_k) + 1)
-
-            if lp_k < len(L_k['data']):
-                if e_k * len(L_k['data']) < ep_k * lp_k:
-                    update_k = True
-                elif lp_k > e_k / (ep_k - e_k):
-                    L_k = subsample(L_k, ceil(((ep_k * lp_k) / e_k) - 1))
-                    update_k = True
-
-        # Classifier h_i
-        update_i = False
-        L_i = Bunch(data=np.array([]), target=np.array([]))
-        e_i = measure_error(h_j, h_k, L['data'])
-
-        if e_i < ep_i:
-            for sample in unlabeled:
-                sample_s = sample.reshape(1, -1)
-                if h_j.predict(sample_s) == h_k.predict(sample_s):
-                    pred = h_i.predict(sample_s)
-                    prev_dat = list(L_i['data'])
-                    prev_tar = list(L_i['target'])
-                    prev_dat.append(sample)
-                    L_i['data'] = np.array(prev_dat)
-                    prev_tar.append(pred)
-                    L_i['target'] = np.array(prev_tar)
-
-            if lp_i == 0:
-                lp_i = floor(e_i / (ep_i - e_i) + 1)
-
-            if lp_i < len(L_i['data']):
-                if e_i * len(L_i['data']) < ep_i * lp_i:
-                    update_i = True
-                elif lp_i > e_i / (ep_i - e_i):
-                    L_i = subsample(L_i, ceil(((ep_i * lp_i) / e_i) - 1))
-                    update_i = True
-
-        # for i \in {1..3} do
-        if update_j:
-            train = np.concatenate((L['data'], L_j['data']), axis=0)
-            test = np.concatenate((L['target'], np.ravel(L_j['target'])),
-                                  axis=0)
-            h_j = KNeighborsClassifier(n_neighbors=3, n_jobs=-1).fit(train,
-                                                                     test)
-            ep_j = e_j
-            lp_j = len(L_j)
-        if update_k:
-            train = np.concatenate((L['data'], L_k['data']), axis=0)
-            test = np.concatenate((L['target'], np.ravel(L_k['target'])),
-                                  axis=0)
-            h_k = DecisionTreeClassifier(random_state=420).fit(train, test)
-            ep_k = e_k
-            lp_k = len(L_k)
-        if update_i:
-            train = np.concatenate((L['data'], L_i['data']), axis=0)
-            test = np.concatenate((L['target'], np.ravel(L_i['target'])),
-                                  axis=0)
-            h_i = RandomForestClassifier(random_state=4200).fit(train, test)
-            ep_i = e_i
-            lp_i = len(L_i)
-
-        if h_i.__hash__() == hash_i and h_j.__hash__() == hash_j and \
-                h_k.__hash__() == hash_k:
-            end = time.time()
-            print(end - start)
-            break
-
-    pred_j = h_j.predict(unlabeled)
-    pred_k = h_k.predict(unlabeled)
-    pred_i = h_i.predict(unlabeled)
-
-    new_labels = []
-    for w, x, y, z in zip(unlabeled, pred_j, pred_k, pred_i):
-        if x == y and x == z:
-            new_labels.append(x)
-        elif x == y or x == z:
-            new_labels.append(x)
-        elif y == z:
-            new_labels.append(y)
-        else:
-            new_labels.append(z)
-
-    return Bunch(data=unlabeled, target=new_labels)
-
-
-if __name__ == '__main__':
-    dataset = load_iris()
-    X = dataset['data']
-    y = dataset['target']
-    # Odd order of attributes...
-    X_train, y_train, X_test, y_test = train_test_split(
-        X, y, test_size=0.8, random_state=42)
-    pred = tri_training(Bunch(data=X_train, target=X_test), Bunch(data=y_train))
-    print(classification_report(y_true=y_test, y_pred=pred['target']))
diff --git a/SemiSupervised Learning Algorithms/TriTraining.py b/SemiSupervised Learning Algorithms/TriTraining.py
new file mode 100644
index 0000000..aa7e47b
--- /dev/null
+++ b/SemiSupervised Learning Algorithms/TriTraining.py
@@ -0,0 +1,213 @@
+#!/usr/bin/env python
+# -*- coding:utf-8 -*-
+# @Filename:    Tri-Training.py
+# @Author:      Daniel Puente Ramírez
+# @Time:        27/12/21 10:25
+# @Version:     2.0
+
+from math import floor, ceil
+
+import numpy as np
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.model_selection import train_test_split
+from sklearn.naive_bayes import GaussianNB
+from sklearn.neighbors import KNeighborsClassifier
+from sklearn.preprocessing import LabelEncoder
+from sklearn.tree import DecisionTreeClassifier
+from sklearn.utils import Bunch
+
+
+class TriTraining:
+    def __init__(self, learn=None, random_state=None):
+        if learn is None or learn not in [*range(3)]:
+            raise ValueError(
+                'The learn algorithm must be passed to the constructor.\n'
+                '\t 1 for 3-NN \n\t 2 for Decision Tree \n\t 3 for Random '
+                'Forest \n\t 4 for Gaussian NB'
+            )
+        if learn == 1:
+            self.hj = KNeighborsClassifier(n_neighbors=3, n_jobs=-1, p=2)
+            self.hk = KNeighborsClassifier(n_neighbors=3, n_jobs=-1, p=2)
+            self.hi = KNeighborsClassifier(n_neighbors=3, n_jobs=-1, p=2)
+        elif learn == 2:
+            self.hj = DecisionTreeClassifier(random_state=random_state)
+            self.hk = DecisionTreeClassifier(random_state=random_state)
+            self.hi = DecisionTreeClassifier(random_state=random_state)
+        elif learn == 3:
+            self.hj = RandomForestClassifier(random_state=random_state)
+            self.hk = RandomForestClassifier(random_state=random_state)
+            self.hi = RandomForestClassifier(random_state=random_state)
+        else:
+            self.hj = GaussianNB()
+            self.hk = GaussianNB()
+            self.hi = GaussianNB()
+
+        self.random_state = random_state if random_state is not None else \
+            np.random.randint(low=0, high=10e5, size=1)[0]
+
+    def measure_error(self, classifier_j, classifier_k, labeled_data):
+        pred_j = classifier_j.predict(labeled_data)
+        pred_k = classifier_k.predict(labeled_data)
+        same = len([0 for x, y in zip(pred_j, pred_k) if x == y])
+        return (len(pred_j) - same) / same
+
+    def subsample(self, l_t, s):
+        np.random.seed(self.random_state)
+        rng = np.random.default_rng()
+        data = np.array(l_t['data'])
+        target = np.array(l_t['target'])
+        samples_index = rng.choice(len(data), size=s, replace=False)
+        samples = data[samples_index]
+        targets = target[samples_index]
+        return Bunch(data=samples, target=targets)
+
+    def fit(self, L, U, y):
+        if len(L) != len(y):
+            raise ValueError(
+                f'The dimension of the labeled data must be the same as the '
+                f'number of labels given. {len(L)} != {len(y)}'
+            )
+
+        le = LabelEncoder()
+        le.fit(y)
+        y = le.transform(y)
+
+        train, _, test, _ = train_test_split(L, y, train_size=floor(
+            len(L) / 3), stratify=y, random_state=self.random_state)
+        h_j = self.hj.fit(train, test)
+        ep_j = 0.5
+        lp_j = 0
+        train, _, test, _ = train_test_split(L, y, train_size=floor(
+            len(L) / 3), stratify=y, random_state=self.random_state)
+        h_k = self.hk.fit(train, test)
+        ep_k = 0.5
+        lp_k = 0
+        train, _, test, _ = train_test_split(L, y, train_size=floor(
+            len(L) / 3), stratify=y, random_state=self.random_state)
+        h_i = self.hi.fit(train, test)
+        ep_i = 0.5
+        lp_i = 0
+
+        while True:
+            hash_i = h_i.__hash__()
+            hash_j = h_j.__hash__()
+            hash_k = h_k.__hash__()
+
+            update_j = False
+            L_j = Bunch(data=[], target=[])
+            e_j = self.measure_error(h_j, h_k, L)
+
+            if e_j < ep_j:
+                for sample in U:
+                    sample_s = sample.reshape(1, -1)
+                    if h_j.predict(sample_s) == h_k.predict(sample_s):
+                        pred = h_i.predict(sample_s)
+                        prev_dat = list(L_j['data'])
+                        prev_tar = list(L_j['target'])
+                        prev_dat.append(sample)
+                        L_j['data'] = np.array(prev_dat)
+                        prev_tar.append(pred)
+                        L_j['target'] = np.array(prev_tar)
+
+                if lp_j == 0:
+                    lp_j = floor(e_j / (ep_j - e_j) + 1)
+
+                if lp_j < len(L_j['data']):
+                    if e_j * len(L_j['data']) < ep_j * lp_j:
+                        update_j = True
+                    elif lp_j > e_j / (ep_j - e_j):
+                        L_j = self.subsample(L_j, ceil(((ep_j * lp_j) / e_j)
+                                                       - 1))
+                        update_j = True
+
+            update_k = False
+            L_k = Bunch(data=np.array([]), target=np.array([]))
+            e_k = self.measure_error(h_j, h_k, L)
+
+            if e_k < ep_k:
+                for sample in U:
+                    sample_s = sample.reshape(1, -1)
+                    if h_j.predict(sample_s) == h_k.predict(sample_s):
+                        pred = h_i.predict(sample_s)
+                        prev_dat = list(L_k['data'])
+                        prev_tar = list(L_k['target'])
+                        prev_dat.append(sample)
+                        L_k['data'] = np.array(prev_dat)
+                        prev_tar.append(pred)
+                        L_k['target'] = np.array(prev_tar)
+
+                if lp_k == 0:
+                    lp_k = floor(e_k / (ep_k - e_k) + 1)
+
+                if lp_k < len(L_k['data']):
+                    if e_k * len(L_k['data']) < ep_k * lp_k:
+                        update_k = True
+                    elif lp_k > e_k / (ep_k - e_k):
+                        L_k = self.subsample(L_k, ceil(((ep_k * lp_k) / e_k)
+                                                       - 1))
+                        update_k = True
+
+            update_i = False
+            L_i = Bunch(data=np.array([]), target=np.array([]))
+            e_i = self.measure_error(h_j, h_k, L)
+
+            if e_i < ep_i:
+                for sample in U:
+                    sample_s = sample.reshape(1, -1)
+                    if h_j.predict(sample_s) == h_k.predict(sample_s):
+                        pred = h_i.predict(sample_s)
+                        prev_dat = list(L_i['data'])
+                        prev_tar = list(L_i['target'])
+                        prev_dat.append(sample)
+                        L_i['data'] = np.array(prev_dat)
+                        prev_tar.append(pred)
+                        L_i['target'] = np.array(prev_tar)
+
+                if lp_i == 0:
+                    lp_i = floor(e_i / (ep_i - e_i) + 1)
+
+                if lp_i < len(L_i['data']):
+                    if e_i * len(L_i['data']) < ep_i * lp_i:
+                        update_i = True
+                    elif lp_i > e_i / (ep_i - e_i):
+                        L_i = self.subsample(L_i, ceil(((ep_i * lp_i) / e_i)
+                                                       - 1))
+                        update_i = True
+
+            if update_j:
+                train = np.concatenate((L, L_j['data']), axis=0)
+                test = np.concatenate((y, np.ravel(L_j['target'])),
+                                      axis=0)
+                h_j = self.hj.fit(train, test)
+                ep_j = e_j
+                lp_j = len(L_j)
+            if update_k:
+                train = np.concatenate((L, L_k['data']), axis=0)
+                test = np.concatenate((y, np.ravel(L_k['target'])),
+                                      axis=0)
+                h_k = self.hk.fit(train, test)
+                ep_k = e_k
+                lp_k = len(L_k)
+            if update_i:
+                train = np.concatenate((L, L_i['data']), axis=0)
+                test = np.concatenate((y, np.ravel(L_i['target'])),
+                                      axis=0)
+                h_i = self.hi.fit(train, test)
+                ep_i = e_i
+                lp_i = len(L_i)
+
+            if h_i.__hash__() == hash_i and h_j.__hash__() == hash_j and \
+                    h_k.__hash__() == hash_k:
+                break
+
+    def predict(self, X):
+        labels = []
+        pred1 = self.hi.predict(X)
+        pred2 = self.hj.predict(X)
+        pred3 = self.hk.predict(X)
+
+        for p in zip(pred1, pred2, pred3):
+            count = np.bincount(p)
+            labels.append(np.where(count == np.amax(count))[0][0])
+
+        return np.array(labels)


Test patch:
diff --git a/SemiSupervised Learning Algorithms/self_training_test_percent.py b/SemiSupervised Learning Algorithms/self_training_test_percent.py
index 6714a10..4dceb55 100644
--- a/SemiSupervised Learning Algorithms/self_training_test_percent.py
+++ b/SemiSupervised Learning Algorithms/self_training_test_percent.py
@@ -3,18 +3,19 @@
 # @Filename:    self_training.py
 # @Author:      Daniel Puente Ramírez
 # @Time:        10/12/21 08:36
-from os import walk
-import os.path
-from testing import arff2sk_dataset
-from math import floor
 import copy
 import csv
-import numpy as np
+import os.path
+from math import floor
+from os import walk
 from statistics import mean
+
+import numpy as np
 from sklearn.metrics import accuracy_score
 from sklearn.model_selection import train_test_split, KFold
-from sklearn.semi_supervised import SelfTrainingClassifier
 from sklearn.neighbors import KNeighborsClassifier
+from sklearn.semi_supervised import SelfTrainingClassifier
+from testing import arff2sk_dataset
 
 
 def main():
@@ -56,7 +57,7 @@ def main():
                 if precision != 1.0:
                     indexes = np.random.choice(len(y_train),
                                                floor(len(y_train) * (
-                                                           1 - precision)),
+                                                       1 - precision)),
                                                replace=False)
                     indexes = np.setdiff1d(np.arange(len(y_train)), indexes)
                     y_train[indexes] = -1
@@ -80,4 +81,4 @@ def main():
 
 
 if __name__ == '__main__':
-    main()
\ No newline at end of file
+    main()
diff --git a/SemiSupervised Learning Algorithms/testing_driver.py b/SemiSupervised Learning Algorithms/testing_driver.py
index c7e2296..81e21cf 100644
--- a/SemiSupervised Learning Algorithms/testing_driver.py
+++ b/SemiSupervised Learning Algorithms/testing_driver.py
@@ -8,15 +8,16 @@
 from sklearn.metrics import classification_report
 from sklearn.model_selection import train_test_split
 
-from CoTraining import CoTraining
+from TriTraining import TriTraining
 
 if __name__ == '__main__':
-    model = CoTraining()
+    model = TriTraining(learn=1, random_state=42)
     iris = load_iris()
     X = iris['data']
     y = iris['target']
-    X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.2,
-                                                        stratify=y)
+    X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=0.4,
+                                                        stratify=y,
+                                                        random_state=42)
     model.fit(L=X_train, U=X_test, y=y_train)
     y_pred = model.predict(X_test)
     print(classification_report(y_test, y_pred))