jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
60 lines
2.2 KiB
Python
60 lines
2.2 KiB
Python
"""Step 1：全量口径事件研究，给出缠论买卖点信号有效性的乐观上界。
|
|
|
|
注意这一步是 in-sample 的：买卖点由「看完全部历史」的一次性 pipeline 产出，
|
|
中枢与笔的最终形态可能包含事后信息。结论需由 Step 2 的 walk-forward 重放校验。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pandas as pd
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
|
|
from lib.bsp_eval import DEFAULT_HORIZONS, forward_returns, run_pipeline, summarize, to_events
|
|
from lib.data import fetch_ohlcv
|
|
|
|
pd.set_option("display.width", 200)
|
|
pd.set_option("display.max_columns", 50)
|
|
|
|
|
|
def main() -> None:
|
|
symbol, tf, limit = "BTC/USDT:USDT", "1h", 50000
|
|
|
|
df = fetch_ohlcv(symbol, tf, limit)
|
|
print(f"[data] {symbol} {tf} rows={len(df)} {df['date'].iloc[0]} -> {df['date'].iloc[-1]}")
|
|
|
|
chan, bsp_list = run_pipeline(df, tf)
|
|
events = to_events(bsp_list, df)
|
|
print(f"[chan] bi={len(chan.bi_list)} seg={len(chan.seg_list)} bsp={len(bsp_list)} events={len(events)}")
|
|
|
|
if not events:
|
|
print("没有产出可交易事件，终止。")
|
|
return
|
|
|
|
fwd = forward_returns(events, df)
|
|
|
|
lag = fwd["lag_bars"]
|
|
print(f"\n[确认滞后] 均值={lag.mean():.1f} 根 中位数={lag.median():.0f} 根 最大={lag.max()} 根")
|
|
print(" 各类型滞后中位数:")
|
|
print(fwd.groupby("bsp_type")["lag_bars"].agg(["count", "median", "mean", "max"]).to_string())
|
|
|
|
summary = summarize(fwd, df)
|
|
print("\n[事件研究] 方向调整后收益（正=盈利）")
|
|
out = summary.copy()
|
|
for col in ("mean", "median", "excess", "mfe", "mae"):
|
|
out[col] = out[col].map(lambda v: f"{v * 100:+.2f}%")
|
|
out["winrate"] = out["winrate"].map(lambda v: f"{v * 100:.0f}%")
|
|
out["tstat"] = out["tstat"].map(lambda v: f"{v:+.2f}")
|
|
print(out.to_string(index=False))
|
|
|
|
Path(__file__).parent.joinpath("out").mkdir(exist_ok=True)
|
|
fwd.to_csv(Path(__file__).parent / "out" / "step1_events.csv", index=False)
|
|
summary.to_csv(Path(__file__).parent / "out" / "step1_summary.csv", index=False)
|
|
print("\n明细已写入 research/out/")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|