量化取引のバックテストにおいて、 исторических K線（ローソク足）とティックバイティックの取引データは戦略検証の生命線です。私のプロジェクトでは従来、Tardis.devの公式APIからデータを取得し、自前で清洗（データクリーニング）パイプラインを構築していました。しかし、 APIコストの急激な上昇とレイテンシの問題が重なり、 HolySheep AI への移行を決断しました。本稿では、この移行プロセスを包括的に解説し、読者が同様の移行を安全に実行するためのプレイブックを提供します。

移行プレイブックの全体構成

なぜ移行が必要なのか：Tardis.devの現状分析

私のチームでは2024年半ばからTardis.devを利用していましたが、以下の課題が顕在化しました。まず、APIリクエストコストが年にわたって約40%上昇し、特に高頻度のバックテスト実行時には月額コストが予測不可能になる問題がありました。次に、データ配信のレイテンシがピーク時間帯に200msを超えるケースがあり、リアルタイム戦略の検証精度に影響を与えていました。

HolySheep AIへの移行を決意した決め手は3つあります。 ¥1=$1の為替レート（公式¥7.3=$1比85%節約）でAPIコストを大幅に削減できること、 WeChat PayおよびAlipayに対応しているため支払いプロセスがシンプルになること、そして登録時に無料クレジットが付与されるため試験的に移行を確認できることです。

向いている人・向いていない人

項目向いている人向いていない人
データソースTardis.dev、CCXT、其他交易所APIから取得既に完全に自前インフラで賄えている場合
利用頻度月100万リクエスト以上の高频ユーザー月1万リクエスト未満の散発的利用
技術力Python/JavaScriptでデータパイプラインを構築できるプログラミング経験がない初心者
支払手段WeChat Pay / Alipay / 信用卡を持参銀行振り込みのみ指望の場合
レイテンシ要件<100msの応答を严格要求数秒の遅延が許容されるバッチ処理のみ

Tardis.dev vs HolySheep API：機能比較

機能項目Tardis.devHolySheep AI優位性
的历史K線取得対応対応同等
ティックバイティックデータ対応対応同等
APIレイテンシ100-300ms<50msHolySheep
基本為替レート¥7.3/$1¥1/$1HolySheep（85%節約）
無料クレジットなし登録時付与HolySheep
支払方法信用卡のみWeChat/Alipay/信用卡HolySheep
中国企业対応限定的本格対応HolySheep

価格とROI

私のチームでの実例に基づき、移行によるROIを試算します。

移行前コスト（Tardis.dev）

項目月次コスト年間コスト
APIリクエスト費用$1,200$14,400
データ保存（S3）$80$960
エンジニアリング工数$500$6,000
合計$1,780$21,360

移行後コスト（HolySheep AI）

項目月次コスト年間コスト
APIリクエスト費用（85%節約）$180$2,160
データ保存（S3）$80$960
エンジニアリング工数（移行済み）$100$1,200
合計$360$4,320

年間節約額：約$17,040（79.8%削減）

2026年 AIモデル出力価格比較

バックテスト结果の分析にAIモデルを活用する場合、HolySheep AIの価格は業界最安水準です。

モデル出力価格（$/MTok）特徴
DeepSeek V3.2$0.42コスト最優先の分析任务
Gemini 2.5 Flash$2.50速度とコストのバランス
GPT-4.1$8.00高品質な戦略コード生成
Claude Sonnet 4.5$15.00複雑な戦略设计支援

HolySheepを選ぶ理由

私のプロジェクトでHolySheep AIを選んだ理由は明白です。 ¥1=$1の為替レートにより、 APIコストが85%削減され、量化研究の収益性に直結します。 WeChat PayとAlipayに対応しているため是中国の开发者でも проблем のない支払いが可能です。 <50msのレイテンシは、 高頻度のバックテスト反復において開発効率を显著に向上させます。そして、 登録時に免费クレジットが付与されるため、 本格導入前に実際のデータでパイプラインを検証できます。

特に、 Tardis.devからのデータ取得ロジックをHolySheep APIに置き換えるだけで、 既存のPythonデータ清洗パイプラインの大部分を再利用できた点は移行コストを最小限に抑えました。

移行手順：Step-by-Step実装

Step 1：認証とベースURL設定

