13.1 曲线的异常点定位

#!/usr/bin/env python3
# file: sim/nm_13_1_anomaly_localize.py
# tier: S
# verify: /opt/homebrew/bin/python3 sim/nm_13_1_anomaly_localize.py
"""卷三 13.1 曲线的异常点定位（S 层）：把「曲线偏了」落到具体的节拍下标与场景。

前几章的度量都是正向的：参数 → 曲线。本章把度量倒过来用，第一步要回答的是——
**一条曲线偏离目标时，偏在哪个节拍上？** 做法是把观测曲线与目标曲线逐项相减得到
残差向量，再对每个非零残差项报出：路径下标、规格下标、节拍 id、所属场景、所属幕、
观测值、目标值、增量 Δ。定位不靠眼睛看曲线，靠残差向量。

三个漂移场景（**都不改真源**，只在内存里造一条「改动后的标注」）：
  A) b07 4→3             —— 在路径上：路径口径与全曲线口径都能定位
  B) b02 2→3 且 b07 4→3  —— 两个异常点：单点归因失效，必须多点定位
  C) b10 5→4             —— 不在路径上：路径口径**看不见**，只有全曲线口径能定位

C 是本节最有用的一条：**可观测性决定了可识别性**——玩家路径没经过的节拍，
在路径曲线上永远是零残差，异常只能从全曲线口径里读出来。

推进复用卷一 sim/na_02_digest.py（只读 import，不改动它）。
口径（proxy）：张力曲线与 MAE 均为结构量，只做相对比较，不声称绝对情绪强度。
全部定点整数，不出现浮点。产出 out/nm_13_1_anomaly_localize.json。
"""
from __future__ import annotations

import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
from na_02_digest import cond_ok, init_state, node_by_id, apply_choice  # noqa: E402

ROOT = os.path.dirname(HERE)
OUT = os.path.join(ROOT, "out")


def load_spec() -> dict:
    with open(os.path.join(ROOT, "config/story_spec.json"), encoding="utf-8") as fh:
        return json.load(fh)


def walk(spec: dict) -> dict:
    """确定性策略：每步选第一个条件成立的选项。同 nm_00 / nm_05_2。"""
    s = init_state(spec)
    for _ in range(64):
        cur = node_by_id(spec, s["node"])
        if cur.get("ending"):
            break
        opts = [c for c in (cur.get("choices") or []) if cond_ok(c.get("when"), s)]
        if not opts:
            break
        s = apply_choice(spec, s, opts[0]["id"])
    return s


def mae_milli(obs: list, tgt: list) -> int:
    return sum(abs(a - b) for a, b in zip(obs, tgt)) * 1000 // len(obs)


def localize(obs: list, tgt: list, idx: list, beats: list) -> list:
    """残差向量 → 异常点清单：路径下标 / 规格下标 / 节拍 / 场景 / 幕 / Δ。"""
    rows = []
    for pos, (o, t) in enumerate(zip(obs, tgt)):
        if o == t:
            continue
        b = beats[idx[pos]]
        rows.append({"pos": pos, "spec_idx": idx[pos], "beat": b["id"], "scene": b["scene"],
                     "act": b["act"], "obs": o, "tgt": t, "delta": o - t})
    return rows


def single_point(rows: list) -> str:
    """单点归因：残差只在 1 个位置非零时异常点唯一；否则单点假设不成立。"""
    if not rows:
        return "无异常"
    if len(rows) == 1:
        return f"唯一（{rows[0]['beat']}）"
    return f"不成立（{len(rows)} 个非零点，需多点定位）"


def fmt(rows: list) -> str:
    return "  ".join(f"pos{r['pos']}/idx{r['spec_idx']} {r['beat']} {r['scene']} act{r['act']} "
                     f"obs{r['obs']} tgt{r['tgt']} Δ{r['delta']:+d}" for r in rows)


