過去の金融データをダウンロードした後、欠落した取引日と異常なローソク足をどう確認する？

過去の金融データをダウンロードした後、ファイルを正常に開けるからといって、データが分析やバックテストに適しているとは限りません。一部の取引日のデータが欠落していたり、同じローソク足が重複してダウンロードされていたり、最高値が始値や終値を下回っていたりする場合があります。こうした問題を事前に発見できなければ、テクニカル指標の計算、売買戦略のシグナル、バックテストの結果に影響する可能性があります。

Historical Market Data API を利用する開発者にとって、データ品質の確認は単にレコード数を数えるだけでは不十分です。ダウンロードしたデータを、想定される取引日、タイムスタンプ、OHLC の価格ルールと照合し、異常なレコードを後から確認できるように保存しておく必要があります。

まず、どの取引日にデータが存在するべきかを確認する

欠落した取引日を調べる際、月曜日から金曜日まで毎日データが存在するはずだと単純に判断してはいけません。株式市場では、祝日、売買停止、臨時休場などによって通常の取引が行われないことがあります。また、市場ごとに取引日カレンダーも異なります。外国為替や暗号資産などの金融商品についても、それぞれの取引時間に合ったルールを使用する必要があります。

そのため、最初に対象市場と金融商品に対応する取引日リストを用意します。日足ローソク足の場合は、実際のデータに含まれる取引日と予定された取引日を比較することで、1日分のデータが欠落しているかを確認できます。1分足や5分足などの日中データでは、取引時間に基づいて想定されるタイムスタンプを生成しなければ、同じ取引日の一部のローソク足が欠落していることを検出できません。

過去の金融データをダウンロードする際は、API のクエリ範囲と1回のリクエストで取得できるデータ件数にも注意が必要です。長期間のデータが必要な場合は、対象 API の仕様に従ってデータを分割取得し、結合後に重複を確認します。1回のリクエストが成功したからといって、必要な期間の履歴データをすべて取得できたとは限りません。

タイムスタンプと OHLC の価格関係を確認する

時間に関するチェックでは、主に3つの問題を確認します。タイムスタンプを解析できないこと、同じ金融商品と時間足でタイムスタンプが重複していること、そして実際の時間範囲が想定と一致していないことです。また、同じ時間基準で比較できるよう、タイムスタンプの単位とタイムゾーンを統一する必要があります。これを怠ると、同じローソク足が別の時刻として扱われたり、誤った取引日に分類されたりする可能性があります。

価格チェックでは、OHLC フィールド間の基本的な関係を利用できます。正常なローソク足は、次の条件を満たす必要があります。

• 高値は始値と終値の両方以上である。
• 安値は始値と終値の両方以下である。
• 高値は安値以上である。
• 始値、高値、安値、終値のすべてが有効な数値である。

価格が突然大きく上昇または下落した場合も、確認すべき対象になります。ただし、それだけでデータエラーと判断してはいけません。実際の相場変動、企業行動、株価調整の基準の違い、データ欠落など、さまざまな原因が考えられます。まず疑わしいレコードに印を付け、元データや該当する市場の状況を確認します。

Python で欠落した取引日と異常なローソク足を検出する

以下では pandas を使用して、単一の金融商品と単一の時間足を対象に、ダウンロードしたデータをチェックします。この例では、履歴データが history_kline.csv に整理されており、bar_time、trade_date、open、high、low、close の6つのフィールドが含まれていることを前提とします。market_calendar.csv には、対象市場で想定される取引日が保存されているものとします。

bar_time は、API のレスポンス形式に応じて、タイムゾーン付きの時刻または標準化された UTC 時刻にあらかじめ統一しておく必要があります。trade_date は、対象市場の取引日ルールに基づいて生成してください。

import numpy as np
import pandas as pd

# ダウンロード済みの履歴ローソク足データを読み込む
df = pd.read_csv("history_kline.csv")

required = {
    "bar_time", "trade_date",
    "open", "high", "low", "close"
}

missing_columns = required - set(df.columns)
if missing_columns:
    raise ValueError(
        f"必須フィールドが不足しています: {sorted(missing_columns)}"
    )

# bar_time は解析可能な時刻形式に統一済みとする
df["bar_time"] = pd.to_datetime(
    df["bar_time"], errors="coerce", utc=True
)

price_cols = ["open", "high", "low", "close"]

for col in price_cols:
    df[col] = pd.to_numeric(df[col], errors="coerce")

