缺口回补：上周的收盘价，这周有 79.2% 会被碰回

我有一份清洗过的 ES 期货 1 分钟 K 线 CSV（Databento 的 ES.c.0 连续合约，列有 ts_event、instrument_id、open、high、low、close，时间是 UTC）。请帮我写 Python 代码，统计“上周（昨天）的收盘价，这周（今天）有没有被碰回来”。规则如下：
1. 时间先转成美东时间（US/Eastern），自动处理夏令时。只看日盘 09:30–16:15，也就是 1 分钟 K 线从 09:30 到 16:14，不含夜盘。
2. 每天的开盘价 = 09:30 那根的开盘价，收盘价 = 16:14 那根的收盘价，最高价 / 最低价 = 日盘所有 K 线的最高 / 最低。
3. 周线：一周 = 周一到周五。这周开盘价 = 这周第一个交易日的开盘价；上周收盘价 = 上周最后一个交易日的收盘价。缺口 = 这周开盘价 − 上周收盘价，正好等于 0 的不算。
4. 碰回：从这周开盘起，只要有一天（一根 K 线）的最高价和最低价把上周收盘价夹在中间就算。向上缺口看最低价 ≤ 上周收盘价，向下缺口看最高价 ≥ 上周收盘价。再记一下是不是第一天就碰回。
5. 缺口大小 = |缺口| ÷ 过去 8 周平均每周的波幅（每周最高 − 最低），分成 0–5%、5–10%、10–15%、15–20%、20–30%、30–50%、50–100%、100% 以上 8 档。
6. 跳过：上一周数据缺失的；上周最后一天和这周第一天的 instrument_id 不同（换了合约）的。一周中间换了合约的话，换约以后的几天不看。
7. 日线：把“周”换成“天”。昨天 = 今天之前最近的一个股市交易日，用纽交所日历判断（pandas 自带的节假日规则，再加上 2012-10-29、2012-10-30、2018-12-05、2025-01-09 临时休市）；如果这个昨天不在数据里，或者两天合约不同，今天就跳过。平均波幅用过去 8 天；“一开始就碰回”改成开盘第一个小时（09:30–10:30）就碰回。
8. 打印周线和日线各自的：缺口数、当期碰回的比例、一开始就碰回的比例、向上和向下缺口分别的碰回率，以及每一档的缺口数和碰回率。
请每一步加中文注释，说明为什么这样做。

# 缺口回补：上周（昨天）的收盘价，这周（今天）有没有被碰回来
# 输入：清洗后的 ES 1 分钟 K 线（Databento ES.c.0，时间是 UTC，含 instrument_id 列）
import numpy as np
import pandas as pd
from pandas.tseries.holiday import (
    AbstractHolidayCalendar, Holiday, GoodFriday, USMartinLutherKingJr, USPresidentsDay,
    USMemorialDay, USLaborDay, USThanksgivingDay, nearest_workday, sunday_to_monday,
)


class NYSE(AbstractHolidayCalendar):  # 纽交所全天休市日
    rules = [
        Holiday("元旦", month=1, day=1, observance=sunday_to_monday),
        USMartinLutherKingJr, USPresidentsDay, GoodFriday, USMemorialDay,
        Holiday("六月节", month=6, day=19, start_date="2022-01-01", observance=nearest_workday),
        Holiday("独立日", month=7, day=4, observance=nearest_workday),
        USLaborDay, USThanksgivingDay,
        Holiday("圣诞节", month=12, day=25, observance=nearest_workday),
    ]


CLOSED = pd.to_datetime(["2012-10-29", "2012-10-30", "2018-12-05", "2025-01-09"])  # 临时休市
EDGES = [0, 5, 10, 15, 20, 30, 50, 100, np.inf]                                   # 缺口大小分档（% 平均波幅）

