Q_Learning_FA.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bd611aca",
   "metadata": {},
   "outputs": [],
   "source": [
    "# imports\n",
    "\n",
    "import asyncio\n",
    "import json\n",
    "import os\n",
    "import matplotlib\n",
    "import neptune.new as neptune\n",
    "import nest_asyncio\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import time\n",
    "\n",
    "from collections import defaultdict\n",
    "from datetime import date\n",
    "from itertools import product\n",
    "from matplotlib import pyplot\n",
    "from poke_env.environment.abstract_battle import AbstractBattle\n",
    "from poke_env.player.battle_order import ForfeitBattleOrder\n",
    "from poke_env.player.player import Player\n",
    "# from poke_env.player.random_player import RandomPlayer\n",
    "from scipy.interpolate import griddata\n",
    "from src.PlayerQLearning import Player as PlayerQLearning\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f537cc4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# global configs\n",
    "\n",
    "debug = True\n",
    "save_to_json_file = True\n",
    "use_validation = False\n",
    "use_neptune = False\n",
    "\n",
    "nest_asyncio.apply()\n",
    "np.random.seed(0)\n",
    "\n",
    "if use_neptune:\n",
    "    run = neptune.init(project='leolellisr/rl-pokeenv',\n",
    "                       api_token='eyJhcGlfYWRkcmVzcyI6Imh0dHBzOi8vYXBwLm5lcHR1bmUuYWkiLCJhcGlfdXJsIjoiaHR0cHM6Ly9hcHAubmVwdHVuZS5haSIsImFwaV9rZXkiOiI1NjY1YmJkZi1hYmM5LTQ3M2QtOGU1ZC1iZTFlNWY4NjE1NDQifQ==',\n",
    "                       tags=[\"Henrique\", \"Q-learning FA\"])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "093e462f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# our team\n",
    "\n",
    "OUR_TEAM = \"\"\"\n",
    "Pikachu-Original (M) @ Light Ball  \n",
    "Ability: Static  \n",
    "EVs: 252 Atk / 4 SpD / 252 Spe  \n",
    "Jolly Nature  \n",
    "- Volt Tackle  \n",
    "- Nuzzle  \n",
    "- Iron Tail  \n",
    "- Knock Off  \n",
    "\n",
    "Charizard @ Life Orb  \n",
    "Ability: Solar Power  \n",
    "EVs: 252 SpA / 4 SpD / 252 Spe  \n",
    "Timid Nature  \n",
    "IVs: 0 Atk  \n",
    "- Flamethrower  \n",
    "- Dragon Pulse  \n",
    "- Roost  \n",
    "- Sunny Day  \n",
    "\n",
    "Blastoise @ White Herb  \n",
    "Ability: Torrent  \n",
    "EVs: 4 Atk / 252 SpA / 252 Spe  \n",
    "Mild Nature  \n",
    "- Scald  \n",
    "- Ice Beam  \n",
    "- Earthquake  \n",
    "- Shell Smash  \n",
    "\n",
    "Venusaur @ Black Sludge  \n",
    "Ability: Chlorophyll  \n",
    "EVs: 252 SpA / 4 SpD / 252 Spe  \n",
    "Modest Nature  \n",
    "IVs: 0 Atk  \n",
    "- Giga Drain  \n",
    "- Sludge Bomb  \n",
    "- Sleep Powder  \n",
    "- Leech Seed  \n",
    "\n",
    "Sirfetch’d @ Aguav Berry  \n",
    "Ability: Steadfast  \n",
    "EVs: 248 HP / 252 Atk / 8 SpD  \n",
    "Adamant Nature  \n",
    "- Close Combat  \n",
    "- Swords Dance  \n",
    "- Poison Jab  \n",
    "- Knock Off  \n",
    "\n",
    "Tauros (M) @ Assault Vest  \n",
    "Ability: Intimidate  \n",
    "EVs: 252 Atk / 4 SpD / 252 Spe  \n",
    "Jolly Nature  \n",
    "- Double-Edge  \n",
    "- Earthquake  \n",
    "- Megahorn  \n",
    "- Iron Head  \n",
    "\"\"\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "12635b18",
   "metadata": {},
   "outputs": [],
   "source": [
    "# opponent's team\n",
    "\n",
    "OP_TEAM = \"\"\"\n",
    "Eevee @ Eviolite  \n",
    "Ability: Adaptability  \n",
    "EVs: 252 HP / 252 Atk / 4 SpD  \n",
    "Adamant Nature  \n",
    "- Quick Attack  \n",
    "- Flail  \n",
    "- Facade  \n",
    "- Wish  \n",
    "\n",
    "Vaporeon @ Leftovers  \n",
    "Ability: Hydration  \n",
    "EVs: 252 HP / 252 Def / 4 SpA  \n",
    "Bold Nature  \n",
    "IVs: 0 Atk  \n",
    "- Scald  \n",
    "- Shadow Ball  \n",
    "- Toxic  \n",
    "- Wish  \n",
    "\n",
    "Sylveon @ Aguav Berry  \n",
    "Ability: Pixilate  \n",
    "EVs: 252 HP / 252 SpA / 4 SpD  \n",
    "Modest Nature  \n",
    "IVs: 0 Atk  \n",
    "- Hyper Voice  \n",
    "- Mystical Fire  \n",
    "- Psyshock  \n",
    "- Calm Mind  \n",
    "\n",
    "Jolteon @ Assault Vest  \n",
    "Ability: Quick Feet  \n",
    "EVs: 252 SpA / 4 SpD / 252 Spe  \n",
    "Timid Nature  \n",
    "IVs: 0 Atk  \n",
    "- Thunderbolt  \n",
    "- Hyper Voice  \n",
    "- Volt Switch  \n",
    "- Shadow Ball  \n",
    "\n",
    "Leafeon @ Life Orb  \n",
    "Ability: Chlorophyll  \n",
    "EVs: 252 Atk / 4 SpD / 252 Spe  \n",
    "Adamant Nature  \n",
    "- Leaf Blade  \n",
    "- Knock Off  \n",
    "- X-Scissor  \n",
    "- Swords Dance  \n",
    "\n",
    "Umbreon @ Iapapa Berry  \n",
    "Ability: Inner Focus  \n",
    "EVs: 252 HP / 4 Atk / 252 SpD  \n",
    "Careful Nature  \n",
    "- Foul Play  \n",
    "- Body Slam  \n",
    "- Toxic  \n",
    "- Wish  \n",
    "\"\"\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f27f14a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "N_STATE_COMPONENTS = 12\n",
    "# num of features = num of state components + action\n",
    "N_FEATURES = N_STATE_COMPONENTS + 1\n",
    "\n",
    "N_OUR_MOVE_ACTIONS = 4\n",
    "N_OUR_SWITCH_ACTIONS = 5\n",
    "N_OUR_ACTIONS = N_OUR_MOVE_ACTIONS + N_OUR_SWITCH_ACTIONS\n",
    "\n",
    "ALL_OUR_ACTIONS = np.array(range(0, N_OUR_ACTIONS))\n",
    "\n",
    "NAME_TO_ID_DICT = {\n",
    "    \"pikachuoriginal\": 0,\n",
    "    \"charizard\": 1,\n",
    "    \"blastoise\": 2,\n",
    "    \"venusaur\": 3,\n",
    "    \"sirfetchd\": 4,\n",
    "    \"tauros\": 5,\n",
    "    \"eevee\": 6,\n",
    "    \"vaporeon\": 7,\n",
    "    \"sylveon\": 8,\n",
    "    \"jolteon\": 9,\n",
    "    \"leafeon\": 10,\n",
    "    \"umbreon\": 11\n",
    "}\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "22dada00",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Max-damage player\n",
    "\n",
    "class MaxDamagePlayer(Player):\n",
    "    def choose_move(self, battle):\n",
    "        if battle.available_moves:\n",
    "            best_move = max(battle.available_moves, key=lambda move: move.base_power)\n",
    "            return self.create_order(best_move)\n",
    "        else:\n",
    "            return self.choose_random_move(battle)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fc1b7c73",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Q-learning FA player\n",
    "\n",
    "class QLearningFAPlayer(PlayerQLearning):\n",
    "    def __init__(self, battle_format, team, n0, alpha0, gamma):\n",
    "        super().__init__(battle_format=battle_format, team=team)\n",
    "        self.N = defaultdict(lambda: np.zeros(N_OUR_ACTIONS))\n",
    "        self.w = np.random.rand(N_FEATURES)\n",
    "        self.n0 = n0\n",
    "        self.alpha0 = alpha0\n",
    "        self.gamma = gamma\n",
    "        self.state = None\n",
    "        self.action = None\n",
    "\n",
    "    def choose_move(self, battle):\n",
    "        if self.state is not None:\n",
    "            # observe R, S'\n",
    "            reward = self.compute_reward(battle)\n",
    "            next_state = self.embed_battle(battle)\n",
    "            # Q-learning\n",
    "            self.N[str(self.state)][self.action] += 1\n",
    "            alpha = self.alpha0 / self.N[str(self.state)][self.action]\n",
    "            delta = \\\n",
    "                reward + self.gamma * self.max_q_approx(next_state, self.w) - self.q_approx(self.state, self.action, self.w)\n",
    "            self.w += alpha * delta * self.x(self.state, self.action)\n",
    "            # S <- S'\n",
    "            self.state = next_state\n",
    "        else:\n",
    "            # S first initialization\n",
    "            self.state = self.embed_battle(battle)\n",
    "\n",
    "        # Choose A from S using epsilon-greedy policy\n",
    "        self.action = self.pi(self.state, self.w)\n",
    "\n",
    "        # if the selected action is not possible, perform a random move instead\n",
    "        if self.action == -1:\n",
    "            return ForfeitBattleOrder()\n",
    "        elif self.action < 4 and self.action < len(battle.available_moves) and not battle.force_switch:\n",
    "            return self.create_order(battle.available_moves[self.action])\n",
    "        elif 0 <= self.action - 4 < len(battle.available_switches):\n",
    "            return self.create_order(battle.available_switches[self.action - 4])\n",
    "        else:\n",
    "            return self.choose_random_move(battle)\n",
    "\n",
    "    def _battle_finished_callback(self, battle):\n",
    "        if use_neptune:\n",
    "            run[f'N0: {self.n0} gamma: {self.gamma} win_acc'].log(self.n_won_battles / len(self._reward_buffer))\n",
    "\n",
    "    ''' Helper functions '''\n",
    "\n",
    "    # feature vector\n",
    "    @staticmethod\n",
    "    def x(state, action):\n",
    "        state = np.array(state).astype(float)\n",
    "        return np.append(state, action)\n",
    "\n",
    "    # q^(S, A, W)\n",
    "    def q_approx(self, state, action, w):\n",
    "        state = np.array(state).astype(float)\n",
    "        return np.dot(self.x(state, action), w)\n",
    "\n",
    "    # max(a, q^(S, a', W))\n",
    "    def max_q_approx(self, state, w):\n",
    "        state = np.array(state).astype(float)\n",
    "        return max(np.array([self.q_approx(state, action, w) for action in range(N_OUR_ACTIONS)]))\n",
    "\n",
    "    # epsilon-greedy policy\n",
    "    def pi(self, state, w):\n",
    "        epsilon = self.n0 / (self.n0 + np.sum(self.N[str(state)]))\n",
    "        # let's get the greedy action. Ties must be broken arbitrarily\n",
    "        q_approx = np.array([self.q_approx(state, action, w) for action in range(N_OUR_ACTIONS)])\n",
    "        greedy_action = np.random.choice(np.where(q_approx == q_approx.max())[0])\n",
    "        action_pick_probability = np.full(N_OUR_ACTIONS, epsilon / N_OUR_ACTIONS)\n",
    "        action_pick_probability[greedy_action] += 1 - epsilon\n",
    "        return np.random.choice(ALL_OUR_ACTIONS, p=action_pick_probability)\n",
    "\n",
    "    # the embed battle is our state\n",
    "    # 12 factors: our active mon, opponent's active mon, 4 moves base power, 4 moves multipliers, remaining mons\n",
    "    @staticmethod\n",
    "    def embed_battle(battle):\n",
    "        # -1 indicates that the move does not have a base power\n",
    "        # or is not available\n",
    "        moves_base_power = -np.ones(4)\n",
    "        moves_dmg_multiplier = np.ones(4)\n",
    "        for i, move in enumerate(battle.available_moves):\n",
    "            moves_base_power[i] = (\n",
    "                    move.base_power / 100\n",
    "            )  # Simple rescaling to facilitate learning\n",
    "            if move.type:\n",
    "                moves_dmg_multiplier[i] = move.type.damage_multiplier(\n",
    "                    battle.opponent_active_pokemon.type_1,\n",
    "                    battle.opponent_active_pokemon.type_2,\n",
    "                )\n",
    "\n",
    "        # We count how many pokemons have not fainted in each team\n",
    "        n_fainted_mon_team = (\n",
    "            len([mon for mon in battle.team.values() if mon.fainted])\n",
    "        )\n",
    "        n_fainted_mon_opponent = (\n",
    "            len([mon for mon in battle.opponent_team.values() if mon.fainted])\n",
    "        )\n",
    "\n",
    "        state = list()\n",
    "        state.append(NAME_TO_ID_DICT[str(battle.active_pokemon).split(' ')[0]])\n",
    "        state.append(NAME_TO_ID_DICT[str(battle.opponent_active_pokemon).split(' ')[0]])\n",
    "        for move_base_power in moves_base_power:\n",
    "            state.append('{0:.2f}'.format(move_base_power))\n",
    "        for move_dmg_multiplier in moves_dmg_multiplier:\n",
    "            state.append('{0:.2f}'.format(move_dmg_multiplier))\n",
    "        state.append(n_fainted_mon_team)\n",
    "        state.append(n_fainted_mon_opponent)\n",
    "\n",
    "        return state\n",
    "\n",
    "    # Computing rewards\n",
    "    def reward_computing_helper(\n",
    "            self,\n",
    "            battle: AbstractBattle,\n",
    "            *,\n",
    "            fainted_value: float = 0.15,\n",
    "            hp_value: float = 0.15,\n",
    "            number_of_pokemons: int = 6,\n",
    "            starting_value: float = 0.0,\n",
    "            status_value: float = 0.15,\n",
    "            victory_value: float = 1.0\n",
    "    ) -> float:\n",
    "        # 1st compute\n",
    "        if battle not in self._reward_buffer:\n",
    "            self._reward_buffer[battle] = starting_value\n",
    "        current_value = 0\n",
    "\n",
    "        # Verify if pokemon have fainted or have status\n",
    "        for mon in battle.team.values():\n",
    "            current_value += mon.current_hp_fraction * hp_value\n",
    "            if mon.fainted:\n",
    "                current_value -= fainted_value\n",
    "            elif mon.status is not None:\n",
    "                current_value -= status_value\n",
    "\n",
    "        current_value += (number_of_pokemons - len(battle.team)) * hp_value\n",
    "\n",
    "        # Verify if opponent pokemon have fainted or have status\n",
    "        for mon in battle.opponent_team.values():\n",
    "            current_value -= mon.current_hp_fraction * hp_value\n",
    "            if mon.fainted:\n",
    "                current_value += fainted_value\n",
    "            elif mon.status is not None:\n",
    "                current_value += status_value\n",
    "\n",
    "        current_value -= (number_of_pokemons - len(battle.opponent_team)) * hp_value\n",
    "\n",
    "        # Verify if we won or lost\n",
    "        if battle.won:\n",
    "            current_value += victory_value\n",
    "        elif battle.lost:\n",
    "            current_value -= victory_value\n",
    "\n",
    "        # Value to return\n",
    "        to_return = current_value - self._reward_buffer[battle]\n",
    "        self._reward_buffer[battle] = current_value\n",
    "        if use_neptune:\n",
    "            run[f'N0: {self.n0} gamma: {self.gamma} reward_buffer'].log(current_value)\n",
    "        return to_return\n",
    "\n",
    "    # Calling reward_computing_helper\n",
    "    def compute_reward(self, battle) -> float:\n",
    "        return self.reward_computing_helper(battle, fainted_value=2, hp_value=1, victory_value=15)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "92c94ce3",
   "metadata": {},
   "outputs": [],
   "source": [
    "# validation player\n",
    "\n",
    "class ValidationPlayer(PlayerQLearning):\n",
    "    def __init__(self, battle_format, team, w):\n",
    "        super().__init__(battle_format=battle_format, team=team)\n",
    "        self.w = w\n",
    "\n",
    "    def choose_move(self, battle):\n",
    "        state = self.embed_battle(battle)\n",
    "        # let's get the greedy action. Ties must be broken arbitrarily\n",
    "        q_approx = np.array([self.q_approx(state, action, self.w) for action in range(N_OUR_ACTIONS)])\n",
    "        action = np.random.choice(np.where(q_approx == q_approx.max())[0])\n",
    "\n",
    "        # if the selected action is not possible, perform a random move instead\n",
    "        if action == -1:\n",
    "            return ForfeitBattleOrder()\n",
    "        elif action < 4 and action < len(battle.available_moves) and not battle.force_switch:\n",
    "            return self.create_order(battle.available_moves[action])\n",
    "        elif 0 <= action - 4 < len(battle.available_switches):\n",
    "            return self.create_order(battle.available_switches[action - 4])\n",
    "        else:\n",
    "            return self.choose_random_move(battle)\n",
    "\n",
    "    def _battle_finished_callback(self, battle):\n",
    "        pass\n",
    "\n",
    "    ''' Helper functions '''\n",
    "\n",
    "    # feature vector\n",
    "    @staticmethod\n",
    "    def x(state, action):\n",
    "        state = np.array(state).astype(float)\n",
    "        return np.append(state, action)\n",
    "\n",
    "    # q^(S, A, W)\n",
    "    def q_approx(self, state, action, w):\n",
    "        state = np.array(state).astype(float)\n",
    "        return np.dot(self.x(state, action), w)\n",
    "\n",
    "    # the embed battle is our state\n",
    "    # 12 factors: our active mon, opponent's active mon, 4 moves base power, 4 moves multipliers, remaining mons\n",
    "    @staticmethod\n",
    "    def embed_battle(battle):\n",
    "        # -1 indicates that the move does not have a base power\n",
    "        # or is not available\n",
    "        moves_base_power = -np.ones(4)\n",
    "        moves_dmg_multiplier = np.ones(4)\n",
    "        for i, move in enumerate(battle.available_moves):\n",
    "            moves_base_power[i] = (\n",
    "                    move.base_power / 100\n",
    "            )  # Simple rescaling to facilitate learning\n",
    "            if move.type:\n",
    "                moves_dmg_multiplier[i] = move.type.damage_multiplier(\n",
    "                    battle.opponent_active_pokemon.type_1,\n",
    "                    battle.opponent_active_pokemon.type_2,\n",
    "                )\n",
    "\n",
    "        # We count how many pokemons have not fainted in each team\n",
    "        fainted_mon_team = (\n",
    "            len([mon for mon in battle.team.values() if mon.fainted])\n",
    "        )\n",
    "        fainted_mon_opponent = (\n",
    "            len([mon for mon in battle.opponent_team.values() if mon.fainted])\n",
    "        )\n",
    "\n",
    "        state = list()\n",
    "        state.append(NAME_TO_ID_DICT[str(battle.active_pokemon).split(' ')[0]])\n",
    "        state.append(NAME_TO_ID_DICT[str(battle.opponent_active_pokemon).split(' ')[0]])\n",
    "        for move_base_power in moves_base_power:\n",
    "            state.append('{0:.2f}'.format(move_base_power))\n",
    "        for move_dmg_multiplier in moves_dmg_multiplier:\n",
    "            state.append('{0:.2f}'.format(move_dmg_multiplier))\n",
    "        state.append(fainted_mon_team)\n",
    "        state.append(fainted_mon_opponent)\n",
    "\n",
    "        return state\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5e2940bd",
   "metadata": {},
   "outputs": [],
   "source": [
    "# global parameters\n",
    "\n",
    "# possible values for num_battles (number of episodes)\n",
    "n_battles_array = [1000, 10000]\n",
    "# exploration schedule from MC, i. e., epsilon(t) = N0 / (N0 + N(S(t)))\n",
    "n0_array = [0.0001, 0.001, 0.01]\n",
    "# possible values for alpha0 (initial learning rate)\n",
    "alpha0_array = [0.01]\n",
    "# possible values for gamma (discount factor)\n",
    "gamma_array = [0.75]\n",
    "\n",
    "\n",
    "list_of_params = [\n",
    "    {\n",
    "        'n_battles': n_battles,\n",
    "        'n0': n0,\n",
    "        'alpha0': alpha0,\n",
    "        'gamma': gamma\n",
    "    } for n_battles, n0, alpha0, gamma in product(n_battles_array, n0_array, alpha0_array, gamma_array)\n",
    "]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fe309652",
   "metadata": {},
   "outputs": [],
   "source": [
    "# json helper functions\n",
    "\n",
    "def save_array_to_json(path_dir, filename, data):\n",
    "    if not os.path.exists(path_dir):\n",
    "        os.makedirs(path_dir)\n",
    "    full_filename = path_dir + \"/\" + filename\n",
    "    # write\n",
    "    with open(full_filename, \"w\") as file:\n",
    "        json.dump(data if isinstance(data, list) else data.tolist(), file)\n",
    "        file.close()\n",
    "\n",
    "\n",
    "def save_dict_to_json(path_dir, filename, data, append=True):\n",
    "    if not os.path.exists(path_dir):\n",
    "        os.makedirs(path_dir)\n",
    "    full_filename = path_dir + \"/\" + filename\n",
    "    if os.path.exists(full_filename) and append:\n",
    "        with open(full_filename, \"r\") as file:\n",
    "            value_dict = json.load(file)\n",
    "            for key in data:\n",
    "                value_dict[key] = data[key] if isinstance(data[key], list) else data[key].tolist()\n",
    "            file.close()\n",
    "    else:\n",
    "        value_dict = dict()\n",
    "        for key in data:\n",
    "            value_dict[key] = data[key] if isinstance(data[key], list) else data[key].tolist()\n",
    "    # write\n",
    "    with open(full_filename, \"w\") as file:\n",
    "        json.dump(value_dict, file)\n",
    "        file.close()\n",
    "\n",
    "\n",
    "def read_array_from_json(path_dir, filename):\n",
    "    full_filename = path_dir + \"/\" + filename\n",
    "    if not os.path.exists(full_filename):\n",
    "        return None\n",
    "    file = open(full_filename, \"r\")\n",
    "    data = json.load(file)\n",
    "    file.close()\n",
    "    return data\n",
    "\n",
    "\n",
    "def read_dict_from_json(path_dir, filename):\n",
    "    full_filename = path_dir + \"/\" + filename\n",
    "    if not os.path.exists(full_filename):\n",
    "        return None\n",
    "    file = open(full_filename, \"r\")\n",
    "    data = json.load(file)\n",
    "    file.close()\n",
    "    return data\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "42a1336f",
   "metadata": {},
   "outputs": [],
   "source": [
    "# main (let's battle!)\n",
    "\n",
    "# training\n",
    "async def do_battle_training():\n",
    "    for params in list_of_params:\n",
    "        start = time.time()\n",
    "        params['player'] = QLearningFAPlayer(battle_format=\"gen8ou\", team=OUR_TEAM, n0=params['n0'], alpha0=params['alpha0'], gamma=params['gamma'])\n",
    "        params['opponent'] = MaxDamagePlayer(battle_format=\"gen8ou\", team=OP_TEAM)\n",
    "        await params['player'].battle_against(opponent=params['opponent'], n_battles=params['n_battles'])\n",
    "        if debug:\n",
    "            print(\"training: num battles (episodes)=%d, N0=%.4f, alpha0=%.2f, gamma=%.2f, wins=%d, winning %%=%.2f, total time=%s sec\" %\n",
    "                  (\n",
    "                      params['n_battles'],\n",
    "                      round(params['n0'], 4),\n",
    "                      round(params['alpha0'], 2),\n",
    "                      round(params['gamma'], 2),\n",
    "                      params['player'].n_won_battles,\n",
    "                      round((params['player'].n_won_battles / params['n_battles']) * 100, 2),\n",
    "                      round(time.time() - start, 2)\n",
    "                  ))\n",
    "\n",
    "        # save w to json file\n",
    "        if save_to_json_file:\n",
    "            today_s = str(date.today())\n",
    "            n_battle_s = str(params['n_battles'])\n",
    "            n0_s = str(round(params['n0'], 4))\n",
    "            alpha0_s = str(round(params['alpha0'], 2))\n",
    "            gamma_s = str(round(params['gamma'], 2))\n",
    "            winning_percentage_s = str(round((params['player'].n_won_battles / params['n_battles']) * 100, 2))\n",
    "            filename = \"W_\" + today_s + \"_\" + n_battle_s + \"_\" + n0_s + \"_\" + alpha0_s + \"_\" + gamma_s + \"_\" + winning_percentage_s + \".json\"\n",
    "            save_array_to_json(\"./Q_Learning_FA_w\", filename, params['player'].w)\n",
    "\n",
    "        # statistics: key: \"n_battles, n0, alpha0, gamma\", values: list of win or lose\n",
    "        key = str(params['n_battles']) + \"_\" + str(round(params['n0'], 4)) + \"_\" + str(round(params['alpha0'], 2)) + \"_\" + str(round(params['gamma'], 2))\n",
    "        winning_status = list()\n",
    "        for battle in params['player']._battles.values():\n",
    "            if battle.won:\n",
    "                winning_status.append(True)\n",
    "            else:\n",
    "                winning_status.append(False)\n",
    "        # save statistics json file (append)\n",
    "        data = dict()\n",
    "        data[key] = winning_status\n",
    "        save_dict_to_json(\"./Q_Learning_FA_statistics\", \"statistics.json\", data)\n",
    "\n",
    "\n",
    "loop = asyncio.get_event_loop()\n",
    "loop.run_until_complete(loop.create_task(do_battle_training()))\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0e7fcb58",
   "metadata": {},
   "outputs": [],
   "source": [
    "# plotting helper functions\n",
    "\n",
    "def plot_2d(path, title, x_label, x_array, y_label, y_array):\n",
    "    # print(\"plotting %s\" % title)\n",
    "    # set labels and plot surface\n",
    "    figure = matplotlib.pyplot.figure(figsize=(20, 10))\n",
    "    ax = figure.gca()\n",
    "    ax.set_title(title)\n",
    "    ax.set_xlabel(x_label)\n",
    "    ax.set_ylabel(y_label)\n",
    "    ax.plot(x_array, y_array)\n",
    "    # pyplot.show()\n",
    "    if not os.path.exists(path):\n",
    "        os.makedirs(path)\n",
    "    filename = path + \"/\" + title + \"_\" + x_label + \"_\" + y_label + \"_\" + \".pdf\"\n",
    "    figure.savefig(filename, dpi=figure.dpi)\n",
    "    pyplot.close(figure)\n",
    "    # print(\"done!\")\n",
    "\n",
    "\n",
    "def plot_3d(path, title, x_label, x_array, y_label, y_array, z_label, z_array):\n",
    "    # print(\"plotting %s\" % title)\n",
    "    xyz = {'x': x_array, 'y': y_array, 'z': z_array}\n",
    "    df = pd.DataFrame(xyz, index=range(len(xyz['x'])))\n",
    "    xv, yv = np.meshgrid(x_array, y_array)\n",
    "    zv = griddata((df['x'], df['y']), df['z'], (xv, yv), method='nearest')\n",
    "    # set labels and plot surface\n",
    "    figure = matplotlib.pyplot.figure(figsize=(20, 10))\n",
    "    ax = figure.gca(projection='3d')\n",
    "    ax.set_title(title)\n",
    "    ax.set_xlabel(x_label)\n",
    "    ax.set_ylabel(y_label)\n",
    "    ax.set_zlabel(z_label)\n",
    "    surface = ax.plot_surface(xv, yv, zv, rstride=1, cstride=1, cmap=matplotlib.cm.coolwarm, linewidth=0,\n",
    "                              antialiased=False)\n",
    "    figure.colorbar(surface)\n",
    "    # pyplot.show()\n",
    "    if not os.path.exists(path):\n",
    "        os.makedirs(path)\n",
    "    filename = path + \"/\" + title + \".pdf\"\n",
    "    figure.savefig(filename, dpi=figure.dpi)\n",
    "    pyplot.close(figure)\n",
    "    # print(\"done!\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6c97a03b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# plot additional statistics\n",
    "\n",
    "def plot_statistics_json(path_dir, filename=\"statistics.json\"):\n",
    "    # plots from statistics.json\n",
    "    statistics = read_dict_from_json(path_dir, filename)\n",
    "    # win/lost vs. episode number\n",
    "    for key in statistics.keys():\n",
    "        key_elements = key.split(\"_\")\n",
    "        n_battles = key_elements[0]\n",
    "        n0 = key_elements[1]\n",
    "        alpha0 = key_elements[2]\n",
    "        gamma = key_elements[3]\n",
    "        value = statistics[key]\n",
    "        plot_2d(path=\"./Q_Learning_FA_plot\",\n",
    "                title=\"acc_victories_n_battles_\" + n_battles + \"_N0_\" + n0 + \"_alpha0_\" + alpha0 + \"_gamma_\" + gamma,\n",
    "                x_label=\"episodes\",\n",
    "                x_array=np.array(range(0, len(value))),\n",
    "                y_label=\"acc victory\",\n",
    "                y_array=np.cumsum(value).astype(np.int))\n",
    "\n",
    "    # winning % by set of parameters\n",
    "    n_battles = \"\"\n",
    "    alpha0 = \"\"\n",
    "    x_values = []\n",
    "    y_values = []\n",
    "    z_values = []\n",
    "    for key in statistics.keys():\n",
    "        key_elements = key.split(\"_\")\n",
    "        n_battles = key_elements[0]\n",
    "        n0 = key_elements[1]\n",
    "        alpha0 = key_elements[2]\n",
    "        gamma = key_elements[3]\n",
    "        value = statistics[key]\n",
    "        x_values.append(n0)\n",
    "        y_values.append(gamma)\n",
    "        z_values.append(value.count(True) / len(value))\n",
    "    plot_3d(path=\"./Q_Learning_FA_plot\",\n",
    "            title=\"winning_percentage_n_battles_\" + n_battles + \"_alpha0_\" + alpha0,\n",
    "            x_label=\"N0\",\n",
    "            x_array=np.array(x_values).astype(np.float),\n",
    "            y_label=\"gamma\",\n",
    "            y_array=np.array(y_values).astype(np.float),\n",
    "            z_label=\"winning %\",\n",
    "            z_array=np.array(z_values))\n",
    "\n",
    "\n",
    "# plots from statistics\n",
    "plot_statistics_json(\"./Q_Learning_FA_statistics\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "335a07d4",
   "metadata": {},
   "outputs": [],
   "source": [
    "# validation\n",
    "\n",
    "async def do_battle_validation(path_dir):\n",
    "    # read from json\n",
    "    for filename in os.listdir(path_dir):\n",
    "        # learned feature vector\n",
    "        w = np.array(read_array_from_json(path_dir, filename))\n",
    "        # params: n_battles, n0, gamma\n",
    "        params = filename.split(\"_\")\n",
    "        n_battles = int(params[2])\n",
    "        n0 = float(params[3])\n",
    "        alpha0 = float(params[4])\n",
    "        gamma = float(params[5])\n",
    "\n",
    "        # validation (play 1/3 of the battles using Q-learned table)\n",
    "        start = time.time()\n",
    "        validation_player = ValidationPlayer(battle_format=\"gen8ou\", team=OUR_TEAM, w=w)\n",
    "        opponent = MaxDamagePlayer(battle_format=\"gen8ou\", team=OP_TEAM)\n",
    "        n_battles_validation = int(n_battles / 3)\n",
    "        await validation_player.battle_against(opponent=opponent, n_battles=n_battles_validation)\n",
    "        print(\"validation: num battles (episodes)=%d, N0=%.4f, alpha0=%.2f, gamma=%.2f, wins=%d, winning %%=%.2f, total time=%s sec\" %\n",
    "              (\n",
    "                  n_battles_validation,\n",
    "                  n0,\n",
    "                  alpha0,\n",
    "                  gamma,\n",
    "                  validation_player.n_won_battles,\n",
    "                  round((validation_player.n_won_battles / n_battles_validation) * 100, 2),\n",
    "                  round(time.time() - start, 2)\n",
    "              ))\n",
    "\n",
    "\n",
    "if use_validation:\n",
    "    loop = asyncio.get_event_loop()\n",
    "    loop.run_until_complete(loop.create_task(do_battle_validation(\"./Q_Learning_FA_w\")))\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "poke_env",
   "language": "python",
   "name": "poke_env"
  },
  "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.6.13"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}