# 1. 無効なタイムスタンプを確認する
invalid_time = df[df["bar_time"].isna()].copy()

# 2. 重複したタイムスタンプを確認する
valid_time = df[df["bar_time"].notna()]
duplicates = valid_time[
    valid_time.duplicated("bar_time", keep=False)
].copy()

# 3. 欠損価格、有限値でない価格、異常な OHLC 関係を確認する
price_array = df[price_cols].to_numpy(dtype=float)
bad_price = pd.Series(
    ~np.isfinite(price_array).all(axis=1),
    index=df.index
)

bad_ohlc = (
    (df["high"] < df[["open", "close"]].max(axis=1))
    | (df["low"] > df[["open", "close"]].min(axis=1))
    | (df["high"] < df["low"])
) & ~bad_price

invalid_prices = df[bad_price].copy()
abnormal_ohlc = df[bad_ohlc].copy()

# 4. 実際の取引日と取引日カレンダーを比較する
calendar = pd.read_csv("market_calendar.csv")

if "trade_date" not in calendar.columns:
    raise ValueError("取引日カレンダーに trade_date フィールドがありません")

expected = pd.to_datetime(
    calendar["trade_date"], errors="coerce"
)

if expected.isna().any():
    raise ValueError("取引日カレンダーに解析できない日付があります")

actual = pd.to_datetime(
    df["trade_date"], errors="coerce"
)

expected_dates = set(expected.dt.strftime("%Y-%m-%d"))
actual_dates = set(actual.dt.strftime("%Y-%m-%d").dropna())

missing_dates = sorted(expected_dates - actual_dates)

# 5. 結果を保存して後から確認できるようにする
invalid_time.to_csv("invalid_timestamps.csv", index=False)
duplicates.to_csv("duplicate_bars.csv", index=False)
invalid_prices.to_csv("invalid_prices.csv", index=False)
abnormal_ohlc.to_csv("abnormal_ohlc.csv", index=False)

pd.Series(
    missing_dates, name="missing_trade_date"
).to_csv("missing_trade_dates.csv", index=False)

print("無効なタイムスタンプ数:", len(invalid_time))
print("重複したローソク足レコード数:", len(duplicates))
print("無効な価格レコード数:", len(invalid_prices))
print("異常な OHLC レコード数:", len(abnormal_ohlc))
print("欠落した取引日数:", len(missing_dates))

実行すると、タイムスタンプ、重複レコード、無効な価格、異常な OHLC、欠落した取引日の一覧をそれぞれ確認できます。疑わしいデータを一律に削除するのではなく、問題の種類ごとに分けて保存することが重要です。

なお、このコードは入力ファイルのフィールド対応付けが完了していることを前提としており、単一の金融商品と単一の時間足を対象としています。複数の金融商品や時間足を含むファイルでは、重複を判定する際に金融商品コード、時間足、タイムスタンプを組み合わせる必要があります。

また、取引日カレンダーとの比較で検出できるのは、基本的に1日分のデータが欠落しているケースです。ある取引日にデータが存在していても、その日の数本の分足が欠落している場合は、取引時間と時間足に基づいて想定されるタイムスタンプを生成し、実際のデータと照合する必要があります。

異常を見つけても、すぐにデータを補完しない

欠落した取引日は通常の休場によるものかもしれませんし、ダウンロード範囲が不完全である可能性もあります。重複レコードは、分割取得したリクエストの時間範囲が重なったことが原因かもしれません。価格の異常についても、実際の市場変動やデータ処理上の問題など、複数の原因が考えられます。異常一覧、元のレスポンス、市場カレンダーを組み合わせて、原因を確認する必要があります。

特に、欠落したすべてのローソク足を直前の終値で自動的に埋めることは避けてください。この処理によって値動きのない記録が人工的に作られ、移動平均、ボラティリティ、売買シグナルが変化する可能性があります。業務上データの補完が必要な場合は、独立したルールを設け、元データを保持するとともに、補完したレコードを明確に識別できるようにしてください。

AllTick を利用して過去の相場データを取得する場合は、AllTick 公式開発者ポータル から該当する API ドキュメントを確認し、金融商品の種類、時間足、クエリ範囲、取得件数の制限を確認できます。履歴データの取得、取引日カレンダーによる検証、異常レコードの確認を分けて管理することで、取得したデータを分析やバックテストにより適した状態に整えられます。