diff --git a/docs/factors/d5_runner_difference_catalogue.md b/docs/factors/d5_runner_difference_catalogue.md
index 783e6d3..dda052d 100644
--- a/docs/factors/d5_runner_difference_catalogue.md
+++ b/docs/factors/d5_runner_difference_catalogue.md
@@ -459,3 +459,124 @@ max|diff|=0.0、NaN 集合互差为 0——平移面板左端效应在真实缓
    （不变），**截面因子 ≤1,000**（实测 707 + 余量）。全局容差不动。*理由*：101 的 cap
    按 jump（bars-only）标定，对近零值多、格子多的截面 OLS 残差系统性偏紧；abs 下限
    把「rel 在近零值上失真」从 cap 压力里剥离，两处都不放宽对真回归的牙。
+
+### 七之六、C5 F1（`amp_marginal_anomaly_vol_20` 5min 残桶）——**真引擎缺陷，已修，永不作为具名类**
+
+**结论先行（给做 C5 audit 的人）**：C5 全量对账里 `amp_marginal_anomaly_vol_20` 的
+panels 腿 **729,029 格类外 finite-vs-finite**，**不是**一个待登记的差异类，而是一个
+**真实引擎缺陷**，已由独立 correctness PR（`fix/amp-marginal-residual-bucket`）消除。
+**不要为它注册任何具名类、不要为它调任何边界参数。** 修复后实测
+`unclassified=0`，且 `float_tail=0 / threshold_flip=0 / flip_contamination=0`
+——它**没有**靠任何容差类兜底。
+
+**机制**：本因子是 11 个分钟因子里**唯一派生 5min bar** 的。
+`resample_intraday_bars` 按 `ceil(bar_end, freq)` 分桶，但 emit 的是桶的**真实跨度**
+（`bar_end` = 最后一个成分的 bar_end），所以成分不齐的桶被 emit 成一根**更短的 bar**
+（残桶），而不是被丢弃。
+
+**残桶只有两个成因，且到不了同一批调用方**：
+
+1. **截断**：materializer 预截到 `available_time <= 14:50`（到 14:49 那根）⇒
+   `[14:46,14:50]` 桶只剩 4 个成分、`bar_end=14:49` 不在网格上 ⇒ **每个交易日一个**
+   ⇒ **这就是 729,029 的全部来源**。
+2. **session 内数据洞**：桶的尾部分钟缺失，同样收在网格外。**这是唯一能在不截断的情况下
+   发生的成因。**
+
+legacy / 冻结几何喂的是**整日** bar ⇒ 成因 1 **不可能**发生 ⇒ 它的暴露只剩成因 2。
+成因 2 实测为空：**全量普查**（`tmp/context/f1_residual_census_full.py`，网格直接取自
+**冻结面板自身的 (date, symbol) 对**、无抽样）——**995/995 只、1,159,263 对**
+（**恰等于 `rows_frozen`**，即普查网格与冻结面板逐格重合）、**残桶 0 个**，缓存 1min bar
+在每个 session 内网格连续。两个成因都排除 ⇒ 过滤器在 legacy 路径上是 **no-op** ⇒
+**冻结 exec 基线与 D1 冻结面板一格未动**。两方独立测得同一结果（实现方本脚本 / 评审自写
+脚本），数字逐项一致。
+
+> ⚠️ **早期版本的普查射程被夸大过，记录在案**：初版 `f1_residual_probe.py` 的 `LIMIT=600`
+> 是在**全部 5,782 个缓存 symbol 目录**上随机抽样，与 995 只评估 universe 的交集只有
+> **102 只（10.3%）**，却被写成"评估窗内 591 只"——读者会读成"995 里的 591"。**这是承重
+> 前提（legacy 路径 no-op 的论据），射程写错等于论据落空**，故改为上面的全量普查。
+>
+> ⚠️ **普查只量成因 2**（它对输入不施加任何 cutoff）。若把它读成"残桶总数为 0"，就会和
+> 729,029 直接矛盾——后者是成因 1，在一个该普查根本没看的几何里。**任何时候证据和已知
+> 事实打架，先怀疑证据在量的是别的东西。**
+>
+> **panels 腿本身已 subsume 这个普查**（995 只 × 1,191 个非 warmup 日全部卡在 1e-12）；
+> 普查的**独特价值是它把另外 19 个 warmup 日也覆盖了**——那 19 天恰是 panels 腿唯一不施加
+> 1e-12 的区域。
+
+**修法**：`factors/compute/minute/amp_marginal_anomaly_vol.py::_complete_grid_bars`，
+判据 `bar_end == bar_end.dt.ceil(freq)`（emit 的 bar_end 与桶键相等 ⟺ 窗口闭合，精确），
+在 compute 里 resample 之后立刻应用。**`resample_intraday_bars` 未改**（通用 data-layer
+原语，残桶行为由 `tests/test_intraday_aggregate.py` 直接覆盖；"完整桶"是**本因子定义**的
+性质）；**也没有在 materializer 打补丁**（那会把因子定义散到调用方，正是本缺陷的形状）。
+
+**逐格实证**（`tmp/context/f1_ab_geometry.py`，同一批真实缓存 bar，000008.SZ × 61 日，
+cache-only）：
+
+| | 整日几何（legacy） | 预截几何（materializer） |
+|---|---|---|
+| 修复前 | `0.004156551477727235` | `0.0041410493817719074` |
+| 修复后 | `0.004156551477727235` | `0.004156551477727235` |
+
+修复前 **51/61 日不同、max\|diff\| 1.42e-04**（值量级 ~4e-3，**有方向**）；修复后
+**0/61 不同**，两种几何收敛到**冻结那一侧**。
+
+**修复后真实对账（cache-only，`stk_mins_live_calls=0`，`covered=995/996`）**：
+
+- **panels rc=0**：`frozen=1159263 new=1205160 equal=1822 within_tol=1140169
+  warmup=17272 float_tail=0 threshold_flip=0 flip_contamination=0
+  nan_footprint=45897 unclassified=0 max_rel_diff=9.030e-01`；
+  warmup by direction `{nan_to_finite: 9109, finite_to_finite: 8163}`。
+- **anchors rc=0**：`5 rows, ok=4 warmup=1 failed=0`；此前 FAIL 的 3 行现在**精确
+  `rel=0.00e+00`**，剩下 1 行是登记的 `warmup_left_extension`。
+- **reports rc=0**：6 份 artifact，差异**全部落在已登记类内**
+  （`registered_addition` / `warmup_aggregate_effect` / `registered_sanity_stem_rename`
+  / `book_view_effect`），**无一处 `spec.*` 值变化**——实测 frozen 18 个 `spec.*` 叶子、
+  new 24 个，两侧都存在的 **18 个值全同**（含 `spec.version` 与 `spec.description`）、
+  0 处删除；**6 处差异是纯新增**（`spec.adjustment` / `spec.lookback_depth` /
+  `spec.overnight_boundary` / `spec.requires[0..2]`），逐条 `_is_registered_addition
+  == True`，属 D1/D5b 契约的**预登记新增，非本 PR 引入**。
+
+**不 bump `spec.version`、不加 `FactorCorrection`（已裁定）**：该字段的语义是
+「**已发布**的值被取代」。#103（jump）是**已发布 artifact 本身就脏**（旧 runner 没做
+截断）⇒ 必须承载更正；F1 相反，**已发布/冻结的那一侧是对的**（整日几何 = 定义忠实的
+路径），缺陷是重构过程中新引擎引入、**从未发布**，修复是让新引擎**回到已发布值**。
+声明一次没发生的 supersession 会往每份 artifact 里写一句假话。连带：**不动
+`spec.description`** ⇒ reports 腿**不出 `spec.*` 值变化**（仅上述 6 处预登记新增）⇒
+**不需要 jump 式注册路由**。
+
+**但确实被污染过的东西要点名**（免得读成"从来没出过错"）：C5 全量跑写出的 live
+`artifacts/reports/factor_eval_amp_marginal_anomaly_vol_20_exec_*` **是脏的**，由缺陷
+引擎产出，已被本次重跑覆盖。**冻结 exec 基线与十一因子 v0.9 表不受影响。** 缺陷的足迹
+**不小**——它触到了几乎每一个 emitted cell；秩 IC 逐日封顶会让聚合看起来动得很小，那是
+**稀释不是无害**。
+
+**verdict 重述（真跑出来的，非推定）**：`Reject / Reject` **未变**，三轴标签全未变。
+IC −0.042601 → −0.042341、ICIR −0.446484 → −0.443614、N_eff 1116.017 → 1115.866、
+NW-t −16.1546 → −16.1323、periods 1199 → 1209；增量 ICIR：`_bookclose`
+**逐位复现冻结的 −0.311985**，decision 书 −0.304787（intended book-view change）。
+⚠️ **这些微小位移不是 F1 修复造成的**——修复是把 served 值搬回冻结那一侧；位移是已登记的
+`warmup_left_extension` 聚合效应（评估日 1199→1209）。
+
+**判据"放行"了什么：09:30 集合竞价 bar —— 已知，且刻意不改**。判据判的是**网格边界那一
+分钟在不在场**，不是成分数。`ceil(09:30, 5min) = 09:30` ⇒ session 首分钟**自成一桶、恰
+1 个成分**、收在网格上 ⇒ **保留**。即**每个交易日都有一根 1 分钟 bar 被当 5min bar 池化**
+（实测 600000.SH 2023-06：20/20 天各恰有一个被保留的 <5 成分桶，且**全是 09:30**；另一条
+独立路径的交叉印证：全量普查在前 200 只上数到 **233,225** 个 (symbol, day) 对，与评审在同
+样 200 只上数到的 <5 成分"完整"桶数**同为 233,225** ⇒ 每 (symbol, 交易日) 恰一个）。
+
+这与上面"部分桶不该被池化"的理由**确实张力**，但**必须原样保留**：两种几何都保留它、
+**冻结基线也保留它** ⇒ 丢掉它会**移动已发布值**，那是**另一个 correctness 问题、要另开 PR
+带自己的证据**，不能夹带进本次修复。**后人不要当 bug 顺手"修"了。**
+
+**本过滤器的射程（写进它自己的 docstring）**：等值判据只在 `freq` 网格与 A 股 session 对齐
+时有意义。实测 `{1, 5, 15, 30}min` **一格不丢**；**`60min` 每天丢一桶**——`ceil(11:30,
+60min) = 12:00`，故 11:01–11:30 收在网格外，**整个上午收盘半小时会被每日静默丢弃**
+（实测 600000.SH 2023-06：60min coarse 100 → kept 80，丢的 `bar_end` 全是 `11:30`）。
+今日仅为**潜在**：`freq` 默认取模块常量 `AMP_ANOMALY_FREQ`（`"5min"`），两个调用点都用默认。
+
+**缺陷幅度（比初版呈现的更重）**：初版 PR 引用的例子（000008.SZ，61 日窗口，max\|diff\|
+1.42e-04）偏小。在**完整评估窗口**上重测 `600000.SH`（1,213 个日期，cache-only）：修复前
+**567/1,213 日不同**、max\|diff\| **4.14e-04**、**max rel `1.456776e-01`**；修复后
+**0/1,213 日不同**、max rel **0.0**。评审独立复测（另写脚本、用 `git show main:` 把修复
+前后两个模块载进同一进程喂逐字节相同输入）得 max rel **1.457e-01**、且**整日几何 5/5
+逐位相等**——"已发布值不移动"由此**直接证得，不靠推理**。
diff --git a/factors/compute/minute/amp_marginal_anomaly_vol.py b/factors/compute/minute/amp_marginal_anomaly_vol.py
index 4bb68c4..943a306 100644
--- a/factors/compute/minute/amp_marginal_anomaly_vol.py
+++ b/factors/compute/minute/amp_marginal_anomaly_vol.py
@@ -12,6 +12,8 @@
   1. bar frequency = 5min, DERIVED from the 1min cache via
      :func:`data.clean.intraday_aggregate.resample_intraday_bars` (a derived bar
      inherits ``available_time = max(source_1min.available_time)`` — PIT-faithful).
