※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。
無期限先物は満期がないため、価格を現物に近づける仕組みが必要です。それが資金調達（Funding）で、一定時間ごとにロングとショートの間で資金をやり取りします。
| 資金調達率 | 市場の状態 | ロング | ショート |
|---|---|---|---|
| プラス（+） | 先物が現物より高い＝買いが過熱 | 支払い | 受取り |
| マイナス（−） | 先物が現物より安い＝売りが過熱 | 受取り | 支払い |
金額はポジション価値 × 資金調達率です。レバレッジには依存せず、建てている想定元本に対してかかります。
歴史的に、仮想通貨の無期限先物の資金調達率はプラスに偏る傾向があります（レバレッジをかけて買いたい需要が構造的に強いため）。つまりロングを持ち続ける戦略は、常に少しずつコストを払っています。
これを計上しないバックテストはロング戦略を体系的に過大評価し、ショート戦略を過小評価します。年率で数%〜十数%の差が生まれるため、無期限先物で数時間以上ポジションを持つ戦略では必ず織り込んでください（記事31）。
逆にこの偏りは収益源にもなり、「現物ロング＋perpショート」で資金調達を受け取り続けるキャリー戦略が成立します（記事35）。
| パラメータ | 必須 | 内容 |
|---|---|---|
| category | 必須 | linear / inverse（現物・オプションには存在しない） |
| symbol | 必須 | 全銘柄一括取得はできない。1銘柄ずつ呼ぶ |
| startTime | 任意 | 単独指定はエラーになる |
| endTime | 任意 | この時刻までの直近200件が返る |
| limit | 任意 | 1〜200（既定200） |
公式ドキュメントに「startTime のみを渡すとエラーが返る」と明記されています。「2024年以降を取りたい」と考えて startTime を指定すると失敗します。
正しい取得方法はendTime を使って新しい方から遡ることです（記事07のklineと同じパターン）。startTime は「ここまで遡ったら終了」というループの終了条件として使い、リクエストには渡しません。
レスポンスは symbol / fundingRate / fundingRateTimestamp の3フィールドのみで、新しい順（降順）に返ります。
"""funding_fetch.py — 資金調達率の履歴を全期間取得する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
重要: startTime単独指定はエラーになるため、endTimeで遡る方式を使う。
"""
from __future__ import annotations
import time
from datetime import datetime, timezone
from pathlib import Path
import pandas as pd
from pybit.unified_trading import HTTP
MAX_LIMIT = 200
CACHE_DIR = Path("data/funding")
def to_ms(dt: str | datetime | None) -> int | None:
if dt is None:
return None
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_funding_history(http: HTTP, symbol: str,
start: str | datetime | None = None,
end: str | datetime | None = None,
category: str = "linear",
sleep: float = 0.1,
verbose: bool = True) -> pd.DataFrame:
"""全期間の資金調達率を取得する。
startを省略すると取引所が持つ最古まで遡る。
"""
start_ms = to_ms(start) or 0
cursor_end = to_ms(end) or int(time.time() * 1000)
rows: list[dict] = []
pages = 0
prev_oldest: int | None = None
while True:
# startTimeは渡さない（単独指定はエラー、併用しても遡れない）
res = http.get_funding_rate_history(
category=category, symbol=symbol,
endTime=cursor_end, limit=MAX_LIMIT,
)
if res["retCode"] != 0:
raise RuntimeError(
f"get_funding_rate_history失敗({symbol}): "
f"{res['retCode']} {res['retMsg']}"
)
page = res["result"]["list"]
if not page:
break
# 降順で返るので最後の要素が最古
oldest = int(page[-1]["fundingRateTimestamp"])
# カーソルが進まなくなったら打ち切る（無限ループ防止）
if prev_oldest is not None and oldest >= prev_oldest:
break
prev_oldest = oldest
rows.extend(page)
pages += 1
if verbose and pages % 20 == 0:
print(f" {pages}ページ / {len(rows)}件 最古 "
f"{datetime.fromtimestamp(oldest / 1000, timezone.utc):%Y-%m-%d}")
if oldest <= start_ms or len(page) < MAX_LIMIT:
break
cursor_end = oldest - 1
time.sleep(sleep)
if not rows:
return pd.DataFrame(columns=["funding_rate"],
index=pd.DatetimeIndex([], tz="UTC", name="ts"))
df = pd.DataFrame(rows)
df["ts"] = pd.to_datetime(df["fundingRateTimestamp"].astype("int64"),
unit="ms", utc=True)
df["funding_rate"] = df["fundingRate"].astype(float)
df = (df[["ts", "funding_rate"]]
.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}: {len(df)}件 "
f"{df.index[0]:%Y-%m-%d} 〜 {df.index[-1]:%Y-%m-%d}")
return df
def load_or_fetch(http: HTTP, symbol: str, start: str = "2023-01-01",
category: str = "linear",
verbose: bool = False) -> pd.DataFrame:
"""Parquetキャッシュ付きの取得（差分更新）。"""
CACHE_DIR.mkdir(parents=True, exist_ok=True)
path = CACHE_DIR / f"{category}_{symbol}.parquet"
existing = pd.read_parquet(path) if path.exists() else pd.DataFrame()
fetch_from = (existing.index[-1].isoformat()
if not existing.empty else start)
fresh = fetch_funding_history(http, symbol, start=fetch_from,
category=category, verbose=verbose)
merged = pd.concat([existing, fresh]) if not existing.empty else fresh
merged = merged[~merged.index.duplicated(keep="last")].sort_index()
tmp = path.with_suffix(".tmp")
merged.to_parquet(tmp, compression="zstd")
tmp.replace(path)
return merged
if __name__ == "__main__":
http = HTTP(testnet=False)
df = fetch_funding_history(http, "BTCUSDT", start="2024-01-01")
print(df.head())
print(f"\n平均 {df['funding_rate'].mean():+.6%} "
f"標準偏差 {df['funding_rate'].std():.6%}")
print(f"プラスの割合 {(df['funding_rate'] > 0).mean():.1%}")
# startTime単独指定がエラーになることの確認
print("\n--- startTime単独指定の挙動 ---")
res = http.get_funding_rate_history(
category="linear", symbol="BTCUSDT",
startTime=int(pd.Timestamp("2025-01-01").timestamp() * 1000),
)
print(f" retCode={res['retCode']} retMsg={res['retMsg']}")
print(" -> endTimeで遡る方式を使う必要がある")「資金調達は8時間ごと」は思い込みです。Bybitでは銘柄によって8時間・4時間・1時間などが混在します。instruments-info の fundingInterval（分単位）から取得してください。
"""annualize.py — 収受間隔を踏まえた正しい年率換算
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
よくある誤り: すべての銘柄を「8時間ごと（1日3回）」と決め打つこと。
実際は4時間・1時間の銘柄もあり、その場合2〜8倍ずれる。
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
import pandas as pd
from pybit.unified_trading import HTTP
HOURS_PER_YEAR = Decimal("8760") # 24 * 365
@dataclass(frozen=True)
class FundingSpec:
symbol: str
interval_minutes: int
@property
def interval_hours(self) -> Decimal:
return Decimal(self.interval_minutes) / 60
@property
def times_per_day(self) -> Decimal:
return Decimal(24) / self.interval_hours
@property
def times_per_year(self) -> Decimal:
return HOURS_PER_YEAR / self.interval_hours
def annualize(self, rate: float | Decimal) -> Decimal:
"""1回分のレートを年率に換算する（単利）。"""
return Decimal(str(rate)) * self.times_per_year
def annualize_compound(self, rate: float | Decimal) -> Decimal:
"""複利での年率換算。レートが大きい場合は差が出る。"""
r = Decimal(str(rate))
n = int(self.times_per_year)
return (Decimal(1) + r) ** n - Decimal(1)
def fetch_funding_specs(http: HTTP, category: str = "linear"
) -> dict[str, FundingSpec]:
"""全銘柄の収受間隔を取得する（1〜2リクエストで済む）。"""
specs: dict[str, FundingSpec] = {}
cursor = None
while True:
kw: dict = {"category": category, "limit": 1000}
if cursor:
kw["cursor"] = cursor
res = http.get_instruments_info(**kw)
if res["retCode"] != 0:
raise RuntimeError(f"instruments-info失敗: {res['retMsg']}")
for info in res["result"]["list"]:
minutes = int(info.get("fundingInterval") or 0)
if minutes > 0:
specs[info["symbol"]] = FundingSpec(info["symbol"], minutes)
cursor = res["result"].get("nextPageCursor")
if not cursor:
break
return specs
def interval_distribution(specs: dict[str, FundingSpec]) -> pd.Series:
"""収受間隔の分布。8時間以外がどれだけあるかを確認する。"""
s = pd.Series([sp.interval_hours for sp in specs.values()],
dtype="float64")
return s.value_counts().sort_index()
if __name__ == "__main__":
http = HTTP(testnet=False)
specs = fetch_funding_specs(http)
print(f"資金調達がある銘柄: {len(specs)}件\n")
print("収受間隔の分布:")
for hours, count in interval_distribution(specs).items():
print(f" {hours:>5.1f}時間ごと（1日{24 / hours:>4.0f}回）: "
f"{count:>4}銘柄")
print("\n--- 同じレート0.01%でも、間隔が違えば年率が変わる ---")
rate = Decimal("0.0001") # 0.01%
print(f"{'間隔':>8}{'年間回数':>10}{'単利年率':>12}{'複利年率':>12}")
print("-" * 44)
for minutes in (60, 240, 480):
sp = FundingSpec("SAMPLE", minutes)
print(f"{sp.interval_hours:>6.0f}h{sp.times_per_year:>10.0f}"
f"{sp.annualize(rate) * 100:>11.2f}%"
f"{sp.annualize_compound(rate) * 100:>11.2f}%")
print("\n※ 8時間前提で計算すると、1時間間隔の銘柄では年率を")
print(" 8分の1に見誤る。必ずfundingIntervalを取得すること。")
for sym in ("BTCUSDT", "ETHUSDT"):
if sym in specs:
sp = specs[sym]
print(f"\n{sym}: {sp.interval_hours:.0f}時間ごと "
f"（年間{sp.times_per_year:.0f}回）")get_tickers は全銘柄の現在の資金調達率と次回収受時刻を1リクエストで返します。履歴が要らない用途ならこれが最も効率的です。
"""funding_scan.py — 全銘柄の資金調達率をスキャンしてランキングする
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
get_tickers（1リクエスト）+ instruments-info（1〜2リクエスト）だけで
全銘柄の年率換算ランキングが作れる。履歴取得は不要。
"""
from __future__ import annotations
from datetime import datetime, timezone
import pandas as pd
from pybit.unified_trading import HTTP
from annualize import fetch_funding_specs
def scan_funding(http: HTTP, category: str = "linear",
min_turnover: float = 10_000_000) -> pd.DataFrame:
"""全銘柄の現在の資金調達率を年率換算して返す。"""
specs = fetch_funding_specs(http, category)
res = http.get_tickers(category=category)
if res["retCode"] != 0:
raise RuntimeError(f"get_tickers失敗: {res['retMsg']}")
rows = []
now_ms = int(datetime.now(timezone.utc).timestamp() * 1000)
for t in res["result"]["list"]:
sym = t["symbol"]
spec = specs.get(sym)
if spec is None:
continue
rate = float(t.get("fundingRate") or 0)
turnover = float(t.get("turnover24h") or 0)
if turnover < min_turnover:
continue
next_ms = int(t.get("nextFundingTime") or 0)
mins_to_next = (next_ms - now_ms) / 60000 if next_ms else float("nan")
bid = float(t.get("bid1Price") or 0)
ask = float(t.get("ask1Price") or 0)
last = float(t.get("lastPrice") or 0)
spread_bps = (ask - bid) / last * 10_000 if last and bid and ask else float("nan")
rows.append({
"symbol": sym,
"funding_rate": rate,
"interval_h": float(spec.interval_hours),
"apr_pct": float(spec.annualize(rate)) * 100,
"next_in_min": mins_to_next,
"turnover24h": turnover,
"spread_bps": spread_bps,
"oi_value": float(t.get("openInterestValue") or 0),
})
return pd.DataFrame(rows)
def report(df: pd.DataFrame, top: int = 12) -> None:
cols = ["symbol", "funding_rate", "interval_h", "apr_pct",
"next_in_min", "spread_bps", "turnover24h"]
with pd.option_context("display.float_format", lambda v: f"{v:,.4f}"):
print("=== 資金調達率 年率上位（ショート側が受け取る） ===")
print(df.nlargest(top, "apr_pct")[cols].to_string(index=False))
print("\n=== 資金調達率 年率下位（ロング側が受け取る） ===")
print(df.nsmallest(top, "apr_pct")[cols].to_string(index=False))
print(f"\n--- 全体の統計（{len(df)}銘柄） ---")
print(f" 年率の中央値 : {df['apr_pct'].median():+.2f}%")
print(f" 年率の平均 : {df['apr_pct'].mean():+.2f}%")
print(f" プラスの割合 : {(df['funding_rate'] > 0).mean():.1%}")
print(f" |年率|>20% の数: {(df['apr_pct'].abs() > 20).sum()}銘柄")
# アービトラージ候補（記事35）: 年率が高くスプレッドが狭い銘柄
cand = df[(df["apr_pct"] > 15) & (df["spread_bps"] < 8)]
if not cand.empty:
print(f"\n--- キャリー候補（年率>15% かつ スプレッド<8bps） ---")
print(cand.nlargest(8, "apr_pct")[
["symbol", "apr_pct", "spread_bps", "turnover24h"]
].to_string(index=False))
print(" ※ 高い資金調達率は、それ自体がリスクの高さを示している。")
print(" 実行可否は記事35の損益分岐計算で判断すること。")
if __name__ == "__main__":
http = HTTP(testnet=False)
report(scan_funding(http))年率100%を超える銘柄は珍しくありませんが、それはその銘柄が異常に過熱していることのシグナルです。価格が急変してデルタニュートラルを維持できなくなる、流動性が枯れて解消できない、資金調達率が突然反転する——といったリスクと引き換えです。
スクリーニングの段階でスプレッドと出来高でフィルタする（上のコードで spread_bps < 8 を条件にしている）のが最低限の防御です。実行判断は記事35の損益分岐計算まで進んでから行ってください。
「今プラスなら次回もプラスか？」——これが分かれば、キャリー戦略のエントリー判断に使えます。自己相関と符号の的中率で測ります。
"""funding_persistence.py — 資金調達率の持続性を統計的に検定する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
目的: 「今の資金調達率から次回を予測できるか」を数値で確認する。
予測できるなら、キャリー戦略のエントリー判断に使える。
"""
from __future__ import annotations
from dataclasses import dataclass
import numpy as np
import pandas as pd
@dataclass
class PersistenceStats:
symbol: str
n: int
mean: float
std: float
positive_ratio: float
acf1: float # 1期ラグの自己相関
acf3: float
sign_hit_rate: float # 符号が次回も同じである割合
half_life: float | None # 平均回帰の半減期（回数）
max_run_positive: int # プラスが連続した最長回数
def line(self) -> str:
hl = f"{self.half_life:.1f}回" if self.half_life else " - "
return (f"{self.symbol:<12}{self.n:>7}{self.mean * 100:>+9.4f}%"
f"{self.positive_ratio:>8.1%}{self.acf1:>8.3f}"
f"{self.sign_hit_rate:>9.1%}{hl:>8}"
f"{self.max_run_positive:>7}")
def longest_run(series: pd.Series, positive: bool = True) -> int:
"""同じ符号が連続した最長回数。"""
mask = (series > 0) if positive else (series < 0)
best = run = 0
for v in mask:
run = run + 1 if v else 0
best = max(best, run)
return best
def ou_half_life(series: pd.Series) -> float | None:
"""AR(1)係数から平均回帰の半減期を求める。
x_t = a + b * x_{t-1} + e のとき、半減期 = -ln(2) / ln(b)
b が1に近いほど持続的、0に近いほどランダム。
"""
x = series.dropna()
if len(x) < 30:
return None
lag = x.shift(1).dropna()
y = x.loc[lag.index]
b = np.polyfit(lag.values, y.values, 1)[0]
if not (0 < b < 1):
return None
return float(-np.log(2) / np.log(b))
def analyse_persistence(df: pd.DataFrame, symbol: str,
col: str = "funding_rate") -> PersistenceStats:
s = df[col].dropna()
if len(s) < 10:
raise ValueError(f"{symbol}: データが不足しています ({len(s)}件)")
sign = np.sign(s)
# 符号が次回も同じである割合（0を除く）
pairs = pd.DataFrame({"now": sign, "next": sign.shift(-1)}).dropna()
pairs = pairs[(pairs["now"] != 0) & (pairs["next"] != 0)]
hit = (pairs["now"] == pairs["next"]).mean() if len(pairs) else float("nan")
return PersistenceStats(
symbol=symbol, n=len(s),
mean=float(s.mean()), std=float(s.std()),
positive_ratio=float((s > 0).mean()),
acf1=float(s.autocorr(lag=1)),
acf3=float(s.autocorr(lag=3)),
sign_hit_rate=float(hit),
half_life=ou_half_life(s),
max_run_positive=longest_run(s, positive=True),
)
def sign_transition_matrix(df: pd.DataFrame,
col: str = "funding_rate") -> pd.DataFrame:
"""符号の遷移確率。プラスの後にプラスが来る確率などが分かる。"""
sign = np.sign(df[col].dropna())
cur = sign.iloc[:-1].reset_index(drop=True)
nxt = sign.iloc[1:].reset_index(drop=True)
tab = pd.crosstab(cur, nxt, normalize="index")
tab.index = [{-1.0: "現在マイナス", 0.0: "現在ゼロ",
1.0: "現在プラス"}.get(i, str(i)) for i in tab.index]
tab.columns = [{-1.0: "次マイナス", 0.0: "次ゼロ",
1.0: "次プラス"}.get(c, str(c)) for c in tab.columns]
return tab
if __name__ == "__main__":
from pybit.unified_trading import HTTP
from funding_fetch import fetch_funding_history
http = HTTP(testnet=False)
symbols = ("BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT")
print(f"{'symbol':<12}{'件数':>7}{'平均':>10}{'正の割合':>8}"
f"{'ACF(1)':>8}{'符号的中':>9}{'半減期':>8}{'最長連続':>7}")
print("-" * 70)
results = []
for sym in symbols:
try:
df = fetch_funding_history(http, sym, start="2024-01-01",
verbose=False)
st = analyse_persistence(df, sym)
results.append((sym, df, st))
print(st.line())
except Exception as e:
print(f"{sym:<12} 取得失敗: {e}")
if results:
sym, df, st = results[0]
print(f"\n=== {sym} の符号遷移確率 ===")
print(sign_transition_matrix(df).to_string(
float_format=lambda v: f"{v:.1%}"))
print("\n--- 解釈 ---")
print(" ACF(1)が高く符号的中率が60%を超えるなら、資金調達率は持続的。")
print(" つまり『今プラスなら次もプラス』が期待でき、")
print(" キャリー戦略（記事35）のエントリー判断に使える。")
print(" ただし持続性が高いことは『反転しない』を意味しない。")
print(" 最長連続回数の後には必ず反転が来ている点に注意。")「現物ロング＋perpショート」で資金調達を受け取り続けた場合の収益を、コストを差し引いた上で試算します。
"""carry_sim.py — デルタニュートラル・キャリーの収益試算
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
構成: 現物ロング + 無期限ショート（価格変動の影響を打ち消す）
収益: 資金調達の受取り − 往復手数料 − スプレッド
"""
from __future__ import annotations
from dataclasses import dataclass
from decimal import Decimal
import pandas as pd
@dataclass
class CarryCost:
"""往復のコスト（bps単位）。実測値を入れること。"""
spot_taker_bps: Decimal = Decimal("10.0") # 現物テイカー 0.1%
perp_taker_bps: Decimal = Decimal("5.5") # perpテイカー 0.055%
spot_maker_bps: Decimal = Decimal("10.0")
perp_maker_bps: Decimal = Decimal("2.0")
spread_bps: Decimal = Decimal("2.0") # 片道のスプレッド損失
def entry_exit_bps(self, use_maker: bool = False) -> Decimal:
"""建てて解消するまでの総コスト（bps）。
現物とperpの両脚で、それぞれ建て・解消の2回ずつ発生する。
"""
if use_maker:
legs = self.spot_maker_bps + self.perp_maker_bps
spread = Decimal(0) # メイカーならスプレッドは払わない
else:
legs = self.spot_taker_bps + self.perp_taker_bps
spread = self.spread_bps * 2
return legs * 2 + spread
@dataclass
class CarryResult:
symbol: str
days: float
funding_collected_bps: Decimal
cost_bps: Decimal
net_bps: Decimal
apr_pct: Decimal
breakeven_days: Decimal | None
events: int
def report(self) -> str:
be = (f"{self.breakeven_days:.1f}日" if self.breakeven_days
else "回収不能")
sign = "利益" if self.net_bps > 0 else "損失"
return (
f"{self.symbol} 保有 {self.days:.1f}日（{self.events}回収受）\n"
f" 資金調達の受取り : {self.funding_collected_bps:>+9.2f} bps\n"
f" 往復コスト : {-self.cost_bps:>+9.2f} bps\n"
f" 正味 : {self.net_bps:>+9.2f} bps ({sign})\n"
f" 年率換算 : {self.apr_pct:>+9.2f} %\n"
f" 損益分岐 : {be}（この期間より短いと赤字）"
)
def simulate_carry(funding: pd.DataFrame, symbol: str,
cost: CarryCost | None = None,
use_maker: bool = False,
start: str | None = None,
end: str | None = None) -> CarryResult:
"""実際の資金調達履歴から、キャリーの正味収益を計算する。
ショート側なので、資金調達率がプラスのとき受け取る。
"""
cost = cost or CarryCost()
s = funding["funding_rate"]
if start:
s = s[s.index >= start]
if end:
s = s[s.index <= end]
if s.empty:
raise ValueError("対象期間にデータがありません")
days = (s.index[-1] - s.index[0]).total_seconds() / 86400
if days <= 0:
days = 1 / 24
# perpショートは、レートがプラスのとき受け取る（符号そのまま）
collected = Decimal(str(s.sum())) * 10_000 # bps
total_cost = cost.entry_exit_bps(use_maker)
net = collected - total_cost
apr = net / Decimal(str(days)) * Decimal("365") / 100
# 1日あたりの平均受取りで、コストを何日で回収できるか
per_day = collected / Decimal(str(days)) if days else Decimal(0)
breakeven = (total_cost / per_day) if per_day > 0 else None
return CarryResult(symbol=symbol, days=days,
funding_collected_bps=collected,
cost_bps=total_cost, net_bps=net, apr_pct=apr,
breakeven_days=breakeven, events=len(s))
def required_rate_for_breakeven(cost: CarryCost, hold_days: float,
interval_hours: float,
use_maker: bool = False) -> Decimal:
"""指定期間の保有でコストを回収するために必要な1回あたりのレート。"""
total_cost_bps = cost.entry_exit_bps(use_maker)
events = Decimal(str(hold_days)) * 24 / Decimal(str(interval_hours))
if events <= 0:
return Decimal(0)
return total_cost_bps / events / 10_000 # 比率に戻す
if __name__ == "__main__":
from pybit.unified_trading import HTTP
from annualize import fetch_funding_specs
from funding_fetch import fetch_funding_history
http = HTTP(testnet=False)
specs = fetch_funding_specs(http)
for sym in ("BTCUSDT", "ETHUSDT"):
try:
df = fetch_funding_history(http, sym, start="2025-01-01",
verbose=False)
except Exception as e:
print(f"{sym}: 取得失敗 {e}")
continue
print("=" * 58)
for label, maker in (("テイカー執行", False), ("メイカー執行", True)):
r = simulate_carry(df, sym, use_maker=maker)
print(f"[{label}]")
print(r.report())
print()
print("=" * 58)
print("損益分岐に必要な1回あたりのレート（保有期間別・8時間間隔）")
cost = CarryCost()
print(f"{'保有':>8}{'テイカー':>14}{'メイカー':>14}")
print("-" * 38)
for d in (0.5, 1, 3, 7, 30):
t = required_rate_for_breakeven(cost, d, 8.0, False)
m = required_rate_for_breakeven(cost, d, 8.0, True)
print(f"{d:>6}日{t * 100:>13.4f}%{m * 100:>13.4f}%")
print("\n※ 短期保有ではコストの回収が難しい。")
print(" 1日以内で解消する前提のキャリーは、ほぼ成立しない。")デルタニュートラルなキャリーでは、現物の建て・解消、perpの建て・解消の4回の執行が発生します。全テイカーだと合計30bps前後になり、年率20%のキャリーでも回収に1週間以上かかります。
だからキャリー戦略ではメイカー執行（PostOnly）が事実上必須です。上のコードでテイカーとメイカーを比較すると、損益分岐日数が数倍変わることが確認できます。実装は記事35で扱います。
"""funding_merge.py — 資金調達をOHLCVに結合し、損益への影響を可視化する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
バックテスト（記事31）に資金調達を織り込む前段。
「ロングを持ち続けた場合、どれだけ削られたか」を実データで確認する。
"""
from __future__ import annotations
import pandas as pd
def attach_funding(bars: pd.DataFrame, funding: pd.DataFrame) -> pd.DataFrame:
"""足に、その足の区間で発生した資金調達率を割り当てる。
足の区間 (prev, current] に含まれる収受を合計する。
複数回入る場合（1時間間隔の銘柄で4時間足など）にも対応。
"""
out = bars.copy()
rates = funding["funding_rate"]
# 各収受を、それが属する足にマッピングする
idx = out.index
assigned = pd.Series(0.0, index=idx, name="funding_rate")
counts = pd.Series(0, index=idx, name="funding_events")
# searchsorted で「その収受時刻以上の最初の足」を求める
positions = idx.searchsorted(rates.index, side="left")
for pos, rate in zip(positions, rates.values):
if pos >= len(idx):
continue
assigned.iloc[pos] += float(rate)
counts.iloc[pos] += 1
out["funding_rate"] = assigned
out["funding_events"] = counts
return out
def long_hold_impact(merged: pd.DataFrame) -> pd.DataFrame:
"""ロングを持ち続けた場合の、価格リターンと資金調達コストを分解する。"""
df = merged.copy()
df["price_return"] = df["close"].pct_change().fillna(0)
# ロングは資金調達率がプラスのとき支払う（マイナスのリターン）
df["funding_cost"] = -df["funding_rate"]
df["net_return"] = df["price_return"] + df["funding_cost"]
df["cum_price"] = (1 + df["price_return"]).cumprod() - 1
df["cum_funding"] = df["funding_cost"].cumsum()
df["cum_net"] = (1 + df["net_return"]).cumprod() - 1
return df
def summarise(df: pd.DataFrame, label: str = "") -> str:
days = (df.index[-1] - df.index[0]).total_seconds() / 86400
price = df["cum_price"].iloc[-1]
fund = df["cum_funding"].iloc[-1]
net = df["cum_net"].iloc[-1]
events = int(df["funding_events"].sum())
share = abs(fund) / abs(price) * 100 if price else float("nan")
return "\n".join([
f"{label} {df.index[0]:%Y-%m-%d} 〜 {df.index[-1]:%Y-%m-%d}"
f"（{days:.0f}日 / 収受{events}回）",
f" 価格リターン : {price:>+9.2%}",
f" 資金調達コスト : {fund:>+9.2%}",
f" 正味リターン : {net:>+9.2%}",
f" 年率の資金調達負担 : {fund / days * 365:>+9.2%}",
f" 価格リターンに対する資金調達の比率: {share:>6.1f}%",
])
if __name__ == "__main__":
from pybit.unified_trading import HTTP
from funding_fetch import fetch_funding_history
from kline_fetch import fetch_klines
http = HTTP(testnet=False)
SYMBOL, START = "BTCUSDT", "2025-01-01"
bars = fetch_klines(http, SYMBOL, interval="60", start=START,
verbose=False)
funding = fetch_funding_history(http, SYMBOL, start=START, verbose=False)
merged = attach_funding(bars, funding)
impact = long_hold_impact(merged)
print(summarise(impact, f"{SYMBOL} ロング保有"))
print("\n--- 月別の資金調達負担 ---")
monthly = impact["funding_cost"].resample("ME").sum()
for ts, v in monthly.items():
bar = "#" * int(abs(v) * 2000)
print(f" {ts:%Y-%m} {v:>+8.3%} {bar}")
print("\n※ この負担を計上しないバックテストは、")
print(" ロング戦略の成績を年率で数%〜十数%過大評価する。")"""funding_monitor.py — 次回収受の直前に、支払いを避ける／受け取りを狙う
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
用途:
- 収受直前に不利なポジションを畳む（支払いを回避）
- 有利な側のポジションを収受直前に建てる
注意: 収受狙いの短期売買は、往復手数料が資金調達を上回ることが多い。
サンプル5の損益分岐計算で必ず確認すること。
"""
from __future__ import annotations
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from decimal import Decimal
@dataclass
class FundingEvent:
symbol: str
rate: Decimal
next_time_ms: int
interval_hours: Decimal
@property
def minutes_to_next(self) -> float:
now_ms = int(time.time() * 1000)
return (self.next_time_ms - now_ms) / 60000
@property
def next_at(self) -> datetime:
return datetime.fromtimestamp(self.next_time_ms / 1000, timezone.utc)
@property
def apr_pct(self) -> Decimal:
times = Decimal("8760") / self.interval_hours
return self.rate * times * 100
def cost_bps_for(self, is_long: bool) -> Decimal:
"""このポジション方向で、次回に支払う（マイナスなら受け取る）bps。"""
return self.rate * 10_000 * (1 if is_long else -1)
def upcoming_funding(http, symbols: list[str], specs: dict,
category: str = "linear") -> list[FundingEvent]:
res = http.get_tickers(category=category)
if res["retCode"] != 0:
raise RuntimeError(f"get_tickers失敗: {res['retMsg']}")
wanted = set(symbols)
out: list[FundingEvent] = []
for t in res["result"]["list"]:
if t["symbol"] not in wanted:
continue
spec = specs.get(t["symbol"])
if spec is None:
continue
out.append(FundingEvent(
symbol=t["symbol"],
rate=Decimal(str(t.get("fundingRate") or 0)),
next_time_ms=int(t.get("nextFundingTime") or 0),
interval_hours=spec.interval_hours,
))
return sorted(out, key=lambda e: e.next_time_ms)
def check_positions_before_funding(client, specs: dict,
warn_minutes: float = 15.0,
warn_bps: Decimal = Decimal("3.0")
) -> list[str]:
"""保有ポジションについて、次回収受の支払い額を警告する。"""
from position import fetch_positions
positions = fetch_positions(client.http)
if not positions:
return []
events = {e.symbol: e for e in upcoming_funding(
client.http, [p.symbol for p in positions], specs)}
warnings: list[str] = []
for p in positions:
ev = events.get(p.symbol)
if ev is None:
continue
cost_bps = ev.cost_bps_for(p.is_long)
cost_usd = p.notional * cost_bps / 10_000
if ev.minutes_to_next <= warn_minutes and cost_bps >= warn_bps:
warnings.append(
f"{p.symbol} {p.side}: {ev.minutes_to_next:.0f}分後に "
f"{cost_bps:.2f} bps（約 {cost_usd:.2f} USDT）支払い "
f"[年率 {ev.apr_pct:+.1f}%]"
)
elif ev.minutes_to_next <= warn_minutes and cost_bps <= -warn_bps:
warnings.append(
f"{p.symbol} {p.side}: {ev.minutes_to_next:.0f}分後に "
f"{-cost_bps:.2f} bps（約 {-cost_usd:.2f} USDT）受取り"
)
return warnings
if __name__ == "__main__":
from client import make_client
from annualize import fetch_funding_specs
c = make_client()
specs = fetch_funding_specs(c.http)
watch = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
print(f"{'symbol':<11}{'レート':>11}{'年率':>10}{'次回まで':>10}"
f"{'次回時刻(UTC)':>18}")
print("-" * 62)
for ev in upcoming_funding(c.http, watch, specs):
print(f"{ev.symbol:<11}{ev.rate * 100:>+10.4f}%"
f"{ev.apr_pct:>+9.1f}%{ev.minutes_to_next:>9.0f}分"
f"{ev.next_at:%Y-%m-%d %H:%M}")
print("\n--- 保有ポジションの支払い予定 ---")
ws = check_positions_before_funding(c, specs, warn_minutes=60,
warn_bps=Decimal("1.0"))
if ws:
for w in ws:
print(f" {w}")
else:
print(" 警告対象はありません")試算だけでなく、実際にいくら払った／受け取ったかを確認できます。get_transaction_log の SETTLEMENT 種別が資金調達です。
"""funding_actual.py — 実際に発生した資金調達を取引ログから集計する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
注意: get_transaction_log の時間範囲は最大7日。
長期間を集計するには7日ずつ窓をずらして繰り返す。
"""
from __future__ import annotations
import time
from datetime import datetime, timedelta, timezone
import pandas as pd
WINDOW_DAYS = 7 # APIの上限
MAX_PAGES = 200
def fetch_settlements(http, days: int = 30,
account_type: str = "UNIFIED") -> pd.DataFrame:
"""指定日数分の資金調達（SETTLEMENT）を7日窓で取得する。"""
end = datetime.now(timezone.utc)
start = end - timedelta(days=days)
rows: list[dict] = []
window_start = start
while window_start < end:
window_end = min(window_start + timedelta(days=WINDOW_DAYS), end)
cursor = None
for _ in range(MAX_PAGES):
kw: dict = {
"accountType": account_type,
"type": "SETTLEMENT", # 資金調達
"startTime": int(window_start.timestamp() * 1000),
"endTime": int(window_end.timestamp() * 1000),
"limit": 50,
}
if cursor:
kw["cursor"] = cursor
res = http.get_transaction_log(**kw)
if res["retCode"] != 0:
print(f"[warn] {window_start:%Y-%m-%d}: "
f"{res['retCode']} {res['retMsg']}")
break
page = res["result"]["list"]
rows.extend(page)
cursor = res["result"].get("nextPageCursor")
if not cursor or not page:
break
time.sleep(0.05)
window_start = window_end
if not rows:
return pd.DataFrame()
df = pd.DataFrame(rows)
df["ts"] = pd.to_datetime(df["transactionTime"].astype("int64"),
unit="ms", utc=True)
for col in ("funding", "fee", "cashFlow", "change", "cashBalance"):
if col in df.columns:
df[col] = pd.to_numeric(df[col], errors="coerce").fillna(0.0)
return df.set_index("ts").sort_index()
def report(df: pd.DataFrame) -> None:
if df.empty:
print("該当期間に資金調達の記録はありません")
return
# fundingフィールドがマイナスなら支払い、プラスなら受取り
total = df["funding"].sum()
paid = df[df["funding"] < 0]["funding"].sum()
received = df[df["funding"] > 0]["funding"].sum()
print(f"期間: {df.index[0]:%Y-%m-%d} 〜 {df.index[-1]:%Y-%m-%d}")
print(f" 記録件数 : {len(df)}")
print(f" 受け取り : {received:>+12,.4f}")
print(f" 支払い : {paid:>+12,.4f}")
print(f" 正味 : {total:>+12,.4f}")
if "symbol" in df.columns:
print("\n--- 銘柄別 ---")
by_symbol = (df.groupby("symbol")["funding"]
.agg(["sum", "count"])
.sort_values("sum"))
for sym, r in by_symbol.iterrows():
print(f" {sym:<14}{r['sum']:>+12,.4f} ({int(r['count'])}回)")
print("\n--- 週別 ---")
weekly = df["funding"].resample("W").sum()
for ts, v in weekly.items():
print(f" {ts:%Y-%m-%d}週 {v:>+10,.4f}")
if __name__ == "__main__":
from client import make_client
c = make_client()
df = fetch_settlements(c.http, days=30)
report(df)
print("\n※ 試算（サンプル5）と実測がずれる場合、")
print(" ポジション量の変動やデルタの乖離が原因。")
print(" 年間の損益集計は記事40で扱う。")get_tickers の fundingRate は次回適用される予定のレートです（確定値ではなく収受時刻まで変動します）。より精度を上げたいなら get_premium_index_price_kline（プレミアム指数）を使います。プレミアム指数は先物と現物指数の乖離を表し、これが資金調達率の算出根拠になるため、収受までの推移を見れば方向は読めます。startTime 単独はエラー、endTime 単独はその時刻までの200件、両方省略で最新200件、という挙動です。endTime を使って新しい方から遡るのが唯一の全期間取得方法です（サンプル1）。記事07のklineも同じパターンなので、ページネーションの実装を共通化できます。symbol が必須なので銘柄ごとに繰り返す必要があります。500銘柄×1年分（8時間間隔なら約1,100件＝6ページ）で約3,000リクエストです。記事05の流量ゲートを通し、Parquetにキャッシュ（サンプル1の load_or_fetch）してください。funding フィールドはマイナスになります。投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません（間接的・付随的な障害・損失も含め一切の責任を負いません）。自己責任でご利用ください。
当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。
当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。
当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。
当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。
当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。
当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。