* 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.
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/<identifier>/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-<stamp>` 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 <id>` to repair a single model directory or `--path <file>`
-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
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
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"]
+++ /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
-[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"]
+++ /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
