Chan/research/live/venue_parity.py

"""前置测量二：venue 对齐——Bitget 与 Binance 的 1m 是不是同一批信号。

研究数据全部来自 Binance，影子交易器却跑在 Bitget。若两家的 1m K 线有差异，
信号集就会不同，而这个差异会被误记到滑点账上——那样收集一两周也不可归因。

所以先把两家同期的 1m 拉齐，跑同一套管线，比三件事：
  K 线层   时间戳缺口、close 价差（bp）、high/low 差异
  信号层   原始 fast_bsp3 的重合率
  过滤后   加 5m 同向过滤后的重合率（这才是实际要交易的那批）

重合率高 → 后面测到的滑点可以直接对照 Binance 回测的 3.9bp 预算。
重合率低 → 必须先补 Bitget 自己的回测基线，否则实验不可归因。

输出 out/venue_parity.csv。Bitget 数据缓存在 live/cache/，重跑不必再拉。
"""
from __future__ import annotations

import argparse
import os
import sys
import time
import warnings
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
RESEARCH = HERE.parent
sys.path.insert(0, str(RESEARCH))
sys.path.insert(0, str(RESEARCH.parent))
pd.set_option("display.width", 240)

CACHE = HERE / "cache"
LTF, HTF = "1m", "5m"
HTF_RATIO = 5
SYMS = ("BTC", "ETH", "SOL")
NUMERIC = ("open", "high", "low", "close", "volume")


def _exchange():
    import ccxt
    # 这台机器在新加坡，直连 Bitget 0.30s。绝不要照抄交接文档里 Mac 的代理配置，
    # 代理会把延迟放大到秒级，测出来的滑点就是代理的账。
    # rateLimit 默认 50ms，连拉上千页 history-candles 会被 429，放宽到 120ms
    return ccxt.bitget({"options": {"defaultType": "swap"},
                        "enableRateLimit": True, "rateLimit": 120})


