Spaces:
Build error
Build error
File size: 1,570 Bytes
201cf4d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 | """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()
|