本稿は、年間約3,400万行におよぶ為替Tickデータ（HistData.com提供）から、市場ミクロ構造（マーケットマイクロストラクチャー）の歪みを抽出し、M1（1分足）～M15（15分足）の個人デイトレードへ転用可能なエッジを検証するための高速解析データ基盤（NumPyマスターバイナリ）の構築ログである。
目的
MT5のリアルタイムTickフィードに含まれるフラグ属性（Bidのみ変化: 98、Askのみ変化: 100、両方更新: 102、気配消失・静寂: 96）は、ブローカーのサーバー上では通常2日程度しか保存されない。また、HistData等の長期ヒストリカルデータはスナップショット型（常にBid/Askの双方が埋められた状態）となっており、そのままではフラグ情報が欠落している。
本プロジェクトの目的は以下の2点にある：
- ミクロ構造の復元：HistDataの連続Tickにおける価格差分（$\Delta Bid, \Delta Ask$）および到達時間差（$\Delta t$）から、MT5同等の板更新状態・気配真空状態を逆算・独自フラグ化する。
- 個人執行に即したデータセット化：HFT（高頻度取引）のミリ秒争いは排除し、ブローカーのレイテンシや約定拒否を回避しつつ、15秒・1分・5分・15分の時間軸で最大騰落幅（Max Rise / Max Fall）を事前結合（カンペ化）した64バイト固定長の超高速NumPyバイナリを構築する。
エッジがあるかないかテストするプラン
構築したデータセットを用いて、以下の3つの仮説に基づくエッジの有無を検証する。
プランA：オーダーフロー・インバランス（OFIR）とダイバージェンス
M1足内部における「Bid単独更新数（売り板叩き・買い指値後退）」と「Ask単独更新数（買い板叩き・売り指値後退）」の比率を算出し、ローソク足の実体方向（陽線/陰線）との乖離（ダイバージェンス）を検知する。例えば、ローソク足は大陽線を作っているにもかかわらず、高値圏でBid単独変化が急増しているケースは、天井の厚い売り指値にブロックされて買いが尽きている兆候とみなし、逆張り（ショート）の優位性を検証する。
プランB：時間間隔による静寂（真空・96）ブレイクアウト / 逆張り
欧州・NYなどの活況セッション中、通常は数百ミリ秒単位で流れるTickが突如として数秒以上（$\Delta t \ge 3.0$秒）途絶える「静寂（流動性の真空状態）」を検知する。リクイディティプロバイダーが板を引いた直後のスプレッド拡大と、その直後に再開された一撃の方向（ブレイクアウト追随）、あるいはオーバーシュート後の全戻し（逆張り）の期待値を評価する。
プランC：連続性の破綻（Streak Exhaustion：食い尽くし・踏み上げ）
98（売りパニック）が$N$回連続した直後に、一転して強烈な100（Ask更新）または102（両方上昇）が発生するパターンを検知する。パニック売りの成行注文が極小の底で一瞬にして吸収（食い尽くし）され、ショートカバーを巻き込んで急上昇する「踏み上げ」の初動を捉える。
CSVからParquet作成
HistData.comから取得した生CSV（約3,400万行）を、高速I/Oと省ディスク容量を両立するApache Parquet形式へ圧縮・型変換するスクリプト。文字列型のタイムスタンプや不要列を整理し、後続のNumPy変換へのロード時間を劇的に短縮する。
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import time
def csv_to_parquet(csv_file_path: str, parquet_file_path: str):
print(f"変換開始: {csv_file_path} -> {parquet_file_path}")
start_time = time.time()
# HistDataのCSVフォーマット仕様 (コロン区切りの日時の場合あり)
# 例: 20251231 165857586,156.655,156.682,0
df = pd.read_csv(
csv_file_path,
sep=';', # HistDataの区切り文字に合わせて設定 (',' または ';')
names=['timestamp', 'bid', 'ask', 'volume'],
dtype={'bid': 'float64', 'ask': 'float64'},
usecols=['timestamp', 'bid', 'ask']
)
# Parquetへ高速出力 (Snappy圧縮)
table = pa.Table.from_pandas(df)
pq.write_table(table, parquet_file_path, compression='snappy')
print(f"✅ 変換完了: {len(df):,} 行 / 処理時間: {time.time() - start_time:.2f} 秒")
if __name__ == '__main__':
csv_to_parquet('DAT_ASCII_USDJPY_T_2025.csv', 'DAT_ASCII_USDJPY_TICK_2025.parquet')
Histdata.comのTickデータには、このGapの大半は市場側の流動性真空ではなく、HistData（または元のブローカー）側のデータ収集サーバーのパケロス、API切断、再接続ラグなどの「インフラ的欠損」があり、Tickと一緒に、gap_reportが同梱されている。意味のないTickの空白期間をバックテストから除外するために、parquetを再作成
import polars as pl
import re
import os
# ==========================================
# 1. Gap レポートテキストのパース
# ==========================================
def parse_gap_report(gap_txt_path: str):
"""
merged_gap_report.txt を読み込み、
gap_start(無効化開始時刻 = Gapの15分前), gap_end(Gap終了時刻) のリストを返す。
"""
pattern = re.compile(r"between (\d{14}) and (\d{14})")
records = []
if not os.path.exists(gap_txt_path):
print(f"⚠️ {gap_txt_path} が見つかりません。Gap除外なしで続行します。")
return []
with open(gap_txt_path, "r", encoding="utf-8") as f:
for line in f:
m = pattern.search(line)
if m:
# ログのフォーマット: YYYYMMDDHHMMSS
start_dt = pl.Series([m.group(1)]).str.strptime(pl.Datetime("ms"), format="%Y%m%d%H%M%S")[0]
end_dt = pl.Series([m.group(2)]).str.strptime(pl.Datetime("ms"), format="%Y%m%d%H%M%S")[0]
import datetime
# カンペ（最大15分）が欠損を跨がないよう、Gap開始の15分前から無効化対象とする
invalid_start = start_dt - datetime.timedelta(minutes=15)
records.append((invalid_start, end_dt))
print(f"✅ Gapログ読み込み完了: {len(records):,} 件の欠損区間を登録")
return records
# ==========================================
# 2. メイン生成処理
# ==========================================
def build_clean_parquet(
csv_path: str,
gap_report_path: str,
output_parquet_path: str
):
print("1. Gapレポートの解析中...")
gap_intervals = parse_gap_report(gap_report_path)
print("2. CSVのスキャンと前処理スキーマ構築中...")
# カラムは4つ。最後の0(status_code)は読み込むが後で捨てる
column_names = ["timestamp", "bid", "ask", "status_code"]
lazy_df = (
pl.scan_csv(
csv_path,
has_header=False,
new_columns=column_names,
)
.with_columns([
# タイムスタンプの整形: "20251201 000000079" -> "20251201 000000.079"
pl.col("timestamp").str.replace(r"(\d{6})(\d{3})$", r"${1}.${2}")
])
.with_columns([
# 正規化された文字列をDatetime(ミリ秒)へパース
pl.col("timestamp").str.strptime(pl.Datetime("ms"), format="%Y%m%d %H%M%S.%f").alias("dt")
])
.with_columns([
# ミリ秒単位の Unix Epoch (NumPyで直接 int64 として使える)
pl.col("dt").dt.epoch(time_unit="ms").alias("epoch_ms"),
# 前後Tickの時間差（ミリ秒）
(pl.col("dt").diff().dt.total_milliseconds().fill_null(0)).alias("delta_ms")
])
# セッション境界（週末や60秒以上の断絶）を検知してセッションIDをインクリメント
.with_columns([
(pl.col("delta_ms") >= 60000).cum_sum().alias("session_id")
])
)
print("3. データクレンジングフラグ（is_clean）の構築中...")
# Gap区間（およびその直前15分）に引っかかるかを判定する条件式を生成
if gap_intervals:
# Polarsの in_between 条件をORで連結
gap_exprs = [
pl.col("dt").is_between(start, end)
for start, end in gap_intervals
]
is_in_gap = pl.any_horizontal(gap_exprs)
final_lazy_df = lazy_df.with_columns([
(~is_in_gap & (pl.col("delta_ms") < 60000)).alias("is_clean")
])
else:
final_lazy_df = lazy_df.with_columns([
(pl.col("delta_ms") < 60000).alias("is_clean")
])
print(f"4. Parquetへストリーミング書き出し中... -> {output_parquet_path}")
(
final_lazy_df.select([
"epoch_ms",
"session_id",
"bid",
"ask",
"delta_ms",
"is_clean" # NumPy側で IS_VALID フラグの基礎として使用
])
.sink_parquet(output_parquet_path, compression="zstd")
)
print("🎉 クリーンなParquetファイルの生成が完了しました！")
if __name__ == "__main__":
build_clean_parquet(
csv_path="sorted_output.csv",
gap_report_path="merged_gap_report.txt",
output_parquet_path="clean_tick_data.parquet"
)NumPyの構造
Core i5-8500（6コア、L3キャッシュ9MB）環境下で、キャッシュスラッシング（破棄競合）を防ぎメモリ帯域を浪費しないため、「1行64バイト固定（int32 × 16列）」「全列int32統一」「NaN完全排除」の構造化バイナリ（tick_master.npy）を設計した。
CPUのキャッシュライン（64バイト）と1行のサイズが1対1で完全一致するため、先頭の判定フラグをロードした瞬間に同一行の全データがL1/L2キャッシュに乗る構造となっている。
| 列番号 | カラム名 | 型 | 単位・形式 | 設計意図・役割 |
|---|---|---|---|---|
| Col 0 | micro_flag | int32 | ビットマスク | 【判定系】状態フラグ（XOR・ハミング距離スキャン対象） |
| Col 1 | streak_len | int32 | 連続回数 | 【案C】同一フラグ（98/100）の連続カウント |
| Col 2 | timestamp_sec | int32 | 経過秒数 | 2025-01-01 00:00:00からのオフセット秒（M1集約キー） |
| Col 3 | delta_ms | int32 | ミリ秒 | 【案B】直前Tickからの経過時間（上限60,000msクリップ済） |
| Col 4 | bid_int | int32 | 価格 × 係数 | 生Bid価格（JPY: ×1,000 / その他: ×100,000） |
| Col 5 | ask_int | int32 | 価格 × 係数 | 生Ask価格 |
| Col 6 | mid_int | int32 | 価格 × 係数 | 基準Mid価格（(Bid + Ask) / 2） |
| Col 7 | spread_int | int32 | 0.1 pips | スプレッド（Ask – Bid） |
| Col 8 | max_rise_15s | int32 | 0.1 pips | 直後15秒以内の最大上昇幅（カンペ） |
| Col 9 | max_fall_15s | int32 | 0.1 pips | 直後15秒以内の最大下落幅（即死判定） |
| Col 10 | max_rise_60s | int32 | 0.1 pips | 直後60秒以内の最大上昇幅（M1足完成・建値移動判定） |
| Col 11 | max_fall_60s | int32 | 0.1 pips | 直後60秒以内の最大下落幅 |
| Col 12 | max_rise_5m | int32 | 0.1 pips | 直後5分以内の最大上昇幅 |
| Col 13 | max_fall_5m | int32 | 0.1 pips | 直後5分以内の最大下落幅 |
| Col 14 | max_rise_15m | int32 | 0.1 pips | 直後15分以内の最大上昇幅（デイトレ本命利確判定） |
| Col 15 | max_fall_15m | int32 | 0.1 pips | 直後15分以内の最大下落幅（許容最大ドローダウン） |
Col 0: micro_flag のビット定義表
最上位ビット（Bit 31）を有効行フラグに割り当て、末尾未確定領域や週末跨ぎの無効データを一括除外できるようにしている。また、POPCNT（ビットカウント命令）を用いたハミング距離スキャンにより、「特定条件と完全一致または1ビット違い（ニアピン）」のTick抽出を1クロックに近い速度で実行可能にしている。
Bit 0 (0x0001): IS_98（Bid単独変化）Bit 1 (0x0002): IS_100（Ask単独変化）Bit 2 (0x0004): IS_102（両方変化）Bit 3 (0x0008): IS_SAME（価格変化なし）Bit 4 (0x0010): IS_96（真空・静寂: $\Delta t \ge 3.0$秒）Bit 5 (0x0020): SPREAD_NORMAL（通常スプレッド: $\le 1.5$ pips）Bit 6 (0x0040): SPREAD_WIDE（スプレッド拡大: $\ge 2.5$ pips）Bit 7 (0x0080): ACTIVE_SESSION（コア流動性セッション）Bit 8 (0x0100): MID_UP（前TickよりMid上昇）Bit 9 (0x0200): MID_DOWN（前TickよりMid下落）Bit 10 (0x0400): STREAK_REVERSAL（同方向連続後の反転トリガー）Bit 11 (0x0800): DIVERGENCE（価格と板変化の逆流）Bit 31 (0x80000000): IS_VALID（有効行フラグ：週末跨ぎ・未確定末尾を除外）
NumPy 生成
Parquetを読み込み、銘柄の桁数（JPY系は3桁、それ以外は5桁）を動的判定してスケーリングを行った後、Numba（LLVM JIT）によってC言語並みの速度で全行スキャンおよび未来15分間の最大騰落（カンペ）を前方探索・書き込みするスクリプト。
import pandas as pd
import numpy as np
from numba import njit
import time
import os
# ==========================================
# 1. 通貨ペア判別・桁数取得関数
# ==========================================
def get_digit(parquet_file: str) -> int:
filename = os.path.basename(parquet_file).upper()
if 'JPY' in filename:
return 3
else:
return 5
# ==========================================
# 2. ビットマスク定義 (32-bit Integer)
# ==========================================
IS_98 = 1 << 0 # Bid単独変化
IS_100 = 1 << 1 # Ask単独変化
IS_102 = 1 << 2 # 両方変化
IS_SAME = 1 << 3 # 変化なし
IS_96 = 1 << 4 # 静寂 (Δt >= 3.0s)
SPREAD_NORMAL = 1 << 5 # 通常スプレッド (<= 1.5 pips)
SPREAD_WIDE = 1 << 6 # スプレッド拡大 (>= 2.5 pips)
ACTIVE_SESSION = 1 << 7 # 活況時間帯
MID_UP = 1 << 8 # Mid上昇
MID_DOWN = 1 << 9 # Mid下落
STREAK_REVERSAL = 1 << 10 # 連続後の反転
DIVERGENCE = 1 << 11 # ダイバージェンス
IS_VALID = 1 << 31 # 0x80000000 (有効行フラグ)
INT32_MIN = -2147483648
# ==========================================
# 3. Numba JIT によるコア演算エンジン
# ==========================================
@njit
def build_tick_master_numba(
N, timestamps, delta_ms_arr, bid_arr, ask_arr, mid_arr, spread_arr,
streak_threshold=5,
silence_ms=3000
):
out = np.zeros((N, 16), dtype=np.int32)
current_streak_type = 0 # 1: 98(Bid変化), 2: 100(Ask変化)
current_streak_len = 0
for i in range(1, N):
bid = bid_arr[i]
ask = ask_arr[i]
prev_bid = bid_arr[i-1]
prev_ask = ask_arr[i-1]
mid = mid_arr[i]
prev_mid = mid_arr[i-1]
t_sec = timestamps[i]
d_ms = delta_ms_arr[i]
spread = spread_arr[i]
flag = 0
# --- 基本 Tick状態判定 ---
d_bid = bid - prev_bid
d_ask = ask - prev_ask
is_98 = (d_bid != 0) and (d_ask == 0)
is_100 = (d_bid == 0) and (d_ask != 0)
if is_98:
flag |= IS_98
elif is_100:
flag |= IS_100
elif (d_bid != 0) and (d_ask != 0):
flag |= IS_102
else:
flag |= IS_SAME
# --- 静寂判定 (96相当) ---
if d_ms >= silence_ms:
flag |= IS_96
# --- スプレッド判定 (整数1 = 0.1 pipsのため、1.5pips=15, 2.5pips=25) ---
if spread <= 15:
flag |= SPREAD_NORMAL
elif spread >= 25:
flag |= SPREAD_WIDE
# --- Mid方向判定 ---
if mid > prev_mid:
flag |= MID_UP
elif mid < prev_mid:
flag |= MID_DOWN
# --- 連続(Streak) と 反転(Reversal) 判定 ---
streak_rev = False
if is_98:
if current_streak_type == 1:
current_streak_len += 1
else:
if current_streak_type == 2 and current_streak_len >= streak_threshold:
streak_rev = True
current_streak_type = 1
current_streak_len = 1
elif is_100:
if current_streak_type == 2:
current_streak_len += 1
else:
if current_streak_type == 1 and current_streak_len >= streak_threshold:
streak_rev = True
current_streak_type = 2
current_streak_len = 1
else:
current_streak_len = 0
current_streak_type = 0
if streak_rev:
flag |= STREAK_REVERSAL
# --- 妥当性判定 (週末・末尾15分) ---
if d_ms > 60000 or (timestamps[N-1] - t_sec < 900):
pass
else:
flag |= IS_VALID
# 左側8列への書き込み
out[i, 0] = flag
out[i, 1] = current_streak_len
out[i, 2] = t_sec
out[i, 3] = min(d_ms, 60000)
out[i, 4] = bid
out[i, 5] = ask
out[i, 6] = mid
out[i, 7] = spread
# --- 右側カンペ (Max Rise / Max Fall) の前方探索 ---
if (flag & IS_VALID) != 0:
max_15, min_15 = mid, mid
max_60, min_60 = mid, mid
max_300, min_300 = mid, mid
max_900, min_900 = mid, mid
j = i + 1
while j < N:
future_t = timestamps[j]
elapsed = future_t - t_sec
if elapsed > 900:
break
f_mid = mid_arr[j]
if elapsed <= 15:
if f_mid > max_15: max_15 = f_mid
if f_mid < min_15: min_15 = f_mid
if elapsed <= 60:
if f_mid > max_60: max_60 = f_mid
if f_mid < min_60: min_60 = f_mid
if elapsed <= 300:
if f_mid > max_300: max_300 = f_mid
if f_mid < min_300: min_300 = f_mid
if f_mid > max_900: max_900 = f_mid
if f_mid < min_900: min_900 = f_mid
j += 1
out[i, 8] = max_15 - mid
out[i, 9] = mid - min_15
out[i, 10] = max_60 - mid
out[i, 11] = mid - min_60
out[i, 12] = max_300 - mid
out[i, 13] = mid - min_300
out[i, 14] = max_900 - mid
out[i, 15] = mid - min_900
else:
out[i, 8:] = INT32_MIN
return out
# ==========================================
# 4. メインパイプライン
# ==========================================
def generate_tick_master(parquet_path: str, output_npy_path: str):
digits = get_digit(parquet_path)
multiplier = 10 ** digits
print(f"銘柄判定結果: 小数点 {digits} 桁 (スケーリング係数: {multiplier:,})")
print("1. Parquetデータの読み込み中...")
start = time.time()
df = pd.read_parquet(parquet_path, columns=['timestamp', 'bid', 'ask'])
df['timestamp'] = pd.to_datetime(df['timestamp'], format='%Y%m%d %H%M%S%f')
base_time = pd.Timestamp('2025-01-01 00:00:00')
timestamps_sec = (df['timestamp'] - base_time).dt.total_seconds().astype(np.int32).values
delta_ms = df['timestamp'].diff().dt.total_seconds() * 1000
delta_ms = np.nan_to_num(delta_ms, nan=0).astype(np.int32)
print(f"2. スケーリング処理 (* {multiplier}) とNumPy配列化...")
bid_int = np.round(df['bid'].values * multiplier).astype(np.int32)
ask_int = np.round(df['ask'].values * multiplier).astype(np.int32)
mid_int = np.round(((df['bid'].values + df['ask'].values) / 2.0) * multiplier).astype(np.int32)
spread_int = ask_int - bid_int
N = len(df)
print(f" => データ準備完了: {N:,} 行 ({time.time() - start:.2f}秒)")
print("3. Numba JITエンジンでの全行・未来カンペ走査開始...")
start_numba = time.time()
out_master = build_tick_master_numba(
N, timestamps_sec, delta_ms, bid_int, ask_int, mid_int, spread_int,
streak_threshold=5,
silence_ms=3000
)
print(f" => 走査完了! ({time.time() - start_numba:.2f}秒)")
print(f"4. NumPyバイナリ (Shape: {out_master.shape}, {out_master.nbytes / 1024**3:.2f} GB) の保存中...")
np.save(output_npy_path, out_master)
print(f"✅ 全工程完了。保存先: {output_npy_path}")
if __name__ == '__main__':
generate_tick_master('DAT_ASCII_USDJPY_TICK_2025.parquet', 'tick_master.npy')
