Bybit APIでローソク足（OHLCV）を全期間取得し
Parquetに保存するPython実装
1000本上限のページネーションと差分更新

# 本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
import pandas as pd
from pybit.unified_trading import HTTP

http = HTTP(testnet=False)          # 公開エンドポイントなのでキー不要

res = http.get_kline(
    category="linear",
    symbol="BTCUSDT",
    interval="60",                  # 1,3,5,15,30,60,120,240,360,720,D,W,M
    limit=1000,                     # 1〜1000（既定200）
)

# V5のklineは [startTime, open, high, low, close, volume, turnover] の文字列配列
df = pd.DataFrame(
    res["result"]["list"],
    columns=["ts", "open", "high", "low", "close", "volume", "turnover"],
)
df = df.astype(float)
df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
df = df.set_index("ts").sort_index()   # 返り値は「新しい順」なので並べ替える

print(df.tail())
print(f"\n{len(df)}本  {df.index[0]} 〜 {df.index[-1]}")

"""kline_fetch.py — 全期間のローソク足を取り切る（逆順ページネーション）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time
from datetime import datetime, timezone

import pandas as pd
from pybit.unified_trading import HTTP

MAX_LIMIT = 1000

# 足の長さ（ミリ秒）。欠損検出とページ送りの安全確認に使う
INTERVAL_MS: dict[str, int] = {
    "1": 60_000, "3": 180_000, "5": 300_000, "15": 900_000, "30": 1_800_000,
    "60": 3_600_000, "120": 7_200_000, "240": 14_400_000, "360": 21_600_000,
    "720": 43_200_000, "D": 86_400_000, "W": 604_800_000,
    # "M" は月ごとに長さが変わるので辞書に入れない
}

KLINE_COLS = ["ts", "open", "high", "low", "close", "volume", "turnover"]


def to_ms(dt: str | datetime) -> int:
    if isinstance(dt, str):
        dt = datetime.fromisoformat(dt)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return int(dt.timestamp() * 1000)


def fetch_klines(
    http: HTTP,
    symbol: str,
    interval: str = "60",
    start: str | datetime | None = None,
    end: str | datetime | None = None,
    category: str = "linear",
    sleep: float = 0.12,
    verbose: bool = True,
) -> pd.DataFrame:
    """startからendまでのローソク足を全件取得してDataFrameで返す。

    startを省略すると、取引所が持っている最古の足まで遡る。
    """
    start_ms = to_ms(start) if start else 0
    cursor_end = to_ms(end) if end else int(time.time() * 1000)

    chunks: list[list[list[str]]] = []
    pages = 0
    prev_oldest: int | None = None

    while True:
        res = http.get_kline(
            category=category, symbol=symbol, interval=interval,
            end=cursor_end, limit=MAX_LIMIT,
        )
        if res["retCode"] != 0:
            raise RuntimeError(f"get_kline失敗: {res['retCode']} {res['retMsg']}")

        rows = res["result"]["list"]
        if not rows:
            break                       # これ以上古いデータが無い

        # 返り値は降順なので、最後の要素が最古
        oldest = int(rows[-1][0])

        # 無限ループ防止: カーソルが進まなくなったら打ち切る
        if prev_oldest is not None and oldest >= prev_oldest:
            if verbose:
                print(f"  [stop] カーソルが進みません（oldest={oldest}）")
            break
        prev_oldest = oldest

        chunks.append(rows)
        pages += 1
        if verbose and pages % 10 == 0:
            got = sum(len(c) for c in chunks)
            print(f"  {pages}ページ / {got:,}本  最古 "
                  f"{datetime.fromtimestamp(oldest / 1000, timezone.utc):%Y-%m-%d %H:%M}")

        if oldest <= start_ms or len(rows) < MAX_LIMIT:
            break

        cursor_end = oldest - 1
        time.sleep(sleep)               # IP制限（5秒600リクエスト）に配慮

    if not chunks:
        return pd.DataFrame(columns=KLINE_COLS).set_index("ts")

    flat = [r for chunk in chunks for r in chunk]
    df = pd.DataFrame(flat, columns=KLINE_COLS).astype(float)
    df["ts"] = pd.to_datetime(df["ts"].astype("int64"), unit="ms", utc=True)

    df = (df.drop_duplicates(subset="ts")     # ページ境界の重複を除去
            .set_index("ts")
            .sort_index())                    # 時系列昇順に直す

    if start_ms:
        df = df[df.index >= pd.Timestamp(start_ms, unit="ms", tz="UTC")]

    if verbose:
        print(f"[完了] {symbol} {interval} : {len(df):,}本  "
              f"{df.index[0]} 〜 {df.index[-1]}")
    return df


if __name__ == "__main__":
    http = HTTP(testnet=False)
    df = fetch_klines(http, "BTCUSDT", interval="60", start="2024-01-01")
    print(df.head())
    print(df.describe())

"""drop_unconfirmed.py — 未確定足の判定と除外
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import time