df = pd.read_csv("ES-1min_标准版.csv", usecols=["ts_event", "instrument_id", "open", "high", "low", "close"])
t = pd.to_datetime(df.ts_event, utc=True).dt.tz_convert("US/Eastern").dt.tz_localize(None)
hm = t.dt.hour * 100 + t.dt.minute
rth = df[(hm >= 930) & (hm < 1615)].assign(date=t.dt.normalize(), h1=hm < 1030)   # 日盘 09:30–16:15
g = rth.groupby("date")
day = g.agg(o=("open", "first"), h=("high", "max"), l=("low", "min"), c=("close", "last"), inst=("instrument_id", "first"))
first_hour = rth[rth.h1].groupby("date").agg(h=("high", "max"), l=("low", "min"))  # 开盘第一个小时 09:30–10:30


def report(name, gaps):
    gaps["档"] = pd.cut(gaps["大小"], EDGES, right=False)
    print(f"== {name}：{len(gaps)} 个缺口，当期碰回 {100 * gaps['碰回'].mean():.1f}%，"
          f"其中一开始就碰回 {100 * gaps['一开始'].mean():.1f}%")
    by_dir = gaps.groupby("向上")["碰回"].mean() * 100
    print(f"向上缺口 {by_dir[True]:.1f}%，向下缺口 {by_dir[False]:.1f}%")
    print(gaps.groupby("档", observed=True).agg(缺口=("碰回", "size"), 碰回=("碰回", "mean"), 一开始=("一开始", "mean")).round(3).to_string())


# —— 周线：这周开盘价 vs 上周收盘价 ——
day["wk"] = day.index - pd.to_timedelta(day.index.dayofweek, unit="D")   # 这周的周一
weeks = []
for wk, d in day.groupby("wk"):
    d = d[d.inst == d.inst.iloc[0]]          # 一周中间换了合约的话，换约以后的几天不看
    weeks.append(dict(wk=wk, d=d, o=d.o.iloc[0], c=d.c.iloc[-1], rng=d.h.max() - d.l.min(),
                      i0=d.inst.iloc[0], i1=d.inst.iloc[-1]))
rows = []
for i in range(8, len(weeks)):
    a, b = weeks[i - 1], weeks[i]
    if (b["wk"] - a["wk"]).days != 7 or a["i1"] != b["i0"]:
        continue                             # 上周缺失，或者两周不是同一个合约
    gap = b["o"] - a["c"]
    if gap == 0:
        continue
    abr = np.mean([w["rng"] for w in weeks[i - 8:i]])   # 过去 8 周平均每周的波幅
    hit = (b["d"].l <= a["c"]) if gap > 0 else (b["d"].h >= a["c"])   # 哪几天碰到了上周收盘价
    rows.append({"向上": gap > 0, "大小": abs(gap) / abr * 100, "碰回": hit.any(), "一开始": hit.iloc[0]})
report("周线", pd.DataFrame(rows))

# —— 日线：今天开盘价 vs 昨天收盘价（昨天 = 上一个股市交易日）——
cal = pd.bdate_range(day.index.min() - pd.Timedelta(days=10), day.index.max())
cal = cal[~cal.isin(NYSE().holidays(cal.min(), cal.max()).union(CLOSED))]
prev = pd.Series(cal[:-1], index=cal[1:]).reindex(day.index)
adr = (day.h - day.l).rolling(8).mean().shift(1)        # 过去 8 天平均每天的波幅
rows = []
for i in range(8, len(day)):
    today, yday = day.iloc[i], day.iloc[i - 1]
    if prev.iloc[i] != day.index[i - 1] or today.inst != yday.inst:
        continue                             # 昨天被清洗掉了，或者两天不是同一个合约
    gap = today.o - yday.c
    if gap == 0:
        continue
    fh = first_hour.loc[day.index[i]]
    up = gap > 0
    rows.append({"向上": up, "大小": abs(gap) / adr.iloc[i] * 100,
                 "碰回": today.l <= yday.c if up else today.h >= yday.c,
                 "一开始": fh.l <= yday.c if up else fh.h >= yday.c})
report("日线", pd.DataFrame(rows))