"""振幅过滤门:回看窗内 range% 须 ≤ 上限(可关)。永期开仓第一关 / 期期共用。""" from __future__ import annotations from typing import Any from ..exchange.candles import AmplitudeHL def evaluate_amplitude_gate( *, filter_enabled: bool, amp: AmplitudeHL | None, max_pct: float, hours: float, ) -> dict[str, Any]: """ 返回: blocked: 过滤开启且未过关(无K线或超限) ok: 展示用是否过关(过滤关视为 True) reason: 拒开文案(未拒则为 None) snapshot: 写入 session._oo_amp 的字典(amp 为 None 时仅含元数据) """ max_p = float(max_pct) hrs = float(hours) filt = bool(filter_enabled) if amp is None: snap = { "high": None, "low": None, "mid": None, "range_pct": None, "hours": hrs, "max_pct": max_p, "filter_enabled": filt, "ok": False if filt else True, } if filt: return { "blocked": True, "ok": False, "reason": f"振幅未过关:无法获取近 {hrs:g}h K 线高低", "snapshot": snap, } return {"blocked": False, "ok": True, "reason": None, "snapshot": snap} range_pct = float(amp.range_pct) over = range_pct > max_p + 1e-12 ok = (not filt) or (not over) snap = { "high": float(amp.high), "low": float(amp.low), "mid": float(amp.mid), "range_pct": range_pct, "hours": hrs, "max_pct": max_p, "filter_enabled": filt, "ok": ok, } if filt and over: return { "blocked": True, "ok": False, "reason": ( f"振幅未过关:{hrs:g}h 内 {range_pct:.2f}% > {max_p:g}%" ), "snapshot": snap, } return {"blocked": False, "ok": ok, "reason": None, "snapshot": snap}