mirror of
https://ghfast.top/https://github.com/aeroxw/easy_tdx_max.git
synced 2026-09-12 23:54:21 +08:00
Justin Gu
a69fa4103a
缓存键此前用数组对象 id 做签名，而引擎每次 run 会重建数组对象， id 逐网格点漂移导致跨点永不命中——ParamGridOptimizer 的指标缓存 全程 0 命中，两段式加速自 v1.25 引入起就没生效过。正确性不受影响 （缓存未命中即走重算路径，结果与无缓存逐位一致），属纯性能回归； test_optimizer_cache_reuse_across_grid_points 因此一直在失败。 - IndicatorCache._atom 数组签名由 (id, shape) 改为 (dtype, shape, blake2b 内容摘要 16 字节)：同值不同对象视为同一 数据；NaN 按字节参与哈希，位模式不同只多算一次、不会误命中 - 删除防 id 复用的 _array_refs 强引用表（内容寻址下不再需要， 缓存不再延长数组生命周期） - 新增缓存级回归测试：同内容不同对象必须命中；原失败测试转绿 - 实测 16 点网格命中率 38%（修复前恒为 0）；全量 pytest 1613 通过、ruff/mypy 全绿
99 lines
3.8 KiB
Python
99 lines
3.8 KiB
Python
"""指标计算缓存（两段式寻优加速，v1.25 新增）。
|
|
|
|
借鉴 backtest-system ``register_two_stage`` 的思路：把**指标计算**（只依赖
|
|
数据 + 指标参数）与**信号组合 + 订单模拟**解耦。网格寻优中同一份 K 线被
|
|
所有网格点共用，而指标只依赖 ``(指标函数, 数据数组, 指标参数)``——
|
|
例如 ``{"fast": [5,10,20], "slow": [10,20,30]}`` 的 9 个点里，
|
|
``MA(close, 5)`` 会被计算 3 次（与每个 slow 组合各一次），实际只需 1 次。
|
|
|
|
:key 设计：``(函数限定名, 参数原子序列)``。数组参数用 ``(dtype, shape,
|
|
内容哈希)`` 做签名——引擎每次 run 会重建数组对象（对象 id 不稳定），
|
|
必须按内容寻址才能跨网格点命中；标量参数直接 repr。哈希用
|
|
blake2b（16 字节摘要），对回测级数组（KB 量级）开销可忽略。跨进程不
|
|
共享（进程池并行模式下各 worker 各自建缓存）。
|
|
|
|
收益上限取决于指标层在回测耗时中的占比；引擎的逐 bar Python 循环无法
|
|
通用缓存，故大网格另配 ``ParamGridOptimizer(workers=N)`` 进程级并行，
|
|
两者叠加使用。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
__all__ = ["IndicatorCache"]
|
|
|
|
|
|
class IndicatorCache:
|
|
"""跨回测运行的指标计算缓存（线程内使用，非线程安全）。"""
|
|
|
|
def __init__(self) -> None:
|
|
self._store: dict[tuple[Any, ...], Any] = {}
|
|
self.hits = 0
|
|
self.misses = 0
|
|
|
|
@property
|
|
def total(self) -> int:
|
|
"""总请求次数。"""
|
|
return self.hits + self.misses
|
|
|
|
def get_or_compute(
|
|
self,
|
|
func: Callable[..., Any],
|
|
args: tuple[Any, ...],
|
|
kwargs: dict[str, Any],
|
|
) -> Any:
|
|
"""命中则返回缓存，未命中则计算并缓存。"""
|
|
key = self._make_key(func, args, kwargs)
|
|
if key in self._store:
|
|
self.hits += 1
|
|
return self._store[key]
|
|
self.misses += 1
|
|
result = func(*args, **kwargs)
|
|
self._store[key] = result
|
|
return result
|
|
|
|
def stats(self) -> dict[str, int | float]:
|
|
"""缓存统计（命中率诊断用）。"""
|
|
total = self.total
|
|
return {
|
|
"hits": self.hits,
|
|
"misses": self.misses,
|
|
"total": total,
|
|
"hit_rate": round(self.hits / total, 4) if total else 0.0,
|
|
}
|
|
|
|
# ── 内部 ─────────────────────────────────────────────────────────────────
|
|
|
|
def _make_key(
|
|
self,
|
|
func: Callable[..., Any],
|
|
args: tuple[Any, ...],
|
|
kwargs: dict[str, Any],
|
|
) -> tuple[Any, ...]:
|
|
parts: list[Any] = [
|
|
getattr(func, "__module__", ""),
|
|
getattr(func, "__qualname__", str(func)),
|
|
]
|
|
parts.extend(self._atom(a) for a in args)
|
|
for k in sorted(kwargs):
|
|
parts.append(k)
|
|
parts.append(self._atom(kwargs[k]))
|
|
return tuple(parts)
|
|
|
|
def _atom(self, a: Any) -> Any:
|
|
"""把单个参数转为可哈希原子。"""
|
|
if isinstance(a, np.ndarray):
|
|
# 内容寻址：同值不同对象必须视为同一数据（引擎每次 run 重建数组，
|
|
# 对象 id 不稳定）。NaN 按字节参与哈希，位模式不同只多算一次，
|
|
# 不会误命中。
|
|
digest = hashlib.blake2b(a.tobytes(), digest_size=16).hexdigest()
|
|
return ("arr", str(a.dtype), a.shape, digest)
|
|
if isinstance(a, int | float | str | bool | None):
|
|
return ("s", type(a).__name__, repr(a))
|
|
return ("o", type(a).__name__, repr(a))
|