jackyu66gitandCursor
6d93975e67
用户追问"你用的是长时间周期分析的？"。原答案是 1m / 18.5 个月 / 3941 笔， 而上一轮的结论是"没效果"——没效果最怕样本不够，这个追问戳到了点子上。 改用 step41 那份 5m/15m/30m 复核：2019-09 → 2026-08，2525 天，8168 笔。 口径能对上——keep = (h1_agree==1) & push 就是深色过滤，出场配置 s2_so8_k2_m48 正是实盘那套 2/3/8/2/48（cfg_name 的分批命名，scale_at 默认 3）。 唯一差异是没加 ATR≥8bp 门控，而 §3.5 实测它在这些周期上几乎不触发 （30m 0%、15m 0.16%、5m 2.6%）。 ① 时段坐实无效。段间毛R极差的置换 p：5m 0.6185、15m 0.6687、30m 0.3105、 合并 0.6445，全部远离显著。1m 那个 p=0.0895 现在看清楚了，就是零分布里运气 略好的一条尾巴。 ② 周末效应没复现，而且符号翻转。1m 上周末−工作日是 -0.132（p=0.0246）， 长样本上 5m +0.014、15m +0.142、30m -0.006、合并 +0.041；周日从 -0.199 （p=0.0079）变成 +0.033。这不是严格的样本外复现——1m 与 5m 是不同的信号 总体——但若"周末流动性薄所以吃亏"是真的市场结构效应，它没有理由只在 1m 上 出现、在 5m 上还反号。 所以上一提交里"周日效应统计上真实"那句要撤回。本步一共跑了约 35 个分组比较 （24 小时 + 两套时段定义 + 7 个星期 + 周末/周日），冒出一个 p=0.008 恰是多重 比较的期望产物。HANDOFF 里已把该结论标为撤回并记下两条教训：报告"无效"之前 先确认样本量撑得起这个"无"；在几十个分组里挑出的最显著那个，默认它是噪声， 除非能在另一个总体上复现。 实践结论不变且更硬：不加任何时间维度的过滤，mom60≥7 仍是唯一值得上的开关。 复核走 --long，复用 step41 已有 feather，未重跑采集。 Co-authored-by: Cursor <cursoragent@cursor.com>
432 lines
19 KiB
Python
432 lines
19 KiB
Python
"""Step 54：开仓时刻落在哪个时段——亚盘 / 欧盘 / 美盘谁更赚。
|
||
|
||
用户问的是时段，但这个问题有两个必须先堵的坑，否则很容易得出一个假的结论：
|
||
|
||
① ATR 混淆。亚盘波动天然低，而低 ATR 的信号因固定成本吃亏是已知结论
|
||
（HANDOFF §3.5：余量 = 净收益 / taker名义额，ATR 越小分母越小）。
|
||
所以"亚盘差"完全可能只是"亚盘 ATR 低"换个说法。必须在 ATR 分层内部再看。
|
||
② mom60 混淆。step53 刚定论 mom60 是强因子（Q4 最差），而美盘开盘那几个
|
||
小时最容易出现已经走完一大段的行情。不控的话时段会借它的力。
|
||
|
||
所以本步的顺序是：先逐小时看（不设边界，边界是人定的、最容易带出想要的结论），
|
||
再聚合成时段，最后在 ATR 与 mom60 的分层内部复核。
|
||
|
||
样本内外沿用 step53 的切法（IS_START），任何只在一边成立的都不认。
|
||
"""
|
||
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().parent
|
||
sys.path.insert(0, str(HERE))
|
||
sys.path.insert(0, str(HERE.parent))
|
||
pd.set_option("display.width", 340)
|
||
|
||
SL = 2.0
|
||
GATE_BP = 8.0
|
||
SRC = HERE / "out" / "step53_pre_entry.feather"
|
||
IS_START = pd.Timestamp("2026-01-30", tz="Asia/Shanghai")
|
||
|
||
# 时段用 UTC 定义。三段等分是加密市场的通行切法，且不重叠——重叠定义会让
|
||
# 同一笔进两个桶，比较就没有意义了。真实的开盘时刻（伦敦 08:00、纽约 13:30）
|
||
# 落在段内而非段首，所以另有一张逐小时表兜底，防止边界把结论切出来
|
||
SESSIONS = [("亚盘", 0, 8), ("欧盘", 8, 16), ("美盘", 16, 24)]
|
||
# 稳健性对照：按真实开盘时刻切，并把欧美重叠那段单列
|
||
SESSIONS_ALT = [("亚盘", 0, 7), ("欧盘", 7, 13), ("欧美重叠", 13, 17),
|
||
("美盘", 17, 21), ("淡时段", 21, 24)]
|
||
|
||
|
||
def stat(g: pd.DataFrame, lab: str, denom: int) -> dict:
|
||
if len(g) < 40:
|
||
return {"分组": lab, "笔数": len(g), "备注": "样本不足"}
|
||
w, o = g.net[g.net > 0].sum(), -g.net[g.net <= 0].sum()
|
||
return {"分组": lab, "笔数": len(g), "占比": f"{len(g)/denom*100:.0f}%",
|
||
"胜率": f"{(g.net > 0).mean()*100:.1f}%",
|
||
"毛R": round(g.gR.mean(), 3), "净均R": round(g.R.mean(), 3),
|
||
"PF": round(w / o, 2) if o > 0 else np.inf,
|
||
"余量bp": round(g.net.mean() / g.tn.mean() * 1e4, 2),
|
||
"中位ATRbp": round(g.atr_bp.median(), 1)}
|
||
|
||
|
||
def label_session(h: pd.Series, table) -> pd.Series:
|
||
out = pd.Series("?", index=h.index, dtype=object)
|
||
for name, a, b in table:
|
||
out[(h >= a) & (h < b)] = name
|
||
return out
|
||
|
||
|
||
def hourly(d: pd.DataFrame) -> None:
|
||
"""逐小时。先看这个再谈时段——时段边界是人定的，逐小时不是。"""
|
||
print("\n" + "=" * 100)
|
||
print("########## 一、逐小时（UTC），不设时段边界 ##########")
|
||
rows = []
|
||
for h, g in d.groupby("utc_h"):
|
||
oos, ins = g[g.date < IS_START], g[g.date >= IS_START]
|
||
rows.append({
|
||
"UTC时": h, "北京时": (h + 8) % 24, "笔数": len(g),
|
||
"毛R": round(g.gR.mean(), 3),
|
||
"净均R": round(g.R.mean(), 3),
|
||
"余量bp": round(g.net.mean() / g.tn.mean() * 1e4, 2),
|
||
"中位ATRbp": round(g.atr_bp.median(), 1),
|
||
"样本外毛R": round(oos.gR.mean(), 3) if len(oos) >= 30 else None,
|
||
"发现期毛R": round(ins.gR.mean(), 3) if len(ins) >= 30 else None,
|
||
})
|
||
t = pd.DataFrame(rows)
|
||
print(t.to_string(index=False))
|
||
print(f"\n每小时平均只有 {len(d)/24:.0f} 笔，单个小时的数是噪声，"
|
||
f"看形状不要看单点。")
|
||
|
||
|
||
def sessions(d: pd.DataFrame, table, title: str) -> None:
|
||
print("\n" + "=" * 100)
|
||
print(f"########## {title} ##########")
|
||
d = d.copy()
|
||
d["seg"] = label_session(d.utc_h, table)
|
||
order = [n for n, _, _ in table]
|
||
for lab, part in (("全样本", d), ("样本外", d[d.date < IS_START]),
|
||
("发现期", d[d.date >= IS_START])):
|
||
rows = [stat(part[part.seg == n], n, len(part)) for n in order]
|
||
print(f"\n--- {lab}（{len(part)} 笔）---")
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
|
||
def control(d: pd.DataFrame, col: str, name: str, table) -> None:
|
||
"""在混淆变量的高/低两半内部各看一次时段。
|
||
|
||
时段若只是 ATR（或 mom60）的代理，分层后段间差异会塌掉。
|
||
"""
|
||
x = d[d[col].notna() & np.isfinite(d[col])].copy()
|
||
x["seg"] = label_session(x.utc_h, table)
|
||
x["层"] = np.where(x[col] >= x[col].median(), f"{name}高", f"{name}低")
|
||
order = [n for n, _, _ in table]
|
||
rows = []
|
||
for lay, g in x.groupby("层"):
|
||
r = {"控制层": lay, "笔数": len(g)}
|
||
for n in order:
|
||
s = g[g.seg == n]
|
||
r[n] = round(s.gR.mean(), 3) if len(s) >= 40 else None
|
||
vals = [r[n] for n in order if r[n] is not None]
|
||
r["极差"] = round(max(vals) - min(vals), 3) if len(vals) > 1 else None
|
||
rows.append(r)
|
||
print(f"\n--- 控 {name}（{col}）后的段间毛R ---")
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
|
||
def permutation(d: pd.DataFrame, table, n_iter: int = 20_000,
|
||
seed: int = 0) -> None:
|
||
"""段间极差有没有超出随机分组的水平。
|
||
|
||
24 个小时聚成 3 段，本来就会因为噪声产生一定的段间差异。不做这一步就没法
|
||
区分"时段有效"与"任意切三份都能切出这么大的差"。
|
||
"""
|
||
rng = np.random.default_rng(seed)
|
||
seg = label_session(d.utc_h, table).to_numpy()
|
||
g = d.gR.to_numpy()
|
||
order = [n for n, _, _ in table]
|
||
obs_means = np.array([g[seg == n].mean() for n in order])
|
||
obs = obs_means.max() - obs_means.min()
|
||
cnt = 0
|
||
for _ in range(n_iter):
|
||
p = rng.permutation(seg)
|
||
m = np.array([g[p == n].mean() for n in order])
|
||
if m.max() - m.min() >= obs:
|
||
cnt += 1
|
||
print(f"\n置换检验：实测段间毛R极差 {obs:.3f}，"
|
||
f"随机打乱 {n_iter} 次里有 {cnt/n_iter*100:.2f}% 达到或超过它 "
|
||
f"→ p = {cnt/n_iter:.4f}")
|
||
|
||
|
||
NAMES = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
|
||
|
||
|
||
def weekday(d: pd.DataFrame) -> None:
|
||
print("\n" + "=" * 100)
|
||
print("########## 四、星期几（按 UTC）##########")
|
||
rows = []
|
||
for k, g in d.groupby("utc_dow"):
|
||
oos, ins = g[g.date < IS_START], g[g.date >= IS_START]
|
||
r = stat(g, NAMES[k], len(d))
|
||
# 样本内外必须并排看。时段那一节正是靠这一列拆穿"美盘最好"的
|
||
r["样本外毛R"] = round(oos.gR.mean(), 3) if len(oos) >= 30 else None
|
||
r["发现期毛R"] = round(ins.gR.mean(), 3) if len(ins) >= 30 else None
|
||
rows.append(r)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
print("\n--- 工作日 vs 周末 ---")
|
||
for lab, part in (("全样本", d), ("样本外", d[d.date < IS_START]),
|
||
("发现期", d[d.date >= IS_START])):
|
||
we = part.utc_dow >= 5
|
||
t = pd.DataFrame([stat(part[~we], "工作日", len(part)),
|
||
stat(part[we], "周末", len(part))])
|
||
print(f"\n{lab}（{len(part)} 笔）")
|
||
print(t.to_string(index=False))
|
||
|
||
|
||
def weekend_controls(d: pd.DataFrame) -> None:
|
||
"""周末效应是不是 ATR / mom60 / 币种 的代理。
|
||
|
||
周末 ATR 中位比工作日低 1bp，而低 ATR 吃亏是已知的，所以必须分层复核。
|
||
"""
|
||
print("\n" + "=" * 100)
|
||
print("########## 五、周末效应的混淆检查 ##########")
|
||
x = d.copy()
|
||
x["周末"] = np.where(x.utc_dow >= 5, "周末", "工作日")
|
||
for col, name in (("atr_bp", "ATR"), ("mom60", "前60根动量"),
|
||
("vpre10", "前10根量")):
|
||
g = x[x[col].notna() & np.isfinite(x[col])].copy()
|
||
g["层"] = np.where(g[col] >= g[col].median(), f"{name}高", f"{name}低")
|
||
rows = []
|
||
for lay, s in g.groupby("层"):
|
||
wd, we = s[s.周末 == "工作日"], s[s.周末 == "周末"]
|
||
if min(len(wd), len(we)) < 40:
|
||
continue
|
||
rows.append({"控制层": lay, "工作日毛R": round(wd.gR.mean(), 3),
|
||
"周末毛R": round(we.gR.mean(), 3),
|
||
"差": round(we.gR.mean() - wd.gR.mean(), 3),
|
||
"工作日余量": round(wd.net.mean()/wd.tn.mean()*1e4, 2),
|
||
"周末余量": round(we.net.mean()/we.tn.mean()*1e4, 2)})
|
||
print(f"\n--- 控 {name} ---")
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
print("\n--- 逐币：周末差是普遍的还是少数币带的 ---")
|
||
rows = []
|
||
for sym, s in x.groupby("sym"):
|
||
wd, we = s[s.周末 == "工作日"], s[s.周末 == "周末"]
|
||
if min(len(wd), len(we)) < 25:
|
||
continue
|
||
rows.append({"币": sym, "工作日笔数": len(wd), "周末笔数": len(we),
|
||
"工作日毛R": round(wd.gR.mean(), 3),
|
||
"周末毛R": round(we.gR.mean(), 3),
|
||
"差": round(we.gR.mean() - wd.gR.mean(), 3)})
|
||
t = pd.DataFrame(rows).sort_values("差")
|
||
print(t.to_string(index=False))
|
||
neg = (t["差"] < 0).sum()
|
||
print(f"\n{neg}/{len(t)} 个币周末更差")
|
||
|
||
|
||
def overlap_with_mom60(d: pd.DataFrame) -> None:
|
||
"""周末的劣势是不是已经被 step53 的 `mom60≥7` 过滤吃掉了。
|
||
|
||
这是决定"要不要再加一条时间过滤"的关键：两个过滤若砍的是同一批单子，
|
||
叠加只会白丢笔数。控制表已有暗示——mom60 低的那层周末差只有 -0.069，
|
||
高的那层 -0.169。
|
||
"""
|
||
print("\n" + "=" * 100)
|
||
print("########## 六、周末效应与 mom60 过滤的重叠 ##########")
|
||
x = d.copy()
|
||
x["周末"] = np.where(x.utc_dow >= 5, "周末", "工作日")
|
||
x["周日"] = np.where(x.utc_dow == 6, "周日", "其余")
|
||
|
||
print("\n--- 高动量（mom60≥7，step53 要砍的那批）在各组里的占比 ---")
|
||
for col in ("周末", "周日"):
|
||
t = x.groupby(col).apply(
|
||
lambda g: pd.Series({
|
||
"笔数": len(g),
|
||
"mom60≥7占比": f"{(g.mom60 >= 7).mean()*100:.1f}%",
|
||
"中位mom60": round(g.mom60.median(), 2)}))
|
||
print(t.to_string())
|
||
|
||
print("\n--- 施加 mom60<7 之后，周末差还剩多少 ---")
|
||
rows = []
|
||
for lab, sub in (("过滤前（全部）", x),
|
||
("过滤后（mom60<7）", x[x.mom60 < 7])):
|
||
for col in ("周末", "周日"):
|
||
a = sub[sub[col] == sub[col].unique()[0]]
|
||
hi = sub[sub[col].isin(["周末", "周日"])]
|
||
lo = sub[sub[col].isin(["工作日", "其余"])]
|
||
if min(len(hi), len(lo)) < 40:
|
||
continue
|
||
rows.append({
|
||
"口径": lab, "对比": f"{col} vs 其余",
|
||
"差组笔数": len(hi), "差组毛R": round(hi.gR.mean(), 3),
|
||
"对照毛R": round(lo.gR.mean(), 3),
|
||
"毛R差": round(hi.gR.mean() - lo.gR.mean(), 3),
|
||
"差组余量": round(hi.net.mean()/hi.tn.mean()*1e4, 2),
|
||
"对照余量": round(lo.net.mean()/lo.tn.mean()*1e4, 2)})
|
||
print(pd.DataFrame(rows).drop_duplicates().to_string(index=False))
|
||
|
||
|
||
def perm_binary(d: pd.DataFrame, mask: np.ndarray, lab: str,
|
||
n_iter: int = 20_000, seed: int = 0) -> None:
|
||
"""两组均值差的置换检验。"""
|
||
rng = np.random.default_rng(seed)
|
||
g = d.gR.to_numpy()
|
||
obs = g[mask].mean() - g[~mask].mean()
|
||
n = int(mask.sum())
|
||
cnt = 0
|
||
for _ in range(n_iter):
|
||
idx = rng.permutation(len(g))[:n]
|
||
m = np.zeros(len(g), bool)
|
||
m[idx] = True
|
||
if abs(g[m].mean() - g[~m].mean()) >= abs(obs):
|
||
cnt += 1
|
||
print(f"\n置换检验（{lab}）：实测毛R差 {obs:+.3f}，"
|
||
f"随机分组 {n_iter} 次里 {cnt/n_iter*100:.2f}% 达到或超过 "
|
||
f"→ p = {cnt/n_iter:.4f}")
|
||
|
||
|
||
def decision(d: pd.DataFrame, cuts: dict) -> None:
|
||
"""若砍掉某些时段，在不同真实滑点下的总R。
|
||
|
||
沿用 step53 的决策表口径：砍掉一批信号既省成本也丢收益，哪边大取决于
|
||
真实滑点，所以必须按滑点扫一遍，不能只报一个数。
|
||
"""
|
||
print("\n" + "=" * 100)
|
||
print("########## 七、把时段做成过滤的决策表（发现期总R）##########")
|
||
ins = d[d.date >= IS_START]
|
||
slips = [0, 5, 8, 10, 12, 15, 18, 20]
|
||
rows = []
|
||
for lab, mask in cuts.items():
|
||
m = mask(ins)
|
||
sub = ins[m]
|
||
r = {"方案": lab, "保留": f"{m.mean()*100:.0f}%"}
|
||
for s in slips:
|
||
# 滑点只打在 taker 腿上，口径与 lib.exit_model 一致
|
||
net = sub.net - sub.tn * s / 1e4
|
||
r[f"{s}bp"] = round((net / (SL * sub.atr_pct)).sum(), 0)
|
||
rows.append(r)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
|
||
SRC41 = HERE / "out" / "step41_exit_tp.feather"
|
||
# step41 的分批出场配置名，与实盘完全一致：SL 2 / 减半 3（scale_at 默认）/
|
||
# runner 8 / runner 止损 2 / 48 根。见 lib.exit_model.cfg_name
|
||
CFG41 = "s2_so8_k2_m48"
|
||
|
||
|
||
def long_history() -> None:
|
||
"""用 5m/15m/30m 的 7 年数据复核时段与星期几。
|
||
|
||
为什么要这一步：1m 只能追到 18.5 个月（3941 笔），而时段结论是"没效果"，
|
||
**没效果最怕的就是样本不够**。step41 那份跨 2019-09 → 2026-08，5m 单周期
|
||
就有 5434 笔，且过滤口径相同（`h1_agree==1 & push` 就是深色），出场配置
|
||
也能对上实盘那套。若长样本上仍然测不出时段效应，"无效"才站得住。
|
||
|
||
⚠️ 口径差一处：这里没加 ATR≥8bp 门控。§3.5 实测该门控在这些周期上几乎
|
||
不触发（30m 0%、15m 0.16%、5m 2.6%），所以影响可忽略，但要记着。
|
||
"""
|
||
from lib.exit_model import fee_of, taker_notional
|
||
|
||
d = pd.read_feather(SRC41)
|
||
d["date"] = pd.to_datetime(d["date"])
|
||
g, r, c = (d[f"{CFG41}_{k}"].to_numpy() for k in ("g", "r", "c"))
|
||
d["net"] = g - fee_of(r, c)
|
||
d["gR"] = g / (SL * d.atr_pct.values)
|
||
d["R"] = d.net / (SL * d.atr_pct.values)
|
||
d["tn"] = taker_notional(r, c)
|
||
d["atr_bp"] = d.atr_pct * 1e4
|
||
utc = d.date.dt.tz_convert("UTC")
|
||
d["utc_h"], d["utc_dow"] = utc.dt.hour, utc.dt.dayofweek
|
||
|
||
print("\n" + "#" * 100)
|
||
print("########## 长样本复核：5m/15m/30m × 7 年 ##########")
|
||
print(f"{len(d)} 笔 · {d.symbol.nunique()} 币 · "
|
||
f"{d.date.min():%Y-%m} → {d.date.max():%Y-%m}"
|
||
f"（{(d.date.max()-d.date.min()).days} 天）")
|
||
print(f"出场配置 {CFG41}（= 实盘的 2/3/8/2/48）· 深色已过滤 · 无 ATR 门控")
|
||
|
||
for tf in ("5m", "15m", "30m"):
|
||
x = d[d.ltf == tf].copy()
|
||
if len(x) < 300:
|
||
continue
|
||
x["seg"] = label_session(x.utc_h, SESSIONS)
|
||
print(f"\n--- {tf}（{len(x)} 笔）时段 ---")
|
||
rows = []
|
||
for n, _, _ in SESSIONS:
|
||
s = x[x.seg == n]
|
||
ins, oos = s[s.group == "样本内"], s[s.group == "样本外"]
|
||
row = stat(s, n, len(x))
|
||
row["样本内毛R"] = round(ins.gR.mean(), 3) if len(ins) >= 30 else None
|
||
row["样本外毛R"] = round(oos.gR.mean(), 3) if len(oos) >= 30 else None
|
||
rows.append(row)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
permutation(x, SESSIONS, n_iter=10_000)
|
||
|
||
we = (x.utc_dow >= 5).to_numpy()
|
||
print(f"\n--- {tf} 工作日 vs 周末 ---")
|
||
print(pd.DataFrame([stat(x[~we], "工作日", len(x)),
|
||
stat(x[we], "周末", len(x))]).to_string(index=False))
|
||
perm_binary(x, we, f"{tf} 周末 vs 工作日", n_iter=10_000)
|
||
|
||
print("\n--- 三周期合并（每笔等权）---")
|
||
d["seg"] = label_session(d.utc_h, SESSIONS)
|
||
print(pd.DataFrame([stat(d[d.seg == n], n, len(d))
|
||
for n, _, _ in SESSIONS]).to_string(index=False))
|
||
permutation(d, SESSIONS, n_iter=10_000)
|
||
we = (d.utc_dow >= 5).to_numpy()
|
||
print(pd.DataFrame([stat(d[~we], "工作日", len(d)),
|
||
stat(d[we], "周末", len(d))]).to_string(index=False))
|
||
perm_binary(d, we, "合并 周末 vs 工作日", n_iter=10_000)
|
||
perm_binary(d, (d.utc_dow == 6).to_numpy(), "合并 周日 vs 其余",
|
||
n_iter=10_000)
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--src", default=str(SRC))
|
||
ap.add_argument("--long", action="store_true",
|
||
help="只跑 5m/15m/30m 的 7 年长样本复核")
|
||
args = ap.parse_args()
|
||
|
||
if args.long:
|
||
long_history()
|
||
return
|
||
|
||
from step50_volume import prep
|
||
|
||
d = pd.read_feather(args.src)
|
||
d["date"] = pd.to_datetime(d["date"])
|
||
d = prep(d[(d.htf == 1.0) & d.lad & (d.atr_bp >= GATE_BP)].copy())
|
||
utc = d.date.dt.tz_convert("UTC")
|
||
d["utc_h"] = utc.dt.hour
|
||
d["utc_dow"] = utc.dt.dayofweek
|
||
|
||
span = (d.date.max() - d.date.min()).days
|
||
print(f"实盘口径 {len(d)} 笔 · {d.sym.nunique()} 币 · 跨 {span} 天 "
|
||
f"({d.date.min():%Y-%m-%d} → {d.date.max():%Y-%m-%d})")
|
||
print(f"样本外 {(d.date < IS_START).sum()} · "
|
||
f"发现期 {(d.date >= IS_START).sum()}")
|
||
print("⚠️ 持仓最长 48 分钟，跨段的笔按**入场时刻**归属——那是唯一可操作的口径")
|
||
|
||
hourly(d)
|
||
sessions(d, SESSIONS, "二、三段等分（UTC 0/8/16）")
|
||
permutation(d, SESSIONS)
|
||
sessions(d, SESSIONS_ALT, "三、稳健性对照：按真实开盘时刻切五段")
|
||
|
||
print("\n" + "=" * 100)
|
||
print("########## 混淆检查：时段是不是 ATR / mom60 的代理 ##########")
|
||
for col, name in (("atr_bp", "ATR"), ("mom60", "前60根动量"),
|
||
("vpre10", "前10根量")):
|
||
control(d, col, name, SESSIONS)
|
||
|
||
weekday(d)
|
||
perm_binary(d, (d.utc_dow >= 5).to_numpy(), "周末 vs 工作日")
|
||
perm_binary(d, (d.utc_dow == 6).to_numpy(), "周日 vs 其余")
|
||
weekend_controls(d)
|
||
overlap_with_mom60(d)
|
||
|
||
decision(d, {
|
||
"等权（现状）": lambda x: np.ones(len(x), bool),
|
||
"砍周日": lambda x: (x.utc_dow != 6).to_numpy(),
|
||
"砍周末": lambda x: (x.utc_dow < 5).to_numpy(),
|
||
"砍欧盘": lambda x: ((x.utc_h < 8) | (x.utc_h >= 16)).to_numpy(),
|
||
"砍 mom60≥7（step53 基准）": lambda x: (x.mom60 < 7).to_numpy(),
|
||
"砍 mom60≥7 + 周日": lambda x: ((x.mom60 < 7) &
|
||
(x.utc_dow != 6)).to_numpy(),
|
||
})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|