run_evolve.py

"""进化入口：跑 N 代遗传算法，输出每代最佳基因与适应度。

用法：.venv/bin/python run_evolve.py [--generations N] [--pop N] [--workers N]
"""

import argparse
import json
import os
from pathlib import Path
import time

from evolve.ga import GA
from status import EVOLVE_STATUS, write_status
from strategies.heuristic import GENES, make_default_genes


class Reporter:
    """把进化进度写成状态快照，供 monitor.py 实时展示。"""

    def __init__(self, path, args, eval_seeds):
        self.path = path
        self.started = time.time()
        self.gen_seconds = []
        self.state = {
            "kind": "evolve", "phase": "baseline",
            "generation": 0, "generations": args.generations,
            "done": 0, "total": args.pop, "history": [], "baseline": None,
            "best_genes": None, "best_detail": None,
            "default_genes": make_default_genes(),
            "gene_bounds": {n: [lo, hi] for n, _d, lo, hi in GENES},
            "config": {"pop": args.pop, "players": args.players,
                       "seeds": list(eval_seeds), "max_ticks": args.max_ticks,
                       "workers": args.workers, "prescreen": args.prescreen,
                       "live": args.live, "multistage": args.multistage,
                       "mid_ticks": args.mid_ticks,
                       "risk_lambda": args.risk_lambda,
                       "holdout_seeds": args.holdout_seeds,
                       "seed_rollover": args.seed_rollover,
                       "long_check_ticks": args.long_check_ticks},
        }
        self.flush()

    def flush(self):
        s = self.state
        s["elapsed"] = time.time() - self.started
        # ETA：已完成代的均速 × 剩余代 + 当前代剩余部分
        if self.gen_seconds:
            per_gen = sum(self.gen_seconds) / len(self.gen_seconds)
            frac = (s["done"] / s["total"]) if s["total"] else 0
            remain = per_gen * (s["generations"] - s["generation"] - frac)
            s["eta"] = max(0.0, remain)
        else:
            s["eta"] = None
        write_status(self.path, s)

    def progress(self, done, total):
        self.state["done"] = done
        self.state["total"] = total
        self.flush()

    def generation_done(self, gen, best, avg, genes, detail, seconds):
        self.gen_seconds.append(seconds)
        s = self.state
        # history 附带每代冠军的 detail（harvested/deposited/res/...）供前端画曲线
        s["history"].append({"gen": gen, "best": best, "avg": avg,
                             "detail": detail or {}})
        if s["best_genes"] is None or best >= max(h["best"] for h in s["history"]):
            s["best_genes"] = genes
            s["best_detail"] = detail
        s["generation"] = gen + 1
        self.flush()


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--generations", type=int, default=8,
                    help="0 = 无限循环（手动/收敛停止）")
    ap.add_argument("--patience", type=int, default=0,
                    help="连续 N 代 best 无提升则停止（0 = 不早停）")
    ap.add_argument("--stop-file", default="results/evolve_stop",
                    help="手动停止标记文件（无限模式下存在即优雅停止）")
    ap.add_argument("--pop", type=int, default=16)
    ap.add_argument("--workers", type=int, default=0,
                    help="并行进程数（0=自动：保留 2 个 CPU 核心）")
    ap.add_argument("--seeds", type=str, default="42,43,44,45")
    ap.add_argument("--seed-pool", type=str, default="",
                    help="滚动种子池（逗号分隔，配合 --seed-rollover；"
                         "默认 = eval_seeds，即不滚动）")
    ap.add_argument("--seed-rollover", type=int, default=0,
                    help="每 N 代从种子池滚动一批新种子（P0#13 过拟合缓解："
                         "固定种子集合会让基因只适应几张地图；滚动后代间 best "
                         "不可直接比，收敛看 holdout 分数）")
    ap.add_argument("--max-ticks", type=int, default=800)
    ap.add_argument("--players", type=int, default=8,
                    help="每局玩家数（被测基因 + 分层对手）。baseline 与进化"
                         "必须用同一个值，否则分数不可比")
    ap.add_argument("--prescreen", type=float, default=0.0,
                    help="预筛比例：先用 1 个种子跑全种群，仅前该比例的个体补齐"
                         "其余种子（0.5 约省 1.6x 时间，代价是偶尔漏掉一个好个体）")
    ap.add_argument("--risk-lambda", type=float, default=0.0,
                    help="风险调整：选择按 fitness - λ×跨种子std 排序"
                         "（>0 惩罚偶尔爆高经常崩盘的策略）")
    ap.add_argument("--holdout-seeds", type=str, default="",
                    help="独立验证种子（逗号分隔，不进选择压力）：每代冠军"
                         "用这些种子复评并记录，防记忆过拟合固定种子")
    ap.add_argument("--long-check-ticks", type=int, default=0,
                    help="每代冠军的长局复评时长（multistage 模式，固定种子）："
                         "看稳态指标（经济枯竭/侦察死区在 5k+ tick 才暴露，"
                         "进化压力内的短局看不到）；0 = 关闭")
    ap.add_argument("--long-check-seeds", type=str, default="77",
                    help="长局复评种子（默认 77，不进选择压力）")
    ap.add_argument("--live", action="store_true",
                    help="线上环境模式：老玩家带兵出生（发育差）+ 偏远环带")
    ap.add_argument("--multistage", action="store_true",
                    help="多阶段评估：0.5×短局(0→600) + "
                         "0.5×真实续局(600→600+mid-ticks)")
    ap.add_argument("--mid-ticks", type=int, default=600,
                    help="多阶段中局时长")
    ap.add_argument("--size", type=int, default=512,
                    help="live 模式世界大小（线上推断 4096+，取 768 更接近比例）")
    ap.add_argument("--out", type=str, default="evolve_results.json")
    ap.add_argument("--init-genes", default=None,
                    help="初始种群基因 JSON（warm start）：围绕该基因扰动生成"
                         "初始种群（代替默认基因——v7 解封基因是已知好起点，"
                         "省去前 5-8 代的爬升）")
    ap.add_argument("--status", default=EVOLVE_STATUS,
                    help="实时进度快照路径，供 monitor.py 读取；传空串关闭")
    args = ap.parse_args()
    if args.workers <= 0:
        args.workers = max(1, (os.cpu_count() or 2) - 2)

    eval_seeds = tuple(int(s) for s in args.seeds.split(","))
    seed_pool = tuple(int(s) for s in args.seed_pool.split(",")) \
        if args.seed_pool else eval_seeds
    os.makedirs("results", exist_ok=True)
    rep = Reporter(args.status, args, eval_seeds) if args.status else None

    ga = GA(pop_size=args.pop, workers=args.workers, eval_seeds=eval_seeds,
            max_ticks=args.max_ticks, seed=0, prescreen=args.prescreen,
            num_players=args.players, live=args.live, size=args.size,
            multistage=args.multistage, mid_ticks=args.mid_ticks,
            risk_lambda=args.risk_lambda,
            holdout_seeds=tuple(int(s) for s in args.holdout_seeds.split(","))
            if args.holdout_seeds else (),
            seed_pool=seed_pool, seed_rollover=args.seed_rollover,
            long_check_ticks=args.long_check_ticks,
            long_check_seeds=tuple(int(s) for s in
                                   args.long_check_seeds.split(",")),
            init_genes=(json.load(open(args.init_genes))
                        if args.init_genes else None),
            progress=rep.progress if rep else None)

    # 评估默认基因作为基线（必须与进化完全同配置：玩家数/种子/tick 数）
    print("=== 评估默认基因（baseline）===")
    t0 = time.time()
    base_fit, base_detail = ga.evaluate_genes_by_seed(
        make_default_genes(), eval_seeds)
    print(f"baseline fitness: {base_fit:.1f} detail: {base_detail}")
    print(f"  ({time.time()-t0:.1f}s)")
    if rep:
        rep.state["baseline"] = base_fit
        rep.state["baseline_detail"] = base_detail
        rep.state["phase"] = "evolving"
        rep.flush()

    ga.init_population()

    history = []
    stop_file = Path(args.stop_file)
    if stop_file.exists():
        stop_file.unlink()   # 启动时清旧标记
    gens = args.generations if args.generations > 0 else None
    gen = 0
    best_overall = -1e9
    best_overall_genes = None
    best_overall_gen = None
    best_overall_is_holdout = False
    stale = 0
    while gens is None or gen < gens:
        t0 = time.time()
        if rep:
            rep.state["generation"] = gen
            rep.state["done"] = 0
            rep.flush()
        best_fit, best_genes, avg = ga.evaluate(generation=gen, verbose=True)
        dt = time.time() - t0
        print(f"gen {gen}: best={best_fit:.1f} avg={avg:.1f} ({dt:.0f}s)")
        if ga.last_holdout:
            print(f"  [holdout] {ga.last_holdout[0]:.1f}")
        history.append({"gen": gen, "best": best_fit, "avg": avg,
                        "holdout": ga.last_holdout[0] if ga.last_holdout
                        else None,
                        "seeds": list(ga.eval_seeds),
                        "genes": best_genes})
        # 跨代比较/落盘/早停的分数：滚动种子模式下代间 best 不可比
        # （各代种子不同），必须用 holdout（固定种子纵向可比）；
        # 未配置 holdout 时退回代内 best（旧行为）
        # A holdout score is the deployment selection metric whenever it is
        # configured, regardless of whether seed rollover is also enabled.
        # Using training ``best`` here made the checkpoint and final JSON pick
        # a different genome than the champion printed at shutdown.
        cmp_fit = ga.last_holdout[0] if ga.last_holdout else best_fit
        # 最佳基因持续落盘（无限模式随时可取当前 best 部署）
        if cmp_fit > best_overall:
            best_overall = cmp_fit
            best_overall_genes = dict(best_genes)
            best_overall_gen = gen
            best_overall_is_holdout = ga.last_holdout is not None
            stale = 0
            _out_p = args.out if args.out.startswith("results/") \
                else os.path.join("results", args.out)
            os.makedirs(os.path.dirname(_out_p), exist_ok=True)
            with open(_out_p, "w") as f:
                json.dump({"baseline_fitness": base_fit,
                           "best": best_overall, "gen": gen,
                           "best_is_holdout": best_overall_is_holdout,
                           "best_gen": best_overall_gen,
                           "best_genes": best_overall_genes,
                           "best_metric": best_overall,
                           "history": history[-200:]}, f, indent=2,
                          ensure_ascii=False)
        else:
            stale += 1
        if rep:
            best_idx = max(range(len(ga.fitness)), key=lambda i: ga.fitness[i])
            rep.generation_done(gen, best_fit, avg, best_genes,
                                ga.last_details[best_idx] if ga.last_details else None,
                                dt)
        # 停止条件：固定代数完成 / 收敛早停 / 手动停止文件
        if gens is not None and gen >= gens - 1:
            break
        if args.patience and stale >= args.patience:
            print(f"收敛早停：连续 {stale} 代无提升")
            break
        if stop_file.exists():
            print("手动停止：检测到停止标记")
            break
        ga.next_generation()
        gen += 1
    ga.close()
    if rep:
        rep.state["phase"] = "done"
        rep.flush()

    # 输出最佳（滚动种子模式下按 holdout 选——代内 best 跨代不可比）
    if any(h.get("holdout") is not None for h in history):
        final_best = max(history, key=lambda h: h.get("holdout") or -1e9)
        print("(按 holdout 分数选择冠军)")
    else:
        final_best = max(history, key=lambda h: h["best"])
    print("\n=== 最佳基因 ===")
    for name, val in final_best["genes"].items():
        print(f"  {name}: {val:.3f}")
    out_path = args.out if args.out.startswith("results/") \
        else os.path.join("results", args.out)
    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    selection_metric = ("holdout" if any(h.get("holdout") is not None
                                         for h in history) else "best")
    final_metric = final_best.get(selection_metric, final_best.get("best"))
    with open(out_path, "w") as f:
        json.dump({"schema_version": 2,
                   "rules_version": "v0.14",
                   "sdk_version": "0.2.9",
                   "baseline_fitness": base_fit,
                   "best": final_metric,
                   "best_gen": final_best["gen"],
                   "best_genes": final_best["genes"],
                   "best_metric": final_metric,
                   "selection_metric": selection_metric,
                   "best_is_holdout": selection_metric == "holdout",
                   "history": history},
                  f, indent=2, ensure_ascii=False)
    print(f"结果已保存: {out_path}")


if __name__ == "__main__":
    main()