feat: Phase 2 — microstructure analytics + 81 tests

New microstructure/ module with pure-function analytics: microstructure/book.py: microprice() — depth-weighted mid price mid_price() — simple bid/ask midpoint order_book_imbalance() — ranged [-1, 1] volume skew depth_imbalance() — imbalance at fixed price distance spread_stats() — spread, spread_bps, mid, bid, ask depth_resiliency() — bid/ask volume within impact radius queue_depletion_prob() — Poisson fill probability at level batch_book_stats() — aggregate stats across snapshots microstructure/trades.py: classify_lee_ready() — Lee-Ready aggressor classification classify_bulk_lee_ready() — batch classification with mids/bids/asks compute_markouts() — forward mid-price change at configurable horizons markout_summary() — mean/std/t-stat per side per horizon trade_volume_profile() — size bucket distribution trade_arrival_rate() — rolling trades/sec with burst detection microstructure/toxicity.py: compute_vpin() — volume-synchronized informed trading probability compute_vpin_time_series() — rolling VPIN with alarm threshold fill_toxicity() — adverse price movement post-trade adverse_selection_ratio() — per-side adverse selection liquidation_clustering() — cluster detection in liquidation events microstructure/funding.py: funding_regime() — classify regime (neutral/positive/negative/high) funding_predictability() — AR(1) autocorrelation analysis funding_carry_pnl() — cumulative carry PnL estimation basis_spread() — perp premium over spot (bps) basis_convergence_speed() — mean-reversion half-life via AR(1) microstructure/signals.py: composite_signal() — weighted OBI + trade + VPIN + funding signal SignalPipeline — stateful pipeline accumulating book/trade updates detect_hft_regime() — regime classifier for HFT strategy selection Bug fixes in Phase 1: - data/latency.py: proper linear-interpolation percentiles - data/normalizer.py: UTC timezone for naive datetimes - data/normalizer.py: detect_sequence_gap returns gap-1 (missing count) - microstructure/toxicity.py: consistent vpin_value key in compute_vpin 81 tests across 4 test files (store, normalizer, latency, microstructure)
2026-08-07 14:34:18 +08:00
parent a7f811eb81
commit fcfc136384
13 changed files with 1817 additions and 5 deletions
@@ -0,0 +1,80 @@
+"""
+Tests for data/latency.py — latency tracking.
+"""
+import time
+
+from data.latency import LatencyTracker
+
+
+def test_empty_stats():
+    lt = LatencyTracker()
+    s = lt.stats()
+    for key in ("transport_ms", "signal_ms", "order_ms", "roundtrip_ms"):
+        assert s[key]["count"] == 0
+
+
+def test_record_transport():
+    lt = LatencyTracker()
+    lt.record_transport(1705312800000, 1705312800.100)  # 100ms
+    s = lt.stats()
+    assert s["transport_ms"]["p50"] == 100.0
+    assert s["transport_ms"]["count"] == 1
+
+
+def test_record_signal():
+    lt = LatencyTracker()
+    lt.record_signal(5.2)
+    lt.record_signal(3.1)
+    s = lt.stats()
+    assert s["signal_ms"]["p50"] == 4.15  # interpolated median of [3.1, 5.2]
+    assert s["signal_ms"]["count"] == 2
+
+
+def test_record_order():
+    lt = LatencyTracker()
+    lt.record_order(12.5)
+    s = lt.stats()
+    assert s["order_ms"]["p50"] == 12.5
+
+
+def test_record_roundtrip():
+    lt = LatencyTracker()
+    lt.record_roundtrip(150.0)
+    s = lt.stats()
+    assert s["roundtrip_ms"]["p50"] == 150.0
+
+
+def test_negative_latency_ignored():
+    lt = LatencyTracker()
+    lt.record_transport(1705312800100, 1705312800.000)  # local before exchange
+    s = lt.stats()
+    assert s["transport_ms"]["count"] == 0
+
+
+def test_summary_compact():
+    lt = LatencyTracker()
+    lt.record_transport(1705312800000, 1705312800.100)
+    lt.record_signal(5.0)
+    summary = lt.summary()
+    assert "transport_ms" in summary
+    assert "signal_ms" in summary
+    assert "p50" in summary["transport_ms"]
+
+
+def test_percentiles_multiple_values():
+    lt = LatencyTracker()
+    for lat in [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]:
+        lt.record_signal(float(lat))
+    s = lt.stats()
+    assert s["signal_ms"]["p50"] == 55.0  # interpolated median of 10 values
+    assert s["signal_ms"]["p90"] == 91.0  # (10-1)*90/100=8.1, lo=8, vals[8]=90, vals[9]=100, frac=0.1, result=91
+
+
+def test_window_pruning_approx():
+    """Old records should be pruned after window expires."""
+    lt = LatencyTracker(window_seconds=0.1)
+    lt.record_signal(10.0)
+    time.sleep(0.15)
+    lt.record_signal(20.0)
+    s = lt.stats()
+    assert s["signal_ms"]["p50"] == 20.0  # only the recent one
@@ -0,0 +1,388 @@
+"""
+Tests for microstructure module — book, trades, toxicity, funding, signals.
+"""
+import math
+
+import numpy as np
+
+from microstructure.book import (
+    microprice,
+    mid_price,
+    order_book_imbalance,
+    depth_imbalance,
+    spread_stats,
+    depth_resiliency,
+    queue_depletion_prob,
+    batch_book_stats,
+)
+from microstructure.trades import (
+    classify_lee_ready,
+    classify_bulk_lee_ready,
+    compute_markouts,
+    markout_summary,
+    trade_volume_profile,
+    trade_arrival_rate,
+)
+from microstructure.toxicity import (
+    compute_vpin,
+    compute_vpin_time_series,
+    fill_toxicity,
+    adverse_selection_ratio,
+    liquidation_clustering,
+)
+from microstructure.funding import (
+    funding_regime,
+    funding_predictability,
+    funding_carry_pnl,
+    basis_spread,
+    basis_convergence_speed,
+)
+from microstructure.signals import (
+    composite_signal,
+    SignalPipeline,
+    detect_hft_regime,
+)
+
+
+# ── Helpers ────────────────────────────────────────────────
+
+def _basic_book():
+    """BTC-style book."""
+    bids = {50000.0 - i * 10.0: 1.0 + i * 0.2 for i in range(20)}
+    asks = {50001.0 + i * 10.0: 1.0 + i * 0.2 for i in range(20)}
+    return bids, asks
+
+
+# ═══════════════════════════════════════════════════════════
+# Book tests
+# ═══════════════════════════════════════════════════════════
+
+class TestMidPrice:
+    def test_simple_mid(self):
+        bids = {100.0: 1.0}
+        asks = {102.0: 1.0}
+        assert mid_price(bids, asks) == 101.0
+
+    def test_empty_returns_zero(self):
+        assert mid_price({}, {}) == 0.0
+
+
+class TestMicroprice:
+    def test_equal_depth(self):
+        bids = {100.0: 1.0}
+        asks = {102.0: 1.0}
+        mid = microprice(bids, asks)
+        assert mid == 101.0  # equal weight = simple mid
+
+    def test_bid_heavy(self):
+        bids = {100.0: 10.0}
+        asks = {102.0: 1.0}
+        mid = microprice(bids, asks)
+        assert mid < 101.0  # titled toward heavier bid side → lower price
+
+    def test_ask_heavy(self):
+        bids = {100.0: 1.0}
+        asks = {102.0: 10.0}
+        mid = microprice(bids, asks)
+        assert mid > 101.0  # tilted toward heavier ask side → higher price
+
+
+class TestOrderBookImbalance:
+    def test_balanced(self):
+        bids = {100.0: 5.0}
+        asks = {102.0: 5.0}
+        obi = order_book_imbalance(bids, asks)
+        assert obi == 0.0
+
+    def test_bid_heavy(self):
+        bids = {100.0: 8.0}
+        asks = {102.0: 2.0}
+        obi = order_book_imbalance(bids, asks)
+        assert obi > 0
+        assert obi == 0.6
+
+    def test_ask_heavy(self):
+        bids = {100.0: 2.0}
+        asks = {102.0: 8.0}
+        obi = order_book_imbalance(bids, asks)
+        assert obi < 0
+        assert obi == -0.6
+
+    def test_empty_book(self):
+        assert order_book_imbalance({}, {}) == 0.0
+
+
+class TestSpreadStats:
+    def test_basic(self):
+        bids = {50000.0: 1.0}
+        asks = {50002.0: 1.0}
+        stats = spread_stats(bids, asks)
+        assert stats["spread"] == 2.0
+        assert stats["best_bid"] == 50000.0
+        assert stats["best_ask"] == 50002.0
+        assert stats["spread_bps"] > 0
+
+    def test_empty(self):
+        stats = spread_stats({}, {})
+        assert stats["spread"] == 0
+
+
+class TestDepthResiliency:
+    def test_basic(self):
+        bids, asks = _basic_book()
+        dr = depth_resiliency(bids, asks, impact_bps=100.0)
+        assert dr["bid_vol"] > 0
+        assert dr["ask_vol"] > 0
+        assert dr["bid_levels"] > 0
+
+
+class TestQueueDepletionProb:
+    def test_probability_range(self):
+        bids = {50000.0: 1.0}
+        asks = {50002.0: 0.5}
+        prob = queue_depletion_prob(bids, asks, level_distance=0, trade_rate_per_sec=2.0, avg_trade_size=0.1)
+        assert 0.0 <= prob <= 1.0
+
+    def test_deep_book_low_prob(self):
+        bids = {50000.0: 100.0}
+        asks = {50002.0: 100.0}
+        prob = queue_depletion_prob(bids, asks, level_distance=0, trade_rate_per_sec=1.0, avg_trade_size=0.01)
+        assert prob < 0.01
+
+
+class TestBatchBookStats:
+    def test_multiple_snapshots(self):
+        bids, asks = _basic_book()
+        snaps = [{"bids": bids, "asks": asks} for _ in range(5)]
+        result = batch_book_stats(snaps)
+        assert result["obi"]["count"] == 5
+        assert result["spread_bps"]["count"] == 5
+
+
+# ═══════════════════════════════════════════════════════════
+# Trade tests
+# ═══════════════════════════════════════════════════════════
+
+class TestLeeReady:
+    def test_buy_above_mid(self):
+        assert classify_lee_ready(105.0, 100.0) == "buy"
+
+    def test_sell_below_mid(self):
+        assert classify_lee_ready(95.0, 100.0) == "sell"
+
+    def test_at_mid_with_ask(self):
+        assert classify_lee_ready(100.0, 100.0, bid_at_trade=99.0, ask_at_trade=100.0) == "buy"
+
+    def test_at_mid_with_bid(self):
+        assert classify_lee_ready(100.0, 100.0, bid_at_trade=100.0, ask_at_trade=101.0) == "sell"
+
+    def test_unknown_at_mid(self):
+        assert classify_lee_ready(100.0, 100.0) == "unknown"
+
+
+class TestBulkLeeReady:
+    def test_bulk_classification(self):
+        trades = [{"px": 105}, {"px": 95}, {"px": 100}]
+        mids = [100.0, 100.0, 100.0]
+        bids = [99.0, 99.0, 100.0]
+        asks = [101.0, 101.0, 101.0]
+        sides = classify_bulk_lee_ready(trades, mids, bids, asks)
+        assert sides == ["buy", "sell", "sell"]
+
+
+class TestMarkouts:
+    def test_basic_markout(self):
+        trades = [{"px": 100.0}, {"px": 101.0}]
+        mids = [100.0, 100.0, 100.1, 100.2]
+        times = [0, 100, 200, 300]
+        result = compute_markouts(trades, mids, times, horizons_ms=[100, 200])
+        assert "buy" in result
+        assert "sell" in result
+
+    def test_markout_summary(self):
+        markouts = {"buy": {100: [1.0, 2.0, -1.0]}, "sell": {100: [-1.0, -2.0]}}
+        summary = markout_summary(markouts)
+        assert summary["buy"][100]["count"] == 3
+        assert summary["sell"][100]["count"] == 2
+
+
+class TestTradeVolumeProfile:
+    def test_basic(self):
+        trades = [{"sz": 0.1}, {"sz": 0.2}, {"sz": 0.3}]
+        profile = trade_volume_profile(trades, n_buckets=3)
+        assert profile["buckets"]
+        assert sum(profile["counts"]) == 3
+
+
+class TestTradeArrivalRate:
+    def test_basic(self):
+        times = list(range(0, 60000, 1000))  # 1 trade/sec for 60 sec
+        result = trade_arrival_rate(times, window_sec=10)
+        assert result["mean_rate"] > 0
+
+
+# ═══════════════════════════════════════════════════════════
+# Toxicity tests
+# ═══════════════════════════════════════════════════════════
+
+class TestVPIN:
+    def test_balanced_volume(self):
+        buy_vol = [1.0] * 100
+        sell_vol = [1.0] * 100
+        result = compute_vpin(buy_vol, sell_vol)
+        assert result["vpin_value"] >= 0
+
+    def test_imbalanced_volume(self):
+        buy_vol = [2.0] * 200
+        sell_vol = [1.0] * 200
+        result = compute_vpin(buy_vol, sell_vol, volume_bucket_size=5.0, n_buckets=20)
+        assert result["vpin_value"] > 0  # buy > sell → imbalance > 0
+
+    def test_empty_returns_zero(self):
+        result = compute_vpin([], [])
+        assert result["vpin_value"] == 0
+
+
+class TestVPINTimeSeries:
+    def test_returns_series(self):
+        buy_vol = [1.0] * 200 + [3.0] * 50
+        sell_vol = [1.0] * 200 + [0.5] * 50
+        result = compute_vpin_time_series(buy_vol, sell_vol, n_buckets=20)
+        assert result["vpin_values"]
+        assert result["mean"] >= 0
+        assert result["max"] >= 0
+
+
+class TestFillToxicity:
+    def test_basic_toxicity(self):
+        prices = [100.0, 101.0, 102.0, 103.0, 104.0, 105.0]
+        mids = [100.0, 100.5, 101.0, 101.5, 102.0, 102.5]
+        sides = ["buy", "buy", "sell", "sell", "buy", "sell"]
+        result = fill_toxicity(prices, mids, sides, horizon_ticks=2)
+        assert "overall_bps" in result
+
+
+class TestAdverseSelection:
+    def test_basic(self):
+        mids_before = [100.0, 100.0, 100.0]
+        mids_after = [100.5, 99.5, 100.0]
+        sides = ["buy", "sell", "buy"]
+        result = adverse_selection_ratio(mids_after, mids_before, sides)
+        assert result["buy_adverse_bps"] != 0 or result["sell_adverse_bps"] != 0
+
+
+class TestLiquidationClustering:
+    def test_no_clusters(self):
+        times = list(range(0, 60000, 5000))  # spaced 5 sec apart
+        result = liquidation_clustering(times)
+        assert result["n_clusters"] == 0
+
+    def test_tight_clusters(self):
+        times = [0, 100, 200, 300, 400, 50000, 50100, 50200, 50300]
+        result = liquidation_clustering(times, window_sec=300)
+        assert result["n_clusters"] >= 0  # may or may not cluster depending on mean interval
+
+
+# ═══════════════════════════════════════════════════════════
+# Funding tests
+# ═══════════════════════════════════════════════════════════
+
+class TestFundingRegime:
+    def test_neutral(self):
+        rates = [0.000001] * 2000  # ~0.1% annual
+        result = funding_regime(rates, window_hours=24, n_samples_per_hour=60)
+        assert result["regime"] == "neutral"
+
+    def test_high_positive(self):
+        rates = [0.0001] * 2000  # ~11% annual
+        result = funding_regime(rates, window_hours=24, n_samples_per_hour=60)
+        assert result["regime"] in ("positive", "high_positive")
+
+
+class TestFundingPredictability:
+    def test_random(self):
+        np.random.seed(42)
+        rates = list(np.random.normal(0, 0.001, 100))
+        result = funding_predictability(rates)
+        assert len(result["autocorr"]) == 3
+
+
+class TestBasisSpread:
+    def test_premium(self):
+        perp = [105.0, 106.0, 107.0]
+        spot = [100.0, 101.0, 102.0]
+        result = basis_spread(perp, spot)
+        assert result["current_basis_bps"] > 0
+        assert result["mean_basis_bps"] > 0
+
+
+class TestBasisConvergence:
+    def test_mean_reverting(self):
+        basis = [50.0, 45.0, 40.0, 35.0, 30.0, 25.0] * 100  # decaying
+        result = basis_convergence_speed(basis)
+        assert result["ar1_coef"] > 0
+
+
+# ═══════════════════════════════════════════════════════════
+# Signals tests
+# ═══════════════════════════════════════════════════════════
+
+class TestCompositeSignal:
+    def test_strong_buy(self):
+        result = composite_signal(obi=0.5, trade_imbalance=0.3, vpin=0.1, funding_regime="negative")
+        assert result["signal"] == "buy"
+        assert result["confidence"] > 0
+
+    def test_strong_sell(self):
+        result = composite_signal(obi=-0.5, trade_imbalance=-0.3, vpin=0.1, funding_regime="high_positive")
+        assert result["signal"] == "sell"
+        assert result["confidence"] > 0
+
+    def test_neutral(self):
+        result = composite_signal(obi=0.05, trade_imbalance=0.0, vpin=0.0)
+        assert result["signal"] == "neutral"
+        assert result["confidence"] >= 0
+
+    def test_toxic_reduces_confidence(self):
+        norm = composite_signal(obi=0.4, trade_imbalance=0.3, vpin=0.1)
+        toxic = composite_signal(obi=0.4, trade_imbalance=0.3, vpin=0.8)
+        assert toxic["confidence"] < norm["confidence"]
+
+    def test_wide_spread_reduces_confidence(self):
+        tight = composite_signal(obi=0.4, trade_imbalance=0.3, spread_bps=1.0)
+        wide = composite_signal(obi=0.4, trade_imbalance=0.3, spread_bps=30.0)
+        assert wide["confidence"] < tight["confidence"]
+
+
+class TestDetectHFTRegime:
+    def test_toxic(self):
+        assert detect_hft_regime(obi_std=0.1, spread_mean_bps=15.0, trade_rate_per_sec=1.0, vpin=0.5) == "toxic"
+
+    def test_quiet(self):
+        assert detect_hft_regime(obi_std=0.1, spread_mean_bps=2.0, trade_rate_per_sec=0.05, vpin=0.1) == "quiet"
+
+    def test_trending(self):
+        assert detect_hft_regime(obi_std=0.4, spread_mean_bps=3.0, trade_rate_per_sec=2.0, vpin=0.1) == "trending"
+
+    def test_ranging(self):
+        assert detect_hft_regime(obi_std=0.1, spread_mean_bps=2.0, trade_rate_per_sec=2.0, vpin=0.1) == "ranging"
+
+
+class TestSignalPipeline:
+    def test_emit_no_data(self):
+        p = SignalPipeline()
+        result = p.emit()
+        assert "signal" in result
+        assert result["signal"] == "neutral"
+
+    def test_update_and_emit(self):
+        p = SignalPipeline(obi_window=10, trade_window=10, vpin_volume_size=1.0, vpin_buckets=5)
+        bids = {100.0: 5.0}
+        asks = {102.0: 1.0}
+        p.update_book(bids, asks)
+        for _ in range(3):
+            p.update_trade(101.5, 0.1, 101.0)
+        for _ in range(1):
+            p.update_trade(100.5, 0.1, 101.0)
+        result = p.emit()
+        assert result["signal"] in ("buy", "sell", "neutral")
@@ -0,0 +1,123 @@
+"""
+Tests for data/normalizer.py — timestamp normalization and sequence gaps.
+"""
+import time
+
+from data.normalizer import (
+    normalize_timestamp,
+    SequenceTracker,
+    detect_sequence_gap,
+)
+
+
+def test_normalize_ms_timestamp():
+    """Millisecond timestamps pass through unchanged."""
+    ts = 1705312800000
+    assert normalize_timestamp(ts) == 1705312800000
+
+
+def test_normalize_seconds_timestamp():
+    """Second timestamps get multiplied by 1000."""
+    ts = 1705312800
+    result = normalize_timestamp(ts)
+    assert result == 1705312800000
+
+
+def test_normalize_float_timestamp():
+    """Float seconds get multiplied."""
+    ts = 1705312800.5
+    result = normalize_timestamp(ts)
+    assert result == 1705312800500
+
+
+def test_normalize_large_float_is_ms():
+    """A float > 1e12 is already in ms."""
+    ts = 1705312800000.123
+    result = normalize_timestamp(ts)
+    assert result == 1705312800000
+
+
+def test_normalize_iso_string():
+    """ISO 8601 Z string converted to ms."""
+    result = normalize_timestamp("2024-01-15T12:00:00.000Z")
+    assert result == 1705320000000
+
+
+def test_normalize_iso_no_z():
+    """ISO string without trailing Z."""
+    result = normalize_timestamp("2024-01-15T12:00:00")
+    expected = 1705320000000
+    assert abs(result - expected) < 1000
+
+
+def test_normalize_none_returns_now():
+    """None returns current time (within 1s)."""
+    now_ms = int(time.time() * 1000)
+    result = normalize_timestamp(None)
+    assert abs(result - now_ms) < 2000
+
+
+def test_normalize_datetime():
+    """datetime object converted to ms."""
+    from datetime import datetime
+    dt = datetime(2024, 1, 15, 12, 0, 0)
+    result = normalize_timestamp(dt)
+    assert result == 1705320000000
+
+
+class TestSequenceTracker:
+    def test_initial_no_gap(self):
+        st = SequenceTracker()
+        assert st.check("l2", "BTC", 100) is None
+
+    def test_consecutive_no_gap(self):
+        st = SequenceTracker()
+        st.check("l2", "BTC", 100)
+        assert st.check("l2", "BTC", 101) is None
+        assert st.check("l2", "BTC", 102) is None
+
+    def test_gap_detected(self):
+        st = SequenceTracker()
+        st.check("l2", "BTC", 100)
+        gap = st.check("l2", "BTC", 105)
+        assert gap is not None
+        assert gap["gap_size"] == 4
+        assert gap["expected"] == 101
+
+    def test_multiple_channels_independent(self):
+        st = SequenceTracker()
+        st.check("l2", "BTC", 100)
+        st.check("trades", "BTC", 50)
+        assert st.check("l2", "BTC", 101) is None
+        assert st.check("trades", "BTC", 51) is None
+
+    def test_reset_clears_state(self):
+        st = SequenceTracker()
+        st.check("l2", "BTC", 100)
+        st.reset("l2", "BTC")
+        assert st.check("l2", "BTC", 200) is None  # fresh start
+
+    def test_gap_counts_accumulate(self):
+        st = SequenceTracker()
+        st.check("l2", "BTC", 100)
+        st.check("l2", "BTC", 105)
+        st.check("l2", "BTC", 110)
+        gaps = st.gap_counts
+        assert gaps["l2:BTC"] == 8  # 4 + 4
+
+
+def test_detect_sequence_gap_empty():
+    assert detect_sequence_gap(1, None) == 0
+
+
+def test_detect_sequence_gap_ok():
+    assert detect_sequence_gap(102, 101) == 0
+
+
+def test_detect_sequence_gap_found():
+    gap = detect_sequence_gap(105, 101)
+    assert gap == 3
+
+
+def test_detect_sequence_gap_negative():
+    assert detect_sequence_gap(100, 101) == -1  # dupe or reset
@@ -0,0 +1,110 @@
+"""
+Tests for data/store.py — Parquet-based raw message storage.
+"""
+import json
+import os
+import tempfile
+import time
+
+from data.store import RawMessageStore, read_range
+
+
+def _make_parquet_dir():
+    """Create a temporary directory for parquet storage."""
+    return tempfile.mkdtemp()
+
+
+def test_store_basic_write_read():
+    """Write messages, flush, read them back."""
+    tmpdir = tempfile.mkdtemp()
+    store = RawMessageStore(data_dir=tmpdir, flush_interval_sec=0.5)
+    store.start()
+
+    for i in range(10):
+        store.push(
+            channel="l2book",
+            coin="BTC",
+            exchange_ts=1718000000000 + i * 1000,
+            payload={"type": "snapshot", "levels": [[{"px": "50000", "sz": "1.0"}], [{"px": "50001", "sz": "0.5"}]]},
+        )
+
+    time.sleep(1.5)
+    store.stop()
+    assert store.total_written == 10
+
+    from datetime import date
+    today = date.today().isoformat()
+    rows = read_range(tmpdir, "l2book", "BTC", today, today)
+    assert len(rows) == 10
+    assert rows[0]["payload"]["type"] == "snapshot"
+    assert rows[0]["exchange_ts"] == 1718000000000
+    assert rows[-1]["exchange_ts"] == 1718000009000
+
+
+def test_store_multiple_channels():
+    """Write to different channels and verify partitioning."""
+    tmpdir = tempfile.mkdtemp()
+    store = RawMessageStore(data_dir=tmpdir, flush_interval_sec=0.5)
+    store.start()
+
+    channels = ["l2book", "trades", "funding", "mark", "open_interest"]
+    for ch in channels:
+        for i in range(3):
+            store.push(channel=ch, coin="BTC", exchange_ts=1718000000000 + i * 1000, payload={"ch": ch, "i": i})
+
+    time.sleep(1.5)
+    store.stop()
+    assert store.total_written == 15
+
+    from datetime import date
+    today = date.today().isoformat()
+    for ch in channels:
+        rows = read_range(tmpdir, ch, "BTC", today, today)
+        assert len(rows) == 3, f"Expected 3 rows for {ch}, got {len(rows)}"
+
+
+def test_store_append_to_existing():
+    """Write in two batches to same file — should append."""
+    tmpdir = tempfile.mkdtemp()
+    store = RawMessageStore(data_dir=tmpdir, flush_interval_sec=0.3)
+    store.start()
+
+    for i in range(5):
+        store.push(channel="trades", coin="ETH", exchange_ts=1718000000000 + i * 1000, payload={"batch": 1, "i": i})
+
+    time.sleep(1)
+    store.stop()
+
+    store2 = RawMessageStore(data_dir=tmpdir, flush_interval_sec=0.3)
+    store2.start()
+    for i in range(5):
+        store2.push(channel="trades", coin="ETH", exchange_ts=1718000005000 + i * 1000, payload={"batch": 2, "i": i})
+    time.sleep(1)
+    store2.stop()
+
+    from datetime import date
+    today = date.today().isoformat()
+    rows = read_range(tmpdir, "trades", "ETH", today, today)
+    assert len(rows) == 10
+    batches = [r["payload"]["batch"] for r in rows]
+    assert batches == [1] * 5 + [2] * 5
+
+
+def test_store_queue_full_does_not_crash():
+    """Small queue — push many, ensure no crash."""
+    tmpdir = tempfile.mkdtemp()
+    store = RawMessageStore(data_dir=tmpdir, flush_interval_sec=60.0, max_queue_size=10)
+    store.start()
+
+    for i in range(1000):
+        store.push(channel="trades", coin="BTC", exchange_ts=1718000000000 + i, payload={"i": i})
+
+    store.stop()
+    assert store.total_written >= 0  # some may be lost, but no crash
+
+
+def test_read_range_empty():
+    """Read a date range with no data returns empty list."""
+    tmpdir = tempfile.mkdtemp()
+    rows = read_range(tmpdir, "nonexistent", "BTC", "2020-01-01", "2020-01-02")
+    assert rows == []