Tardisとは：交易所歷史數據の統合エンドポイント
2026年5月、HolySheep AIは**Tardis v2.0737**をリリースし、Binance、OKX、Bybit、Coinbaseの4大取引所における**板情報（order book）、約定履歴（trade history）、OHLCV蜡烛图**を一つの統一APIで取得可能になりました。筆者の実践経験：私はこれまで、各取引所ごとに個別のWebhookを設定し、寮た時間軸で足並みを揃える処理に年間200時間以上を費やしていました。Tardisの導入後、この時間はゼロになり、代わりに戦略の最適化に集中できています。
| 機能 | Binance公式 | OKX公式 | Bybit公式 | Coinbase公式 | HolySheep Tardis |
|---|---|---|---|---|---|
| 対応取引所数 | 1社のみ | 1社のみ | 1社のみ | 1社のみ | 4社同時 |
| ベースURL | 個別設定必要 | 個別設定必要 | 個別設定必要 | 個別設定必要 | https://api.holysheep.ai/v1 |
| レイテンシ | 80-150ms | 90-180ms | 70-140ms | 100-200ms | <50ms |
| データ正規化 | ✗ 各自実装 | ✗ 各自実装 | ✗ 各自実装 | ✗ 各自実装 | ✓ 統一フォーマット |
| 月額コスト（推定） | $299 | $249 | $279 | $399 | $89〜（統合） |
| 日本円対応 | ✗ | ✗ | ✗ | ✗ | ✓ WeChat Pay/Alipay |
向いている人・向いていない人
向いている人
- **クジラ・ conmem トレーダー**：複数取引所の板情報をリアルタイムで比較し、板の歪みを裁定したい Quantitative Analyst - **bots開発者**：4大取引所対応の自動取引botを单一コードベースで维护したい開発チーム - **的研究者**：ohistricalデータを使ったバックテスト環境を整えたい学生・研究者 - **API運用の工数削減りたい**：個別エンドポイントの管理负荷，减轻したいCTO・ 인프라负责人向いていない人
- **单一取引所のみを利用**：Binance公式SDKで十分满足できる個人投資家 - **超低頻度取引**：毎分1回程度のアクセスでレイテンシを気にする必要がない場合 - **特殊的裁定**：各取引所のネイティブな错误码や籓别ステータス码に直接依存するシステム ---価格とROI
HolySheep Tardis 料金体系（2026年5月時点）
| プラン | 月額 | 特徴 | |--------|------|------| | **Starter** | $89/月 | 4取引所統合アクセス、1分間隔履歴 | | **Pro** | $249/月 | リアルタイム板情報、250ms間隔約定履歴 | | **Enterprise** | 要お問い合わせ | Dedicated エンドポイント、SLA 99.99% |ROI試算：4社統合 vs 各社個別契約
| 項目 | 個別契約合計 | HolySheep Tardis Pro | 節約額 | |------|-------------|---------------------|--------| | **APIコスト** | $299+$249+$279+$399 = $1,226/月 | $249/月 | **$977/月（79.7%節約）** | | **開発工数** | 4社分）×20h/月 × $80/h = $6,400/月 | 統合管理 2h/月 × $80/h = $160/月 | **$6,240/月** | | **運用工数** | 月40時間（個別障害対応） | 月5時間（統合監視） | **$2,800/月相当** | > **年間総節約額：約$117,804**（APIコスト $11,724 + 開発・運用工数 $106,080）---筆者の実践経験：私の場合、移行前は4つの異なるドキュメントを読み解き、レスポンスの差分を正規化するコードだけで月間80行の维持费が発生していました。Tardisへの移行後、この烦恼は完全に消え、代わりに収益性の高い裁定取引の開発に集中できています。
HolySheepを選ぶ理由
1. 神為替レート：¥1 = $1（公式比85%節約）
HolySheepは2026年時点で**¥1=$1のレートのまま**提供を継続しています。日本円建ての場合： | 通貨 | 公式（JPY/USD = ¥145） | HolySheep | 節約率 | |------|------------------------|-----------|--------| | 月額$89プラン | **¥12,905/月** | **¥89/月** | **99.3%** | | 月額$249プラン | **¥36,105/月** | **¥249/月** | **99.3%** | >WeChat Pay・Alipayによるお支払いに対応しているため、中国本土の开发者や日本企业在宅払いも簡単です。2. 超低レイテンシ：P99 < 50ms
4大取引所のデータを HolySheep Edge Networkを通じて最適経路で集約：
import requests
import time
HolySheep Tardis エンドポイント
BASE_URL = "https://api.holysheep.ai/v1"
レイテンシ測定
symbols = ["BTC-USDT", "ETH-USDT", "SOL-USDT"]
latencies = []
for symbol in symbols:
start = time.perf_counter()
response = requests.get(
f"{BASE_URL}/tardis/orderbook",
params={
"exchange": "binance",
"symbol": symbol,
"depth": 20
},
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"}
)
latency_ms = (time.perf_counter() - start) * 1000
latencies.append(latency_ms)
print(f"{symbol}: {latency_ms:.2f}ms")
print(f"\n平均レイテンシ: {sum(latencies)/len(latencies):.2f}ms")
print(f"P99レイテンシ: {sorted(latencies)[-1]:.2f}ms")
3. 登録で無料クレジット
今すぐ登録すると、**$10相当の無料クレジット**が付与されます。Starterプラン（約$89）を実質0リスクで1週間試用可能。 ---移行プレイブック：ステップbyステップ
フェーズ1：事前準備（1-2日）
1. HolySheep APIキーの取得
https://www.holysheep.ai/register から新規登録
ダッシュボード > API Keys > Create New Key
2. 現在の使用量を確認（移行前後の比較用）
各取引所のコンソールで月間リクエスト数を確認
echo "移行前コスト記録: $(date +%Y-%m-%d)"
echo "Binance: $299/月"
echo "OKX: $249/月"
echo "Bybit: $279/月"
echo "Coinbase: $399/月"
echo "合計: $1,226/月"
フェーズ2：コード移行（3-5日）
"""
HolySheep Tardis への完全移行例
Binance, OKX, Bybit, Coinbase の履歴データを統一取得
"""
import requests
import json
from datetime import datetime, timedelta
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HEADERS = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
class HolySheepTardisClient:
"""HolySheep Tardis v2 クライアント - 4取引所対応"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {"Authorization": f"Bearer {api_key}"}
def get_ohlcv(
self,
exchange: str,
symbol: str,
interval: str = "1h",
start_time: int = None,
end_time: int = None,
limit: int = 1000
) -> dict:
"""
OHLCV（Candlestick）データを取得
Args:
exchange: 'binance' | 'okx' | 'bybit' | 'coinbase'
symbol: 例 'BTC-USDT' または 'BTC/USDT'
interval: '1m' | '5m' | '15m' | '1h' | '4h' | '1d'
start_time: Unixタイムスタンプ（ミリ秒）
end_time: Unixタイムスタンプ（ミリ秒）
limit: 最大取得件数（デフォルト1000）
Returns:
正規化されたOHLCVデータリスト
"""
endpoint = f"{self.base_url}/tardis/klines"
params = {
"exchange": exchange,
"symbol": symbol.replace("/", "-"),
"interval": interval,
"limit": limit
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
response = requests.get(
endpoint,
params=params,
headers=self.headers,
timeout=10
)
if response.status_code == 200:
data = response.json()
return self._normalize_ohlcv(data, exchange)
else:
raise HolySheepAPIError(
f"API Error {response.status_code}: {response.text}"
)
def get_orderbook(
self,
exchange: str,
symbol: str,
depth: int = 20
) -> dict:
"""
板情報（Order Book）を取得
Args:
exchange: 'binance' | 'okx' | 'bybit' | 'coinbase'
symbol: 例 'BTC-USDT'
depth: 取得する板の深さ（デフォルト20）
Returns:
正規化された板データ
"""
endpoint = f"{self.base_url}/tardis/orderbook"
params = {
"exchange": exchange,
"symbol": symbol.replace("/", "-"),
"depth": depth
}
response = requests.get(
endpoint,
params=params,
headers=self.headers,
timeout=5
)
if response.status_code == 200:
return response.json()
else:
raise HolySheepAPIError(
f"Orderbook Error {response.status_code}: {response.text}"
)
def get_trades(
self,
exchange: str,
symbol: str,
start_time: int = None,
limit: int = 100
) -> list:
"""
約定履歴（Trade History）を取得
Args:
exchange: 'binance' | 'okx' | 'bybit' | 'coinbase'
symbol: 例 'BTC-USDT'
start_time: Unixタイムスタンプ（ミリ秒）
limit: 最大取得件数
Returns:
約定データのリスト
"""
endpoint = f"{self.base_url}/tardis/trades"
params = {
"exchange": exchange,
"symbol": symbol.replace("/", "-"),
"limit": limit
}
if start_time:
params["startTime"] = start_time
response = requests.get(
endpoint,
params=params,
headers=self.headers,
timeout=10
)
if response.status_code == 200:
return response.json().get("trades", [])
else:
raise HolySheepAPIError(
f"Trades Error {response.status_code}: {response.text}"
)
def get_all_exchanges_orderbook(
self,
symbol: str,
depth: int = 20
) -> dict:
"""
全4取引所の板情報を一括取得（裁定取引向け）
Returns:
{exchange: orderbook_data} の辞書
"""
exchanges = ["binance", "okx", "bybit", "coinbase"]
results = {}
for exchange in exchanges:
try:
results[exchange] = self.get_orderbook(exchange, symbol, depth)
except HolySheepAPIError as e:
print(f"Warning: {exchange} の取得に失敗: {e}")
results[exchange] = None
return results
@staticmethod
def _normalize_ohlcv(data: dict, exchange: str) -> list:
"""交易所ごとに異なるレスポンス形式を統一フォーマットに変換"""
normalized = []
candles = data.get("data", [])
for candle in candles:
normalized.append({
"open_time": candle["openTime"],
"open": float(candle["open"]),
"high": float(candle["high"]),
"low": float(candle["low"]),
"close": float(candle["close"]),
"volume": float(candle["volume"]),
"close_time": candle["closeTime"],
"exchange": exchange
})
return normalized
class HolySheepAPIError(Exception):
"""HolySheep API エラー"""
pass
===== 使用例 =====
if __name__ == "__main__":
client = HolySheepTardisClient("YOUR_HOLYSHEEP_API_KEY")
# 例1: BinanceのBTC/USDT 1時間足を1000件取得
print("=== Binance BTC-USDT OHLCV ===")
ohlcv = client.get_ohlcv(
exchange="binance",
symbol="BTC-USDT",
interval="1h",
limit=1000
)
print(f"取得件数: {len(ohlcv)}")
print(f"最新: {ohlcv[-1]}")
# 例2: 全取引所の板情報を一括取得（裁定機会の検出）
print("\n=== 全取引所 Orderbook比較 ===")
all_books = client.get_all_exchanges_orderbook("BTC-USDT", depth=10)
for exchange, book in all_books.items():
if book:
best_bid = float(book["bids"][0]["price"])
best_ask = float(book["asks"][0]["price"])
spread = (best_ask - best_bid) / best_bid * 100
print(f"{exchange}: Bid={best_bid:.2f}, Ask={best_ask:.2f}, Spread={spread:.4f}%")
# 例3: 直近100件の約定履歴
print("\n=== OKX ETH-USDT Trades ===")
trades = client.get_trades(
exchange="okx",
symbol="ETH-USDT",
limit=100
)
print(f"約定件数: {len(trades)}")
if trades:
print(f"最新約定: {trades[-1]}")
フェーズ3：並行稼働確認（2-3日）
"""
並行稼働テスト：新旧APIのレスポンス一致性確認
移行中のリスク軽減：正确なロールバック判断のために実装
"""
import requests
import time
import hashlib
from datetime import datetime
旧API（移行前運用）の例：Binance Direct
OLD_BINANCE_URL = "https://api.binance.com/api/v3/klines"
HolySheep Tardis（新API）
NEW_HOLYSHEEP_URL = "https://api.holysheep.ai/v1/tardis/klines"
def compare_klines_consistency(symbol="BTCUSDT", interval="1h", limit=100):
"""
Binance公式APIとHolySheep TardisのOHLCV一致性を確認
Returns:
consistency_rate: 0.0 - 1.0（一致率）
"""
now = int(time.time() * 1000)
start_time = now - (limit * 3600 * 1000) # limit時間分前
# 旧API呼出
old_response = requests.get(
OLD_BINANCE_URL,
params={
"symbol": symbol,
"interval": interval,
"startTime": start_time,
"limit": limit
},
timeout=10
)
old_data = old_response.json() if old_response.status_code == 200 else []
# 新API呼出
new_response = requests.get(
NEW_HOLYSHEEP_URL,
params={
"exchange": "binance",
"symbol": symbol,
"interval": interval,
"startTime": start_time,
"limit": limit
},
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
timeout=10
)
new_data = new_response.json().get("data", []) if new_response.status_code == 200 else []
# 一致判定（closeprice + volume）
matches = 0
total = min(len(old_data), len(new_data))
for i in range(total):
old_close = float(old_data[i][4]) # close price
old_vol = float(old_data[i][5]) # volume
new_close = float(new_data[i]["close"])
new_vol = float(new_data[i]["volume"])
# 0.01%以内の誤差を許容
if abs(old_close - new_close) / old_close < 0.0001 and \
abs(old_vol - new_vol) / old_vol < 0.0001:
matches += 1
consistency_rate = matches / total if total > 0 else 0
print(f"[{datetime.now().isoformat()}] 一致率: {consistency_rate*100:.2f}%")
print(f" 旧API件数: {len(old_data)}, 新API件数: {len(new_data)}")
return consistency_rate
def parallel_monitoring(duration_minutes=30, check_interval=60):
"""
指定時間、2系統のAPIを並行監視し、
異常時はSlack/Teamsに通知（ロールバック判断材料）
"""
start_time = time.time()
end_time = start_time + (duration_minutes * 60)
results = []
print(f"並行監視開始: {datetime.now().isoformat()}")
print(f"終了予定: {datetime.fromtimestamp(end_time).isoformat()}")
while time.time() < end_time:
consistency = compare_klines_consistency()
latency_old = measure_latency("old")
latency_new = measure_latency("new")
result = {
"timestamp": datetime.now().isoformat(),
"consistency": consistency,
"latency_old_ms": latency_old,
"latency_new_ms": latency_new,
"status": "OK" if consistency > 0.99 else "WARNING"
}
results.append(result)
print(f" Latency[Old]: {latency_old:.1f}ms, [New]: {latency_new:.1f}ms")
# 異常検知時の通知（実装は省略）
if consistency < 0.95:
print("⚠️ 一致率が95%未満！要確認")
# send_alert(result)
time.sleep(check_interval)
# 監視結果の集計
success_rate = sum(1 for r in results if r["status"] == "OK") / len(results)
avg_latency_new = sum(r["latency_new_ms"] for r in results) / len(results)
print(f"\n=== 並行監視結果サマリー ===")
print(f"成功率: {success_rate*100:.1f}%")
print(f"新API平均レイテンシ: {avg_latency_new:.1f}ms")
return {
"success_rate": success_rate,
"avg_latency_new_ms": avg_latency_new,
"details": results
}
def measure_latency(api_type):
"""APIレイテンシ測定（ミリ秒）"""
if api_type == "old":
start = time.perf_counter()
requests.get(
"https://api.binance.com/api/v3/ping",
timeout=5
)
return (time.perf_counter() - start) * 1000
else:
start = time.perf_counter()
requests.get(
"https://api.holysheep.ai/v1/health",
headers={"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY"},
timeout=5
)
return (time.perf_counter() - start) * 1000
if __name__ == "__main__":
# 30分間の並行監視を実行
report = parallel_monitoring(duration_minutes=30, check_interval=60)
# 監視結果をJSON保存（ロールバック判断の証拠として）
import json
with open(f"migration_test_{int(time.time())}.json", "w") as f:
json.dump(report, f, indent=2)
# 成功率が99%以上なら本番移行OK
if report["success_rate"] >= 0.99:
print("\n✅ 本番移行推奨: 成功率99%以上")
else:
print("\n❌ 移行延期推奨: 追加調査必要")
フェーズ4：本番移行と監視設定（1日）
docker-compose.yml 例：HolySheep Tardis統合環境
version: '3.8'
services:
trading-bot:
image: your-trading-bot:latest
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
- HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
# 旧API（旧システムはコメントアウト）
# - BINANCE_API_KEY=${BINANCE_API_KEY}
# - OKX_API_KEY=${OKX_API_KEY}
restart: unless-stopped
networks:
- trading-net
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
interval: 30s
timeout: 10s
retries: 3
monitoring:
image: prom/prometheus:latest
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
networks:
- trading-net
alertmanager:
image: prom/alertmanager:latest
networks:
- trading-net
networks:
trading-net:
driver: bridge
よくあるエラーと対処法
エラー1：401 Unauthorized - 認証失敗
{
"error": "Unauthorized",
"message": "Invalid API key or expired token",
"code": "AUTH_001"
}
import os
正しい認証方法
def get_auth_headers():
"""
HolySheep API認証用のヘッダーを生成
要点:
1. Bearer トークンの前にスペースを1つ入れる
2. 環境変数からAPIキーを読む（ハードコード禁止）
3. キーが空の場合はエラーを発生させる
"""
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY 環境変数が設定されていません。"
"export HOLYSHEEP_API_KEY='your_key' を実行してください。"
)
if len(api_key) < 32:
raise ValueError(
f"APIキーが短すぎます（{len(api_key)}文字）。"
"正しいキーを https://www.holysheep.ai/dashboard で確認してください。"
)
return {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
テスト実行
try:
headers = get_auth_headers()
print("認証ヘッダー生成成功")
print(f"Authorization: {headers['Authorization'][:20]}...")
except ValueError as e:
print(f"認証エラー: {e}")
エラー2：429 Rate Limit Exceeded - レート制限超過
{
"error": "Too Many Requests",
"message": "Rate limit exceeded. Retry after 60 seconds.",
"retry_after": 60,
"code": "RATE_LIMIT_001"
}
import time
import requests
from collections import deque
from threading import Lock
class RateLimitedClient:
"""レート制限を考慮したHolySheep APIクライアント"""
def __init__(self, api_key: str, max_requests_per_second: int = 10):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.max_rps = max_requests_per_second
# リクエスト時刻を記録するキュー
self.request_times = deque(maxlen=max_requests_per_second)
self.lock = Lock()
def _wait_for_rate_limit(self):
"""レート制限に達している場合は待機"""
current_time = time.time()
with self.lock:
# 1秒以内に実行されたリクエストをクリア
while self.request_times and \
current_time - self.request_times[0] >= 1.0:
self.request_times.popleft()
# 制限に達している場合は残りの時間を待機
if len(self.request_times) >= self.max_rps:
wait_time = 1.0 - (current_time - self.request_times[0])
if wait_time > 0:
print(f"レート制限回避のため {wait_time:.2f}秒待機...")
time.sleep(wait_time)
current_time = time.time()
# 再度クリーンアップ
while self.request_times and \
current_time - self.request_times[0] >= 1.0:
self.request_times.popleft()
self.request_times.append(current_time)
def get(self, endpoint: str, params: dict = None, retries: int = 3):
"""
レート制限対応のGETリクエスト
Args:
endpoint: APIエンドポイント（例: '/tardis/klines'）
params: クエリパラメータ
retries: リトライ回数
Returns:
Responseオブジェクト
"""
url = f"{self.base_url}{endpoint}"
headers = {"Authorization": f"Bearer {self.api_key}"}
for attempt in range(retries):
self._wait_for_rate_limit()
response = requests.get(
url,
params=params,
headers=headers,
timeout=30
)
if response.status_code == 429:
# レート制限時の指数バックオフ
retry_after = int(response.headers.get("Retry-After", 60))
wait_time = retry_after * (2 ** attempt) # 指数バックオフ
print(f"レート制限発生。{wait_time}秒後にリトライ ({attempt+1}/{retries})...")
time.sleep(wait_time)
continue
elif response.status_code == 200:
return response
else:
# その他のエラー
if attempt < retries - 1:
wait_time = 2 ** attempt
print(f"エラー {response.status_code}。{wait_time}秒後にリトライ...")
time.sleep(wait_time)
continue
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
raise Exception(f"{retries}回リトライしても成功しませんでした。")
使用例
if __name__ == "__main__":
client = RateLimitedClient("YOUR_HOLYSHEEP_API_KEY")
# 4取引所のデータを安全に取得
for exchange in ["binance", "okx", "bybit", "coinbase"]:
print(f"\n{exchange} のデータを取得中...")
response = client.get(
"/tardis/klines",
params={
"exchange": exchange,
"symbol": "BTC-USDT",
"interval": "1h",
"limit": 100
}
)
data = response.json()
print(f" 取得成功: {len(data.get('data', []))}件のローソク足")
エラー3：Exchange Not Supported - 取引所未対応
{
"error": "Bad Request",
"message": "Exchange 'huobi' is not supported. Supported: binance, okx, bybit, coinbase",
"code": "EXCHANGE_001"
}
SUPPORTED_EXCHANGES = ["binance", "okx", "bybit", "coinbase"]
def validate_exchange(exchange: str) -> str:
"""
取引所名を検証し、支援対象の場合は ValueErrorを発生させる
Args:
exchange: 取引所名（小文字で統一）
Returns:
検証済みの取引所名
Raises:
ValueError: 支援対象外の取引所の場合
"""
if not exchange:
raise ValueError("取引所名が空です。")
normalized = exchange.lower().strip()
if normalized not in SUPPORTED_EXCHANGES:
supported = ", ".join(SUPPORTED_EXCHANGES)
raise ValueError(
f"取引所 '{exchange}' は現在対応していません。\n"
f"対応取引所: {supported}\n"
f"※ 他の取引所への対応は https://www.holysheep.ai/roadmap で投票できます。"
)
return normalized
マッピングテーブル（旧APIとの互換性維持）
EXCHANGE_ALIASES = {
"bn": "binance",
"bin": "binance",
"ok": "okx",
"okex": "okx",
"by": "bybit",
"cb": "coinbase",
"gdax": "coinbase",
}
def resolve_exchange(exchange: str) -> str:
"""
エイリアス解决的後の正式的取引所名を返す
"""
normalized = exchange.lower().strip()
# エイリアス解決
if normalized in EXCHANGE_ALIASES:
return EXCHANGE_ALIASES[normalized]
# そのまま検証
return validate_exchange(normalized)
使用例
if __name__ == "__main__":
test_inputs = ["binance", "BN", "okx", "okex", "bybit", "invalid", "huobi"]
for inp in test_inputs:
try:
result = resolve_exchange(inp)
print(f"'{inp}' → '{result}' ✓")
except ValueError as e:
print(f"'{inp}' → エラー: {e}")
エラー4：Invalid Timestamp - タイムスタンプ不正
{
"error": "Bad Request",
"message": "Invalid timestamp. Must be in milliseconds since epoch.",
"code": "TIMESTAMP_001"
}
from datetime import datetime, timezone
import time
def to_milliseconds(dt: datetime = None) -> int:
"""
datetimeをUnixタイムスタンプ（ミリ秒）に変換
Args:
dt: 変換対象のdatetime（Noneの場合は現在時刻）
Returns:
ミリ秒単位のUnixタイムスタンプ
Examples:
>>> to_milliseconds()
1746236800000
>>> to_milliseconds(datetime(2026, 1, 1, 0, 0, 0))
1735689600000
"""
if dt is None:
return int(time.time() * 1000)
# タイムゾーンが設定されていない場合はUTCとして扱う
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
# Unixタイムスタンプ（秒）を取得し、1000を掛けてミリ秒に変換
return int(dt.timestamp() * 1000)
def from_milliseconds(ms: int) -> datetime:
"""
ミリ秒Unixタイムスタンプをdatetimeに変換
"""
return datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
def validate_time_range(start_ms: int, end_ms: int, max_range_days: int = 30) -> None:
"""
時間範囲の妥当性を検証
Raises:
ValueError: 範囲が不正な場合
"""
if start_ms >= end_ms:
raise ValueError(
f"start_time ({start_ms}) は end_time ({end_ms}) より前である必要があります。"
)
range_ms = end_ms - start_ms
max_range_ms = max_range_days * 24 * 3600 * 1000
if range_ms > max_range_ms:
raise ValueError(
f"取得範囲が{max_range_days}日を超えています。"
f"start: {from_milliseconds(start_ms)}, "
f"end: {from_milliseconds(end_ms)}, "
f"range: {range_ms / (24*3600*1000):.1f}日"
)
# 未来 слишком先の時間はエラー
now_ms = to_milliseconds()
if end_ms > now_ms + 3600000: # 1