danielmjrr
160ed2f948
Memory-mapped tickstores, a nogil numba replay kernel validated trade-for-trade against an independent reference implementation, purged walk-forward CV, an append-only FDR registry, tick-exact labels, and online changepoint detection. 129 tests, 4 runnable examples on synthetic data, CI for linux/macos.
195 lines
8.1 KiB
Python
195 lines
8.1 KiB
Python
"""TickStore round-trips and bar derivation against a brute-force pandas oracle."""
|
|
import json
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from tickreplay.store import TIMEFRAMES, TickStore, derive_bars
|
|
|
|
|
|
def _random_ticks(n=4000, seed=7, start="2024-01-02T00:00:00", max_gap_s=600):
|
|
"""Irregularly spaced ticks: gaps up to max_gap_s leave many buckets empty."""
|
|
rng = np.random.default_rng(seed)
|
|
t0 = np.datetime64(start, "ns").astype("int64")
|
|
time_ns = (t0 + np.cumsum(rng.integers(1, max_gap_s * 10**9, size=n))).astype("int64")
|
|
bid = 2000.0 + np.cumsum(rng.normal(0.0, 0.1, n))
|
|
ask = bid + rng.uniform(0.01, 0.05, n)
|
|
return time_ns, bid, ask
|
|
|
|
|
|
def _oracle_bars(time_ns, bid, ask, timeframe):
|
|
tf_ns = TIMEFRAMES[timeframe] * 10**9
|
|
df = pd.DataFrame({"bucket": time_ns // tf_ns, "bid": bid, "spread": ask - bid})
|
|
out = df.groupby("bucket", sort=True).agg(
|
|
open=("bid", "first"), high=("bid", "max"), low=("bid", "min"),
|
|
close=("bid", "last"), spread_mean=("spread", "mean"), tick_count=("bid", "size"))
|
|
out.insert(0, "time", pd.to_datetime(out.index.to_numpy() * tf_ns, unit="ns"))
|
|
return out.reset_index(drop=True)
|
|
|
|
|
|
def _month_frame(start, n, seed, max_gap_s=3600):
|
|
rng = np.random.default_rng(seed)
|
|
t0 = np.datetime64(start, "ns").astype("int64")
|
|
times = t0 + np.cumsum(rng.integers(1, max_gap_s * 10**9, size=n))
|
|
bid = 2000.0 + rng.normal(0.0, 1.0, n)
|
|
return pd.DataFrame({"time": pd.to_datetime(times, unit="ns"),
|
|
"bid": bid, "ask": bid + 0.03})
|
|
|
|
|
|
@pytest.mark.parametrize("timeframe", ["M1", "M5", "M15", "M30", "H1", "H4", "D1"])
|
|
def test_derive_bars_matches_pandas_oracle(timeframe):
|
|
time_ns, bid, ask = _random_ticks()
|
|
bars = derive_bars(time_ns, bid, ask, timeframe)
|
|
oracle = _oracle_bars(time_ns, bid, ask, timeframe)
|
|
|
|
assert len(oracle) > 1
|
|
assert len(bars) == len(oracle)
|
|
assert np.array_equal(bars["time"].to_numpy(), oracle["time"].to_numpy())
|
|
for col in ("open", "high", "low", "close"):
|
|
assert np.array_equal(bars[col].to_numpy(), oracle[col].to_numpy())
|
|
np.testing.assert_allclose(bars["spread_mean"].to_numpy(),
|
|
oracle["spread_mean"].to_numpy(), rtol=1e-12)
|
|
assert np.array_equal(bars["tick_count"].to_numpy(), oracle["tick_count"].to_numpy())
|
|
|
|
|
|
def test_derive_bars_emits_no_bar_for_empty_buckets():
|
|
time_ns, bid, ask = _random_ticks()
|
|
tf_ns = TIMEFRAMES["M1"] * 10**9
|
|
bars = derive_bars(time_ns, bid, ask, "M1")
|
|
|
|
spanned = int(time_ns[-1] // tf_ns - time_ns[0] // tf_ns) + 1
|
|
assert len(bars) < spanned
|
|
occupied = np.unique(time_ns // tf_ns)
|
|
assert np.array_equal(bars["time"].to_numpy().astype("int64") // tf_ns, occupied)
|
|
|
|
|
|
def test_bar_times_are_epoch_aligned():
|
|
time_ns, bid, ask = _random_ticks()
|
|
for timeframe, seconds in TIMEFRAMES.items():
|
|
times = derive_bars(time_ns, bid, ask, timeframe)["time"]
|
|
assert (times.to_numpy().astype("int64") % (seconds * 10**9) == 0).all()
|
|
|
|
h4 = derive_bars(time_ns, bid, ask, "H4")["time"].dt
|
|
assert set(h4.hour.unique()) == {0, 4, 8, 12, 16, 20}
|
|
assert (h4.minute == 0).all()
|
|
assert (h4.second == 0).all()
|
|
|
|
d1 = derive_bars(time_ns, bid, ask, "D1")["time"]
|
|
assert len(d1) > 1
|
|
assert (d1 == d1.dt.normalize()).all()
|
|
|
|
|
|
@pytest.mark.parametrize("timeframe", ["M1", "M15", "H4"])
|
|
def test_tick_spans_partition_every_tick(timeframe):
|
|
time_ns, bid, ask = _random_ticks()
|
|
bars = derive_bars(time_ns, bid, ask, timeframe)
|
|
first = bars["tick_first"].to_numpy()
|
|
last = bars["tick_last"].to_numpy()
|
|
count = bars["tick_count"].to_numpy()
|
|
|
|
assert first[0] == 0
|
|
assert last[-1] == len(time_ns) - 1
|
|
assert (last >= first).all()
|
|
assert np.array_equal(first[1:], last[:-1] + 1)
|
|
assert np.array_equal(count, last - first + 1)
|
|
assert count.sum() == len(time_ns)
|
|
|
|
bucket = time_ns // (TIMEFRAMES[timeframe] * 10**9)
|
|
assert np.array_equal(bucket[first], bucket[last])
|
|
assert np.array_equal(bars["time"].to_numpy().astype("int64"),
|
|
bucket[first] * TIMEFRAMES[timeframe] * 10**9)
|
|
|
|
|
|
def test_write_open_roundtrip(tmp_path):
|
|
time_ns, bid, ask = _random_ticks(n=500, seed=1)
|
|
store_dir = tmp_path / "store"
|
|
written = TickStore.write(store_dir, time_ns, bid, ask, {"symbol": "XAUUSD"})
|
|
expected_range = [str(np.datetime64(int(time_ns[0]), "ns")),
|
|
str(np.datetime64(int(time_ns[-1]), "ns"))]
|
|
|
|
for store in (written, TickStore.open(store_dir)):
|
|
assert len(store) == len(time_ns)
|
|
assert np.array_equal(np.asarray(store.time_ns), time_ns)
|
|
assert np.array_equal(np.asarray(store.bid), bid)
|
|
assert np.array_equal(np.asarray(store.ask), ask)
|
|
assert store.manifest["n_ticks"] == len(time_ns)
|
|
assert store.manifest["time_range"] == expected_range
|
|
assert store.manifest["symbol"] == "XAUUSD"
|
|
|
|
assert json.loads((store_dir / "manifest.json").read_text()) == written.manifest
|
|
|
|
|
|
def test_write_rejects_non_monotonic_time(tmp_path):
|
|
backwards = np.array([10, 30, 20], dtype="int64") * 10**9
|
|
with pytest.raises(ValueError, match="non-monotonic"):
|
|
TickStore.write(tmp_path / "bad", backwards, [1.0, 2.0, 3.0], [1.1, 2.1, 3.1])
|
|
|
|
repeated = np.array([10, 20, 20, 30], dtype="int64") * 10**9
|
|
store = TickStore.write(tmp_path / "ok", repeated, [1.0] * 4, [1.1] * 4)
|
|
assert len(store) == 4
|
|
|
|
|
|
def test_build_concatenates_monthly_parquet_files(tmp_path):
|
|
months = {"2024-01": _month_frame("2024-01-02T00:00:00", 50, 0),
|
|
"2024-02": _month_frame("2024-02-01T00:00:00", 40, 1),
|
|
"2024-03": _month_frame("2024-03-01T00:00:00", 30, 2)}
|
|
parquet_dir = tmp_path / "parquet"
|
|
parquet_dir.mkdir()
|
|
for month, frame in months.items():
|
|
frame.to_parquet(parquet_dir / f"ticks_{month}.parquet", index=False)
|
|
|
|
store = TickStore.build(parquet_dir, tmp_path / "store")
|
|
expected = pd.concat(months.values(), ignore_index=True)
|
|
|
|
assert len(store) == len(expected) == 120
|
|
assert np.array_equal(np.asarray(store.time_ns),
|
|
expected["time"].to_numpy().astype("datetime64[ns]").astype("int64"))
|
|
assert np.array_equal(np.asarray(store.bid), expected["bid"].to_numpy())
|
|
assert np.array_equal(np.asarray(store.ask), expected["ask"].to_numpy())
|
|
assert store.manifest["n_ticks"] == 120
|
|
assert store.manifest["sources"] == ["ticks_2024-01.parquet", "ticks_2024-02.parquet",
|
|
"ticks_2024-03.parquet"]
|
|
assert store.manifest["time_range"][0].startswith("2024-01")
|
|
assert store.manifest["time_range"][1].startswith("2024-03")
|
|
assert TickStore.open(tmp_path / "store").manifest == store.manifest
|
|
|
|
|
|
def test_build_rejects_non_monotonic_time_within_a_file(tmp_path):
|
|
frame = _month_frame("2024-01-02T00:00:00", 20, 0)
|
|
frame.loc[10, "time"] = frame.loc[2, "time"]
|
|
parquet_dir = tmp_path / "parquet"
|
|
parquet_dir.mkdir()
|
|
frame.to_parquet(parquet_dir / "ticks_2024-01.parquet", index=False)
|
|
|
|
with pytest.raises(ValueError, match="non-monotonic"):
|
|
TickStore.build(parquet_dir, tmp_path / "store")
|
|
|
|
|
|
def test_build_rejects_time_going_backwards_across_files(tmp_path):
|
|
parquet_dir = tmp_path / "parquet"
|
|
parquet_dir.mkdir()
|
|
_month_frame("2024-02-01T00:00:00", 20, 0).to_parquet(
|
|
parquet_dir / "ticks_2024-01.parquet", index=False)
|
|
_month_frame("2024-01-02T00:00:00", 20, 1).to_parquet(
|
|
parquet_dir / "ticks_2024-02.parquet", index=False)
|
|
|
|
with pytest.raises(ValueError, match="non-monotonic"):
|
|
TickStore.build(parquet_dir, tmp_path / "store")
|
|
|
|
|
|
def test_build_requires_at_least_one_parquet_file(tmp_path):
|
|
(tmp_path / "empty").mkdir()
|
|
with pytest.raises(ValueError, match="no ticks_"):
|
|
TickStore.build(tmp_path / "empty", tmp_path / "store")
|
|
|
|
|
|
def test_bars_are_cached_per_timeframe(tmp_path):
|
|
time_ns, bid, ask = _random_ticks(n=800, seed=3)
|
|
store = TickStore.write(tmp_path / "store", time_ns, bid, ask)
|
|
|
|
m5 = store.bars("M5")
|
|
assert store.bars("M5") is m5
|
|
assert store.bars("M15") is not m5
|
|
pd.testing.assert_frame_equal(m5, derive_bars(time_ns, bid, ask, "M5"))
|