okx-codex-trader/tests/test_validate_data_pipeline.py

import importlib.util
import sys
from pathlib import Path

from okx_codex_trader.models import Candle


def load_data_validation_module():
    path = Path(__file__).resolve().parents[1] / "scripts" / "validate_data_pipeline.py"
    spec = importlib.util.spec_from_file_location("validate_data_pipeline", path)
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    sys.modules[spec.name] = module
    spec.loader.exec_module(module)
    return module


def build_candles(count: int) -> list[Candle]:
    return [
        Candle(
            symbol="BTC-USDT-SWAP",
            ts=index * 60_000,
            open=100.0 + index,
            high=101.0 + index,
            low=99.0 + index,
            close=100.5 + index,
            volume=1_000.0 + index,
        )
        for index in range(count)
    ]


def test_audit_candles_passes_clean_ohlcv_pipeline():
    module = load_data_validation_module()
    audit = module.audit_candles(candles=build_candles(80), symbol="BTC-USDT-SWAP", bar="1m")

    assert audit.passed
    assert audit.rows == 80
    assert audit.duplicate_timestamps == 0
    assert audit.unexpected_intervals == 0
    assert audit.invalid_ohlc_rows == 0
    assert audit.invalid_volume_rows == 0
    assert audit.cache_roundtrip_matches
    assert audit.frame_conversion_matches


def test_audit_candles_fails_duplicate_gap_and_invalid_ohlcv():
    module = load_data_validation_module()
    candles = build_candles(5)
    candles[2] = Candle(
        symbol="BTC-USDT-SWAP",
        ts=candles[1].ts,
        open=100.0,
        high=99.0,
        low=101.0,
        close=100.0,
        volume=-1.0,
    )
    audit = module.audit_candles(candles=candles, symbol="BTC-USDT-SWAP", bar="1m")

    assert not audit.passed
    assert audit.duplicate_timestamps == 1
    assert audit.non_increasing_steps == 1
    assert audit.unexpected_intervals == 2
    assert audit.invalid_ohlc_rows == 1
    assert audit.invalid_volume_rows == 1


def test_audit_candles_reports_live_overlap_mismatches():
    module = load_data_validation_module()
    candles = build_candles(3)
    live = [
        candles[0],
        Candle(
            symbol="BTC-USDT-SWAP",
            ts=candles[1].ts,
            open=candles[1].open,
            high=candles[1].high,
            low=candles[1].low,
            close=candles[1].close + 1.0,
            volume=candles[1].volume,
        ),
    ]

    audit = module.audit_candles(candles=candles, symbol="BTC-USDT-SWAP", bar="1m", live_candles=live)

    assert not audit.passed
    assert audit.live_overlap_rows == 2
    assert audit.live_mismatches == 1