--- /dev/null
+#!/usr/bin/env python3
+"""Deduplicate a FreqAI ``historic_predictions.pkl`` store on ``date_pred``.
+
+FreqAI persists rolling predictions per pair in
+``user_data/models/<identifier>/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 ``<name>.original-<stamp>``."""
+ 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/<identifier>/")
+ 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())
--- /dev/null
+#!/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 <this file>``); 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())
repositories / freqai-strategies.git / commitdiff
