KNN.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "89581652",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "\n",
    "fFloat = open(\"iris.csv\", \"r\")\n",
    "dataset = np.loadtxt(fFloat, delimiter=\",\")\n",
    "fFloat.close()\n",
    "\n",
    "x = dataset[:,0:4]\n",
    "y = dataset[:,4]\n",
    "percentTrainingset = 0.8\n",
    "np.random.seed(42)\n",
    "TrainSet = np.random.choice(x.shape[0], int(x.shape[0]*percentTrainingset), replace=False)\n",
    "XTrain = x[TrainSet,:]\n",
    "YTrain = y[TrainSet]\n",
    "TestSet = np.delete(np.arange(0, len(y)), TrainSet)\n",
    "XTest = x[TestSet,:]\n",
    "YTest = y[TestSet]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "626e887c",
   "metadata": {},
   "outputs": [],
   "source": [
    "def plainKNNclassification(xTrain, yTrain, xQuery, k, normOrd=None):\n",
    "    xMin = xTrain.min(axis=0)\n",
    "    xMax = xTrain.max(axis=0)\n",
    "    xTrain = (xTrain - xMin) / (xMax-xMin)\n",
    "    xQuery = (xQuery - xMin) / (xMax - xMin)\n",
    "    diff = xTrain - xQuery\n",
    "    dist = np.linalg.norm(diff, axis=1, ord=normOrd)\n",
    "    knearest = np.argsort(dist)[:k]\n",
    "    (classification, counts) = np.unique(YTrain[knearest], return_counts=True)\n",
    "    chosenClass = np.argmax(counts)\n",
    "    return (classification[chosenClass])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "1eb66663",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[4.9 2.5 4.5 1.7] wurde als 2 statt 3 klassifiziert\n",
      "[7.2 3.  5.8 1.6] wurde als 2 statt 3 klassifiziert\n"
     ]
    }
   ],
   "source": [
    "errors = 0\n",
    "for i in range(len(YTest)):\n",
    "    myClass = plainKNNclassification(XTrain, YTrain, XTest[i,:], 3)\n",
    "    if myClass != YTest[i]:\n",
    "        errors = errors + 1\n",
    "        print('%s wurde als %d statt %d klassifiziert' % (str(XTest[i,:]), myClass, YTest[i]))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "2c10914f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Motivate weighted average of closest nodes using two moon problems\n",
    "def twoMoonsProblems(SamplesPerMoon=240, pNoise=2):\n",
    "    tMoon0 = np.linspace(0, np.pi, SamplesPerMoon)\n",
    "    tMoon1 = np.linspace(0, np.pi, SamplesPerMoon)\n",
    "    Moon0x = np.cos(tMoon0)\n",
    "    Moon0y = np.sin(tMoon0)\n",
    "    Moon1x = 1 - np.cos(tMoon1)\n",
    "    Moon1y = 0.5 - np.sin(tMoon1)\n",
    "    \n",
    "    X = np.vstack((np.append(Moon0x, Moon1x), np.append(Moon0y, Moon1y))).T\n",
    "    X = X + pNoise/100*np.random.normal(size=X.shape)\n",
    "    Y = np.hstack([np.zeros(SamplesPerMoon), np.ones(SamplesPerMoon)])\n",
    "    \n",
    "    return X, Y"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "5ed86ea9",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Further, we utilize a kd tree to reduce the lookup time\n",
    "from scipy.spatial import KDTree\n",
    "\n",
    "class knnRegression:\n",
    "    def fit(self, X, Y):\n",
    "        self.xMin = X.min(axis=0)\n",
    "        self.xMax = x.max(axis=0)\n",
    "        self.XTrain = (X - self.xMin) / (self.xMax - self.xMin)\n",
    "        self.kdTree = KDTree(self.XTrain)\n",
    "        self.YTrain = Y\n",
    "        \n",
    "    def predict(self, X, k=3, smear = 1):\n",
    "        X = (X - self.xMin) / (self.xMax - self.xMin) # scaling\n",
    "        (dist, neighbours) = self.kdTree.query(X,k) # find k closest neighbors in X using kdtree\n",
    "        distsum = np.sum( 1/(dist+smear/k), axis=1) \n",
    "        distsum = np.repeat(distsum[:,None],k,axis=1)\n",
    "        dist = (1/distsum)*1/(dist+smear/k)\n",
    "        y = np.sum( dist*YTrain[neighbours], axis=1)\n",
    "        return(y)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "441e2684",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0.024800797417177438\n"
     ]
    }
   ],
   "source": [
    "samples = 5000\n",
    "pNoise = 1\n",
    "myK = 3\n",
    "mysmear = 0.5\n",
    "\n",
    "np.random.seed(42)\n",
    "x = np.random.rand(samples, 2)\n",
    "y = np.tanh( 500*( (1/16) - (x[:,0]-0.5)**2 - (x[:,1]-0.5)**2) )\n",
    "Noise = np.random.normal(size=len(y))\n",
    "y = (1+Noise*pNoise/100)*y\n",
    "\n",
    "percentTrainingset = 0.8\n",
    "TrainSet = np.random.choice(x.shape[0], int(x.shape[0]*percentTrainingset), replace=False)\n",
    "XTrain = x[TrainSet, :]\n",
    "YTrain = y[TrainSet]\n",
    "TestSet = np.delete(np.arange(0, len(y)), TrainSet)\n",
    "XTest = x[TestSet,:]\n",
    "YTest = y[TestSet]\n",
    "\n",
    "myRegression = knnRegression()\n",
    "\n",
    "myRegression.fit(XTrain, YTrain)\n",
    "yP = myRegression.predict(XTest, k=myK, smear=mysmear)\n",
    "diff = yP - YTest\n",
    "MAE = np.mean(np.abs(diff))\n",
    "print(MAE) # Mean absolute error"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "id": "b643a1b7",
   "metadata": {},
   "outputs": [],
   "source": [
    "# TODO: Adapt classification into a similar class format using a weighted distance choice\n",
    "\n",
    "class knnClassification:\n",
    "    def fit(self, X, Y):\n",
    "        self.xMin = X.min(axis=0)\n",
    "        self.xMax = x.max(axis=0)\n",
    "        self.XTrain = (X - self.xMin) / (self.xMax - self.xMin)\n",
    "        self.kdTree = KDTree(self.XTrain)\n",
    "        self.YTrain = Y\n",
    "        \n",
    "    def predict(self, X, k=3, smear=1):\n",
    "        X = (X - self.xMin) / (self.xMax - self.xMin) # scaling\n",
    "        (dist, neighbours) = self.kdTree.query(X,k) # find k closest neighbors in X using kdtree\n",
    "        distsum = np.sum( 1/(dist+smear/k))  # Sum up all distances, including smear\n",
    "        distsum = np.repeat(distsum,k) # Repeat distance sum k-times [[a,a,a],[b,b,b],[c,c,c]]\n",
    "        dist = (1/distsum)*1/(dist+smear/k) # Reciprocal of distsum multipled with reciprocal of dist+smear/k\n",
    "        classTotalDistances = {}\n",
    "        for neighbor in range(len(neighbours)):\n",
    "            if(YTrain[neighbours[neighbor]] in classTotalDistances):\n",
    "                classTotalDistances[YTrain[neighbours[neighbor]]] += abs(dist[neighbor])\n",
    "            else:\n",
    "                classTotalDistances[YTrain[neighbours[neighbor]]] = abs(dist[neighbor])\n",
    "        return max(classTotalDistances, key=classTotalDistances.get)\n",
    "        \n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "91a12ca0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[4.9 2.5 4.5 1.7] wurde als 2 statt 3 klassifiziert\n",
      "[7.2 3.  5.8 1.6] wurde als 2 statt 3 klassifiziert\n"
     ]
    }
   ],
   "source": [
    "errors = 0\n",
    "knn = knnClassification()\n",
    "knn.fit(XTrain, YTrain)\n",
    "for i in range(len(YTest)):\n",
    "    myClass = knn.predict(XTest[i,:])\n",
    "    if myClass != YTest[i]:\n",
    "        errors = errors + 1\n",
    "        print('%s wurde als %d statt %d klassifiziert' % (str(XTest[i,:]), myClass, YTest[i]))"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}