Files
gplan-2/qa-data/analyze.py
T
DKChan 4719c61c67 feat(board): L1 小怪板统一 10×10 + 实体名修正大地精——板尺寸策略改判批(用户拍板 2026-09-24,数值后置前收口)
- board_gen.tres:仅 (1,2) 12×12→10×10(首版误缩 boss d4 已回滚),L1 板序列 10/10/10/10/12 单调;「板随战斗时长」原则(第 6 轮)作废,改判「小怪板恒定、boss 板独大」
- enemy.tres:(1,3) 熊地精双雄→大地精(enemy_id bugbear→hobgoblin,仅本行)——DND 阶梯 地精→大地精→熊地精 与每只血量 4-8→14→50 同构;「双雄」数量词与 [2,3] 变量骰矛盾退役;hp/池/数量零改动(用户裁决只改名)
- smoke_all.gd:回放注释同步
- qa-data:calibrate 重跑(L1 五行 10×10+boss 121)、MC 修正批读数(report_U2_*)、iteration_log 第 8 批、analyze.py BASE 自定位修复
- 冒烟 537 PASS/0 FAIL;perfect L1 通过 85.7%、boss 97%/10 轮/清图 41%(12×12 行为回归);L2/L3 读数含另一会话双刀调参基线(大奖励 5/L2 收紧)非本批影响
2026-09-24 15:46:38 +08:00

