optimise global fallbcak score

src/warchron/controller/presenter.py

@@ -30,6 +30,7 @@ class Presenter:
         campaign: Campaign | None = None,
         round: Round | None = None,
     ) -> TieDialogData:
+        # TODO display Nth place
         if ctx.context_type == ContextType.WAR:
             if ctx.objective_id:
                 obj = war.objectives[ctx.objective_id]

src/warchron/controller/round_controller.py

@@ -86,6 +86,7 @@ class RoundController:
                     rnd,
                     part.id,
                 )
+                # TODO clarify status icons
                 if alloc.priority != ChoiceStatus.NONE:
                     priority_icon = QIcon(
                         Icons.get_pixmap(IconName[alloc.priority.name])

src/warchron/model/history.py

@@ -1,20 +0,0 @@
-from __future__ import annotations
-from typing import Dict, Tuple
-
-from warchron.model.campaign import Campaign
-
-
-class CampaignHistory:
-
-    @staticmethod
-    def build_match_count(campaign: Campaign) -> Dict[Tuple[str, str], int]:
-        counts: Dict[Tuple[str, str], int] = {}
-        for rnd in campaign.rounds:
-            for battle in rnd.battles.values():
-                p1 = battle.player_1_id
-                p2 = battle.player_2_id
-                if not p1 or not p2:
-                    continue
-                key = (p1, p2) if p1 < p2 else (p2, p1)
-                counts[key] = counts.get(key, 0) + 1
-        return counts

src/warchron/model/pairing.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
-from typing import Dict, List, Tuple
+from typing import Dict, List, Tuple, TYPE_CHECKING
+from collections import defaultdict
 from dataclasses import dataclass
 from uuid import uuid4
-
+from copy import deepcopy
 import random
 
 from warchron.constants import ContextType, ScoreKind, ChoiceStatus, AllocationType
@@ -18,7 +19,9 @@ from warchron.model.scoring import ScoreComputer
 from warchron.model.tiebreaking import TieBreaker, TieContext, ResolveTiesCallback
 from warchron.model.war_event import TieResolved
 from warchron.model.scoring import ParticipantScore
-from warchron.model.history import CampaignHistory
+
+if TYPE_CHECKING:
+    from warchron.model.campaign import Campaign
 
 ScoredBattle = Tuple[int, Battle]
 
@@ -30,6 +33,14 @@ class AllocationResult:
     fallback: bool
 
 
+@dataclass
+class _FallbackState:
+    assignments: List[Tuple[str, str]]  # (pid, battle_id)
+    remaining: List[str]
+    occupancy: Dict[str, List[str]]
+    score: int
+
+
 class Pairing:
 
     @staticmethod
@@ -260,15 +271,58 @@ class Pairing:
             context.participants,
         )
         ordered: List[str] = []
-        for group in ranked_groups:
+        refined_groups = Pairing._refine_with_war_ranking(
+            war,
+            ranked_groups,
+        )
+        for group in refined_groups:
             shuffled_group = list(group)
-            # TODO improve tie break with history parsing
             random.shuffle(shuffled_group)
             ordered.extend(
                 campaign.war_to_campaign_part_id(pid) for pid in shuffled_group
             )
         return ordered[:places]
 
