jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""Step 23：费率分档下的小周期可行性 + 交易额测算。
|
||
|
||
背景：此前统一按 0.08% 双边（VIP0 taker、无返佣）计费，1m/5m 因此判负。
|
||
但实际有 50% 返佣、且交易量越大费率越低，成本可能低至 0.02% 甚至更低。
|
||
1m 的毛收益本来就是正的（+0.047%），所以这条线值得重算。
|
||
|
||
本步：严格口径重跑 1m~30m，记录毛收益，再对多档费率做 what-if，
|
||
并测算「刷量路线」每年能产生多少名义交易额（决定能爬到哪个 VIP 档）。
|
||
"""
|
||
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
|
||
MAX_ROWS = {"1m": 1_200_000}
|
||
|
||
# (名称, 双边费率, 说明)。滑点单列，因为 maker 路线不吃滑点。
|
||
FEE_TIERS = [
|
||
("0.080%", 0.00080, "VIP0 taker 无返佣（原基准）"),
|
||
("0.040%", 0.00040, "taker + 50%返佣"),
|
||
("0.030%", 0.00030, "VIP2 taker + 返佣"),
|
||
("0.020%", 0.00020, "VIP4 taker 或 maker 双边"),
|
||
("0.010%", 0.00010, "maker + 返佣"),
|
||
("0.000%", 0.00000, "maker + 高返佣（理论下限）"),
|
||
]
|
||
# Binance USDT 永续 30 天交易量门槛（美元）
|
||
VIP_TIERS = [("VIP1", 15e6), ("VIP2", 50e6), ("VIP3", 100e6),
|
||
("VIP4", 600e6), ("VIP5", 1000e6)]
|
||
|
||
|
||
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, MAX_ROWS.get(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
|
||
|
||
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)
|
||
s = signals_to_frame(extract_fx_signals(chan_h, chan_h.dataframe))
|
||
sig = attach_htf_context(sig, cdf, htf_fx_timeline(s, chan_h.dataframe), pref)
|
||
|
||
entries = list(zip(sig["entry_idx"].astype(int), sig["direction"].astype(int)))
|
||
# 费率事后套用，这里先跑零费率拿毛收益；仍用次根开盘的真实成交节奏
|
||
tr = run_trades(cdf, entries, SL, TP, MAXB, fee=0.0, entry_delay=1)
|
||
if tr.empty:
|
||
return None
|
||
m = sig.set_index("entry_idx")
|
||
tr["symbol"] = sym
|
||
tr["ltf"] = 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
|
||
return {"task": f"{sym} {ltf}", "trades": tr}
|
||
except Exception as e:
|
||
return {"task": f"{sym} {ltf}", "error": repr(e)[:200]}
|
||
|
||
|
||
def desc(r: np.ndarray, label: str) -> dict:
|
||
if len(r) < 5:
|
||
return {}
|
||
win, loss = 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"{win.sum() / abs(loss.sum()):.2f}" if len(loss) else "inf",
|
||
"t值": f"{r.mean() / (sd / np.sqrt(len(r))):+.2f}",
|
||
}
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--pairs", default="1m:5m:15m,5m:15m:1h,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("--slippage-bp", type=float, default=1.0,
|
||
help="taker 路线的单边滑点(bp)，maker 路线设 0")
|
||
ap.add_argument("--reuse", action="store_true",
|
||
help="跳过回测，直接用上次存下的毛收益明细做费率 what-if")
|
||
args = ap.parse_args()
|
||
|
||
cache = HERE / "out" / "step23_gross.csv"
|
||
if args.reuse and cache.exists():
|
||
allt = pd.read_csv(cache, parse_dates=["date"])
|
||
print(f"[复用] {cache.name}，{len(allt)} 笔\n")
|
||
else:
|
||
pairs = [tuple(p.split(":")) for p in args.pairs.split(",") if p]
|
||
syms = [s.strip() for s in args.symbols.split(",") if s.strip()]
|
||
tasks = [(s, *p) for p in pairs for s in syms]
|
||
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}/{len(tasks)}] 跳过 {(r or {}).get('task', futs[f])} "
|
||
f"{(r or {}).get('error', '')}", flush=True)
|
||
continue
|
||
res.append(r)
|
||
print(f" [{i}/{len(tasks)}] {r['task']} — {len(r['trades'])} 笔", flush=True)
|
||
|
||
if not res:
|
||
print("无结果")
|
||
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"])
|
||
slip = args.slippage_bp / 10000.0
|
||
|
||
print("\n" + "=" * 120)
|
||
print("########## 1. 毛收益（零成本），大级别同向 ##########")
|
||
a1 = allt["h1_agree"] == 1
|
||
rows = [desc(g["gross"].to_numpy(), f"{tf} 同向")
|
||
for tf, g in allt[a1].groupby("ltf")]
|
||
print(pd.DataFrame([r for r in rows if r]).to_string(index=False))
|
||
print(" 毛收益为正 = 结构本身有 alpha，能否落地全看成本能压到多少。")
|
||
|
||
print(f"\n########## 2. 各费率档 × 各级别的 PF（含 {args.slippage_bp}bp 滑点）##########")
|
||
grid = {}
|
||
for tf, g in allt[a1].groupby("ltf"):
|
||
gr = g["gross"].to_numpy()
|
||
for name, fee, _ in FEE_TIERS:
|
||
r = gr - fee - slip
|
||
win, loss = r[r > 0], r[r <= 0]
|
||
grid.setdefault(name, {})[tf] = (
|
||
win.sum() / abs(loss.sum()) if len(loss) else np.inf
|
||
)
|
||
gp = pd.DataFrame(grid).T
|
||
gp = gp[[c for c in ["1m", "5m", "15m", "30m"] if c in gp.columns]]
|
||
print(gp.round(2).to_string())
|
||
|
||
print(f"\n########## 3. 各费率档 × 各级别的 t值 ##########")
|
||
grid = {}
|
||
for tf, g in allt[a1].groupby("ltf"):
|
||
gr = g["gross"].to_numpy()
|
||
for name, fee, _ in FEE_TIERS:
|
||
r = gr - fee - slip
|
||
sd = r.std(ddof=1)
|
||
grid.setdefault(name, {})[tf] = r.mean() / (sd / np.sqrt(len(r)))
|
||
tp_ = pd.DataFrame(grid).T
|
||
tp_ = tp_[[c for c in ["1m", "5m", "15m", "30m"] if c in tp_.columns]]
|
||
print(tp_.round(2).to_string())
|
||
print(" t>2 才算统计显著。找每个级别转正/转显著的临界费率。")
|
||
|
||
print("\n########## 4. 盈亏平衡费率（PF=1 所需的双边成本上限）##########")
|
||
rows = []
|
||
for tf, g in allt[a1].groupby("ltf"):
|
||
gr = g["gross"].to_numpy()
|
||
be = gr.mean() - slip # 均收益扣滑点后还能承受的费率
|
||
# 找 t=2 对应的费率
|
||
sd = gr.std(ddof=1) / np.sqrt(len(gr))
|
||
be_t2 = gr.mean() - 2 * sd - slip
|
||
rows.append({
|
||
"级别": tf, "笔数": len(gr),
|
||
"毛均收益": f"{gr.mean() * 100:+.4f}%",
|
||
"盈亏平衡费率": f"{be * 100:.4f}%",
|
||
"t=2所需费率": f"{be_t2 * 100:.4f}%" if be_t2 > 0 else "达不到",
|
||
"现实最低0.02%可行": "是" if be > 0.0002 else "否",
|
||
})
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
|
||
print("\n########## 5. 刷量测算：每 1 万 USDT 本金、单笔 1% 风险 ##########")
|
||
print(" 名义仓位 = 风险预算 / 止损距离；交易额 = 名义仓位 × 2（开+平）")
|
||
rows = []
|
||
for tf, g in allt[a1].groupby("ltf"):
|
||
yrs = (g["date"].max() - g["date"].min()).days / 365.25
|
||
if yrs <= 0:
|
||
continue
|
||
rp = g["risk_pct"].to_numpy()
|
||
rp = np.clip(rp, 0.002, None) # 止损过窄会把杠杆算爆，截断
|
||
lev = np.clip(0.01 / rp, 0, 20) # 单笔名义仓位 / 本金，上限 20x
|
||
turn_per_10k = lev.sum() * 2 * 10000 / yrs / 3 # 除以3=单币口径
|
||
rows.append({
|
||
"级别": tf, "笔数/年/币": f"{len(g) / yrs / 3:.0f}",
|
||
"均杠杆": f"{lev.mean():.1f}x",
|
||
"年交易额/万U本金": f"${turn_per_10k / 1e6:.2f}M",
|
||
"月交易额/万U": f"${turn_per_10k / 12 / 1e6:.2f}M",
|
||
})
|
||
tb = pd.DataFrame(rows)
|
||
print(tb.to_string(index=False))
|
||
|
||
print("\n########## 6. 爬到各 VIP 档所需本金（按 3 币同跑、30天量）##########")
|
||
rows = []
|
||
for tf, g in allt[a1].groupby("ltf"):
|
||
yrs = (g["date"].max() - g["date"].min()).days / 365.25
|
||
if yrs <= 0:
|
||
continue
|
||
rp = np.clip(g["risk_pct"].to_numpy(), 0.002, None)
|
||
lev = np.clip(0.01 / rp, 0, 20)
|
||
# 3 币合计、每万 U 本金的 30 天交易额
|
||
m30 = lev.sum() * 2 * 10000 / yrs / 12
|
||
row = {"级别": tf, "月交易额/万U(3币)": f"${m30 / 1e6:.2f}M"}
|
||
for vname, need in VIP_TIERS:
|
||
row[vname] = f"${need / m30 * 10000 / 1e6:.1f}M" if m30 > 0 else "—"
|
||
rows.append(row)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print(" 表内数字 = 达到该 VIP 档所需本金。低周期刷量效率高但对本金仍有要求。")
|
||
|
||
print("\n########## 7. 双路线组合：低周期刷量 + 30m 主仓 ##########")
|
||
for name, fee, note in FEE_TIERS:
|
||
parts = []
|
||
for tf in ("1m", "5m", "30m"):
|
||
g = allt[a1 & (allt.ltf == tf)]
|
||
if g.empty:
|
||
continue
|
||
r = g["gross"].to_numpy() - fee - slip
|
||
win, loss = r[r > 0], r[r <= 0]
|
||
pf = win.sum() / abs(loss.sum()) if len(loss) else np.inf
|
||
parts.append(f"{tf} PF={pf:.2f} 均={r.mean() * 100:+.3f}%")
|
||
print(f" {name} ({note})")
|
||
print(f" {' | '.join(parts)}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|