Unified arena solver for arenas

CHANGELOG.md

@@ -1,3 +1,7 @@
+# `2.4b3`
+
+* **Unified arena solver for arenas**
+
 # `2.4b2`
 
 * Fix critical bug in `bestmobabot.dataclasses_.ShopSlot`

bestmobabot/bot.py

@@ -13,12 +13,13 @@
 
 import requests
 
-from bestmobabot import arena, constants
+from bestmobabot import constants
 from bestmobabot.api import API, AlreadyError, InvalidResponseError, NotEnoughError, NotFoundError, OutOfRetargetDelta
+from bestmobabot.arena import ArenaSolver, reduce_grand_arena, reduce_normal_arena
 from bestmobabot.database import Database
 from bestmobabot.dataclasses_ import Hero, Mission, Quests, Replay, User
 from bestmobabot.enums import BattleType
-from bestmobabot.logging_ import log_heroes, log_rewards, logger
+from bestmobabot.logging_ import log_rewards, logger
 from bestmobabot.model import Model
 from bestmobabot.resources import get_heroic_mission_ids, mission_name, shop_name
 from bestmobabot.settings import Settings
@@ -128,7 +129,7 @@ def start(self):
             Task(next_run_at=Task.at(hour=21, minute=30, tz=self.user.tz), execute=self.farm_quests),
 
             # Recurring tasks.
