prediction_engine/python_bridge/tests/test_orchestrator.py

"""Smoke test: write ticks to mmap, read with Python, score with KAN."""
import struct
import tempfile
import time
from pathlib import Path

from prediction_engine.python_bridge.mmap_reader import (
    HEADER_FMT, HEADER_SIZE, MAGIC, TICK_FMT, TICK_SIZE, MmapReader,
)
from prediction_engine.python_bridge.kan_scorer import MarketFeatureExtractor, KANFastScorer


def test_end_to_end_smoke():
    path = tempfile.mktemp(suffix=".mmap")
    capacity = 100
    file_size = HEADER_SIZE + capacity * TICK_SIZE
    with open(path, "wb") as f:
        f.write(struct.pack(HEADER_FMT, MAGIC, capacity, 2, 0))
        f.write(struct.pack(TICK_FMT, 0, 0, 0.62, 100.0, int(time.time() * 1e6), 111))
        f.write(struct.pack(TICK_FMT, 1, 1, 0.35, 200.0, int(time.time() * 1e6), 222))
        f.write(b"\x00" * (file_size - HEADER_SIZE - 2 * TICK_SIZE))

    reader = MmapReader(path)
    reader.open()
    ticks = reader.read_new_ticks()
    assert len(ticks) == 2

    extractor = MarketFeatureExtractor()
    features = extractor.extract({
        "yes_price": ticks[0].price,
        "no_price": ticks[1].price,
        "spread": abs(1.0 - ticks[0].price - ticks[1].price),
        "volume_ratio": ticks[0].size / max(ticks[1].size, 1),
        "time_to_event_hours": 24.0,
        "venue_count": 2,
    })

    scorer = KANFastScorer(in_features=6)
    result = scorer.score(features)
    assert "confidence" in result
    assert "kk_phase" in result

    total_cost = ticks[0].price + ticks[1].price
    assert total_cost < 1.0

    reader.close()
    Path(path).unlink()