+    @staticmethod
+    def _refine_with_war_ranking(
+        war: War,
+        ranked_groups: List[List[str]],
+    ) -> List[List[str]]:
+        from warchron.model.checking import ResultChecker
+
+        scores = ScoreComputer.compute_scores(
+            war,
+            ContextType.WAR,
+            war.id,
+        )
+
+        def value_getter(score: ParticipantScore) -> int:
+            return score.victory_points
+
+        war_ranking = ResultChecker.get_effective_ranking(
+            war,
+            ContextType.WAR,
+            war.id,
+            ScoreKind.VP,
+            scores,
+            value_getter,
+        )
+        rank_map: Dict[str, int] = {}
+        for rank, group, _ in war_ranking:
+            for pid in group:
+                rank_map[pid] = rank
+        refined: List[List[str]] = []
+        for group in ranked_groups:
+            if len(group) <= 1:
+                refined.append(group)
+                continue
+            buckets: Dict[int, List[str]] = defaultdict(list)
+            for pid in group:
+                buckets[rank_map.get(pid, 10**9)].append(pid)
+            for rank in sorted(buckets.keys()):
+                refined.append(buckets[rank])
+        return refined
+
     @staticmethod
     def _assign_fallback(
         war: War,
@@ -278,37 +332,88 @@ class Pairing:
         campaign = war.get_campaign_by_round(round.id)
         if campaign is None:
             raise DomainError("Campaign not found")
-        match_counts = CampaignHistory.build_match_count(campaign)
+        match_counts = Pairing.build_match_count(campaign)
-        random.shuffle(remaining)
+        available_battles = round.get_battles_with_places()
-        for pid in list(remaining):
+        if not available_battles:
-            available = round.get_battles_with_places()
+            raise DomainError("No available battle remaining")
-            if not available:
+        occupancy = {
-                raise DomainError("No available battle remaining")
+            b.sector_id: [p for p in (b.player_1_id, b.player_2_id) if p is not None]
-            scored: List[ScoredBattle] = [
+            for b in round.battles.values()
-                (
+        }
-                    Pairing._fallback_score(pid, battle, match_counts),
+        shuffled_remaining = list(remaining)
-                    battle,
+        random.shuffle(shuffled_remaining)
-                )
+        initial = _FallbackState(
-                for battle in available
+            assignments=[],
-            ]
+            remaining=shuffled_remaining,
-            scored.sort(key=lambda x: x[0])
+            occupancy=occupancy,
-            best_score = scored[0][0]
+            score=0,
-            best_battles = [b for s, b in scored if s == best_score]
+        )
-            chosen = sorted(best_battles, key=lambda b: b.sector_id)[0]
+        best = Pairing._search_best_fallback(
-            chosen.assign_participant(pid)
+            initial,
+            match_counts,
+            beam_width=10,
+        )
+        battle_map = {b.sector_id: b for b in round.battles.values()}
+        for pid, bid in best.assignments:
+            battle_map[bid].assign_participant(pid)
             remaining.remove(pid)
 
     @staticmethod
-    def _fallback_score(
+    def _search_best_fallback(
+        initial: _FallbackState,
+        match_counts: Dict[Tuple[str, str], int],
+        beam_width: int,
+    ) -> _FallbackState:
+        states = [initial]
+        while True:
+            new_states: List[_FallbackState] = []
+            progress = False
+            for state in states:
+                if not state.remaining:
+                    new_states.append(state)
+                    continue
+                progress = True
+                pid = state.remaining[0]
+                next_remaining = state.remaining[1:]
+                battles = Pairing._available_virtual_battles(state)
+                random.shuffle(battles)
+                for sector_id in battles:
+                    occupants = state.occupancy[sector_id]
+                    if len(occupants) >= 2:
+                        continue
+                    score = Pairing._fallback_score_virtual(
+                        pid,
+                        occupants,
+                        match_counts,
+                    )
+                    new_occupancy = deepcopy(state.occupancy)
+                    new_occupancy[sector_id] = occupants + [pid]
+                    new_states.append(
+                        _FallbackState(
+                            assignments=state.assignments + [(pid, sector_id)],
+                            remaining=next_remaining,
+                            occupancy=new_occupancy,
+                            score=state.score + score,
+                        )
+                    )
+            if not progress:
+                break
+            new_states.sort(key=lambda s: (s.score, random.random()))
+            states = new_states[:beam_width]
+        return min(states, key=lambda s: s.score)
+
+    @staticmethod
+    def _available_virtual_battles(state: _FallbackState) -> List[str]:
+        return [sector_id for sector_id, occ in state.occupancy.items() if len(occ) < 2]
+
+    @staticmethod
+    def _fallback_score_virtual(
         pid: str,
-        battle: Battle,
+        occupants: List[str],
         match_counts: Dict[Tuple[str, str], int],
     ) -> int:
         rematch_weight = 1
         occupancy_weight = 1
-        occupants = [
-            p for p in (battle.player_1_id, battle.player_2_id) if p is not None
-        ]
         occupancy_penalty = len(occupants)
         rematch_penalty = 0
         for opp in occupants:
@@ -316,6 +421,19 @@ class Pairing:
             rematch_penalty += match_counts.get(key, 0)
         return rematch_penalty * rematch_weight + occupancy_penalty * occupancy_weight
 
+    @staticmethod
+    def build_match_count(campaign: Campaign) -> Dict[Tuple[str, str], int]:
+        counts: Dict[Tuple[str, str], int] = {}
+        for rnd in campaign.rounds:
+            for battle in rnd.battles.values():
+                p1 = battle.player_1_id
+                p2 = battle.player_2_id
+                if not p1 or not p2:
+                    continue
+                key = (p1, p2) if p1 < p2 else (p2, p1)
+                counts[key] = counts.get(key, 0) + 1
+        return counts
+
     @staticmethod
     def get_allocation_kind(
         war: War,

src/warchron/model/round.py

@@ -166,6 +166,7 @@ class Round:
         return any(b.is_finished() for b in self.battles.values())
 
     def all_battles_finished(self) -> bool:
+        # TODO exception for participant alone
         return all(
             b.winner_id is not None or b.is_draw() for b in self.battles.values()
         )

src/warchron/view/view.py

@@ -201,6 +201,7 @@ class View(QtWidgets.QMainWindow, Ui_MainWindow):
             self.on_delete_item(ItemType.PLAYER, player_id)
 
     def display_players(self, players: List[ParticipantOption]) -> None:
+        # TODO display stats (war, campaign battles...)
         table = self.playersTable
         table.setSortingEnabled(False)
         table.setRowCount(len(players))