From: Jérôme Benoit Date: Wed, 12 Aug 2026 20:54:36 +0000 (+0200) Subject: fix(quickadapter): guard against freqtrade duplicate date_pred regeneration (#202) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=a249a1c2321f46c25284d9e7eb38693a40d1f312;p=freqai-strategies.git fix(quickadapter): guard against freqtrade duplicate date_pred regeneration (#202) * revert(scripts): drop historic-predictions dedup tool Reverts the squash-merged PR #201 (commit c20d23a2). A one-shot on-disk deduplication cannot durably fix the issue: freqtrade 2026.7 regenerates the duplicate date_pred in memory on restart. It is superseded by an in-memory guard in the QuickAdapter model (next commit). * fix(quickadapter): guard against freqtrade duplicate date_pred regeneration freqtrade 2026.7 FreqaiDataDrawer.set_initial_return_values trims the new prediction window by position (new_pred.iloc[len(common_dates):]) instead of by date_pred value. When the persisted history overlaps the tail of the analysis window (the normal restart geometry), the trim misaligns and the following pd.concat re-creates duplicate date_pred rows, which the validate="m:1" merge in attach_return_values_to_return_dataframe rejects with MergeError, blocking the pair. A one-shot on-disk dedup cannot fix this since the duplicate is regenerated in memory on restart. Install an idempotent in-memory guard from the model module that wraps set_initial_return_values, append_model_predictions and attach_return_values_to_return_dataframe to keep date_pred unique, preferring a real prediction over a zero/NaN placeholder and preserving NaT rows. Validated against freqtradeorg/freqtrade:stable_freqai (2026.7) across all restart geometries. * fix(quickadapter): keep original row order in date_pred dedup Sort the kept rows by their original position (_order) instead of by date_pred (_dp). The date sort pushed interior NaT rows to the tail and could change which rows tail(N) returns to the strategy, dropping a real candle and injecting a placeholder. Ordering by original position preserves freqtrade insertion order minus the removed duplicates; date_pred uniqueness is already guaranteed by drop_duplicates. * style(quickadapter): type the date_pred dedup wrappers Annotate the three FreqaiDataDrawer wrappers to match the module conventions and freqtrade signatures (data_drawer.py:286/343/413). * docs(quickadapter): sharpen date_pred duplicate attribution in guard docstring State that set_initial_return_values counts the overlap but trims new_pred by position, so a crash-persisted or overlapping store breaks the contiguous head-prefix assumption and reintroduces duplicate date_pred; drop the inaccurate "normal restart geometry" framing. * docs(quickadapter): correct date_pred duplicate attribution Lead with the robust primary cause: a store already holding duplicate date_pred rows survives set_initial_return_values unchanged (its pd.concat preserves hist_preds) and trips the validate="m:1" merge. Demote the positional-trim path to a secondary case and drop the inaccurate claim that it needs non-monotonic dates: a gapped or tail-overlapping store triggers it with strictly monotonic dates too (proven). Anchor the data_drawer.py source lines. * docs(quickadapter): tighten date_pred dedup guard docstrings Move the freqtrade-bug rationale from the pure _dedupe helper (contract only: unique non-NaT date_pred, most-informative row, NaT preserved, order kept) to _install, where the patch rationale belongs, and condense it with source anchors. Net -13 doc lines; matches the module terse-docstring style. * fix(quickadapter): drop NaT date_pred and restore chronological dedup order Two correctness defects in the date_pred dedup guard reached the validate="m:1" merge in attach_return_values_to_return_dataframe (data_drawer.py:429-431): - The early return only checked duplicate valid dates, so a frame with >=2 NaT date_pred was returned unchanged; the merge then raised MergeError ("Merge keys are not unique in right dataset") because pandas counts repeated null keys as non-unique, even with no NaT on the left. - Ordering the kept rows by original position (_order) left the frame non-chronological, so the tail(N) trim in set_initial_return_values and append_model_predictions could keep stale dates and evict recent ones. Drop NaT date_pred rows (a NaT matches no candle -- freqtrade always assigns real dates at set_initial_return_values, data_drawer.py:305 -- and only harms the m:1 merge) and order survivors chronologically by normalized date_pred; the most-informative row per timestamp is still kept. This supersedes the _order sort introduced in f4b95a7b: removing the NaT rows makes that reorder unnecessary and the tail chronologically correct. --- diff --git a/README.md b/README.md index dab4604..ad80f36 100644 --- a/README.md +++ b/README.md @@ -489,38 +489,6 @@ _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/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index c0d38c4..eb96bb2 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -35,6 +35,7 @@ from datasieve.transforms import SKLearnWrapper from freqtrade.enums import TRADE_MODES from freqtrade.exceptions import DependencyException from freqtrade.freqai.base_models.BaseRegressionModel import BaseRegressionModel +from freqtrade.freqai.data_drawer import FreqaiDataDrawer from freqtrade.freqai.data_kitchen import FreqaiDataKitchen from numpy.typing import NDArray from optuna.storages import JournalStorage @@ -115,6 +116,96 @@ from Utils import ( zigzag, ) +_DATE_PRED_DEDUP_SENTINEL = "_quickadapter_date_pred_dedup_patched" + + +def _dedupe_historic_predictions_on_date_pred(frame: pd.DataFrame) -> pd.DataFrame: + """Return ``frame`` with a unique, chronologically ordered ``date_pred``, + keeping the most informative row per timestamp (a real prediction outranks a + zero/NaN placeholder). ``NaT`` ``date_pred`` rows are dropped: they match no + candle, and two or more of them break the ``validate="m:1"`` merge in + ``attach_return_values_to_return_dataframe`` (data_drawer.py:429-431) since + pandas treats repeated null keys as non-unique. + """ + date_pred = pd.to_datetime(frame["date_pred"], utc=True, errors="coerce") + valid = date_pred.notna() + if valid.all() and not date_pred[valid].duplicated().any(): + return frame + work = frame.reset_index(drop=True) + content = [column for column in work.columns if column not in ("date_pred", "date")] + block = work[content] + numeric = block.apply(pd.to_numeric, errors="coerce") + is_numeric = numeric.notna() + informative = (block.notna() & is_numeric & numeric.ne(0)) | (block.notna() & ~is_numeric) + work = work.assign( + _dp=date_pred.to_numpy(), + _score=informative.sum(axis=1).to_numpy(), + _nonnull=block.notna().sum(axis=1).to_numpy(), + _order=work.index.to_numpy(), + ) + contested = work[valid.to_numpy()].sort_values( + ["_dp", "_score", "_nonnull", "_order"], kind="stable" + ) + kept = contested.drop_duplicates("_dp", keep="last") + return kept.drop(columns=["_dp", "_score", "_nonnull", "_order"]).reset_index(drop=True) + + +def _install_date_pred_dedup_patch() -> None: + """Keep ``FreqaiDataDrawer``'s per-pair prediction store free of duplicate + ``date_pred`` rows, which freqtrade 2026.7 does not deduplicate and its + ``validate="m:1"`` merge (data_drawer.py:429-431) then rejects with a + ``MergeError``. Duplicates persist across a crash or are re-created by the + positional trim in ``set_initial_return_values`` (data_drawer.py:319-321). + + Re-verify the three wrapped signatures against ``data_drawer.py`` on every + freqtrade bump. + """ + if getattr(FreqaiDataDrawer, _DATE_PRED_DEDUP_SENTINEL, False): + return + original_set_initial = FreqaiDataDrawer.set_initial_return_values + original_append = FreqaiDataDrawer.append_model_predictions + original_attach = FreqaiDataDrawer.attach_return_values_to_return_dataframe + + def set_initial_return_values( + self, pair: str, pred_df: pd.DataFrame, dataframe: pd.DataFrame + ) -> None: + original_set_initial(self, pair, pred_df, dataframe) + frame = _dedupe_historic_predictions_on_date_pred(self.historic_predictions[pair]) + self.historic_predictions[pair] = frame + self.model_return_values[pair] = frame.tail(len(dataframe.index)).reset_index(drop=True) + + def append_model_predictions( + self, + pair: str, + predictions: pd.DataFrame, + do_preds: NDArray[np.int_], + dk: FreqaiDataKitchen, + strat_df: pd.DataFrame, + ) -> None: + original_append(self, pair, predictions, do_preds, dk, strat_df) + frame = _dedupe_historic_predictions_on_date_pred(self.historic_predictions[pair]) + self.historic_predictions[pair] = frame + self.model_return_values[pair] = frame.tail(len(strat_df.index)).reset_index(drop=True) + + def attach_return_values_to_return_dataframe( + self, pair: str, dataframe: pd.DataFrame + ) -> pd.DataFrame: + self.model_return_values[pair] = _dedupe_historic_predictions_on_date_pred( + self.model_return_values[pair] + ) + return original_attach(self, pair, dataframe) + + FreqaiDataDrawer.set_initial_return_values = set_initial_return_values + FreqaiDataDrawer.append_model_predictions = append_model_predictions + FreqaiDataDrawer.attach_return_values_to_return_dataframe = ( + attach_return_values_to_return_dataframe + ) + setattr(FreqaiDataDrawer, _DATE_PRED_DEDUP_SENTINEL, True) + + +_install_date_pred_dedup_patch() + + OptunaSampler = Literal["tpe", "auto", "nsgaii", "nsgaiii"] ScalerType = Literal["minmax", "maxabs", "standard", "robust"] DensityAggregation = Literal["power_mean", "quantile", "min", "max"] diff --git a/scripts/historic_predictions_deduplicate.py b/scripts/historic_predictions_deduplicate.py deleted file mode 100755 index 4f0a8ef..0000000 --- a/scripts/historic_predictions_deduplicate.py +++ /dev/null @@ -1,312 +0,0 @@ -#!/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 deleted file mode 100644 index a29e489..0000000 --- a/scripts/pyproject.toml +++ /dev/null @@ -1,9 +0,0 @@ -[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 deleted file mode 100755 index b90fee9..0000000 --- a/scripts/tests/test_historic_predictions_deduplicate.py +++ /dev/null @@ -1,375 +0,0 @@ -#!/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())