104 lines
3.6 KiB
Python
104 lines
3.6 KiB
Python
"""策略代码版本快照（§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()
|