kingchenc
cb6da4d737
* feat(data-layer): Resampler (candle resampling) in all 10 languages Second data-layer feature (F3): resample candles into a higher timeframe. - Native (Node.js/WASM): new Resampler(timeframe) -> update(o,h,l,c,v,ts): Candle|null + flush(): Candle|null. Python the same -> tuple|None. - C ABI: wickra_resampler_new/update/flush/free (update has the multi-output shape so the generators auto-emit it; flush is bespoke). Go Update -> (Candle, bool) + Flush; C# Candle? Update/Flush; Java Candle update/flush; R update() generic + a flush() S3 method (extends base::flush); C/C++ direct. - Cross-language golden (testdata/golden/data_resampled.csv): the shared input candles resampled into 5-unit buckets, the final partial bucket via flush, pinned bit-for-bit across every binding. Verified locally in all 10 (3 candles for the 5-unit smoke; 16 for the golden). The WickraCandle output record is shared with the tick aggregator (deduped). * test(node): exclude data-layer types from the indicator completeness contract The Resampler exposes update(), so the completeness test flagged it as an indicator and required batch/reset/isReady/warmupPeriod, which a data-layer type does not have. Exclude TickAggregator and Resampler like the bar builders.
66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""Cross-language data-layer parity for the Python binding: replay the shared
|
|
golden tick stream through the TickAggregator and check the candles against the
|
|
Rust-generated fixtures, with and without gap filling. Fixtures are produced by
|
|
``cargo run -p wickra-examples --bin gen_golden``.
|
|
"""
|
|
import csv
|
|
import os
|
|
|
|
import pytest
|
|
import wickra as ta
|
|
|
|
HERE = os.path.dirname(__file__)
|
|
GOLDEN = os.path.normpath(os.path.join(HERE, "..", "..", "..", "testdata", "golden"))
|
|
|
|
|
|
def _read(name):
|
|
with open(os.path.join(GOLDEN, name + ".csv"), newline="") as f:
|
|
rows = list(csv.reader(f))
|
|
return [[float(x) for x in r] for r in rows[1:] if r]
|
|
|
|
|
|
TICKS = _read("data_ticks")
|
|
|
|
|
|
def _run(gap_fill):
|
|
agg = ta.TickAggregator(1000, gap_fill=gap_fill)
|
|
out = []
|
|
for price, size, ts in TICKS:
|
|
out.extend(agg.push(price, size, int(ts)))
|
|
return out
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"gap_fill,fixture",
|
|
[(False, "data_candles"), (True, "data_candles_gap")],
|
|
)
|
|
def test_tick_aggregator_matches_golden(gap_fill, fixture):
|
|
got = _run(gap_fill)
|
|
want = _read(fixture)
|
|
assert len(got) == len(want)
|
|
for i, (g, w) in enumerate(zip(got, want)):
|
|
for j in range(6):
|
|
tol = 1e-9 * max(1.0, abs(w[j]))
|
|
assert abs(g[j] - w[j]) <= tol, f"row {i} col {j}: {g[j]} vs {w[j]}"
|
|
|
|
|
|
INPUT = _read("input") # open,high,low,close,volume (timestamp = row index)
|
|
|
|
|
|
def test_resampler_matches_golden():
|
|
r = ta.Resampler(5)
|
|
got = []
|
|
for i, (o, h, l, c, v) in enumerate(INPUT):
|
|
candle = r.update(o, h, l, c, v, i)
|
|
if candle is not None:
|
|
got.append(candle)
|
|
f = r.flush()
|
|
if f is not None:
|
|
got.append(f)
|
|
want = _read("data_resampled")
|
|
assert len(got) == len(want)
|
|
for i, (g, w) in enumerate(zip(got, want)):
|
|
for j in range(6):
|
|
tol = 1e-9 * max(1.0, abs(w[j]))
|
|
assert abs(g[j] - w[j]) <= tol, f"row {i} col {j}: {g[j]} vs {w[j]}"
|