File size: 4,580 Bytes
6b66ac0 | 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 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Tests for the Level-2 large-resting-order factor."""
from __future__ import annotations
import math
import pytest
from scanner.factor_sources import StubDataSource
from scanner.l2_factor import (
BIG_SIZE,
CLIP_RANGE,
SPOOF_AGE_THRESH,
TOP_LEVELS,
_book_lying,
compute_l2_factor,
compute_l2_factors,
)
def _book(bids, asks, age=5.0):
"""Wrap bid/ask lists in the dict schema the factor expects."""
return {
"ticker": "TEST",
"ts": "2026-06-02T14:30:00Z",
"bids": [[p, s, "NSDQ", age] for p, s in bids],
"asks": [[p, s, "NSDQ", age] for p, s in asks],
}
class _StaticSource:
"""Data source that returns the same book for every ticker."""
def __init__(self, book):
self._book = book
def get_l2_snapshot(self, ticker):
return self._book
def test_balanced_book_is_near_zero():
book = _book(
[(100.0, 1000), (99.99, 800), (99.98, 600)],
[(100.01, 1000), (100.02, 800), (100.03, 600)],
)
f = compute_l2_factor("X", source=_StaticSource(book))
assert -0.5 < f < 0.5, f"expected near-zero for balanced book, got {f}"
def test_bid_heavy_book_is_positive():
book = _book(
[(100.0, 50000), (99.99, 40000), (99.98, 30000), (99.97, 20000), (99.96, 10000)],
[(100.01, 200), (100.02, 200), (100.03, 200), (100.04, 200), (100.05, 200)],
)
f = compute_l2_factor("X", source=_StaticSource(book))
assert f > 0.3, f"expected positive for bid-heavy book, got {f}"
def test_ask_heavy_book_is_negative():
book = _book(
[(100.0, 200), (99.99, 200), (99.98, 200), (99.97, 200), (99.96, 200)],
[(100.01, 50000), (100.02, 40000), (100.03, 30000), (100.04, 20000), (100.05, 10000)],
)
f = compute_l2_factor("X", source=_StaticSource(book))
assert f < -0.3, f"expected negative for ask-heavy book, got {f}"
def test_clipped_to_range():
book = _book(
[(100.0, 1_000_000)] * TOP_LEVELS,
[(100.01, 1)] * TOP_LEVELS,
)
f = compute_l2_factor("X", source=_StaticSource(book))
assert -CLIP_RANGE <= f <= CLIP_RANGE
def test_spoofed_orders_ignored():
"""Orders with age < SPOOF_AGE_THRESH should be filtered out."""
book = _book(
[(100.0, 50000)], # one big bid
[(100.01, 200)],
)
# With normal ages - all included, factor should be bid-heavy
f_normal = compute_l2_factor("X", source=_StaticSource(book))
assert f_normal > 0.2
# With spoof ages (0.1s) - the big bid is filtered, book looks thin
book_spoofed = {
"ticker": "X",
"ts": "2026-06-02T14:30:00Z",
"bids": [[100.0, 50000, "NSDQ", 0.1]], # spoof!
"asks": [[100.01, 200, "NSDQ", 5.0]],
}
f_spoofed = compute_l2_factor("X", source=_StaticSource(book_spoofed))
# Spoofed book should be weaker (closer to 0)
assert abs(f_spoofed) < abs(f_normal)
def test_book_lying_detection():
assert _book_lying(0.7, -0.005) is True # bid-heavy but falling
assert _book_lying(0.7, 0.0) is False # bid-heavy and flat
assert _book_lying(0.3, 0.005) is True # ask-heavy but rising
assert _book_lying(0.5, 0.0) is False # neutral
def test_book_lying_discounts_factor():
book = _book(
[(100.0, 50000), (99.99, 40000), (99.98, 30000), (99.97, 20000), (99.96, 10000)],
[(100.01, 200), (100.02, 200), (100.03, 200), (100.04, 200), (100.05, 200)],
)
f_honest = compute_l2_factor("X", recent_return=0.001, source=_StaticSource(book))
f_lying = compute_l2_factor("X", recent_return=-0.01, source=_StaticSource(book))
# "Lying" book should produce a smaller absolute factor
assert abs(f_lying) < abs(f_honest)
def test_empty_book_returns_zero():
f = compute_l2_factor("X", source=_StaticSource(None))
assert f == 0.0
f = compute_l2_factor("X", source=_StaticSource({"bids": [], "asks": []}))
assert f == 0.0
def test_batch_returns_all_tickers():
book = _book(
[(100.0, 1000), (99.99, 800)],
[(100.01, 1000), (100.02, 800)],
)
src = _StaticSource(book)
out = compute_l2_factors(["A", "B", "C"], source=src)
assert set(out.keys()) == {"A", "B", "C"}
for v in out.values():
assert -CLIP_RANGE <= v <= CLIP_RANGE
def test_stub_data_source_synthesises():
src = StubDataSource(stub_dir="/nonexistent")
book = src.get_l2_snapshot("AAPL")
assert book is not None
assert "bids" in book and "asks" in book
assert len(book["bids"]) == 10
assert all(len(b) == 4 for b in book["bids"])
|