开仓前那段：量要小、势要有但不能过头，换掉 vr60 那个开关
用户提出看开仓之前的量与趋势方向。这是第三个位置——step50 管信号根、 step52 管持仓中，这里管入场前。step48 采集端补 vpre10/vpre30/mom10/mom60。 先纠正一个错的心智模型（我和用户都以为的）：B4/S4 不是回抽后进场。mom10 中位 +3.47 ATR，为负的只占 6%。信号触发时价格在前 10 根里已经顺着你的方向 走了三个多 ATR，2 ATR 止损是架在一段已经走完的行情后面。这也解释了 step50： 信号根是突破根，放量 = 追在最后一棒上。 ① 入场前的量单调，越小越好（样本外毛R，Q1→Q4）：1.474 / 1.266 / 1.156 / 0.573。控 vr60 后仍成立（低量层 −0.284、高量层 −0.541），与 vr60 相关只有 +0.301，不是同一件事换个说法。 ② 入场前的动量是驼峰形，不是单调（mom60 样本外毛R，Q1→Q4）：1.252 / 1.473 / 1.141 / 0.603。势要有——完全没动过的 Q1 也不如 Q2；但不能过头—— Q4 在发现期余量只剩 1.44bp，等于不能做。 驼峰形意味着中位数二分法会把它测没：控制表里 mom10 的毛R差是 +0.044， 看着无效，那是二分把 Q1+Q2 和 Q3+Q4 各自平均了。对非单调因子不要用中位数 分层做检验。 ③ 砍 mom60≥7 全面优于 §3.39 定的砍 vr60≥4（发现期）： 等权 保留 100% 盈亏平衡 13.5bp R夏普 0.325 回撤 16.8 总R 596.6 砍 vr60≥4 保留 69% 盈亏平衡 15.6bp R夏普 0.411 回撤 10.8 总R 514.9 砍 mom60≥7 保留 78% 盈亏平衡 17.5bp R夏普 0.476 回撤 7.7 总R 657.7 每一项都赢，还多留 9 个点的笔数。更要紧的是总R 比不砍还高——被砍掉那 22% 期望为负，砍掉不是花钱买稳健，是纯赚。样本外同向。所以这个开关不需要等 影子测量：它在 0~20bp 每一档都不输等权。 附滑点决策表：5~12bp 区间砍 mom60≥7 通吃，只有 ≥15bp 才该上「三个都砍」， 而那时策略本身已在生死线上。 阈值 7 和 1.5 是贴着 Q4 边界取的整数，不是搜出来的，但也不是完全无关于数据 （看过分位表才取的整），上线前应确认阈值附近没有断崖敏感。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
@@ -0,0 +1,169 @@
|
||||
"""Step 53：开仓**之前**那段的量与走势方向。
|
||||
|
||||
用户提出的第三个位置。前两个已有定论：
|
||||
step50 信号根放量 → 差（那根你是买方，付的是资金已推到的价）
|
||||
step52 持仓中放量 → 好（那是资金来接你的货）
|
||||
本步问：入场之前那几根呢？资金是不是已经在里面了。
|
||||
|
||||
这跟 B4/S4 的结构直接相关。它是回抽后转强，所以入场前那几根**通常逆着你走**
|
||||
（mom10 多为负，负得多 = 回抽深）。那么「回抽时缩量」与「回抽时放量」是有
|
||||
明确含义的区分——前者是没人卖，后者是真有人在卖。
|
||||
|
||||
四个因子，全部在信号根收盘时可知：
|
||||
vpre10 / vpre30 前 10 / 30 根的相对量均值（已 shift(1)，不含信号根）
|
||||
mom10 / mom60 顺方向动量，ATR 为单位
|
||||
|
||||
⚠️ 必须控 `vr60`（信号根自身的量）。step50 已证明它是强负因子，不控的话
|
||||
前段量会通过相关性借它的力，看着有效其实是同一件事。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import os
|
||||
import sys
|
||||
import warnings
|
||||
from concurrent.futures import ProcessPoolExecutor, as_completed
|
||||
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
|
||||
OUT = HERE / "out" / "step53_pre_entry.feather"
|
||||
IS_START = pd.Timestamp("2026-01-30", tz="Asia/Shanghai")
|
||||
FACTORS = [("vpre10", "前10根量"), ("vpre30", "前30根量"),
|
||||
("mom10", "前10根顺向动量"), ("mom60", "前60根顺向动量")]
|
||||
|
||||
|
||||
def collect(sym: str, rows: int):
|
||||
import warnings as _w
|
||||
_w.filterwarnings("ignore")
|
||||
sys.path.insert(0, str(HERE))
|
||||
sys.path.insert(0, str(HERE.parent))
|
||||
from step48_signal_timing import collect as _c
|
||||
return _c(sym, rows)
|
||||
|
||||
|
||||
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)}
|
||||
|
||||
|
||||
def quartiles(d: pd.DataFrame, col: str, name: str, label: str) -> None:
|
||||
x = d[d[col].notna() & np.isfinite(d[col])]
|
||||
if len(x) < 300:
|
||||
print(f" {name}: 样本不足")
|
||||
return
|
||||
x = x.copy()
|
||||
x["bin"] = pd.qcut(x[col], 4, labels=["Q1最低", "Q2", "Q3", "Q4最高"])
|
||||
rows = [stat(g, str(b), len(x)) for b, g in x.groupby("bin", observed=True)]
|
||||
t = pd.DataFrame(rows)
|
||||
med = x.groupby("bin", observed=True)[col].median().round(2).to_dict()
|
||||
t.insert(1, "中位", [med.get(b) for b in t["分组"]])
|
||||
print(f"\n--- {label} / {name}（{col}）---")
|
||||
print(t.to_string(index=False))
|
||||
|
||||
|
||||
def control_vr60(d: pd.DataFrame, col: str, name: str) -> None:
|
||||
"""在信号根量的高/低两半内部各切一次，看因子是否还独立成立。"""
|
||||
x = d[d[col].notna() & np.isfinite(d[col]) & d.vr60.notna()].copy()
|
||||
x["vr60半"] = np.where(x.vr60 >= x.vr60.median(), "信号根高量", "信号根低量")
|
||||
rows = []
|
||||
for half, g in x.groupby("vr60半"):
|
||||
g = g.copy()
|
||||
g["h"] = pd.qcut(g[col], 2, labels=["低", "高"])
|
||||
lo, hi = g[g.h == "低"], g[g.h == "高"]
|
||||
if min(len(lo), len(hi)) < 40:
|
||||
continue
|
||||
rows.append({"因子": name, "控制层": half, "笔数": len(g),
|
||||
"低组毛R": round(lo.gR.mean(), 3), "高组毛R": round(hi.gR.mean(), 3),
|
||||
"毛R差": round(hi.gR.mean() - lo.gR.mean(), 3),
|
||||
"低组余量": round(lo.net.mean() / lo.tn.mean() * 1e4, 2),
|
||||
"高组余量": round(hi.net.mean() / hi.tn.mean() * 1e4, 2)})
|
||||
if rows:
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--symbols", default="BTC,BNB,ETH,SOL,LINK,LTC,AVAX,XRP,DOGE,ADA")
|
||||
ap.add_argument("--rows", type=int, default=800_000)
|
||||
ap.add_argument("--workers", type=int, default=3)
|
||||
ap.add_argument("--reuse", action="store_true")
|
||||
args = ap.parse_args()
|
||||
|
||||
if args.reuse and OUT.exists():
|
||||
d = pd.read_feather(OUT)
|
||||
else:
|
||||
syms = [s.strip() for s in args.symbols.split(",")]
|
||||
print(f"[开仓前的量与方向] {len(syms)} 币 × {args.rows} 根 1m\n", flush=True)
|
||||
parts = []
|
||||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||||
fut = {ex.submit(collect, s, args.rows): s for s in syms}
|
||||
for i, f in enumerate(as_completed(fut), 1):
|
||||
r = f.result()
|
||||
print(f" [{i}/{len(syms)}] {fut[f]} {0 if r is None else len(r)}", flush=True)
|
||||
if r is not None:
|
||||
parts.append(r)
|
||||
if not parts:
|
||||
print("无结果")
|
||||
return
|
||||
d = pd.concat(parts, ignore_index=True)
|
||||
d.to_feather(OUT)
|
||||
|
||||
from step50_volume import prep
|
||||
|
||||
d["date"] = pd.to_datetime(d["date"])
|
||||
d = prep(d[(d.htf == 1.0) & d.lad & (d.atr_bp >= GATE_BP)].copy())
|
||||
oos, ins = d[d.date < IS_START], d[d.date >= IS_START]
|
||||
print(f"\n实盘口径 {len(d)} 笔 | 样本外 {len(oos)} 发现期 {len(ins)}")
|
||||
print(f"回抽确认：mom10 中位 {d.mom10.median():+.2f} ATR，"
|
||||
f"为负的占 {(d.mom10 < 0).mean()*100:.0f}%")
|
||||
|
||||
print("\n" + "=" * 118)
|
||||
print("########## 一、四个因子各自看（样本外 / 发现期）##########")
|
||||
for col, name in FACTORS:
|
||||
for lab, part in (("样本外", oos), ("发现期", ins)):
|
||||
quartiles(part, col, name, lab)
|
||||
|
||||
print("\n" + "=" * 118)
|
||||
print("########## 二、控信号根自身的量（vr60）后是否还成立 ##########")
|
||||
for col, name in FACTORS:
|
||||
control_vr60(d, col, name)
|
||||
|
||||
print("\n" + "=" * 118)
|
||||
print("########## 三、缩量回抽 vs 放量回抽 ##########")
|
||||
x = d[d.vpre10.notna() & np.isfinite(d.vpre10) & d.mom10.notna()].copy()
|
||||
x["回抽"] = np.where(x.mom10 < 0, "回抽（逆向）", "顺向进场")
|
||||
x["前段量"] = np.where(x.vpre10 >= x.vpre10.median(), "放量", "缩量")
|
||||
rows = []
|
||||
for a in ("回抽（逆向）", "顺向进场"):
|
||||
for b in ("缩量", "放量"):
|
||||
g = x[(x.回抽 == a) & (x.前段量 == b)]
|
||||
rows.append(stat(g, f"{a} × {b}", len(x)))
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
|
||||
print("\n 与信号根量的相关性（防止是同一件事换个说法）")
|
||||
for col, name in FACTORS:
|
||||
v = x[[col, "vr60"]].dropna()
|
||||
print(f" {name:<14} vs vr60 r = {v[col].corr(v.vr60):+.3f}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user