jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
233 lines
9.6 KiB
Python
233 lines
9.6 KiB
Python
"""Step 27：小级别三买入场后，大级别也给出三买，该不该改止盈。
|
|
|
|
现在的出场是入场瞬间就锁死的 1.5/3.0 ATR，持仓期间无论出现什么新证据都不动。
|
|
但按区间套的逻辑，小级别三买只是「可能转折」，等大级别同向三买落地，
|
|
趋势的证据强度完全变了，此时仍用原止盈就把最大的一段行情让掉了。
|
|
|
|
对照四种出场：
|
|
A 固定 TP 3.0 ATR
|
|
B 放大止盈 等到大级别同向确认 -> TP 6.0 ATR
|
|
C 放大+保本 同上，并把止损收到成本价
|
|
D 更大止盈 TP 9.0 ATR
|
|
另附 E 移动止损 作为「让利润奔跑」的另一种实现。
|
|
"""
|
|
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", 300)
|
|
|
|
SL, TP, MAXB = 1.5, 3.0, 48
|
|
FEE, SLIP = 0.0004, 0.0001
|
|
|
|
|
|
def run_one(task: tuple) -> dict | None:
|
|
import warnings as _w
|
|
_w.filterwarnings("ignore")
|
|
sys.path.insert(0, str(HERE))
|
|
sys.path.insert(0, str(HERE.parent))
|
|
|
|
from chanlun import TF_DF
|
|
from lib.breakout import run_trades
|
|
from lib.data import fetch_ohlcv
|
|
from lib.fast_bsp3 import find_fast_bsp3
|
|
from lib.fx_signal import extract_fx_signals, signals_to_frame
|
|
from lib.nested_bsp import attach_htf_context, htf_fx_timeline
|
|
from lib.nested_level import build_htf_zones
|
|
|
|
sym, ltf, h1, h2 = task
|
|
try:
|
|
df_l = fetch_ohlcv(f"{sym}/USDT:USDT", ltf, 10**9)
|
|
if df_l is None or len(df_l) < 3000:
|
|
return None
|
|
chan_l = TF_DF(df_l, 1, ltf)
|
|
cdf = chan_l.dataframe
|
|
sig = find_fast_bsp3(cdf, build_htf_zones(cdf, ltf, chan=chan_l))
|
|
if sig.empty or len(sig) < 10:
|
|
return None
|
|
|
|
ts_l = cdf["timestamp"].to_numpy()
|
|
boost_bsp = np.zeros(len(cdf)) # 大级别三买
|
|
boost_fx = np.zeros(len(cdf)) # 大级别分型
|
|
|
|
for tf, pref in ((h1, "h1"), (h2, "h2")):
|
|
df_h = fetch_ohlcv(f"{sym}/USDT:USDT", tf, 10**9)
|
|
if df_h is None or len(df_h) < 300:
|
|
continue
|
|
chan_h = TF_DF(df_h, 1, tf)
|
|
cdf_h = chan_h.dataframe
|
|
s = signals_to_frame(extract_fx_signals(chan_h, cdf_h))
|
|
tl = htf_fx_timeline(s, cdf_h)
|
|
sig = attach_htf_context(sig, cdf, tl, pref)
|
|
|
|
if pref == "h1":
|
|
ts_h = cdf_h["timestamp"].to_numpy()
|
|
period = int(np.median(np.diff(ts_h))) if len(ts_h) > 1 else 0
|
|
# 大级别三买：同样跑一遍中枢+快速三买，收盘后才可用
|
|
sh = find_fast_bsp3(cdf_h, build_htf_zones(cdf_h, tf, chan=chan_h))
|
|
if not sh.empty:
|
|
bts = ts_h[sh["entry_idx"].to_numpy().astype(int)] + period
|
|
pos = np.searchsorted(ts_l, bts, side="left")
|
|
for p, d in zip(pos, sh["direction"].to_numpy()):
|
|
if 0 <= p < len(boost_bsp):
|
|
boost_bsp[p] = d
|
|
# 大级别分型确认（更频繁的弱证据）
|
|
pos = np.searchsorted(ts_l, tl["confirm_ts"].to_numpy(), side="left")
|
|
for p, d in zip(pos, tl["direction"].to_numpy()):
|
|
if 0 <= p < len(boost_fx):
|
|
boost_fx[p] = d
|
|
|
|
entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
|
|
m = sig.drop_duplicates("entry_idx").set_index("entry_idx")
|
|
variants = {
|
|
"A 固定TP3": dict(),
|
|
"B 大级别三买->TP6": dict(boost_dir=boost_bsp, tp_boost=2.0),
|
|
"C 大级别三买->TP6+保本": dict(boost_dir=boost_bsp, tp_boost=2.0,
|
|
boost_breakeven=True),
|
|
"D 大级别三买->TP9": dict(boost_dir=boost_bsp, tp_boost=3.0),
|
|
"E 大级别分型->TP6": dict(boost_dir=boost_fx, tp_boost=2.0),
|
|
"F 移动止损": dict(trail=True),
|
|
}
|
|
frames = []
|
|
for name, kw in variants.items():
|
|
tr = run_trades(cdf, entries, SL, TP, MAXB, fee=0.0, entry_delay=1, **kw)
|
|
if tr.empty:
|
|
continue
|
|
tr["variant"], tr["symbol"], tr["ltf"] = name, sym, ltf
|
|
tr["date"] = cdf["date"].to_numpy()[tr["entry_idx"].to_numpy()]
|
|
for c in ("h1_agree", "h2_agree"):
|
|
tr[c] = tr["entry_idx"].map(m[c]) if c in m.columns else np.nan
|
|
frames.append(tr)
|
|
return {"task": f"{sym} {ltf}", "trades": pd.concat(frames, ignore_index=True)}
|
|
except Exception as e:
|
|
return {"task": f"{sym} {ltf}", "error": repr(e)[:200]}
|
|
|
|
|
|
def desc(g: pd.DataFrame, label: str) -> dict:
|
|
r = g["gross"].to_numpy() - FEE - SLIP
|
|
if len(r) < 20:
|
|
return {}
|
|
w, o = r[r > 0], r[r <= 0]
|
|
sd = r.std(ddof=1)
|
|
return {
|
|
"出场方式": label, "笔数": len(r),
|
|
"胜率": f"{(r > 0).mean() * 100:.1f}%",
|
|
"均收益": f"{r.mean() * 100:+.3f}%",
|
|
"中位": f"{np.median(r) * 100:+.3f}%",
|
|
"PF": f"{w.sum() / abs(o.sum()):.2f}" if len(o) else "inf",
|
|
"偏度": f"{pd.Series(r).skew():.2f}",
|
|
"t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
|
|
"均持有": f"{g['bars_held'].mean():.0f}",
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--pairs", default="15m:1h:4h,30m:2h:4h")
|
|
ap.add_argument("--symbols", default="BTC,ETH,SOL")
|
|
ap.add_argument("--workers", type=int, default=6)
|
|
ap.add_argument("--reuse", action="store_true")
|
|
args = ap.parse_args()
|
|
|
|
cache = HERE / "out" / "step27_dynexit.csv"
|
|
if args.reuse and cache.exists():
|
|
allt = pd.read_csv(cache, parse_dates=["date"])
|
|
print(f"[复用] {len(allt)} 笔\n")
|
|
else:
|
|
tasks = [(s, *tuple(p.split(":")))
|
|
for p in args.pairs.split(",") if p
|
|
for s in args.symbols.split(",")]
|
|
print(f"[动态出场] {len(tasks)} 个任务\n", flush=True)
|
|
res = []
|
|
with ProcessPoolExecutor(max_workers=args.workers) as ex:
|
|
futs = {ex.submit(run_one, t): t for t in tasks}
|
|
for i, f in enumerate(as_completed(futs), 1):
|
|
r = f.result()
|
|
if r is None or "error" in (r or {}):
|
|
print(f" [{i}] 跳过 {(r or {}).get('error', '')}", flush=True)
|
|
continue
|
|
res.append(r)
|
|
print(f" [{i}/{len(tasks)}] {r['task']}", flush=True)
|
|
if not res:
|
|
return
|
|
allt = pd.concat([r["trades"] for r in res], ignore_index=True)
|
|
allt.to_csv(cache, index=False)
|
|
|
|
allt["date"] = pd.to_datetime(allt["date"])
|
|
a = allt[allt["h1_agree"] == 1]
|
|
|
|
print("=" * 115)
|
|
print("########## 1. 六种出场方式（大级别同向，全级别合并）##########")
|
|
print(pd.DataFrame([r for r in
|
|
[desc(g, v) for v, g in a.groupby("variant")] if r]
|
|
).to_string(index=False))
|
|
|
|
print("\n########## 2. 分级别 ##########")
|
|
for tf, g in a.groupby("ltf"):
|
|
rows = [desc(x, f"{tf} {v}") for v, x in g.groupby("variant")]
|
|
print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
|
|
|
|
print("\n########## 3. 触发率：多少笔真的等到了大级别确认 ##########")
|
|
rows = []
|
|
for v in ("B 大级别三买->TP6", "E 大级别分型->TP6"):
|
|
g = a[a.variant == v]
|
|
if g.empty or "boosted" not in g.columns:
|
|
continue
|
|
b = g["boosted"].astype(bool)
|
|
rows.append({"变体": v, "总笔数": len(g), "触发数": int(b.sum()),
|
|
"触发率": f"{b.mean() * 100:.1f}%"})
|
|
if rows:
|
|
print(pd.DataFrame(rows).to_string(index=False))
|
|
|
|
print("\n########## 4. 只看触发了的那批：改单到底有没有用 ##########")
|
|
base = a[a.variant == "A 固定TP3"].set_index(["symbol", "ltf", "entry_idx"])
|
|
for v in ("B 大级别三买->TP6", "C 大级别三买->TP6+保本", "E 大级别分型->TP6"):
|
|
g = a[a.variant == v]
|
|
if g.empty or "boosted" not in g.columns:
|
|
continue
|
|
gb = g[g["boosted"].astype(bool)].set_index(["symbol", "ltf", "entry_idx"])
|
|
if len(gb) < 20:
|
|
continue
|
|
common = gb.index.intersection(base.index)
|
|
if len(common) < 20:
|
|
continue
|
|
r_new = gb.loc[common, "gross"].to_numpy() - FEE - SLIP
|
|
r_old = base.loc[common, "gross"].to_numpy() - FEE - SLIP
|
|
d = r_new - r_old
|
|
sd = d.std(ddof=1)
|
|
print(f" {v}: n={len(d)} 改单后均收益 {r_new.mean() * 100:+.3f}% vs "
|
|
f"原 {r_old.mean() * 100:+.3f}% 差 {d.mean() * 100:+.3f}% "
|
|
f"t={d.mean() / (sd / np.sqrt(len(d))):+.2f}" if sd else "")
|
|
|
|
print("\n########## 5. 资金曲线 ##########")
|
|
for v, g in a.groupby("variant"):
|
|
g = g.sort_values("date")
|
|
r = g["gross"].to_numpy() - FEE - SLIP
|
|
lev = np.clip(0.01 / np.clip(g["risk_pct"].to_numpy(), 0.002, None), 0, 20)
|
|
pnl = r * lev
|
|
eq = np.cumprod(1 + pnl)
|
|
yrs = (g["date"].max() - g["date"].min()).days / 365.25
|
|
dd = (1 - eq / np.maximum.accumulate(eq)).max()
|
|
print(f" {v:>22}: 年化 {(eq[-1] ** (1 / yrs) - 1) * 100:+6.1f}% "
|
|
f"回撤 {dd * 100:4.1f}% Sharpe "
|
|
f"{pnl.mean() / pnl.std(ddof=1) * np.sqrt(len(pnl) / yrs):4.2f}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|