feat(strategy): 代码版本快照(§12.6补,方案B发起时快照·QuantConnect轻量版)——用户拍板:解决「实例参数有快照但代码没有,历史运行无法回溯当时跑的哪版代码」——①sanguo_api/code_versions.py:发起时全文落盘data/strategy_code_versions/{file}.{hash8}.py(md5内容寻址天然去重,原子替换防并发坏);发起四入口(paper/live/CTA/组合回测)全快照,账户加code_hash列(ALTER迁移),回测经spec→run_meta.code_hash落档案②查看:GET /strategy/code-versions列表+单版本全文;代码编辑页「版本历史」抽屉(MonacoDiff左右对比当前,主题同款)③「代码已变更」角标:策略库档案行(enriched code_changed)+全景每run标v哈希·一致/已变更④双轨对账加「代码版本一致」第五指标(对账FAIL先查这行,两边代码不同价差必然大);路径穿越防护(版本号只认8位hex)+6测试;927绿+build绿;「用当时代码重跑」留P2 [vps]

sanguo_api/code_versions.py

@@ -0,0 +1,103 @@
+"""策略代码版本快照（§12.6 补：发起时快照，QuantConnect 轻量版）。
+
+痛点：实例参数有发起快照，策略代码没有——历史运行无法回溯「当时跑的哪版代码」。
+方案：发起回测/模拟/实盘那一刻，把当时代码全文按内容寻址落盘
+（data/strategy_code_versions/{file}.{hash8}.py，同内容只存一份天然去重），
+运行记录只存 8 位哈希；「代码已变更」= 运行时哈希 ≠ 当前文件哈希。
+"""
+from __future__ import annotations
+
+import hashlib
+import os
+import re
+import time
+
+# 测试用 monkeypatch 改 _DIR
+_DIR: str = os.environ.get(
+    "SANGUO_CODE_VERSIONS",
+    os.path.join(
+        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+        "data", "strategy_code_versions",
+    ),
+)
+_H8 = re.compile(r"^[0-9a-f]{8}$")  # 防路径穿越：版本号只认 8 位十六进制
+
+
+def _version_path(code_file: str, h8: str) -> str:
+    return os.path.join(_DIR, f"{code_file}.{h8}.py")
+
+
+def snapshot_code(code_file: str) -> dict | None:
+    """读当前代码 → md5 → 未存过则落盘快照。
+
+    返回 {code_file, code_hash, code_version(8位), saved_at}；文件不存在返回 None。
+    """
+    from .strategy_registry import read_strategy_file
+
+    if not code_file:
+        return None
+    try:
+        src = read_strategy_file(code_file)["code"]
+    except Exception:
+        return None
+    full = hashlib.md5(src.encode("utf-8")).hexdigest()
+    h8 = full[:8]
+    os.makedirs(_DIR, exist_ok=True)
+    path = _version_path(code_file, h8)
+    if not os.path.exists(path):
+        tmp = path + ".tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            f.write(src)
+        os.replace(tmp, path)  # 原子替换防并发写坏
+    return {"code_file": code_file, "code_hash": full,
+            "code_version": h8, "saved_at": time.strftime("%Y-%m-%d %H:%M:%S")}
+
+
+def current_hash(code_file: str) -> str | None:
+    """当前文件 8 位哈希（只算不存）。文件不存在返回 None。"""
+    from .strategy_registry import read_strategy_file
+
+    try:
+        src = read_strategy_file(code_file)["code"]
+    except Exception:
+        return None
+    return hashlib.md5(src.encode("utf-8")).hexdigest()[:8]
+
+
+def code_changed(code_file: str, code_hash: str | None) -> bool | None:
+    """运行时哈希 vs 当前文件。哈希缺失（早期账户/未解析文件）返回 None=不可判。"""
+    if not code_hash:
+        return None
+    cur = current_hash(code_file)
+    return None if cur is None else code_hash[:8] != cur
+
+
+def list_versions(code_file: str) -> list[dict]:
+    """该文件全部快照（新→旧）：{code_version, saved_at(文件mtime), size}。"""
+    if not os.path.isdir(_DIR):
+        return []
+    prefix = f"{code_file}."
+    out = []
+    for fn in os.listdir(_DIR):
+        if not (fn.startswith(prefix) and fn.endswith(".py")):
+            continue
+        h8 = fn[len(prefix):-3]
+        if not _H8.match(h8):
+            continue
+        st = os.stat(os.path.join(_DIR, fn))
+        out.append({"code_version": h8,
+                    "saved_at": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(st.st_mtime)),
+                    "size": st.st_size})
+    out.sort(key=lambda v: v["saved_at"], reverse=True)
+    return out
+
+
+def read_version(code_file: str, h8: str) -> str | None:
+    """读某版本全文。版本号非法（路径穿越尝试）或不存在返回 None。"""
+    if not _H8.match(h8 or ""):
+        return None
+    path = _version_path(code_file, h8)
+    if not os.path.exists(path):
+        return None
+    with open(path, encoding="utf-8") as f:
+        return f.read()