import pandas as pd

from kline_fetch import INTERVAL_MS


def drop_unconfirmed(df: pd.DataFrame, interval: str,
                     now_ms: int | None = None) -> pd.DataFrame:
    """まだ閉じていない足を落とす。

    足が確定しているのは「足の開始時刻 + 足の長さ <= 現在時刻」のとき。
    """
    if df.empty:
        return df
    step = INTERVAL_MS.get(interval)
    if step is None:
        # 月足など長さが可変の場合は、単純に最後の1本を落とす
        return df.iloc[:-1]

    now = now_ms if now_ms is not None else int(time.time() * 1000)
    cutoff = pd.Timestamp(now - step, unit="ms", tz="UTC")
    return df[df.index <= cutoff]


def is_bar_closed(bar_start: pd.Timestamp, interval: str) -> bool:
    step = INTERVAL_MS.get(interval)
    if step is None:
        return False
    return bar_start.value // 1_000_000 + step <= int(time.time() * 1000)


if __name__ == "__main__":
    from pybit.unified_trading import HTTP

    from kline_fetch import fetch_klines

    http = HTTP(testnet=False)
    raw = fetch_klines(http, "BTCUSDT", interval="60", start="2026-07-01",
                       verbose=False)
    clean = drop_unconfirmed(raw, "60")

    print(f"取得       : {len(raw)}本  最終足 {raw.index[-1]}")
    print(f"確定足のみ : {len(clean)}本  最終足 {clean.index[-1]}")
    print(f"除外       : {len(raw) - len(clean)}本")

"""timezone_util.py — UTC基準で保持し、表示のみJSTへ変換する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

JST = "Asia/Tokyo"


def to_jst(df: pd.DataFrame) -> pd.DataFrame:
    """表示用。indexをJSTに変換したコピーを返す（計算には使わない）。"""
    out = df.copy()
    out.index = out.index.tz_convert(JST)
    return out


def assert_utc(df: pd.DataFrame) -> None:
    """データの取り違えを早期に検出するアサーション。"""
    if not isinstance(df.index, pd.DatetimeIndex):
        raise TypeError("indexがDatetimeIndexではありません")
    if df.index.tz is None:
        raise ValueError("naiveなDatetimeIndexです。UTCを明示してください")
    if str(df.index.tz) not in ("UTC", "utc"):
        raise ValueError(f"indexがUTCではありません: {df.index.tz}")
    if not df.index.is_monotonic_increasing:
        raise ValueError("indexが昇順ではありません（sort_index()を忘れています）")

"""gap_check.py — 欠損足の検出とレポート
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

from kline_fetch import INTERVAL_MS

PANDAS_FREQ = {
    "1": "1min", "3": "3min", "5": "5min", "15": "15min", "30": "30min",
    "60": "1h", "120": "2h", "240": "4h", "360": "6h", "720": "12h",
    "D": "1D", "W": "1W",
}


