※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。
| 価格 | 何を表すか | 使う場面 |
|---|---|---|
| 最終価格 （Last Price） |
その取引所で実際に約定した最新価格 | 利確のトリガー、板の中心 |
| マーク価格 （Mark Price） |
インデックス価格に基づく公正価格 | 清算判定・含み損益の評価（記事22） |
| インデックス価格 （Index Price） |
複数の現物取引所の加重平均 | ベーシスの基準、マーク価格の元 |
最終価格が瞬間的に急落しても、マーク価格が動いていなければ清算されません。逆もあります。「板が薄い時間帯にヒゲで刈られた」ように見える場面でも、実際にはマーク価格基準で判定されています。
この違いを理解していないと、損切りのトリガー選択（記事14の triggerBy）やバックテストの清算再現（記事31）が誤ります。損切りは MarkPrice、利確は LastPrice が基本です。
"""price_kinds.py — 3種類の価格を同時に取得し、乖離を測る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
@dataclass(frozen=True)
class PriceSnapshot:
symbol: str
last: Decimal
mark: Decimal
index: Decimal
bid: Decimal
ask: Decimal
ts_ms: int
@property
def mid(self) -> Decimal:
return (self.bid + self.ask) / 2 if self.bid and self.ask else self.last
@property
def basis_bps(self) -> Decimal:
"""最終価格 − インデックス価格。無期限のプレミアム。"""
if self.index <= 0:
return Decimal(0)
return (self.last - self.index) / self.index * 10_000
@property
def mark_gap_bps(self) -> Decimal:
"""最終価格 − マーク価格。清算判定とのズレの目安。"""
if self.mark <= 0:
return Decimal(0)
return (self.last - self.mark) / self.mark * 10_000
@property
def mark_index_bps(self) -> Decimal:
"""マーク価格 − インデックス価格。"""
if self.index <= 0:
return Decimal(0)
return (self.mark - self.index) / self.index * 10_000
def line(self) -> str:
return (f"{self.symbol:<12}{self.last:>13}{self.mark:>13}"
f"{self.index:>13}{self.basis_bps:>+10.2f}b"
f"{self.mark_gap_bps:>+10.2f}b")
def _d(v, default: str = "0") -> Decimal:
if v in (None, "", "null"):
return Decimal(default)
return Decimal(str(v))
def fetch_prices(http, symbol: str,
category: str = "linear") -> PriceSnapshot:
res = http.get_tickers(category=category, symbol=symbol)
if res["retCode"] != 0:
raise RuntimeError(f"get_tickers失敗: {res['retMsg']}")
t = res["result"]["list"][0]
return PriceSnapshot(
symbol=t["symbol"],
last=_d(t.get("lastPrice")),
mark=_d(t.get("markPrice") or t.get("lastPrice")),
index=_d(t.get("indexPrice") or t.get("lastPrice")),
bid=_d(t.get("bid1Price")),
ask=_d(t.get("ask1Price")),
ts_ms=int(res.get("time") or 0),
)
def scan_basis(http, category: str = "linear",
min_turnover: float = 20_000_000,
top: int = 15) -> list[PriceSnapshot]:
"""全銘柄のベーシスをスキャンする（1リクエスト）。"""
res = http.get_tickers(category=category)
if res["retCode"] != 0:
raise RuntimeError(f"get_tickers失敗: {res['retMsg']}")
out = []
for t in res["result"]["list"]:
if float(t.get("turnover24h") or 0) < min_turnover:
continue
index = _d(t.get("indexPrice"))
if index <= 0:
continue
out.append(PriceSnapshot(
symbol=t["symbol"], last=_d(t.get("lastPrice")),
mark=_d(t.get("markPrice")), index=index,
bid=_d(t.get("bid1Price")), ask=_d(t.get("ask1Price")),
ts_ms=int(res.get("time") or 0),
))
return sorted(out, key=lambda p: -abs(p.basis_bps))[:top]
if __name__ == "__main__":
from client import make_client
c = make_client()
print(f"{'symbol':<12}{'最終':>13}{'マーク':>13}{'インデックス':>13}"
f"{'ベーシス':>11}{'マーク乖離':>11}")
print("-" * 74)
for sym in ("BTCUSDT", "ETHUSDT", "SOLUSDT"):
print(fetch_prices(c.http, sym).line())
print("\n=== ベーシスが大きい銘柄（過熱の指標） ===")
for p in scan_basis(c.http):
print(p.line())
print("\n※ ベーシスが大きくプラス = 先物が現物より高い = 買いが過熱。")
print(" この状態は資金調達率がプラスになる要因（記事10）。")
print(" マーク乖離が大きい局面では、最終価格ベースのストップ判定と")
print(" 実際の清算判定がズレる（記事22）。")ベーシス＝無期限価格 − 現物指数です。この乖離が資金調達率を通じて是正される、というのが無期限先物の設計思想です。
"""basis_history.py — ベーシスの時系列を作り、統計を取る
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
3種類のklineを組み合わせる（記事07）。
mark / index / premium は5フィールド（出来高なし）である点に注意。
"""
from __future__ import annotations
import pandas as pd
FULL_COLS = ["ts", "open", "high", "low", "close", "volume", "turnover"]
PRICE_COLS = ["ts", "open", "high", "low", "close"]
def _to_df(rows: list[list[str]]) -> pd.DataFrame:
"""フィールド数から自動判別してDataFrame化する（記事07と同じ）。"""
if not rows:
return pd.DataFrame()
cols = FULL_COLS if len(rows[0]) >= 7 else PRICE_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_basis_series(http, symbol: str = "BTCUSDT",
interval: str = "60", limit: int = 1000,
category: str = "linear") -> pd.DataFrame:
"""最終・マーク・インデックス・プレミアム指数を結合する。"""
kw = dict(category=category, 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"])
df = pd.DataFrame({
"last": last["close"],
"mark": mark["close"],
"index": index["close"],
"premium_index": prem["close"],
"volume": last.get("volume"),
}).dropna(subset=["last", "index"])
df["basis_bps"] = (df["last"] - df["index"]) / df["index"] * 10_000
df["mark_gap_bps"] = (df["last"] - df["mark"]) / df["mark"] * 10_000
df["mark_index_bps"] = (df["mark"] - df["index"]) / df["index"] * 10_000
return df
def basis_stats(df: pd.DataFrame) -> dict:
b = df["basis_bps"].dropna()
if b.empty:
return {"error": "データなし"}
return {
"件数": len(b),
"平均(bps)": round(float(b.mean()), 2),
"中央値(bps)": round(float(b.median()), 2),
"標準偏差(bps)": round(float(b.std()), 2),
"プラスの割合": f"{(b > 0).mean():.1%}",
"5%点": round(float(b.quantile(0.05)), 2),
"95%点": round(float(b.quantile(0.95)), 2),
"最大乖離": round(float(b.abs().max()), 2),
"自己相関(1)": round(float(b.autocorr(1)), 3),
}
def basis_percentile(df: pd.DataFrame, window: int = 720) -> pd.Series:
"""ベーシスの相対水準（0〜1）。極端な局面の検出に使う。"""
return df["basis_bps"].rolling(window, min_periods=window // 4).rank(pct=True)
if __name__ == "__main__":
from client import make_client
c = make_client()
for symbol in ("BTCUSDT", "ETHUSDT"):
df = fetch_basis_series(c.http, symbol)
if df.empty:
continue
print(f"=== {symbol} のベーシス（{len(df)}本の1時間足） ===")
for k, v in basis_stats(df).items():
print(f" {k:<14}: {v}")
pct = basis_percentile(df)
latest = df.iloc[-1]
print(f"\n 直近: ベーシス {latest['basis_bps']:+.2f} bps "
f"（過去30日の {pct.iloc[-1]:.0%} 分位）")
print()
print("※ ベーシスが95%分位を超える局面は『買いが過熱』の状態。")
print(" 資金調達率も高くなりやすく、キャリー戦略の好機だが")
print(" 同時に急変のリスクも高い（記事35）。")| パラメータ | 必須 | 内容 |
|---|---|---|
| category | 必須 | linear / inverse（現物にOIはない） |
| symbol | 必須 | 全銘柄一括は不可。1銘柄ずつ |
| intervalTime | 必須 | 5min / 15min / 30min / 1h / 4h / 1d |
| limit | 任意 | 1〜200（既定50） |
| cursor | 任意 | ページネーション |
"""open_interest.py — OI（未決済建玉）の取得と分析
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
OIは「まだ決済されていない契約の総数」。
増加 = 新規参入、減少 = 決済。価格と組み合わせて需給を読む。
"""
from __future__ import annotations
import time
from datetime import datetime, timezone
import pandas as pd
VALID_INTERVALS = ("5min", "15min", "30min", "1h", "4h", "1d")
MAX_LIMIT = 200
def fetch_open_interest(http, symbol: str, interval: str = "1h",
start: str | None = None,
category: str = "linear",
max_pages: int = 200,
sleep: float = 0.1,
verbose: bool = False) -> pd.DataFrame:
"""OIの履歴を取得する（cursorページネーション）。"""
if interval not in VALID_INTERVALS:
raise ValueError(f"intervalTimeは {VALID_INTERVALS} のいずれか")
start_ms = None
if start:
start_ms = int(pd.Timestamp(start, tz="UTC").timestamp() * 1000)
rows: list[dict] = []
cursor = None
for page in range(max_pages):
kw: dict = {"category": category, "symbol": symbol,
"intervalTime": interval, "limit": MAX_LIMIT}
if cursor:
kw["cursor"] = cursor
res = http.get_open_interest(**kw)
if res["retCode"] != 0:
raise RuntimeError(f"get_open_interest失敗: {res['retMsg']}")
page_rows = res["result"]["list"]
if not page_rows:
break
rows.extend(page_rows)
oldest = min(int(r["timestamp"]) for r in page_rows)
if start_ms and oldest <= start_ms:
break
cursor = res["result"].get("nextPageCursor")
if not cursor:
break
if verbose and page % 10 == 0:
print(f" {page + 1}ページ / {len(rows)}件")
time.sleep(sleep)
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
df["ts"] = pd.to_datetime(df["timestamp"].astype("int64"), unit="ms",
utc=True)
df["open_interest"] = pd.to_numeric(df["openInterest"], errors="coerce")
df = (df[["ts", "open_interest"]]
.drop_duplicates(subset="ts")
.set_index("ts").sort_index())
if start_ms:
df = df[df.index >= pd.Timestamp(start_ms, unit="ms", tz="UTC")]
return df
def attach_price(oi: pd.DataFrame, bars: pd.DataFrame) -> pd.DataFrame:
"""OIに価格を結合する（4象限分類の準備）。"""
df = oi.join(bars[["close"]], how="inner")
df["oi_change"] = df["open_interest"].diff()
df["oi_change_pct"] = df["open_interest"].pct_change()
df["price_change"] = df["close"].diff()
df["price_change_pct"] = df["close"].pct_change()
df["oi_value"] = df["open_interest"] * df["close"]
return df.dropna()
def oi_stats(df: pd.DataFrame) -> dict:
oi = df["open_interest"]
return {
"件数": len(oi),
"最新OI": f"{oi.iloc[-1]:,.0f}",
"最新OI額(USDT)": f"{df['oi_value'].iloc[-1]:,.0f}"
if "oi_value" in df else "-",
"期間平均": f"{oi.mean():,.0f}",
"最大": f"{oi.max():,.0f}",
"最小": f"{oi.min():,.0f}",
"直近の変化率": f"{oi.pct_change().iloc[-1]:+.2%}",
"価格との相関": (round(float(df["oi_change_pct"]
.corr(df["price_change_pct"])), 3)
if "price_change_pct" in df else "-"),
}
if __name__ == "__main__":
from client import make_client
from kline_fetch import fetch_klines
c = make_client()
SYMBOL = "BTCUSDT"
oi = fetch_open_interest(c.http, SYMBOL, "1h", start="2026-06-01",
verbose=True)
print(f"\nOI: {len(oi)}件 {oi.index[0]} 〜 {oi.index[-1]}")
bars = fetch_klines(c.http, SYMBOL, interval="60", start="2026-06-01",
verbose=False)
merged = attach_price(oi, bars)
print(f"\n=== {SYMBOL} のOI統計 ===")
for k, v in oi_stats(merged).items():
print(f" {k:<16}: {v}")
print("\n※ OI変化と価格変化の相関だけでは方向は読めない。")
print(" 4象限分類（サンプル4）で意味を持たせる。")OI単体では意味が薄いですが、価格変化と組み合わせると「誰が動いたか」の解釈ができます。これはテクニカル分析で古くから使われる枠組みです。
| 価格 | OI | 解釈 | 含意 |
|---|---|---|---|
| 上昇 | 増加 | 新規の買い | トレンドに資金が入っている |
| 上昇 | 減少 | 売りの決済（踏み上げ） | ショートカバー主導。持続性は低め |
| 下落 | 増加 | 新規の売り | 下方向に資金が入っている |
| 下落 | 減少 | 買いの決済（投げ） | ロング解消主導。売り枯れの兆候 |
"""oi_quadrant.py — 価格×OIの4象限分類
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
注意: この分類は「解釈の枠組み」であり、予測力を保証するものではない。
シグナルとして使うなら記事32の検証を必ず通すこと。
"""
from __future__ import annotations
from enum import Enum
import numpy as np
import pandas as pd
class Quadrant(str, Enum):
NEW_LONG = "新規買い" # 価格↑ OI↑
SHORT_COVER = "売り決済" # 価格↑ OI↓
NEW_SHORT = "新規売り" # 価格↓ OI↑
LONG_LIQUIDATION = "買い決済" # 価格↓ OI↓
NEUTRAL = "中立"
def classify(price_change: float, oi_change: float,
price_threshold: float = 0.0,
oi_threshold: float = 0.0) -> Quadrant:
"""1本分の4象限分類。閾値で微小な変化を中立にできる。"""
if abs(price_change) <= price_threshold or abs(oi_change) <= oi_threshold:
return Quadrant.NEUTRAL
if price_change > 0:
return Quadrant.NEW_LONG if oi_change > 0 else Quadrant.SHORT_COVER
return Quadrant.NEW_SHORT if oi_change > 0 else Quadrant.LONG_LIQUIDATION
def classify_series(df: pd.DataFrame,
price_pct_threshold: float = 0.001,
oi_pct_threshold: float = 0.001) -> pd.DataFrame:
"""時系列全体を分類する。閾値は相対変化率で指定する。"""
out = df.copy()
out["quadrant"] = [
classify(p, o, price_pct_threshold, oi_pct_threshold).value
for p, o in zip(out["price_change_pct"], out["oi_change_pct"])
]
return out
def quadrant_summary(df: pd.DataFrame) -> pd.DataFrame:
"""象限ごとの出現率と、その後のリターンを集計する。
「新規買いの後は上がりやすいか」を実データで確認できる。
"""
out = df.copy()
# 次の足のリターン（未来を見ないよう shift(-1) で明示的に扱う）
out["fwd_return"] = out["close"].pct_change().shift(-1)
agg = out.groupby("quadrant").agg(
件数=("quadrant", "count"),
出現率=("quadrant", lambda s: len(s) / len(out)),
次足リターン平均=("fwd_return", "mean"),
次足リターン中央値=("fwd_return", "median"),
勝率=("fwd_return", lambda s: (s > 0).mean()),
)
return agg.sort_values("件数", ascending=False)
def recent_regime(df: pd.DataFrame, window: int = 24) -> str:
"""直近N本の象限分布から、現在の需給状態を要約する。"""
recent = df["quadrant"].tail(window)
counts = recent.value_counts(normalize=True)
dominant = counts.index[0]
share = counts.iloc[0]
interp = {
Quadrant.NEW_LONG.value:
"新規の買いが優勢。上昇に資金が入っている",
Quadrant.SHORT_COVER.value:
"ショートカバー主導。買い戻しが一巡すると勢いが落ちやすい",
Quadrant.NEW_SHORT.value:
"新規の売りが優勢。下落に資金が入っている",
Quadrant.LONG_LIQUIDATION.value:
"ロング解消主導。投げが出ている",
Quadrant.NEUTRAL.value: "方向感が乏しい",
}
return (f"直近{window}本: {dominant}が{share:.0%}\n"
f" -> {interp.get(dominant, '')}")
if __name__ == "__main__":
from client import make_client
from kline_fetch import fetch_klines
from open_interest import attach_price, fetch_open_interest
c = make_client()
SYMBOL = "BTCUSDT"
oi = fetch_open_interest(c.http, SYMBOL, "1h", start="2026-05-01")
bars = fetch_klines(c.http, SYMBOL, interval="60", start="2026-05-01",
verbose=False)
merged = classify_series(attach_price(oi, bars))
print(f"=== {SYMBOL} の4象限分布（{len(merged)}本） ===")
print(quadrant_summary(merged).to_string(
float_format=lambda v: f"{v:+.4%}" if abs(v) < 1 else f"{v:,.0f}"))
print(f"\n{recent_regime(merged)}")
print("\n--- 注意 ---")
print(" 次足リターンの差が小さければ、この分類に予測力はない。")
print(" 『解釈として納得できる』ことと『使える』ことは別。")
print(" シグナルにするなら記事32のWFA/DSRで検証すること。")"""long_short_ratio.py — ロング/ショート比率の取得と極値検出
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
period: 5min / 15min / 30min / 1h / 4h / 1d
limit : 1〜500（既定50）
注意: buyRatio + sellRatio = 1 になる「アカウント数の比率」。
建玉量の比率ではない点に注意。
"""
from __future__ import annotations
import time
import numpy as np
import pandas as pd
VALID_PERIODS = ("5min", "15min", "30min", "1h", "4h", "1d")
MAX_LIMIT = 500
def fetch_long_short_ratio(http, symbol: str, period: str = "1h",
category: str = "linear",
max_pages: int = 50,
sleep: float = 0.1) -> pd.DataFrame:
if period not in VALID_PERIODS:
raise ValueError(f"periodは {VALID_PERIODS} のいずれか")
rows: list[dict] = []
cursor = None
for _ in range(max_pages):
kw: dict = {"category": category, "symbol": symbol,
"period": period, "limit": MAX_LIMIT}
if cursor:
kw["cursor"] = cursor
res = http.get_long_short_ratio(**kw)
if res["retCode"] != 0:
raise RuntimeError(f"get_long_short_ratio失敗: {res['retMsg']}")
page = res["result"]["list"]
if not page:
break
rows.extend(page)
cursor = res["result"].get("nextPageCursor")
if not cursor:
break
time.sleep(sleep)
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
df["ts"] = pd.to_datetime(df["timestamp"].astype("int64"), unit="ms",
utc=True)
df["buy_ratio"] = pd.to_numeric(df["buyRatio"], errors="coerce")
df["sell_ratio"] = pd.to_numeric(df["sellRatio"], errors="coerce")
df = (df[["ts", "buy_ratio", "sell_ratio"]]
.drop_duplicates(subset="ts")
.set_index("ts").sort_index())
# ロング/ショートの倍率（1.0で均衡）
df["ls_ratio"] = df["buy_ratio"] / df["sell_ratio"].replace(0, np.nan)
df["skew"] = df["buy_ratio"] - df["sell_ratio"] # -1 〜 +1
return df
def detect_extremes(df: pd.DataFrame, window: int = 720,
high_pct: float = 0.95,
low_pct: float = 0.05) -> pd.DataFrame:
"""比率の相対的な極値を検出する。
絶対値で「ロング70%は多い」と判断するのは危険。
銘柄ごとに平常時の水準が違うため、相対分位で見る。
"""
out = df.copy()
out["pct_rank"] = out["skew"].rolling(
window, min_periods=window // 4).rank(pct=True)
out["extreme_long"] = out["pct_rank"] >= high_pct
out["extreme_short"] = out["pct_rank"] <= low_pct
return out
def contrarian_check(df: pd.DataFrame, bars: pd.DataFrame,
forward: int = 24) -> dict:
"""「偏りが極端なとき逆張りが効くか」を実データで確認する。
ポジションが片方に偏りすぎると反転しやすい、という仮説の検証。
"""
merged = detect_extremes(df).join(bars[["close"]], how="inner").dropna()
if merged.empty:
return {"error": "結合できるデータがありません"}
# forward本先までのリターン（未来参照は集計のみに使う）
merged["fwd"] = merged["close"].shift(-forward) / merged["close"] - 1
long_ext = merged.loc[merged["extreme_long"], "fwd"].dropna()
short_ext = merged.loc[merged["extreme_short"], "fwd"].dropna()
base = merged["fwd"].dropna()
def summarise(s: pd.Series, label: str) -> dict:
if len(s) < 10:
return {label: f"サンプル不足({len(s)})"}
return {
f"{label}_件数": len(s),
f"{label}_平均": f"{s.mean():+.3%}",
f"{label}_勝率": f"{(s > 0).mean():.1%}",
}
out = {"全体_平均": f"{base.mean():+.3%}",
"全体_勝率": f"{(base > 0).mean():.1%}"}
out.update(summarise(long_ext, "ロング偏り極大後"))
out.update(summarise(short_ext, "ショート偏り極大後"))
return out
if __name__ == "__main__":
from client import make_client
from kline_fetch import fetch_klines
c = make_client()
SYMBOL = "BTCUSDT"
lsr = fetch_long_short_ratio(c.http, SYMBOL, "1h")
if lsr.empty:
raise SystemExit("データを取得できませんでした")
print(f"=== {SYMBOL} のロングショート比率（{len(lsr)}件） ===")
print(f" 期間 : {lsr.index[0]} 〜 {lsr.index[-1]}")
print(f" 直近 : ロング {lsr['buy_ratio'].iloc[-1]:.1%} / "
f"ショート {lsr['sell_ratio'].iloc[-1]:.1%} "
f"(倍率 {lsr['ls_ratio'].iloc[-1]:.2f})")
print(f" 期間平均 : ロング {lsr['buy_ratio'].mean():.1%}")
print(f" 偏りの幅 : {lsr['skew'].min():+.3f} 〜 {lsr['skew'].max():+.3f}")
ext = detect_extremes(lsr)
print(f"\n 極端なロング偏り : {ext['extreme_long'].sum()}回")
print(f" 極端なショート偏り: {ext['extreme_short'].sum()}回")
bars = fetch_klines(c.http, SYMBOL, interval="60",
start=str(lsr.index[0].date()), verbose=False)
print(f"\n=== 逆張りの有効性（24本先のリターン） ===")
for k, v in contrarian_check(lsr, bars).items():
print(f" {k:<24}: {v}")
print("\n※ 極値後のリターンが全体平均と大差なければ、")
print(" この指標に逆張りの予測力はない。")
print(" 『多くの人が信じている指標』と『機能する指標』は別。")buyRatio / sellRatio はポジションを持っているアカウント数の比率です。「ロング70%」は人数が70%という意味で、建玉の金額が70%とは限りません。
小口の個人が多数ロング、大口が少数でショート、という状況では比率が偏っていても金額では均衡していることがあります。金額ベースの偏りを見たいならOI（サンプル3）と組み合わせてください。
資金調達率はプレミアム指数（先物と現物指数の乖離）を平均化して算出されます。収受時刻までの推移を見れば、次回のレートの方向が読めます。
"""premium_forecast.py — プレミアム指数から次回資金調達率の傾向を読む
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
注意: 正確な算出式（クランプ・平均化の期間・金利成分）は
取引所の仕様に依存する。ここでは「方向と大きさの目安」を
得ることを目的とし、完全な予測は行わない。
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timezone
import numpy as np
import pandas as pd
from basis_history import _to_df
@dataclass
class FundingOutlook:
symbol: str
current_rate: float # get_tickers が返す次回予定レート
next_funding_ms: int
premium_now: float
premium_mean: float # 収受期間中の平均プレミアム
premium_trend: float # プレミアムの傾き
minutes_to_next: float
@property
def direction(self) -> str:
if self.premium_trend > 1e-5:
return "上昇（レートは高くなる方向）"
if self.premium_trend < -1e-5:
return "低下（レートは低くなる方向）"
return "横ばい"
def report(self) -> str:
return "\n".join([
f"{self.symbol}",
f" 次回予定レート : {self.current_rate * 100:+.5f}%",
f" 次回まで : {self.minutes_to_next:.0f}分",
f" 現在のプレミアム: {self.premium_now * 100:+.5f}%",
f" 期間平均 : {self.premium_mean * 100:+.5f}%",
f" 推移 : {self.direction}",
])
def fetch_premium_series(http, symbol: str, interval: str = "5",
limit: int = 200,
category: str = "linear") -> pd.DataFrame:
"""プレミアム指数のkline（5フィールド）を取得する。"""
res = http.get_premium_index_price_kline(
category=category, symbol=symbol, interval=interval, limit=limit)
if res["retCode"] != 0:
raise RuntimeError(f"premium kline失敗: {res['retMsg']}")
return _to_df(res["result"]["list"])
def outlook(http, symbol: str, interval_hours: float = 8.0,
category: str = "linear") -> FundingOutlook:
ticker = http.get_tickers(category=category,
symbol=symbol)["result"]["list"][0]
rate = float(ticker.get("fundingRate") or 0)
next_ms = int(ticker.get("nextFundingTime") or 0)
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
prem = fetch_premium_series(http, symbol, interval="5",
category=category)
if prem.empty:
raise RuntimeError("プレミアム指数を取得できません")
# 直近の収受期間ぶんだけを見る
bars_in_period = int(interval_hours * 60 / 5)
window = prem["close"].tail(max(bars_in_period, 12))
# 傾き（線形回帰の係数）
x = np.arange(len(window), dtype=float)
trend = float(np.polyfit(x, window.to_numpy(), 1)[0]) if len(window) > 2 else 0.0
return FundingOutlook(
symbol=symbol, current_rate=rate, next_funding_ms=next_ms,
premium_now=float(prem["close"].iloc[-1]),
premium_mean=float(window.mean()),
premium_trend=trend,
minutes_to_next=(next_ms - now_ms) / 60000 if next_ms else float("nan"),
)
def compare_premium_and_funding(http, symbol: str,
category: str = "linear") -> pd.DataFrame:
"""過去のプレミアム平均と、実際に確定した資金調達率を比較する。
「プレミアムから資金調達率をどれだけ説明できるか」を測る。
"""
from funding_fetch import fetch_funding_history
funding = fetch_funding_history(http, symbol, start="2026-06-01",
category=category, verbose=False)
if funding.empty:
return pd.DataFrame()
prem = fetch_premium_series(http, symbol, interval="60", limit=1000)
if prem.empty:
return pd.DataFrame()
rows = []
for ts, rate in funding["funding_rate"].items():
# 収受時刻の直前8時間のプレミアム平均
window = prem["close"].loc[ts - pd.Timedelta(hours=8):ts]
if window.empty:
continue
rows.append({"ts": ts, "funding_rate": float(rate),
"premium_mean": float(window.mean()),
"premium_last": float(window.iloc[-1])})
df = pd.DataFrame(rows).set_index("ts")
if len(df) > 3:
df.attrs["corr"] = float(df["funding_rate"].corr(df["premium_mean"]))
return df
if __name__ == "__main__":
from client import make_client
c = make_client()
for symbol in ("BTCUSDT", "ETHUSDT"):
try:
print(outlook(c.http, symbol).report())
print()
except Exception as e:
print(f"{symbol}: {e}\n")
print("=== プレミアム指数と確定レートの関係 ===")
df = compare_premium_and_funding(c.http, "BTCUSDT")
if not df.empty:
corr = df.attrs.get("corr")
print(f" サンプル {len(df)}件 / 相関 {corr:.3f}")
print(df.tail(6).to_string(float_format=lambda v: f"{v:+.6f}"))
print("\n 相関が高ければ、プレミアムの推移から次回レートの")
print(" 方向を読める（記事35のエントリー判断に使える）。")
else:
print(" データを取得できませんでした")"""market_snapshot.py — 需給データを1つのDataFrameに統合する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
価格・OI・比率・ベーシス・資金調達率を横断的に並べ、
銘柄選定やレジーム判定の材料にする。
"""
from __future__ import annotations
from dataclasses import dataclass, field
import numpy as np
import pandas as pd
@dataclass
class SymbolSnapshot:
symbol: str
last: float
change_24h: float
turnover_24h: float
open_interest: float
oi_value: float
funding_rate: float
funding_apr: float
basis_bps: float
spread_bps: float
long_ratio: float = float("nan")
oi_change_24h: float = float("nan")
@property
def oi_to_turnover(self) -> float:
"""OI ÷ 24h売買代金。高いほど『持ち合い』、低いほど回転が速い。"""
if self.turnover_24h <= 0:
return float("nan")
return self.oi_value / self.turnover_24h
def line(self) -> str:
return (f"{self.symbol:<12}{self.change_24h:>+8.2%}"
f"{self.turnover_24h / 1e9:>9.2f}B"
f"{self.oi_value / 1e6:>10.0f}M"
f"{self.funding_apr:>+9.1f}%{self.basis_bps:>+9.1f}b"
f"{self.spread_bps:>8.2f}b{self.oi_to_turnover:>8.2f}")
def build_snapshot(http, category: str = "linear",
min_turnover: float = 20_000_000,
with_long_short: bool = False,
top: int = 20) -> list[SymbolSnapshot]:
"""全銘柄の需給スナップショットを作る（基本は2リクエスト）。"""
from annualize import fetch_funding_specs
specs = fetch_funding_specs(http, category)
res = http.get_tickers(category=category)
if res["retCode"] != 0:
raise RuntimeError(f"get_tickers失敗: {res['retMsg']}")
out: list[SymbolSnapshot] = []
for t in res["result"]["list"]:
turnover = float(t.get("turnover24h") or 0)
if turnover < min_turnover:
continue
symbol = t["symbol"]
spec = specs.get(symbol)
rate = float(t.get("fundingRate") or 0)
last = float(t.get("lastPrice") or 0)
index = float(t.get("indexPrice") or 0)
bid = float(t.get("bid1Price") or 0)
ask = float(t.get("ask1Price") or 0)
basis = ((last - index) / index * 10_000) if index > 0 else float("nan")
spread = ((ask - bid) / last * 10_000) if last and bid and ask else float("nan")
apr = float(spec.annualize(rate)) * 100 if spec else float("nan")
out.append(SymbolSnapshot(
symbol=symbol, last=last,
change_24h=float(t.get("price24hPcnt") or 0),
turnover_24h=turnover,
open_interest=float(t.get("openInterest") or 0),
oi_value=float(t.get("openInterestValue") or 0),
funding_rate=rate, funding_apr=apr,
basis_bps=basis, spread_bps=spread,
))
out.sort(key=lambda s: -s.turnover_24h)
out = out[:top]
# ロングショート比率は1銘柄ずつなのでオプション扱い
if with_long_short:
from long_short_ratio import fetch_long_short_ratio
for s in out:
try:
lsr = fetch_long_short_ratio(http, s.symbol, "1h",
category, max_pages=1)
if not lsr.empty:
s.long_ratio = float(lsr["buy_ratio"].iloc[-1])
except Exception:
pass
return out
def regime_summary(snaps: list[SymbolSnapshot]) -> str:
"""市場全体の需給レジームを要約する。"""
if not snaps:
return "データがありません"
aprs = np.array([s.funding_apr for s in snaps
if np.isfinite(s.funding_apr)])
basis = np.array([s.basis_bps for s in snaps
if np.isfinite(s.basis_bps)])
changes = np.array([s.change_24h for s in snaps])
positive_funding = (aprs > 0).mean() if aprs.size else float("nan")
median_apr = np.median(aprs) if aprs.size else float("nan")
if positive_funding > 0.8 and median_apr > 15:
regime = "強い買い過熱（キャリーの好機だが急変リスクも高い）"
elif positive_funding > 0.6:
regime = "やや買い優勢（平常）"
elif positive_funding < 0.3:
regime = "売り優勢（ロング側が資金調達を受け取れる）"
else:
regime = "中立"
return "\n".join([
f"対象 {len(snaps)}銘柄",
f" 資金調達がプラスの割合 : {positive_funding:.0%}",
f" 年率の中央値 : {median_apr:+.1f}%",
f" ベーシス中央値 : {np.median(basis):+.1f} bps"
if basis.size else "",
f" 24h変化率の中央値 : {np.median(changes):+.2%}",
f" レジーム判定 : {regime}",
])
if __name__ == "__main__":
from client import make_client
c = make_client()
snaps = build_snapshot(c.http, top=20)
print(f"{'symbol':<12}{'24h変化':>8}{'売買代金':>10}"
f"{'OI額':>11}{'資金年率':>9}{'ベーシス':>10}"
f"{'スプレッド':>9}{'OI/代金':>8}")
print("-" * 78)
for s in snaps:
print(s.line())
print()
print(regime_summary(snaps))
print("\n※ OI/売買代金 が高い = 建玉が滞留している（動きが出ると大きい）")
print(" 低い = 回転が速い（デイトレード主体）")| 注意点 | 内容 |
|---|---|
| この取引所のデータのみ | BybitのOI・比率はBybitの中だけの数値。市場全体の需給ではない |
| 比率は人数ベース | buyRatio は建玉量ではなくアカウント数の比率 |
| OIの単位 | linearは基軸通貨、inverseはUSD建て。openInterestValue（金額）で比較する |
| 4象限は解釈であって予測ではない | 「納得できる」ことと「機能する」ことは別（記事32で検証） |
| リクエスト数 | OIも比率も銘柄ごと。全銘柄を取ると数百リクエスト（記事05） |
| ベーシスと資金調達は連動 | 両方をシグナルに使うと実質的に同じ情報を二重に使うことになる |
"""test_snapshot.py — 需給データの計算が正しいことを検証する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
pytest test_snapshot.py -v
"""
from __future__ import annotations
from decimal import Decimal
import pandas as pd
import pytest
from oi_quadrant import Quadrant, classify
from price_kinds import PriceSnapshot
def make_price(last: str, mark: str, index: str) -> PriceSnapshot:
return PriceSnapshot(symbol="TESTUSDT", last=Decimal(last),
mark=Decimal(mark), index=Decimal(index),
bid=Decimal(last), ask=Decimal(last), ts_ms=0)
def test_basis_sign():
"""先物が現物より高ければベーシスはプラス。"""
p = make_price("101", "100.5", "100")
assert p.basis_bps == Decimal("100") # +1% = 100bps
p = make_price("99", "99.5", "100")
assert p.basis_bps == Decimal("-100")
def test_zero_index_does_not_crash():
"""インデックス価格が0でもゼロ割りしないこと。"""
p = make_price("100", "100", "0")
assert p.basis_bps == Decimal(0)
assert p.mark_index_bps == Decimal(0)
def test_mark_gap():
p = make_price("100.5", "100", "100")
assert p.mark_gap_bps == Decimal("50") # +0.5% = 50bps
@pytest.mark.parametrize("price,oi,expected", [
(0.01, 0.01, Quadrant.NEW_LONG),
(0.01, -0.01, Quadrant.SHORT_COVER),
(-0.01, 0.01, Quadrant.NEW_SHORT),
(-0.01, -0.01, Quadrant.LONG_LIQUIDATION),
])
def test_quadrant_classification(price, oi, expected):
assert classify(price, oi) is expected
def test_quadrant_threshold_makes_neutral():
"""閾値を超えない微小な変化は中立になること。"""
assert classify(0.0001, 0.01, price_threshold=0.001) is Quadrant.NEUTRAL
assert classify(0.01, 0.0001, oi_threshold=0.001) is Quadrant.NEUTRAL
def test_long_short_ratio_sums_to_one():
"""buyRatio + sellRatio は1になる（アカウント数の比率）。"""
df = pd.DataFrame({"buy_ratio": [0.6, 0.55], "sell_ratio": [0.4, 0.45]})
total = df["buy_ratio"] + df["sell_ratio"]
assert (total - 1.0).abs().max() < 1e-9
def test_oi_value_scales_with_price():
"""OI額 = OI × 価格 という関係。"""
oi = 1000.0
price = 120_000.0
assert oi * price == 120_000_000.0quadrant_summary() で、実際に次足リターンに差が出るかを確認してください。contrarian_check() は、まさにその仮説を検証する関数です。極端な偏りの後のリターンが全体平均と大差なければ、この指標に予測力はありません。mark_index_bps で両者の差を監視できます。get_open_interest は銘柄ごとなので、500銘柄なら500リクエストです。記事05の流量ゲートを通してください。get_tickers の openInterest / openInterestValue で1リクエストで全銘柄取れます（サンプル7）。履歴が必要な場合だけ個別に取得する、という使い分けが効率的です。投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません（間接的・付随的な障害・損失も含め一切の責任を負いません）。自己責任でご利用ください。
当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。
当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。
当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。
当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。
当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。
当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。