feat(phase2): optimizer and scenario compiler (P3)
This commit is contained in:
@@ -0,0 +1,318 @@
|
||||
"""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)
|
||||
Reference in New Issue
Block a user