def find_gaps(df: pd.DataFrame, interval: str) -> pd.DataFrame:
    """欠損している足の開始時刻を列挙する。"""
    step = INTERVAL_MS.get(interval)
    if step is None or df.empty:
        return pd.DataFrame(columns=["missing_from", "missing_to", "bars"])

    freq = PANDAS_FREQ[interval]
    full = pd.date_range(df.index[0], df.index[-1], freq=freq, tz="UTC")
    missing = full.difference(df.index)
    if len(missing) == 0:
        return pd.DataFrame(columns=["missing_from", "missing_to", "bars"])

    # 連続する欠損を1区間にまとめる
    groups, start, prev = [], missing[0], missing[0]
    for t in missing[1:]:
        if (t - prev).total_seconds() * 1000 > step:
            groups.append((start, prev))
            start = t
        prev = t
    groups.append((start, prev))

    return pd.DataFrame(
        [{"missing_from": a, "missing_to": b,
          "bars": int((b - a).total_seconds() * 1000 / step) + 1}
         for a, b in groups]
    )


def fill_gaps(df: pd.DataFrame, interval: str, method: str = "ffill") -> pd.DataFrame:
    """欠損足を埋める。用途によって埋め方を変える。

    method="ffill" : 直前の終値で埋め、出来高は0にする（指標計算向け）
    method="none"  : 埋めない（バックテストは「取引が無かった」ままが正しい）
    """
    if method == "none" or df.empty:
        return df

    out = df.reindex(pd.date_range(df.index[0], df.index[-1],
                                   freq=PANDAS_FREQ[interval], tz="UTC"))
    # 価格は直前の終値で埋める
    out["close"] = out["close"].ffill()
    for col in ("open", "high", "low"):
        out[col] = out[col].fillna(out["close"])
    # 出来高は0（取引が無かったのだから0が正しい。ffillは誤り）
    for col in ("volume", "turnover"):
        if col in out.columns:
            out[col] = out[col].fillna(0.0)
    return out


if __name__ == "__main__":
    from pybit.unified_trading import HTTP

    from kline_fetch import fetch_klines

    http = HTTP(testnet=False)
    df = fetch_klines(http, "BTCUSDT", interval="1", start="2026-07-25",
                      verbose=False)

    gaps = find_gaps(df, "1")
    print(f"取得 {len(df):,}本 / 欠損区間 {len(gaps)}件")
    if not gaps.empty:
        print(gaps.head(10).to_string())
        print(f"欠損足の合計: {gaps['bars'].sum()}本 "
              f"({gaps['bars'].sum() / (len(df) + gaps['bars'].sum()):.2%})")

"""kline_store.py — ローソク足のParquetストア（差分更新つき）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install pyarrow
"""
from __future__ import annotations

from pathlib import Path

import pandas as pd
from pybit.unified_trading import HTTP

from drop_unconfirmed import drop_unconfirmed
from gap_check import find_gaps
from kline_fetch import fetch_klines


class KlineStore:
    """symbol / interval ごとに1ファイル。確定足のみを保存する。"""

    def __init__(self, root: str | Path = "data/kline", category: str = "linear"):
        self.root = Path(root)
        self.category = category
        self.root.mkdir(parents=True, exist_ok=True)

    def path(self, symbol: str, interval: str) -> Path:
        return self.root / f"{self.category}_{symbol}_{interval}.parquet"

    # ---- 読み込み ------------------------------------------------------
    def load(self, symbol: str, interval: str) -> pd.DataFrame:
        p = self.path(symbol, interval)
        if not p.exists():
            return pd.DataFrame()
        return pd.read_parquet(p)

    # ---- 更新 ----------------------------------------------------------
    def update(self, http: HTTP, symbol: str, interval: str = "60",
               start: str = "2020-01-01", verbose: bool = True) -> pd.DataFrame:
        """未取得の区間だけを取得して追記する。"""
        existing = self.load(symbol, interval)

        if existing.empty:
            fetch_from = start
            if verbose:
                print(f"[新規] {symbol} {interval} を {start} から取得します")
        else:
            # 最終足の「次の足」から取得すればよい
            last = existing.index[-1]
            fetch_from = last.isoformat()
            if verbose:
                print(f"[差分] {symbol} {interval} : 既存{len(existing):,}本 "
                      f"(最終 {last}) 以降を取得")

        fresh = fetch_klines(http, symbol, interval=interval, start=fetch_from,
                             category=self.category, verbose=verbose)
        fresh = drop_unconfirmed(fresh, interval)      # 未確定足は保存しない

        if fresh.empty:
            if verbose:
                print("  新しい確定足はありません")
            return existing

        merged = (pd.concat([existing, fresh])
                    if not existing.empty else fresh)
        merged = merged[~merged.index.duplicated(keep="last")].sort_index()

        # 原子的に書き換える（書き込み中の中断で壊さない）
        p = self.path(symbol, interval)
        tmp = p.with_suffix(".parquet.tmp")
        merged.to_parquet(tmp, compression="zstd")
        tmp.replace(p)

        if verbose:
            added = len(merged) - len(existing)
            size_mb = p.stat().st_size / 1024 / 1024
            print(f"  +{added:,}本 -> 計{len(merged):,}本  ({size_mb:.1f} MB)")
        return merged

    # ---- 検査 ----------------------------------------------------------
    def inspect(self, symbol: str, interval: str) -> None:
        df = self.load(symbol, interval)
        if df.empty:
            print(f"{symbol} {interval}: データなし")
            return
        gaps = find_gaps(df, interval)
        print(f"{symbol} {interval}: {len(df):,}本  "
              f"{df.index[0]:%Y-%m-%d} 〜 {df.index[-1]:%Y-%m-%d}  "
              f"欠損区間{len(gaps)}件")


