qlearning_tictactoe.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 76,
   "id": "f1a7f899",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "from random import randint"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "id": "4262a1e9",
   "metadata": {},
   "outputs": [],
   "source": [
    "loops = {0: (0,0), 1: (1, 0), 2: (2, 0), 3: (0,1), 4: (1, 1), 5: (2, 1), 6: (0, 2), 7: (1, 2), 8: (2, 2)}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 163,
   "id": "76608325",
   "metadata": {},
   "outputs": [],
   "source": [
    "def random_boards():\n",
    "    pos_boards = []\n",
    "    boards = []\n",
    "    for i in range(100):\n",
    "        l = []\n",
    "        b = {(0,0): randint(0,2),\n",
    "            (1,0): randint(0,2),\n",
    "            (2,0): randint(0,2),\n",
    "            (0,1): randint(0,2),\n",
    "            (1,1): randint(0,2),\n",
    "            (2,1): randint(0,2),\n",
    "            (0,2): randint(0,2),\n",
    "            (1,2): randint(0,2),\n",
    "            (2,2): randint(0,2)}\n",
    "        one = False\n",
    "        for pos, val in b.items():\n",
    "            if val == 0:\n",
    "                new_board = b.copy()\n",
    "                new_board[pos] = 1\n",
    "                l.append(new_board)\n",
    "                one = True\n",
    "        if one:\n",
    "            pos_boards.append(l)\n",
    "            boards.append(b)\n",
    "    return boards, pos_boards"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 164,
   "id": "a1b13d60",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_hash(board):\n",
    "    return hash(frozenset(board.items()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 165,
   "id": "ab7724a4",
   "metadata": {},
   "outputs": [],
   "source": [
    "def random_state_hash(boards):\n",
    "    hashes = {}\n",
    "    for b in boards:\n",
    "        for j in b:\n",
    "            b = get_hash(j)\n",
    "            hashes[b] = randint(0,1000)\n",
    "    return hashes"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 166,
   "id": "81d27c7c",
   "metadata": {},
   "outputs": [],
   "source": [
    "boards, pos_boards = random_boards()\n",
    "state_hash_value = random_state_hash(pos_boards)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 167,
   "id": "9acea690",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_best_next_board(next_board, best_next_board):\n",
    "    hash_next_board = get_hash(next_board)\n",
    "    hash_best_next_board = get_hash(best_next_board)\n",
    "#     print(hash_next_board)\n",
    "#     print(hash_best_next_board)\n",
    "#     print(state_hash_value[hash_next_board])\n",
    "#     print(state_hash_value[hash_best_next_board])\n",
    "\n",
    "    if state_hash_value[hash_next_board] > state_hash_value[hash_best_next_board]:\n",
    "        #print(state_hash_value[hash_next_board])\n",
    "        #print(next_board)\n",
    "        return next_board\n",
    "    else: \n",
    "        #print(state_hash_value[hash_best_next_board])\n",
    "        #print(best_next_board)\n",
    "        return best_next_board"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 173,
   "id": "3bbb7a0e",
   "metadata": {},
   "outputs": [],
   "source": [
    "## recursion best position\n",
    "def choose(i, board_local, last):\n",
    "    ## the condition here is just to test out the removal of a value, \n",
    "    ## we should remove the positions where a player already played\n",
    "    next_board = board_local.copy()\n",
    "    \n",
    "    # if you went to all possibilities\n",
    "    if i == 0:\n",
    "        next_board[loops[last]] = 1\n",
    "        return next_board\n",
    "    \n",
    "    if next_board[loops[i]] != 0:\n",
    "        ## pass this possibility\n",
    "        best_next_board = choose(i-1, board_local, last)\n",
    "        return best_next_board\n",
    "        \n",
    "    next_board[loops[i]] = 1\n",
    "    best_next_board = choose(i-1, board_local, i)\n",
    "    best_next_board = get_best_next_board(next_board, best_next_board)\n",
    "    return best_next_board"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 176,
   "id": "13509aa9",
   "metadata": {
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "#Test all the boards and check each time in pos_board that the return is the max\n",
    "for b in boards:\n",
    "    best = choose(8, b, 0)\n",
    "    real_max = state_hash_value[get_hash(best)]\n",
    "    pos = pos_boards[i]\n",
    "    calc_max = 0\n",
    "    for p in pos:\n",
    "        tmp = state_hash_value[get_hash(best)]\n",
    "        calc_max = tmp if tmp > calc_max else calc_max\n",
    "    assert calc_max == real_max, \"error on the max\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 159,
   "id": "1ba79f2f",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{(0, 0): 1,\n",
       " (1, 0): 1,\n",
       " (2, 0): 1,\n",
       " (0, 1): 2,\n",
       " (1, 1): 1,\n",
       " (2, 1): 2,\n",
       " (0, 2): 1,\n",
       " (1, 2): 1,\n",
       " (2, 2): 1}"
      ]
     },
     "execution_count": 159,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "best"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 270,
   "id": "87151ff8",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "-4764985156899373092"
      ]
     },
     "execution_count": 270,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "hash(frozenset(best.items()))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "id": "7eb2724e",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pickle\n",
    "\n",
    "BOARD_ROWS = 3\n",
    "BOARD_COLS = 3\n",
    "\n",
    "\n",
    "class State:\n",
    "    def __init__(self, p1, p2):\n",
    "        self.board = np.zeros((BOARD_ROWS, BOARD_COLS))\n",
    "        self.p1 = p1\n",
    "        self.p2 = p2\n",
    "        self.isEnd = False\n",
    "        self.boardHash = None\n",
    "        # init p1 plays first\n",
    "        self.playerSymbol = 1\n",
    "\n",
    "    # get unique hash of current board state\n",
    "    def getHash(self):\n",
    "        self.boardHash = str(self.board.reshape(BOARD_COLS * BOARD_ROWS))\n",
    "        return self.boardHash\n",
    "\n",
    "    def winner(self):\n",
    "        # row\n",
    "        for i in range(BOARD_ROWS):\n",
    "            if sum(self.board[i, :]) == 3:\n",
    "                self.isEnd = True\n",
    "                return 1\n",
    "            if sum(self.board[i, :]) == -3:\n",
    "                self.isEnd = True\n",
    "                return -1\n",
    "        # col\n",
    "        for i in range(BOARD_COLS):\n",
    "            if sum(self.board[:, i]) == 3:\n",
    "                self.isEnd = True\n",
    "                return 1\n",
    "            if sum(self.board[:, i]) == -3:\n",
    "                self.isEnd = True\n",
    "                return -1\n",
    "        # diagonal\n",
    "        diag_sum1 = sum([self.board[i, i] for i in range(BOARD_COLS)])\n",
    "        diag_sum2 = sum([self.board[i, BOARD_COLS - i - 1] for i in range(BOARD_COLS)])\n",
    "        diag_sum = max(abs(diag_sum1), abs(diag_sum2))\n",
    "        if diag_sum == 3:\n",
    "            self.isEnd = True\n",
    "            if diag_sum1 == 3 or diag_sum2 == 3:\n",
    "                return 1\n",
    "            else:\n",
    "                return -1\n",
    "\n",
    "        # tie\n",
    "        # no available positions\n",
    "        if len(self.availablePositions()) == 0:\n",
    "            self.isEnd = True\n",
    "            return 0\n",
    "        # not end\n",
    "        self.isEnd = False\n",
    "        return None\n",
    "\n",
    "    def availablePositions(self):\n",
    "        positions = []\n",
    "        for i in range(BOARD_ROWS):\n",
    "            for j in range(BOARD_COLS):\n",
    "                if self.board[i, j] == 0:\n",
    "                    positions.append((i, j))  # need to be tuple\n",
    "        return positions\n",
    "\n",
    "    def updateState(self, position):\n",
    "        self.board[position] = self.playerSymbol\n",
    "        # switch to another player\n",
    "        self.playerSymbol = -1 if self.playerSymbol == 1 else 1\n",
    "\n",
    "    # only when game ends\n",
    "    def giveReward(self):\n",
    "        result = self.winner()\n",
    "        # backpropagate reward\n",
    "        if result == 1:\n",
    "            self.p1.feedReward(1)\n",
    "            self.p2.feedReward(0)\n",
    "        elif result == -1:\n",
    "            self.p1.feedReward(0)\n",
    "            self.p2.feedReward(1)\n",
    "        else:\n",
    "            self.p1.feedReward(0.1)\n",
    "            self.p2.feedReward(0.5)\n",
    "\n",
    "    # board reset\n",
    "    def reset(self):\n",
    "        self.board = np.zeros((BOARD_ROWS, BOARD_COLS))\n",
    "        self.boardHash = None\n",
    "        self.isEnd = False\n",
    "        self.playerSymbol = 1\n",
    "\n",
    "    def play(self, rounds=100):\n",
    "        for i in range(rounds):\n",
    "            if i % 1000 == 0:\n",
    "                print(self.isEnd)\n",
    "                print(\"Rounds {}\".format(i))\n",
    "            while not self.isEnd:\n",
    "                # Player 1\n",
    "                positions = self.availablePositions()\n",
    "                p1_action = self.p1.chooseAction(positions, self.board, self.playerSymbol)\n",
    "                # take action and upate board state\n",
    "                self.updateState(p1_action)\n",
    "                board_hash = self.getHash()\n",
    "                self.p1.addState(board_hash)\n",
    "                # check board status if it is end\n",
    "\n",
    "                win = self.winner()\n",
    "                if win is not None:\n",
    "                    # self.showBoard()\n",
    "                    # ended with p1 either win or draw\n",
    "                    self.giveReward()\n",
    "                    self.p1.reset()\n",
    "                    self.p2.reset()\n",
    "                    self.reset()\n",
    "                    break\n",
    "\n",
    "                else:\n",
    "                    # Player 2\n",
    "                    positions = self.availablePositions()\n",
    "                    p2_action = self.p2.chooseAction(positions, self.board, self.playerSymbol)\n",
    "                    self.updateState(p2_action)\n",
    "                    board_hash = self.getHash()\n",
    "                    self.p2.addState(board_hash)\n",
    "\n",
    "                    win = self.winner()\n",
    "                    if win is not None:\n",
    "                        # self.showBoard()\n",
    "                        # ended with p2 either win or draw\n",
    "                        self.giveReward()\n",
    "                        self.p1.reset()\n",
    "                        self.p2.reset()\n",
    "                        self.reset()\n",
    "                        break\n",
    "\n",
    "    # play with human\n",
    "    def play2(self):\n",
    "        while not self.isEnd:\n",
    "            # Player 1\n",
    "            positions = self.availablePositions()\n",
    "            p1_action = self.p1.chooseAction(positions, self.board, self.playerSymbol)\n",
    "            # take action and upate board state\n",
    "            self.updateState(p1_action)\n",
    "            self.showBoard()\n",
    "            # check board status if it is end\n",
    "            win = self.winner()\n",
    "            if win is not None:\n",
    "                if win == 1:\n",
    "                    print(self.p1.name, \"wins!\")\n",
    "                else:\n",
    "                    print(\"tie!\")\n",
    "                self.reset()\n",
    "                break\n",
    "\n",
    "            else:\n",
    "                # Player 2\n",
    "                positions = self.availablePositions()\n",
    "                p2_action = self.p2.chooseAction(positions)\n",
    "\n",
    "                self.updateState(p2_action)\n",
    "                self.showBoard()\n",
    "                win = self.winner()\n",
    "                if win is not None:\n",
    "                    if win == -1:\n",
    "                        print(self.p2.name, \"wins!\")\n",
    "                    else:\n",
    "                        print(\"tie!\")\n",
    "                    self.reset()\n",
    "                    break\n",
    "\n",
    "    def showBoard(self):\n",
    "        # p1: x  p2: o\n",
    "        for i in range(0, BOARD_ROWS):\n",
    "            print('-------------')\n",
    "            out = '| '\n",
    "            for j in range(0, BOARD_COLS):\n",
    "                if self.board[i, j] == 1:\n",
    "                    token = 'x'\n",
    "                if self.board[i, j] == -1:\n",
    "                    token = 'o'\n",
    "                if self.board[i, j] == 0:\n",
    "                    token = ' '\n",
    "                out += token + ' | '\n",
    "            print(out)\n",
    "        print('-------------')\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "id": "91fda4cc",
   "metadata": {},
   "outputs": [],
   "source": [
    "class Player:\n",
    "    def __init__(self, name, exp_rate=0.3):\n",
    "        self.name = name\n",
    "        self.states = []  # record all positions taken\n",
    "        self.lr = 0.2\n",
    "        self.exp_rate = exp_rate\n",
    "        self.decay_gamma = 0.9\n",
    "        self.states_value = {}  # state -> value\n",
    "\n",
    "    def getHash(self, board):\n",
    "        boardHash = str(board.reshape(BOARD_COLS * BOARD_ROWS))\n",
    "        return boardHash\n",
    "\n",
    "    def chooseAction(self, positions, current_board, symbol):\n",
    "        if np.random.uniform(0, 1) <= self.exp_rate:\n",
    "            # take random action\n",
    "            idx = np.random.choice(len(positions))\n",
    "            action = positions[idx]\n",
    "        else:\n",
    "            value_max = -999\n",
    "            for p in positions:\n",
    "                next_board = current_board.copy()\n",
    "                next_board[p] = symbol\n",
    "                next_boardHash = self.getHash(next_board)\n",
    "                value = 0 if self.states_value.get(next_boardHash) is None else self.states_value.get(next_boardHash)\n",
    "                # print(\"value\", value)\n",
    "                if value >= value_max:\n",
    "                    value_max = value\n",
    "                    action = p\n",
    "        # print(\"{} takes action {}\".format(self.name, action))\n",
    "        return action\n",
    "\n",
    "    # append a hash state\n",
    "    def addState(self, state):\n",
    "        self.states.append(state)\n",
    "\n",
    "    # at the end of game, backpropagate and update states value\n",
    "    def feedReward(self, reward):\n",
    "        for st in reversed(self.states):\n",
    "            if self.states_value.get(st) is None:\n",
    "                self.states_value[st] = 0\n",
    "            self.states_value[st] += self.lr * (self.decay_gamma * reward - self.states_value[st])\n",
    "            reward = self.states_value[st]\n",
    "\n",
    "    def reset(self):\n",
    "        self.states = []\n",
    "\n",
    "    def savePolicy(self):\n",
    "        fw = open('policy_' + str(self.name), 'wb')\n",
    "        pickle.dump(self.states_value, fw)\n",
    "        fw.close()\n",
    "\n",
    "    def loadPolicy(self, file):\n",
    "        fr = open(file, 'rb')\n",
    "        self.states_value = pickle.load(fr)\n",
    "        fr.close()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "a11e8084",
   "metadata": {},
   "outputs": [],
   "source": [
    "class HumanPlayer:\n",
    "    def __init__(self, name):\n",
    "        self.name = name\n",
    "\n",
    "    def chooseAction(self, positions):\n",
    "        while True:\n",
    "            row = int(input(\"Input your action row:\"))\n",
    "            col = int(input(\"Input your action col:\"))\n",
    "            action = (row, col)\n",
    "            if action in positions:\n",
    "                return action\n",
    "\n",
    "    # append a hash state\n",
    "    def addState(self, state):\n",
    "        pass\n",
    "\n",
    "    # at the end of game, backpropagate and update states value\n",
    "    def feedReward(self, reward):\n",
    "        pass\n",
    "\n",
    "    def reset(self):\n",
    "        pass"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "id": "da62ab5e",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "training...\n",
      "False\n",
      "Rounds 0\n"
     ]
    }
   ],
   "source": [
    "# training\n",
    "p1 = Player(\"p1\")\n",
    "p2 = Player(\"p2\")\n",
    "\n",
    "st = State(p1, p2)\n",
    "print(\"training...\")\n",
    "st.play(1000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "id": "ff75205a",
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-------------\n",
      "|   |   |   | \n",
      "-------------\n",
      "|   |   |   | \n",
      "-------------\n",
      "| x |   |   | \n",
      "-------------\n",
      "Input your action row:1\n",
      "Input your action col:1\n",
      "-------------\n",
      "|   |   |   | \n",
      "-------------\n",
      "|   | o |   | \n",
      "-------------\n",
      "| x |   |   | \n",
      "-------------\n",
      "-------------\n",
      "|   |   |   | \n",
      "-------------\n",
      "|   | o |   | \n",
      "-------------\n",
      "| x |   | x | \n",
      "-------------\n",
      "Input your action row:2\n",
      "Input your action col:1\n",
      "-------------\n",
      "|   |   |   | \n",
      "-------------\n",
      "|   | o |   | \n",
      "-------------\n",
      "| x | o | x | \n",
      "-------------\n",
      "-------------\n",
      "|   |   | x | \n",
      "-------------\n",
      "|   | o |   | \n",
      "-------------\n",
      "| x | o | x | \n",
      "-------------\n",
      "Input your action row:0\n",
      "Input your action col:0\n",
      "-------------\n",
      "| o |   | x | \n",
      "-------------\n",
      "|   | o |   | \n",
      "-------------\n",
      "| x | o | x | \n",
      "-------------\n",
      "-------------\n",
      "| o |   | x | \n",
      "-------------\n",
      "|   | o | x | \n",
      "-------------\n",
      "| x | o | x | \n",
      "-------------\n",
      "p1 wins!\n"
     ]
    }
   ],
   "source": [
    "p2 = HumanPlayer(\"human\")\n",
    "\n",
    "st = State(p1, p2)\n",
    "st.play2()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "31711426",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}