352 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 冒险全流程零加成 600 局三维分析(2026-09-21 采集批 → 分析会话)
# 输入:fulllog_normal_300.jsonl / fulllog_perfect_300.jsonl(playtest_logger 全量 JSONL)
# runsum_600.jsonl(逐局摘要)
# 输出:stdout 报告 + analysis_*.csv
# 口径:
# 局切分 = run 行;战斗切分 = node(battle/boss) 开、battle_end FIFO 收(延迟发出——同 log_analysis.py 口径)
# 棋盘完成度% = Σ_各板(该板累计安全格翻开数 ÷ 该板安全格总数) × 100(跨板简单相加,用户 2026-09-21 口径)
# 安全格总数校准 = 该 (level,depth) 清图板样本的累计 safe 数中位数(清图时恰等于总数);无样本回退
# w*h − round(0.25*(w*h−9))(board_gen 表口径,生成期删雷修正会造成小幅高估)
# 承伤 = step 行 source=enemy_action × type=damage 合计;占生效 max_hp 比 = ÷ 该场首快照 hero.max_hp
# 档位映射(tier):L1 normal=d0,d1 elite=d2,d3 boss=d4;L2/L3 normal=d0,d1,d6,d7 elite=d4,d5 boss=d9
import json, csv, statistics as st, collections, sys, io, os
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import os
BASE = os.path.dirname(os.path.abspath(__file__)) + os.sep # 自定位(原硬编码旧 worktree 绝对路径已废)
RUNSUM = os.environ.get('RUNSUM', 'runsum_600.jsonl')
OUT_SUFFIX = os.environ.get('OUT_SUFFIX', '')
TIER = {
(1, 0): 'normal', (1, 1): 'normal', (1, 2): 'elite', (1, 3): 'elite', (1, 4): 'boss',
(2, 0): 'normal', (2, 1): 'normal', (2, 4): 'elite', (2, 5): 'elite',
(2, 6): 'normal', (2, 7): 'normal', (2, 9): 'boss',
(3, 0): 'normal', (3, 1): 'normal', (3, 4): 'elite', (3, 5): 'elite',
(3, 6): 'normal', (3, 7): 'normal', (3, 9): 'boss',
}
FALLBACK_SAFE = {100: 77, 144: 110, 169: 129, 196: 149} # w*h → 表口径 safe 数
def parse_log(path):
"""返回 battles 列表 + runs 列表(逐局)"""
battles = [] # 完成的战斗
open_q = [] # FIFO 未收 battle_end 的战斗
cur = None # 当前 node 上下文(battle/boss 开启)
run = None
runs = []
last_level = 0
def new_battle(node):
return {
'seed': run['seed'], 'policy': run['policy'],
'level': None, 'depth': node.get('depth'), 'node_type': node['type'],
'snaps': [], 'dmg_taken': 0, 'dmg_var': 0, 'dmg_mine': 0, 'dmg_flagmiss': 0,
'dmg_by_round': collections.Counter(),
'heal': 0, 'dmg_out': collections.Counter(), 'refreshes': 0,
'result': '', 'hero_hp_end': -1, 'gp': 0,
}
with open(path, encoding='utf-8') as fh:
for line in fh:
d = json.loads(line)
k = d.get('kind')
if k == 'run':
run = {'seed': d['seed'], 'policy': d['policy'], 'rests': []}
runs.append(run)
elif k == 'node':
if d['type'] in ('battle', 'boss'):
cur = new_battle(d)
open_q.append(cur)
else:
cur = None
if d['type'] == 'rest' and run is not None:
run.setdefault('rests', []).append({'healed': int(d.get('healed', 0)), 'level': last_level})
elif k == 'settlement' and open_q:
b = open_q[-1]
b['snaps'].append(d)
b['level'] = d.get('level')
last_level = d.get('level')
elif k == 'step':
tgt = open_q[-1] if open_q else None
if tgt is None:
continue
t, s, v = d.get('type'), d.get('source', ''), int(d.get('value', 0))
if t == 'damage':
if s == 'enemy_action':
tgt['dmg_taken'] += v
tgt['dmg_by_round'][int(d.get('round', 0))] += v
elif s in ('safe_cell', 'flag_hit', 'hero_attack', 'clear_reward'):
tgt['dmg_out'][s] += v
elif s == 'mine_hit':
tgt['dmg_mine'] += v
elif s == 'flag_miss':
tgt['dmg_flagmiss'] += v
else:
tgt['dmg_var'] += v # 地形变异(荆棘/岩浆等 display_name 作 source)
elif t == 'heal':
tgt['heal'] += v
elif k == 'action':
if open_q and d.get('op') == 'refresh':
open_q[-1]['refreshes'] += 1
elif k == 'battle_end':
if open_q:
b = open_q.pop(0) # FIFO 对账(battle_end 延迟发出)
b['result'] = d.get('result', '')
b['hero_hp_end'] = int(d.get('hero_hp', -1))
b['gp'] = int(d.get('gp_dropped', 0))
battles.append(b)
elif k == 'level_end':
pass
return battles, runs
def board_progress(snaps):
"""逐板累计口径:seq → {revealed(末值), mines(累计触雷), w, h, cleared_at(触雷修正后安全翻开)}
revealed = board.revealed(累计翻开,含触雷格);round_facts.mines = 该轮触雷数(R₀ 永不触雷,无归属歧义);
round_facts.safe 不含初始展开 R₀(首板 install 事件不入轮 facts——battle_orchestrator L121),不可作完成度分子"""
prog = {}
for s in snaps:
bd = s.get('board', {})
seq = int(bd.get('seq', 0))
rf = s.get('round_facts', {})
p = prog.setdefault(seq, {'revealed': 0, 'mines': 0, 'w': int(bd.get('w', 0)), 'h': int(bd.get('h', 0)), 'cleared_safe': -1})
p['revealed'] = max(p['revealed'], int(bd.get('revealed', 0)))
p['mines'] += int(rf.get('mines', 0))
if rf.get('path') == 'cleared':
p['cleared_safe'] = p['revealed'] - p['mines']
return prog
def battle_stats(b, safe_total_map, safe_by_wh, r0_map=None):
"""单场派生指标"""
snaps = b['snaps']
if not snaps:
return None
first = snaps[0]
lv, dp = b['level'], b['depth']
tier = TIER.get((lv, dp), '?')
rounds = max(int(s.get('round', 0)) for s in snaps)
max_hp = int(first['hero']['max_hp'])
atk = int(first['hero']['atk'])
foes = first.get('enemies', [])
enemy_hp = sum(int(e['max_hp']) for e in foes)
enemy_ids = '+'.join(sorted({e['id'] for e in foes})) or '-'
prog = board_progress(snaps)
cleared = sum(1 for p in prog.values() if p['cleared_safe'] >= 0)
mines = sum(p['mines'] for p in prog.values())
flags = sum(int(s.get('round_facts', {}).get('flag_events', 0)) for s in snaps)
paths = collections.Counter(s.get('round_facts', {}).get('path', '') for s in snaps)
# 完成度 = Σ_各板 min(安全翻开, 安全总数)/安全总数 × 100(跨板简单相加,用户口径)
comp = 0.0
dug_safe = 0.0
r0_base = (r0_map or {}).get((lv, dp), 0.0)
for seq, p in prog.items():
tot = safe_total_map.get((lv, dp, p['w'], p['h'])) or safe_by_wh.get((p['w'], p['h'])) or FALLBACK_SAFE.get(p['w'] * p['h'], max(1, p['w'] * p['h']))
rs = max(p['revealed'] - p['mines'], 0)
comp += 100.0 * min(rs, tot) / tot
dug_safe += max(rs - r0_base, 0) # 主动挖掘翻开的安全格(扣 R₀ 免费展开均值)
dmg_pct = 100.0 * b['dmg_taken'] / max(1, max_hp)
hero_out = sum(b['dmg_out'].values())
entry_hp = int(first['hero']['hp']) + b['dmg_by_round'].get(int(first.get('round', 1)), 0) # 快照反推法(首快照=首回合末)
return {
'policy': b['policy'], 'seed': b['seed'], 'level': lv, 'depth': dp, 'tier': tier,
'result': b['result'], 'rounds': rounds, 'comp': comp, 'cleared': cleared, 'dug_safe': dug_safe,
'boards': len(prog), 'dmg_taken': b['dmg_taken'], 'dmg_pct': dmg_pct,
'dmg_var': b['dmg_var'], 'dmg_mine': b['dmg_mine'], 'dmg_flagmiss': b['dmg_flagmiss'],
'heal': b['heal'], 'hero_out': hero_out, 'out_per_round': hero_out / max(1, rounds),
'dmg_out': dict(b['dmg_out']), 'mines': mines, 'flags': flags, 'refreshes': b['refreshes'],
'max_hp': max_hp, 'atk': atk, 'enemy_hp': enemy_hp, 'enemy_ids': enemy_ids,
'paths': dict(paths), 'entry_hp': entry_hp, 'hero_hp_end': b['hero_hp_end'],
}
def main():
files = sys.argv[1:] or ['fulllog_normal_300.jsonl', 'fulllog_perfect_300.jsonl']
# 第一遍:清图板校准 safe_total(清图时点 revealed−mines 恰=安全格总数;生成期删雷修正 ⇒ 表口径不可用)
all_battles = {}
for fn in files:
pol = 'perfect' if 'perfect' in fn else 'normal'
key = fn.replace('.jsonl', '')
bs, runs = parse_log(BASE + fn)
all_battles[key] = (bs, runs, pol)
calib = collections.defaultdict(list) # (level,depth,w,h) → 清图样本
calib_wh = collections.defaultdict(list) # (w,h) → 清图样本(跨深度回退池)
for key, (bs, _, _) in all_battles.items():
for b in bs:
for seq, p in board_progress(b['snaps']).items():
if p['cleared_safe'] >= 0:
calib[(b['level'], b['depth'], p['w'], p['h'])].append(p['cleared_safe'])
calib_wh[(p['w'], p['h'])].append(p['cleared_safe'])
safe_total_map = {k: int(st.median(v)) for k, v in calib.items()}
safe_by_wh = {k: int(st.median(v)) for k, v in calib_wh.items()}
# 生成器真值分母(smoke/calibrate_boards.gd 200 板/行——权威口径;日志清图校准仅作交叉核对)
gen_safe = {}
gen_r0 = {}
with open(BASE + 'board_calibration.csv', encoding='utf-8') as f:
for rec in csv.DictReader(f):
k = (int(rec['level']), int(rec['depth']))
gen_safe[(k[0], k[1], int(rec['w']), int(rec['h']))] = int(rec['safe_med'])
gen_r0[k] = (float(rec['r0_mean']), int(rec['safe_med']))
safe_total_map.update(gen_safe)
print('== safe_total 分母(生成器真值 calibrate_boards 200板/行;R₀ 免费展开基线)==')
for k in sorted(gen_r0):
r0, sm = gen_r0[k]
print(f" L{k[0]} d{k[1]}: safe={sm} R₀均={r0:.1f} ({100*r0/sm:.0f}% 免费展开)")
for k in sorted(calib):
print(f" [日志清图校准对照] L{k[0]} d{k[1]} {k[2]}x{k[3]}: safe={int(st.median(calib[k]))} (样本{len(calib[k])}, min={min(calib[k])}, max={max(calib[k])})")
rows = []
r0_map = {k: v[0] for k, v in gen_r0.items()}
for key, (bs, _, _) in all_battles.items():
for b in bs:
r = battle_stats(b, safe_total_map, safe_by_wh, r0_map)
if r:
rows.append(r)
# ---- CSV 落盘 ----
with open(BASE + 'analysis_battles' + OUT_SUFFIX + '.csv', 'w', newline='', encoding='utf-8') as f:
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
w.writeheader()
for r in rows:
w.writerow(r)
# ---- R1 局级 ----
runsums = [json.loads(l) for l in open(BASE + RUNSUM, encoding='utf-8')]
print('\n== R1 局级通关率(N=300/策略,零加成)==')
for pol in ('normal', 'perfect'):
rs = [r for r in runsums if r['policy'] == pol]
n = len(rs)
if n == 0:
continue
comp = sum(1 for r in rs if r['cleared'])
stop = collections.Counter(r['stopped_at'] for r in rs if not r['cleared'])
pass1 = n - stop.get(1, 0)
pass2 = n - stop.get(1, 0) - stop.get(2, 0)
print(f" {pol}: 通关3关 {comp}/{n} = {100*comp/n:.1f}% | 止步L1 {stop.get(1,0)} ({100*stop.get(1,0)/n:.1f}%) "
f"止步L2 {stop.get(2,0)} ({100*stop.get(2,0)/n:.1f}%) 止步L3 {stop.get(3,0)} ({100*stop.get(3,0)/n:.1f}%)")
print(f" 逐关通过率: L1 {100*pass1/n:.1f}% · L2(过L1后累计) {100*pass2/n:.1f}% · 条件通过 L1→L2 {100*pass2/max(1,pass1):.1f}% · L2→L3 {100*comp/max(1,pass2):.1f}%")
# ---- R2 战斗级 ----
def q(a, p):
a = sorted(a)
return a[min(len(a) - 1, int(p * len(a)))] if a else 0
print('\n== R2 战斗级(按 策略×关卡×档位)==')
hdr = f" {'策略':<8}{'L':>2} {'档位':<7}{'场数':>5}{'胜率':>7}{'轮数均/中':>10}{'完成度均/中/p90':>16}{'挖安全格/场':>10}{'踩雷/场':>8}{'清图≥1':>7}{'承伤%均':>8}{'输出/轮':>8}"
print(hdr)
for pol in ('normal', 'perfect'):
for lv in (1, 2, 3):
for tier in ('normal', 'elite', 'boss'):
g = [r for r in rows if r['policy'] == pol and r['level'] == lv and r['tier'] == tier]
if not g:
continue
wins = sum(1 for r in g if r['result'] == 'victory')
comps = [r['comp'] for r in g]
rds = [r['rounds'] for r in g]
dps = [r['dmg_pct'] for r in g]
clr = sum(1 for r in g if r['cleared'] >= 1)
opr = [r['out_per_round'] for r in g]
print(f" {pol:<8}{lv:>2} {tier:<7}{len(g):>5}{100*wins/len(g):>6.0f}%"
f"{st.mean(rds):>6.1f}/{q(rds,0.5):<4.0f}{st.mean(comps):>8.0f}/{q(comps,0.5):<4.0f}/{q(comps,0.9):<5.0f}"
f"{st.mean([r['dug_safe'] for r in g]):>10.1f}{st.mean([r['mines'] for r in g]):>8.2f}"
f"{100*clr/len(g):>6.0f}%{st.mean(dps):>7.1f}%{st.mean(opr):>8.1f}")
print('\n== R2b 完成度分布桶(占该组场数%)==')
buckets = [(0, 70), (70, 90), (90, 100), (100, 130), (130, 10**9)]
labels = ['<70', '70-90', '90-100', '100-130', '>130']
print(f" {'组':<24}" + ''.join(f"{l:>9}" for l in labels))
for pol in ('normal', 'perfect'):
for lv in (1, 2, 3):
for tier in ('normal', 'elite', 'boss'):
g = [r for r in rows if r['policy'] == pol and r['level'] == lv and r['tier'] == tier]
if not g:
continue
cnt = [sum(1 for r in g if lo <= r['comp'] < hi) for lo, hi in buckets]
print(f" {pol+' L'+str(lv)+' '+tier:<24}" + ''.join(f"{100*c/len(g):>8.0f}%" for c in cnt))
# ---- R3 成长数值 ----
print('\n== R3 战斗成长数值(按关卡)==')
def mean_or(g, key, fmt='{:.1f}'):
v = [r[key] for r in g]
return fmt.format(st.mean(v)) if v else '—'
for pol in ('normal', 'perfect'):
for lv in (1, 2, 3):
g = [r for r in rows if r['policy'] == pol and r['level'] == lv]
if not g:
continue
mhp = {r['max_hp'] for r in g}
atks = {r['atk'] for r in g}
parts = []
for tier in ('normal', 'elite', 'boss'):
gt = [r for r in g if r['tier'] == tier]
parts.append(f"{tier}敌HP均={mean_or(gt, 'enemy_hp')}")
print(f" {pol} L{lv}: 英雄max_hp={sorted(mhp)} atk={sorted(atks)} | " + ' '.join(parts) +
f" | 英雄输出/轮均={mean_or(g, 'out_per_round', '{:.2f}')} 敌输出/场均={mean_or(g, 'dmg_taken')}"
f" 变异伤/场均={mean_or(g, 'dmg_var')} 雷伤/场均={mean_or(g, 'dmg_mine')}")
print('\n== R3b 伤害输出构成(全战斗合计占比)==')
for pol in ('normal', 'perfect'):
tot = collections.Counter()
for r in rows:
if r['policy'] == pol:
tot.update(r['dmg_out'])
s = sum(tot.values())
print(f" {pol}: " + ' '.join(f"{k}={100*v/s:.1f}%" for k, v in tot.most_common()))
# 逐深度敌HP与轮数(成长曲线细目)
print('\n== R3c 逐深度明细(normal 策略)==')
print(f" {'L.d':<6}{'敌HP均':>7}{'轮均':>6}{'完成度均':>9}{'承伤%均':>8}{'胜率':>7}")
seen = set()
for r in rows:
seen.add((r['level'], r['depth']))
for lv, dp in sorted(seen):
g = [r for r in rows if r['policy'] == 'normal' and r['level'] == lv and r['depth'] == dp]
if not g:
continue
wins = sum(1 for r in g if r['result'] == 'victory')
print(f" {lv}.{dp:<4}{st.mean([r['enemy_hp'] for r in g]):>7.1f}{st.mean([r['rounds'] for r in g]):>6.1f}"
f"{st.mean([r['comp'] for r in g]):>9.0f}{st.mean([r['dmg_pct'] for r in g]):>7.1f}%{100*wins/len(g):>6.0f}%")
# 败北战斗画像
print('\n== R4 败北战斗画像(result=defeat 的战斗)==')
for pol in ('normal', 'perfect'):
g = [r for r in rows if r['policy'] == pol and r['result'] == 'defeat']
if not g:
continue
by = collections.Counter((r['level'], r['tier']) for r in g)
print(f" {pol}: 败北场数={len(g)} 分布=" + ' '.join(f"L{k[0]}{k[1]}×{v}" for k, v in sorted(by.items())))
ent = [r['entry_hp'] / max(1, r['max_hp']) for r in g]
print(f" 进场HP占比 均={100*st.mean(ent):.0f}% 中位={100*q(ent,0.5):.0f}% | 败北场承伤%均={st.mean([r['dmg_pct'] for r in g]):.0f}%")
# ---- R5 逐关 HP 经济(成长系统的账本面)----
print('\n== R5 逐关 HP 经济(每局每关:承伤合计 vs 治疗+休息;开局回满,关间大奖励+10上限回满)==')
bstat_by_run = collections.defaultdict(list)
for r in rows:
bstat_by_run[(r['policy'], r['seed'])].append(r)
print(f" {'策略':<9}{'L':>2} {'局数':>4} {'战斗/局':>7} {'敌伤/局':>8} {'变异+雷伤/局':>10} {'战斗治疗/局':>9} {'休息恢复/局':>9} {'净消耗/局':>9} {'期初上限':>7}")
for key, (bs, runs, pol) in all_battles.items():
for lv in (1, 2, 3):
n = 0
acc = collections.Counter()
for run in runs:
rl = [r for r in bstat_by_run.get((pol, run['seed']), []) if r['level'] == lv]
if not rl:
continue
n += 1
acc['battles'] += len(rl)
acc['edmg'] += sum(r['dmg_taken'] for r in rl)
acc['var'] += sum(r['dmg_var'] + r['dmg_mine'] + r['dmg_flagmiss'] for r in rl)
acc['heal'] += sum(r['heal'] for r in rl)
acc['rest'] += sum(x['healed'] for x in run['rests'] if x['level'] == lv)
acc['maxhp'] = rl[0]['max_hp']
if not n:
continue
print(f" {pol:<9}{lv:>2} {n:>4} {acc['battles']/n:>7.1f} {acc['edmg']/n:>8.1f} {acc['var']/n:>10.1f}"
f" {acc['heal']/n:>9.1f} {acc['rest']/n:>9.1f} {(acc['edmg']+acc['var']-acc['heal']-acc['rest'])/n:>9.1f} {acc['maxhp']:>7}")
if __name__ == '__main__':
main()