if __name__ == "__main__":
    http = HTTP(testnet=False)
    store = KlineStore()

    for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
        store.update(http, sym, interval="60", start="2023-01-01")
        store.inspect(sym, "60")

"""query_duckdb.py — Parquet群にSQLを直接投げて全銘柄を横断集計する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

  pip install duckdb
"""
from __future__ import annotations

import duckdb

# ファイル名から銘柄名を取り出しつつ、全Parquetを1つのテーブルとして扱う
BASE = """
    SELECT
        regexp_extract(filename, '_([A-Z0-9]+)_', 1) AS symbol,
        ts, open, high, low, close, volume, turnover
    FROM read_parquet('data/kline/linear_*_60.parquet', filename = true)
"""


def monthly_stats() -> None:
    """銘柄×月ごとのボラティリティと出来高を集計する。"""
    q = f"""
    WITH k AS ({BASE}),
    r AS (
        SELECT symbol, ts, turnover,
               ln(close / lag(close) OVER (PARTITION BY symbol ORDER BY ts)) AS ret
        FROM k
    )
    SELECT
        symbol,
        date_trunc('month', ts)                       AS month,
        count(*)                                      AS bars,
        round(stddev_samp(ret) * sqrt(24 * 365) * 100, 1) AS ann_vol_pct,
        round(sum(turnover) / 1e9, 2)                 AS turnover_bn
    FROM r
    WHERE ret IS NOT NULL
    GROUP BY symbol, month
    ORDER BY month DESC, turnover_bn DESC
    LIMIT 20
    """
    print(duckdb.sql(q))


def correlation_matrix() -> None:
    """銘柄間の日次リターン相関（ペアトレードの候補探索・記事34の前段）。"""
    q = f"""
    WITH k AS ({BASE}),
    d AS (
        SELECT symbol, date_trunc('day', ts) AS d, last(close ORDER BY ts) AS close
        FROM k GROUP BY symbol, d
    ),
    r AS (
        SELECT symbol, d,
               ln(close / lag(close) OVER (PARTITION BY symbol ORDER BY d)) AS ret
        FROM d
    )
    SELECT a.symbol AS sym_a, b.symbol AS sym_b,
           round(corr(a.ret, b.ret), 3) AS correlation,
           count(*) AS n
    FROM r a JOIN r b ON a.d = b.d AND a.symbol < b.symbol
    WHERE a.ret IS NOT NULL AND b.ret IS NOT NULL
    GROUP BY sym_a, sym_b
    HAVING count(*) >= 60
    ORDER BY correlation DESC
    LIMIT 15
    """
    print(duckdb.sql(q))


if __name__ == "__main__":
    print("=== 銘柄×月の年率ボラと売買代金 ===")
    monthly_stats()
    print("\n=== 日次リターン相関の高いペア ===")
    correlation_matrix()