def main() -> int:
    spec = load_spec()
    beats = spec["beats"]
    m = spec["metrics"]
    base = [b["tension"] for b in beats]
    tgt_full = list(m["target_tension_curve"])

    s = walk(spec)
    visited = {node_by_id(spec, nid)["scene"] for nid in s["path"]}
    idx = [i for i, b in enumerate(beats) if b["scene"] in visited]
    base_path = [base[i] for i in idx]
    tgt_path = [tgt_full[i] for i in idx]

    print(f"起点 {spec['meta']['start']}  路径节点 {len(s['path'])}  终局 {s['node']}")
    print(f"路径节拍 {len(idx)} / {len(beats)}  {' '.join(beats[i]['id'] for i in idx)}")
    print(f"路径曲线 {' '.join(str(x) for x in base_path)}")
    print(f"目标曲线 {' '.join(str(x) for x in tgt_path)}")
    print(f"基线定位 {len(localize(base_path, tgt_path, idx, beats))} 个异常点；"
          f"路径 MAE {mae_milli(base_path, tgt_path)} 毫，"
          f"全曲线 MAE {mae_milli(base, tgt_full)} 毫（当前规格自洽）")

    scenarios = [
        ("A", "b07 4→3（在路径上）", {6: 3}),
        ("B", "b02 2→3 且 b07 4→3（两个异常点）", {1: 3, 6: 3}),
        ("C", "b10 5→4（不在路径上）", {9: 4}),
    ]
    report = {"path_nodes": len(s["path"]), "final_node": s["node"],
              "beats_on_path": [beats[i]["id"] for i in idx],
              "base_path_curve": base_path, "target_path_curve": tgt_path,
              "base_path_mae_milli": mae_milli(base_path, tgt_path),
              "base_full_mae_milli": mae_milli(base, tgt_full), "scenarios": []}

    for tag, desc, drift in scenarios:
        obs_full = list(base)
        for i, v in drift.items():
            obs_full[i] = v
        obs_path = [obs_full[i] for i in idx]
        rows_full = localize(obs_full, tgt_full, list(range(len(beats))), beats)
        rows_path = localize(obs_path, tgt_path, idx, beats)
        mae_full, mae_path = mae_milli(obs_full, tgt_full), mae_milli(obs_path, tgt_path)

        print("")
        print(f"场景 {tag}：{desc}")
        print(f"  路径曲线   {' '.join(str(x) for x in obs_path)}")
        print(f"  路径异常   {len(rows_path)} 个：{fmt(rows_path) if rows_path else '（无——路径口径看不见）'}")
        print(f"  路径 MAE   {mae_path} 毫（{len(idx)} 拍）")
        print(f"  全曲线异常 {len(rows_full)} 个：{fmt(rows_full)}")
        print(f"  全曲线 MAE {mae_full} 毫（{len(beats)} 拍）")
        print(f"  单点归因   路径 {single_point(rows_path)} / 全曲线 {single_point(rows_full)}")

        report["scenarios"].append({
            "tag": tag, "desc": desc, "drift": {str(k): v for k, v in drift.items()},
            "path_curve": obs_path, "path_anomalies": rows_path, "path_mae_milli": mae_path,
            "full_anomalies": rows_full, "full_mae_milli": mae_full,
            "path_single_point": single_point(rows_path),
            "full_single_point": single_point(rows_full)})

    assert all(1 <= v <= 5 for v in base), "张力越界"
    assert report["base_path_mae_milli"] == 0 and report["base_full_mae_milli"] == 0

    os.makedirs(OUT, exist_ok=True)
    with open(os.path.join(OUT, "nm_13_1_anomaly_localize.json"), "w", encoding="utf-8") as fh:
        json.dump(report, fh, ensure_ascii=False, indent=2)
        fh.write("\n")
    print("")
    print("wrote out/nm_13_1_anomaly_localize.json")
    print("OK nm_13_1_anomaly_localize：3 个漂移场景已定位，离路径异常需全曲线口径")
    return 0


if __name__ == "__main__":
    sys.exit(main())

起点 d_pier_1  路径节点 26  终局 e_lit
路径节拍 11 / 16  b01 b02 b04 b05 b06 b07 b09 b11 b12 b15 b16
路径曲线 1 2 3 2 3 4 4 3 4 4 2
目标曲线 1 2 3 2 3 4 4 3 4 4 2
基线定位 0 个异常点；路径 MAE 0 毫，全曲线 MAE 0 毫（当前规格自洽）

场景 A：b07 4→3（在路径上）
  路径曲线   1 2 3 2 3 3 4 3 4 4 2
  路径异常   1 个：pos5/idx6 b07 tide_cave act2 obs3 tgt4 Δ-1
  路径 MAE   90 毫（11 拍）
  全曲线异常 1 个：pos6/idx6 b07 tide_cave act2 obs3 tgt4 Δ-1
  全曲线 MAE 62 毫（16 拍）
  单点归因   路径 唯一（b07） / 全曲线 唯一（b07）

场景 B：b02 2→3 且 b07 4→3（两个异常点）
  路径曲线   1 3 3 2 3 3 4 3 4 4 2
  路径异常   2 个：pos1/idx1 b02 tavern act1 obs3 tgt2 Δ+1  pos5/idx6 b07 tide_cave act2 obs3 tgt4 Δ-1
  路径 MAE   181 毫（11 拍）
  全曲线异常 2 个：pos1/idx1 b02 tavern act1 obs3 tgt2 Δ+1  pos6/idx6 b07 tide_cave act2 obs3 tgt4 Δ-1
  全曲线 MAE 125 毫（16 拍）
  单点归因   路径 不成立（2 个非零点，需多点定位） / 全曲线 不成立（2 个非零点，需多点定位）

场景 C：b10 5→4（不在路径上）
  路径曲线   1 2 3 2 3 4 4 3 4 4 2
  路径异常   0 个：（无——路径口径看不见）
  路径 MAE   0 毫（11 拍）
  全曲线异常 1 个：pos9/idx9 b10 drowned_bell act2 obs4 tgt5 Δ-1
  全曲线 MAE 62 毫（16 拍）
  单点归因   路径 无异常 / 全曲线 唯一（b10）

wrote out/nm_13_1_anomaly_localize.json
OK nm_13_1_anomaly_localize：3 个漂移场景已定位，离路径异常需全曲线口径