{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [],
"source": [
"import numpy as np\n",
"from scipy.special import expit\n",
"from scipy.optimize import minimize\n",
"from copy import deepcopy\n",
"from sklearn.base import BaseEstimator, ClassifierMixin\n",
"from sklearn.datasets import load_iris\n",
"from sklearn.ensemble import BaggingClassifier as skBaggingClassifier\n",
"#from sklearn.linear_model import LogisticRegression"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"def accuracy_score(y_true, y_pred):\n",
" return np.mean(y_true == y_pred)"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [],
"source": [
"class LogisticRegression(BaseEstimator, ClassifierMixin):\n",
" def __init__(self, C=1.0):\n",
" self.C = C\n",
"\n",
" def _encode(self, y):\n",
" classes = np.unique(y)\n",
" y_train = np.full((y.shape[0], len(classes)), -1)\n",
" for i, c in enumerate(classes):\n",
" y_train[y == c, i] = 1\n",
" if len(classes) == 2:\n",
" y_train = y_train[:, 1].reshape(-1, 1)\n",
" return classes, y_train\n",
"\n",
" @staticmethod\n",
" def _cost_grad(w, X, y, alpha):\n",
" def _log_logistic(x):\n",
" if x > 0:\n",
" return -np.log(1 + np.exp(-x))\n",
" else:\n",
" return x - np.log(1 + np.exp(x))\n",
" yz = y * (np.dot(X, w[:-1]) + w[-1])\n",
" cost = -np.sum(np.vectorize(_log_logistic)(yz)) + 0.5 * alpha * np.dot(w[:-1], w[:-1])\n",
" grad = np.zeros(len(w))\n",
" t = (expit(yz) - 1) * y\n",
" grad[:-1] = np.dot(X.T, t) + alpha * w[:-1]\n",
" grad[-1] = np.sum(t)\n",
" return cost, grad\n",
"\n",
" def _solve_lbfgs(self, X, y):\n",
" result = np.zeros((y.shape[1], X.shape[1] + 1))\n",
" for i in range(y.shape[1]):\n",
" cur_y = y[:, i]\n",
" w0 = np.zeros(X.shape[1] + 1)\n",
" res = minimize(fun=self._cost_grad, jac=True, x0=w0,\n",
" args=(X, cur_y, 1 / self.C), method='L-BFGS-B')\n",
" result[i] = res.x\n",
" return result[:, :-1], result[:, -1]\n",
"\n",
" def fit(self, X, y):\n",
" self.classes_, y_train = self._encode(y)\n",
" self.coef_, self.intercept_ = self._solve_lbfgs(X, y_train)\n",
" return self\n",
"\n",
" def decision_function(self, X):\n",
" scores = np.dot(X, self.coef_.T) + self.intercept_\n",
" if scores.shape[1] == 1:\n",
" return scores.ravel()\n",
" else:\n",
" return scores\n",
"\n",
" def predict(self, X):\n",
" scores = self.decision_function(X)\n",
" if len(scores.shape) == 1:\n",
" indices = (scores > 0).astype(int)\n",
" else:\n",
" indices = np.argmax(scores, axis=1)\n",
" return self.classes_[indices]\n",
"\n",
" def predict_proba(self, X):\n",
" scores = self.decision_function(X)\n",
" prob = expit(scores)\n",
" if len(scores.shape) == 1:\n",
" prob = np.vstack((1 - prob, prob)).T\n",
" else:\n",
" prob /= np.sum(prob, axis=1)[:, np.newaxis]\n",
" return prob"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"class BaggingClassifier():\n",
" def __init__(self, base_estimator, n_estimators, oob_score, random_state):\n",
" self.base_estimator = base_estimator\n",
" self.n_estimators = n_estimators\n",
" self.oob_score = oob_score\n",
" self.random_state = random_state\n",
"\n",
" def fit(self, X, y):\n",
" self.classes_, y_train = np.unique(y, return_inverse=True)\n",
" self.n_classes_ = len(self.classes_)\n",
" MAX_INT = np.iinfo(np.int32).max\n",
" rng = np.random.RandomState(self.random_state)\n",
" self._seeds = rng.randint(MAX_INT, size=self.n_estimators)\n",
" self.estimators_ = []\n",
" self.estimators_samples_ = []\n",
" for i in range(self.n_estimators):\n",
" est = deepcopy(self.base_estimator)\n",
" rng = np.random.RandomState(self._seeds[i])\n",
" sample_indices = rng.randint(0, X.shape[0], X.shape[0])\n",
" self.estimators_samples_.append(sample_indices)\n",
" est.fit(X[sample_indices], y_train[sample_indices])\n",
" self.estimators_.append(est)\n",
" if self.oob_score:\n",
" self._set_oob_score(X, y_train)\n",
" return self\n",
"\n",
" def _set_oob_score(self, X, y):\n",
" predictions = np.zeros((X.shape[0], self.n_classes_))\n",
" for i in range(self.n_estimators):\n",
" mask = np.ones(X.shape[0], dtype=bool)\n",
" mask[self.estimators_samples_[i]] = False\n",
" predictions[mask] += self.estimators_[i].predict_proba(X[mask])\n",
" self.oob_decision_function_ = predictions / np.sum(predictions, axis=1)[:, np.newaxis]\n",
" self.oob_score_ = accuracy_score(y, np.argmax(predictions, axis=1))\n",
"\n",
" def predict_proba(self, X):\n",
" proba = np.zeros((X.shape[0], self.n_classes_))\n",
" for i in range(self.n_estimators):\n",
" proba += self.estimators_[i].predict_proba(X)\n",
" proba /= self.n_estimators\n",
" return proba\n",
"\n",
" def predict(self, X):\n",
" proba = self.predict_proba(X)\n",
" return self.classes_[np.argmax(proba, axis=1)]"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [],
"source": [
"X, y = load_iris(return_X_y=True)\n",
"clf1 = BaggingClassifier(base_estimator=LogisticRegression(),\n",
" n_estimators=100, oob_score=True, random_state=0).fit(X, y)\n",
"clf2 = skBaggingClassifier(base_estimator=LogisticRegression(),\n",
" n_estimators=100, oob_score=True, random_state=0).fit(X, y)\n",
"assert np.allclose(clf1._seeds, clf2._seeds)\n",
"assert np.array_equal(clf1.estimators_samples_, clf2.estimators_samples_)\n",
"for i in range(clf1.n_estimators):\n",
" assert np.allclose(clf1.estimators_[i].coef_, clf2.estimators_[i].coef_)\n",
" assert np.allclose(clf1.estimators_[i].intercept_, clf2.estimators_[i].intercept_)\n",
"assert np.allclose(clf1.oob_decision_function_, clf2.oob_decision_function_)\n",
"assert np.allclose(clf1.oob_score_, clf2.oob_score_)\n",
"prob1 = clf1.predict_proba(X)\n",
"prob2 = clf2.predict_proba(X)\n",
"assert np.allclose(prob1, prob2)\n",
"pred1 = clf1.predict(X)\n",
"pred2 = clf2.predict(X)\n",
"assert np.array_equal(pred1, pred2)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "dev",
"language": "python",
"name": "dev"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.7.3"
}
},
"nbformat": 4,
"nbformat_minor": 2
}