From 0a0fd2f68243cd3c50f4b11c72ca02cddfe8fbab Mon Sep 17 00:00:00 2001 From: jack Date: Fri, 28 Aug 2026 06:17:47 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BD=B1=E5=AD=90=E4=BF=A1=E5=8F=B7=E8=B5=B0?= =?UTF-8?q?=E5=A2=9E=E9=87=8F=EF=BC=9A=E6=B8=85=E7=A9=BA=E5=8D=81=E5=B8=81?= =?UTF-8?q?=20560ms=20=E2=86=92=20247ms?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit compute() 每根新建 TF_DF 换成按 (symbol, timeframe) 缓存的流式对象。worker 进程被复用，所以缓存跨根存活；2 worker 轮流拿 10 币，每个 worker 最终缓存 全部 20 条流，实测内存开销落在噪声里（597→598MiB）。 ## 前提先验，否则整个改动建立在沙子上 init_stream/append_bar **没有 trim**，dataframe 靠 pd.concat 无界增长。所以 增量必然让窗口每根 +1，只能周期性重建拉回，两次重建之间窗口是 [W, W+500] 而非恒定 W。于是必须先证明 compute() 输出对窗口长度不敏感——否则增量等于 静默换掉一批信号，不报错不崩。 verify_window_sens.py：三币 75 个信号窗口，+200/+500/+1000 三档全部逐字段 一致。step39 说的是「命中率在 2000 根饱和」，饱和不等于不变，这是两回事。 ## 对拍 verify_incr_parity.py：三币 1,800 根、21 个命中、各跨 1 次重建边界，逐字段 零分歧。不能引用 HANDOFF §5.5——那验的是 bsp_list 那条链的整体哈希，而这里 是 find_fast_bsp3 那条链，且流式对象跨根复用，状态污染只会让信号悄悄换一批。 对拍顺带定论一件读代码定不了的事：cal_bi_list **不依赖** klc.trend。 init_stream/append_bar 从不调 cal_trend（它只在 get_klc_list 里），所以追加 出来的 klc 其 trend 恒为 UNKNOWN，而批量构建的有值；两者结果逐字段相同。 HANDOFF §5.5 那句「bi.py:221 读 klc.trend，笔的计算依赖它」不成立——221 行 在 cal_trend 自己的循环里，读的是它自身的序列状态。 ## 重建不走 init_stream init_stream 是逐行 dataframe.iloc[idx]，正是引擎提速刚修掉的反模式：2001 根 要 238.5ms，而批量 lean 只 74.3ms，慢 3.2 倍。第一版用它重建，10 个币启动时 各来一次，清空反而涨到 1686ms。改用 TF_DF(df, lean=True) 重建，append_bar 靠 _ensure_stream_state 就能接上。 ## 实测 append_bar 21.8ms vs 批量 lean 重建 77.7ms = 3.56x，与研究侧测的 3.7x 一致。 拆解：add_indicators 全表 7.5ms（34%，为加一根重算 2001 行）+ cal_bi_list 整表重扫 11.1ms（51%）+ concat 1.4ms。这两项都在引擎侧，值得反馈。 十币 / 2 核：清空 560→247ms，排队 92→10ms，纯计算 219→108ms。判定从 「加 worker 无用，唯一出路是增量」变成「宽裕，无需优化」。 注意 inner 108ms 里 chan 构建只占约 22ms，其余是 build_htf_zones / find_fast_bsp3 / attach_htf_context。**瓶颈已不在 chan 构建**，再压增量收益 有限。 stream_bars 落到 latency CSV：恒等于 2001 说明缺口判定在每根都回退重建、 增量静默失效，这一点从耗时上看不出是哪一环。实测窗口稳定长大。 Co-authored-by: Cursor --- research/live/shadow_hb.py | 10 +- research/live/shadow_signal.py | 70 ++++++++++++- research/live/verify_incr_parity.py | 154 ++++++++++++++++++++++++++++ research/live/verify_lean_parity.py | 9 +- research/live/verify_window_sens.py | 115 +++++++++++++++++++++ 5 files changed, 349 insertions(+), 9 deletions(-) create mode 100644 research/live/verify_incr_parity.py create mode 100644 research/live/verify_window_sens.py diff --git a/research/live/shadow_hb.py b/research/live/shadow_hb.py index b7fae66..6d9c265 100644 --- a/research/live/shadow_hb.py +++ b/research/live/shadow_hb.py @@ -363,7 +363,8 @@ class Shadow: # compute_ms 含排队；queue_ms/inner_ms 把它拆开，用来判断加核有没有用 "compute_ms", "queue_ms", "inner_ms", "n_bars", "n_hits", - "n_pass", "atr_bp", "lag_med_ms", "lag_ok"]) + "n_pass", "atr_bp", "lag_med_ms", "lag_ok", + "stream_bars"]) # 无条件漂移：每根都记，用来和信号根上的条件漂移对照 self.f_drf, self.w_drf = _writer(d / "shadow_drift.csv", [ "site", "sym", "kline_ts", "delay_label", "delay_ms", @@ -571,7 +572,7 @@ class Shadow: t0 = time.perf_counter() payload = (df_l[NUM_COLS].values.tolist(), - df_h[NUM_COLS].values.tolist(), baseline, time.time()) + df_h[NUM_COLS].values.tolist(), baseline, time.time(), sym) loop = asyncio.get_running_loop() from shadow_signal import compute_packed try: @@ -610,7 +611,10 @@ class Shadow: "queue_ms": res.get("queue_ms"), "inner_ms": res.get("inner_ms"), "n_bars": res.get("n_bars", 0), "n_hits": len(hits), "n_pass": n_pass, "atr_bp": atr_bp, - "lag_med_ms": lag_med, "lag_ok": int(lag_ok)}) + "lag_med_ms": lag_med, "lag_ok": int(lag_ok), + # 增量流当前窗口。恒等于 2001 说明缺口判定在每根都 + # 回退重建，增量静默失效——只从耗时上看不出是哪一环 + "stream_bars": res.get("stream_bars")}) self.f_lat.flush() if baseline is not None and np.isfinite(baseline): diff --git a/research/live/shadow_signal.py b/research/live/shadow_signal.py index 1c22f40..7071d81 100644 --- a/research/live/shadow_signal.py +++ b/research/live/shadow_signal.py @@ -40,10 +40,64 @@ for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"): LEAN = os.environ.get("SHADOW_LEAN", "1") not in ("0", "", "false") +INCR = os.environ.get("SHADOW_INCR", "1") not in ("0", "", "false") + +# 增量流缓存。worker 进程被复用，所以这个 dict 跨根存活。 +# 键是 (symbol, timeframe)——2 个 worker 轮流拿 10 个币，每个 worker 最终会 +# 缓存全部 10 个币，共 20 条流。 +_STREAMS: dict[tuple, tuple] = {} + +# 两次重建之间允许窗口长多少根。 +# +# init_stream/append_bar **没有 trim**：dataframe 靠 pd.concat 无界增长。所以 +# 增量必然让窗口每根 +1，只能周期性 init_stream 拉回。取 500 的两个理由： +# 1. append_bar 里 rebuild_bi_zs 要整表重扫笔，是 O(n)。窗口涨 25% 成本也涨 +# 约 25%，500/2001 正好把这个膨胀压在 25% 以内。 +# 2. 重建约 51ms、追加约 14ms，摊到 500 根上重建只加 0.07ms/根。 +# 前提「输出对窗口长度不敏感」由 verify_window_sens.py 验过（+200/+500/+1000 +# 全部逐字段一致），否则这个方案等于静默换掉一批信号。 +MAX_GROW = 500 + + +def _chan_for(key: tuple, df, tf: str, lean: bool): + """拿该窗口对应的 chan 对象，能增量就增量，否则重建。 + + 三种情况必须回退到全量重建，否则会拿一个状态不对的流去出信号： + + 缓存没有 首次见到这个币 + 窗口已长过阈值 见 MAX_GROW + 缓存末根不在新窗口 说明中间断了很多根（或时间戳回退），接不上 + + 第三种是最要紧的。2 个 worker 轮流拿 10 个币，某个 worker 可能隔几根才再 + 看到同一个币，那几根要补齐；但若缺口大到超出窗口，就没法补，只能重建。 + 不检查而直接 append 会把不连续的 K 线接在一起，笔和中枢全错且不报错。 + """ + from chanlun import TF_DF + + ts = df["timestamp"].to_numpy("int64") + st = _STREAMS.get(key) + if st is not None: + chan, last_ts, base_n = st + if len(chan.dataframe) <= base_n + MAX_GROW and last_ts >= ts[0] \ + and last_ts <= ts[-1] and (ts == last_ts).any(): + for _, row in df[df["timestamp"] > last_ts].iterrows(): + chan.append_bar(row) + _STREAMS[key] = (chan, int(ts[-1]), base_n) + return chan + + # 重建走**批量** init_TF_DF，不用 init_stream。init_stream 是逐行 + # `dataframe.iloc[idx]`，正是引擎提速刚修掉的反模式：实测 2001 根要 + # 238.5ms，而批量 lean 只要 74.3ms，慢 3.2 倍。 + # append_bar 能接在批量构建的对象上——_ensure_stream_state 会补出 + # _klc_feed_last_klu，其余列表 init_TF_DF 都建好了。 + chan = TF_DF(df.copy(), 1, tf, lean=lean) + _STREAMS[key] = (chan, int(ts[-1]), len(df)) + return chan def compute(df_l, df_h, entry_px: float | None = None, - lean: bool | None = None) -> dict: + lean: bool | None = None, sym: str | None = None, + incr: bool | None = None) -> dict: """在 df_l 的最后一根上找信号。df_l/df_h 都只含已收盘 K 线。 entry_px 是次根开盘价（回测 entry_delay=1 的成交价），用作 atr_pct 的 @@ -66,6 +120,8 @@ def compute(df_l, df_h, entry_px: float | None = None, import pandas as pd lean = LEAN if lean is None else lean + # 没有 sym 就无法给流分键，只能走全量——对拍脚本会用这条路径当基准 + incr = (INCR if incr is None else incr) and sym is not None try: from chanlun import TF_DF @@ -75,7 +131,8 @@ def compute(df_l, df_h, entry_px: float | None = None, from lib.nested_level import build_htf_zones from lib.shadow_budget import ATR_GATE_BP - chan_l = TF_DF(df_l, 1, "1m", lean=lean) + chan_l = _chan_for((sym, "1m"), df_l, "1m", lean) if incr \ + else TF_DF(df_l, 1, "1m", lean=lean) cdf = chan_l.dataframe last = len(cdf) - 1 base = {"last_idx": last, "n_bars": int(len(df_l)), "hits": [], @@ -111,7 +168,8 @@ def compute(df_l, df_h, entry_px: float | None = None, # 5m 同向。算不出时 h1_agree 记 0，该信号自然不会通过 pass_all if df_h is not None and len(df_h) > 0: - chan_h = TF_DF(df_h, 1, "5m", lean=lean) + chan_h = _chan_for((sym, "5m"), df_h, "5m", lean) if incr \ + else TF_DF(df_h, 1, "5m", lean=lean) hdf = chan_h.dataframe tl = htf_fx_timeline( signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf) @@ -180,11 +238,15 @@ def compute_packed(payload: tuple) -> dict: t_start = time.time() l_rows, h_rows, entry_px, *rest = payload t_submit = rest[0] if rest else None + sym = rest[1] if len(rest) > 1 else None t0 = time.perf_counter() df_l = _rebuild(l_rows) df_h = _rebuild(h_rows) if h_rows else None - out = compute(df_l, df_h, entry_px) + out = compute(df_l, df_h, entry_px, sym=sym) + # 落盘这两个数才能在线看出增量是否在生效：走了重建的根 grown 会等于窗口 + st = _STREAMS.get((sym, "1m")) + out["stream_bars"] = int(len(st[0].dataframe)) if st else None out["inner_ms"] = int((time.perf_counter() - t0) * 1000) # 同一台机器，父子进程时钟一致，可直接相减 out["queue_ms"] = int((t_start - t_submit) * 1000) \ diff --git a/research/live/verify_incr_parity.py b/research/live/verify_incr_parity.py new file mode 100644 index 0000000..57ab500 --- /dev/null +++ b/research/live/verify_incr_parity.py @@ -0,0 +1,154 @@ +"""逐根对拍「全量重算」与「增量追加」，并量提速。 + +## 为什么必须逐根对拍，不能引用 HANDOFF §5.5 + +§5.5 验的是 step46 那批用例（走 bsp_list 那条链），且是「追加 150~200 根 vs +全量重建」的整体哈希。影子路径不同： + + - 走 find_fast_bsp3 + build_htf_zones + htf_fx_timeline + attach_htf_context + - 流式对象**跨根复用**，而 worker 轮流拿多个币，同一条流可能隔几根才被 + 再次追加。状态污染只会让信号悄悄换一批，不报错、不崩 + +而且代码阅读已经暴露一处偏差：`init_stream/append_bar` 从不调 `cal_trend` +（它只在 `get_klc_list` 里），所以增量路径下 `klc.trend` 恒为 UNKNOWN。 +HANDOFF 说「笔的计算依赖 klc.trend」——若为真，增量的笔就和全量不同。 +那句话所引的 bi.py:221 其实在 `cal_trend` 自己的循环里，不是 `cal_bi_list` +的依赖。**这条只能由对拍来定论**，不能靠读代码。 + +## 判据 + +逐字段相同，排除 last_idx/n_bars（随窗口长度必然变，见 verify_window_sens） +与计时字段。数量相同而标志不同一样算失败。 + +模拟真实调用模式：连续推进，且每根都按「全量」和「增量」各算一次，增量那侧 +复用同一条流。 + + python research/live/verify_incr_parity.py --syms BTC,ETH,SOL --n 300 +""" +from __future__ import annotations + +import argparse +import os +import sys +import time +import warnings +from pathlib import Path + +import numpy as np +import pandas as pd + +warnings.filterwarnings("ignore") +for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"): + os.environ.setdefault(_v, "1") + +HERE = Path(__file__).resolve() +sys.path.insert(0, str(HERE.parents[1])) +sys.path.insert(0, str(HERE.parents[2])) +sys.path.insert(0, str(HERE.parent)) + +BASE_L, BASE_H = 2001, 801 + + +def run_one(sym: str, cache: Path, n: int, start_at: int | None) -> dict: + import shadow_signal as ss + from verify_lean_parity import SKIP, WINDOW_KEYS, canon, load, signal_bars + + skip = SKIP + WINDOW_KEYS + ("stream_bars",) + l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache) + l_ts = l_all["timestamp"].to_numpy("int64") + h_ts = h_all["timestamp"].to_numpy("int64") + + # 从最后一个信号根往前 n 根开始，保证这段里一定有信号分支被执行 + if start_at is None: + try: + sb = signal_bars(sym, cache) + sb = sb[(sb > BASE_L + n) & (sb < len(l_all) - 1)] + start_at = int(sb[-1]) - n + 5 if len(sb) else BASE_L + 10 + except Exception: + start_at = BASE_L + 10 + ends = [e for e in range(start_at, start_at + n) if e < len(l_all) - 1] + if not ends: + raise RuntimeError("窗口不足") + + ss._STREAMS.clear() + same = diff = 0 + t_full = t_incr = 0.0 + n_hits = 0 + first = None + rebuilds = 0 + prev_grown = 0 + + for e in ends: + hi = int(np.searchsorted(h_ts, l_ts[e], side="right")) + df_l = l_all.iloc[e - BASE_L + 1:e + 1] + df_h = h_all.iloc[max(0, hi - BASE_H):hi] + entry = float(l_all["open"].to_numpy(float)[e + 1]) + + t0 = time.perf_counter() + rf = ss.compute(df_l.copy(), df_h.copy(), entry, incr=False) + t_full += time.perf_counter() - t0 + + t0 = time.perf_counter() + ri = ss.compute(df_l.copy(), df_h.copy(), entry, sym=sym, incr=True) + t_incr += time.perf_counter() - t0 + + grown = len(ss._STREAMS[(sym, "1m")][0].dataframe) + if grown <= prev_grown: + rebuilds += 1 + prev_grown = grown + + n_hits += len(rf.get("hits") or []) + if canon(rf, skip) == canon(ri, skip): + same += 1 + else: + diff += 1 + if first is None: + first = (e, canon(rf, skip), canon(ri, skip)) + + k = len(ends) + print(f" {k} 根 · 一致 {same} · 不一致 {diff} · 命中 {n_hits} 个 · " + f"重建 {rebuilds} 次 · 末窗 {prev_grown} 根") + print(f" 单根 全量 {t_full / k * 1000:.1f}ms → " + f"增量 {t_incr / k * 1000:.1f}ms " + f"（{t_full / max(t_incr, 1e-9):.2f}x）") + if first: + e, a, b = first + print(f" ⚠ 首个分歧 idx={e}\n 全量: {a[:300]}\n 增量: {b[:300]}") + ss._STREAMS.clear() + del l_all, h_all + return {"sym": sym, "n": k, "same": same, "diff": diff, "hits": n_hits, + "full_ms": t_full / k * 1000, "incr_ms": t_incr / k * 1000} + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--syms", default="BTC,ETH,SOL") + ap.add_argument("--cache", default="research/live/cache") + ap.add_argument("--n", type=int, default=300) + ap.add_argument("--start", type=int, default=None) + a = ap.parse_args() + + rows = [] + for sym in a.syms.split(","): + print(f"\n{'=' * 70}\n{sym}") + try: + rows.append(run_one(sym, Path(a.cache), a.n, a.start)) + except Exception as e: + print(f" 跳过：{e!r}") + if not rows: + return + d = pd.DataFrame(rows) + print(f"\n\n{'=' * 70}\n汇总\n") + print(f" 对拍 {int(d['n'].sum()):,} 根 · 不一致 {int(d['diff'].sum())} · " + f"命中 {int(d['hits'].sum())} 个") + print(f" 单根 全量 {d['full_ms'].mean():.1f}ms → " + f"增量 {d['incr_ms'].mean():.1f}ms " + f"（{d['full_ms'].sum() / max(d['incr_ms'].sum(), 1e-9):.2f}x）") + if int(d["diff"].sum()) == 0: + print("\n 逐字段一致，增量可以上线。") + else: + print("\n ⛔ 有分歧，不要上线。增量流的状态与全量重建不等价。") + + +if __name__ == "__main__": + main() diff --git a/research/live/verify_lean_parity.py b/research/live/verify_lean_parity.py index 61814eb..e86f3e3 100644 --- a/research/live/verify_lean_parity.py +++ b/research/live/verify_lean_parity.py @@ -50,7 +50,12 @@ def load(sym: str, tf: str, cache: Path) -> pd.DataFrame: return pd.read_feather(c[0]) -def canon(d: dict) -> str: +# 随窗口长度必然改变的记账字段。比「窗口长度会不会改信号」时要排除它们， +# 否则一定 0/12 不一致，而那是记账字段在变，不是信号在变 +WINDOW_KEYS = ("last_idx", "n_bars") + + +def canon(d: dict, skip: tuple = SKIP) -> str: """把返回值规范化成可比较的字符串。 浮点直接比会被末位差异误判，但 lean 走的是同一段算术、不该有任何差异， @@ -68,7 +73,7 @@ def canon(d: dict) -> str: return norm(v.item()) return v return json.dumps({k: norm(v) for k, v in sorted(d.items()) - if k not in SKIP}, sort_keys=True, ensure_ascii=False) + if k not in skip}, sort_keys=True, ensure_ascii=False) def signal_bars(sym: str, cache: Path) -> np.ndarray: diff --git a/research/live/verify_window_sens.py b/research/live/verify_window_sens.py new file mode 100644 index 0000000..aa73aed --- /dev/null +++ b/research/live/verify_window_sens.py @@ -0,0 +1,115 @@ +"""compute() 的输出对窗口长度是否不变——增量路径的前提。 + +## 为什么这是增量路径的前提 + +`init_stream/append_bar` 没有 trim：`dataframe` 靠 `pd.concat` 无界增长。所以 +增量方案必然意味着**窗口会长大**（每根 +1），只能靠周期性 `init_stream` 重建 +拉回。于是在两次重建之间，实际窗口是 [W, W+slack] 而不是恒定 W。 + +这就把一个问题摆在前面：如果 `compute()` 的输出随窗口长度变化，增量路径等于 +静默把信号换了一批——不报错、不崩，只是测的不再是回测那批信号。 + +step39 的结论是「命中率在 2000 根饱和」。**饱和不等于不变**：再加根数不再提高 +命中率，与「结果逐字段相同」是两回事。所以要单独验。 + +判据是逐字段相同，不是命中数相同。数量相同而方向或滤网标志不同，会让影子测 +的是另一批信号。 + + python research/live/verify_window_sens.py --syms BTC,ETH,SOL +""" +from __future__ import annotations + +import argparse +import os +import sys +import warnings +from pathlib import Path + +import numpy as np +import pandas as pd + +warnings.filterwarnings("ignore") +for _v in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"): + os.environ.setdefault(_v, "1") + +HERE = Path(__file__).resolve() +sys.path.insert(0, str(HERE.parents[1])) +sys.path.insert(0, str(HERE.parents[2])) +sys.path.insert(0, str(HERE.parent)) + +BASE_L, BASE_H = 2001, 801 +# 增量在两次重建之间会长这么多。取 500 是因为它对应约 8 小时， +# 重建摊薄后单根成本仍接近纯追加 +GROW = (0, 200, 500, 1000) + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--syms", default="BTC,ETH,SOL") + ap.add_argument("--cache", default="research/live/cache") + ap.add_argument("--n", type=int, default=40) + a = ap.parse_args() + + from shadow_signal import compute + from verify_lean_parity import (SKIP, WINDOW_KEYS, canon, load, + signal_bars) + + # last_idx/n_bars 必然随窗口长度变。不排除的话结果一定是「全不一致」， + # 而那说明的是记账字段在变，不是信号在变 + skip = SKIP + WINDOW_KEYS + + cache = Path(a.cache) + print("同一根上，只改窗口长度，比对 compute() 的全部返回字段") + print(f"基准窗口 1m×{BASE_L} + 5m×{BASE_H}；增量会让它长大，故试 " + f"+{GROW[1:]}\n") + + tot = {g: [0, 0] for g in GROW[1:]} # [相同, 不同] + for sym in a.syms.split(","): + l_all, h_all = load(sym, "1m", cache), load(sym, "5m", cache) + l_ts = l_all["timestamp"].to_numpy("int64") + h_ts = h_all["timestamp"].to_numpy("int64") + try: + sb = signal_bars(sym, cache) + except Exception as e: + print(f"{sym} 取信号根失败：{e!r}") + continue + need = BASE_L + max(GROW) + sb = sb[(sb > need) & (sb < len(l_all) - 1)][-a.n:] + if len(sb) == 0: + print(f"{sym} 可用信号根不足") + continue + + res: dict[int, list[str]] = {g: [] for g in GROW} + for e in sb: + hi = int(np.searchsorted(h_ts, l_ts[e], side="right")) + entry = float(l_all["open"].to_numpy(float)[e + 1]) + for g in GROW: + df_l = l_all.iloc[e - (BASE_L + g) + 1:e + 1] + df_h = h_all.iloc[max(0, hi - (BASE_H + g // 5)):hi] + res[g].append(canon(compute(df_l.copy(), df_h.copy(), entry), + skip=skip)) + + print(f"{sym} {len(sb)} 根信号窗口") + for g in GROW[1:]: + same = sum(1 for x, y in zip(res[0], res[g]) if x == y) + tot[g][0] += same + tot[g][1] += len(sb) - same + print(f" +{g:>4} 根 → 一致 {same}/{len(sb)}" + + ("" if same == len(sb) else " ⚠ 有分歧")) + del l_all, h_all + + print(f"\n{'=' * 66}\n汇总\n") + for g in GROW[1:]: + s, d = tot[g] + print(f" 窗口 +{g:>4} 根：一致 {s} · 不一致 {d}") + worst = max(GROW[1:], key=lambda g: tot[g][1]) + if tot[worst][1] == 0: + print("\n 窗口长度不影响输出，增量路径的前提成立。") + print(" 可以按「长到 +N 根再 init_stream 重建」摊薄成本。") + else: + print("\n ⛔ 窗口长度会改变输出，增量路径会静默换掉一批信号。") + print(" 此时要么每根都重建（等于没有增量），要么先把窗口效应本身收口。") + + +if __name__ == "__main__": + main()