723 lines
33 KiB
Python
723 lines
33 KiB
Python
#!/usr/bin/env python3
|
|
"""Recalculate the fixed-seed build-policy anti-dominance fixture."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass, replace
|
|
from decimal import Decimal, ROUND_FLOOR
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
|
|
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_FIXTURE = REPOSITORY_ROOT / "docs/product/fixtures/ANTI_DOMINANCE_FIXED_SEEDS.json"
|
|
POLICIES = ("refresh_first", "hire_first", "dismiss_first", "no_replace")
|
|
COMPARISON_ORDER = (
|
|
"route_cleared",
|
|
"ending_coin",
|
|
"ending_credit",
|
|
"ending_live_recruits",
|
|
"counter_coverage",
|
|
)
|
|
DIFFICULTY_REWARD_MULTIPLIERS = {
|
|
"Wayfinder": Decimal("1.00"),
|
|
"Standard": Decimal("1.00"),
|
|
"Oathbound": Decimal("1.15"),
|
|
}
|
|
SEED_ID_RE = re.compile(r"^AD-FS-\d{3}-[A-Z0-9-]+$")
|
|
|
|
|
|
class FixtureError(ValueError):
|
|
"""Raised when a fixture or recalculated result violates the contract."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Outcome:
|
|
route_cleared: bool
|
|
ending_coin: int
|
|
ending_credit: int
|
|
ending_live_recruits: int
|
|
deployed_recruits: int
|
|
required_deployed_recruits: int
|
|
counter_coverage: int
|
|
counter_required: int
|
|
battle_coin: int
|
|
market_coin_spent: int
|
|
dismissal_rebate: int
|
|
credit_used: int
|
|
line_capacity: int
|
|
actions: Tuple[str, ...]
|
|
|
|
def comparison_key(self) -> Tuple[int, int, int, int, int]:
|
|
return (
|
|
int(self.route_cleared),
|
|
self.ending_coin,
|
|
self.ending_credit,
|
|
self.ending_live_recruits,
|
|
self.counter_coverage,
|
|
)
|
|
|
|
def as_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"route_cleared": self.route_cleared,
|
|
"ending_coin": self.ending_coin,
|
|
"ending_credit": self.ending_credit,
|
|
"ending_live_recruits": self.ending_live_recruits,
|
|
"deployed_recruits": self.deployed_recruits,
|
|
"required_deployed_recruits": self.required_deployed_recruits,
|
|
"counter_coverage": self.counter_coverage,
|
|
"counter_required": self.counter_required,
|
|
"battle_coin": self.battle_coin,
|
|
"market_coin_spent": self.market_coin_spent,
|
|
"dismissal_rebate": self.dismissal_rebate,
|
|
"credit_used": self.credit_used,
|
|
"line_capacity": self.line_capacity,
|
|
"actions": list(self.actions),
|
|
}
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise FixtureError(message)
|
|
|
|
|
|
def require_keys(value: Mapping[str, Any], expected: Iterable[str], label: str) -> None:
|
|
expected_set = set(expected)
|
|
actual_set = set(value)
|
|
require(
|
|
actual_set == expected_set,
|
|
"{} fields {} != {}".format(label, sorted(actual_set), sorted(expected_set)),
|
|
)
|
|
|
|
|
|
def require_int(value: Any, label: str, minimum: Optional[int] = None) -> int:
|
|
require(isinstance(value, int) and not isinstance(value, bool), "{} must be an integer".format(label))
|
|
if minimum is not None:
|
|
require(value >= minimum, "{} must be at least {}".format(label, minimum))
|
|
return value
|
|
|
|
|
|
def line_capacity(cleared_rounds: int) -> int:
|
|
return min(10, 4 + cleared_rounds // 2)
|
|
|
|
|
|
def offer_level(round_number: int) -> int:
|
|
return min(10, 1 + (round_number - 1) // 3)
|
|
|
|
|
|
def refresh_cost(prior_refreshes: int, paid_locks: int) -> int:
|
|
return min(24, 6 + 3 * prior_refreshes + 2 * paid_locks)
|
|
|
|
|
|
def hire_list_price(level: int, profession_tier: int) -> int:
|
|
return 10 + 4 * level + 10 * profession_tier
|
|
|
|
|
|
def equipment_list_price(region_band: int, tier: int) -> int:
|
|
return 12 + 3 * region_band + 6 * tier
|
|
|
|
|
|
def dismissal_rebate(coin_price_paid: int, survived_battles: int) -> int:
|
|
if survived_battles < 2:
|
|
return 0
|
|
return min(10, coin_price_paid // 4)
|
|
|
|
|
|
def death_credit(original_list_price: int, coin_price_paid: int) -> int:
|
|
return min(original_list_price, 10 + coin_price_paid // 2)
|
|
|
|
|
|
def battle_coin(region_band: int, round_number: int, risk_tier: int, difficulty: str) -> int:
|
|
multiplier = DIFFICULTY_REWARD_MULTIPLIERS[difficulty]
|
|
base = 18 + 2 * round_number + 2 * region_band + 7 * risk_tier
|
|
return int((Decimal(base) * multiplier).to_integral_value(rounding=ROUND_FLOOR))
|
|
|
|
|
|
def validate_recruit_offer(offer: Mapping[str, Any], label: str) -> None:
|
|
require_keys(
|
|
offer,
|
|
("offer_id", "level", "profession_tier", "tags", "protected", "list_price"),
|
|
label,
|
|
)
|
|
require(isinstance(offer["offer_id"], str) and offer["offer_id"], "{}.offer_id must be text".format(label))
|
|
level = require_int(offer["level"], "{}.level".format(label), 1)
|
|
profession_tier = require_int(offer["profession_tier"], "{}.profession_tier".format(label), 0)
|
|
require(profession_tier in (0, 1), "{}.profession_tier must be 0 or 1".format(label))
|
|
require(isinstance(offer["protected"], bool), "{}.protected must be boolean".format(label))
|
|
require(isinstance(offer["tags"], list) and offer["tags"], "{}.tags must be nonempty".format(label))
|
|
require(all(isinstance(tag, str) and tag for tag in offer["tags"]), "{}.tags must contain text".format(label))
|
|
require(len(set(offer["tags"])) == len(offer["tags"]), "{}.tags repeat".format(label))
|
|
expected = hire_list_price(level, profession_tier)
|
|
require(offer["list_price"] == expected, "{} list price {} != {}".format(label, offer["list_price"], expected))
|
|
|
|
|
|
def validate_equipment_offer(offer: Mapping[str, Any], label: str) -> None:
|
|
require_keys(
|
|
offer,
|
|
("offer_id", "region_band", "tier", "tags", "legal_user", "list_price"),
|
|
label,
|
|
)
|
|
require(isinstance(offer["offer_id"], str) and offer["offer_id"], "{}.offer_id must be text".format(label))
|
|
region_band = require_int(offer["region_band"], "{}.region_band".format(label), 0)
|
|
tier = require_int(offer["tier"], "{}.tier".format(label), 0)
|
|
require(region_band <= 7, "{}.region_band exceeds 7".format(label))
|
|
require(tier <= 3, "{}.tier exceeds 3".format(label))
|
|
require(offer["legal_user"] is True, "{}.legal_user must be true".format(label))
|
|
require(isinstance(offer["tags"], list) and offer["tags"], "{}.tags must be nonempty".format(label))
|
|
require(all(isinstance(tag, str) and tag for tag in offer["tags"]), "{}.tags must contain text".format(label))
|
|
require(len(set(offer["tags"])) == len(offer["tags"]), "{}.tags repeat".format(label))
|
|
expected = equipment_list_price(region_band, tier)
|
|
require(offer["list_price"] == expected, "{} list price {} != {}".format(label, offer["list_price"], expected))
|
|
|
|
|
|
def validate_battle_offer(
|
|
offer: Mapping[str, Any], state: Mapping[str, Any], label: str
|
|
) -> None:
|
|
require_keys(
|
|
offer,
|
|
(
|
|
"route_id",
|
|
"locked",
|
|
"risk_tier",
|
|
"enemy_line",
|
|
"required_tags",
|
|
"minimum_deployed",
|
|
"casualties_on_clear",
|
|
"casualties_on_failure",
|
|
"battle_coin",
|
|
),
|
|
label,
|
|
)
|
|
require(isinstance(offer["route_id"], str) and offer["route_id"], "{}.route_id must be text".format(label))
|
|
risk_tier = require_int(offer["risk_tier"], "{}.risk_tier".format(label), 0)
|
|
require(risk_tier <= 2, "{}.risk_tier exceeds 2".format(label))
|
|
require(isinstance(offer["locked"], bool), "{}.locked must be boolean".format(label))
|
|
require(isinstance(offer["required_tags"], list), "{}.required_tags must be a list".format(label))
|
|
require(all(isinstance(tag, str) and tag for tag in offer["required_tags"]), "{}.required_tags must contain text".format(label))
|
|
require(len(set(offer["required_tags"])) == len(offer["required_tags"]), "{}.required_tags repeat".format(label))
|
|
require_int(offer["minimum_deployed"], "{}.minimum_deployed".format(label), 1)
|
|
require(offer["minimum_deployed"] <= state["expected_line_capacity"], "{}.minimum_deployed exceeds Line capacity".format(label))
|
|
require_int(offer["casualties_on_clear"], "{}.casualties_on_clear".format(label), 0)
|
|
require_int(offer["casualties_on_failure"], "{}.casualties_on_failure".format(label), 0)
|
|
enemy_line = offer["enemy_line"]
|
|
require(isinstance(enemy_line, list) and enemy_line, "{}.enemy_line must be nonempty".format(label))
|
|
orders = [enemy.get("order") for enemy in enemy_line]
|
|
require(orders == list(range(1, len(enemy_line) + 1)), "{}.enemy_line order must be contiguous".format(label))
|
|
for index, enemy in enumerate(enemy_line):
|
|
require_keys(enemy, ("enemy_id", "order", "threat_tags"), "{}.enemy_line[{}]".format(label, index))
|
|
require(isinstance(enemy.get("enemy_id"), str), "{}.enemy_line[{}] needs enemy_id".format(label, index))
|
|
require(isinstance(enemy.get("threat_tags"), list), "{}.enemy_line[{}] needs threat_tags".format(label, index))
|
|
require(all(isinstance(tag, str) and tag for tag in enemy["threat_tags"]), "{}.enemy_line[{}] threat tags must contain text".format(label, index))
|
|
expected = battle_coin(
|
|
state["region_band"], state["round"], risk_tier, state["difficulty"]
|
|
)
|
|
require(offer["battle_coin"] == expected, "{} battle Coin {} != {}".format(label, offer["battle_coin"], expected))
|
|
|
|
|
|
def validate_board(board: Mapping[str, Any], state: Mapping[str, Any], label: str) -> None:
|
|
require(set(board) == {"recruit_offers", "equipment_offers", "battle_offers"}, "{} has incomplete or extra rows".format(label))
|
|
require(len(board["recruit_offers"]) == 3, "{} must contain three Recruit offers".format(label))
|
|
require(len(board["equipment_offers"]) == 4, "{} must contain four equipment offers".format(label))
|
|
require(len(board["battle_offers"]) == 3, "{} must contain three battle offers".format(label))
|
|
for index, offer in enumerate(board["recruit_offers"]):
|
|
validate_recruit_offer(offer, "{}.recruit_offers[{}]".format(label, index))
|
|
expected_level = max(1, state["expected_offer_level"] - 1) if offer["protected"] else state["expected_offer_level"]
|
|
require(offer["level"] == expected_level, "{}.recruit_offers[{}] level mismatch".format(label, index))
|
|
for index, offer in enumerate(board["equipment_offers"]):
|
|
validate_equipment_offer(offer, "{}.equipment_offers[{}]".format(label, index))
|
|
require(offer["region_band"] == state["region_band"], "{}.equipment_offers[{}] Region band mismatch".format(label, index))
|
|
for index, offer in enumerate(board["battle_offers"]):
|
|
validate_battle_offer(offer, state, "{}.battle_offers[{}]".format(label, index))
|
|
require(
|
|
{offer["risk_tier"] for offer in board["battle_offers"]} == {0, 1, 2},
|
|
"{} must expose one offer at each risk tier".format(label),
|
|
)
|
|
require(any(offer["profession_tier"] == 0 for offer in board["recruit_offers"]), "{} needs a base Recruit".format(label))
|
|
equipment_tags = {tag for offer in board["equipment_offers"] for tag in offer["tags"]}
|
|
require(len(equipment_tags) >= 3, "{} equipment row needs at least three behavior tags".format(label))
|
|
offer_ids = [offer["offer_id"] for row in ("recruit_offers", "equipment_offers") for offer in board[row]]
|
|
route_ids = [offer["route_id"] for offer in board["battle_offers"]]
|
|
require(len(set(offer_ids)) == len(offer_ids), "{} Recruit/equipment offer IDs repeat".format(label))
|
|
require(len(set(route_ids)) == len(route_ids), "{} route IDs repeat".format(label))
|
|
|
|
|
|
def find_route(board: Mapping[str, Any], route_id: str) -> Mapping[str, Any]:
|
|
matches = [offer for offer in board["battle_offers"] if offer["route_id"] == route_id]
|
|
require(len(matches) == 1, "route {} must appear exactly once on a board".format(route_id))
|
|
return matches[0]
|
|
|
|
|
|
def casualty_credit(seed: Mapping[str, Any]) -> int:
|
|
state = seed["state"]
|
|
credit = state["replacement_credit_before_casualties"]
|
|
for casualty in seed["casualties"]:
|
|
grant = death_credit(casualty["original_list_price"], casualty["coin_price_paid"])
|
|
require(casualty["expected_credit_grant"] == grant, "{} casualty grant mismatch".format(seed["seed_id"]))
|
|
credit = min(54, credit + grant)
|
|
return credit
|
|
|
|
|
|
def validate_seed(seed: Mapping[str, Any]) -> None:
|
|
seed_id = seed.get("seed_id")
|
|
require(isinstance(seed_id, str) and SEED_ID_RE.fullmatch(seed_id) is not None, "invalid stable seed_id {}".format(seed_id))
|
|
require_keys(
|
|
seed,
|
|
(
|
|
"seed_id",
|
|
"description",
|
|
"state",
|
|
"living_roster",
|
|
"casualties",
|
|
"protected_replacement_offer_id",
|
|
"current_board",
|
|
"refresh_boards",
|
|
"selected_route_id",
|
|
"expected_winners",
|
|
"expected_outcomes",
|
|
),
|
|
seed_id,
|
|
)
|
|
require(isinstance(seed["description"], str) and seed["description"], "{} description must be text".format(seed_id))
|
|
state = seed["state"]
|
|
require_keys(
|
|
state,
|
|
(
|
|
"market_open",
|
|
"round",
|
|
"cleared_rounds",
|
|
"region_band",
|
|
"difficulty",
|
|
"coin",
|
|
"replacement_credit_before_casualties",
|
|
"expected_replacement_credit",
|
|
"prior_refreshes",
|
|
"paid_locks",
|
|
"expected_line_capacity",
|
|
"expected_offer_level",
|
|
"expected_refresh_cost",
|
|
),
|
|
"{}.state".format(seed_id),
|
|
)
|
|
require(state["market_open"] is True, "{} Market must be open".format(seed_id))
|
|
round_number = require_int(state["round"], "{}.round".format(seed_id), 1)
|
|
require(round_number <= 12, "{}.round exceeds 12".format(seed_id))
|
|
cleared_rounds = require_int(state["cleared_rounds"], "{}.cleared_rounds".format(seed_id), 0)
|
|
require(cleared_rounds == round_number - 1, "{} cleared_rounds must precede round".format(seed_id))
|
|
require(state["difficulty"] in DIFFICULTY_REWARD_MULTIPLIERS, "{} invalid difficulty".format(seed_id))
|
|
require_int(state["region_band"], "{}.region_band".format(seed_id), 0)
|
|
require(state["region_band"] <= 7, "{}.region_band exceeds 7".format(seed_id))
|
|
for key in ("coin", "replacement_credit_before_casualties", "expected_replacement_credit", "prior_refreshes", "paid_locks"):
|
|
require_int(state[key], "{}.{}".format(seed_id, key), 0)
|
|
require(state["replacement_credit_before_casualties"] <= 54, "{} starting Credit exceeds cap".format(seed_id))
|
|
require(state["expected_replacement_credit"] <= 54, "{} resulting Credit exceeds cap".format(seed_id))
|
|
require(state["paid_locks"] <= 3, "{} paid_locks exceeds cap".format(seed_id))
|
|
require(state["expected_line_capacity"] == line_capacity(cleared_rounds), "{} line capacity mismatch".format(seed_id))
|
|
require(state["expected_offer_level"] == offer_level(round_number), "{} offer level mismatch".format(seed_id))
|
|
expected_refresh = refresh_cost(state["prior_refreshes"], state["paid_locks"])
|
|
require(state["expected_refresh_cost"] == expected_refresh, "{} refresh cost mismatch".format(seed_id))
|
|
|
|
require(isinstance(seed["casualties"], list), "{} casualties must be a list".format(seed_id))
|
|
for casualty in seed["casualties"]:
|
|
require_keys(
|
|
casualty,
|
|
("recruit_id", "original_list_price", "coin_price_paid", "expected_credit_grant"),
|
|
"{}.casualties".format(seed_id),
|
|
)
|
|
require(isinstance(casualty["recruit_id"], str) and casualty["recruit_id"], "{} casualty ID must be text".format(seed_id))
|
|
for key in ("original_list_price", "coin_price_paid", "expected_credit_grant"):
|
|
require_int(casualty[key], "{}.casualties.{}".format(seed_id, key), 0)
|
|
require(state["expected_replacement_credit"] == casualty_credit(seed), "{} post-casualty Credit mismatch".format(seed_id))
|
|
|
|
require(isinstance(seed["living_roster"], list) and seed["living_roster"], "{} needs a living roster".format(seed_id))
|
|
roster_ids = set()
|
|
deployed = 0
|
|
for recruit in seed["living_roster"]:
|
|
require_keys(
|
|
recruit,
|
|
("recruit_id", "assignment", "tags", "coin_price_paid", "survived_battles"),
|
|
"{}.living_roster".format(seed_id),
|
|
)
|
|
recruit_id = recruit.get("recruit_id")
|
|
require(isinstance(recruit_id, str) and recruit_id not in roster_ids, "{} roster IDs must be unique".format(seed_id))
|
|
roster_ids.add(recruit_id)
|
|
require(recruit.get("assignment") in ("deployed", "reserve"), "{} invalid roster assignment".format(seed_id))
|
|
deployed += int(recruit["assignment"] == "deployed")
|
|
require(isinstance(recruit.get("tags"), list), "{} roster tags must be explicit".format(seed_id))
|
|
require(all(isinstance(tag, str) and tag for tag in recruit["tags"]), "{} roster tags must contain text".format(seed_id))
|
|
require_int(recruit.get("coin_price_paid"), "{}.coin_price_paid".format(seed_id), 0)
|
|
require_int(recruit.get("survived_battles"), "{}.survived_battles".format(seed_id), 0)
|
|
require(deployed <= state["expected_line_capacity"], "{} deployment exceeds Line capacity".format(seed_id))
|
|
|
|
validate_board(seed["current_board"], state, "{}.current_board".format(seed_id))
|
|
require(len(seed["refresh_boards"]) == 1, "{} must declare exactly the first refresh board".format(seed_id))
|
|
validate_board(seed["refresh_boards"][0], state, "{}.refresh_boards[0]".format(seed_id))
|
|
|
|
current_route = find_route(seed["current_board"], seed["selected_route_id"])
|
|
refreshed_route = find_route(seed["refresh_boards"][0], seed["selected_route_id"])
|
|
require(current_route["locked"] is True, "{} selected route must be locked".format(seed_id))
|
|
require(current_route == refreshed_route, "{} locked route changed on refresh".format(seed_id))
|
|
locked_count = sum(
|
|
int(offer.get("locked", False))
|
|
for row in seed["current_board"].values()
|
|
for offer in row
|
|
)
|
|
require(locked_count == state["paid_locks"], "{} paid_locks does not match locked offers".format(seed_id))
|
|
|
|
current_protected = [offer for offer in seed["current_board"]["recruit_offers"] if offer["protected"]]
|
|
refreshed_protected = [offer for offer in seed["refresh_boards"][0]["recruit_offers"] if offer["protected"]]
|
|
protected_id = seed["protected_replacement_offer_id"]
|
|
if seed["casualties"]:
|
|
require(protected_id is not None, "{} casualty state needs a protected offer".format(seed_id))
|
|
require(len(current_protected) == len(refreshed_protected) == 1, "{} needs one protected offer per board".format(seed_id))
|
|
require(current_protected[0]["offer_id"] == protected_id, "{} protected offer ID mismatch".format(seed_id))
|
|
require(current_protected[0] == refreshed_protected[0], "{} protected offer changed on refresh".format(seed_id))
|
|
expected_level = max(1, state["expected_offer_level"] - 1)
|
|
require(current_protected[0]["level"] == expected_level, "{} protected offer level mismatch".format(seed_id))
|
|
else:
|
|
require(protected_id is None and not current_protected and not refreshed_protected, "{} cannot have a protected offer without casualty input".format(seed_id))
|
|
|
|
require(set(seed["expected_outcomes"]) == set(POLICIES), "{} expected outcomes must name all policies".format(seed_id))
|
|
require(isinstance(seed["expected_winners"], list) and seed["expected_winners"], "{} expected_winners must be nonempty".format(seed_id))
|
|
require(set(seed["expected_winners"]).issubset(set(POLICIES)), "{} expected_winners name an unknown policy".format(seed_id))
|
|
|
|
|
|
def validate_fixture(fixture: Mapping[str, Any]) -> None:
|
|
require_keys(
|
|
fixture,
|
|
("schema_version", "fixture_id", "policy_order", "comparison_order", "seeds"),
|
|
"fixture",
|
|
)
|
|
require(fixture.get("schema_version") == 1, "schema_version must be 1")
|
|
require(fixture.get("fixture_id") == "ABG-ANTI-DOMINANCE-FIXED-SEEDS-V1", "unexpected fixture_id")
|
|
require(tuple(fixture.get("policy_order", ())) == POLICIES, "policy_order mismatch")
|
|
require(tuple(fixture.get("comparison_order", ())) == COMPARISON_ORDER, "comparison_order mismatch")
|
|
seeds = fixture.get("seeds")
|
|
require(isinstance(seeds, list) and len(seeds) == 5, "fixture must contain exactly five seeds")
|
|
seed_ids = [seed.get("seed_id") for seed in seeds]
|
|
require(len(set(seed_ids)) == 5, "seed IDs must be unique")
|
|
for seed in seeds:
|
|
validate_seed(seed)
|
|
|
|
|
|
def active_tags(roster: Iterable[Mapping[str, Any]], equipment_tags: Iterable[str]) -> set:
|
|
tags = set(equipment_tags)
|
|
for recruit in roster:
|
|
if recruit["assignment"] == "deployed":
|
|
tags.update(recruit["tags"])
|
|
return tags
|
|
|
|
|
|
def deployed_count(roster: Iterable[Mapping[str, Any]]) -> int:
|
|
return sum(1 for recruit in roster if recruit["assignment"] == "deployed")
|
|
|
|
|
|
def best_equipment(
|
|
offers: Iterable[Mapping[str, Any]], coin: int, required_tags: set, build_tags: set
|
|
) -> Optional[Mapping[str, Any]]:
|
|
candidates = []
|
|
for offer in offers:
|
|
gain = len((set(offer["tags"]) & required_tags) - build_tags)
|
|
if gain > 0 and offer["list_price"] <= coin:
|
|
candidates.append((-gain, offer["list_price"], offer["offer_id"], offer))
|
|
return min(candidates)[3] if candidates else None
|
|
|
|
|
|
def hire_candidate(
|
|
offers: Iterable[Mapping[str, Any]], coin: int, credit: int, required_tags: set, build_tags: set
|
|
) -> Optional[Mapping[str, Any]]:
|
|
candidates = []
|
|
for offer in offers:
|
|
coin_cost = max(0, offer["list_price"] - credit)
|
|
if coin_cost > coin:
|
|
continue
|
|
gain = len((set(offer["tags"]) & required_tags) - build_tags)
|
|
candidates.append(
|
|
(-int(offer["protected"]), -gain, coin_cost, offer["list_price"], offer["offer_id"], offer)
|
|
)
|
|
return min(candidates)[5] if candidates else None
|
|
|
|
|
|
def simulate_policy(seed: Mapping[str, Any], policy: str) -> Outcome:
|
|
require(policy in POLICIES, "unknown policy {}".format(policy))
|
|
state = seed["state"]
|
|
board = seed["current_board"]
|
|
roster: List[Dict[str, Any]] = copy.deepcopy(seed["living_roster"])
|
|
equipment_tags: set = set()
|
|
coin = state["coin"]
|
|
credit = casualty_credit(seed)
|
|
capacity = line_capacity(state["cleared_rounds"])
|
|
route = find_route(board, seed["selected_route_id"])
|
|
required_tags = set(route["required_tags"])
|
|
market_coin_spent = 0
|
|
rebate_total = 0
|
|
credit_used_total = 0
|
|
actions: List[str] = []
|
|
|
|
if policy == "refresh_first":
|
|
cost = refresh_cost(state["prior_refreshes"], state["paid_locks"])
|
|
if coin >= cost:
|
|
coin -= cost
|
|
market_coin_spent += cost
|
|
actions.append("refresh:{}".format(cost))
|
|
board = seed["refresh_boards"][0]
|
|
route = find_route(board, seed["selected_route_id"])
|
|
else:
|
|
actions.append("skip_refresh:INSUFFICIENT_COIN")
|
|
build_tags = active_tags(roster, equipment_tags)
|
|
equipment = best_equipment(board["equipment_offers"], coin, required_tags, build_tags)
|
|
if equipment is not None:
|
|
coin -= equipment["list_price"]
|
|
market_coin_spent += equipment["list_price"]
|
|
equipment_tags.update(equipment["tags"])
|
|
actions.append("buy_equipment:{}:{}".format(equipment["offer_id"], equipment["list_price"]))
|
|
|
|
elif policy == "hire_first":
|
|
build_tags = active_tags(roster, equipment_tags)
|
|
recruit = hire_candidate(board["recruit_offers"], coin, credit, required_tags, build_tags)
|
|
if recruit is None:
|
|
actions.append("skip_hire:INSUFFICIENT_RESOURCES")
|
|
else:
|
|
credit_used = min(credit, recruit["list_price"])
|
|
coin_cost = recruit["list_price"] - credit_used
|
|
credit -= credit_used
|
|
coin -= coin_cost
|
|
market_coin_spent += coin_cost
|
|
credit_used_total += credit_used
|
|
assignment = "deployed" if deployed_count(roster) < capacity else "reserve"
|
|
roster.append(
|
|
{
|
|
"recruit_id": recruit["offer_id"],
|
|
"assignment": assignment,
|
|
"tags": list(recruit["tags"]),
|
|
"coin_price_paid": coin_cost,
|
|
"survived_battles": 0,
|
|
}
|
|
)
|
|
actions.append(
|
|
"hire:{}:coin={}:credit={}".format(recruit["offer_id"], coin_cost, credit_used)
|
|
)
|
|
|
|
elif policy == "dismiss_first":
|
|
if len(roster) < 2:
|
|
actions.append("skip_dismiss:FINAL_LIVE_RECRUIT")
|
|
else:
|
|
reserves = [recruit for recruit in roster if recruit["assignment"] == "reserve"]
|
|
if not reserves:
|
|
actions.append("skip_dismiss:NO_ELIGIBLE_RESERVE")
|
|
else:
|
|
target = min(
|
|
reserves,
|
|
key=lambda recruit: (
|
|
len(set(recruit["tags"]) & required_tags),
|
|
-dismissal_rebate(recruit["coin_price_paid"], recruit["survived_battles"]),
|
|
recruit["recruit_id"],
|
|
),
|
|
)
|
|
rebate = dismissal_rebate(target["coin_price_paid"], target["survived_battles"])
|
|
roster.remove(target)
|
|
coin += rebate
|
|
rebate_total += rebate
|
|
actions.append("dismiss:{}:rebate={}".format(target["recruit_id"], rebate))
|
|
build_tags = active_tags(roster, equipment_tags)
|
|
equipment = best_equipment(board["equipment_offers"], coin, required_tags, build_tags)
|
|
if equipment is not None:
|
|
coin -= equipment["list_price"]
|
|
market_coin_spent += equipment["list_price"]
|
|
equipment_tags.update(equipment["tags"])
|
|
actions.append("buy_equipment:{}:{}".format(equipment["offer_id"], equipment["list_price"]))
|
|
|
|
elif policy == "no_replace":
|
|
build_tags = active_tags(roster, equipment_tags)
|
|
equipment = best_equipment(board["equipment_offers"], coin, required_tags, build_tags)
|
|
if equipment is not None:
|
|
coin -= equipment["list_price"]
|
|
market_coin_spent += equipment["list_price"]
|
|
equipment_tags.update(equipment["tags"])
|
|
actions.append("buy_equipment:{}:{}".format(equipment["offer_id"], equipment["list_price"]))
|
|
|
|
build_tags = active_tags(roster, equipment_tags)
|
|
coverage = len(required_tags & build_tags)
|
|
deployed = deployed_count(roster)
|
|
cleared = coverage == len(required_tags) and deployed >= route["minimum_deployed"]
|
|
casualties = route["casualties_on_clear"] if cleared else route["casualties_on_failure"]
|
|
require(casualties <= len(roster), "{} {} casualty count exceeds roster".format(seed["seed_id"], policy))
|
|
reward = battle_coin(state["region_band"], state["round"], route["risk_tier"], state["difficulty"]) if cleared else 0
|
|
return Outcome(
|
|
route_cleared=cleared,
|
|
ending_coin=coin + reward,
|
|
ending_credit=credit,
|
|
ending_live_recruits=len(roster) - casualties,
|
|
deployed_recruits=deployed,
|
|
required_deployed_recruits=route["minimum_deployed"],
|
|
counter_coverage=coverage,
|
|
counter_required=len(required_tags),
|
|
battle_coin=reward,
|
|
market_coin_spent=market_coin_spent,
|
|
dismissal_rebate=rebate_total,
|
|
credit_used=credit_used_total,
|
|
line_capacity=capacity,
|
|
actions=tuple(actions),
|
|
)
|
|
|
|
|
|
def winners(outcomes: Mapping[str, Outcome]) -> Tuple[str, ...]:
|
|
best = max(outcome.comparison_key() for outcome in outcomes.values())
|
|
return tuple(policy for policy in POLICIES if outcomes[policy].comparison_key() == best)
|
|
|
|
|
|
def assert_no_weak_dominance(results: Mapping[str, Mapping[str, Outcome]]) -> None:
|
|
for left in POLICIES:
|
|
for right in POLICIES:
|
|
if left == right:
|
|
continue
|
|
comparisons = [
|
|
(
|
|
seed_outcomes[left].comparison_key(),
|
|
seed_outcomes[right].comparison_key(),
|
|
)
|
|
for seed_outcomes in results.values()
|
|
]
|
|
if all(left_key >= right_key for left_key, right_key in comparisons) and any(
|
|
left_key > right_key for left_key, right_key in comparisons
|
|
):
|
|
raise FixtureError("policy {} weakly dominates {}".format(left, right))
|
|
|
|
|
|
def audit_fixture(fixture: Mapping[str, Any]) -> Dict[str, Dict[str, Outcome]]:
|
|
validate_fixture(fixture)
|
|
results: Dict[str, Dict[str, Outcome]] = {}
|
|
win_counts = {policy: 0 for policy in POLICIES}
|
|
loss_counts = {policy: 0 for policy in POLICIES}
|
|
for seed in fixture["seeds"]:
|
|
seed_id = seed["seed_id"]
|
|
seed_outcomes = {policy: simulate_policy(seed, policy) for policy in POLICIES}
|
|
results[seed_id] = seed_outcomes
|
|
actual_winners = winners(seed_outcomes)
|
|
require(list(actual_winners) == seed["expected_winners"], "{} winners {} != {}".format(seed_id, list(actual_winners), seed["expected_winners"]))
|
|
for policy in POLICIES:
|
|
actual = seed_outcomes[policy].as_dict()
|
|
expected = seed["expected_outcomes"][policy]
|
|
require(actual == expected, "{} {} outcome mismatch\nactual={}\nexpected={}".format(seed_id, policy, json.dumps(actual, sort_keys=True), json.dumps(expected, sort_keys=True)))
|
|
if policy in actual_winners:
|
|
win_counts[policy] += 1
|
|
else:
|
|
loss_counts[policy] += 1
|
|
for policy in POLICIES:
|
|
require(win_counts[policy] >= 1, "policy {} never wins".format(policy))
|
|
require(loss_counts[policy] >= 1, "policy {} never loses".format(policy))
|
|
assert_no_weak_dominance(results)
|
|
return results
|
|
|
|
|
|
def result_payload(
|
|
fixture: Mapping[str, Any], results: Mapping[str, Mapping[str, Outcome]]
|
|
) -> Dict[str, Any]:
|
|
return {
|
|
"fixture_id": fixture["fixture_id"],
|
|
"comparison_order": list(COMPARISON_ORDER),
|
|
"seeds": [
|
|
{
|
|
"seed_id": seed["seed_id"],
|
|
"winners": list(winners(results[seed["seed_id"]])),
|
|
"outcomes": {
|
|
policy: results[seed["seed_id"]][policy].as_dict()
|
|
for policy in POLICIES
|
|
},
|
|
}
|
|
for seed in fixture["seeds"]
|
|
],
|
|
}
|
|
|
|
|
|
def render_summary(
|
|
fixture: Mapping[str, Any], results: Mapping[str, Mapping[str, Outcome]]
|
|
) -> str:
|
|
payload = result_payload(fixture, results)
|
|
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
|
|
digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
lines = [
|
|
"ABG_ANTI_DOMINANCE fixture={} seeds=5 policies=4".format(fixture["fixture_id"])
|
|
]
|
|
win_counts = {policy: 0 for policy in POLICIES}
|
|
for seed in fixture["seeds"]:
|
|
seed_id = seed["seed_id"]
|
|
seed_winners = winners(results[seed_id])
|
|
for policy in seed_winners:
|
|
win_counts[policy] += 1
|
|
metrics = []
|
|
for policy in POLICIES:
|
|
outcome = results[seed_id][policy]
|
|
metrics.append(
|
|
"{}:clear={},coin={},credit={},live={},coverage={}/{}".format(
|
|
policy,
|
|
int(outcome.route_cleared),
|
|
outcome.ending_coin,
|
|
outcome.ending_credit,
|
|
outcome.ending_live_recruits,
|
|
outcome.counter_coverage,
|
|
outcome.counter_required,
|
|
)
|
|
)
|
|
lines.append("SEED {} winners={} {}".format(seed_id, ",".join(seed_winners), " ".join(metrics)))
|
|
lines.append(
|
|
"WINS {}".format(" ".join("{}={}".format(policy, win_counts[policy]) for policy in POLICIES))
|
|
)
|
|
lines.append("RESULT_SHA256 {}".format(digest))
|
|
lines.append("ABG_ANTI_DOMINANCE_OK no_weak_dominance=1")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def run_self_test(fixture: Mapping[str, Any]) -> None:
|
|
results = audit_fixture(fixture)
|
|
dominant = copy.deepcopy(results)
|
|
for seed_outcomes in dominant.values():
|
|
maximum_coin = max(outcome.ending_coin for outcome in seed_outcomes.values())
|
|
seed_outcomes["refresh_first"] = replace(
|
|
seed_outcomes["refresh_first"], route_cleared=True, ending_coin=maximum_coin + 1000
|
|
)
|
|
try:
|
|
assert_no_weak_dominance(dominant)
|
|
except FixtureError as error:
|
|
require("weakly dominates" in str(error), "dominant negative matrix failed for the wrong reason")
|
|
else:
|
|
raise FixtureError("deliberately dominant in-memory matrix was accepted")
|
|
print("ABG_ANTI_DOMINANCE_SELF_TEST_OK seeds=5 dominant_matrix_rejected=1")
|
|
|
|
|
|
def load_fixture(path: Path) -> Dict[str, Any]:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as error:
|
|
raise FixtureError("cannot load {}: {}".format(path, error))
|
|
|
|
|
|
def parse_arguments(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE, help="fixed-seed JSON fixture")
|
|
parser.add_argument("--self-test", action="store_true", help="run canonical and dominant negative fixtures")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: Optional[Sequence[str]] = None) -> int:
|
|
arguments = parse_arguments(argv)
|
|
try:
|
|
fixture = load_fixture(arguments.fixture)
|
|
if arguments.self_test:
|
|
run_self_test(fixture)
|
|
else:
|
|
results = audit_fixture(fixture)
|
|
print(render_summary(fixture, results))
|
|
except FixtureError as error:
|
|
print("ABG_ANTI_DOMINANCE_FAILED {}".format(error))
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|