From b010b73d60af22e82ccc13388f79bd6b39c80dd8 Mon Sep 17 00:00:00 2001 From: luisleo526 Date: Thu, 16 Jul 2026 06:37:56 +0800 Subject: [PATCH] Derive chart opens from first traded minute --- README.md | 4 +- scripts/derive_corpus_feeds.py | 151 +++++++++++++--- scripts/derive_corpus_feeds_self_test.py | 213 +++++++++++++++++++++++ tests/CMakeLists.txt | 7 + 4 files changed, 345 insertions(+), 30 deletions(-) create mode 100644 scripts/derive_corpus_feeds_self_test.py diff --git a/README.md b/README.md index 2cfb740..6df8ef6 100644 --- a/README.md +++ b/README.md @@ -228,6 +228,8 @@ If you encounter day-boundary alignment issues or want to force the engine to pr - CMake ≥ 3.16 - A C++17 compiler (GCC ≥ 9, Clang ≥ 10, Apple Clang ≥ 12) - [Eigen 3.3+](https://eigen.tuxfamily.org/) — used for matrix-typed PineScript. The build will fetch Eigen via CMake `FetchContent` if no system install is found. +- Python 3 when the test suite is enabled (the default). Library-only builds + can pass `-DPINEFORGE_BUILD_TESTS=OFF`. ### Build + test @@ -237,7 +239,7 @@ cmake --build build -j ctest --test-dir build --output-on-failure ``` -Expect 39 tests to pass. The largest (`test_integration`, `test_request_security`) take a few hundred milliseconds; everything else completes faster. +Expect 105 tests to pass. The largest (`test_integration`, `test_request_security`) take a few hundred milliseconds; everything else completes faster. ### Install diff --git a/scripts/derive_corpus_feeds.py b/scripts/derive_corpus_feeds.py index 22c416e..7bed8e0 100644 --- a/scripts/derive_corpus_feeds.py +++ b/scripts/derive_corpus_feeds.py @@ -18,11 +18,16 @@ comparable), and the harness's window-bounds fallback -Resample rule: open=first, high=max, low=min, close=last, volume=sum -(rounded to 6dp), timestamp=bucket start; a trailing partial bucket is -kept. Idempotent: files are only rewritten when missing or older than -the source feed. Import ensure_derived() or run as a script. +Resample rule: open=first positive-volume row (or first row when the +bucket is entirely zero-volume), high=max, low=min, close=last, +volume=sum (rounded to 6dp), timestamp=bucket start; a trailing partial +bucket is kept. Idempotent: files are only rewritten when missing, +older than the source feed, or stamped with an older derivation rule. +Import ensure_derived() or run as a script. """ +import math +import os +import tempfile from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent @@ -30,6 +35,11 @@ DERIVED_DIR = REPO_ROOT / "corpus" / "data" / "derived" DERIVED_15M = DERIVED_DIR / "ohlcv_ETH-USDT-USDT_15m.csv" DERIVED_15M_WINDOW = DERIVED_DIR / "ohlcv_ETH-USDT-USDT_15m_window.csv" +DERIVATION_STAMP = DERIVED_DIR / ".derive_corpus_feeds.version" + +# Bump whenever the materialization semantics change. This invalidates +# already-generated feeds even when the committed 1m source is unchanged. +DERIVATION_CACHE_VERSION = "v2:first-positive-volume-open" # Comparison window (epoch ms, inclusive), pinned from the historical # window-only 15m feed the corpus used to ship: first bar 2025-04-20 @@ -53,6 +63,105 @@ def _stale(target: Path, source: Path) -> bool: or target.stat().st_mtime < source.stat().st_mtime) +def _derived_cache_is_fresh(source: Path, targets, stamp: Path) -> bool: + """Return whether outputs match both the source and derivation rule.""" + if any(_stale(target, source) for target in targets): + return False + try: + return stamp.read_text().strip() == DERIVATION_CACHE_VERSION + except OSError: + return False + + +def _parse_ohlcv_row(line: str, row_number: int): + fields = line.rstrip("\r\n").split(",") + if len(fields) != 6: + raise ValueError( + f"row {row_number}: expected 6 comma-separated fields, " + f"got {len(fields)}") + + ts_s, o, h, l, c, v = fields + try: + ts = int(ts_s) + except ValueError as exc: + raise ValueError( + f"row {row_number}: invalid timestamp {ts_s!r}") from exc + + values = {} + for name, text in (("open", o), ("high", h), ("low", l), + ("close", c), ("volume", v)): + try: + value = float(text) + except ValueError as exc: + raise ValueError( + f"row {row_number}: invalid {name} {text!r}") from exc + if not math.isfinite(value): + raise ValueError( + f"row {row_number}: {name} must be finite, got {text!r}") + values[name] = value + + if values["volume"] < 0.0: + raise ValueError( + f"row {row_number}: volume must be non-negative, got {v!r}") + return (ts, o, values["high"], values["low"], c, + values["volume"]) + + +def _resample_15m(lines): + """Aggregate header-free 1m CSV rows into 15m buckets.""" + buckets = [] # [ts, o, h, l, c, v] per 900s bucket, in order + cur_key = None + has_positive_volume = False + for row_number, line in enumerate(lines, start=2): + ts, o, hf, lf, c, vf = _parse_ohlcv_row(line, row_number) + key = ts - ts % 900_000 + if key != cur_key: + buckets.append([key, o, hf, lf, c, vf]) + cur_key = key + has_positive_volume = vf > 0.0 + else: + b = buckets[-1] + if not has_positive_volume and vf > 0.0: + b[1] = o + has_positive_volume = True + if hf > b[2]: + b[2] = hf + if lf < b[3]: + b[3] = lf + b[4] = c + b[5] += vf + return buckets + + +def _format_bucket(bucket) -> str: + ts, o, h, l, c, v = bucket + return f"{ts},{o},{_fmt(h)},{_fmt(l)},{c},{_fmt(round(v, 6))}" + + +def _write_atomic(path: Path, text: str) -> None: + # Every checkout rebuilds once when DERIVATION_CACHE_VERSION changes. Use + # a private same-directory temporary so two independently launched + # validators cannot race over one fixed ``.new`` path. Same-directory + # replacement preserves the atomic rename guarantee. + fd, tmp_name = tempfile.mkstemp( + dir=path.parent, + prefix=f".{path.name}.", + suffix=".new", + text=True, + ) + tmp = Path(tmp_name) + try: + with os.fdopen(fd, "w") as fh: + fh.write(text) + os.chmod(tmp, path.stat().st_mode & 0o777 if path.exists() else 0o644) + tmp.replace(path) + finally: + try: + tmp.unlink() + except FileNotFoundError: + pass + + def ensure_derived(verbose: bool = False) -> None: """Create/refresh the derived feeds. Cheap no-op when up to date.""" if not SOURCE_1M.exists(): @@ -65,49 +174,33 @@ def ensure_derived(verbose: bool = False) -> None: f"{SOURCE_1M} is suspiciously small — likely an unsmudged " "Git LFS pointer. Run: git lfs install && git lfs pull " "(inside the corpus submodule).") - if not (_stale(DERIVED_15M, SOURCE_1M) - or _stale(DERIVED_15M_WINDOW, SOURCE_1M)): + outputs = (DERIVED_15M, DERIVED_15M_WINDOW) + if _derived_cache_is_fresh(SOURCE_1M, outputs, DERIVATION_STAMP): return if verbose: print(f"[derive] resampling {SOURCE_1M.name} -> 15m ...") - buckets = [] # [ts, o, h, l, c, v] per 900s bucket, in order - cur_key = None with SOURCE_1M.open() as fh: next(fh) # header - for line in fh: - ts_s, o, h, l, c, v = line.rstrip("\n").split(",") - ts = int(ts_s) - key = ts - ts % 900_000 - if key != cur_key: - buckets.append([key, o, float(h), float(l), c, float(v)]) - cur_key = key - else: - b = buckets[-1] - hf, lf = float(h), float(l) - if hf > b[2]: - b[2] = hf - if lf < b[3]: - b[3] = lf - b[4] = c - b[5] += float(v) + buckets = _resample_15m(fh) DERIVED_DIR.mkdir(parents=True, exist_ok=True) full_lines = [HEADER] window_lines = [HEADER] - for ts, o, h, l, c, v in buckets: - row = f"{ts},{o},{_fmt(h)},{_fmt(l)},{c},{_fmt(round(v, 6))}" + for bucket in buckets: + row = _format_bucket(bucket) full_lines.append(row) + ts = bucket[0] if WINDOW_START_MS <= ts <= WINDOW_END_MS: window_lines.append(row) for path, lines in ((DERIVED_15M, full_lines), (DERIVED_15M_WINDOW, window_lines)): - tmp = path.with_suffix(".csv.new") - tmp.write_text("\n".join(lines) + "\n") - tmp.replace(path) + _write_atomic(path, "\n".join(lines) + "\n") if verbose: print(f"[derive] wrote {path.relative_to(REPO_ROOT)} " f"({len(lines) - 1} bars)") + # Publish the cache version only after both feed replacements succeed. + _write_atomic(DERIVATION_STAMP, DERIVATION_CACHE_VERSION + "\n") if __name__ == "__main__": diff --git a/scripts/derive_corpus_feeds_self_test.py b/scripts/derive_corpus_feeds_self_test.py new file mode 100644 index 0000000..a83b0e7 --- /dev/null +++ b/scripts/derive_corpus_feeds_self_test.py @@ -0,0 +1,213 @@ +#!/usr/bin/env python3 +"""Focused regression tests for deterministic corpus-feed derivation.""" +from __future__ import annotations + +import hashlib +import os +import tempfile +import threading +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from unittest import mock + +import derive_corpus_feeds as derive + + +def _legacy_resample_15m(lines: list[str]): + """Previous open=first algorithm, used only for no-prefix equality.""" + buckets = [] + cur_key = None + for line in lines: + ts_s, o, h, l, c, v = line.rstrip("\n").split(",") + ts = int(ts_s) + key = ts - ts % 900_000 + if key != cur_key: + buckets.append([key, o, float(h), float(l), c, float(v)]) + cur_key = key + else: + bucket = buckets[-1] + bucket[2] = max(bucket[2], float(h)) + bucket[3] = min(bucket[3], float(l)) + bucket[4] = c + bucket[5] += float(v) + return buckets + + +class ResampleOpenTests(unittest.TestCase): + def test_positive_first_row_is_unchanged(self) -> None: + lines = [ + "60000,100,102,99,101,2\n", + "120000,200,203,98,202,3\n", + ] + bucket = derive._resample_15m(lines)[0] + self.assertEqual(bucket, [0, "100", 203.0, 98.0, "202", 5.0]) + + def test_zero_volume_lower_prefix_uses_first_real_open(self) -> None: + lines = [ + "0,90,91,89,90,0\n", + "60000,100,101,99,100,4\n", + "120000,105,106,104,105,2\n", + ] + bucket = derive._resample_15m(lines)[0] + self.assertEqual(bucket, [0, "100", 106.0, 89.0, "105", 6.0]) + + def test_zero_volume_higher_prefix_uses_first_real_open(self) -> None: + lines = [ + "0,110,111,109,110,0\n", + "60000,100,101,99,100,4\n", + "120000,95,96,94,95,0\n", + ] + bucket = derive._resample_15m(lines)[0] + self.assertEqual(bucket, [0, "100", 111.0, 94.0, "95", 4.0]) + + def test_all_zero_bucket_falls_back_to_first_row(self) -> None: + lines = [ + "0,90,91,89,90,0\n", + "60000,110,111,109,110,0\n", + ] + bucket = derive._resample_15m(lines)[0] + self.assertEqual(bucket, [0, "90", 111.0, 89.0, "110", 0.0]) + + def test_no_synthetic_prefix_is_exactly_legacy_equal(self) -> None: + lines = [ + "0,100.125,102,99,101,1.25\n", + "60000,101,103,98,102,0\n", + "900000,200,201,199,200.5,2.125\n", + "960000,201,202,198,199.5,3.25\n", + ] + actual = derive._resample_15m(lines) + expected = _legacy_resample_15m(lines) + self.assertEqual(actual, expected) + self.assertEqual( + [derive._format_bucket(bucket) for bucket in actual], + [ + "0,100.125,103,98,102,1.25", + "900000,200,202,198,199.5,5.375", + ], + ) + + +class InputPolicyTests(unittest.TestCase): + def test_malformed_and_nonfinite_rows_fail_closed(self) -> None: + cases = { + "field count": ("0,1,2\n", "expected 6"), + "timestamp": ("bad,1,1,1,1,1\n", "invalid timestamp"), + "volume text": ("0,1,1,1,1,nope\n", "invalid volume"), + "volume nan": ("0,1,1,1,1,nan\n", "volume must be finite"), + "volume positive infinity": ( + "0,1,1,1,1,inf\n", "volume must be finite"), + "volume negative infinity": ( + "0,1,1,1,1,-inf\n", "volume must be finite"), + "negative volume": ( + "0,1,1,1,1,-0.001\n", "volume must be non-negative"), + "nonfinite price": ("0,1,nan,1,1,1\n", "high must be finite"), + } + for name, (line, message) in cases.items(): + with self.subTest(name=name): + with self.assertRaisesRegex(ValueError, message): + derive._resample_15m([line]) + + +class CacheVersionTests(unittest.TestCase): + def test_old_version_cannot_preserve_newer_stale_feeds(self) -> None: + with tempfile.TemporaryDirectory() as tmp_s: + root = Path(tmp_s) + source = root / "source.csv" + derived_dir = root / "derived" + full = derived_dir / "full.csv" + window = derived_dir / "window.csv" + stamp = derived_dir / ".version" + derived_dir.mkdir() + + # ensure_derived deliberately rejects tiny/LFS-pointer-like inputs. + # A large ignored header keeps this fixture cheap and exercises the + # real materialization path without weakening that production guard. + source.write_text( + "h" * 1_000_001 + + "\n" + + f"{derive.WINDOW_START_MS},90,91,89,90,0\n" + + f"{derive.WINDOW_START_MS + 60_000},100,101,99,100,2\n" + ) + source_before = hashlib.sha256(source.read_bytes()).hexdigest() + full.write_text("stale-full\n") + window.write_text("stale-window\n") + stamp.write_text("v1:first-row-open\n") + + # Outputs are newer than the source, so only the rule-version stamp + # can force regeneration. + os.utime(source, (100, 100)) + for path in (full, window, stamp): + os.utime(path, (200, 200)) + + with mock.patch.multiple( + derive, + SOURCE_1M=source, + DERIVED_DIR=derived_dir, + DERIVED_15M=full, + DERIVED_15M_WINDOW=window, + DERIVATION_STAMP=stamp, + ): + derive.ensure_derived() + expected = ( + derive.HEADER + + "\n" + + f"{derive.WINDOW_START_MS},100,101,89,100,2\n" + ) + self.assertEqual(full.read_text(), expected) + self.assertEqual(window.read_text(), expected) + self.assertEqual( + stamp.read_text(), derive.DERIVATION_CACHE_VERSION + "\n") + self.assertEqual( + hashlib.sha256(source.read_bytes()).hexdigest(), + source_before, + ) + + # A current stamp and fresh outputs are a true byte-preserving + # no-op, including file mtimes. + before = { + path: (path.read_bytes(), path.stat().st_mtime_ns) + for path in (full, window, stamp) + } + derive.ensure_derived() + after = { + path: (path.read_bytes(), path.stat().st_mtime_ns) + for path in (full, window, stamp) + } + self.assertEqual(after, before) + + +class AtomicWriteTests(unittest.TestCase): + def test_parallel_writers_do_not_share_a_temporary_path(self) -> None: + """Mutation-kill the former fixed ``.new`` race.""" + with tempfile.TemporaryDirectory() as tmp_s: + root = Path(tmp_s) + target = root / "derived.csv" + barrier = threading.Barrier(2) + seen_sources: list[Path] = [] + seen_lock = threading.Lock() + original_replace = Path.replace + + def synchronized_replace(source: Path, destination: Path): + with seen_lock: + seen_sources.append(source) + barrier.wait(timeout=5) + return original_replace(source, destination) + + with mock.patch.object(Path, "replace", synchronized_replace): + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit(derive._write_atomic, target, value) + for value in ("alpha\n", "beta\n") + ] + for future in futures: + future.result(timeout=5) + + self.assertEqual(len(seen_sources), 2) + self.assertEqual(len(set(seen_sources)), 2) + self.assertIn(target.read_text(), {"alpha\n", "beta\n"}) + self.assertEqual(list(root.glob(".derived.csv.*.new")), []) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index fcc66cf..c4cebde 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -104,6 +104,13 @@ set(TEST_SOURCES ) find_package(Threads REQUIRED) +find_package(Python3 COMPONENTS Interpreter REQUIRED) + +add_test( + NAME test_derive_corpus_feeds + COMMAND ${Python3_EXECUTABLE} + ${PROJECT_SOURCE_DIR}/scripts/derive_corpus_feeds_self_test.py +) # When PINEFORGE_ENABLE_COVERAGE is ON we also instrument the test # binaries — header-only code (Series, na, color, log, math::pine_random)