"""bulk_fetch.py — 複数銘柄を安全に並列取得する（IP制限に配慮）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from pybit.unified_trading import HTTP

from kline_store import KlineStore

# IP制限は5秒600リクエスト。安全側に倒して秒間30リクエストを上限にする。
# 履歴取得は急ぐ処理ではないので、余裕を持たせるほうが確実。
MAX_RPS = 30
WORKERS = 4


class RateGate:
    """秒間リクエスト数を制限する簡易ゲート（詳細な実装は記事05）。"""

    def __init__(self, rps: int):
        self.min_interval = 1.0 / rps
        self._lock = threading.Lock()
        self._next_at = 0.0

    def wait(self) -> None:
        with self._lock:
            now = time.monotonic()
            if now < self._next_at:
                time.sleep(self._next_at - now)
                now = time.monotonic()
            self._next_at = now + self.min_interval


gate = RateGate(MAX_RPS)


class GatedHTTP:
    """HTTPをラップして、全リクエストをゲート経由にする。"""

    def __init__(self, http: HTTP):
        self._http = http

    def __getattr__(self, name):
        attr = getattr(self._http, name)
        if not callable(attr):
            return attr

        def wrapped(*args, **kwargs):
            gate.wait()
            return attr(*args, **kwargs)

        return wrapped


def liquid_symbols(http: HTTP, min_turnover: float = 20_000_000) -> list[str]:
    rows = http.get_tickers(category="linear")["result"]["list"]
    rows = [r for r in rows if r["symbol"].endswith("USDT")
            and float(r["turnover24h"]) >= min_turnover]
    rows.sort(key=lambda r: float(r["turnover24h"]), reverse=True)
    return [r["symbol"] for r in rows]


def main() -> None:
    raw_http = HTTP(testnet=False)
    http = GatedHTTP(raw_http)
    store = KlineStore()

    symbols = liquid_symbols(raw_http)
    print(f"{len(symbols)}銘柄を取得します（並列{WORKERS} / 上限{MAX_RPS}rps）")

    started = time.time()
    ok, ng = 0, []

    with ThreadPoolExecutor(max_workers=WORKERS) as pool:
        futures = {
            pool.submit(store.update, http, sym, "60", "2024-01-01", False): sym
            for sym in symbols
        }
        for i, fut in enumerate(as_completed(futures), 1):
            sym = futures[fut]
            try:
                df = fut.result()
                ok += 1
                print(f"[{i:>3}/{len(symbols)}] {sym:<12} {len(df):>7,}本")
            except Exception as e:
                ng.append((sym, f"{type(e).__name__}: {e}"))
                print(f"[{i:>3}/{len(symbols)}] {sym:<12} 失敗: {e}")

    print(f"\n完了 {ok}銘柄 / 失敗 {len(ng)}件 / {time.time() - started:.0f}秒")
    for sym, err in ng:
        print(f"  {sym}: {err}")


if __name__ == "__main__":
    main()

"""resample.py — OHLCVを正しくリサンプリングする（JST日足の作り方も含む）
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

OHLCV_AGG = {
    "open": "first",
    "high": "max",
    "low": "min",
    "close": "last",
    "volume": "sum",
    "turnover": "sum",
}


def resample_ohlcv(df: pd.DataFrame, freq: str, tz: str | None = None,
                   drop_partial: bool = True) -> pd.DataFrame:
    """OHLCVを上位足に集約する。

    freq       : pandasの頻度文字列（"4h", "1D", "1W" など）
    tz         : "Asia/Tokyo" を渡すとJST基準の区切りになる（日足・週足で重要）
    drop_partial: 区間が埋まっていない最後の足を落とす
    """
    if df.empty:
        return df

    work = df.copy()
    if tz:
        work.index = work.index.tz_convert(tz)

    cols = {k: v for k, v in OHLCV_AGG.items() if k in work.columns}
    out = work.resample(freq, label="left", closed="left").agg(cols)

    # 取引が無かった区間は openがNaNになる。これは「足が存在しない」ので落とす
    out = out.dropna(subset=["open"])

    if drop_partial and len(out) > 0:
        # 最後の区間が現在進行中なら落とす
        step = pd.tseries.frequencies.to_offset(freq)
        last_start = out.index[-1]
        if last_start + step > work.index[-1] + pd.Timedelta(milliseconds=1):
            out = out.iloc[:-1]

    if tz:
        out.index = out.index.tz_convert("UTC")   # 保持はUTCに戻す
    return out