+     A derived bar counts only if its window actually REACHED its grid boundary
+     (``bar_end`` on the ``freq`` grid); a residual bucket is dropped whole.
   2. lookback window N = 20 trading days (the symbol's own days that have bars,
      trailing and INCLUDING ``d``).
   3. anomaly threshold = ``1{|Δamp_t| > μ + σ}`` with μ / σ = mean / ddof=1 std of
@@ -25,7 +27,9 @@
 Definition (per symbol, per panel date ``d``):
 
   1. Take the symbol's most recent ``N`` (=20) trading days INCLUDING ``d`` of 5min
-     bars. PIT truncation (standing authorization): keep only bars with
+     bars. A derived bar is kept only if it is a COMPLETE ``freq`` window — its
+     ``bar_end`` (= the last constituent 1min bar_end) sits ON the ``freq`` grid.
+     PIT truncation (standing authorization): keep only bars with
      ``available_time <= (that bar's trade_date + decision_time)`` (default 14:50),
      so every day is truncated to its own [session-open, 14:50].
   2. Per bar: ``amp = high/low - 1`` (drop a bar unless ``low > 0`` and
@@ -84,6 +88,81 @@ def _minute_requires(*fields: str) -> tuple[PanelField, ...]:
     return tuple(PanelField(f, source=STK_MINS_1MIN) for f in fields)
 
 
+def _complete_grid_bars(coarse: pd.DataFrame, freq: str) -> pd.DataFrame:
+    """Drop the RESIDUAL buckets of a derived-bar frame: keep complete windows only.
+
+    :func:`resample_intraday_bars` buckets each 1min bar by ``ceil(bar_end, freq)``
+    and then reports the bucket's REAL span, so the emitted ``bar_end`` is the last
+    constituent's ``bar_end`` — equal to the bucket's grid boundary when the window
+    closed, strictly before it when the bucket ran out of 1min bars. The test applied
+    here is that equality, and it should be read literally: it asks whether THE
+    GRID-BOUNDARY MINUTE IS PRESENT, not how many constituents the bucket has.
+
+    WHY this factor needs it (and why it lives here, not in ``resample_intraday_bars``).
+    Every statistic downstream is BAR-LENGTH SENSITIVE — ``amp = high/low - 1`` and
+    ``r = close_t/close_{t-1} - 1`` both scale with how long the bar ran — and they are
+    POOLED across a 20-day window, so one short bar per day shifts the pool's μ / σ and
+    the selected-bar return std systematically, not randomly. A 4-minute bar is not a
+    5min bar and must not be pooled with them.
+
+    WHY it is load-bearing rather than incidental. The rule is a property of the
+    FACTOR DEFINITION ("5min bars, truncated at 14:50"), and it must hold whatever
+    the CALLER hands in:
+
+    * whole-day 1min bars (the legacy runner): the 14:46..14:50 bucket is complete,
+      so it survives here and is then dropped by the availability rule below
+      (``available_time`` 14:51 > the 14:50 cutoff) — the last bar of the day is the
+      complete 14:45 bucket. The 14:45-14:50 window is genuinely UNKNOWABLE at 14:50,
+      which is why dropping it is definition-faithful rather than a loss;
+    * 1min bars ALREADY PIT-truncated by the caller (the D4 materializer truncates to
+      ``available_time <= 14:50``, i.e. through the 14:49 bar): the same bucket now
+      holds only 4 constituents and ends at 14:49, so it is a residual bucket. Without
+      this filter it would enter as a fifth "5min bar" of every single day, which is
+      precisely the load-geometry dependence this test removes;
+    * a bar frame with a mid-session data gap: a bucket whose trailing minutes are
+      missing likewise ends off-grid and is dropped. This case is the one behavioural
+      change for whole-day callers, and it is deliberate under the same argument —
+      a partial bar's amplitude and return are not comparable with a full bar's.
+
+    Those are the ONLY two causes of a residual bucket, and they do not reach the same
+    callers: TRUNCATION cannot arise from whole-day input, so a whole-day caller's only
+    exposure is the DATA GAP. Censused over the FULL evaluation grid — the frozen
+    panel's own (date, symbol) pairs, 995/995 symbols and 1,159,263 pairs, no
+    sampling — the gap case is EMPTY: the cached 1min bars are grid-contiguous within
+    each session, so on this cache the filter drops nothing for a whole-day caller.
+    Do not restate that census as "there are no residual buckets": it says nothing
+    about the truncated geometry, where there is one per trading day.
+
+    A bucket with an EARLIER minute missing but its last minute present still closes
+    on the grid and is KEPT: the test is about the window closing, not about
+    constituent count.
+
+    WHAT THIS DELIBERATELY LETS THROUGH — the OPENING AUCTION bar. ``ceil(09:30,
+    5min)`` is 09:30, so the session's first minute forms a bucket of its own holding
+    exactly ONE constituent, ends on the grid, and is kept. So one 1-minute bar is
+    pooled as a "5min bar" every single trading day (measured: on 600000.SH over
+    2023-06, 20 of 20 days have exactly one kept sub-5-constituent bucket and it is
+    always 09:30). That sits awkwardly beside the bar-length argument above, and it is
+    STILL LEFT ALONE ON PURPOSE: both geometries keep it and so does the frozen
+    baseline, so dropping it would MOVE PUBLISHED VALUES — a separate correctness
+    question needing its own PR and its own evidence, not a change to smuggle in here.
+
+    RANGE OF THIS GUARD (§ house rule: a guard states what it does not cover). The
+    equality test is only meaningful when the ``freq`` grid aligns with the A-share
+    session. Measured on real bars, ``{1, 5, 15, 30}min`` drop nothing, but ``60min``
+    drops one bucket per day: ``ceil(11:30, 60min)`` is 12:00, so the 11:01–11:30
+    half-hour ends off-grid and this filter would DISCARD THE WHOLE MORNING CLOSE
+    every day. That is latent, not live — ``freq`` defaults to the module constant
+    ``AMP_ANOMALY_FREQ`` ("5min") and both call sites use the default — but a caller
+    passing 60min would get silent daily data loss rather than an error.
+
+    Returns a fresh frame; never mutates ``coarse``.
+    """
+    bar_end = coarse["bar_end"]
+    complete = (bar_end == bar_end.dt.ceil(freq)).to_numpy()
+    return coarse.loc[complete].copy()
+
+
 def _anomaly_vol_cut(
     dabs: np.ndarray,
     ret: np.ndarray,
@@ -164,9 +243,17 @@ def compute_amp_marginal_anomaly_vol(
 
     Takes normalized 1min ``bars``, DERIVES ``freq`` (default 5min) bars from them via
     :func:`resample_intraday_bars` (so a derived bar is usable only once EVERY
-    constituent 1min bar is available), PIT-truncates them at ``decision_time``, forms
-    the within-day ``|Δamp|`` / ``r`` pairs, and returns the trailing-``lookback_days``
-    anomaly-bar return std. See the module docstring for the LOCKED definition.
+    constituent 1min bar is available), keeps only the COMPLETE ``freq`` windows
+    (:func:`_complete_grid_bars` — a residual bucket is not a ``freq`` bar and its
+    shorter amplitude / return must not be pooled with full ones), PIT-truncates them
+    at ``decision_time``, forms the within-day ``|Δamp|`` / ``r`` pairs, and returns
+    the trailing-``lookback_days`` anomaly-bar return std. See the module docstring
+    for the LOCKED definition.
+
+    The completeness filter makes the value INDEPENDENT OF THE CALLER'S LOADING
+    GEOMETRY: passing whole-day bars and passing bars the caller already truncated at
+    ``decision_time`` yield the same value, because the bucket straddling the cutoff
+    is dropped either way (as a post-cutoff complete bar, or as a residual one).
 
     Args:
         bars: normalized 1min bars (:mod:`data.clean.intraday_schema`),
@@ -201,6 +288,13 @@ def compute_amp_marginal_anomaly_vol(
     # DERIVE the coarse (5min) bars from 1min FIRST: available_time = max source, so a
     # coarse bar enters only once all its 1min constituents are available.
     coarse = resample_intraday_bars(bars, freq)
+    if len(coarse) == 0:
+        return empty_factor_series(name)
+    # Keep COMPLETE freq windows only. This is the factor's own definition rule, so
+    # it is applied to the derived bars regardless of whether the caller pre-truncated
+    # the 1min input -- see ``_complete_grid_bars`` for why a partial bar must never
+    # be pooled with full ones.
+    coarse = _complete_grid_bars(coarse, freq)
     if len(coarse) == 0:
         return empty_factor_series(name)
 
diff --git a/tests/test_amp_marginal_residual_bucket.py b/tests/test_amp_marginal_residual_bucket.py
new file mode 100644
index 0000000..99f5563
--- /dev/null
+++ b/tests/test_amp_marginal_residual_bucket.py
@@ -0,0 +1,275 @@
+"""PR-E residual-bucket correctness guard: a 4-minute bar is not a 5min bar.
+
+``amp_marginal_anomaly_vol`` is the ONE minute factor of the eleven that DERIVES
+coarser (5min) bars from the 1min cache. ``resample_intraday_bars`` buckets by
+``ceil(bar_end, freq)`` and reports each bucket's REAL span, so a bucket that runs
+out of 1min bars is emitted as a SHORTER bar (a residual bucket) rather than
+dropped. Every statistic this factor computes is bar-length sensitive
+(``amp = high/low - 1``, ``r = close_t/close_{t-1} - 1``) and they are POOLED over
+20 trading days, so a short bar shifts the pool's mean / std and the selected-bar
+return std SYSTEMATICALLY.
+
+That made the factor's value depend on the CALLER'S LOADING GEOMETRY:
+
+* the legacy runner handed in WHOLE-DAY 1min bars, so the 14:46..14:50 bucket was
+  complete and was then removed by the availability rule (its ``available_time``
+  is 14:51 > the 14:50 cutoff) — the last visible bar of each day was the complete
+  14:45 bucket;
+* the D4 materializer PRE-TRUNCATES the 1min bars to ``available_time <= 14:50``
+  (i.e. through the 14:49 bar), so the same bucket held four constituents, ended
+  at 14:49 and passed the availability rule — every single day gained a fifth
+  "5min bar" that was really four minutes long.
+
+The fix keeps only the derived bars whose window reached its ``freq`` grid
+boundary. These tests pin the three directions of that rule and, in the
+whole-day / gapless case, that the filter is a NO-OP — the legacy path's values
+do not move.
+
+Definition note: the rule is about the WINDOW CLOSING, not about counting five
+constituents. A bucket missing some of its minutes still closes on the grid as
+long as the grid-boundary minute is present, and is kept;
+``test_bucket_with_a_leading_gap_is_kept_because_its_window_still_closed`` pins
+that so a future "constituent count == 5" reading cannot creep in silently.
+
+The most visible consequence is the OPENING AUCTION: ``ceil(09:30, 5min)`` is
+09:30, so that minute forms a bucket of its own with exactly ONE constituent and
+is KEPT. That is one 1-minute bar pooled as a "5min bar" every trading day, in
+BOTH geometries and in the frozen baseline alike. It is deliberately left alone
+here -- see ``_complete_grid_bars``.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from data.clean.intraday_aggregate import resample_intraday_bars
+from data.clean.intraday_schema import normalize_intraday_bars
+from factors.compute.minute.amp_marginal_anomaly_vol import (
+    AMP_ANOMALY_FREQ,
+    _complete_grid_bars,
+    compute_amp_marginal_anomaly_vol,
+)
+from factors.view_lag import minute_decision_cutoff
+
+_SYM = "000905.SZ"
+#: small gates so a compact fixture still produces a finite value; the DEFAULT
+#: gates (460 / 20) are exercised by tests/test_amp_marginal_anomaly_vol_factor.py.
+_GATES = {"lookback_days": 20, "min_pool": 6, "min_selected": 2}
+
+
+def _rows_to_bars(rows: list[tuple]) -> pd.DataFrame:
+    """rows = [(bar_end, symbol, high, low, close), ...] -> normalized 1min bars."""
+    df = pd.DataFrame(
+        {
+            "time": pd.to_datetime([r[0] for r in rows]),
+            "symbol": [r[1] for r in rows],
+            "open": [r[4] for r in rows],
+            "high": [r[2] for r in rows],
+            "low": [r[3] for r in rows],
+            "close": [r[4] for r in rows],
+            "volume": [100.0] * len(rows),
+            "amount": [100.0] * len(rows),
+        }
+    )
+    return normalize_intraday_bars(df, freq="1min")
+
+
+def _tail_session(day: str, *, first: str = "13:31", last: str = "15:00") -> list[tuple]:
+    """One session's 1min bars over ``[first, last]``, minute by minute.
+
+    Amplitudes and closes vary bar to bar (a deterministic saw pattern seeded by
+    the day) so the pooled ``|Δamp|`` has real spread and the anomaly selection is
+    not degenerate; the exact numbers do not matter, only that they differ.
+    """
+    start = pd.Timestamp(f"{day} {first}")
+    end = pd.Timestamp(f"{day} {last}")
+    n = int((end - start) / pd.Timedelta(minutes=1)) + 1
+    seed = int(pd.Timestamp(day).dayofyear)
+    rows = []
+    for i in range(n):
+        amp = 0.002 + 0.001 * ((i * 7 + seed) % 11)
+        close = 100.0 + 0.1 * ((i * 5 + seed) % 13)
+        rows.append(
+            (start + pd.Timedelta(minutes=i), _SYM, 100.0 * (1.0 + amp), 100.0, close)
+        )
+    return rows
+
+
+def _whole_day_bars(days: list[str]) -> pd.DataFrame:
+    rows: list[tuple] = []
+    for day in days:
+        rows.extend(_tail_session(day))
+    return _rows_to_bars(rows)
+
+
+def _visible_coarse_bar_ends(bars: pd.DataFrame, day: str) -> list[pd.Timestamp]:
+    """The 5min ``bar_end``s the factor actually pools on ``day``, in order.
+
+    Mirrors the factor's own front half (derive -> completeness filter ->
+    availability rule) so a test can name the last bar of the day.
+    """
+    coarse = _complete_grid_bars(resample_intraday_bars(bars, AMP_ANOMALY_FREQ), "5min")
+    cut = minute_decision_cutoff(coarse, decision_time="14:50:00")
+    ends = pd.DatetimeIndex(cut["bar_end"])
+    return sorted(ends[ends.normalize() == pd.Timestamp(day)])
+
+
+_DAYS = ["2021-07-01", "2021-07-02", "2021-07-05"]
+
+
+# --------------------------------------------------------------------------- #
+# Direction 1 — the legacy (whole-day, gapless) path does not move
+# --------------------------------------------------------------------------- #
+def test_whole_day_gapless_bars_leave_the_completeness_filter_a_noop():
+    """On whole-day gapless input EVERY derived bucket already closes on the grid.
+
+    This is the bit-identity claim for the legacy runner path stated as a property
+    of the data rather than as a diff against deleted code: if the filter removes
+    nothing, it cannot have changed any legacy value.
+    """
+    bars = _whole_day_bars(_DAYS)
+    coarse = resample_intraday_bars(bars, AMP_ANOMALY_FREQ)
+    kept = _complete_grid_bars(coarse, "5min")
+    # PRECONDITION: the frame is non-trivial, so "nothing was dropped" is not
+    # vacuously true of an empty frame.
+    assert len(coarse) == 3 * 18  # 13:35, 13:40, ... 15:00 buckets per day
+    pd.testing.assert_frame_equal(kept, coarse)
+
+
+def test_whole_day_last_visible_bar_of_the_day_is_the_1445_bucket():
+    bars = _whole_day_bars(_DAYS)
+    for day in _DAYS:
+        ends = _visible_coarse_bar_ends(bars, day)
+        assert ends[-1] == pd.Timestamp(f"{day} 14:45"), day
+        assert ends[0] == pd.Timestamp(f"{day} 13:35"), day
+        assert len(ends) == 15, day
+
+
+# --------------------------------------------------------------------------- #
+# Direction 2 — a caller that pre-truncates gets the SAME value (the real defect)
+# --------------------------------------------------------------------------- #
+def test_pretruncated_input_drops_the_1449_residual_bucket():
+    bars = _whole_day_bars(_DAYS)
+    pre = minute_decision_cutoff(bars, decision_time="14:50:00")
+    # PRECONDITION: the pre-truncated 1min input really does end at 14:49, i.e.
+    # the fixture reproduces the materializer's geometry.
+    assert pd.DatetimeIndex(pre["bar_end"]).max() == pd.Timestamp("2021-07-05 14:49")
+    # PRECONDITION: without the completeness filter that geometry DOES produce a
+    # residual 14:49 bucket -- the thing being guarded exists in this fixture.
+    raw_coarse = resample_intraday_bars(pre, AMP_ANOMALY_FREQ)
+    raw_ends = pd.DatetimeIndex(raw_coarse["bar_end"])
+    assert pd.Timestamp("2021-07-01 14:49") in set(raw_ends)
+
+    for day in _DAYS:
+        ends = _visible_coarse_bar_ends(pre, day)
+        assert ends[-1] == pd.Timestamp(f"{day} 14:45"), day
+        assert len(ends) == 15, day
+
+
+def test_factor_value_is_independent_of_caller_pretruncation():
+    """The load-geometry invariance the fix exists to restore, asserted bitwise."""
+    bars = _whole_day_bars(_DAYS)
+    pre = minute_decision_cutoff(bars, decision_time="14:50:00")
+    whole = compute_amp_marginal_anomaly_vol(bars, **_GATES)
+    truncated = compute_amp_marginal_anomaly_vol(pre, **_GATES)
+    # PRECONDITION: a finite value exists on both sides, so equality cannot come
+    # from two all-NaN series comparing equal.
+    assert whole.notna().any()
+    pd.testing.assert_series_equal(whole, truncated)
+
+
+# --------------------------------------------------------------------------- #
+# Direction 3 — a data gap makes a residual bucket too, and it goes as well
+# --------------------------------------------------------------------------- #
+def test_missing_data_residual_bucket_is_dropped_from_a_whole_day_frame():
+    """A bucket whose trailing minutes are absent ends off-grid and is dropped.
+
+    This is the one behavioural change the fix makes for a WHOLE-DAY caller, and
+    it is deliberate: the bucket really is shorter than 5 minutes, so pooling its
+    amplitude and return with full bars is the same error the 14:49 bucket was.
+    """
+    rows = [r for r in _tail_session("2021-07-01") if r[0].strftime("%H:%M") not in
+            {"14:39", "14:40"}]
+    bars = _rows_to_bars(rows)
+    raw_ends = set(pd.DatetimeIndex(resample_intraday_bars(bars, AMP_ANOMALY_FREQ)["bar_end"]))
+    # PRECONDITION: the gap really did produce an off-grid bucket ending 14:38.
+    assert pd.Timestamp("2021-07-01 14:38") in raw_ends
+    assert pd.Timestamp("2021-07-01 14:40") not in raw_ends
+
+    kept_ends = set(
+        pd.DatetimeIndex(
+            _complete_grid_bars(
+                resample_intraday_bars(bars, AMP_ANOMALY_FREQ), "5min"
+            )["bar_end"]
+        )
+    )
+    assert pd.Timestamp("2021-07-01 14:38") not in kept_ends
+    assert pd.Timestamp("2021-07-01 14:35") in kept_ends  # untouched neighbour stays
+
+
+def test_bucket_with_a_leading_gap_is_kept_because_its_window_still_closed():
+    """A bucket missing its LEADING minutes still closes on the grid, so it is kept.
+
+    Pins that the rule tests the bucket's END, not its constituent count -- a
+    "count == 5" reading would silently drop these and diverge from the legacy
+    path on every illiquid symbol.
+
+    The gap is deliberately at the FRONT of the bucket (14:36/14:37 of the
+    14:36..14:40 window). That is what separates this rule from a span-based one:
+    a leading gap moves ``bar_start`` while leaving ``bar_end`` on the grid, so
+    ``bar_end - bar_start == freq`` would reject the bucket while the real rule
+    keeps it.
+    """
+    rows = [r for r in _tail_session("2021-07-01") if r[0].strftime("%H:%M") not in
+            {"14:36", "14:37"}]
+    bars = _rows_to_bars(rows)
+    coarse = resample_intraday_bars(bars, AMP_ANOMALY_FREQ)
+    kept = _complete_grid_bars(coarse, "5min")
+    ends = set(pd.DatetimeIndex(kept["bar_end"]))
+    # PRECONDITION: the 14:40 bucket really is short two constituents -- its
+    # bar_start (min over the constituents) has moved from 14:35 to 14:37, so this
+    # bucket genuinely has a hole rather than merely existing.
+    holed = coarse[pd.DatetimeIndex(coarse["bar_end"]) == pd.Timestamp("2021-07-01 14:40")]
+    assert len(holed) == 1
+    assert holed["bar_start"].iloc[0] == pd.Timestamp("2021-07-01 14:37")
+    assert pd.Timestamp("2021-07-01 14:40") in ends
+
+
+# --------------------------------------------------------------------------- #
+# Purity
+# --------------------------------------------------------------------------- #
+def test_complete_grid_bars_does_not_mutate_its_input():
+    bars = _whole_day_bars(["2021-07-01"])
+    coarse = resample_intraday_bars(minute_decision_cutoff(bars), AMP_ANOMALY_FREQ)
+    before = coarse.copy(deep=True)
+    out = _complete_grid_bars(coarse, "5min")
+    # PRECONDITION: this call really did drop something (else purity is vacuous).
+    assert len(out) < len(coarse)
+    pd.testing.assert_frame_equal(coarse, before)
+
+
+def test_completeness_filter_survives_an_all_residual_frame():
+    """Every bucket residual -> empty coarse frame -> honest empty factor series."""
+    rows = [
+        (pd.Timestamp("2021-07-01 09:31"), _SYM, 101.0, 100.0, 100.5),
+        (pd.Timestamp("2021-07-01 09:36"), _SYM, 101.0, 100.0, 100.5),
+    ]
+    bars = _rows_to_bars(rows)
+    coarse = resample_intraday_bars(bars, AMP_ANOMALY_FREQ)
+    assert len(coarse) == 2  # buckets 09:35 and 09:40, both ending off-grid
+    assert _complete_grid_bars(coarse, "5min").empty
+    out = compute_amp_marginal_anomaly_vol(bars, **_GATES)
+    assert out.empty
+    assert list(out.index.names) == ["date", "symbol"]
+
+
+def test_derived_bar_availability_still_comes_from_the_source_maximum():
+    """The completeness filter must not disturb the PIT provenance of what remains."""
+    bars = _whole_day_bars(["2021-07-01"])
+    kept = _complete_grid_bars(resample_intraday_bars(bars, AMP_ANOMALY_FREQ), "5min")
+    row = kept[pd.DatetimeIndex(kept["bar_end"]) == pd.Timestamp("2021-07-01 14:45")]
+    assert len(row) == 1
+    assert row["available_time"].iloc[0] == pd.Timestamp("2021-07-01 14:46")
+    assert row["bar_start"].iloc[0] == pd.Timestamp("2021-07-01 14:40")
+    assert np.isfinite(float(row["high"].iloc[0]))