jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
"""Step 3：滞后成本曲线——量化「若能更早识别信号，收益能回来多少」。
|
|
|
|
做法：把入场点设在分型后第 k 根K线，k 从 0 扫到 30，看收益随 k 的衰减。
|
|
k 是反事实的（实时并不能在分型后第 k 根就确认信号），但它给出了
|
|
「压缩确认滞后」这条优化路线的收益天花板。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from lib.bsp_eval import run_pipeline
|
|
from lib.data import fetch_ohlcv
|
|
from lib.walkforward import analyze_stability, replay
|
|
|
|
pd.set_option("display.width", 240)
|
|
|
|
from chanlun.core.ChanEnum import Chan_BSP_DIR
|
|
|
|
SYMBOL, TF, WINDOW, STEP = "BTC/USDT:USDT", "1h", 3000, 4
|
|
HOLD = 20 # 固定持有期
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--zs", default="seg", choices=["seg", "pure"])
|
|
ap.add_argument("--hold", type=int, default=HOLD)
|
|
args = ap.parse_args()
|
|
hold = args.hold
|
|
|
|
df = fetch_ohlcv(SYMBOL, TF, 50000)
|
|
_, final_bsp = run_pipeline(df, TF, zs_source=args.zs)
|
|
final_keys = {
|
|
(str(b.type).replace("Chan_BSP_TYPE.", ""),
|
|
1 if b.dir == Chan_BSP_DIR.BUY else -1, str(b.end_time))
|
|
for b in final_bsp if b.is_sure and b.sure_time is not None
|
|
}
|
|
result = replay(df, TF, window=WINDOW, step=STEP, zs_source=args.zs,
|
|
cache_key=f"{SYMBOL.replace('/', '_').replace(':', '-')}_{TF}")
|
|
life = analyze_stability(result, final_keys, df, window=WINDOW)
|
|
valid = life[~life["truncated"]].copy()
|
|
valid = valid.dropna(subset=["fx_idx"])
|
|
valid["fx_idx"] = valid["fx_idx"].astype(int)
|
|
|
|
closes = df["close"].to_numpy(dtype=float)
|
|
n = len(df)
|
|
|
|
print(f"[中枢来源] {args.zs}")
|
|
print(f"[样本] 有效信号 = {len(valid)} 持有期 = {hold} 根")
|
|
print(f"[实际] 实时首见滞后中位数 = {valid['observed_lag'].median():.0f} 根\n")
|
|
|
|
def curve(sub: pd.DataFrame, label: str) -> None:
|
|
out = []
|
|
for k in (0, 2, 4, 6, 8, 10, 12, 16, 20, 25, 30):
|
|
rets = []
|
|
for _, r in sub.iterrows():
|
|
i = int(r["fx_idx"]) + k
|
|
j = i + hold
|
|
if i >= n or j >= n:
|
|
continue
|
|
d = int(r["direction"])
|
|
rets.append(d * (closes[j] - closes[i]) / closes[i])
|
|
if not rets:
|
|
continue
|
|
a = np.array(rets)
|
|
sd = a.std(ddof=1)
|
|
out.append({
|
|
"k": k, "n": len(a), "mean": a.mean(), "winrate": (a > 0).mean(),
|
|
"tstat": a.mean() / (sd / np.sqrt(len(a))) if sd else np.nan,
|
|
})
|
|
o = pd.DataFrame(out)
|
|
o["mean"] = o["mean"].map(lambda v: f"{v * 100:+.2f}%")
|
|
o["winrate"] = o["winrate"].map(lambda v: f"{v * 100:.0f}%")
|
|
o["tstat"] = o["tstat"].map(lambda v: f"{v:+.2f}")
|
|
print(f"--- {label} ---")
|
|
print(o.to_string(index=False))
|
|
print()
|
|
|
|
curve(valid, "全部信号")
|
|
curve(valid[valid.direction == 1], "仅做多信号 (B1/B2/B3)")
|
|
curve(valid[valid.direction == -1], "仅做空信号 (S1/S2/S3)")
|
|
for t in sorted(valid["bsp_type"].unique()):
|
|
sub = valid[valid.bsp_type == t]
|
|
if len(sub) >= 25:
|
|
curve(sub, f"{t} (n={len(sub)})")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|