research: 因果回放推翻一类反手(step59)，线段顶点验证识别有效但不够(step60)
step59 逐根回放：init_stream 预热后逐根 append_bar，每根重算笔中枢与 bsp， 记录信号首现根并按它入场。信号身份用「类型+极值KLC时刻」而非 sure_time， 后者正是会被重画的字段。必须逐根，分段重建等于多给信息。 结果把 §3.395 推翻了：召回 100%、幻影 0，即存在性是因果的；但首现根比全量 sure_time 晚中位 15 根、P90 31 根，0% 能准时拿到。按真实首现根入场， B1 反手 PF 2.35→0.92、S1 2.75→1.30，t 0.20/0.64 完全不显著。 教训：sure_time 是引擎事后标注的确认时刻，不等于可执行时刻。任何拿它当 entry 的回测都要先过逐根回放。 step60 用线段终点当标准答案验证识别本身。对照组取所有同向笔端点——B1 按构造 就长在笔低点上，不设这个基准任何绝对命中率都无法解读。5m 上 B1 命中 30.3% 对基准 15.0%，提升 2.02 倍，S1 1.81 倍；B3/S3 恰为 0%，符合三类长在趋势 中段的预期，两者互为标签有效性旁证。 所以识别是对的，但精度只有 30%，且那 70% 噪声 PF 只有 0.08。更关键的是即使 用未来函数把精度提到 100%，命中组也只有 PF 0.70~0.83，仍不赚钱——因为入场价 已在结构底上方 2.90 ATR。一类线三层逐层否定后到此为止。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
+91
-1
@@ -1086,7 +1086,15 @@ B4 跑出 PF 1.83~2.06 / t +8.5~+10.7。所以 fast B1 的负数不是管线接
|
||||
头早就写了「B4 不只是 B3 提前几根，两者统计性质符号相反」，一类的实验说明
|
||||
反过来也成立：**把反转信号提前，它还是反转信号。**
|
||||
|
||||
### 3.395 ⭐ 一类反手：识别没错，但发出的时点让它变成了反向信号（step55/57）
|
||||
### 3.395 ⛔ 一类反手：**已被 step59 推翻，不要用本节的 PF**（step55/57）
|
||||
|
||||
> **本节结论作废。** 下面的 PF 2.3~3.0 用 `sure_time+1` 当入场时点，而 §3.396
|
||||
> 的逐根回放证明实盘要等中位 **+15 根**才知道信号存在（0% 能准时拿到），
|
||||
> 按真实首现根入场后 PF 塌到 0.92 / 1.30、t 0.20 / 0.64。
|
||||
> 保留本节是为了记录推理链和那个「胜率低于随机 = 信号指反了」的判据，
|
||||
> 它本身是对的、也可复用；错的是把 `sure_time` 当成了可执行时刻。
|
||||
|
||||
|
||||
|
||||
用户问「是不是一类二类的识别错了」。查下来结论是：**代码没写错，但信号该反着用。**
|
||||
|
||||
@@ -1171,6 +1179,88 @@ B4 跑出 PF 1.83~2.06 / t +8.5~+10.7。所以 fast B1 的负数不是管线接
|
||||
时点上，与背驰无关，那这个信号就该重新命名和重新设计。
|
||||
3. **与 B4 的重叠度与相关性**。若只是换名字的 B4，就没有增量价值。
|
||||
|
||||
### 3.396 ⭐ 因果回放：信号存在性是真的，时点是假的（step59）
|
||||
|
||||
§3.395 的 PF 太好，先查因果性——增量模块文件头自己写着「最后一笔 is_sure
|
||||
**允许收回**」，§5.41 也记了中枢右边缘会重画。
|
||||
|
||||
**做法**：`init_stream` 预热 2 万根，随后逐根 `append_bar`，每根之后重算
|
||||
`cal_bi_zs_list_pure` + `find_all_bsp`，记录每个信号**第一次出现**在哪一根；
|
||||
入场用首现根，而不是事后的 `sure_time`。信号身份用「类型 + 极值 KLC 时刻」，
|
||||
**不能用 `sure_time` 当键**——它正是会被重画的字段。
|
||||
|
||||
必须逐根。分段重建（每 S 根算一次）等于在第 t 根多给了 S 根的信息，
|
||||
测出来的因果性是假的。
|
||||
|
||||
| 检查 | 结果 |
|
||||
|---|---|
|
||||
| 召回（全量信号在回放中出现过的比例） | **100%**（B1 50/50，S1 65/65） |
|
||||
| 幻影（回放发过、全量没有的） | **0** |
|
||||
| 首现根 − 全量 `sure_time` | 中位 **+15 根**，P25 +11，P75 +22，P90 +31 |
|
||||
| **准时拿到的比例** | **0%** |
|
||||
|
||||
| 反手 · 5m | 全量口径 | **回放首现口径** |
|
||||
|---|---|---|
|
||||
| B1 | PF 2.35 | **0.92**（t 0.20，余量 −1.37bp） |
|
||||
| S1 | PF 2.75 | **1.30**（t 0.64，余量 +4.21bp） |
|
||||
|
||||
**结论**：信号的**存在性**是因果的（不重画、不消失），但**时点**不是。
|
||||
实盘要晚中位 15 根、5m 上 75 分钟才知道它存在，而那笔交易吃的正是
|
||||
「反弹已走完」这个窗口，晚 75 分钟就没了。t 值 0.20/0.64 完全不显著。
|
||||
|
||||
⚠️ **可复用的教训**：`sure_time` 是引擎事后标注的确认时刻，**不等于可执行时刻**。
|
||||
任何用它当 entry 的回测都要先过逐根回放。样本 115 笔偏薄，但机制清楚、
|
||||
点估计从 2.35 塌到 0.92，方向不存在疑义。
|
||||
|
||||
### 3.397 线段顶点当标准答案：识别是对的，但精度和时点都不够（step60）
|
||||
|
||||
用户提出用线段终点当趋势反转的标准答案——是未来函数，但用作**标签**验证
|
||||
检测器而非用于交易，能把「检测器对不对」和「能不能交易」分开。
|
||||
|
||||
**对照组是关键**：B1 按构造就长在笔的低点上，而笔低点本来就有概率撞上线段底。
|
||||
所以问的是「在所有同向笔端点里，B1 这个标签把命中率提高了多少倍」。
|
||||
|
||||
| 5m | 样本 | 命中线段顶点 | 相对笔端点的提升 |
|
||||
|---|---|---|---|
|
||||
| 笔端点（基准） | 48229 | 15.0% | 1.00 |
|
||||
| **B1** | 623 | **30.3%** | **2.02×** |
|
||||
| **S1** | 644 | 27.2% | 1.81× |
|
||||
| B3 / S3 | 2806 | **0.0%** | 0.00 |
|
||||
|
||||
15m 同向：B1 28.1%（1.81×）、S1 27.3%（1.76×）。
|
||||
|
||||
各容差档（±0~±5 根）数字完全相同，不是 bug：线段终点**本身就是某根笔的终点**，
|
||||
两者天然按笔端点对齐，放宽容差不会多命中。B3/S3 恰好 0% 也是对的——
|
||||
三类长在回抽笔的端点上，按定义就在趋势中段，不该命中反转点。这两条互为
|
||||
标签有效性的旁证。
|
||||
|
||||
**所以检测器是有信息量的：它把「这个笔低点是真反转」的概率翻了一倍。**
|
||||
但还有两道坎：
|
||||
|
||||
| 5m · 按原方向抄底 | 笔数 | 胜率 | PF | t值 |
|
||||
|---|---|---|---|---|
|
||||
| **未命中**线段顶点 | 903 | 8.7% | **0.08** | −42.0 |
|
||||
| **命中**线段顶点 | 364 | 36.3% | **0.70** | −3.2 |
|
||||
|
||||
15m：未命中 PF 0.08、命中 PF 0.83（t −0.36）。
|
||||
|
||||
**精度只有 30%，而那 70% 的噪声是灾难性的（PF 0.08）。更要命的是：即使用
|
||||
未来函数把精度提到 100%，也只到 PF 0.70~0.83，仍不赚钱。**
|
||||
|
||||
原因是 §3.398 的几何：入场价已在结构底上方 **2.90 ATR**（step58）。
|
||||
即使你真站在线段底上，从底部上方 2.9 ATR 处用 2 ATR 止损做多，结构本身就是
|
||||
负期望的。§3.396 的回放还要在此之上再晚 15 根。
|
||||
|
||||
**一类的完整诊断（三层，逐层否定）**
|
||||
|
||||
1. 识别对不对 —— **对**，2× 提升（step60）
|
||||
2. 精度够不够 —— **不够**，30%；且用未来函数选到 100% 也只到 PF 0.83
|
||||
3. 时点来不来得及 —— **来不及**，全量口径已晚到入场价高出结构底 2.9 ATR，
|
||||
实盘回放再晚 15 根
|
||||
|
||||
**判定：一类线到此为止。** 病不在识别，在于「等笔确认」这个机制天然把信号
|
||||
推到了结构失效之后，而这是 `find_all_bsp` 的固有属性，不是可调的参数。
|
||||
|
||||
### 3.4 alpha 的来源（step32 消融）
|
||||
|
||||
逐条拆掉 `fast_bsp3` 的条件后发现：**alpha 完全来自缠论中枢的上下文定位，
|
||||
|
||||
@@ -0,0 +1,236 @@
|
||||
"""因果性回放：一类反手的信号在当时真的发得出来吗？
|
||||
|
||||
§3.395 的 PF 2.3~3.0 建立在全量数据一次算完的 `bsp_list` 上。但增量模块的
|
||||
文件头自己写着「笔必须整表重扫：**最后一笔 is_sure 允许收回**」，§5.41 也记了
|
||||
中枢右边缘会重画。若信号是事后才浮现的，那个 PF 就是幻觉。
|
||||
|
||||
**做法**：用 `init_stream` 预热，随后逐根 `append_bar`，每根之后重算
|
||||
`cal_bi_zs_list_pure` + `find_all_bsp`，记录每个信号**第一次出现**在哪一根。
|
||||
入场用那一根（的次根开盘），而不是事后的 `sure_time` —— 实盘只能这样。
|
||||
|
||||
必须逐根，不能分段重建：在第 t 根用 `data[0:t+S]` 重算等于多给了 S 根的信息，
|
||||
测出来的因果性是假的。
|
||||
|
||||
**三个要看的量**
|
||||
召回 全量算出的信号，有多少在回放中真的出现过（没出现的是事后才浮现）
|
||||
幻影 回放中出现、但全量里没有的（当时发了、后来被重画掉）
|
||||
代价 回放首现根 vs 全量 sure_time 的滞后；以及按首现根入场的实际 PF
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
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")
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
sys.path.insert(0, str(HERE.parent))
|
||||
|
||||
OUT = HERE / "out" / "step59_replay.feather"
|
||||
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
|
||||
WANT = {"B1": 1, "S1": -1}
|
||||
|
||||
|
||||
def sig_key(b) -> tuple | None:
|
||||
"""信号身份用「类型 + 极值 KLC 的结束时刻」。
|
||||
|
||||
不能用 sure_time 当身份：它正是会被重画的字段，用它做键会把同一个信号
|
||||
在不同根上算成两个。极值点稳定得多。
|
||||
"""
|
||||
t = getattr(b.type, "name", str(b.type))
|
||||
if t not in WANT:
|
||||
return None
|
||||
return (t, str(b.klc.end_time))
|
||||
|
||||
|
||||
def replay(sym: str, tf: str, rows: int, warm: int, steps: int) -> pd.DataFrame | None:
|
||||
from chanlun import TF_DF
|
||||
from lib.data import fetch_ohlcv
|
||||
|
||||
df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
|
||||
if df is None or len(df) < warm + steps + 100:
|
||||
print(f" {sym} {tf} 数据不足 {0 if df is None else len(df)}", flush=True)
|
||||
return None
|
||||
df = df.iloc[-(warm + steps):].reset_index(drop=True)
|
||||
|
||||
# ---- 全量口径：一次算完，作为对照 ----
|
||||
full = TF_DF(df, 1, tf, lean=False)
|
||||
fz = full.cal_bi_zs_list_pure(full.bi_list)
|
||||
full_sig = {}
|
||||
for b in (full.find_all_bsp(full.bi_list, fz) or []):
|
||||
k = sig_key(b)
|
||||
if k and b.sure_time is not None:
|
||||
full_sig[k] = str(b.sure_time)
|
||||
|
||||
# ---- 回放口径：逐根追加，记录首现根 ----
|
||||
chan = TF_DF(df.iloc[:warm].copy(), 1, tf, lean=False)
|
||||
chan.init_stream(df.iloc[:warm].copy(), 1, tf)
|
||||
first_seen: dict[tuple, int] = {}
|
||||
for i in range(warm, len(df)):
|
||||
chan.append_bar(df.iloc[i])
|
||||
try:
|
||||
zs = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||||
bsp = chan.find_all_bsp(chan.bi_list, zs) if zs else []
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
for b in (bsp or []):
|
||||
k = sig_key(b)
|
||||
if k and k not in first_seen:
|
||||
first_seen[k] = i
|
||||
|
||||
dser = pd.to_datetime(full.dataframe["date"])
|
||||
if dser.dt.tz is not None:
|
||||
dser = dser.dt.tz_localize(None)
|
||||
didx = pd.DatetimeIndex(dser)
|
||||
|
||||
def to_i(ts) -> int:
|
||||
t = pd.Timestamp(ts)
|
||||
return int(didx.searchsorted(t.tz_localize(None) if t.tz else t))
|
||||
|
||||
rec = []
|
||||
for k in set(full_sig) | set(first_seen):
|
||||
t, ext_t = k
|
||||
i_seen = first_seen.get(k)
|
||||
# 只统计回放窗口内的：预热段的信号本来就不在考察范围
|
||||
i_ext = to_i(ext_t)
|
||||
if i_ext < warm - 200:
|
||||
continue
|
||||
rec.append({
|
||||
"sym": sym, "tf": tf, "type": t, "dir": WANT[t],
|
||||
"in_full": k in full_sig, "in_replay": i_seen is not None,
|
||||
"i_ext": i_ext,
|
||||
"i_seen": -1 if i_seen is None else i_seen,
|
||||
"i_sure_full": to_i(full_sig[k]) if k in full_sig else -1,
|
||||
})
|
||||
r = pd.DataFrame(rec)
|
||||
if r.empty:
|
||||
return None
|
||||
|
||||
# 按回放首现根入场，跑与实盘一致的出场
|
||||
from lib.exit_model import cfg_name, walk_exits
|
||||
cdf = full.dataframe
|
||||
atr = cdf["atr"].to_numpy(float)
|
||||
cl = cdf["close"].to_numpy(float)
|
||||
live = r[r.in_replay & (r.i_seen < len(cdf) - 2)].copy()
|
||||
live = live[np.isfinite(atr[live.i_seen.values])
|
||||
& (atr[live.i_seen.values] > 0)]
|
||||
if not live.empty:
|
||||
cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
|
||||
# 反手：§3.395 判定该反着做
|
||||
t_ = pd.DataFrame({"entry_idx": live.i_seen.values,
|
||||
"direction": -live.dir.values})
|
||||
res = walk_exits(cdf, t_, [SL], [RUNNER], [MAXB], scale_at=SCALE_AT,
|
||||
runners=(RUNNER,), runner_stops=(RSTOP,))
|
||||
if len(res) == len(live):
|
||||
for c in ("g", "r", "c"):
|
||||
live[c] = res[f"{cfg}_{c}"].to_numpy()
|
||||
live["atr_pct"] = atr[live.i_seen.values] / cl[live.i_seen.values]
|
||||
r = r.merge(live[["i_ext", "type", "g", "r", "c", "atr_pct"]],
|
||||
on=["i_ext", "type"], how="left")
|
||||
return r
|
||||
|
||||
|
||||
def report(d: pd.DataFrame) -> None:
|
||||
from lib.exit_model import fee_of, taker_notional
|
||||
|
||||
print("\n" + "=" * 92)
|
||||
print("########## 一、召回与幻影 ##########")
|
||||
rows = []
|
||||
for (tf, t), x in d.groupby(["tf", "type"]):
|
||||
full = x[x.in_full]
|
||||
rep = x[x.in_replay]
|
||||
both = x[x.in_full & x.in_replay]
|
||||
rows.append({
|
||||
"tf": tf, "类型": t,
|
||||
"全量信号": len(full), "回放信号": len(rep),
|
||||
"召回": f"{len(both)/max(len(full),1)*100:.1f}%",
|
||||
"事后才浮现": len(full) - len(both),
|
||||
"幻影(被重画掉)": len(rep) - len(both),
|
||||
})
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
print("\n召回 = 全量算出的信号里，回放中真的出现过的比例。"
|
||||
"\n幻影 = 回放中发过、全量里却没有的 —— 实盘会照做，回测却看不见它。")
|
||||
|
||||
print("\n" + "=" * 92)
|
||||
print("########## 二、时点代价：回放首现 vs 全量 sure_time ##########")
|
||||
b = d[d.in_full & d.in_replay].copy()
|
||||
b["delay"] = b.i_seen - b.i_sure_full
|
||||
for tf, x in b.groupby("tf"):
|
||||
q = x.delay.quantile([.25, .5, .75, .9])
|
||||
print(f" {tf} 中位 {q[.5]:+.0f} 根 P25 {q[.25]:+.0f} "
|
||||
f"P75 {q[.75]:+.0f} P90 {q[.9]:+.0f} "
|
||||
f"｜ 早于或等于全量的占比 {(x.delay <= 0).mean()*100:.0f}%")
|
||||
print(" 正值 = 回放比全量晚知道，实盘要在更差的价位入场。")
|
||||
|
||||
if "g" not in d.columns:
|
||||
return
|
||||
print("\n" + "=" * 92)
|
||||
print("########## 三、真正能落地的收益：按回放首现根入场（反手）##########")
|
||||
x = d.dropna(subset=["g"]).copy()
|
||||
rows = []
|
||||
for (tf, t), g in x.groupby(["tf", "type"]):
|
||||
if len(g) < 30:
|
||||
continue
|
||||
net = g.g.values - fee_of(g.r.values, g.c.values)
|
||||
gR = g.g.values / (SL * g.atr_pct.values)
|
||||
R = net / (SL * g.atr_pct.values)
|
||||
tn = taker_notional(g.r.values, g.c.values)
|
||||
w, o = net[net > 0].sum(), -net[net <= 0].sum()
|
||||
rows.append({
|
||||
"tf": tf, "类型(反手)": t, "笔数": len(g),
|
||||
"胜率": f"{(net > 0).mean()*100:.1f}%",
|
||||
"毛R": round(gR.mean(), 3), "净均R": round(R.mean(), 3),
|
||||
"PF": round(w / o, 2) if o > 0 else np.inf,
|
||||
"余量bp": round(net.mean() / tn.mean() * 1e4, 2),
|
||||
"t值": round(gR.mean() / (gR.std(ddof=1) / np.sqrt(len(g))), 2),
|
||||
})
|
||||
if rows:
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
print("\n对照 · §3.395 全量口径 5m：B1反手 PF 2.35 / S1反手 2.75")
|
||||
print("若这里明显掉下来，说明那个 PF 吃了右边缘重画的红利，不可落地。")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--symbols", default="BTC,ETH,SOL")
|
||||
ap.add_argument("--tf", default="5m")
|
||||
ap.add_argument("--rows", type=int, default=60_000)
|
||||
ap.add_argument("--warm", type=int, default=20_000)
|
||||
ap.add_argument("--steps", type=int, default=10_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():
|
||||
report(pd.read_feather(OUT))
|
||||
return
|
||||
syms = [x.strip() for x in args.symbols.split(",")]
|
||||
print(f"[因果回放] {len(syms)} 币 × {args.steps} 根逐根重放"
|
||||
f"（每根都要重算笔中枢与 bsp，慢是必然的）\n", flush=True)
|
||||
parts = []
|
||||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||||
fut = {ex.submit(replay, s, args.tf, args.rows, args.warm,
|
||||
args.steps): s for s in syms}
|
||||
for i, f in enumerate(as_completed(fut), 1):
|
||||
r = f.result()
|
||||
print(f" [{i}/{len(syms)}] {fut[f]} "
|
||||
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)
|
||||
OUT.parent.mkdir(exist_ok=True)
|
||||
d.to_feather(OUT)
|
||||
report(d)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,242 @@
|
||||
"""一类买卖点到底有没有找到真正的趋势反转点？用线段顶点当标准答案。
|
||||
|
||||
用户提出：线段的终点就是该级别的趋势反转点，虽然它是未来函数，但可以拿来当
|
||||
**标签**验证检测器，而不是拿来交易。这能把两件事分开：
|
||||
|
||||
检测器对不对 B1 是否真的落在趋势反转底上
|
||||
能不能交易 step59 已证否（实盘首现比 sure_time 晚中位 15 根，PF 塌到 0.92）
|
||||
|
||||
若检测器对而只是慢，那问题是延迟，还有救；若检测器本身就没找到反转点，
|
||||
这条线整个是死的。
|
||||
|
||||
⚠️ **对照组是这个测试的全部意义**。B1 按构造就长在笔的低点上，而笔低点本来
|
||||
就有一定概率撞上线段底。所以要问的不是「B1 命中率多少」，而是
|
||||
**「在所有同向笔端点里，B1 这个标签把命中率提高了多少倍」**。
|
||||
没有这个基准，任何绝对数字都可以随便解读。
|
||||
（同样的坑 fast_bsp 文档头踩过：回抽极值命中笔端点 19.3%，看着不低，
|
||||
但随机基准是 22%，其实是负贡献。）
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
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")
|
||||
HERE = Path(__file__).resolve().parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
sys.path.insert(0, str(HERE.parent))
|
||||
|
||||
OUT = HERE / "out" / "step60_seg_truth.feather"
|
||||
SL, SCALE_AT, RUNNER, RSTOP, MAXB = 2.0, 3.0, 8.0, 2.0, 48
|
||||
TOL = [0, 1, 2, 3, 5]
|
||||
|
||||
|
||||
def collect(sym: str, tf: str, rows: int) -> pd.DataFrame | None:
|
||||
from chanlun import TF_DF
|
||||
from chanlun.core.ChanEnum import Chan_BSP_TYPE, Chan_SEG_DIR
|
||||
from lib.data import fetch_ohlcv
|
||||
from lib.exit_model import cfg_name, walk_exits
|
||||
|
||||
try:
|
||||
df = fetch_ohlcv(f"{sym}/USDT:USDT", tf, rows)
|
||||
if df is None or len(df) < 5_000:
|
||||
return None
|
||||
chan = TF_DF(df, 1, tf, lean=False)
|
||||
cdf = chan.dataframe
|
||||
bz = chan.cal_bi_zs_list_pure(chan.bi_list)
|
||||
if not bz:
|
||||
return None
|
||||
bsp = chan.find_all_bsp(chan.bi_list, bz) or []
|
||||
|
||||
dser = pd.to_datetime(cdf["date"])
|
||||
if dser.dt.tz is not None:
|
||||
dser = dser.dt.tz_localize(None)
|
||||
didx = pd.DatetimeIndex(dser)
|
||||
n = len(cdf)
|
||||
|
||||
def to_i(ts) -> int:
|
||||
t = pd.Timestamp(ts)
|
||||
return int(didx.searchsorted(t.tz_localize(None) if t.tz else t))
|
||||
|
||||
# ---- 标准答案：线段终点。下降线段终点=真底，上升线段终点=真顶 ----
|
||||
seg_bot, seg_top = [], []
|
||||
for sg in getattr(chan, "seg_list", []) or []:
|
||||
if sg.end_time is None:
|
||||
continue
|
||||
i = to_i(sg.end_time)
|
||||
if not (0 <= i < n):
|
||||
continue
|
||||
(seg_bot if sg.dir == Chan_SEG_DIR.DOWN else seg_top).append(i)
|
||||
if not seg_bot or not seg_top:
|
||||
return None
|
||||
truth = {1: np.array(sorted(seg_bot)), -1: np.array(sorted(seg_top))}
|
||||
|
||||
def near(i: int, d: int, tol: int) -> bool:
|
||||
a = truth[d]
|
||||
k = int(np.searchsorted(a, i))
|
||||
for j in (k - 1, k):
|
||||
if 0 <= j < len(a) and abs(int(a[j]) - i) <= tol:
|
||||
return True
|
||||
return False
|
||||
|
||||
rec = []
|
||||
# ---- 对照组：所有笔端点。B1 本就长在笔低点上，基准必须同源 ----
|
||||
for bi in chan.bi_list:
|
||||
if not getattr(bi, "is_sure", False) or bi.end_klc is None:
|
||||
continue
|
||||
d = 1 if str(bi.dir).endswith("DOWN") else -1 # 下降笔终点=低点
|
||||
i = to_i(bi.end_klc.end_time)
|
||||
if not (0 <= i < n):
|
||||
continue
|
||||
rec.append({"kind": "笔端点", "dir": d, "i_ext": i, "i_sure": -1})
|
||||
|
||||
want = {Chan_BSP_TYPE.B1: ("B1", 1), Chan_BSP_TYPE.S1: ("S1", -1),
|
||||
Chan_BSP_TYPE.B3: ("B3", 1), Chan_BSP_TYPE.S3: ("S3", -1)}
|
||||
for b in bsp:
|
||||
tag = want.get(b.type)
|
||||
if tag is None or b.sure_time is None:
|
||||
continue
|
||||
name, d = tag
|
||||
i_ext, i_sure = to_i(b.klc.end_time), to_i(b.sure_time)
|
||||
if not (0 <= i_ext < n and 0 <= i_sure < n):
|
||||
continue
|
||||
rec.append({"kind": name, "dir": d, "i_ext": i_ext,
|
||||
"i_sure": i_sure})
|
||||
|
||||
r = pd.DataFrame(rec)
|
||||
for tol in TOL:
|
||||
r[f"hit{tol}"] = [near(i, d, tol)
|
||||
for i, d in zip(r.i_ext, r.dir)]
|
||||
|
||||
# 命中线段顶点的那批一类，按原方向（抄底）做能不能赚
|
||||
atr = cdf["atr"].to_numpy(float)
|
||||
cl = cdf["close"].to_numpy(float)
|
||||
sig = r[(r.kind.isin(["B1", "S1"])) & (r.i_sure >= 0)
|
||||
& (r.i_sure < n - 2)].copy()
|
||||
sig = sig[np.isfinite(atr[sig.i_sure.values])
|
||||
& (atr[sig.i_sure.values] > 0)]
|
||||
if not sig.empty:
|
||||
cfg = cfg_name(SL, RUNNER, MAXB, RSTOP)
|
||||
res = walk_exits(cdf, pd.DataFrame({
|
||||
"entry_idx": sig.i_sure.values,
|
||||
"direction": sig.dir.values}), [SL], [RUNNER], [MAXB],
|
||||
scale_at=SCALE_AT, runners=(RUNNER,), runner_stops=(RSTOP,))
|
||||
if len(res) == len(sig):
|
||||
for c in ("g", "r", "c"):
|
||||
sig[c] = res[f"{cfg}_{c}"].to_numpy()
|
||||
sig["atr_pct"] = (atr[sig.i_sure.values]
|
||||
/ cl[sig.i_sure.values])
|
||||
r = r.merge(sig[["i_ext", "kind", "g", "r", "c", "atr_pct"]],
|
||||
on=["i_ext", "kind"], how="left")
|
||||
r["sym"], r["tf"] = sym, tf
|
||||
return r
|
||||
except Exception as e: # noqa: BLE001
|
||||
print(f" {sym} {tf} 失败: {type(e).__name__}: {e}", flush=True)
|
||||
return None
|
||||
|
||||
|
||||
def report(d: pd.DataFrame) -> None:
|
||||
from lib.exit_model import fee_of, taker_notional
|
||||
|
||||
for tf, x in d.groupby("tf"):
|
||||
print("\n" + "#" * 92)
|
||||
print(f"########## {tf} · 线段顶点作为标准答案 ##########")
|
||||
print("\n【命中率】i_ext 落在同向线段终点 ±tol 根内的比例")
|
||||
rows = []
|
||||
for kind in ["笔端点", "B1", "S1", "B3", "S3"]:
|
||||
g = x[x.kind == kind]
|
||||
if len(g) < 30:
|
||||
continue
|
||||
row = {"信号": kind, "样本": len(g)}
|
||||
for tol in TOL:
|
||||
row[f"±{tol}根"] = f"{g[f'hit{tol}'].mean()*100:.1f}%"
|
||||
rows.append(row)
|
||||
t = pd.DataFrame(rows)
|
||||
print(t.to_string(index=False))
|
||||
|
||||
base = x[x.kind == "笔端点"]
|
||||
print("\n【提升倍数】相对「所有同向笔端点」这个基准。"
|
||||
"≈1 就是没有信息量")
|
||||
rows = []
|
||||
for kind in ["B1", "S1", "B3", "S3"]:
|
||||
g = x[x.kind == kind]
|
||||
if len(g) < 30:
|
||||
continue
|
||||
row = {"信号": kind}
|
||||
for tol in TOL:
|
||||
b = base[base.dir.isin(g.dir.unique())][f"hit{tol}"].mean()
|
||||
row[f"±{tol}根"] = (round(g[f"hit{tol}"].mean() / b, 2)
|
||||
if b > 0 else np.nan)
|
||||
rows.append(row)
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
|
||||
if "g" not in x.columns:
|
||||
continue
|
||||
print("\n【命中 vs 未命中】一类按原方向（抄底/摸顶）做的表现，"
|
||||
"tol=±2 根")
|
||||
y = x[x.kind.isin(["B1", "S1"])].dropna(subset=["g"]).copy()
|
||||
if len(y) < 60:
|
||||
continue
|
||||
rows = []
|
||||
for hit, g in y.groupby(y.hit2):
|
||||
net = g.g.values - fee_of(g.r.values, g.c.values)
|
||||
gR = g.g.values / (SL * g.atr_pct.values)
|
||||
tn = taker_notional(g.r.values, g.c.values)
|
||||
w, o = net[net > 0].sum(), -net[net <= 0].sum()
|
||||
rows.append({
|
||||
"命中线段顶点": "是" if hit else "否", "笔数": len(g),
|
||||
"胜率": f"{(net > 0).mean()*100:.1f}%",
|
||||
"毛R": round(gR.mean(), 3),
|
||||
"PF": round(w / o, 2) if o > 0 else np.inf,
|
||||
"余量bp": round(net.mean() / tn.mean() * 1e4, 2),
|
||||
"t值": round(gR.mean() / (gR.std(ddof=1)
|
||||
/ np.sqrt(len(g))), 2),
|
||||
})
|
||||
print(pd.DataFrame(rows).to_string(index=False))
|
||||
print("\n判读：若「命中」那组按原方向做显著为正，说明检测器是对的、"
|
||||
"只是掺了太多噪声，\n值得找实时可判的过滤器；若两组都为负，"
|
||||
"说明即使真站在线段底上，\n这个入场时点也已经太晚了。")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--symbols", default="BTC,ETH,SOL,LINK,DOGE")
|
||||
ap.add_argument("--tfs", default="5m,15m")
|
||||
ap.add_argument("--rows", type=int, default=200_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():
|
||||
report(pd.read_feather(OUT))
|
||||
return
|
||||
syms = [s.strip() for s in args.symbols.split(",")]
|
||||
tfs = [t.strip() for t in args.tfs.split(",")]
|
||||
parts = []
|
||||
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
||||
fut = {ex.submit(collect, s, t, args.rows): (s, t)
|
||||
for s in syms for t in tfs}
|
||||
for i, f in enumerate(as_completed(fut), 1):
|
||||
r = f.result()
|
||||
s, t = fut[f]
|
||||
print(f" [{i}/{len(fut)}] {s} {t} "
|
||||
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)
|
||||
OUT.parent.mkdir(exist_ok=True)
|
||||
d.to_feather(OUT)
|
||||
report(d)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user