Chan/research/live/parity_env.py

"""环境等价性验证：同一份切片，在 .venv 与 Hummingbot 容器里必须算出同一组信号。

为什么要单独验这件事：容器里是 Python 3.13.14 + pandas 3.0.5 + numpy 2.4.6，
本机 .venv 是 Python 3.14.4 + pandas 3.0.5 + numpy 2.5.2。pandas 同版本，
numpy 差一个小版本。信号已经被证明对 0.25bp 的数据扰动极度敏感（扰动会换掉
一半信号），所以浮点或 groupby 顺序上的任何细微差异都可能改变信号集合——
必须实测，不能推断。

用 --dump 先从 .venv 导出切片成 CSV，两个环境再读同一个 CSV，
这样数据来源差异为零，比出来的就是纯计算差异。

    .venv/bin/python research/live/parity_env.py --dump      # 导出切片
    .venv/bin/python research/live/parity_env.py --run       # 本机计算
    docker run ... /app/parity_env.py --run --tag container  # 容器计算
"""
from __future__ import annotations

import argparse
import json
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))

LTF, HTF = "1m", "5m"
WINDOW = 2000          # step39 定下的窗口：命中率在此饱和
HTF_WINDOW = 800
SYMS = ("BTC", "ETH", "SOL")


def slice_dir() -> Path:
    """容器里挂在 /out，本机是 research/out。"""
    p = Path("/out")
    return p if p.is_dir() else RESEARCH / "out"


def dump() -> None:
    """从 Bitget 缓存导出末尾切片，供两个环境共用。

    用 Bitget 而非 Binance 的数据：目标场地就是 Bitget，且这批缓存正是
    bitget_baseline.py 算余量时用的同一份，口径可直接对上。
    """
    d = slice_dir() / "parity_slices"
    d.mkdir(parents=True, exist_ok=True)
    cache = HERE / "cache"
    for s in SYMS:
        for tf, n in ((LTF, WINDOW), (HTF, HTF_WINDOW)):
            src = cache / f"bitget_{s}_{tf}_30d.feather"
            if not src.exists():
                src = cache / f"bitget_{s}_{tf}_210d.feather"
            if not src.exists():
                print(f"  {s} {tf}: 无缓存 {src.name}，跳过")
                continue
            df = pd.read_feather(src)
            if df is None or df.empty:
                print(f"  {s} {tf}: 缓存为空，跳过")
                continue
            out = df.tail(n).reset_index(drop=True)
            f = d / f"{s}_{tf}.csv"
            # 用 float 全精度写出，避免导出环节就引入舍入差异
            out.to_csv(f, index=False, float_format="%.10f")
            print(f"  {s} {tf}: {len(out)} 根 -> {f.name}")
    print(f"\n切片目录 {d}")


def signals(df_ltf: pd.DataFrame, df_htf: pd.DataFrame) -> dict:
    """复用 step39 的时点重建口径，返回信号下标与耗时。"""
    from chanlun import TF_DF
    from lib.fast_bsp3 import find_fast_bsp3
    from lib.nested_level import build_htf_zones

    t0 = time.perf_counter()
    chan = TF_DF(df_ltf, 1, LTF)
    cdf = chan.dataframe
    zones = build_htf_zones(cdf, LTF, chan=chan)
    raw: list[int] = []
    if not zones.empty:
        sig = find_fast_bsp3(cdf, zones.reset_index(drop=True))
        if sig is not None and not sig.empty:
            col = "idx" if "idx" in sig.columns else sig.columns[0]
            raw = sorted(int(x) for x in sig[col].to_numpy())
    dt = time.perf_counter() - t0

    # 中枢边界是信号定义的核心中间量，一并指纹化：
    # 若信号相同但中枢不同，说明差异只是被过滤掉了，仍是隐患
    zsig = None
    if not zones.empty:
        num = zones.select_dtypes(include=[np.number])
        zsig = float(np.nansum(num.to_numpy(dtype=float)))

    return {"n_bars": int(len(df_ltf)), "n_signals": len(raw),
            "signal_idx": raw, "zone_checksum": zsig,
            "n_zones": int(len(zones)), "compute_s": round(dt, 4)}