if __name__ == "__main__":
    from pybit.unified_trading import HTTP

    from kline_fetch import fetch_klines

    http = HTTP(testnet=False)
    m1 = fetch_klines(http, "BTCUSDT", interval="1", start="2026-07-20",
                      verbose=False)
    print(f"1分足: {len(m1):,}本")

    h4 = resample_ohlcv(m1, "4h")
    print(f"\n4時間足: {len(h4)}本")
    print(h4.tail(3).to_string())

    # UTC日足とJST日足の違いを見る
    d_utc = resample_ohlcv(m1, "1D")
    d_jst = resample_ohlcv(m1, "1D", tz="Asia/Tokyo")
    print(f"\nUTC日足 終値: {d_utc['close'].tail(3).tolist()}")
    print(f"JST日足 終値: {d_jst['close'].tail(3).tolist()}")
    print("  -> 同じ日でも終値が違う（区切り時刻が9時間ずれるため）")

"""price_kinds.py — 4種類のローソク足を取得して比較する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd
from pybit.unified_trading import HTTP

# mark / index / premium は5フィールド（出来高なし）
PRICE_ONLY_COLS = ["ts", "open", "high", "low", "close"]
FULL_COLS = ["ts", "open", "high", "low", "close", "volume", "turnover"]


def _to_df(rows: list[list[str]]) -> pd.DataFrame:
    """フィールド数から自動判別してDataFrame化する（仕様差を吸収）。"""
    if not rows:
        return pd.DataFrame()
    cols = FULL_COLS if len(rows[0]) >= 7 else PRICE_ONLY_COLS
    df = pd.DataFrame([r[: len(cols)] for r in rows], columns=cols).astype(float)
    df["ts"] = pd.to_datetime(df["ts"].astype("int64"), unit="ms", utc=True)
    return df.set_index("ts").sort_index()


def fetch_all_price_kinds(http: HTTP, symbol: str = "BTCUSDT",
                          interval: str = "60", limit: int = 200) -> pd.DataFrame:
    kw = dict(category="linear", symbol=symbol, interval=interval, limit=limit)

    last = _to_df(http.get_kline(**kw)["result"]["list"])
    mark = _to_df(http.get_mark_price_kline(**kw)["result"]["list"])
    index = _to_df(http.get_index_price_kline(**kw)["result"]["list"])
    prem = _to_df(http.get_premium_index_price_kline(**kw)["result"]["list"])

    out = pd.DataFrame({
        "last": last["close"],
        "mark": mark["close"],
        "index": index["close"],
        "premium_idx": prem["close"],
        "volume": last["volume"],
    })
    # ベーシス: 無期限価格が現物指数からどれだけ乖離しているか
    out["basis_bps"] = (out["last"] - out["index"]) / out["index"] * 10_000
    # マーク価格と最終価格の乖離（清算判定とのズレの目安）
    out["mark_gap_bps"] = (out["last"] - out["mark"]) / out["mark"] * 10_000
    return out.dropna()


if __name__ == "__main__":
    http = HTTP(testnet=False)
    df = fetch_all_price_kinds(http)

    print(df.tail(5).to_string())
    print(f"\nベーシス(bps)      : 平均 {df['basis_bps'].mean():+.2f} / "
          f"標準偏差 {df['basis_bps'].std():.2f} / "
          f"最大 {df['basis_bps'].abs().max():.2f}")
    print(f"マーク乖離(bps)    : 平均 {df['mark_gap_bps'].mean():+.2f} / "
          f"最大 {df['mark_gap_bps'].abs().max():.2f}")
    print("\n※ マーク乖離が大きい局面では、最終価格ベースのストップ判定と"
          "実際の清算判定がズレる")