-            Task(next_run_at=Task.every_n_minutes(24 * 60 // 5, self.settings.bot.arena.schedule_offset), execute=self.attack_arena),  # noqa
+            Task(next_run_at=Task.every_n_minutes(24 * 60 // 5, self.settings.bot.arena.schedule_offset), execute=self.attack_normal_arena),  # noqa
             Task(next_run_at=Task.every_n_minutes(24 * 60 // 5, self.settings.bot.arena.schedule_offset), execute=self.attack_grand_arena),  # noqa
             Task(next_run_at=Task.every_n_hours(6), execute=self.farm_mail),
             Task(next_run_at=Task.every_n_hours(6), execute=self.check_freebie),
@@ -332,37 +333,42 @@ def train_arena_model(self):
             n_last_battles=self.settings.bot.arena.last_battles,
         ).train()
 
-    def attack_arena(self):
+    # FIXME: refactor.
+    def attack_normal_arena(self):
         """
         Совершает бой на арене.
         """
-        logger.info('Attacking arena…')
+        logger.info('Attacking normal arena…')
         model, heroes = self.check_arena(constants.TEAM_SIZE)
 
         # Pick an enemy and select attackers.
-        enemy, attackers, probability = arena.Arena(
+        solution = ArenaSolver(
             model=model,
             user_clan_id=self.user.clan_id,
             heroes=heroes,
-            get_enemies_page=self.api.find_arena_enemies,
-            settings=self.settings,
-        ).select_enemy()
-
-        # Debugging.
-        logger.info(f'Enemy name: "{enemy.user.name}".')
-        logger.info(f'Enemy place: {enemy.place}.')
-        logger.info(f'Probability: {100.0 * probability:.1f}%.')
-        log_heroes('Attackers:', attackers)
-        log_heroes(f'Defenders:', enemy.heroes)
+            max_iterations=self.settings.bot.arena.normal_max_pages,
+            n_keep_solutions=self.settings.bot.arena.normal_keep_solutions,
+            n_generate_solutions=self.settings.bot.arena.normal_generate_solutions,
+            n_generations_count_down=self.settings.bot.arena.normal_generations_count_down,
+            early_stop=self.settings.bot.arena.early_stop,
+            get_enemies=self.api.find_arena_enemies,
+            friendly_clans=self.settings.bot.arena.friendly_clans,
+            reduce_probabilities=reduce_normal_arena,
+        ).solve()
+        solution.log()
 
         # Attack!
-        result, quests = self.api.attack_arena(enemy.user.id, self.get_hero_ids(attackers))
+        result, quests = self.api.attack_arena(
+            solution.enemy.user_id,
+            self.get_hero_ids(solution.attackers[0]),
+        )
 
         # Collect results.
         result.log()
         logger.info('Current place: {}.', result.state.arena_place)
         self.farm_quests(quests)
 
+    # FIXME: refactor.
     def attack_grand_arena(self):
         """
         Совершает бой на гранд арене.
@@ -371,27 +377,25 @@ def attack_grand_arena(self):
         model, heroes = self.check_arena(constants.GRAND_SIZE)
 
         # Pick an enemy and select attackers.
-        enemy, attacker_teams, probability = arena.GrandArena(
+        solution = ArenaSolver(
             model=model,
             user_clan_id=self.user.clan_id,
             heroes=heroes,
-            get_enemies_page=self.api.find_grand_enemies,
-            settings=self.settings,
-        ).select_enemy()
-
-        # Debugging.
-        logger.info(f'Enemy name: "{enemy.user.name}".')
-        logger.info(f'Enemy place: {enemy.place}.')
-        logger.info(f'Probability: {100.0 * probability:.1f}%.')
-        for i, (attackers, defenders) in enumerate(zip(attacker_teams, enemy.heroes), start=1):
-            logger.info(f'Battle #{i}.')
-            log_heroes('Attackers:', attackers)
-            log_heroes('Defenders:', defenders)
+            max_iterations=self.settings.bot.arena.grand_max_pages,
+            n_keep_solutions=self.settings.bot.arena.grand_keep_solutions,
+            n_generate_solutions=self.settings.bot.arena.grand_generate_solutions,
+            n_generations_count_down=self.settings.bot.arena.grand_generations_count_down,
+            early_stop=self.settings.bot.arena.early_stop,
+            get_enemies=self.api.find_grand_enemies,
+            friendly_clans=self.settings.bot.arena.friendly_clans,
+            reduce_probabilities=reduce_grand_arena,
+        ).solve()
+        solution.log()
 
         # Attack!
-        result, quests = self.api.attack_grand(enemy.user.id, [
-            [attacker.id for attacker in attackers]
-            for attackers in attacker_teams
+        result, quests = self.api.attack_grand(solution.enemy.user_id, [
+            self.get_hero_ids(attackers)
+            for attackers in solution.attackers
         ])
 
         # Collect results.

bestmobabot/dataclasses_.py

@@ -128,7 +128,8 @@ def __lt__(self, other: Any) -> Any:
         return NotImplemented
 
     def __str__(self):
-        return f'{"⭐" * self.star} {resources.hero_name(self.id)} ({self.level}) {COLORS.get(self.color, self.color)}'
+        stars = '🌟' if self.star > 5 else '⭐' * self.star
+        return f'{stars} {resources.hero_name(self.id)} ({self.level}) {COLORS.get(self.color, self.color)}'
 
 
 Team = List[Hero]
@@ -177,6 +178,9 @@ class Config:
     def is_from_clans(self, clans: Iterable[str]) -> bool:
         return (self.clan_id and self.clan_id in clans) or (self.clan_title and self.clan_title in clans)
 
+    def __str__(self) -> str:
+        return f'«{self.name}» from «{self.clan_title}»'
+
 
 class BaseArenaEnemy(BaseModel):
     user_id: str
@@ -193,6 +197,9 @@ class Config:
     def teams(self) -> List[Team]:
         raise NotImplementedError()
 
+    def __str__(self) -> str:
+        return f'{self.user} at place {self.place}'
+
 
 class ArenaEnemy(BaseArenaEnemy):
     heroes: List[Hero]

bestmobabot/itertools_.py

@@ -1,15 +1,26 @@
+"""
+More `itertools`.
+"""
+
 from __future__ import annotations
 
-from typing import Iterable, Iterator, TypeVar
+import math
+from typing import Any, Iterable, Iterator, List, TypeVar
+
+from loguru import logger
 
 T = TypeVar('T')
 
 
-class CoolDown(Iterator[T]):
-    def __init__(self, iterable: Iterable[T], interval: int):
+class CountDown(Iterator[T]):
+    """
+    Limits iteration number unless `reset` is called.
+    """
+
+    def __init__(self, iterable: Iterable[T], n_iterations: int):
         self.iterator = iter(iterable)
-        self.interval = interval
-        self.iterations_left = interval
+        self.n_iterations = n_iterations
+        self.iterations_left = n_iterations
         self.is_fresh = True
 
     def __iter__(self) -> Iterator[T]:
@@ -22,6 +33,46 @@ def __next__(self) -> T:
         self.is_fresh = False
         return next(self.iterator)
 
+    def __int__(self) -> int:
+        return self.iterations_left
+
     def reset(self):
-        self.iterations_left = self.interval
+        self.iterations_left = self.n_iterations
         self.is_fresh = True
+
+
+def secretary_max(items: Iterable[T], n: int, early_stop: Any = None) -> T:
+    """
+    Select best item while lazily iterating over the items.
+    https://en.wikipedia.org/wiki/Secretary_problem#Deriving_the_optimal_policy
+    """
+    r = int(n / math.e) + 1
+
+    max_item = None
+
+    for i, item in enumerate(items, start=1):
+        # Check early stop condition.
+        if early_stop is not None and item >= early_stop:
+            logger.trace('Early stop.')
+            return item
+        # If it's the last item, just return it.
+        if i == n:
+            logger.trace('Last item.')
+            return item
+        # Otherwise, check if the item is better than previous ones.
+        if max_item is None or item >= max_item:
+            if i >= r:
+                # Better than (r - 1) previous ones, return it.
+                logger.trace('Better than {} previous ones.', r - 1)
+                return item
+            # Otherwise, update the best key.
+            max_item = item
+
+    raise RuntimeError('unreachable code')
+
+
+def slices(n: int, length: int) -> List[slice]:
+    """
+    Make a list of continuous slices. E.g. `[slice(0, 2), slice(2, 4), slice(4, 6)]`.
+    """
+    return [slice(i * length, (i + 1) * length) for i in range(n)]

bestmobabot/settings.py

@@ -20,17 +20,24 @@ class TelegramSettings(BaseModel):
 
 
 class ArenaSettings(BaseModel):
-    skip_clans: Set[str] = []  # names or clan IDs which must be skipped during enemy search
-    early_stop: confloat(ge=0.0, le=1.0) = 0.95  # minimal win probability to stop enemy search
+    # Shared settings.
     schedule_offset: timedelta = timedelta()  # arena task schedule offset
-    teams_limit: conint(ge=1) = 20000  # number of the most powerful teams tested
-    grand_generations_cool_down: conint(ge=1) = 25  # maximum number of GA iterations without any improvement
-    max_pages: conint(ge=1) = 15  # maximal number of pages during enemy search
-    max_grand_pages: conint(ge=1) = 15  # maximal number of pages during grand enemy search
-    randomize_grand_defenders: bool = False
+    friendly_clans: Set[str] = []  # names or clan IDs which must be skipped during enemy search
+    early_stop: confloat(ge=0.0, le=1.0) = 0.95  # minimal win probability to stop enemy search
+    last_battles: conint(ge=1) = constants.MODEL_N_LAST_BATTLES  # use last N battles for training
+
+    # Normal arena.
+    normal_max_pages: conint(ge=1) = 15  # maximal number of pages during normal enemy search
+    normal_generations_count_down: conint(ge=1) = 5
+    normal_generate_solutions: conint(ge=1) = 750
+    normal_keep_solutions: conint(ge=1) = 250
+
+    # Grand arena.
+    grand_max_pages: conint(ge=1) = 15  # maximal number of pages during grand enemy search
+    grand_generations_count_down: conint(ge=1) = 25  # maximum number of GA iterations without any improvement
     grand_generate_solutions: conint(ge=1) = 1250
     grand_keep_solutions: conint(ge=1) = 250
-    last_battles: conint(ge=1) = constants.MODEL_N_LAST_BATTLES  # use last N battles for training
+    randomize_grand_defenders: bool = False
 
 
 # noinspection PyMethodParameters

setup.cfg

@@ -3,6 +3,7 @@ line_length = 120
 multi_line_output = 3
 include_trailing_comma = true
 known_third_party = ujson
+known_standard_library = dataclasses
 
 [flake8]
 max_line_length = 120

setup.py

@@ -4,7 +4,7 @@
 
 setuptools.setup(
     name='bestmobabot',
-    version='2.4b2',
+    version='2.4b3',
     author='Pavel Perestoronin',
     author_email='eigenein@gmail.com',
     description='Hero Wars game bot 🏆',

tests/test_arena.py → tests/test_itertools.py

@@ -1,8 +1,18 @@
+from __future__ import annotations
+
 from typing import List
 
 import pytest
 
-from bestmobabot.arena import ranges, secretary_max
+from bestmobabot.itertools_ import secretary_max, slices
+
+
+@pytest.mark.parametrize('n, length, expected', [
+    (1, 5, [slice(0, 5)]),
+    (3, 5, [slice(0, 5), slice(5, 10), slice(10, 15)]),
+])
+def test_slices(n: int, length: int, expected: List[slice]):
+    assert slices(n, length) == expected
 
 
 @pytest.mark.parametrize('items, expected, next_', [
@@ -44,10 +54,3 @@ def test_secretary_max_early_stop(items, early_stop, expected, next_):
     iterator = iter(items)
     assert secretary_max(iterator, len(items), early_stop=early_stop) == expected
     assert next(iterator, None) == next_  # test the iterator position
-
-
-@pytest.mark.parametrize('n_ranges, range_size, expected', [
-    (3, 5, [range(0, 5), range(5, 10), range(10, 15)]),
-])
-def test_ranges(n_ranges: int, range_size: int, expected: List[range]):
-    assert list(ranges(n_ranges, range_size)) == expected