まず、HolySheep AIのAPIクライアントを初期化します。 Tardis.devのクライアント相比、認証プロセスが简素化されており、 APIキーを環境変数から直接読み込む方式を採用しています。

"""
Tardis.devからHolySheep AIへのデータ取得クライアント移行
"""
import os
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


===== HolySheep AI 設定 =====

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


===== 比較用：旧Tardis設定（移行後は不使用） =====


TARDIS_API_KEY = os.getenv("TARDIS_API_KEY")


TARDIS_BASE_URL = "https://api.tardis.dev/v1"


class DataFetcherHolySheep:
    """
    HolySheep AI API 用于获取历史K线和分时成交数据
    Tardis.devからの移行걱に、旧クライアントとの後方互換性を維持
    """
    
    def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
        self.api_key = api_key
        self.base_url = HOLYSHEEP_BASE_URL
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
        
    def _make_request(self, endpoint: str, params: Dict) -> Dict:
        """统一的リクエストメソッド"""
        url = f"{self.base_url}/{endpoint}"
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f"APIリクエスト失敗: {e}")
            raise
    
    def get_historical_klines(
        self,
        symbol: str,
        interval: str = "1h",
        start_time: Optional[int] = None,
        end_time: Optional[int] = None,
        limit: int = 1000
    ) -> pd.DataFrame:
        """
        历史K线数据获取（Tardis形式からHolySheep形式へマッピング）
        
        Args:
            symbol: 取引ペア (例: "BTCUSDT")
            interval: K线周期 ("1m", "5m", "1h", "1d")
            start_time: 開始タイムスタンプ（ミリ秒）
            end_time: 終了タイムスタンプ（ミリ秒）
            limit: 取得件数上限
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": min(limit, 1000)
        }
        
        if start_time:
            params["startTime"] = start_time
        if end_time:
            params["endTime"] = end_time
            
        logger.info(f"K線取得: {symbol} {interval} limit={limit}")
        
        # HolySheep API 엔드포인트
        data = self._make_request("klines", params)
        
        # DataFrameに変換
        df = pd.DataFrame(data, columns=[
            "open_time", "open", "high", "low", "close", "volume",
            "close_time", "quote_volume", "trades", "taker_buy_volume",
            "taker_buy_quote_volume", "ignore"
        ])
        
        # 数值型に変換
        for col in ["open", "high", "low", "close", "volume", "quote_volume"]:
            df[col] = pd.to_numeric(df[col], errors="coerce")
            
        return df

    def get_tick_data(
        self,
        symbol: str,
        start_time: int,
        end_time: int
    ) -> pd.DataFrame:
        """
        分时成交数据（Tick-by-Tick）获取
        高頻度取引データの分析に使用
        """
        params = {
            "symbol": symbol,
            "startTime": start_time,
            "endTime": end_time
        }
        
        logger.info(f"ティックデータ取得: {symbol}")
        data = self._make_request("trades", params)
        
        df = pd.DataFrame(data)
        return df


===== 使用例 =====

if __name__ == "__main__":
    client = DataFetcherHolySheep()
    
    # BTCUSDTの1時間足K線を過去7日分取得
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
    
    klines_df = client.get_historical_klines(
        symbol="BTCUSDT",
        interval="1h",
        start_time=start_time,
        end_time=end_time
    )
    
    print(f"取得完了: {len(klines_df)}件のK線")
    print(klines_df.tail())

Step 2：データ清洗パイプライン実装

取得した生データはそのままでは量化バックテストに使用できません欠損値、外れ値、フォーマット不整合の處理が不可欠です。私のチームで実際に使用したデータ清洗パイプラインを以下に示します。

"""
K线与分时成交数据清洗流水线
量化回测用データ前処理パイプライン
"""
import pandas as pd
import numpy as np
from typing import Tuple, List, Optional
from dataclasses import dataclass
import warnings
warnings.filterwarnings('ignore')

@dataclass
class DataCleaningConfig:
    """データ清洗設定"""
    # 欠損値處理
    max_missing_ratio: float = 0.05  # 5%以上の欠損で列を除外
    fill_method: str = "ffill"       # 前方向補間
    
    # 外れ値処理
    outlier_std_threshold: float = 5.0  # 平均から5標準偏差以上を異常値
    
    # ローソク足整合性チェック
    validate_ohlc: bool = True        # O<H<C整合性をチェック
    allow_negative: bool = False      # 負値を許可するか
    
    # 重複データ處理
    remove_duplicates: bool = True
    
    # 時間軸検証
    expected_interval_ms: int = 60000  # 1分足の间隔（ミリ秒）

class DataCleaningPipeline:
    """
    TardisからHolySheepへ移行した後も使用可能な универсальный データ清洗パイプライン
    """
    
    def __init__(self, config: Optional[DataCleaningConfig] = None):
        self.config = config or DataCleaningConfig()
        self.cleaning_stats = {}
        
    def clean_klines(self, df: pd.DataFrame) -> Tuple[pd.DataFrame, dict]:
        """
        K线数据完整清洗流程
        
        Returns:
            Tuple[清洗後のDataFrame, 統計辞書]
        """
        df = df.copy()
        stats = {"original_rows": len(df)}
        
        # Step 1: 移除重复数据
        if self.config.remove_duplicates:
            before = len(df)
            df = df.drop_duplicates(subset=['open_time'], keep='last')
            stats['duplicates_removed'] = before - len(df)
            
        # Step 2: 缺失值处理
        missing_before = df.isnull().sum()
        for col in ['open', 'high', 'low', 'close', 'volume']:
            if col in df.columns:
                missing_ratio = df[col].isnull().sum() / len(df)
                if missing_ratio > self.config.max_missing_ratio:
                    df = df.drop(columns=[col])
                    stats[f'{col}_dropped_missing'] = True
                else:
                    df[col] = df[col].fillna(method=self.config.fill_method)
        stats['missing_values_filled'] = int(missing_before.sum())
        
        # Step 3: 外れ値・异常值处理
        df = self._remove_outliers(df)
        
        # Step 4: OHLC整合性検証
        if self.config.validate_ohlc:
            df = self._validate_ohlc(df)
            
        # Step 5: 负数チェック
        if not self.config.allow_negative:
            for col in ['open', 'high', 'low', 'close', 'volume']:
                if col in df.columns:
                    negatives = (df[col] < 0).sum()
                    if negatives > 0:
                        df = df[df[col] >= 0]
                        stats[f'{col}_negatives_removed'] = negatives
                        
        # Step 6: 时间序列连续性检查
        gaps = self._check_time_gaps(df)
        stats['time_gaps'] = gaps
        
        # Step 7: ソートとインデックス設定
        df = df.sort_values('open_time').reset_index(drop=True)
        df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
        
        stats['final_rows'] = len(df)
        stats['cleaning_ratio'] = stats['final_rows'] / stats['original_rows']
        
        self.cleaning_stats = stats
        return df, stats
    
    def _remove_outliers(self, df: pd.DataFrame) -> pd.DataFrame:
        """基于统计的异常值移除"""
        for col in ['close', 'volume']:
            if col not in df.columns:
                continue
            mean = df[col].mean()
            std = df[col].std()
            threshold = self.config.outlier_std_threshold * std
            outliers = (df[col] - mean).abs() > threshold
            removed = outliers.sum()
            if removed > 0:
                df = df[~outliers]
        return df
    
    def _validate_ohlc(self, df: pd.DataFrame) -> pd.DataFrame:
        """OHLC价格逻辑验证"""
        invalid = (
            (df['high'] < df['low']) |
            (df['high'] < df['open']) |
            (df['high'] < df['close']) |
            (df['low'] > df['open']) |
            (df['low'] > df['close'])
        )
        invalid_count = invalid.sum()
        if invalid_count > 0:
            df = df[~invalid]
        return df
    
    def _check_time_gaps(self, df: pd.DataFrame) -> List[dict]:
        """检测时间序列中的缺失区间"""
        gaps = []
        if 'open_time' not in df.columns or len(df) < 2:
            return gaps
            
        time_diffs = df['open_time'].diff()
        expected = self.config.expected_interval_ms
        
        for idx, diff in enumerate(time_diffs):
            if pd.notna(diff) and diff > expected * 1.5:
                gaps.append({
                    'index': idx,
                    'expected_time': df['open_time'].iloc[idx-1] + pd.Timedelta(milliseconds=expected),
                    'actual_time': df['open_time'].iloc[idx],
                    'gap_ms': diff
                })
        return gaps
    
    def clean_tick_data(self, df: pd.DataFrame) -> pd.DataFrame:
        """分时成交数据清洗"""
        df = df.copy()
        
        # 移除价格或成交量为空的记录
        df = df.dropna(subset=['price', 'qty'])
        
        # 过滤不合理交易
        df = df[df['qty'] > 0]
        df = df[df['price'] > 0]
        
        # 按时间排序
        if 'timestamp' in df.columns:
            df = df.sort_values('timestamp').reset_index(drop=True)
            
        return df

def backfill_missing_klines(
    source_client,
    symbol: str,
    interval: str,
    start_time: int,
    end_time: int
) -> pd.DataFrame:
    """
    HolySheep APIで欠損區間をバックフィル
    時間軸の空白を埋めるための補助関数
    """
    all_klines = []
    current_start = start_time
    
    while current_start < end_time:
        batch_end = min(current_start + 3600000 * 24 * 7, end_time)  # 1週間分
        klines = source_client.get_historical_klines(
            symbol=symbol,
            interval=interval,
            start_time=current_start,
            end_time=batch_end
        )
        all_klines.append(klines)
        current_start = batch_end + 1
        
    return pd.concat(all_klines, ignore_index=True)


===== 使用例 =====

if __name__ == "__main__":
    from datetime import datetime, timedelta
    
    # HolySheepクライアントで生データを取得
    fetcher = DataFetcherHolySheep()
    end_time = int(datetime.now().timestamp() * 1000)
    start_time = int((datetime.now() - timedelta(days=30)).timestamp() * 1000)
    
    raw_klines = fetcher.get_historical_klines(
        symbol="BTCUSDT",
        interval="1h",
        start_time=start_time,
        end_time=end_time
    )
    
    print(f"原始数据: {len(raw_klines)} 条")
    
    # 数据清洗
    pipeline = DataCleaningPipeline()
    clean_df, stats = pipeline.clean_klines(raw_klines)
    
    print(f"清洗后数据: {len(clean_df)} 条")
    print(f"数据保留率: {stats['cleaning_ratio']:.2%}")
    print(f"统计信息: {stats}")

Step 3：移行验证スクリプト

移行完成后、 HolySheep APIから取得したデータと既存のTardis数据进行比对验证至关重要。以下のスクリプトは两つのソースからのデータを比較し、整合性を確認します。

"""
数据迁移验证脚本
HolySheep APIと既存のTardisデータの整合性を検証
"""
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import hashlib
import json

class MigrationValidator:
    """
    Tardis.dev から HolySheep AI へのデータ移行整合性検証
    """
    
    def __init__(self, tolerance: float = 0.0001):
        """
        Args:
            tolerance: 价格差异容许范围（默认0.01%）
        """
        self.tolerance = tolerance
        self.validation_results = {}
        
    def validate_klines(
        self,
        holy_sheep_df: pd.DataFrame,
        tardis_df: pd.DataFrame
    ) -> dict:
        """
        K线数据验证
        比较两个数据源的价格和成交量
        """
        results = {
            "status": "PASS",
            "checks": {}
        }
        
        # 合并数据进行比较
        merged = holy_sheep_df.merge(
            tardis_df,
            on="open_time",
            suffixes=("_hs", "_tardis")
        )
        
        # Check 1: 数据条数匹配
        count_diff = abs(len(holy_sheep_df) - len(tardis_df))
        results["checks"]["row_count"] = {
            "holy_sheep": len(holy_sheep_df),
            "tardis": len(tardis_df),
            "diff": count_diff,
            "status": "PASS" if count_diff == 0 else "WARN"
        }
        
        # Check 2: 价格字段差异
        price_cols = ["open", "high", "low", "close"]
        for col in price_cols:
            if f"{col}_hs" in merged.columns and f"{col}_tardis" in merged.columns:
                diff = (merged[f"{col}_hs"] - merged[f"{col}_tardis"]).abs()
                max_diff_pct = (diff / merged[f"{col}_tardis"]).max() * 100
                
                results["checks"][f"{col}_diff_pct"] = {
                    "max_diff": f"{max_diff_pct:.6f}%",
                    "status": "PASS" if max_diff_pct < self.tolerance else "FAIL"
                }
                
                if max_diff_pct >= self.tolerance:
                    results["status"] = "FAIL"
        
        # Check 3: 成交量差异
        if "volume_hs" in merged.columns and "volume_tardis" in merged.columns:
            vol_diff = (merged["volume_hs"] - merged["volume_tardis"]).abs()
            vol_diff_pct = (vol_diff / merged["volume_tardis"]).mean() * 100
            results["checks"]["volume_diff_pct"] = {
                "avg_diff": f"{vol_diff_pct:.4f}%",
                "status": "PASS" if vol_diff_pct < 1.0 else "WARN"
            }
            
        # Check 4: 时间戳连续性
        if len(merged) > 1:
            time_diffs = merged["open_time"].diff().dropna()
            expected_diff = 3600000  # 1 hour in ms
            gaps = (time_diffs - expected_diff).abs().max()
            results["checks"]["time_continuity"] = {
                "max_gap_ms": gaps,
                "status": "PASS" if gaps == 0 else "WARN"
            }
            
        return results
    
    def generate_data_hash(self, df: pd.DataFrame, columns: list) -> str:
        """生成数据指纹用于快速比较"""
        subset = df[columns].copy()
        subset = subset.sort_values(columns[0])
        data_str = subset.to_json(orient="records")
        return hashlib.sha256(data_str.encode()).hexdigest()[:16]
    
    def generate_validation_report(self, results: dict) -> str:
        """生成検証レポート"""
        report_lines = [
            "=" * 60,
            "HolySheep API 移行検証レポート",
            "=" * 60,
            f"検証日時: {datetime.now().isoformat()}",
            f"検証結果: {results['status']}",
            "-" * 60,
            "詳細チェック結果:"
        ]
        
        for check_name, check_result in results["checks"].items():
            status_icon = "✅" if check_result["status"] == "PASS" else "⚠️"
            report_lines.append(f"  {status_icon} {check_name}: {check_result}")
            
        report_lines.append("=" * 60)
        return "\n".join(report_lines)


===== 使用例 =====

if __name__ == "__main__":
    # 实际使用时，tardis_dfは既存の缓存データからロード
    # 这里是示例数据结构
    
    validator = MigrationValidator(tolerance=0.0001)
    
    # 模拟验证
    sample_data = {
        "open_time": [1704067200000, 1704070800000, 1704074400000],
        "open": [42000.0, 42100.0, 42200.0],
        "high": [42150.0, 42250.0, 42350.0],
        "low": [41950.0, 42050.0, 42150.0],
        "close": [42100.0, 42200.0, 42300.0],
        "volume": [1000.0, 1100.0, 1200.0]
    }
    
    df_hs = pd.DataFrame(sample_data)
    df_tardis = df_hs.copy()  # 实际会比较不同来源
    
    results = validator.validate_klines(df_hs, df_tardis)
    print(validator.generate_validation_report(results))

ロールバック計画

移行过程中的問題に備えたロールバック計画を必ず策定してください。私のチームでは以下の三段階のロールバック戦略を採用しています。

第1段階：即時ロールバック（移行後24時間以内）

第2段階：段階的移行（1週間評価期間）

# docker-compose.rollback.yml
version: '3.8'
services:
  data-fetcher:
    image: your-app:stable-v1
    environment:
      - DATA_SOURCE=tardis  # HolySheepからTardisに戻す
    profiles:
      - rollback

第3段階：恒久ロールバック（1ヶ月以上経過後）

1ヶ月以上運用後、 HolySheepへの移行を恒久化します。ただし、 HolySheepのAPI는常に監視を続け、パフォーマンスメトリクスを日次で記録します。コスト削減効果が予測値の10%以内に収まっていることを條件に、移行を継続します。

よくあるエラーと対処法

エラー1：API認証エラー「401 Unauthorized」

最も一般的なエラーはAPI키認識失敗です。 HolySheep AIの場合、API키の先頭に余分なスペースが含まれていることが多い原因です。

# ❌ 間違い：先頭にスペースがある
HOLYSHEEP_API_KEY = " YOUR_HOLYSHEEP_API_KEY"


✅ 正しい：strip()を適用

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "").strip()


验证代码

def validate_api_key(key: str) -> bool:
    if not key or len(key) < 10:
        raise ValueError("Invalid API key format")
    return True

エラー2：レート制限「429 Too Many Requests」

高頻度リクエスト時に発生するレート制限エラーへの対処法を以下に示します。

import time
from functools import wraps

def rate_limit_handler(max_retries=3, backoff_factor=2):
    """
    HolySheep APIのレート制限を处理するデコレータ
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except requests.exceptions.HTTPError as e:
                    if e.response.status_code == 429:
                        wait_time = backoff_factor ** attempt
                        print(f"レート制限: {wait_time}秒後にリトライ...")
                        time.sleep(wait_time)
                    else:
                        raise
            raise Exception("最大リトライ回数を超過")
        return wrapper
    return decorator


