319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""Strategy search and playbook optimization (docs/phase-2-llm-architecture.md
|
||
sections 3.1, 7, 10; P3.2/P3.3).
|
||
|
||
``evaluate`` measures one strategy against one scenario and returns its
|
||
``StrategyBenchmark`` (danger metrics, win counts, and the party's 3-sigma
|
||
win-rate band). ``optimize`` searches the strategy space in three stages - a
|
||
coarse 24-cell grid over the target/spell/charge/class-ability levers, a
|
||
retreat-threshold refinement of the top grid cells, then any LLM
|
||
``candidates`` - and returns a ranked ``PlaybookEntry`` tuple.
|
||
``strategy_name`` renders a deterministic label for a spec and
|
||
``format_playbook`` renders the ranked playbook as aligned ASCII columns.
|
||
Deterministic: the only randomness comes from the fixed ``seed`` passed to
|
||
the sweep runner.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, replace
|
||
from typing import TYPE_CHECKING
|
||
|
||
if TYPE_CHECKING:
|
||
from collections.abc import Sequence
|
||
|
||
from pf1e_simulator.objectives import DangerMetrics
|
||
from pf1e_simulator.runner import EncounterSpec
|
||
from pf1e_simulator.scenarios import ScenarioSpec
|
||
from pf1e_simulator.sweep import SweepCellResult
|
||
|
||
from pf1e_simulator.scenarios import build_encounter
|
||
from pf1e_simulator.strategy import StrategySpec, compile_strategy
|
||
from pf1e_simulator.sweep import SweepCell, SweepRunner
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StrategyBenchmark:
|
||
"""The measured outcome of one strategy on one scenario (P3.2)."""
|
||
|
||
strategy: StrategySpec
|
||
scenario_name: str
|
||
danger: DangerMetrics
|
||
wins: dict[str, int]
|
||
draws: int
|
||
avg_rounds: float
|
||
win_rate_band: tuple[float, float]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PlaybookEntry:
|
||
"""One ranked row of an optimized playbook (P3.2/P3.3)."""
|
||
|
||
rank: int
|
||
strategy: StrategySpec
|
||
benchmark: StrategyBenchmark
|
||
|
||
|
||
def strategy_name(strategy: StrategySpec) -> str:
|
||
"""A deterministic label for `strategy` from its non-default levers.
|
||
|
||
Levers render in field order (target_priority, spell_policy, use_charges,
|
||
use_class_abilities, retreat_hp_fraction), joined with ", "; the `name`
|
||
field is never part of the label. An all-default spec is "default".
|
||
"""
|
||
levers: list[str] = []
|
||
if strategy.target_priority != "nearest":
|
||
levers.append(strategy.target_priority)
|
||
if strategy.spell_policy != "default":
|
||
levers.append(f"spell_policy={strategy.spell_policy}")
|
||
if not strategy.use_charges:
|
||
levers.append("use_charges=false")
|
||
if not strategy.use_class_abilities:
|
||
levers.append("use_class_abilities=false")
|
||
if strategy.retreat_hp_fraction != 0.0:
|
||
levers.append(f"retreat={strategy.retreat_hp_fraction}")
|
||
if not levers:
|
||
return "default"
|
||
return ", ".join(levers)
|
||
|
||
|
||
def _grid_specs() -> tuple[StrategySpec, ...]:
|
||
"""The coarse 24-cell grid: 3 targets x 2 spell policies x 2 x 2, retreat 0."""
|
||
specs: list[StrategySpec] = []
|
||
for target_priority in ("nearest", "weakest_ac", "lowest_hp"):
|
||
for spell_policy in ("default", "none"):
|
||
for use_charges in (True, False):
|
||
for use_class_abilities in (True, False):
|
||
raw = StrategySpec(
|
||
target_priority=target_priority,
|
||
spell_policy=spell_policy,
|
||
use_charges=use_charges,
|
||
use_class_abilities=use_class_abilities,
|
||
)
|
||
specs.append(replace(raw, name=strategy_name(raw)))
|
||
return tuple(specs)
|
||
|
||
|
||
def _refine_specs(top: Sequence[StrategySpec]) -> tuple[StrategySpec, ...]:
|
||
"""Retreat-threshold refinements of the top grid specs (3 per spec)."""
|
||
refined: list[StrategySpec] = []
|
||
for spec in top:
|
||
for fraction in (0.25, 0.5, 0.75):
|
||
raw = replace(spec, retreat_hp_fraction=fraction)
|
||
refined.append(replace(raw, name=strategy_name(raw)))
|
||
return tuple(refined)
|
||
|
||
|
||
def _rank_key(
|
||
strategy: StrategySpec, danger: DangerMetrics
|
||
) -> tuple[float, float, float, float, str]:
|
||
"""Sort key: best first (survive high, hp lost low, tail high, rounds low)."""
|
||
return (
|
||
-danger.p_survives,
|
||
danger.expected_hp_lost,
|
||
-danger.worst_decile_party_hp,
|
||
danger.expected_rounds,
|
||
strategy_name(strategy),
|
||
)
|
||
|
||
|
||
def _cell(
|
||
scenario: ScenarioSpec,
|
||
encounter: EncounterSpec,
|
||
spec: StrategySpec,
|
||
runs: int,
|
||
seed: int,
|
||
) -> SweepCell:
|
||
"""One sweep cell for `spec` under its compiled strategy."""
|
||
return SweepCell(
|
||
scenario=scenario.name,
|
||
strategy=strategy_name(spec),
|
||
spec=encounter,
|
||
policy=compile_strategy(spec),
|
||
runs=runs,
|
||
seed=seed,
|
||
)
|
||
|
||
|
||
def _cells(
|
||
scenario: ScenarioSpec,
|
||
encounter: EncounterSpec,
|
||
specs: Sequence[StrategySpec],
|
||
runs: int,
|
||
seed: int,
|
||
) -> tuple[SweepCell, ...]:
|
||
"""One cell per spec, in spec order."""
|
||
return tuple(_cell(scenario, encounter, spec, runs, seed) for spec in specs)
|
||
|
||
|
||
def _benchmark(
|
||
strategy: StrategySpec,
|
||
scenario_name: str,
|
||
result: SweepCellResult,
|
||
party: str,
|
||
) -> StrategyBenchmark:
|
||
"""Build the benchmark for one strategy from its sweep cell result."""
|
||
return StrategyBenchmark(
|
||
strategy=strategy,
|
||
scenario_name=scenario_name,
|
||
danger=result.danger,
|
||
wins=result.report.wins,
|
||
draws=result.report.draws,
|
||
avg_rounds=result.report.avg_rounds,
|
||
win_rate_band=result.report.win_rate_band(party),
|
||
)
|
||
|
||
|
||
def evaluate( # noqa: PLR0913 - pinned Phase-2 API (6 knobs)
|
||
strategy: StrategySpec,
|
||
scenario: ScenarioSpec,
|
||
*,
|
||
party: str,
|
||
runs: int,
|
||
seed: int,
|
||
workers: int = 4,
|
||
) -> StrategyBenchmark:
|
||
"""Measure one strategy on one scenario (P3.2).
|
||
|
||
Compiles `strategy` and runs a single sweep cell of `runs` seeded battles;
|
||
the benchmark carries the party's danger metrics, win counts, and 3-sigma
|
||
win-rate band.
|
||
"""
|
||
encounter = build_encounter(scenario)
|
||
runner = SweepRunner(party=party, workers=workers)
|
||
(result,) = runner.run((_cell(scenario, encounter, strategy, runs, seed),))
|
||
return _benchmark(strategy, scenario.name, result, party)
|
||
|
||
|
||
def optimize( # noqa: PLR0913 - pinned Phase-2 API (7 knobs)
|
||
scenario: ScenarioSpec,
|
||
*,
|
||
party: str,
|
||
runs: int,
|
||
seed: int,
|
||
workers: int = 4,
|
||
max_total_runs: int | None = None,
|
||
candidates: Sequence[StrategySpec] = (),
|
||
) -> tuple[PlaybookEntry, ...]:
|
||
"""Search the strategy space and return a ranked playbook (P3.2/P3.3).
|
||
|
||
Stage 1 sweeps the coarse 24-cell grid (target priority x spell policy x
|
||
charges x class abilities, retreat 0); stage 2 refines the top 3 grid
|
||
cells over retreat thresholds 0.25/0.5/0.75; stage 3 adds the LLM
|
||
`candidates`, deduplicated by field equality against every spec already
|
||
run. Entries are ranked by (-p_survives, +expected_hp_lost,
|
||
-worst_decile_party_hp, +avg_rounds, strategy_name). `max_total_runs`
|
||
caps the total Monte Carlo cost before anything executes (mirrors
|
||
SweepRunner.run). The signature is pinned by the Phase-2 API, hence the
|
||
PLR0913 noqa.
|
||
"""
|
||
encounter = build_encounter(scenario)
|
||
runner = SweepRunner(party=party, workers=workers)
|
||
|
||
stage1_specs = _grid_specs()
|
||
seen: set[StrategySpec] = set(stage1_specs)
|
||
stage3_specs: list[StrategySpec] = []
|
||
for spec in candidates:
|
||
if spec not in seen:
|
||
seen.add(spec)
|
||
stage3_specs.append(spec)
|
||
|
||
total_runs = (len(stage1_specs) + 9 + len(stage3_specs)) * runs
|
||
if max_total_runs is not None and total_runs > max_total_runs:
|
||
msg = (
|
||
f"optimize budget exceeded: {total_runs} runs > "
|
||
f"{max_total_runs} max_total_runs"
|
||
)
|
||
raise ValueError(msg)
|
||
|
||
stage1_results = runner.run(_cells(scenario, encounter, stage1_specs, runs, seed))
|
||
stage1_ranked = sorted(
|
||
zip(stage1_specs, stage1_results, strict=True),
|
||
key=lambda pair: _rank_key(pair[0], pair[1].danger),
|
||
)
|
||
top3 = [spec for spec, _ in stage1_ranked[:3]]
|
||
stage2_specs = _refine_specs(top3)
|
||
stage2_results = runner.run(_cells(scenario, encounter, stage2_specs, runs, seed))
|
||
|
||
seen = set(stage1_specs) | set(stage2_specs)
|
||
stage3_specs = []
|
||
for spec in candidates:
|
||
if spec not in seen:
|
||
seen.add(spec)
|
||
stage3_specs.append(spec)
|
||
stage3_results = (
|
||
runner.run(_cells(scenario, encounter, stage3_specs, runs, seed))
|
||
if stage3_specs
|
||
else ()
|
||
)
|
||
|
||
pairs = [
|
||
*zip(stage1_specs, stage1_results, strict=True),
|
||
*zip(stage2_specs, stage2_results, strict=True),
|
||
*zip(stage3_specs, stage3_results, strict=True),
|
||
]
|
||
benchmarks = [
|
||
_benchmark(spec, scenario.name, result, party) for spec, result in pairs
|
||
]
|
||
ordered = sorted(benchmarks, key=lambda b: _rank_key(b.strategy, b.danger))
|
||
return tuple(
|
||
PlaybookEntry(rank=index, strategy=benchmark.strategy, benchmark=benchmark)
|
||
for index, benchmark in enumerate(ordered, start=1)
|
||
)
|
||
|
||
|
||
def _win_rate_text(benchmark: StrategyBenchmark, party: str) -> str:
|
||
"""The party's win rate with its 3-sigma band, e.g. "93.0 [89.5, 96.5]"."""
|
||
rate = benchmark.wins.get(party, 0) / benchmark.danger.runs
|
||
low, high = benchmark.win_rate_band
|
||
return f"{rate * 100:.1f} [{low * 100:.1f}, {high * 100:.1f}]"
|
||
|
||
|
||
def format_playbook(entries: Sequence[PlaybookEntry], party: str) -> str:
|
||
"""Render a ranked playbook as aligned plain-text columns (ASCII only).
|
||
|
||
One row per entry in order, under a header line and a separator row of
|
||
dashes. Percentages and HP values use one decimal; the Win% column shows
|
||
the party's win rate with its 3-sigma band. Column widths are computed
|
||
from the content; columns are separated by a single space.
|
||
"""
|
||
headers = (
|
||
"Rank",
|
||
"Strategy",
|
||
"Runs",
|
||
"P(survive)",
|
||
"P(TPK)",
|
||
"P(>=1 death)",
|
||
"HP lost",
|
||
"Worst-decile HP",
|
||
"Rounds",
|
||
"Win% (3σ)",
|
||
)
|
||
rows = [
|
||
(
|
||
str(entry.rank),
|
||
entry.benchmark.strategy.name,
|
||
str(entry.benchmark.danger.runs),
|
||
f"{entry.benchmark.danger.p_survives:.1%}",
|
||
f"{entry.benchmark.danger.p_tpk:.1%}",
|
||
f"{entry.benchmark.danger.p_any_death:.1%}",
|
||
f"{entry.benchmark.danger.expected_hp_lost:.1f}",
|
||
f"{entry.benchmark.danger.worst_decile_party_hp:.1f}",
|
||
f"{entry.benchmark.danger.expected_rounds:.1f}",
|
||
_win_rate_text(entry.benchmark, party),
|
||
)
|
||
for entry in entries
|
||
]
|
||
widths = [
|
||
max(len(header), *(len(row[index]) for row in rows))
|
||
for index, header in enumerate(headers)
|
||
]
|
||
|
||
def _line(values: Sequence[str]) -> str:
|
||
return " ".join(
|
||
value.ljust(width) for value, width in zip(values, widths, strict=True)
|
||
)
|
||
|
||
lines = [_line(headers), _line(["-" * width for width in widths])]
|
||
lines.extend(_line(row) for row in rows)
|
||
return "\n".join(lines)
|