jackyu66gitandCursor
7f393b93ed
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。 Co-authored-by: Cursor <cursoragent@cursor.com>
145 lines
5.5 KiB
Python
145 lines
5.5 KiB
Python
"""Step 6：多周期结构普查。
|
|
|
|
核查三个论断：
|
|
1. 大周期分型/笔够用，线段滞后太大 —— 实测各级结构的确认滞后。
|
|
2. BTC 趋势时很难有 15 分钟以上的中枢 —— 实测各周期中枢密度。
|
|
3. 必须区分趋势与盘整 —— 按缠论定义（单中枢=盘整，多个同向中枢=趋势）统计占比。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
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.data import fetch_ohlcv
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from chanlun import TF_DF
|
|
from chanlun.core.ChanEnum import Chan_BI_DIR, Chan_ZS_DIR
|
|
|
|
pd.set_option("display.width", 240)
|
|
|
|
SYMBOL = "BTC/USDT:USDT"
|
|
TFS = ["15m", "1h", "4h", "1d"]
|
|
BARS_PER_DAY = {"15m": 96, "1h": 24, "4h": 6, "1d": 1}
|
|
|
|
|
|
def lag_stats(items, idx_of, attr_end="end_time") -> tuple[float, float, int]:
|
|
"""结构确认滞后：sure_time 与结构终点之间隔了几根K线。"""
|
|
lags = []
|
|
for it in items:
|
|
if not getattr(it, "is_sure", False) or getattr(it, "sure_time", None) is None:
|
|
continue
|
|
e, s = str(getattr(it, attr_end, "")), str(it.sure_time)
|
|
if e in idx_of and s in idx_of:
|
|
lags.append(idx_of[s] - idx_of[e])
|
|
if not lags:
|
|
return np.nan, np.nan, 0
|
|
a = np.array(lags)
|
|
return float(np.median(a)), float(a.mean()), len(a)
|
|
|
|
|
|
def classify_regime(bi_zs_list) -> list[str]:
|
|
"""缠论口径：相邻同向且不重叠的中枢构成趋势，否则为盘整。"""
|
|
if len(bi_zs_list) < 2:
|
|
return ["盘整"] * len(bi_zs_list)
|
|
out = ["盘整"]
|
|
for i in range(1, len(bi_zs_list)):
|
|
prev, cur = bi_zs_list[i - 1], bi_zs_list[i]
|
|
if cur.zd > prev.zg:
|
|
out.append("上涨趋势")
|
|
elif cur.zg < prev.zd:
|
|
out.append("下跌趋势")
|
|
else:
|
|
out.append("盘整")
|
|
return out
|
|
|
|
|
|
def main() -> None:
|
|
rows_struct, rows_zs, rows_regime = [], [], []
|
|
|
|
for tf in TFS:
|
|
df = fetch_ohlcv(SYMBOL, tf, 10**9)
|
|
idx_of = {t: i for i, t in enumerate(df["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}
|
|
days = len(df) / BARS_PER_DAY[tf]
|
|
|
|
chan = TF_DF(df, 1, tf)
|
|
fx_count = sum(1 for k in chan.klc_list if getattr(k, "fx_confirmed", False))
|
|
bi_lag_med, bi_lag_mean, bi_n = lag_stats(chan.bi_list, idx_of)
|
|
seg_lag_med, seg_lag_mean, seg_n = lag_stats(chan.seg_list, idx_of)
|
|
|
|
zs_pure = chan.cal_bi_zs_list_pure(chan.bi_list)
|
|
zs_seg = chan.cal_bi_zs(chan.seg_list)
|
|
|
|
rows_struct.append({
|
|
"周期": tf, "K线数": len(df), "跨度(天)": round(days),
|
|
"KLC": len(chan.klc_list), "分型": fx_count,
|
|
"笔": len(chan.bi_list), "线段": len(chan.seg_list),
|
|
"笔滞后(中位)": bi_lag_med, "线段滞后(中位)": seg_lag_med,
|
|
"线段滞后(均值)": round(seg_lag_mean, 1) if seg_lag_mean == seg_lag_mean else np.nan,
|
|
})
|
|
|
|
rows_zs.append({
|
|
"周期": tf,
|
|
"pure中枢": len(zs_pure), "seg中枢": len(zs_seg),
|
|
"pure每百天": round(len(zs_pure) / days * 100, 1),
|
|
"seg每百天": round(len(zs_seg) / days * 100, 1),
|
|
"笔/中枢": round(len(chan.bi_list) / max(len(zs_pure), 1), 1),
|
|
})
|
|
|
|
regimes = classify_regime(zs_pure)
|
|
cnt = pd.Series(regimes).value_counts()
|
|
total = max(len(regimes), 1)
|
|
rows_regime.append({
|
|
"周期": tf,
|
|
"中枢数": len(zs_pure),
|
|
"盘整": cnt.get("盘整", 0),
|
|
"上涨趋势": cnt.get("上涨趋势", 0),
|
|
"下跌趋势": cnt.get("下跌趋势", 0),
|
|
"趋势占比": f"{(cnt.get('上涨趋势', 0) + cnt.get('下跌趋势', 0)) / total * 100:.0f}%",
|
|
})
|
|
|
|
print("########## 1. 各周期结构数量与确认滞后（单位：K线）##########")
|
|
print(pd.DataFrame(rows_struct).to_string(index=False))
|
|
print("\n 说明：笔滞后 = 笔终点到笔被确认之间的K线数；线段同理。")
|
|
|
|
print("\n########## 2. 中枢密度 ##########")
|
|
print(pd.DataFrame(rows_zs).to_string(index=False))
|
|
|
|
print("\n########## 3. 趋势 / 盘整分布（pure 中枢，缠论口径）##########")
|
|
print(pd.DataFrame(rows_regime).to_string(index=False))
|
|
|
|
# 中枢的时间跨度：BTC 趋势中「难有大级别中枢」的直接证据
|
|
print("\n########## 4. 中枢持续时间分布（pure 中枢，单位：K线）##########")
|
|
rows = []
|
|
for tf in TFS:
|
|
df = fetch_ohlcv(SYMBOL, tf, 10**9)
|
|
idx_of = {t: i for i, t in enumerate(df["date"].dt.strftime("%Y-%m-%d %H:%M:%S"))}
|
|
chan = TF_DF(df, 1, tf)
|
|
zs_list = chan.cal_bi_zs_list_pure(chan.bi_list)
|
|
spans = []
|
|
for zs in zs_list:
|
|
bis = getattr(zs, "bi_list", [])
|
|
if not bis:
|
|
continue
|
|
s, e = str(bis[0].start_time), str(bis[-1].end_time)
|
|
if s in idx_of and e in idx_of:
|
|
spans.append(idx_of[e] - idx_of[s])
|
|
if spans:
|
|
a = np.array(spans)
|
|
rows.append({
|
|
"周期": tf, "中枢数": len(a),
|
|
"跨度中位": int(np.median(a)), "跨度均值": round(a.mean(), 1),
|
|
"最短": int(a.min()), "最长": int(a.max()),
|
|
"覆盖K线占比": f"{a.sum() / len(df) * 100:.0f}%",
|
|
})
|
|
print(pd.DataFrame(rows).to_string(index=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|