def run(tag: str) -> None:
    d = slice_dir() / "parity_slices"
    res = {"tag": tag,
           "python": sys.version.split()[0],
           "pandas": pd.__version__,
           "numpy": np.__version__}
    per = {}
    for s in SYMS:
        f_ltf, f_htf = d / f"{s}_{LTF}.csv", d / f"{s}_{HTF}.csv"
        if not f_ltf.exists():
            print(f"  {s}: 缺切片 {f_ltf}")
            continue
        df_ltf = pd.read_csv(f_ltf)
        df_htf = pd.read_csv(f_htf) if f_htf.exists() else pd.DataFrame()
        # date 存的是时间戳字符串，chanlun 的 kline builder 要真 datetime；
        # timestamp 是毫秒整数，保持数值不动
        for df in (df_ltf, df_htf):
            if not df.empty and "date" in df.columns:
                df["date"] = pd.to_datetime(df["date"], utc=True)
        r = signals(df_ltf, df_htf)
        per[s] = r
        print(f"  {s}: {r['n_signals']} 信号 · {r['n_zones']} 中枢 · "
              f"{r['compute_s']}s · zone_checksum={r['zone_checksum']}")
    res["per_symbol"] = per

    out = slice_dir() / f"parity_env_{tag}.json"
    out.write_text(json.dumps(res, indent=2, ensure_ascii=False))
    print(f"\n[{tag}] python {res['python']} pandas {res['pandas']} "
          f"numpy {res['numpy']}")
    print(f"产物写入 {out}")


def compare(a: str, b: str) -> None:
    d = slice_dir()
    ra = json.loads((d / f"parity_env_{a}.json").read_text())
    rb = json.loads((d / f"parity_env_{b}.json").read_text())
    print(f"{a}: python {ra['python']} pandas {ra['pandas']} numpy {ra['numpy']}")
    print(f"{b}: python {rb['python']} pandas {rb['pandas']} numpy {rb['numpy']}")
    print("\n########## 信号集合是否逐一相同 ##########")
    ok = True
    for s in SYMS:
        pa, pb = ra["per_symbol"].get(s), rb["per_symbol"].get(s)
        if not pa or not pb:
            print(f"  {s}: 缺结果，跳过")
            continue
        same_sig = pa["signal_idx"] == pb["signal_idx"]
        same_zone = pa["n_zones"] == pb["n_zones"]
        # checksum 是浮点求和，允许相对 1e-9 的差；超出即为真实分歧
        za, zb = pa["zone_checksum"], pb["zone_checksum"]
        same_cs = (za is None and zb is None) or (
            za is not None and zb is not None
            and abs(za - zb) <= 1e-9 * max(1.0, abs(za)))
        ok = ok and same_sig and same_zone and same_cs
        print(f"  {s}: 信号 {'一致' if same_sig else '不一致'}"
              f"（{pa['n_signals']} vs {pb['n_signals']}）· "
              f"中枢 {'一致' if same_zone else '不一致'}"
              f"（{pa['n_zones']} vs {pb['n_zones']}）· "
              f"checksum {'一致' if same_cs else '不一致'}")
        if not same_sig:
            sa, sb = set(pa["signal_idx"]), set(pb["signal_idx"])
            print(f"      仅 {a} 有: {sorted(sa - sb)[:10]}")
            print(f"      仅 {b} 有: {sorted(sb - sa)[:10]}")
        print(f"      耗时 {pa['compute_s']}s vs {pb['compute_s']}s")
    print(f"\n结论：{'两环境等价，可直接在容器内算信号' if ok else '存在分歧，需把信号计算固定在单一环境'}")


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--dump", action="store_true")
    ap.add_argument("--run", action="store_true")
    ap.add_argument("--tag", default="venv")
    ap.add_argument("--compare", nargs=2, metavar=("A", "B"))
    args = ap.parse_args()
    if args.dump:
        dump()
    if args.run:
        run(args.tag)
    if args.compare:
        compare(*args.compare)


if __name__ == "__main__":
    main()