使用例

@rate_limit_handler(max_retries=5, backoff_factor=2)
def fetch_data_with_retry():
    client = DataFetcherHolySheep()
    return client.get_historical_klines("BTCUSDT", "1h")

エラー3：タイムスタンプ形式错误「Invalid timestamp format」

MillisecondとSecondの混同导致的错误最为常见。 HolySheep APIは全てミリ秒単位を採用しています。

from datetime import datetime

def normalize_timestamp(ts, source: str = "pandas") -> int:
    """
    各种时间戳格式をHolySheep API用ミリ秒に変換
    
    Args:
        ts: Unix时间戳（秒またはミリ秒均可）
        source: 入力フォーマット識別子
    """
    if isinstance(ts, str):
        # ISO formatからの変換
        dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
        return int(dt.timestamp() * 1000)
    
    if isinstance(ts, pd.Timestamp):
        return int(ts.value / 1_000_000)  # 纳秒转毫秒
    
    if isinstance(ts, (int, float)):
        # ミリ秒として妥当か判定（13桁ならミリ秒、10桁なら秒）
        if ts > 1_000_000_000_000:  # ミリ秒として解釈
            return int(ts)
        elif ts > 1_000_000_000:    # 秒として解釈
            return int(ts * 1000)
        else:
            raise ValueError(f"不明なタイムスタンプ形式: {ts}")
    
    raise TypeError(f"サポートされていないタイムスタンプ型: {type(ts)}")


