From: Jérôme Benoit Date: Wed, 12 Aug 2026 17:00:57 +0000 (+0200) Subject: feat(scripts): add historic-predictions deduplication tool X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=c20d23a2;p=freqai-strategies.git feat(scripts): add historic-predictions deduplication tool Add scripts/historic_predictions_deduplicate.py to repair FreqAI historic_predictions.pkl stores that a crash-recovery backfill can leave with duplicate date_pred rows, which break the prediction merge for that pair. - Keeps the most informative row per date_pred (informative cells, then non-null count, then most-recent write); deduplicates only non-NaT keys. - Dry-run by default; --apply writes atomically (temp + fsync + os.replace, preserving mode and best-effort owner) after copying the original aside as .original-. Exits 1 when any target is skipped as unreadable. - Repairs the sibling historic_predictions.backup.pkl as well. - Portable single-file tool with a bundled test suite run via uv. --- diff --git a/README.md b/README.md index ad80f36..dab4604 100644 --- a/README.md +++ b/README.md @@ -489,6 +489,38 @@ _Cronjob setup (daily check at 3:00 AM):_ 0 3 * * * cd /path/to/freqai-strategies/ReforceXY && ./docker-upgrade.sh >> user_data/logs/docker-upgrade.log 2>&1 ``` +**Repair duplicated FreqAI historic predictions:** + +FreqAI stores rolling predictions per pair in +`user_data/models//historic_predictions.pkl`. After a brutal stop, +its crash recovery can leave duplicate `date_pred` rows. FreqAI's left merge on +`date_pred` then multiplies the affected candle rows instead of matching them +one-to-one, corrupting that pair's analysis. This tool removes the duplicates, +keeping the most informative row per `date_pred`. FreqAI also +mirrors the store to `historic_predictions.backup.pkl` on every clean save and +falls back to it only when the primary is truncated (loading it raises +`EOFError`), so the tool repairs the backup too. Stop the bot first (a running +bot re-persists the in-memory state), and run it inside the container so it uses +the same pandas that wrote the file. Before each file is rewritten, its original +is copied aside as a timestamped `.original-` file; omit `--apply` to +preview (dry-run). + +```shell +cd quickadapter # or ReforceXY +docker compose stop +docker compose run --rm -T --entrypoint python freqtrade - \ + < ../scripts/historic_predictions_deduplicate.py # dry-run +docker compose run --rm -T --entrypoint python freqtrade - --apply \ + < ../scripts/historic_predictions_deduplicate.py +``` + +Pass `--identifier ` to repair a single model directory or `--path ` +for one file; the default scans every `historic_predictions.pkl` and +`historic_predictions.backup.pkl` under `user_data/models/`. + +The tool exits `1` when it skipped any unreadable store, so a scripted or cron +run can detect a partial repair, and `0` otherwise, including a clean dry-run. + --- ## Note diff --git a/scripts/historic_predictions_deduplicate.py b/scripts/historic_predictions_deduplicate.py new file mode 100755 index 0000000..4f0a8ef --- /dev/null +++ b/scripts/historic_predictions_deduplicate.py @@ -0,0 +1,312 @@ +#!/usr/bin/env python3 +"""Deduplicate a FreqAI ``historic_predictions.pkl`` store on ``date_pred``. + +FreqAI persists rolling predictions per pair in +``user_data/models//historic_predictions.pkl`` (a ``dict`` mapping a +pair to a ``pandas.DataFrame`` keyed on the candle timestamp copied into +``date_pred``). After a brutal stop (SIGKILL/OOM/power loss), FreqAI's clean-exit +save is skipped and, on restart, its backfill can leave the store with the same +``date_pred`` appearing more than once. FreqAI then merges predictions onto the +candle frame with a left join, so duplicate keys silently multiply the affected +candle rows; only builds that add a ``validate="m:1"`` guard raise ``MergeError`` +instead, which stops that pair from being analyzed. + +This tool removes the duplicate ``date_pred`` rows, keeping the most informative +row per timestamp: rows are ranked by informative cells (non-null and, for +numerics, non-zero), then by non-null count, then by most-recent write. A real +prediction is therefore never discarded in favour of an all-NaN placeholder; a +real all-zero row and a zero-filled placeholder are byte-identical, so collapsing +them loses nothing. FreqAI also mirrors the store to +``historic_predictions.backup.pkl`` on every clean save and falls back to it +only when the primary is truncated (loading it raises ``EOFError``), so the tool +deduplicates the backup as well. + +Run it inside the freqtrade container so it uses the same pandas that wrote the +file; the host pandas may be unable to unpickle it. Stop the bot first: a running +bot holds the store in memory and would re-persist the duplicated state. +""" + +from __future__ import annotations + +import argparse +import os +import pickle +import shutil +import stat +import sys +from datetime import datetime, timezone +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import pandas as pd + +DEFAULTS: dict[str, Any] = { + "user_data": "/freqtrade/user_data", + "models_dirname": "models", + "store_filename": "historic_predictions.pkl", + "backup_filename": "historic_predictions.backup.pkl", + "quarantine_tag": "original", + "quarantine_tie_break_limit": 99, +} + +# ``date`` and ``date_pred`` are timestamps, never prediction content; excluding +# them keeps the informativeness score focused on actual prediction values. +_CONTENT_EXCLUDE = ("date_pred", "date") + + +def _content_columns(frame: pd.DataFrame) -> list[str]: + return [column for column in frame.columns if column not in _CONTENT_EXCLUDE] + + +def _informative_score(frame: pd.DataFrame) -> pd.Series: + """Count informative cells per row (non-null and, for numerics, non-zero). + + FreqAI backfill placeholders are all-zero or all-NaN except ``date_pred``, so + a plain non-null count would rank a zero-filled placeholder as high as a real + prediction. Treating zero as non-informative lets a real row outrank a zero + placeholder; ties against an all-NaN placeholder are then broken by the + non-null count (see ``_nonnull_count``). + """ + columns = _content_columns(frame) + if not columns: + return pd.Series(0, index=frame.index, dtype="int64") + block = frame[columns] + numeric = block.apply(pd.to_numeric, errors="coerce") + is_numeric = numeric.notna() + informative = (block.notna() & is_numeric & numeric.ne(0)) | (block.notna() & ~is_numeric) + return informative.sum(axis=1).astype("int64") + + +def _nonnull_count(frame: pd.DataFrame) -> pd.Series: + """Count non-null content cells per row (tie-break below informativeness). + + A real all-zero row has non-null zeros; an all-NaN placeholder does not, so + this keeps the real row when both score zero on informativeness. + """ + columns = _content_columns(frame) + if not columns: + return pd.Series(0, index=frame.index, dtype="int64") + return frame[columns].notna().sum(axis=1).astype("int64") + + +def deduplicate_pair(frame: pd.DataFrame) -> tuple[pd.DataFrame, int]: + """Return the frame with unique non-NaT ``date_pred`` and the removed count. + + Only rows with a valid (non-NaT) ``date_pred`` can duplicate FreqAI's + many-to-one merge key, so NaT rows are preserved untouched. + """ + if frame is None or frame.empty or "date_pred" not in frame.columns: + return frame, 0 + normalized = pd.to_datetime(frame["date_pred"], utc=True, errors="coerce") + valid = normalized.notna() + if not normalized[valid].duplicated().any(): + return frame, 0 + work = frame.reset_index(drop=True) + work = work.assign( + _dp=normalized.to_numpy(), + _valid=valid.to_numpy(), + _score=_informative_score(work).to_numpy(), + _nonnull=_nonnull_count(work).to_numpy(), + _order=work.index.to_numpy(), + ) + # Among duplicate timestamps keep the most informative row, breaking ties by + # non-null count (a real all-zero row beats an all-NaN placeholder) then by + # the latest original position; NaT rows are carried through untouched. + contested = work[work["_valid"]].sort_values( + ["_dp", "_score", "_nonnull", "_order"], kind="stable" + ) + kept_contested = contested.drop_duplicates(subset="_dp", keep="last") + kept = pd.concat([kept_contested, work[~work["_valid"]]]) + kept = kept.sort_values("_dp", kind="stable", na_position="last") + removed = len(frame) - len(kept) + result = kept.drop(columns=["_dp", "_valid", "_score", "_nonnull", "_order"]).reset_index( + drop=True + ) + return result, removed + + +def deduplicate_store( + store: dict[str, pd.DataFrame], +) -> tuple[dict[str, pd.DataFrame], list[dict[str, Any]], int]: + """Deduplicate every pair; return the new store, a report, and total removed.""" + new_store: dict[str, pd.DataFrame] = {} + report: list[dict[str, Any]] = [] + total_removed = 0 + for pair in sorted(store): + frame = store[pair] + rows_before = 0 if frame is None else len(frame) + deduplicated, removed = deduplicate_pair(frame) + new_store[pair] = deduplicated + total_removed += removed + report.append( + { + "pair": pair, + "rows_before": rows_before, + "removed": removed, + "rows_after": rows_before - removed, + } + ) + if removed: + after_normalized = pd.to_datetime(deduplicated["date_pred"], utc=True, errors="coerce") + if after_normalized[after_normalized.notna()].duplicated().any(): + raise AssertionError(f"[{pair}] duplicate date_pred remain after dedup") + return new_store, report, total_removed + + +def _quarantine_original(path: Path, now: datetime) -> Path: + """Copy the pre-dedup file aside as ``.original-``.""" + stamp = now.strftime("%Y%m%dT%H%M%S%fZ") + base = f"{path.name}.{DEFAULTS['quarantine_tag']}-{stamp}" + for index in range(DEFAULTS["quarantine_tie_break_limit"] + 1): + suffix = "" if index == 0 else f"-{index}" + candidate = path.with_name(f"{base}{suffix}") + if not candidate.exists(): + shutil.copy2(path, candidate) + return candidate + raise FileExistsError(path) + + +def _atomic_write_pickle(store: dict[str, pd.DataFrame], path: Path) -> None: + """Write the store atomically (temp + fsync + os.replace), preserving mode/owner. + + Uses the stdlib ``pickle``: FreqAI writes the store with joblib's vendored + cloudpickle, but a plain ``dict`` of DataFrames pickles to a standard stream + that ``pickle`` round-trips and FreqAI's ``cloudpickle.load`` reads back; + standalone ``cloudpickle`` is not importable in the freqtrade image. + """ + temporary_path = path.with_name(f".{path.name}.{uuid4().hex}.tmp") + try: + existing_stat = path.stat() + file_descriptor = os.open(temporary_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o666) + with os.fdopen(file_descriptor, mode="wb") as write_file: + temporary_stat = os.fstat(write_file.fileno()) + if ( + temporary_stat.st_uid != existing_stat.st_uid + or temporary_stat.st_gid != existing_stat.st_gid + ): + # Best-effort: a non-root process on a cross-uid bind mount lacks + # CAP_CHOWN; the store was never chowned before, so failure here + # must not abort the repair. + try: + os.fchown( + write_file.fileno(), + existing_stat.st_uid + if temporary_stat.st_uid != existing_stat.st_uid + else -1, + existing_stat.st_gid + if temporary_stat.st_gid != existing_stat.st_gid + else -1, + ) + except PermissionError: + pass + os.fchmod(write_file.fileno(), stat.S_IMODE(existing_stat.st_mode)) + pickle.dump(store, write_file, protocol=pickle.HIGHEST_PROTOCOL) + write_file.flush() + os.fsync(write_file.fileno()) + os.replace(temporary_path, path) + except BaseException: + temporary_path.unlink(missing_ok=True) + raise + + +def load_store(path: Path) -> dict[str, pd.DataFrame]: + with path.open("rb") as read_file: + store = pickle.load(read_file) + if not isinstance(store, dict): + raise TypeError(f"{path}: expected a dict of DataFrames, got {type(store)!r}") + return store + + +def find_targets(user_data: Path, identifier: str | None, path: str | None) -> list[Path]: + if path is not None: + target = Path(path) + if not target.is_file(): + raise FileNotFoundError(target) + return [target] + filenames = (DEFAULTS["store_filename"], DEFAULTS["backup_filename"]) + models = user_data / DEFAULTS["models_dirname"] + if identifier is not None: + directory = models / identifier + existing = [directory / name for name in filenames if (directory / name).is_file()] + if not existing: + raise FileNotFoundError(directory / DEFAULTS["store_filename"]) + return existing + found: list[Path] = [] + for name in filenames: + found.extend(models.glob(f"*/{name}")) + return sorted(found) + + +def _parse_args(argv: list[str] | None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--user-data", + default=DEFAULTS["user_data"], + help="user_data directory (container path); default %(default)s", + ) + selector = parser.add_mutually_exclusive_group() + selector.add_argument("--identifier", help="repair only models//") + selector.add_argument("--path", help="repair one explicit historic_predictions.pkl") + parser.add_argument( + "--apply", + action="store_true", + help="write changes; without it the tool only reports (dry-run)", + ) + return parser.parse_args(argv) + + +def _print_report(path: Path, report: list[dict[str, Any]], removed: int, apply: bool) -> None: + mode = "apply" if apply else "dry-run" + print(f"# {path} | mode={mode} | pandas={pd.__version__}") + print(f"{'pair':<24}{'rows_before':>12}{'removed':>10}{'rows_after':>12}") + for row in report: + if row["removed"]: + print( + f"{row['pair']:<24}{row['rows_before']:>12}" + f"{row['removed']:>10}{row['rows_after']:>12}" + ) + print(f"# total duplicate rows removed: {removed}") + + +def main(argv: list[str] | None = None) -> int: + args = _parse_args(argv) + user_data = Path(args.user_data) + targets = find_targets(user_data, args.identifier, args.path) + if not targets: + print( + f"# no {DEFAULTS['store_filename']} found under " + f"{user_data}/{DEFAULTS['models_dirname']}/", + file=sys.stderr, + ) + return 0 + changed_files = 0 + skipped_files = 0 + for target in targets: + try: + store = load_store(target) + except (OSError, EOFError, pickle.UnpicklingError, TypeError) as error: + skipped_files += 1 + print(f"# {target}: skipped unreadable store: {error!r}", file=sys.stderr) + continue + new_store, report, removed = deduplicate_store(store) + _print_report(target, report, removed, args.apply) + if removed and args.apply: + quarantine = _quarantine_original(target, datetime.now(timezone.utc)) + _atomic_write_pickle(new_store, target) + print(f"# quarantined original to {quarantine.name}; wrote deduplicated store") + changed_files += 1 + elif removed: + print("# dry-run: re-run with --apply to write changes", file=sys.stderr) + print( + f"# files scanned: {len(targets)} | files changed: {changed_files} " + f"| files skipped: {skipped_files}" + ) + return 1 if skipped_files else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/pyproject.toml b/scripts/pyproject.toml new file mode 100644 index 0000000..a29e489 --- /dev/null +++ b/scripts/pyproject.toml @@ -0,0 +1,9 @@ +[tool.ruff] +line-length = 100 +target-version = "py311" + +# Restrict to Pyflakes and core pycodestyle E4/E7/E9: modern ruff enables UP by +# default, and UP017 would rewrite the repo idiom timezone.utc to datetime.UTC, so +# keeping opinionated rewrites opt-in preserves it. +[tool.ruff.lint] +select = ["E4", "E7", "E9", "F"] diff --git a/scripts/tests/test_historic_predictions_deduplicate.py b/scripts/tests/test_historic_predictions_deduplicate.py new file mode 100755 index 0000000..b90fee9 --- /dev/null +++ b/scripts/tests/test_historic_predictions_deduplicate.py @@ -0,0 +1,375 @@ +#!/usr/bin/env python3 +"""Tests for the historic_predictions deduplication tool. + +The dedup logic is pure pandas and version-agnostic, so these run under any +pandas 2.x/3.x. Runnable with pytest or directly (``python ``); the +container ships pandas but not necessarily pytest. +""" + +from __future__ import annotations + +import contextlib +import importlib.util +import inspect +import io +import pickle +import unittest +from datetime import datetime, timezone +from pathlib import Path + +import pandas as pd + +_MODULE_PATH = Path(__file__).resolve().parent.parent / "historic_predictions_deduplicate.py" +_spec = importlib.util.spec_from_file_location("hp_dedup", _MODULE_PATH) +assert _spec and _spec.loader +hp = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(hp) + + +def _row(date_pred: str, extrema: float, do_predict: int, close: float) -> dict: + return { + "date_pred": pd.Timestamp(date_pred, tz="UTC"), + "&s-extrema": extrema, + "do_predict": do_predict, + "close_price": close, + } + + +def _placeholder(date_pred: str, *, zero_filled: bool) -> dict: + value = 0 if zero_filled else float("nan") + return { + "date_pred": pd.Timestamp(date_pred, tz="UTC"), + "&s-extrema": value, + "do_predict": value, + "close_price": value, + } + + +def test_real_then_zero_placeholder_keeps_real() -> None: + frame = pd.DataFrame( + [ + _row("2026-08-12 12:10:00", 0.9, 1, 100.0), + _placeholder("2026-08-12 12:15:00", zero_filled=True), + _row("2026-08-12 12:15:00", 0.7, -1, 101.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 + assert not result["date_pred"].duplicated().any() + kept = result[result["date_pred"] == pd.Timestamp("2026-08-12 12:15:00", tz="UTC")] + assert kept["&s-extrema"].iloc[0] == 0.7 # real row kept, not the zero placeholder + + +def test_nan_placeholder_dropped() -> None: + frame = pd.DataFrame( + [ + _placeholder("2026-08-12 12:15:00", zero_filled=False), + _row("2026-08-12 12:15:00", 0.6, 1, 100.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 + assert result["&s-extrema"].iloc[0] == 0.6 + + +def test_two_real_rows_keeps_most_recent() -> None: + frame = pd.DataFrame( + [ + _row("2026-08-12 12:15:00", 0.5, 1, 100.0), + _row("2026-08-12 12:15:00", 0.7, 1, 100.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 + assert result["&s-extrema"].iloc[0] == 0.7 # latest write among equal completeness + + +def test_triple_dup_real_in_middle() -> None: + frame = pd.DataFrame( + [ + _placeholder("2026-08-12 12:15:00", zero_filled=True), + _row("2026-08-12 12:15:00", 0.8, 1, 100.0), + _placeholder("2026-08-12 12:15:00", zero_filled=False), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 2 + assert result["&s-extrema"].iloc[0] == 0.8 + + +def test_clean_pair_is_noop() -> None: + frame = pd.DataFrame( + [ + _row("2026-08-12 12:10:00", 0.1, 1, 100.0), + _row("2026-08-12 12:15:00", 0.2, 1, 101.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 0 + assert result.equals(frame) + + +def test_sorted_ascending_after_dedup() -> None: + frame = pd.DataFrame( + [ + _row("2026-08-12 12:20:00", 0.3, 1, 100.0), + _row("2026-08-12 12:15:00", 0.5, 1, 100.0), + _row("2026-08-12 12:15:00", 0.7, 1, 100.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 + assert list(result["date_pred"]) == sorted(result["date_pred"]) + + +def test_idempotent() -> None: + frame = pd.DataFrame( + [ + _placeholder("2026-08-12 12:15:00", zero_filled=True), + _row("2026-08-12 12:15:00", 0.7, 1, 100.0), + ] + ) + once, _ = hp.deduplicate_pair(frame) + twice, removed2 = hp.deduplicate_pair(once) + assert removed2 == 0 + assert twice.equals(once) + + +def test_edge_cases_do_not_crash() -> None: + empty, removed = hp.deduplicate_pair(pd.DataFrame()) + assert removed == 0 and empty.empty + no_key = pd.DataFrame([{"&s-extrema": 0.1}]) + result, removed = hp.deduplicate_pair(no_key) + assert removed == 0 and result.equals(no_key) + nat = pd.DataFrame( + [ + { + "date_pred": pd.NaT, + "&s-extrema": 0.1, + "do_predict": 1, + "close_price": 1.0, + }, + { + "date_pred": pd.NaT, + "&s-extrema": 0.2, + "do_predict": 1, + "close_price": 1.0, + }, + ] + ) + result, removed = hp.deduplicate_pair(nat) + assert removed == 0 # distinct NaT rows are preserved (never match the merge) + assert len(result) == 2 + + +def test_store_roundtrip_and_atomic_write(tmp_path: Path) -> None: + store = { + "SUI/USD:USD": pd.DataFrame( + [ + _row("2026-08-12 12:10:00", 0.9, 1, 100.0), + _placeholder("2026-08-12 12:15:00", zero_filled=True), + _row("2026-08-12 12:15:00", 0.7, -1, 101.0), + ] + ), + "XRP/USD:USD": pd.DataFrame([_row("2026-08-12 12:10:00", 0.1, 1, 1.0)]), + } + path = tmp_path / "historic_predictions.pkl" + with path.open("wb") as handle: + pickle.dump(store, handle) + loaded = hp.load_store(path) + new_store, report, removed = hp.deduplicate_store(loaded) + assert removed == 1 + hp._atomic_write_pickle(new_store, path) + reloaded = hp.load_store(path) + for frame in reloaded.values(): + assert not frame["date_pred"].duplicated().any() + assert {row["pair"] for row in report} == set(store) + + +def test_real_all_zero_row_beats_nan_placeholder() -> None: + frame = pd.DataFrame( + [ + _row("2026-08-12 12:15:00", 0.0, 0, 0.0), # real, all-zero, written first + _placeholder("2026-08-12 12:15:00", zero_filled=False), # all-NaN, later + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 + assert result["&s-extrema"].notna().all() # NaN placeholder dropped, real kept + + +def test_distinct_nat_rows_preserved_with_real_dup() -> None: + frame = pd.DataFrame( + [ + {"date_pred": pd.NaT, "&s-extrema": 0.11, "do_predict": 1, "close_price": 10.0}, + {"date_pred": pd.NaT, "&s-extrema": 0.22, "do_predict": 1, "close_price": 20.0}, + _row("2026-08-12 12:15:00", 0.3, 1, 30.0), + _row("2026-08-12 12:15:00", 0.4, 1, 30.0), + ] + ) + result, removed = hp.deduplicate_pair(frame) + assert removed == 1 # only the real duplicate collapses; both NaT rows kept + assert len(result) == 3 + assert {0.11, 0.22} <= set(result["&s-extrema"]) + + +def _dup_store() -> dict: + return { + "SUI/USD:USD": pd.DataFrame( + [ + _row("2026-08-12 12:10:00", 0.9, 1, 100.0), + _placeholder("2026-08-12 12:15:00", zero_filled=True), + _row("2026-08-12 12:15:00", 0.7, -1, 101.0), + ] + ) + } + + +def test_find_targets_glob_includes_backup_excludes_quarantine(tmp_path: Path) -> None: + model = tmp_path / "models" / "id1" + model.mkdir(parents=True) + (model / "historic_predictions.pkl").touch() + (model / "historic_predictions.backup.pkl").touch() + (model / "historic_predictions.pkl.original-20260812T000000000000Z").touch() + names = {p.name for p in hp.find_targets(tmp_path, None, None)} + assert names == {"historic_predictions.pkl", "historic_predictions.backup.pkl"} + assert {p.name for p in hp.find_targets(tmp_path, "id1", None)} == names + + +def test_find_targets_path_mode_single_file(tmp_path: Path) -> None: + target = tmp_path / "historic_predictions.pkl" + target.touch() + assert hp.find_targets(tmp_path, None, str(target)) == [target] + + +def test_quarantine_original_unique_names(tmp_path: Path) -> None: + target = tmp_path / "historic_predictions.pkl" + target.write_bytes(b"payload") + now = datetime(2026, 8, 12, tzinfo=timezone.utc) + first = hp._quarantine_original(target, now) + second = hp._quarantine_original(target, now) # same stamp -> -1 suffix + assert first.exists() and second.exists() and first != second + assert first.read_bytes() == b"payload" + + +def test_print_report_values() -> None: + report = [{"pair": "SUI/USD:USD", "rows_before": 3, "removed": 1, "rows_after": 2}] + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + hp._print_report(Path("/x/historic_predictions.pkl"), report, 1, apply=True) + out = buffer.getvalue() + assert "mode=apply" in out + assert "SUI/USD:USD" in out + assert "# total duplicate rows removed: 1" in out + + +def test_main_dry_run_writes_nothing(tmp_path: Path) -> None: + path = tmp_path / "historic_predictions.pkl" + with path.open("wb") as handle: + pickle.dump(_dup_store(), handle) + before = path.read_bytes() + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + code = hp.main(["--path", str(path)]) + assert code == 0 + assert path.read_bytes() == before + assert not list(tmp_path.glob("*.original-*")) + assert "files changed: 0" in buffer.getvalue() + + +def test_main_apply_dedups_and_quarantines(tmp_path: Path) -> None: + path = tmp_path / "historic_predictions.pkl" + with path.open("wb") as handle: + pickle.dump(_dup_store(), handle) + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + code = hp.main(["--path", str(path), "--apply"]) + assert code == 0 + quarantined = list(tmp_path.glob("historic_predictions.pkl.original-*")) + assert len(quarantined) == 1 + for frame in hp.load_store(path).values(): + assert not frame["date_pred"].duplicated().any() + assert "files changed: 1" in buffer.getvalue() + replay = io.StringIO() + with contextlib.redirect_stdout(replay): + hp.main(["--path", str(path), "--apply"]) + assert "files changed: 0" in replay.getvalue() # idempotent + + +def test_main_skips_unreadable_target(tmp_path: Path) -> None: + models = tmp_path / "models" + bad = models / "id1" + bad.mkdir(parents=True) + (bad / "historic_predictions.pkl").write_bytes(b"not a pickle") + good = models / "id2" + good.mkdir(parents=True) + with (good / "historic_predictions.pkl").open("wb") as handle: + pickle.dump(_dup_store(), handle) + out, err = io.StringIO(), io.StringIO() + with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err): + code = hp.main(["--user-data", str(tmp_path), "--apply"]) + assert code == 1 + assert "files skipped: 1" in out.getvalue() + assert "files changed: 1" in out.getvalue() + assert "skipped unreadable store" in err.getvalue() + for frame in hp.load_store(good / "historic_predictions.pkl").values(): + assert not frame["date_pred"].duplicated().any() + + +def test_cloudpickle_reads_stdlib_pickle(tmp_path: Path) -> None: + if importlib.util.find_spec("cloudpickle") is None: + raise unittest.SkipTest("cloudpickle absent on host; interop exercised in-container only") + import cloudpickle + + store = {"SUI/USD:USD": pd.DataFrame([_row("2026-08-12 12:10:00", 0.9, 1, 100.0)])} + path = tmp_path / "historic_predictions.pkl" + with path.open("wb") as handle: + pickle.dump(store, handle) + hp._atomic_write_pickle(store, path) + with path.open("rb") as handle: + reloaded = cloudpickle.load(handle) + assert list(reloaded) == ["SUI/USD:USD"] + assert not reloaded["SUI/USD:USD"]["date_pred"].duplicated().any() + + +def test_informative_score_ignores_nat_datetime_cells() -> None: + frame = pd.DataFrame( + { + "&s-extrema": [0.0, 0.0], + "signal_time": pd.to_datetime([pd.Timestamp("2026-01-01", tz="UTC"), pd.NaT]), + } + ) + assert list(hp._informative_score(frame)) == [1, 0] # NaT datetime cell not informative + + +def _run_all() -> int: + tests = [value for name, value in sorted(globals().items()) if name.startswith("test_")] + failures = 0 + skipped = 0 + import tempfile + + for test in tests: + try: + if "tmp_path" in inspect.signature(test).parameters: + with tempfile.TemporaryDirectory() as directory: + test(Path(directory)) + else: + test() + except unittest.SkipTest as reason: + skipped += 1 + print(f"SKIP {test.__name__}: {reason}") + continue + except Exception as error: # noqa: BLE001 - self-test reporter + failures += 1 + print(f"FAIL {test.__name__}: {error!r}") + continue + print(f"PASS {test.__name__}") + summary = f"\n{len(tests) - failures - skipped}/{len(tests) - skipped} passed" + if skipped: + summary += f", {skipped} skipped" + print(summary) + return 1 if failures else 0 + + +if __name__ == "__main__": + raise SystemExit(_run_all())