def fetch_bitget(sym: str, tf: str, days: int, refresh: bool = False) -> pd.DataFrame:
    """分页拉 Bitget 永续 K 线，落盘缓存，列结构对齐 lib/data.py。"""
    CACHE.mkdir(parents=True, exist_ok=True)
    path = CACHE / f"bitget_{sym}_{tf}_{days}d.feather"
    if path.exists() and not refresh:
        return pd.read_feather(path)

    ex = _exchange()
    pair = f"{sym}/USDT:USDT"
    period_ms = ex.parse_timeframe(tf) * 1000
    t0 = time.perf_counter()
    calls = 0

    # 远端 history-candles 每页硬上限 200 根，而 ccxt 会按 limit 推算 endTime，
    # 只把窗口末尾的 200 根还给你。若照 limit=1000 步进，每页就白丢 800 根——
    # 210 天曾因此只拿到应有量的 31%。故分页一律按 200 走。
    PAGE = 200

    def one(since: int) -> list:
        """单次取数并退避重试。history-candles 连拉上千次会触发 429。"""
        for attempt in range(6):
            try:
                return ex.fetch_ohlcv(pair, tf, since=since, limit=PAGE)
            except Exception as e:
                if attempt == 5:
                    raise
                wait = 2 ** attempt
                print(f"    {sym} {tf}: {type(e).__name__}，{wait}s 后重试",
                      flush=True)
                time.sleep(wait)
        return []

    def page(start: int, stop: int) -> list:
        """向前分页。Bitget 把 since 当开区间，故每次从上一批最后一根重取，
        边界少的那一根靠去重消化。"""
        nonlocal calls
        got, since = [], start
        while since < stop:
            batch = one(since)
            calls += 1
            if len(batch) < 2:
                break
            got.extend(batch)
            if batch[-1][0] <= since:
                break
            since = batch[-1][0]
            if calls % 200 == 0:
                print(f"    {sym} {tf}: {len(got)} 根 / {calls} 次请求", flush=True)
        return got

    now = ex.milliseconds()
    rows = page(now - days * 86_400_000, now)

    # 补缺口：远端接口一次只给 200 根，个别区段仍可能漏，逐个补到补不动为止
    for _ in range(5):
        ts = np.unique(np.array([r[0] for r in rows], dtype="int64"))
        if len(ts) < 2:
            break
        holes = np.where(np.diff(ts) > period_ms)[0]
        if not len(holes):
            break
        before = len(ts)
        for i in holes:
            rows.extend(page(int(ts[i]), int(ts[i + 1])))
        if len(np.unique([r[0] for r in rows])) <= before:
            break

    df = pd.DataFrame(rows, columns=["timestamp", *NUMERIC])
    df["timestamp"] = df["timestamp"].astype("int64")
    for c in NUMERIC:
        df[c] = pd.to_numeric(df[c], errors="coerce")
    df = (df.dropna(subset=list(NUMERIC))
            .drop_duplicates(subset=["timestamp"])
            .sort_values("timestamp")
            .reset_index(drop=True))
    df["date"] = (pd.to_datetime(df["timestamp"], unit="ms", utc=True)
                    .dt.tz_convert("Asia/Shanghai"))
    df = df[["timestamp", "date", *NUMERIC]]
    gap = int(((np.diff(df["timestamp"].to_numpy()) // period_ms) - 1).clip(0).sum())
    print(f"    {sym} {tf}: {len(df)} 根，{calls} 次请求，"
          f"{time.perf_counter() - t0:.1f}s，残余缺口 {gap} 根", flush=True)
    df.to_feather(path)
    return df


def pipeline(df_l: pd.DataFrame, df_h: pd.DataFrame) -> pd.DataFrame:
    """全量口径跑一遍：原始信号 + 5m 同向过滤，返回带时间戳的信号表。"""
    from chanlun import TF_DF
    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

    chan_l = TF_DF(df_l, 1, LTF)
    cdf = chan_l.dataframe
    zones = build_htf_zones(cdf, LTF, chan=chan_l)
    if zones.empty:
        return pd.DataFrame()
    sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
    if sig.empty:
        return pd.DataFrame()

    chan_h = TF_DF(df_h, 1, HTF)
    hdf = chan_h.dataframe
    tl = htf_fx_timeline(signals_to_frame(extract_fx_signals(chan_h, hdf)), hdf)
    full = attach_htf_context(sig, cdf, tl, "h1")
    full["entry_ts"] = cdf["timestamp"].to_numpy()[full["entry_idx"].astype(int)]
    return full


def overlap(a: pd.DataFrame, b: pd.DataFrame, period_ms: int,
            tol_bars: int = 1) -> dict:
    """按时间戳比对两个信号集。方向也必须一致才算命中。"""
    if a.empty or b.empty:
        return {"a": len(a), "b": len(b), "同根": np.nan, f"±{tol_bars}根": np.nan}
    bt = b["entry_ts"].to_numpy()
    bd = b["direction"].to_numpy()
    exact = near = 0
    for ts, d in zip(a["entry_ts"].to_numpy(), a["direction"].to_numpy()):
        hit = np.where((bt == ts) & (bd == d))[0]
        if len(hit):
            exact += 1
            near += 1
            continue
        if np.any((np.abs(bt - ts) <= tol_bars * period_ms) & (bd == d)):
            near += 1
    return {"a": len(a), "b": len(b),
            "同根": exact / len(a), f"±{tol_bars}根": near / len(a)}


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--days", type=int, default=30)
    ap.add_argument("--symbols", default="BTC,ETH,SOL")
    ap.add_argument("--warmup", type=int, default=2000,
                    help="丢弃前若干根的信号，避开中枢左边界效应")
    ap.add_argument("--refresh", action="store_true")
    args = ap.parse_args()

    from lib.data import load_local

    syms = [s.strip() for s in args.symbols.split(",")]
    period_ms = 60_000
    print(f"[venue 对齐] {syms} · 近 {args.days} 天 1m · "
          f"预热丢弃 {args.warmup} 根\n", flush=True)

    bar_rows, sig_rows = [], []
    for sym in syms:
        print(f"── {sym}", flush=True)
        bg_l = fetch_bitget(sym, LTF, args.days, args.refresh)
        bg_h = fetch_bitget(sym, HTF, args.days, args.refresh)

        pair = f"{sym}/USDT:USDT"
        bn_l_all = load_local(pair, LTF)
        bn_h_all = load_local(pair, HTF)
        if bn_l_all is None or bn_h_all is None:
            print(f"  跳过：本地无 Binance 数据")
            continue

        # 只比两家都有的那段时间
        lo = max(bg_l["timestamp"].min(), bn_l_all["timestamp"].min())
        hi = min(bg_l["timestamp"].max(), bn_l_all["timestamp"].max())
        bg_l = bg_l[(bg_l.timestamp >= lo) & (bg_l.timestamp <= hi)].reset_index(drop=True)
        bn_l = bn_l_all[(bn_l_all.timestamp >= lo) & (bn_l_all.timestamp <= hi)].reset_index(drop=True)
        bg_h = bg_h[bg_h.timestamp <= hi].reset_index(drop=True)
        bn_h = bn_h_all[(bn_h_all.timestamp >= bg_h["timestamp"].min())
                        & (bn_h_all.timestamp <= hi)].reset_index(drop=True)

        span_d = (hi - lo) / 86_400_000
        expect = int((hi - lo) / period_ms) + 1
        # K 线层比对
        m = bg_l.merge(bn_l, on="timestamp", suffixes=("_bg", "_bn"))
        dc = (m["close_bg"] - m["close_bn"]) / m["close_bn"] * 1e4
        dh = (m["high_bg"] - m["high_bn"]) / m["high_bn"] * 1e4
        dl = (m["low_bg"] - m["low_bn"]) / m["low_bn"] * 1e4
        bar_rows.append({
            "品种": sym, "重叠天数": round(span_d, 1),
            "Bitget根数": len(bg_l), "Binance根数": len(bn_l),
            "应有根数": expect,
            "Bitget缺口": expect - len(bg_l), "Binance缺口": expect - len(bn_l),
            "共有根数": len(m),
            "close中位差": f"{dc.median():+.2f}bp",
            "close绝对差P95": f"{dc.abs().quantile(.95):.2f}bp",
            "high绝对差P95": f"{dh.abs().quantile(.95):.2f}bp",
            "low绝对差P95": f"{dl.abs().quantile(.95):.2f}bp",
        })
        print(f"  K线：重叠 {span_d:.1f} 天，共有 {len(m)} 根，"
              f"close 中位差 {dc.median():+.2f}bp，P95 {dc.abs().quantile(.95):.2f}bp",
              flush=True)

        # 信号层比对
        t0 = time.perf_counter()
        s_bg = pipeline(bg_l, bg_h)
        s_bn = pipeline(bn_l, bn_h)
        print(f"  管线跑完 {time.perf_counter() - t0:.1f}s", flush=True)
        if s_bg.empty or s_bn.empty:
            print("  信号为空，跳过信号层")
            continue

        cut_bg = bg_l["timestamp"].to_numpy()[min(args.warmup, len(bg_l) - 1)]
        cut_bn = bn_l["timestamp"].to_numpy()[min(args.warmup, len(bn_l) - 1)]
        cut = max(cut_bg, cut_bn)
        s_bg = s_bg[s_bg.entry_ts >= cut]
        s_bn = s_bn[s_bn.entry_ts >= cut]
        f_bg = s_bg[s_bg["h1_agree"] == 1]
        f_bn = s_bn[s_bn["h1_agree"] == 1]

        for tag, x, y in (("原始", s_bg, s_bn), ("5m同向后", f_bg, f_bn)):
            o1 = overlap(x, y, period_ms)   # Bitget 的信号有多少在 Binance 也有
            o2 = overlap(y, x, period_ms)   # 反向
            sig_rows.append({
                "品种": sym, "口径": tag,
                "Bitget信号": o1["a"], "Binance信号": o1["b"],
                "BG→BN同根": f"{o1['同根'] * 100:.1f}%",
                "BG→BN±1根": f"{o1['±1根'] * 100:.1f}%",
                "BN→BG同根": f"{o2['同根'] * 100:.1f}%",
                "BN→BG±1根": f"{o2['±1根'] * 100:.1f}%",
            })
            print(f"  {tag}：Bitget {o1['a']} 笔 / Binance {o1['b']} 笔，"
                  f"同根重合 {o1['同根'] * 100:.1f}%，±1根 {o1['±1根'] * 100:.1f}%",
                  flush=True)

    if not bar_rows:
        print("无结果")
        return

    print("\n" + "=" * 120)
    print("########## 1. K 线层 ##########")
    tb_bar = pd.DataFrame(bar_rows)
    print(tb_bar.to_string(index=False))
    print("  缺口是「应有根数 − 实际根数」，永续在极端行情或维护时会漏推。")

    print("\n########## 2. 信号层 ##########")
    tb_sig = pd.DataFrame(sig_rows)
    print(tb_sig.to_string(index=False))

    print("\n########## 结论 ##########")
    fin = tb_sig[tb_sig["口径"] == "5m同向后"]
    if not fin.empty:
        v = fin["BG→BN同根"].str.rstrip("%").astype(float)
        print(f"  过滤后口径的同根重合率：{v.min():.1f}% ~ {v.max():.1f}%，"
              f"均值 {v.mean():.1f}%")
        print("  高 → 滑点可直接对照 Binance 回测的 3.9bp 预算；")
        print("  低 → 必须先补 Bitget 自己的 1m 回测基线，否则实验不可归因。")

    out_dir = RESEARCH / "out"
    tb_bar.to_csv(out_dir / "venue_parity_bars.csv", index=False)
    tb_sig.to_csv(out_dir / "venue_parity_signals.csv", index=False)
    print(f"\n产物写入 {out_dir}/venue_parity_bars.csv 与 venue_parity_signals.csv")


if __name__ == "__main__":
    main()