jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
200 lines
7.8 KiB
Python
200 lines
7.8 KiB
Python
"""Step 9：笔端点的实时可预测性上界。
|
||
|
||
Step 8 确立了两件事：
|
||
- 笔端点分型事后看有强收益，但那是标签泄漏（端点的定义就要求它之后反向走出一笔）。
|
||
- 然而预测任务本身合法：分型 t+1 就确认，端点身份 t+10 才揭晓，中间 9 根价差真实可得。
|
||
|
||
本步用 LightGBM 吃掉全部实时可得特征，做 purged walk-forward，
|
||
回答唯一重要的问题：按预测概率选出的信号，扣费后到底赚不赚钱。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
import warnings
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
warnings.filterwarnings("ignore")
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
|
||
from lib.data import fetch_ohlcv
|
||
from lib.fx_signal import add_forward_returns, extract_fx_signals, signals_to_frame
|
||
from lib.nested_level import annotate_position, build_htf_zones
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
from chanlun import TF_DF
|
||
|
||
pd.set_option("display.width", 240)
|
||
HORIZONS = (3, 5, 10, 20)
|
||
FEE = 0.0008 # 双边手续费+滑点的保守估计
|
||
|
||
|
||
def build_features(sig: pd.DataFrame, cdf: pd.DataFrame) -> pd.DataFrame:
|
||
"""全部特征在 confirm_idx 当根即可得，无未来信息。"""
|
||
f = sig.sort_values("fx_idx").reset_index(drop=True).copy()
|
||
ci = f["confirm_idx"].to_numpy().astype(int)
|
||
|
||
f["gap"] = f["fx_idx"].diff().fillna(0)
|
||
same = f.groupby("direction")["fx_idx"].diff()
|
||
f["gap_same"] = same.fillna(0)
|
||
|
||
# 本段涨跌幅：从上一反向分型到本分型
|
||
f["seg_amp"] = np.abs(f["price"].diff()) / f["price"].clip(lower=1e-9)
|
||
f["new_extreme"] = np.where(
|
||
f["direction"] == 1, f["price"] <= f["prev_extreme"], f["price"] >= f["prev_extreme"]
|
||
).astype(float)
|
||
f["area_norm"] = f["seg_macd_area"] / f["price"].clip(lower=1e-9)
|
||
f["ratio_f"] = f["ratio"].replace([np.inf, -np.inf], np.nan).fillna(1.0).clip(0, 5)
|
||
f["is_div_f"] = f["is_divergence"].astype(float)
|
||
|
||
close = cdf["close"].to_numpy(dtype=float)
|
||
|
||
def col(name: str) -> np.ndarray:
|
||
return cdf[name].to_numpy(dtype=float)[ci] if name in cdf.columns else np.zeros(len(ci))
|
||
|
||
c_at = close[ci]
|
||
f["rsi"] = col("rsi")
|
||
f["bbp30"] = col("bbp30")
|
||
f["bbp120"] = col("bbp120")
|
||
f["bbp365"] = col("bbp365")
|
||
f["atr_pct"] = col("atr") / np.clip(c_at, 1e-9, None)
|
||
f["vol_ratio"] = col("volume_ratio")
|
||
f["macdhist_n"] = col("macdhist") / np.clip(c_at, 1e-9, None) * 100
|
||
f["macd_n"] = col("macd") / np.clip(c_at, 1e-9, None) * 100
|
||
for e in ("ema24", "ema52", "ema104"):
|
||
f[f"d_{e}"] = c_at / np.clip(col(e), 1e-9, None) - 1
|
||
f["ema_slope"] = col("ema24") / np.clip(col("ema52"), 1e-9, None) - 1
|
||
f["hour"] = pd.to_datetime(cdf["date"].to_numpy()[ci]).hour
|
||
|
||
# 近端动量与波动
|
||
for w in (6, 24, 72):
|
||
prev = np.maximum(ci - w, 0)
|
||
f[f"mom_{w}"] = (c_at - close[prev]) / np.clip(close[prev], 1e-9, None)
|
||
f["dir_mom24"] = f["direction"] * f["mom_24"]
|
||
|
||
return f
|
||
|
||
|
||
FEATURES = [
|
||
"direction", "lag", "gap", "gap_same", "seg_amp", "new_extreme",
|
||
"area_norm", "ratio_f", "is_div_f",
|
||
"rsi", "bbp30", "bbp120", "bbp365", "atr_pct", "vol_ratio",
|
||
"macdhist_n", "macd_n", "d_ema24", "d_ema52", "d_ema104", "ema_slope",
|
||
"hour", "mom_6", "mom_24", "mom_72", "dir_mom24",
|
||
"zone_pos", "near_support", "near_resistance", "inside_zone",
|
||
]
|
||
|
||
|
||
def main() -> None:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--symbol", default="BTC/USDT:USDT")
|
||
ap.add_argument("--tf", default="1h")
|
||
ap.add_argument("--htf", default="4h")
|
||
ap.add_argument("--folds", type=int, default=5)
|
||
ap.add_argument("--purge", type=int, default=60, help="训练/测试之间隔离的信号数")
|
||
args = ap.parse_args()
|
||
|
||
import lightgbm as lgb
|
||
|
||
df = fetch_ohlcv(args.symbol, args.tf, 10**9)
|
||
chan = TF_DF(df, 1, args.tf)
|
||
cdf = chan.dataframe
|
||
idx_of = {t: i for i, t in enumerate(cdf["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}
|
||
|
||
sig = signals_to_frame(extract_fx_signals(chan, cdf))
|
||
sig = add_forward_returns(sig, cdf, HORIZONS)
|
||
zones = build_htf_zones(fetch_ohlcv(args.symbol, args.htf, 10**9), args.htf)
|
||
sig = annotate_position(sig, cdf, zones, tol=0.015)
|
||
|
||
endpoint_idx = set()
|
||
for bi in chan.bi_list:
|
||
for klc in (bi.start_klc, bi.end_klc):
|
||
k = str(getattr(klc, "end_time", "")) if klc is not None else ""
|
||
if k in idx_of:
|
||
endpoint_idx.add(idx_of[k])
|
||
sig["is_endpoint"] = sig["fx_idx"].isin(endpoint_idx).astype(int)
|
||
|
||
f = build_features(sig, cdf)
|
||
for c in ("near_support", "near_resistance", "inside_zone"):
|
||
f[c] = f[c].astype(float)
|
||
f = f.dropna(subset=[f"ret_{h}" for h in HORIZONS]).reset_index(drop=True)
|
||
|
||
print(f"[样本] {len(f)} 个分型 端点率 {f['is_endpoint'].mean() * 100:.1f}% "
|
||
f"特征 {len(FEATURES)} 个")
|
||
print(f"[时间] {cdf['date'].iloc[0]} -> {cdf['date'].iloc[-1]}")
|
||
print(f"[设置] {args.folds} 折 purged walk-forward，隔离 {args.purge} 个信号，"
|
||
f"单边成本 {FEE * 100 / 2:.2f}%\n")
|
||
|
||
X = f[FEATURES].to_numpy(dtype=float)
|
||
y = f["is_endpoint"].to_numpy()
|
||
n = len(f)
|
||
|
||
oof = np.full(n, np.nan)
|
||
start = int(n * 0.4)
|
||
bounds = np.linspace(start, n, args.folds + 1).astype(int)
|
||
for k in range(args.folds):
|
||
te_lo, te_hi = bounds[k], bounds[k + 1]
|
||
tr_hi = max(te_lo - args.purge, 0)
|
||
if tr_hi < 500:
|
||
continue
|
||
model = lgb.LGBMClassifier(
|
||
n_estimators=400, learning_rate=0.03, num_leaves=31,
|
||
min_child_samples=60, subsample=0.8, colsample_bytree=0.8,
|
||
reg_lambda=1.0, verbose=-1, random_state=42,
|
||
)
|
||
model.fit(X[:tr_hi], y[:tr_hi])
|
||
oof[te_lo:te_hi] = model.predict_proba(X[te_lo:te_hi])[:, 1]
|
||
|
||
f["pred"] = oof
|
||
ev = f.dropna(subset=["pred"]).copy()
|
||
print(f"[样本外] {len(ev)} 个信号参与评估 实际端点率 {ev['is_endpoint'].mean() * 100:.1f}%\n")
|
||
|
||
from sklearn.metrics import roc_auc_score
|
||
print(f"[判别力] 样本外 AUC = {roc_auc_score(ev['is_endpoint'], ev['pred']):.4f} "
|
||
f"(0.5=无判别力)\n")
|
||
|
||
print("########## 按预测概率分档：端点率 与 实际收益 ##########")
|
||
rows = []
|
||
for q, name in [(0.99, "top 1%"), (0.95, "top 5%"), (0.90, "top 10%"),
|
||
(0.80, "top 20%"), (0.70, "top 30%"), (0.50, "top 50%")]:
|
||
thr = ev["pred"].quantile(q)
|
||
g = ev[ev["pred"] >= thr]
|
||
if len(g) < 20:
|
||
continue
|
||
r = {"分档": name, "n": len(g), "端点率": f"{g['is_endpoint'].mean() * 100:.1f}%"}
|
||
for h in HORIZONS:
|
||
v = g[f"ret_{h}"].to_numpy()
|
||
net = v.mean() - FEE
|
||
sd = v.std(ddof=1)
|
||
t = v.mean() / (sd / np.sqrt(len(v))) if sd else np.nan
|
||
r[f"{h}根净收益"] = f"{net * 100:+.3f}%"
|
||
r[f"{h}根t"] = f"{t:+.1f}"
|
||
rows.append(r)
|
||
print(pd.DataFrame(rows).to_string(index=False))
|
||
print(f"\n 净收益 = 毛收益 - {FEE * 100:.2f}% 成本；t 值基于毛收益。")
|
||
|
||
print("\n########## 特征重要性（末折）##########")
|
||
model = lgb.LGBMClassifier(
|
||
n_estimators=400, learning_rate=0.03, num_leaves=31,
|
||
min_child_samples=60, subsample=0.8, colsample_bytree=0.8,
|
||
reg_lambda=1.0, verbose=-1, random_state=42,
|
||
)
|
||
cut = int(n * 0.8)
|
||
model.fit(X[:cut], y[:cut])
|
||
imp = pd.DataFrame({"特征": FEATURES, "增益": model.booster_.feature_importance("gain")})
|
||
imp = imp.sort_values("增益", ascending=False).head(15)
|
||
imp["增益"] = imp["增益"].map(lambda v: f"{v:,.0f}")
|
||
print(imp.to_string(index=False))
|
||
|
||
out = Path(__file__).parent / "out" / f"step9_pred_{args.tf}_{args.htf}.csv"
|
||
out.parent.mkdir(exist_ok=True)
|
||
ev.to_csv(out, index=False)
|
||
print(f"\n明细已写入 {out}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|