※ Bybit V5は機能追加のペースが速いAPIです。パラメータの最終的な正解は必ず公式ドキュメントで確認してください。本記事のコードはすべてテストネットで動作確認しています。
| 課題 | 自宅PC | VPS |
|---|---|---|
| 24時間稼働 | スリープ・OS更新で停止 | 常時稼働 |
| 停電・回線断 | ポジションが放置される | データセンターの冗長回線 |
| レイテンシ | 自宅回線経由で数十〜百ms超 | 近いリージョンなら数ms〜 |
| 時刻精度 | OSの設定次第 | chronyで常時同期 |
| IP固定 | 動的IPだとAPIキーのIP制限が使えない | 固定IPでキーを保護（記事01） |
| コスト | 追加費用なし | 月数百円〜 |
VPSを使う最大の理由はレイテンシではなく「止まらないこと」です。botが落ちている間、TP/SLを自作トレールで管理していれば損切りが動きません。
だから記事14で述べたとおり、取引所側に広めの損切りを置いておくことが前提です。VPSと自動復帰は「落ちる確率を下げる」対策であり、「落ちない保証」ではありません。落ちた場合の被害を限定する仕組み（取引所側SL）と併用してください。
「シンガポールが速い」といった伝聞ではなく、自分で測ってから決めてください。
"""measure_latency.py — Bybit APIへのレイテンシを実測する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
VPSの候補それぞれでこれを走らせ、結果を比較してから契約する。
無料期間や時間課金を使えば、契約前に測れることも多い。
"""
from __future__ import annotations
import socket
import statistics
import ssl
import time
from dataclasses import dataclass, field
import requests
HOSTS = {
"REST (api.bybit.com)": ("api.bybit.com", 443),
"REST 代替 (api.bytick.com)": ("api.bytick.com", 443),
"WS (stream.bybit.com)": ("stream.bybit.com", 443),
}
REST_ENDPOINT = "https://api.bybit.com/v5/market/time"
@dataclass
class LatencyResult:
label: str
samples: list[float] = field(default_factory=list)
failures: int = 0
def stats(self) -> dict:
if len(self.samples) < 2:
return {"n": len(self.samples), "error": "サンプル不足"}
s = sorted(self.samples)
def pct(q: float) -> float:
return s[min(int(len(s) * q), len(s) - 1)]
return {
"n": len(s),
"min": round(s[0], 2),
"p50": round(statistics.median(s), 2),
"p90": round(pct(0.90), 2),
"p99": round(pct(0.99), 2),
"max": round(s[-1], 2),
"stdev": round(statistics.stdev(s), 2),
"fail": self.failures,
}
def line(self) -> str:
st = self.stats()
if "error" in st:
return f" {self.label:<30} {st['error']}"
return (f" {self.label:<30} n={st['n']:<4} "
f"min {st['min']:>7.2f} p50 {st['p50']:>7.2f} "
f"p90 {st['p90']:>7.2f} p99 {st['p99']:>7.2f} "
f"σ {st['stdev']:>6.2f} 失敗 {st['fail']}")
def tcp_connect_ms(host: str, port: int, timeout: float = 5.0) -> float:
"""TCP接続の確立までの時間（ms）。ネットワーク距離の指標。"""
t0 = time.perf_counter()
with socket.create_connection((host, port), timeout=timeout):
pass
return (time.perf_counter() - t0) * 1000
def tls_handshake_ms(host: str, port: int, timeout: float = 5.0) -> float:
"""TLSハンドシェイクまで含めた時間（ms）。
REST発注では毎回これが発生しうるため（コネクション再利用しない場合）、
WS発注（記事29）との差を理解するのに役立つ。
"""
ctx = ssl.create_default_context()
t0 = time.perf_counter()
with socket.create_connection((host, port), timeout=timeout) as sock:
with ctx.wrap_socket(sock, server_hostname=host):
pass
return (time.perf_counter() - t0) * 1000
def rest_roundtrip_ms(session: requests.Session,
timeout: float = 5.0) -> tuple[float, float]:
"""REST往復時間と、サーバー時刻とのズレ（ms）を返す。"""
t0 = time.perf_counter()
r = session.get(REST_ENDPOINT, timeout=timeout)
rtt = (time.perf_counter() - t0) * 1000
r.raise_for_status()
server_ms = int(r.json()["result"]["timeNano"]) / 1_000_000
local_ms = time.time() * 1000 - rtt / 2 # 片道分を補正
return rtt, server_ms - local_ms
def run_all(samples: int = 30, interval: float = 0.3) -> None:
print("=" * 78)
print("Bybit APIへのレイテンシ計測（この環境から）")
print("=" * 78)
# 1) TCP / TLS
for label, (host, port) in HOSTS.items():
tcp = LatencyResult(f"TCP {label}")
tls = LatencyResult(f"TLS {label}")
for _ in range(samples):
for res, fn in ((tcp, tcp_connect_ms), (tls, tls_handshake_ms)):
try:
res.samples.append(fn(host, port))
except Exception:
res.failures += 1
time.sleep(interval)
print(tcp.line())
print(tls.line())
# 2) REST往復（コネクション再利用あり）
session = requests.Session()
rest = LatencyResult("REST 往復（keep-alive）")
skews: list[float] = []
for _ in range(samples):
try:
rtt, skew = rest_roundtrip_ms(session)
rest.samples.append(rtt)
skews.append(skew)
except Exception:
rest.failures += 1
time.sleep(interval)
print(rest.line())
if skews:
med = statistics.median(skews)
print(f"\n サーバー時刻とのズレ（中央値）: {med:+.1f} ms")
if abs(med) > 1000:
print(" [警告] 1秒以上ズレています。chronyを設定してください")
print(" （retCode 10002 の原因になります）")
print("\n--- 判断の目安 ---")
print(" p50 が 50ms 未満 : 十分（ほとんどの戦略で問題なし）")
print(" p50 が 50〜150ms : 許容（数分足以上の戦略なら問題なし）")
print(" p50 が 150ms 超 : マーケットメイク等の低遅延戦略には不向き")
print(" σ（標準偏差）が大きい: 回線が不安定。値の大小より安定性が重要")
print("\n ※ TLSハンドシェイクの時間が大きい環境では、")
print(" コネクション再利用（Session）とWS発注（記事29）の効果が大きい")
if __name__ == "__main__":
run_all(samples=30)p50が20msでもp99が2000msなら、その回線は使えません。標準偏差（σ）とp99を見てください。安定して80msの回線のほうが、平均20msで時々2秒かかる回線より運用しやすいです。
またほとんどの戦略ではレイテンシは問題になりません。1時間足のトレンドフォロー（記事33）に100msの差は無関係です。レイテンシが効くのはマーケットメイク（記事37）やWS発注（記事29）を使う低遅延戦略だけです。「速いから」ではなく「安定していて安いから」で選んでください。
「クラウドだから時刻は正確」は誤りです。仮想化環境ではホストのクロックドリフト、ライブマイグレーション、サスペンドからの復帰などで数秒ずれます。
ズレが recv_window（既定5000ms）を超えるとすべてのリクエストが retCode 10002 で落ちます。しかも「たまに出る」形で現れるため、原因を掴みにくい。chronyの導入は必須です。
#!/usr/bin/env bash
# setup_chrony.sh — 時刻同期の設定（Ubuntu / Debian）
# 本スクリプトは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
set -euo pipefail
echo "=== 1) systemd-timesyncd を止めて chrony に統一する ==="
# 両方動かすと競合するため、片方に絞る
sudo systemctl stop systemd-timesyncd 2>/dev/null || true
sudo systemctl disable systemd-timesyncd 2>/dev/null || true
echo "=== 2) chrony をインストール ==="
sudo apt-get update -qq
sudo apt-get install -y chrony
echo "=== 3) 設定を書き込む ==="
sudo tee /etc/chrony/conf.d/bybit-bot.conf > /dev/null <<'CONF'
# 複数のNTPプールを使う（1つが落ちても同期が続く）
pool time.cloudflare.com iburst maxsources 4
pool ntp.nict.jp iburst maxsources 2
pool pool.ntp.org iburst maxsources 2
# 起動直後の大きなズレは即座に補正する（stepでジャンプさせる）
# 取引botでは「徐々に合わせる」より即座に合わせるほうが安全
makestep 0.1 3
# ドリフト補正を有効化
driftfile /var/lib/chrony/chrony.drift
rtcsync
# ログ（時刻ズレの調査に使う）
logdir /var/log/chrony
log tracking measurements statistics
CONF
echo "=== 4) 再起動して即座に同期させる ==="
sudo systemctl enable chrony
sudo systemctl restart chrony
sleep 3
sudo chronyc makestep
echo "=== 5) 確認 ==="
chronyc tracking
echo
chronyc sources -v | head -20
echo
echo "System time の行が 0.001秒（1ms）以内なら良好です。"
echo "0.1秒を超えている場合は、しばらく待ってから再確認してください。"
echo
echo "定期監視には monitor_clock.py（サンプル3）をcronに登録してください。""""monitor_clock.py — 時刻ズレを監視し、閾値超過で通知する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
cronで5分ごとに実行する:
*/5 * * * * cd /opt/bybit-bot && .venv/bin/python monitor_clock.py
"""
from __future__ import annotations
import json
import statistics
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
import requests
SERVER_TIME_URL = "https://api.bybit.com/v5/market/time"
STATE_PATH = Path("data/clock_state.json")
WARN_MS = 500.0 # これを超えたら警告
CRITICAL_MS = 2000.0 # これを超えたら危険（recv_window 5000msの40%）
@dataclass
class ClockStatus:
offset_ms: float
rtt_ms: float
chrony_offset_ms: float | None
chrony_synced: bool
@property
def level(self) -> str:
a = abs(self.offset_ms)
if a >= CRITICAL_MS:
return "critical"
if a >= WARN_MS:
return "warn"
return "ok"
def report(self) -> str:
lines = [
f"Bybitサーバーとのズレ : {self.offset_ms:+8.1f} ms "
f"[{self.level}]",
f"RTT（中央値） : {self.rtt_ms:8.1f} ms",
]
if self.chrony_offset_ms is not None:
lines.append(f"chronyのシステム誤差 : "
f"{self.chrony_offset_ms:+8.3f} ms "
f"（同期 {'OK' if self.chrony_synced else 'NG'}）")
else:
lines.append("chronyの状態 : 取得できません"
"（インストールされていない可能性）")
if self.level != "ok":
lines.append("")
lines.append(" 対処: sudo chronyc makestep")
lines.append(" （それでも直らなければNTP設定を確認）")
return "\n".join(lines)
def measure_offset(samples: int = 5) -> tuple[float, float]:
"""Bybitサーバー時刻とのズレとRTTを測る（中央値）。"""
offsets, rtts = [], []
session = requests.Session()
for _ in range(samples):
t0 = time.time()
r = session.get(SERVER_TIME_URL, timeout=10)
t1 = time.time()
r.raise_for_status()
server_ms = int(r.json()["result"]["timeNano"]) / 1_000_000
local_mid_ms = (t0 + t1) / 2 * 1000
offsets.append(server_ms - local_mid_ms)
rtts.append((t1 - t0) * 1000)
time.sleep(0.2)
return statistics.median(offsets), statistics.median(rtts)
def chrony_status() -> tuple[float | None, bool]:
"""chronycの出力からシステム誤差と同期状態を取得する。"""
try:
out = subprocess.run(["chronyc", "-c", "tracking"],
capture_output=True, text=True, timeout=5)
if out.returncode != 0:
return None, False
# CSV形式: refid,ip,stratum,ref_time,system_time,last_offset,...
fields = out.stdout.strip().split(",")
system_time_sec = float(fields[4])
stratum = int(fields[2])
return system_time_sec * 1000, (0 < stratum < 16)
except Exception:
return None, False
def check() -> ClockStatus:
offset, rtt = measure_offset()
chrony_offset, synced = chrony_status()
return ClockStatus(offset_ms=offset, rtt_ms=rtt,
chrony_offset_ms=chrony_offset, chrony_synced=synced)
def save_history(status: ClockStatus, keep: int = 500) -> None:
"""履歴を残しておくと、ドリフトの傾向が見える。"""
STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
history = []
if STATE_PATH.exists():
try:
history = json.loads(STATE_PATH.read_text(encoding="utf-8"))
except Exception:
history = []
history.append({"ts": time.time(), "offset_ms": status.offset_ms,
"level": status.level})
history = history[-keep:]
tmp = STATE_PATH.with_suffix(".tmp")
tmp.write_text(json.dumps(history), encoding="utf-8")
tmp.replace(STATE_PATH)
def try_resync() -> bool:
"""chronycで即座に補正を試みる（要sudo権限）。"""
try:
r = subprocess.run(["sudo", "-n", "chronyc", "makestep"],
capture_output=True, text=True, timeout=10)
return r.returncode == 0
except Exception:
return False
if __name__ == "__main__":
import sys
status = check()
print(status.report())
save_history(status)
if status.level == "critical":
print("\n[critical] 自動補正を試みます")
if try_resync():
time.sleep(2)
after = check()
print("\n補正後:")
print(after.report())
if after.level == "critical":
sys.exit(2)
else:
print(" 補正に失敗（sudo権限が無いか、chronyが未導入）")
sys.exit(2)
elif status.level == "warn":
sys.exit(1)# Dockerfile — Bybit bot用の実行イメージ
# 本ファイルは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
# ---- ビルドステージ（依存のコンパイルをここで済ませる） ----------------
FROM python:3.12-slim AS builder
ENV PIP_NO_CACHE_DIR=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1
WORKDIR /build
# 依存のインストールだけを先に行う（レイヤキャッシュを効かせる）
COPY requirements.txt .
RUN python -m venv /opt/venv \
&& /opt/venv/bin/pip install --upgrade pip \
&& /opt/venv/bin/pip install -r requirements.txt
# ---- 実行ステージ -------------------------------------------------------
FROM python:3.12-slim
# タイムゾーンはUTCに固定する。
# ログとデータをUTCで統一しておけば、日足の境界（記事07）で混乱しない
ENV TZ=UTC \
PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
PATH="/opt/venv/bin:$PATH"
# ca-certificates はHTTPS通信に必須。tzdata は時刻表示用
RUN apt-get update \
&& apt-get install -y --no-install-recommends ca-certificates tzdata curl \
&& rm -rf /var/lib/apt/lists/*
# rootで動かさない（コンテナ侵害時の影響を限定する）
RUN useradd --create-home --shell /bin/bash --uid 10001 bot
COPY --from=builder /opt/venv /opt/venv
WORKDIR /app
COPY --chown=bot:bot . /app
# データディレクトリは永続化ボリュームにマウントする
RUN mkdir -p /app/data /app/logs && chown -R bot:bot /app/data /app/logs
USER bot
# ヘルスチェック: 自己診断スクリプト（記事06）が通るかを確認する
HEALTHCHECK --interval=120s --timeout=30s --start-period=60s --retries=3 \
CMD python -c "import sys; from monitor_clock import check; \
sys.exit(0 if check().level != 'critical' else 1)"
# 停止シグナルを正しく受け取れるよう exec 形式で書く
# （シェル形式だとSIGTERMがPythonに届かず、graceful shutdownが動かない）
ENTRYPOINT ["python", "-u"]
CMD ["run_trend_bot.py"]CMD python run_bot.py（シェル形式）と書くと、PID 1 が /bin/sh になり、SIGTERMがPythonプロセスに転送されません。結果、docker stop したときに記事30のgraceful shutdownが実行されず、未約定注文が残ります。
必ずexec形式（JSON配列）で書いてください。上のDockerfileは ENTRYPOINT ["python", "-u"] という配列形式にしています。-u はログをバッファリングせず即座に出すためです（Dockerのログで見えなくなるのを防ぐ）。
# docker-compose.yml — botと補助プロセスの構成
# 本ファイルは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
services:
# ---- 売買bot本体 ----------------------------------------------------
bot:
build: .
image: bybit-bot:latest
container_name: bybit-bot
restart: unless-stopped # 手動停止以外は自動復帰
env_file:
- .env # APIキー（イメージには含めない）
environment:
BYBIT_ENV: mainnet
BYBIT_ALLOW_LIVE: "1" # 記事04の多層ガード
TZ: UTC
volumes:
- ./data:/app/data # 注文台帳・履歴（永続化が必須）
- ./logs:/app/logs
- ./config:/app/config:ro # 設定は読み取り専用でマウント
- ./LIVE_ENABLED:/app/LIVE_ENABLED:ro # 本番許可トークン
command: ["run_trend_bot.py"]
stop_grace_period: 60s # graceful shutdownの猶予（記事30）
logging:
driver: json-file
options:
max-size: "50m" # ログでディスクを埋めない
max-file: "5"
deploy:
resources:
limits:
memory: 1G # メモリリークで巻き込まれないよう上限
healthcheck:
test: ["CMD", "python", "-c", "import sys; from monitor_clock import check; sys.exit(0 if check().level != 'critical' else 1)"]
interval: 120s
timeout: 30s
retries: 3
start_period: 60s
# ---- 履歴の同期（記事18）。売買botとは別プロセスにする ---------------
# 理由: 履歴の一括取得は数千リクエストになり、
# 同じプロセスだとIP制限で売買まで止まる（記事05）
history-sync:
image: bybit-bot:latest
container_name: bybit-history-sync
restart: unless-stopped
env_file:
- .env
environment:
BYBIT_ENV: mainnet
TZ: UTC
volumes:
- ./data:/app/data
- ./logs:/app/logs
entrypoint: ["/bin/sh", "-c"]
command:
- |
while true; do
python -u scripts/sync_history.py || echo "sync failed"
sleep 3600
done
logging:
driver: json-file
options:
max-size: "20m"
max-file: "3"
# ---- 時刻監視（サンプル3）------------------------------------------
clock-monitor:
image: bybit-bot:latest
container_name: bybit-clock
restart: unless-stopped
volumes:
- ./data:/app/data
entrypoint: ["/bin/sh", "-c"]
command:
- |
while true; do
python -u monitor_clock.py || echo "clock drift detected"
sleep 300
done
logging:
driver: json-file
options:
max-size: "10m"
max-file: "2"記事05で述べたとおり、IP制限（5秒600リクエスト）を超えると約10分間すべてのAPIが止まります。履歴の一括取得（記事18の7日窓ループ）は数千リクエストになるため、売買botと同じプロセスで走らせると売買まで止まるリスクがあります。
プロセスを分けても同じIPなので制限は共有されますが、少なくとも売買botのイベントループが履歴取得でブロックされることは防げます。より安全にするなら、履歴同期を別VPSまたは別IPで動かしてください。
Dockerの restart: unless-stopped でコンテナは復帰しますが、VPS自体が再起動したときにDockerサービスごと起動させるにはsystemdが必要です。
# bybit-bot.service — docker-composeをsystemdで管理する
# 本ファイルは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
#
# 設置:
# sudo cp bybit-bot.service /etc/systemd/system/
# sudo systemctl daemon-reload
# sudo systemctl enable --now bybit-bot
#
# 確認:
# systemctl status bybit-bot
# journalctl -u bybit-bot -f
[Unit]
Description=Bybit trading bot (docker compose)
Documentation=https://codetribe.xsrv.jp/ea/tips-bybit-bot-vps
# Dockerとネットワークと「時刻同期」の後に起動する。
# chronyより先に起動すると、時計がズレた状態で発注しかねない
After=docker.service network-online.target chrony.service
Wants=network-online.target
Requires=docker.service
[Service]
Type=oneshot
RemainAfterExit=yes
WorkingDirectory=/opt/bybit-bot
# 起動前に時刻を合わせる（10002の予防）
ExecStartPre=/usr/bin/chronyc makestep
ExecStartPre=/bin/sleep 3
ExecStart=/usr/bin/docker compose up -d --remove-orphans
# graceful shutdownの猶予を十分に取る（記事30の停止処理）
ExecStop=/usr/bin/docker compose down --timeout 90
ExecReload=/usr/bin/docker compose up -d --remove-orphans
TimeoutStartSec=300
TimeoutStopSec=120
# 失敗したら再試行する
Restart=on-failure
RestartSec=30
# ログはjournaldへ
StandardOutput=journal
StandardError=journal
SyslogIdentifier=bybit-bot
[Install]
WantedBy=multi-user.target#!/usr/bin/env bash
# setup_vps.sh — VPSの初期構築（Ubuntu 24.04想定）
# 本スクリプトは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
set -euo pipefail
APP_DIR=/opt/bybit-bot
echo "=== 1) パッケージ更新と基本ツール ==="
sudo apt-get update -qq
sudo apt-get install -y ca-certificates curl git logrotate ufw
echo "=== 2) Docker のインストール ==="
if ! command -v docker >/dev/null 2>&1; then
sudo install -m 0755 -d /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg \
| sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \
https://download.docker.com/linux/ubuntu $(. /etc/os-release && echo "$VERSION_CODENAME") stable" \
| sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt-get update -qq
sudo apt-get install -y docker-ce docker-ce-cli containerd.io \
docker-buildx-plugin docker-compose-plugin
sudo usermod -aG docker "$USER"
echo " -> 再ログインするとsudoなしでdockerが使えます"
fi
echo "=== 3) 時刻同期（サンプル2）==="
bash ./setup_chrony.sh
echo "=== 4) ファイアウォール（SSHのみ許可）==="
# botは外部からの接続を受けないため、着信は原則すべて閉じる
sudo ufw default deny incoming
sudo ufw default allow outgoing
sudo ufw allow 22/tcp comment 'SSH'
sudo ufw --force enable
sudo ufw status verbose
echo "=== 5) アプリディレクトリ ==="
sudo mkdir -p "$APP_DIR"
sudo chown "$USER:$USER" "$APP_DIR"
mkdir -p "$APP_DIR"/{data,logs,config}
chmod 700 "$APP_DIR/data" # 注文台帳は他ユーザーに読ませない
echo "=== 6) グローバルIPを表示（APIキーのIP制限に登録する）==="
echo -n " このVPSのグローバルIP: "
curl -s https://api.ipify.org || echo "取得失敗"
echo
echo " -> このIPをBybitのAPIキー設定に登録してください（記事01）"
echo
echo "=== 残りの手順 ==="
cat <<'STEPS'
1) リポジトリを配置
cd /opt/bybit-bot && git clone <your-repo> .
2) .env を作成（APIキー。権限600）
nano .env && chmod 600 .env
3) 本番許可トークンを作成（記事04の多層ガード）
touch LIVE_ENABLED
4) 設定を確認
python -c "from config_model import load_settings; print(load_settings().banner())"
5) 自己診断（記事06）
docker compose run --rm bot scripts/diagnose.py
6) systemdに登録
sudo cp bybit-bot.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now bybit-bot
7) ログを確認
docker compose logs -f bot
STEPS| 方法 | 安全性 | 手間 | 向く場面 |
|---|---|---|---|
.env（権限600） |
最低限。VPS侵害時は読まれる | 最小 | 個人運用の標準 |
| Docker secrets | 環境変数より漏れにくい | 中（Swarm前提） | 複数ノード運用 |
| SOPS + age | 暗号化してリポジトリに置ける | 中 | CI経由のデプロイ |
| クラウドのSecret Manager | 高い | 大 | 本格運用 |
暗号化方式を凝るより、記事01で述べた「出金権限を付けない」「IP制限を掛ける」のほうが効果が大きいです。この2つを守れば、キーが漏れても資金を外部に持ち出されることはありません。
逆に、出金権限付きのキーをどれだけ厳重に暗号化しても、復号する鍵はそのVPSにあります。権限を絞ることが唯一の根本対策です。
#!/usr/bin/env bash
# secrets_setup.sh — シークレットの保護と、SOPS+ageによる暗号化管理
# 本スクリプトは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
set -euo pipefail
APP_DIR=${APP_DIR:-/opt/bybit-bot}
cd "$APP_DIR"
echo "=== 1) .env の権限を絞る（最低限これだけは必須）==="
if [ -f .env ]; then
chmod 600 .env
echo " .env -> 600 (本人のみ読み書き)"
ls -l .env
else
echo " [警告] .env がありません"
fi
echo
echo "=== 2) .gitignore の確認 ==="
REQUIRED=(".env" ".venv/" "data/" "logs/" "LIVE_ENABLED" "secrets/" "*.pem")
touch .gitignore
for pat in "${REQUIRED[@]}"; do
if ! grep -qxF "$pat" .gitignore; then
echo "$pat" >> .gitignore
echo " 追加: $pat"
fi
done
echo
echo "=== 3) リポジトリにキーが混入していないか確認 ==="
if git rev-parse --git-dir >/dev/null 2>&1; then
# 追跡されているファイルにAPIキーらしき文字列が無いか探す
if git grep -nE "(BYBIT_[A-Z_]*API_(KEY|SECRET)\s*=\s*[A-Za-z0-9]{10,})" \
-- . ':!*.md' 2>/dev/null; then
echo " [危険] コミット済みファイルにキーらしき値があります"
echo " 該当キーを即座に無効化し、新しいキーを発行してください"
exit 1
fi
echo " 問題は見つかりませんでした"
fi
echo
echo "=== 4) SOPS + age で暗号化管理する（任意）==="
cat <<'SOPS'
暗号化した .env をリポジトリに置きたい場合の手順:
# インストール
sudo apt-get install -y age
curl -LO https://github.com/getsops/sops/releases/latest/download/sops-v3.9.0.linux.amd64
sudo install -m 755 sops-v3.9.0.linux.amd64 /usr/local/bin/sops
# 鍵を生成（この秘密鍵はVPSの外に出さない）
mkdir -p ~/.config/sops/age
age-keygen -o ~/.config/sops/age/keys.txt
chmod 600 ~/.config/sops/age/keys.txt
grep "public key" ~/.config/sops/age/keys.txt
# .sops.yaml に公開鍵を書く
cat > .sops.yaml <<'YAML'
creation_rules:
- path_regex: \.env\.enc$
age: age1xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
YAML
# 暗号化（.env.enc はコミットしてよい）
sops --encrypt .env > .env.enc
# 復号（起動前に実行）
export SOPS_AGE_KEY_FILE=~/.config/sops/age/keys.txt
sops --decrypt .env.enc > .env && chmod 600 .env
SOPS
echo
echo "=== 5) 最終確認 ==="
echo " 以下をBybitの管理画面で確認してください（記事01）:"
echo " [ ] 出金権限がオフになっている <- 最重要"
echo " [ ] このVPSのIPが許可リストに登録されている"
echo " [ ] キーの有効期限を把握している"
echo " [ ] 分析用と売買用でキーを分けている"# /etc/logrotate.d/bybit-bot
# アプリが直接ファイルに書くログをローテーションする。
# Dockerのstdoutログは docker-compose.yml の logging オプションで制御済み。
/opt/bybit-bot/logs/*.log {
daily
rotate 30
maxsize 100M
compress
delaycompress
missingok
notifempty
copytruncate # プロセスを止めずに切り替える
dateext
dateformat -%Y%m%d
su bot bot
}
/opt/bybit-bot/logs/*.jsonl {
daily
rotate 90 # 構造化ログは長めに残す（障害調査で使う）
maxsize 200M
compress
delaycompress
missingok
notifempty
copytruncate
su bot bot
}意外に多い障害原因がディスク満杯です。ログが無制限に増えると、SQLiteの書き込みが失敗し、注文台帳（記事17）が更新できなくなります。
対策は3層です。(1) Dockerの max-size / max-file（compose側で設定済み）、(2) logrotate（上の設定）、(3) ディスク使用率の監視（次のサンプル）。ログレベルをDEBUGにしたまま放置すると数日で埋まります。
"""healthcheck.py — VPSの稼働状態を点検し、異常を通知する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
cronで10分ごとに実行する:
*/10 * * * * cd /opt/bybit-bot && .venv/bin/python healthcheck.py
"""
from __future__ import annotations
import os
import shutil
import subprocess
import time
from dataclasses import dataclass, field
from pathlib import Path
APP_DIR = Path(os.getenv("APP_DIR", "/opt/bybit-bot"))
CONTAINERS = ("bybit-bot", "bybit-history-sync")
DISK_WARN_PCT = 80.0
DISK_CRIT_PCT = 90.0
MEM_WARN_PCT = 85.0
LEDGER_STALE_SEC = 3600.0
@dataclass
class Finding:
level: str # ok / warn / critical
name: str
detail: str
def line(self) -> str:
icon = {"ok": " ", "warn": "! ", "critical": "!!"}.get(self.level, " ")
return f"{icon} {self.name:<22} {self.detail}"
@dataclass
class HealthReport:
findings: list[Finding] = field(default_factory=list)
def add(self, level: str, name: str, detail: str) -> None:
self.findings.append(Finding(level, name, detail))
@property
def worst(self) -> str:
levels = {f.level for f in self.findings}
if "critical" in levels:
return "critical"
if "warn" in levels:
return "warn"
return "ok"
def report(self) -> str:
lines = [f"VPS健全性チェック [{self.worst}]", "-" * 56]
lines += [f.line() for f in self.findings]
return "\n".join(lines)
# ---- 各チェック ---------------------------------------------------------
def check_disk(rep: HealthReport) -> None:
usage = shutil.disk_usage(str(APP_DIR if APP_DIR.exists() else "/"))
pct = usage.used / usage.total * 100
free_gb = usage.free / 1024 ** 3
detail = f"使用率 {pct:.1f}% / 空き {free_gb:.1f} GB"
level = ("critical" if pct >= DISK_CRIT_PCT
else "warn" if pct >= DISK_WARN_PCT else "ok")
rep.add(level, "ディスク", detail)
def check_memory(rep: HealthReport) -> None:
try:
info = {}
for line in Path("/proc/meminfo").read_text().splitlines():
key, _, rest = line.partition(":")
info[key] = int(rest.strip().split()[0])
total = info["MemTotal"]
available = info.get("MemAvailable", info.get("MemFree", 0))
pct = (total - available) / total * 100
level = "warn" if pct >= MEM_WARN_PCT else "ok"
rep.add(level, "メモリ",
f"使用率 {pct:.1f}% / 空き {available / 1024 ** 2:.2f} GB")
except Exception as e:
rep.add("warn", "メモリ", f"取得できません: {e}")
def check_containers(rep: HealthReport) -> None:
for name in CONTAINERS:
try:
r = subprocess.run(
["docker", "inspect", "-f",
"{{.State.Status}}|{{.State.Health.Status}}|{{.RestartCount}}",
name],
capture_output=True, text=True, timeout=10)
if r.returncode != 0:
rep.add("critical", f"container {name}", "存在しません")
continue
status, health, restarts = (r.stdout.strip().split("|") + ["", ""])[:3]
level = "ok" if status == "running" else "critical"
if health == "unhealthy":
level = "critical"
try:
if int(restarts or 0) > 5:
level = "warn" if level == "ok" else level
except ValueError:
pass
rep.add(level, f"container {name}",
f"{status} / health={health or '-'} / 再起動{restarts}回")
except FileNotFoundError:
rep.add("warn", "docker", "コマンドが見つかりません")
return
except Exception as e:
rep.add("warn", f"container {name}", f"確認失敗: {e}")
def check_clock(rep: HealthReport) -> None:
try:
from monitor_clock import check as clock_check
st = clock_check()
rep.add(st.level, "時刻同期",
f"ズレ {st.offset_ms:+.0f} ms / RTT {st.rtt_ms:.0f} ms")
except Exception as e:
rep.add("warn", "時刻同期", f"確認失敗: {e}")
def check_ledger(rep: HealthReport) -> None:
"""注文台帳が更新されているか（botが実際に動いているか）。"""
path = APP_DIR / "data" / "orders.sqlite"
if not path.exists():
rep.add("warn", "注文台帳", "ファイルがありません")
return
age = time.time() - path.stat().st_mtime
level = "warn" if age > LEDGER_STALE_SEC else "ok"
rep.add(level, "注文台帳",
f"最終更新 {age / 60:.0f}分前 / "
f"{path.stat().st_size / 1024:.0f} KB")
def check_live_token(rep: HealthReport) -> None:
"""本番許可トークン（記事04）の存在確認。"""
exists = (APP_DIR / "LIVE_ENABLED").exists()
env = os.getenv("BYBIT_ENV", "?")
if env == "mainnet" and not exists:
rep.add("critical", "本番許可トークン",
"LIVE_ENABLED がありません（botが起動できません）")
else:
rep.add("ok", "本番許可トークン",
f"{'あり' if exists else 'なし'} (BYBIT_ENV={env})")
def notify(report: HealthReport) -> None:
"""critical時のみ通知する（記事39の通知モジュールを使う）。"""
if report.worst != "critical":
return
try:
from notify import Notifier
Notifier().send(f"VPS異常\n```\n{report.report()}\n```", force=True)
except Exception as e:
print(f"[通知失敗] {e}")
if __name__ == "__main__":
import sys
rep = HealthReport()
check_disk(rep)
check_memory(rep)
check_containers(rep)
check_clock(rep)
check_ledger(rep)
check_live_token(rep)
print(rep.report())
notify(rep)
sys.exit({"ok": 0, "warn": 1, "critical": 2}[rep.worst])# .github/workflows/deploy.yml
# テストを通したものだけVPSにデプロイする。
# 本ファイルは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
name: test-and-deploy
on:
push:
branches: [main]
workflow_dispatch:
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: "3.12"
cache: pip
- name: 依存をインストール
run: |
pip install -r requirements.txt
pip install pytest pytest-asyncio
# ネットワーク不要のテストだけをCIで走らせる（記事04のモック）
- name: 単体テスト
env:
BYBIT_ENV: testnet
# pytest実行中は本番接続が禁止される（記事04の多層ガード）
run: pytest tests/ -v -m "not integration"
- name: ルックアヘッドテスト
run: pytest tests/test_lookahead.py -v
- name: 設定ファイルの検証
run: python -c "from config_model import load_settings; load_settings('testnet')"
deploy:
needs: test
runs-on: ubuntu-latest
# mainへのpushかつテスト通過時のみ
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: SSHでデプロイ
env:
SSH_KEY: ${{ secrets.VPS_SSH_KEY }}
VPS_HOST: ${{ secrets.VPS_HOST }}
VPS_USER: ${{ secrets.VPS_USER }}
run: |
mkdir -p ~/.ssh
echo "$SSH_KEY" > ~/.ssh/id_ed25519
chmod 600 ~/.ssh/id_ed25519
ssh-keyscan -H "$VPS_HOST" >> ~/.ssh/known_hosts
ssh -i ~/.ssh/id_ed25519 "$VPS_USER@$VPS_HOST" bash -s <<'REMOTE'
set -euo pipefail
cd /opt/bybit-bot
echo "=== 現在の状態を記録（ロールバック用）==="
git rev-parse HEAD > /tmp/bybit-prev-sha
echo "=== 取得 ==="
git fetch --all
git reset --hard origin/main
echo "=== ビルド ==="
docker compose build
echo "=== 自己診断（記事06）==="
# 起動前に接続と権限を確認する
docker compose run --rm bot scripts/diagnose.py
echo "=== 再起動（graceful shutdownを待つ）==="
# systemd経由にすることで、停止処理の猶予が確保される
sudo systemctl restart bybit-bot
echo "=== 起動確認 ==="
sleep 20
docker compose ps
docker compose logs --tail 50 bot
# ヘルスチェックが通らなければロールバック
if ! .venv/bin/python healthcheck.py; then
echo "!! ヘルスチェック失敗。ロールバックします"
git reset --hard "$(cat /tmp/bybit-prev-sha)"
docker compose build
sudo systemctl restart bybit-bot
exit 1
fi
echo "=== デプロイ完了 ==="
REMOTE再起動すると記事17の起動時リコンサイルが実行され、未確定注文の照会と孤児注文の検出が行われます。これは正常な動作です。
逆に言えば、リコンサイルが実装されていない状態でデプロイを繰り返すのは危険です。停止中に約定した注文を認識できず、ポジションを二重に建てる可能性があります。CIでデプロイを自動化する前に、記事17の仕組みを入れてください。
| 項目 | 確認内容 | 参照 |
|---|---|---|
| APIキー | 出金権限オフ・VPSのIPを登録・用途別に分離 | 記事01 |
| 本番ガード | LIVE_ENABLED と BYBIT_ALLOW_LIVE=1 の両方 | 記事04 |
| 時刻同期 | chronyが稼働・ズレが1ms以内・定期監視 | 本記事 |
| 取引所側SL | botが落ちても効く損切りを設置 | 記事14 |
| リコンサイル | 起動時に注文とポジションを照合 | 記事17・20 |
| graceful shutdown | docker stop で注文がキャンセルされる | 記事30 |
| 永続化 | data/ をボリュームにマウント・バックアップ | 記事17・18 |
| ログ | ローテーション設定・キーがマスクされている | 記事06 |
| 監視 | ディスク・メモリ・コンテナ・時刻・台帳更新 | 本記事 |
| 通知 | critical時にDiscord/LINEへ飛ぶ | 記事39 |
| ファイアウォール | 着信はSSHのみ | 本記事 |
| レート分離 | 履歴取得を売買botと別プロセスに | 記事05 |
systemd のユニットファイルで ExecStart=/opt/bybit-bot/.venv/bin/python run_bot.py とすれば十分です。chronyc tracking の "System time" が1ms以内か。(3) コンテナ内の時刻——Dockerはホストの時刻を使うため通常は問題ありませんが、docker exec bybit-bot date で確認してください。orderLinkId のプレフィックスで注文を分離し（記事17）、レートゲートを共有する設計にしてください。0 4 * * * tar czf /backup/data-$(date +\%F).tgz -C /opt/bybit-bot data のようなcronで十分です。ただしバックアップにAPIキーを含めないよう注意してください（.env は別管理に）。投資は自己判断・自己責任でお願いいたします。本ブログの情報・ツール・ソフトウェアを参考・使用して、取引をされて何らかの損失を被られても、当サイトでは責任は一切負いません（間接的・付随的な障害・損失も含め一切の責任を負いません）。自己責任でご利用ください。
当サイトからダウンロードしたプログラム・EAは本人利用に限り完全無料でご利用いただけます。プログラムの全てまたは一部の複製・コピー・リバースエンジニアリング・改変・2次配布・商用利用・転売・販売等は一切禁止いたします。違反を発見した場合は法的処置を行います。
当サイト上に存在するソフトウェア・ツール・文章・画像・動画等の著作物を無断転載することを禁止します。引用の範囲を超えるものについては、法的処置を行います。
当サイトで配布するツール・ソフトウェアは無料提供ですのでサポートなどは行っておりません(逆に丁寧にサポートしているサイトはほとんど違法サイトですのでご注意ください。投資助言・代理業登録なしでサポート・助言することは法律で禁止されています)。万が一、ツールやソフトウェアに不具合や問題があっても当サイトでは修正・対応などができない場合があります。ご利用の環境でトラブルが発生しても当サイトでは助言することはできませんので自己責任でご利用ください。検索するなどして自分で解決してください。
当サイトでは取引所・証券会社へのリンクを貼っていますが、これは情報として提供するものであり、勧誘行為などは一切行っておりません。そのため、取引所・証券会社の登録方法・使い方・オススメ金融機関・オススメ金融商品などを説明した記事はサイト内にありませんので、ご自身で調べた上で判断ください(本当はサイト内で丁寧に説明したいのですが、仲介・取次ぎにあたる可能性があり第一種金融商品取引業登録を行っていない場合、違法となる可能性があるため出来ません。ご容赦ください)。
当サイトからリンクやバナーなどによって他のサイトに移動された場合、移動先サイトで提供される情報、サービス等について一切の責任を負いません。当サイト掲載情報やリンク先の掲載情報の合法性、正確性、道徳性、最新性、適切性、著作権の許諾や有無など、その内容については一切の保証を致しかねます。
当サイトは、ご訪問いただいた時間、訪問者のアクセス元のホスト名、IPアドレス、OSとブラウザのバージョン、リンク元のURL、モニターの情報等のパソコン環境情報を収集し、アクセス解析をしております。この目的は訪問者が検索サイトからご訪問いただいた場合、どのような検索で当サイトに来訪いただいたかの解析にのみ使用しており、第3者に公開はいたしません。また、個人を特定するものではありません。