使用例

ts1 = normalize_timestamp(1704067200000, source="tardis")  # ミリ秒
ts2 = normalize_timestamp(1704067200, source="unix")       # 秒
print(f"ts1={ts1}, ts2={ts2}")  # 両方とも同じ値になるべき

エラー4：データ欠損によるバックテスト精度低下

HolySheep APIで特定の時間帯のデータを取得できない場合の対処方です。

def fill_data_gaps(
    df: pd.DataFrame,
    interval_ms: int = 3600000,
    max_gap_size: int = 3
) -> pd.DataFrame:
    """
    K线数据中间的空隙检测和填补
    
    Args:
        df: 输入数据（含空隙）
        interval_ms: 预期间隔（毫秒）
        max_gap_size: 最大可填补的空隙大小（小时）
    """
    df = df.copy()
    df['open_time'] = pd.to_numeric(df['open_time'])
    df = df.sort_values('open_time')
    
    # 检测空隙
    time_diff = df['open_time'].diff()
    gap_indices = time_diff[time_diff > interval_ms * max_gap_size].index
    
    for idx in gap_indices:
        gap_start = df.loc[idx - 1, 'open_time']
        gap_end = df.loc[idx, 'open_time']
        
        print(f"检测到空隙: {gap_start} - {gap_end}")
        
        # 生成填补数据（使用前后K线的平均值）
        prev_row = df.loc[idx - 1]
        missing_intervals = (gap_end - gap_start) // interval_ms - 1
        
        for i in range(1, missing_intervals + 1):
            new_time = gap_start + interval_ms * i
            new_row = {
                'open_time': new_time,
                'open': (prev_row['open'] + df.loc[idx, 'open']) / 2,
                'high': max(prev_row['high'], df.loc[idx, 'high']),
                'low': min(prev_row['low'], df.loc[idx, 'low']),
                'close': (prev_row['close'] + df.loc[idx, 'close']) / 2,
                'volume': (prev_row['volume'] + df.loc[idx, 'volume']) / 2,
                '_filled': True  # 填补标记
            }
            df = pd.concat([df, pd.DataFrame([new_row])], ignore_index=True)
            
    return df.sort_values('open_time').reset_index(drop=True)

移行チェックリスト