01.5.kNN-torch1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# PyTorch1实现kNN分类和回归"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "python version: sys.version_info(major=3, minor=7, micro=2, releaselevel='final', serial=0)\n",
      "torch 1.6.0\n"
     ]
    }
   ],
   "source": [
    "import sys\n",
    "from collections import Counter\n",
    "from collections.abc import Iterable\n",
    "\n",
    "import numpy as np\n",
    "import torch\n",
    "from sklearn.datasets import load_iris, load_boston\n",
    "from sklearn.model_selection import train_test_split\n",
    "\n",
    "\n",
    "print(\"python version:\", sys.version_info)\n",
    "print(torch.__name__, torch.__version__)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. PyTorch实现kNN基类"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TorchKNNBase:\n",
    "    \"\"\"KNN basic class with PyTorch.\n",
    "    \n",
    "    Attributes:\n",
    "        n_neighbors: A int number, number of neighbors.\n",
    "        _metric: A method object, choose from {_manhattan_distance, _euclidean_distance, _chebyshev_distance}.\n",
    "        _X_train: feature data for training. A tf.Tensor matrix of (n_samples, n_features) shape, \n",
    "            data type must be continuous value type. \n",
    "        _y_train: label data for training. A tf.Tensor array of (n_samples, ) shape, \n",
    "            data type must be discrete value.\n",
    "    \"\"\"\n",
    "    def __new__(cls, *args, **kwargs):\n",
    "        raise Exception(\"Can't instantiate an object from TorchKNNBase! \")\n",
    "        \n",
    "    def __init__(self, n_neighbors=5, metric=\"euclidean\"):\n",
    "        \"\"\"Init method.\n",
    "    \n",
    "        Args:\n",
    "            n_neighbors: int, optional (default = 5), the integer must greater then 0.\n",
    "                Number of neighbors to use by default for :meth:`kneighbors` queries.\n",
    "            metric: {\"manhattan\", \"euclidean\", \"chebyshev\"}, optional, default 'euclidean'.\n",
    "        \n",
    "        Raises:\n",
    "            ValueError: metric value is out of options.\n",
    "            AssertionError: n_neighbors value is not a integer or n_neighbors > 0. \n",
    "        \"\"\"\n",
    "        assert isinstance(n_neighbors, int) and n_neighbors > 0\n",
    "        self.n_neighbors = n_neighbors\n",
    "        if metric == \"manhattan\":\n",
    "            self._metric = self._manhattan_distance\n",
    "        elif metric == \"euclidean\":\n",
    "            self._metric = self._euclidean_distance\n",
    "        elif metric == \"chebyshev\":\n",
    "            self._metric = self._chebyshev_distance\n",
    "        else:\n",
    "            raise ValueError(f'No such metric as {metric}, please option from: {\"manhattan\", \"euclidean\", \"chebyshev\"}')\n",
    "        self._X_train, self._y_train = [None] * 2\n",
    "    \n",
    "    def fit(self, X_train, y_train):\n",
    "        \"\"\"method for training model. \n",
    "        \n",
    "        Args:\n",
    "            X_train: A matrix of (n_samples, n_features) shape, data type must be continuous value type. \n",
    "            y_train: A array of (n_samples, ) shape.\n",
    "        \n",
    "        Raises:\n",
    "            AssertionError: X_train value or y_train value with a mismatched shape.\n",
    "        \"\"\"\n",
    "        assert isinstance(X_train, Iterable) and isinstance(y_train, Iterable)\n",
    "        assert len(X_train) == len(y_train)\n",
    "        self._X_train = torch.as_tensor(X_train, dtype=torch.float32)\n",
    "        self._y_train = y_train if isinstance(y_train, torch.Tensor) else torch.as_tensor(y_train)\n",
    "    \n",
    "    def predict(self, X_test):\n",
    "        \"\"\"predict test data.\n",
    "        \n",
    "        Args:\n",
    "            X_test: A np.ndarray matrix of (n_samples, n_features) shape, \n",
    "                or a np.ndarray array of (n_features, ) shape.\n",
    "            \n",
    "        Returns:\n",
    "            A list for samples predictions or a single prediction.\n",
    "        \n",
    "        Raises:\n",
    "            ValueError: X_test value with a mismatched shape.\n",
    "        \"\"\"\n",
    "        assert isinstance(X_test, Iterable)\n",
    "        X_test = torch.as_tensor(X_test, dtype=torch.float32)\n",
    "        \n",
    "        if X_test.shape == (self._X_train.shape[1], ):\n",
    "            y_pred = self._predict_sample(X_test)\n",
    "        elif X_test.shape[1] == self._X_train.shape[1]: \n",
    "            y_pred = []\n",
    "            for sample in X_test:\n",
    "                y_pred.append(self._predict_sample(sample))\n",
    "        else:\n",
    "            raise ValueError(\"Mismatched shape for X_test\")\n",
    "        return y_pred\n",
    "    \n",
    "    def _manhattan_distance(self, x):\n",
    "        return torch.sum(torch.abs(self._X_train - x), dim=1)    \n",
    "    \n",
    "    def _euclidean_distance(self, x):\n",
    "        return torch.sqrt(torch.sum(torch.square(self._X_train - x), dim=1))\n",
    "    \n",
    "    def _chebyshev_distance(self, x):\n",
    "        return torch.max(torch.abs(self._X_train - x), dim=1)\n",
    "    \n",
    "    def _find_k_labels(self, sample):\n",
    "        distance = self._metric(sample)\n",
    "        _, k_nearest_index = torch.topk(distance, self.n_neighbors, largest=False)\n",
    "        k_labels = self._y_train[k_nearest_index]\n",
    "        return k_labels\n",
    "    \n",
    "    def _predict_sample(self, sample):\n",
    "        raise Exception(\"Can call predict method for NumpyKNNBase object! \")\n",
    "        \n",
    "    def _score_validation(self, X_test, y_test):\n",
    "        assert isinstance(X_test, Iterable) and isinstance(y_test, Iterable)\n",
    "        assert len(X_test) == len(y_test)\n",
    "        y_test = y_test if isinstance(y_test, torch.Tensor) else torch.as_tensor(y_test)\n",
    "        return X_test, y_test"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. PyTorch实现kNN Classifier"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TorchKNNClassifier(TorchKNNBase):\n",
    "    \"\"\"kNN Classifier with PyTorch, explicitly inherits from TorchKNNBase already.\n",
    "    \n",
    "    Attributes:\n",
    "        n_neighbors: A int number, number of neighbors.\n",
    "        _metric: A method object, choose from {_manhattan_distance, _euclidean_distance, _chebyshev_distance}.\n",
    "        _X_train: feature data for training. A torch.Tensor matrix of (n_samples, n_features) shape, \n",
    "            data type must be continuous value type. \n",
    "        _y_train: label data for training. A torch.Tensor array of (n_samples, ) shape, \n",
    "            data type must be discrete value.\n",
    "    \"\"\"\n",
    "    def __new__(cls, *args, **kwargs):\n",
    "        return object.__new__(cls, *args, **kwargs)\n",
    "    \n",
    "    def score(self, X_test, y_test):\n",
    "        \"\"\"Use test dataset to evaluate the trained model.\n",
    "        \n",
    "        Args:\n",
    "            X_test: A np.ndarray matrix of (n_samples, n_features) shape.\n",
    "            y_test: A np.ndarray array of (n_samples, ) shape. data type must be\n",
    "                discrete value.\n",
    "        Returns:\n",
    "            return accuracy, a float number. accuracy = correct_count / y_test.shape[0]\n",
    "        \"\"\"\n",
    "        X_test, y_test = self._score_validation(X_test, y_test)\n",
    "        \n",
    "        y_pred = torch.as_tensor(self.predict(X_test))\n",
    "        correct_count = torch.sum(y_pred == y_test)\n",
    "        accuracy = correct_count.item() / y_test.shape[0]\n",
    "        return accuracy\n",
    "    \n",
    "    def _predict_sample(self, sample):\n",
    "        k_labels = self._find_k_labels(sample)\n",
    "        pred = Counter(k_labels.numpy()).most_common(1)[0][0]\n",
    "        return pred"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1.0"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "X_data, y_data = load_iris(return_X_y=True)\n",
    "X_train, X_test, y_train, y_test = train_test_split(X_data, y_data, shuffle=True, random_state=1)\n",
    "cls = TorchKNNClassifier()\n",
    "cls.fit(X_train, y_train)\n",
    "cls.score(X_test, y_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. PyTorch实现kNN Regressor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "class TorchKNNRegressor(TorchKNNBase): \n",
    "    \"\"\"kNN Regressor with Pytorch, explicitly inherits from TorchKNNBase already.\n",
    "    \n",
    "    Attributes:\n",
    "        n_neighbors: A int number, number of neighbors.\n",
    "        _metric: A method object, choose from {_manhattan_distance, _euclidean_distance, _chebyshev_distance}.\n",
    "        _X_train: feature data for training. A torch.Tensr matrix of (n_samples, n_features) shape, \n",
    "            data type must be continuous value type. \n",
    "        _y_train: label data for training. A torch.Tensor array of (n_samples, ) shape, \n",
    "            data type must be discrete value.\n",
    "    \"\"\"\n",
    "    def __new__(cls, *args, **kwargs):\n",
    "        return object.__new__(cls, *args, **kwargs)\n",
    "    \n",
    "    def score(self, X_test, y_test):\n",
    "        \"\"\"Use test dataset to evaluate the trained model.\n",
    "        \n",
    "        Args:\n",
    "            X_test: A np.ndarray matrix of (n_samples, n_features) shape.\n",
    "            y_test: A np.ndarray array of (n_samples, ) shape. data type must be\n",
    "                discrete value.\n",
    "        Returns:\n",
    "            return R^2, R^2 = 1 - u / v. u = sum((y_pred - y_true)^2), v = sum((y_true - y_true_mean)^2)\n",
    "        \"\"\"\n",
    "        X_test, y_test = self._score_validation(X_test, y_test)\n",
    "        \n",
    "        y_pred = torch.as_tensor(self.predict(X_test))\n",
    "        y_true_mean = torch.mean(y_test)\n",
    "        u = torch.sum(torch.square(y_pred - y_test), dim=0)\n",
    "        v = torch.sum(torch.square(y_test - y_true_mean), dim=0)\n",
    "        r_squared = 1 - u / v\n",
    "        return r_squared.item()\n",
    "    \n",
    "    def _predict_sample(self, sample):\n",
    "        k_labels = self._find_k_labels(sample)\n",
    "        pred = torch.mean(k_labels)\n",
    "        return pred.item()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 3.1 boston房价数据集验证算法"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.5281871701699631"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "X_data, y_data = load_boston(return_X_y=True)\n",
    "X_train, X_test, y_train, y_test = train_test_split(X_data, y_data, shuffle=True, random_state=1)\n",
    "cls = TorchKNNRegressor()\n",
    "cls.fit(X_train, y_train)\n",
    "cls.score(X_test, y_test)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. 总结"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "1. PyTorch API\n",
    "    > torch.sum(input, dim)   \n",
    "    > torch.mean(input, dim)  \n",
    "    > torch.max(input, dim)  \n",
    "    >  \n",
    "    > torch.topk(input, k, dim, largest)  # 求topk独家神器\n",
    "    >  \n",
    "    > tensor.item()  \n",
    "    > tensor.numpy()  \n",
    "    >  \n",
    "    > torch.sqrt()  \n",
    "    > torch.square()  \n",
    "    > torch.abs()  \n",
    "    > torch.as_tensor(data, dtype)  \n",
    "    >  \n",
    "    > torch.float32\n",
    "    \n",
    "2. Numpy vs TensorFlow2 vs PyTorch1\n",
    "    > |Numpy|TensorFlow2|PyTorch1|\n",
    "    > |-|-|-|\n",
    "    > |np.sum(a, axis)|tf.reduce_sum(tensor, axis)|torch.sum(tensor, dim)|\n",
    "    > |np.mean(a, axis)|tf.reduce_mean(tensor, axis)|torch.mean(tensor, dim)|  \n",
    "    > |np.max(a, axis)|tf.reduce_max(input_tensor, axis)|torch.max(tensor, dim)|\n",
    "    > |np.argsort(a, axis)|tf.argsort(input_tensor, axis)|torch.argsort(input, dim)|\n",
    "    > |-|tf.math.top_k(input, k, sorted, name)|torch.topk(input, k, dim, largest, sorted)| \n",
    "    > |tensor1[tensor2]|tf.gather(params=tensor1, indices=tensor2)|tensor1[tensor2]|\n",
    "    > |1|a = tf.constant(1) a.numpy() == 1| a = torch.Tensor(1) a.numpy() == np.ndarray(1)|\n",
    "    > |np.array(object, dtype)|tf.constant(value, dtype), tf.Variable()|torch.as_tensor(data, dtype)|\n",
    "    > |np.float32|tf.float32|torch.float32|\n",
    "    \n",
    "3. tf2转torch1之基础API（一）\n",
    "    > tf.reduce_改torch.\n",
    "    > axis改dim\n",
    "    > tf.constant改torch.as_tensor\n",
    "    > tf.float32改torch.float32\n",
    "    > tf.cast(tensor, dtype=tf.float32)改tensor.to(torch.float32)  \n",
    "    > torch.IntTensor不能直接除以一个int值！tf.Tensor可以，并返回float64的Tensor"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 作业\n",
    "\n",
    "1. 惯例，敲两遍以上。  \n",
    "2. 重新梳理1.1～1.5节的内容，整理笔记，画思维导图，变成自己的知识。\n",
    "3. 试着实现kd树"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 相关链接\n",
    "\n",
    "<a href=\"./01.1.kNN.ipynb\" style=\"\"> 1.1 kNN k近邻算法原理 </a>  \n",
    "<a href=\"./01.2.kNN-sklearn.ipynb\" style=\"\"> 1.2 sklearn中使用kNN做分类、回归任务 </a>  \n",
    "<a href=\"./01.3.kNN-numpy.ipynb\" style=\"\"> 1.3 numpy实现kNN分类和回归 </a>    \n",
    "<a href=\"./01.4.kNN-tf2.ipynb\"> 1.4 TensorFlow2实现kNN分类和回归 </a>    \n",
    "  \n",
    "<a href=\"./02.1.LinearRegression.ipynb\"> 2.1 Linear Regression线性回归算法原理 </a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 项目源码  \n",
    "\n",
    "https://github.com/LossJ     \n",
    "进入后点击Statistic-Machine-Learning"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 其他"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "备注：学校/公司-真实姓名。  \n",
    "**不要问你自己的项目遇到的问题！不要问课堂以外的问题！不要问课上留的作业！**  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./imgs/1.WeChatQRCode.jpg\" width=\"450px\" align=\"left\">"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}