From 300b7c766c775fd3a229463e2f8a09a592223637 Mon Sep 17 00:00:00 2001 From: =?utf8?q?J=C3=A9r=C3=B4me=20Benoit?= Date: Fri, 2 Oct 2026 17:04:15 +0200 Subject: [PATCH] test(quickadapter): strengthen reviewed consumer contracts (#283) --- README.md | 5 +- quickadapter/.coveragerc | 2 +- quickadapter/tests/test_coverage_floor.py | 56 ++++++++----- quickadapter/tests/test_label_transformer.py | 7 ++ .../tests/test_regressor_data_split.py | 32 ++++---- .../tests/test_regressor_selection.py | 39 +++++++-- .../tests/test_regressor_weight_support.py | 4 +- quickadapter/tests/test_strategy_targets.py | 80 ++++++++++++++++--- .../tests/test_utils_config_migration.py | 57 +++++-------- quickadapter/tests/test_utils_indicators.py | 54 ++++++++----- .../tests/test_utils_regressor_plumbing.py | 16 ---- quickadapter/tests/test_utils_zigzag.py | 68 ++++++++++++++-- quickadapter/user_data/strategies/Utils.py | 12 +-- 13 files changed, 291 insertions(+), 141 deletions(-) diff --git a/README.md b/README.md index 2ff429d..6e2c31a 100644 --- a/README.md +++ b/README.md @@ -549,8 +549,9 @@ the floor it justifies, a floor below the current minimum, a floor that is not a percentage, a `precision` coarse enough to round the total past the floor, a disabled branch trace, a disabled namespace walk, a source tree that is not the measured one, a `fail_under` that has drifted into the inert `[run]` section, the -three of coverage's default source exclusions, a coverage pragma in production -source, and any `omit`, `include`, `exclude_lines`, `exclude_also` or +source-level `no cover` or `no branch` pragmas (with or without a colon), +ellipsis-only bodies, additional `TYPE_CHECKING` blocks beyond the existing +import-only block, and any `omit`, `include`, `exclude_lines`, `exclude_also` or `partial_*` of production code. The shared `scripts/run-coverage.sh` runner is exercised with real passing and failing coverage reports and a failing test suite; the checks assert process exit status, not workflow token spelling. diff --git a/quickadapter/.coveragerc b/quickadapter/.coveragerc index a091c98..dbdf96e 100644 --- a/quickadapter/.coveragerc +++ b/quickadapter/.coveragerc @@ -8,5 +8,5 @@ relative_files = True show_missing = True precision = 1 include_namespace_packages = True -# measured 2026-10-01 at 71.5% — raise only, never lower +# measured 2026-10-02 at 71.6% — raise only, never lower fail_under = 68 diff --git a/quickadapter/tests/test_coverage_floor.py b/quickadapter/tests/test_coverage_floor.py index 5276962..1ae523d 100644 --- a/quickadapter/tests/test_coverage_floor.py +++ b/quickadapter/tests/test_coverage_floor.py @@ -5,6 +5,7 @@ import os import re import subprocess import unittest +from pathlib import Path from qa_support import COVERAGERC, REPO_ROOT, QaTestCase, temporary_directory @@ -16,7 +17,7 @@ OMIT = "omit" # The lowest floor the project will accept. Raising it is free; dropping below this is the # silent gate disabling that `.coveragerc` and README.md:532 both forbid. MINIMUM_FLOOR = 67.0 -PRAGMA = re.compile(r"pragma\s*:\s*no\s*cover", re.IGNORECASE) +PRAGMA = re.compile(r"#\s*pragma\s*:?\s*no\s*(?:cover|branch)\b", re.IGNORECASE) # The other two of coverage.py's three DEFAULT_EXCLUDE patterns. ELLIPSIS_BODY = re.compile(r"^\s*(((async )?def .*?)?[\])]+(\s*->.*?)?:\s*)?\.\.\.\s*(#|$)") TYPE_CHECKING = re.compile(r"^\s*if (typing\.)?TYPE_CHECKING:") @@ -151,42 +152,37 @@ class CoverageFloorTest(QaTestCase): "the annotated measurement must clear the floor it justifies", ) - def test_no_production_source_widens_coverage_s_default_exclusions(self): - # The source side of the same lever the config-side keys above cover. coverage.py - # always applies a DEFAULT_EXCLUDE of three patterns, and each removes statements from - # the denominator without any config change at all: a `# pragma: no cover` comment, a - # `...`-only body, and an `if TYPE_CHECKING:` block. Measured for the pragma: one on - # `_normalize_final_take_profit_state` took QuickAdapterV3.py from 58.8 % to 60.2 % and - # the total from 69.3 % to 69.6 %, with the suite green. - # - # The tree is NOT free of the lever today: `if TYPE_CHECKING:` appears in Utils.py and - # is excluded today. A type-checking block is legitimate — it holds imports that do not - # run — so it is permitted up to a pinned count rather than banned outright. The other - # two patterns have no legitimate use here, so they must stay at zero. + def _assert_source_exclusions(self, tree: Path, *, type_checking_blocks: int = 0): + # Statement and partial-branch exemptions can change the gate without a config edit. + # The measured tree permits its one import-only TYPE_CHECKING block, but no pragmas + # or ellipsis-only bodies. Synthetic consumer trees have no type-only imports. counts = { - "pragma: no cover": 0, + "coverage pragma": 0, "ellipsis body": 0, - "if TYPE_CHECKING:": 1, + "if TYPE_CHECKING:": type_checking_blocks, } found = dict.fromkeys(counts, 0) - for path in sorted(MEASURED_TREE.rglob("*.py")): + for path in sorted(tree.rglob("*.py")): for line in path.read_text().splitlines(): if PRAGMA.search(line): - found["pragma: no cover"] += 1 + found["coverage pragma"] += 1 elif ELLIPSIS_BODY.match(line): found["ellipsis body"] += 1 elif TYPE_CHECKING.match(line): found["if TYPE_CHECKING:"] += 1 self.assertEqual(counts, found, "these shrink the denominator without any config change") - def _run_coverage(self, floor, *, tests_pass=True): + def test_no_production_source_widens_coverage_s_default_exclusions(self): + self._assert_source_exclusions(MEASURED_TREE, type_checking_blocks=1) + + def _run_coverage(self, floor, *, tests_pass=True, source_text=None): with temporary_directory() as directory: source = directory / "source" tests = directory / "tests" source.mkdir() tests.mkdir() (source / "sample.py").write_text( - "def choose(flag):\n if flag:\n return 1\n return 2\n" + source_text or "def choose(flag):\n if flag:\n return 1\n return 2\n" ) expected = 1 if tests_pass else 2 (tests / "test_sample.py").write_text( @@ -226,6 +222,28 @@ class CoverageFloorTest(QaTestCase): result = self._run_coverage(50, tests_pass=False) self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + def test_source_exemptions_that_bypass_a_real_report_are_refused(self): + # No branch suppresses missing arcs: combined coverage rises from 4/6 to 5/6 + # despite one missing statement. No cover excludes the entire untaken clause. + baseline = self._run_coverage(80) + self.assertEqual(baseline.returncode, 2, baseline.stdout + baseline.stderr) + for kind in ("branch", "cover"): + for separator in (":", ""): + with self.subTest(kind=kind, separator=separator): + pragma = f"# pragma{separator} no {kind}" + if_comment = f" {pragma}" if kind == "branch" else "" + else_comment = f" {pragma}" if kind == "cover" else "" + source_text = ( + f"def choose(flag):\n if flag:{if_comment}\n" + f" return 1\n else:{else_comment}\n return 2\n" + ) + with temporary_directory() as source: + (source / "sample.py").write_text(source_text) + with self.assertRaises(AssertionError): + self._assert_source_exclusions(source) + result = self._run_coverage(80, source_text=source_text) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + def test_the_source_is_the_measured_tree(self): self.assertEqual( [line.strip() for line in self.parser["run"]["source"].splitlines() if line.strip()], diff --git a/quickadapter/tests/test_label_transformer.py b/quickadapter/tests/test_label_transformer.py index b9dd5fa..28f1234 100644 --- a/quickadapter/tests/test_label_transformer.py +++ b/quickadapter/tests/test_label_transformer.py @@ -83,6 +83,13 @@ class LabelTransformerTest(QaTestCase): # Median 2, quartiles 1 and 3: the outlier does not set the scale. np.testing.assert_allclose(scaled, [-1.0, -0.5, 0.0, 0.5, 49.0], rtol=1e-12) + def test_mmad_uses_the_midpoint_medians_for_an_even_nonconstant_sample(self): + series = np.array([0.0, 2.0, 4.0, 100.0]) + fitted = transformer("mmad", "none", mmad_scaling_factor=1.0) + scaled, *_ = fitted.fit_transform(series) + # Median 3; sorted absolute deviations [1, 1, 3, 97] have MAD 2. + np.testing.assert_allclose(scaled, [-1.5, -0.5, 0.5, 48.5], rtol=1e-12, atol=0.0) + def test_inverse_transform_recovers_the_input_for_every_scaler_pair(self): matrix = label_matrix() for standardization in STANDARDIZATION_TYPES: diff --git a/quickadapter/tests/test_regressor_data_split.py b/quickadapter/tests/test_regressor_data_split.py index 1f64c43..16c2b55 100644 --- a/quickadapter/tests/test_regressor_data_split.py +++ b/quickadapter/tests/test_regressor_data_split.py @@ -595,19 +595,7 @@ class TrainTestSplitCausalPurgeTest(QaTestCase): class TimeSeriesSplitCausalPurgeTest(QaTestCase): - """The chronological split, whose entire body had no test at all. - - `_make_timeseries_split_datasets` is reachable through the documented - `data_split_parameters.method='timeseries_split'`, and coverage reported 0 of its 62 - statements executed. Its own availability comparison at - QuickAdapterRegressorV3.py:3108 could go from strict to inclusive, or be deleted, with - the suite green. - - Deliberately NOT a subclass of the class above. That one's expectations are written around - a ten-row test tail starting at position 30, where TimeSeriesSplit's last fold here is five - rows starting at 35; inheriting it made five tests assert the wrong geometry and fail. The - two blocks are siblings with their own arithmetic, so each carries its own fixtures. - """ + """Final-fold gap and label-availability boundaries for chronological splitting.""" ROWS = 40 N_SPLITS = 3 @@ -659,11 +647,19 @@ class TimeSeriesSplitCausalPurgeTest(QaTestCase): self.features, self.labels, self.weights, self._Kitchen(), unfiltered ) - def test_the_last_fold_is_the_chronological_tail(self): - result = self._split(self.unfiltered, horizon=0) - test_index = list(result["test_features"].index) - self.assertEqual(test_index, list(range(self.ROWS - self.TEST_SIZE, self.ROWS))) - self.assertTrue(max(result["train_features"].index) < min(test_index)) + def test_the_last_fold_preserves_the_gap_and_the_chronological_tail(self): + # With no emitted availability, these cases exercise only the splitter gap. + # A horizon of three widens a zero requested gap to three, ending training at row 31. + # The availability-column purge is exercised separately below. + for horizon, gap, expected_train in ( + (0, 2, list(range(33))), + (3, 3, list(range(32))), + (3, 0, list(range(32))), + ): + with self.subTest(horizon=horizon, gap=gap): + result = self._split(self.unfiltered, horizon=horizon, gap=gap) + self.assertEqual(list(result["train_features"].index), expected_train) + self.assertEqual(list(result["test_features"].index), [35, 36, 37, 38, 39]) def test_a_row_known_at_or_after_the_first_test_row_is_purged(self): # The boundary this block exists for, and the one coverage could not see at all. diff --git a/quickadapter/tests/test_regressor_selection.py b/quickadapter/tests/test_regressor_selection.py index e476cb4..62c61ce 100644 --- a/quickadapter/tests/test_regressor_selection.py +++ b/quickadapter/tests/test_regressor_selection.py @@ -7,7 +7,7 @@ import optuna import pandas as pd from EnumErrors import enum_error_message from LabelTransformer import EXTREMA_SELECTION_METHODS, SKIMAGE_THRESHOLD_METHODS -from qa_support import REGRESSOR_MODULE, QaTestCase +from qa_support import REGRESSOR_MODULE, QaTestCase, model_config, temporary_directory from quickadapter.user_data.freqaimodels.QuickAdapterRegressorV3 import QuickAdapterRegressorV3 @@ -1252,13 +1252,19 @@ class RegressorSelectionTest(QaTestCase): self.assertEqual(minima.tolist(), [-1.0]) self.assertEqual(maxima.tolist(), [1.0]) - def test_ranking_keeps_the_most_extreme_peaks_first(self): + def test_ranking_distinguishes_interior_peaks_from_full_series_extrema(self): series = pd.Series([0.1, -0.5, 0.9, -0.2, 0.3, 0.05, -0.95, 0.7]) - for selection_method in ("rank_extrema", "rank_peaks"): + # The final 0.7 is a raw extremum but not an interior peak; 0.3 is a peak. + for selection_method, maxima_indices, maxima_values in ( + ("rank_extrema", [2, 7], [0.9, 0.7]), + ("rank_peaks", [2, 4], [0.9, 0.3]), + ): with self.subTest(selection_method=selection_method): minima, maxima = QuickAdapterRegressorV3.get_pred_min_max(series, selection_method) + self.assertEqual(minima.index.tolist(), [6, 1, 3]) self.assertEqual(minima.tolist(), [-0.95, -0.5, -0.2]) - self.assertEqual(maxima.iloc[0], 0.9) + self.assertEqual(maxima.index.tolist(), maxima_indices) + self.assertEqual(maxima.tolist(), maxima_values) def test_the_keep_fraction_shrinks_the_candidate_pool_to_the_extremes(self): wave = pd.Series([0.0, 1.0, 0.0, -1.0, 0.0, 2.0, 0.0, -2.0, 0.0]) @@ -1385,11 +1391,28 @@ class RegressorSelectionTest(QaTestCase): # ------------------------------------------------------------------ threshold surfaces - def test_the_median_surface_medianises_the_candidate_sets(self): + def test_the_median_prediction_thresholds_use_the_selected_candidate_sets(self): series = pd.Series([0.1, -0.5, 0.9, -0.2, 0.3, 0.05, -0.95, 0.7]) - minimum, maximum = QuickAdapterRegressorV3.median_min_max(series, "rank_extrema") - self.assertAlmostEqual(float(minimum), -0.5, places=9) - self.assertAlmostEqual(float(maximum), 0.8, places=9) + with temporary_directory() as temp: + model = QuickAdapterRegressorV3(config=model_config(temp)) + for selection_method, expected in ( + ("rank_extrema", (-0.5, 0.8)), + ("rank_peaks", (-0.5, 0.6)), + ): + with self.subTest(selection_method=selection_method): + minimum, maximum = model.min_max_pred( + "prediction", + { + "selection_method": selection_method, + "threshold_method": "median", + "keep_fraction": 1.0, + }, + pd.DataFrame({"prediction": series}), + fit_live_predictions_candles=8, + label_period_candles=1, + ) + self.assertAlmostEqual(float(minimum), expected[0], places=9) + self.assertAlmostEqual(float(maximum), expected[1], places=9) def test_the_soft_extremum_surface_hardens_itself_as_alpha_grows(self): series = pd.Series([0.1, -0.5, 0.9, -0.2, 0.3, 0.05, -0.95, 0.7]) diff --git a/quickadapter/tests/test_regressor_weight_support.py b/quickadapter/tests/test_regressor_weight_support.py index afa1b51..72a2483 100644 --- a/quickadapter/tests/test_regressor_weight_support.py +++ b/quickadapter/tests/test_regressor_weight_support.py @@ -126,7 +126,9 @@ class EnforceSupportTest(QaTestCase): def test_one_ulp_below_a_threshold_is_refused(self): with self.assertRaisesRegex(ValueError, "effective_sample_size"): - self._enforce(np.ones(4), np.ones(4), min_effective_sample_size=4.5) + self._enforce( + np.ones(4), np.ones(4), min_effective_sample_size=np.nextafter(4.0, np.inf) + ) def test_too_few_pivots_is_refused(self): with self.assertRaisesRegex(ValueError, "pivot_equivalent_count"): diff --git a/quickadapter/tests/test_strategy_targets.py b/quickadapter/tests/test_strategy_targets.py index 2cad15c..423c5a6 100644 --- a/quickadapter/tests/test_strategy_targets.py +++ b/quickadapter/tests/test_strategy_targets.py @@ -46,6 +46,25 @@ def runtime(tmp, **overrides): return model +def causal_target_strategy(tmp, smoothing_method): + """Explicit label and weight settings for the smoothing dependency fixture.""" + model = runtime( + tmp, + freqai={ + "feature_parameters": { + "label_period_candles": 1, + "label_natr_multiplier": 10.5, + "label_horizon_candles": 18, + "causal_mode": True, + }, + "label_smoothing": {"default": {"method": smoothing_method, "window_candles": 5}}, + "label_weighting": {"default": {"strategy": "uniform", "fill_method": "zero"}}, + }, + ) + model.bot_start() + return model + + def candles(rows: int = 6, **columns) -> pd.DataFrame: """A strictly rising OHLCV frame, extended with the named columns.""" index = np.arange(rows) @@ -558,17 +577,58 @@ class StrategyTargetsTest(QaTestCase): self.assertNotIn(EXTREMA_COLUMN, frame.columns) self.assertTrue(result[EXTREMA_COLUMN].notna().any()) - def test_every_labelled_candle_carries_at_least_one_candle_of_lookahead(self): + def test_smoothed_availability_covers_both_passes_of_the_five_candle_kernel(self): with temporary_directory() as temp: - model = runtime(temp) - model.bot_start() - result = model.set_freqai_targets(zigzag(), {"pair": PAIR}) - horizon = result[LABEL_HORIZON_COLUMN].to_numpy() - labelled = result[EXTREMA_COLUMN].to_numpy() != 0.0 - self.assertTrue(labelled.any()) - self.assertTrue(np.isfinite(horizon).all()) - self.assertTrue((horizon == np.round(horizon)).all()) - self.assertTrue((horizon[labelled] >= 1).all()) + frame = zigzag() + raw = causal_target_strategy(temp, "none").set_freqai_targets( + frame.copy(), {"pair": PAIR} + ) + smoothed = causal_target_strategy(temp, "gaussian").set_freqai_targets( + frame.copy(), {"pair": PAIR} + ) + # Two passes through a length-five FIR kernel span nine input candles: + # offsets -4 through +4, not the single-pass centered radius of two. + positions = np.arange(len(frame)) + for horizon_column in ( + LABEL_HORIZON_COLUMN, + "s-extrema_weight_known_at_lookahead", + ): + with self.subTest(horizon_column=horizon_column): + raw_available_at = positions + raw[horizon_column].to_numpy() + expected = [ + max(raw_available_at[row - 4 : row + 5]) - row + for row in range(4, len(frame) - 4) + ] + np.testing.assert_array_equal(smoothed[horizon_column].iloc[4:-4], expected) + + def test_smoothed_labels_and_weights_are_reproducible_at_their_declared_availability(self): + with temporary_directory() as temp: + model = causal_target_strategy(temp, "gaussian") + frame = zigzag() + full = model.set_freqai_targets(frame.copy(), {"pair": PAIR}) + for value_column, horizon_column in ( + (EXTREMA_COLUMN, LABEL_HORIZON_COLUMN), + (LABEL_WEIGHT_COLUMN, "s-extrema_weight_known_at_lookahead"), + ): + for row in (26, 34, 43): + with self.subTest(value_column=value_column, row=row): + available_at = row + int(full[horizon_column].iloc[row]) + prefix = model.set_freqai_targets( + frame.iloc[: available_at + 1].copy(), {"pair": PAIR} + ) + self.assertAlmostEqual( + prefix[value_column].iloc[row], full[value_column].iloc[row], places=12 + ) + # These rows change when only one future candle is present, + # so the prefix comparison cannot pass on an inert fixture. + premature = model.set_freqai_targets( + frame.iloc[: row + 2].copy(), {"pair": PAIR} + ) + self.assertNotAlmostEqual( + premature[value_column].iloc[row], + full[value_column].iloc[row], + places=12, + ) def test_a_flat_price_series_produces_no_labels_and_no_label_weights(self): flat = pd.DataFrame( diff --git a/quickadapter/tests/test_utils_config_migration.py b/quickadapter/tests/test_utils_config_migration.py index 278f833..e150ff0 100644 --- a/quickadapter/tests/test_utils_config_migration.py +++ b/quickadapter/tests/test_utils_config_migration.py @@ -3,10 +3,10 @@ import logging import unittest -from qa_support import QaTestCase +from qa_support import QaTestCase, model_config, temporary_directory +from QuickAdapterV3 import QuickAdapterV3 from Utils import ( _MISSING, - CONFIG_DEPRECATIONS, _delete_path, _get_path, _set_path, @@ -129,23 +129,25 @@ class UtilsConfigMigrationTest(QaTestCase): self.assertEqual(sorted(config["freqai"]), ["label_smoothing"]) - def test_k_the_deprecation_table_is_a_tuple_of_well_formed_unique_entries(self): - self.assertIsInstance(CONFIG_DEPRECATIONS, tuple) - self.assertEqual( - len({entry[0] for entry in CONFIG_DEPRECATIONS}), - len(CONFIG_DEPRECATIONS), - "two entries share an old path, so the second can never fire", - ) - for old_path, new_path, predicate, guidance in CONFIG_DEPRECATIONS: - with self.subTest(old_path=old_path): - self.assertIsInstance(old_path, str) - self.assertIn(".", old_path) - self.assertNotEqual(old_path, new_path) - self.assertTrue(new_path is None or (isinstance(new_path, str) and "." in new_path)) - self.assertTrue(predicate is None or callable(predicate)) - self.assertTrue(guidance is None or isinstance(guidance, str)) - # A key with nowhere to go must say why; a plain rename has nothing to explain. - self.assertEqual(new_path is None, guidance is not None) + def test_legacy_reversal_decay_migrates_to_the_public_canonical_key(self): + config = {"reversal_confirmation": {"decay_ratio": 0.8}} + + migrate_config(config, LOGGER) + + self.assertEqual(config, {"reversal_confirmation": {"decay_fraction": 0.8}}) + + def test_strategy_resolves_migrated_decay_and_preserves_a_canonical_override(self): + for section, expected in ( + ({"decay_ratio": 0.8}, 0.8), + ({"decay_ratio": 0.8, "decay_fraction": 0.7}, 0.7), + ): + with self.subTest(section=section), temporary_directory() as root: + config = model_config(root, reversal_confirmation=section) + + strategy = QuickAdapterV3(config) + + self.assertEqual(strategy.reversal_confirmation["decay_fraction"], expected) + self.assertEqual(config["reversal_confirmation"], {"decay_fraction": expected}) def test_a_renamed_section_reaches_its_keys_through_the_second_hop(self): # The table is order-dependent by construction: entry 0 renames @@ -173,23 +175,8 @@ class UtilsConfigMigrationTest(QaTestCase): ) self.assertNotIn("extrema_weighting", config["freqai"]) - def test_the_table_still_encodes_a_rename_chain(self): - # The structural half, which is a fact about the table rather than about behaviour: - # at least one entry's old path must sit under an earlier entry's new path, or the - # behavioural test above has stopped testing a chain. - inherited = [ - old_path - for old_path, _, _, _ in CONFIG_DEPRECATIONS - for _, earlier_new, _, _ in CONFIG_DEPRECATIONS - if earlier_new is not None - and (old_path == earlier_new or old_path.startswith(f"{earlier_new}.")) - ] - self.assertGreater(len(inherited), 0, "the table no longer encodes a rename chain") - def test_a_two_hop_rename_reaches_its_final_key(self): - # The chain test above covers ONE section rename plus a key move. The table also - # contains two-hop chains inside a single section, and asserting that "a chain exists" - # plus testing one of them says nothing about the others. + # Section renames and two-hop key renames must both reach their final public keys. # `threshold_outlier` -> `outlier_threshold_quantile` -> `outlier_quantile` for old_key, expected in (("threshold_outlier", "outlier_quantile"),): with self.subTest(key=old_key): diff --git a/quickadapter/tests/test_utils_indicators.py b/quickadapter/tests/test_utils_indicators.py index 692aeeb..9c77e7c 100644 --- a/quickadapter/tests/test_utils_indicators.py +++ b/quickadapter/tests/test_utils_indicators.py @@ -7,6 +7,7 @@ import unittest import numpy as np import pandas as pd +import talib.abstract as ta from numpy.testing import assert_allclose, assert_array_equal from qa_support import QaTestCase, ohlcv_frame from Utils import ( @@ -25,7 +26,6 @@ from Utils import ( get_distance, get_ma_fn, get_price_fn, - get_zl_ma_fn, midpoint, price_retracement_percent, smma, @@ -395,23 +395,31 @@ class UtilsIndicatorsTest(QaTestCase): with self.subTest(plateau=plateau): self.assertNotIn(plateau, fractal_highs) - def test_each_moving_average_name_resolves_to_its_own_callable(self): - callables = [get_ma_fn(name) for name in MA_MODES] - for name, function in zip(MA_MODES, callables, strict=True): - with self.subTest(name=name): - self.assertTrue(callable(function)) - self.assertIs(get_ma_fn(name), function) - self.assertEqual(len(set(callables)), len(MA_MODES)) - self.assertEqual(len({get_zl_ma_fn(name) for name in MA_MODES}), len(MA_MODES)) + def test_weighted_and_adaptive_average_names_produce_the_named_native_average(self): + prices = pd.Series([1.0, 1.2, 1.1, 1.4, 1.3, 1.7, 1.6, 1.9, 1.8, 2.1, 2.0, 2.3]) + for name, native_average in (("wma", ta.WMA), ("kama", ta.KAMA)): + for period in (3, 6): + with self.subTest(name=name, period=period): + assert_allclose( + get_ma_fn(name)(prices, timeperiod=period), + native_average(prices, timeperiod=period), + rtol=1e-12, + atol=0.0, + equal_nan=True, + ) def test_an_unknown_moving_average_name_falls_back_to_the_simple_average(self): - # The registry is a .get with the SMA default, so an unrecognised name is not an - # error: it silently resolves to the first entry. Pinned because a raise here would - # break every caller that forwards a user-supplied mamode. - sma = get_ma_fn(MA_MODES[0]) + prices = pd.Series([1.0, 1.2, 1.1, 1.4, 1.3, 1.7, 1.6, 1.9, 1.8, 2.1, 2.0, 2.3]) + expected = ta.SMA(prices, timeperiod=6) for name in ("bogus", "", "SMA", "sma ", None, 0): with self.subTest(name=repr(name)): - self.assertIs(get_ma_fn(name), sma) + assert_allclose( + get_ma_fn(name)(prices, timeperiod=6), + expected, + rtol=1e-12, + atol=0.0, + equal_nan=True, + ) def test_each_price_name_resolves_to_its_own_callable(self): callables = [get_price_fn(name) for name in PRICE_MODES] @@ -641,13 +649,23 @@ class UtilsIndicatorsTest(QaTestCase): ) def test_the_oscillator_is_the_gap_between_its_two_averages(self): - frame = ohlcv_frame([100.0, 101, 102, 101.5, 103, 104, 103.5, 105, 106, 105.5, 107, 108]) + frame = ohlcv_frame( + [100.0, 101, 102, 101.5, 103, 104, 103.5, 105, 106, 105.5, 107, 108] * 5 + ) close = frame.get("close") - for mamode in MA_MODES: + for mamode, native_average in ( + ("sma", ta.SMA), + ("ema", ta.EMA), + ("wma", ta.WMA), + ("dema", ta.DEMA), + ("tema", ta.TEMA), + ("trima", ta.TRIMA), + ("kama", ta.KAMA), + ("t3", ta.T3), + ): for ma1_length, ma2_length in ((3, 6), (2, 8)): with self.subTest(mamode=mamode, ma1=ma1_length, ma2=ma2_length): - average = get_ma_fn(mamode) - expected = average(close, ma1_length) - average(close, ma2_length) + expected = native_average(close, ma1_length) - native_average(close, ma2_length) assert_allclose( ewo(frame, ma1_length, ma2_length, mamode=mamode), expected, diff --git a/quickadapter/tests/test_utils_regressor_plumbing.py b/quickadapter/tests/test_utils_regressor_plumbing.py index 84205d1..9c800d3 100644 --- a/quickadapter/tests/test_utils_regressor_plumbing.py +++ b/quickadapter/tests/test_utils_regressor_plumbing.py @@ -461,11 +461,6 @@ class UtilsRegressorPlumbingTest(QaTestCase): np.testing.assert_array_equal(eval_set[0][1].to_numpy(), targets.to_numpy()) self.assertEqual(eval_weights[0].ndim, 1) self.assertEqual(eval_weights[0].shape, (len(features),)) - # eval_set[0] must be the caller's own frames: fit_regressor ravel y but not X, - # and a copy here would silently drop the caller's index alignment. - self.assertIs(eval_set[0][0], features) - self.assertIs(eval_set[0][1], targets) - self.assertIs(eval_weights[0], weights) def test_a_non_positive_test_size_suppresses_evaluation_entirely(self): features, targets, weights = validation_split() @@ -477,17 +472,6 @@ class UtilsRegressorPlumbingTest(QaTestCase): (None, None), ) - def test_the_weight_vector_is_passed_through_without_reshaping_or_validation(self): - # The helper never inspects the weights: alignment is the caller's contract - # (both call sites pass a vector already indexed to the validation rows), so - # this pins that a mismatched shape is forwarded rather than repaired. - features, targets, _ = validation_split() - - eval_set, eval_weights = make_test_set_and_weights(features, targets, np.ones((2, 3)), 0.25) - - self.assertEqual(eval_weights[0].shape, (2, 3)) - self.assertIs(eval_set[0][0], features) - if __name__ == "__main__": unittest.main() diff --git a/quickadapter/tests/test_utils_zigzag.py b/quickadapter/tests/test_utils_zigzag.py index fccac55..06b8fb4 100644 --- a/quickadapter/tests/test_utils_zigzag.py +++ b/quickadapter/tests/test_utils_zigzag.py @@ -17,6 +17,8 @@ from Utils import ( ZigzagResult, _adapt_label_generator, _zigzag, + compose_sample_weights, + compute_label_weights, generate_label_data, register_label_generator, soft_extremum, @@ -73,6 +75,29 @@ def zigzag_on(*legs: tuple[float, float, int]) -> ZigzagResult: return zigzag_result(price_path(*legs)) +def w_leg_threshold_ratios(frame: pd.DataFrame) -> np.ndarray: + """Measure the two completed W legs using Wilder ATR, independently of zigzag.""" + highs = frame["high"].to_numpy() + lows = frame["low"].to_numpy() + closes = frame["close"].to_numpy() + true_ranges = np.maximum.reduce( + [highs[1:] - lows[1:], abs(highs[1:] - closes[:-1]), abs(lows[1:] - closes[:-1])] + ) + atr = np.full(len(frame), np.nan) + atr[NATR_PERIOD] = true_ranges[:NATR_PERIOD].mean() + for position in range(NATR_PERIOD + 1, len(frame)): + atr[position] = ( + atr[position - 1] * (NATR_PERIOD - 1) + true_ranges[position - 1] + ) / NATR_PERIOD + natr = atr / closes + natr[:NATR_PERIOD] = natr[NATR_PERIOD] + log_thresholds = np.log1p(NATR_MULTIPLIER * natr) + # Both endpoint candles belong to each leg. Prices use peak highs and trough lows. + amplitudes = [abs(np.log(lows[29] / highs[0])), abs(np.log(highs[69] / lows[29]))] + denominators = [np.median(log_thresholds[:30]), np.median(log_thresholds[29:70])] + return np.asarray(amplitudes) / denominators + + def nested_bar_frame(second_close: float, second_spread: float) -> pd.DataFrame: """Return a frame whose first two wide bars clear both initial thresholds on one candle.""" closes = np.concatenate( @@ -225,7 +250,38 @@ class UtilsZigzagTest(QaTestCase): self.assertEqual(result.volume_weighted_efficiency_ratios[:2], [1.0, 1.0]) # Constant volume makes the average volume per candle equal its median. self.assertEqual(result.volume_rates[:2], [1.0, 1.0]) - self.assertTrue(np.all(np.asarray(result.amplitude_threshold_ratios[:-1]) > 0.0)) + + def test_amplitude_threshold_ratios_divide_each_leg_by_its_log_volatility_threshold(self): + frame = ohlcv_frame(price_path(*W_LEGS)) + result = _zigzag(frame, **PARAMS) + + np.testing.assert_allclose( + result.amplitude_threshold_ratios, + [*w_leg_threshold_ratios(frame), np.nan], + rtol=1e-12, + atol=0.0, + ) + + def test_generated_amplitude_threshold_ratios_preserve_relative_training_importance(self): + frame = ohlcv_frame(price_path(*W_LEGS)) + label = generate_label_data(frame, EXTREMA_COLUMN, PARAMS, LOGGER) + label_weights = compute_label_weights( + len(frame), + label.indices, + label.metrics, + {"strategy": "amplitude_threshold_ratio", "fill_method": "zero"}, + logger=LOGGER, + ) + sample_weights = compose_sample_weights( + np.ones(len(frame)), label_weights, logger=LOGGER, context="W legs" + ) + + expected = w_leg_threshold_ratios(frame) + self.assertAlmostEqual( + float(sample_weights[29] / sample_weights[0]), + float(expected[1] / expected[0]), + places=12, + ) def test_the_volume_rate_compares_the_leg_average_to_its_median(self): closes = price_path(*W_LEGS) @@ -470,12 +526,10 @@ class UtilsZigzagTest(QaTestCase): self.assertTrue(label.series.index.equals(index)) def test_label_weights_survive_an_index_that_is_not_a_position(self): - # The consumer side of the same contract. `compute_label_weights` casts the pivot - # indices to int and bounds them by n_values, so index labels raise TypeError on the - # DatetimeIndex that FreqAI actually passes, and are dropped as out-of-range on a - # shifted integer index, leaving every weight at zero with only a warning. - from Utils import compute_label_weights - + # The consumer casts pivot indices to int and bounds them by n_values. Datetime + # labels can fail conversion; shifted integer labels can shift or discard pivots. + # This fixture starts at 500, so every misused label is out of range and all weights + # would be zero. A smaller shift can instead leave nonzero weights on wrong rows. closes = price_path(*W_LEGS) expected = generate_label_data(ohlcv_frame(closes), EXTREMA_COLUMN, PARAMS, LOGGER) for name, index in ( diff --git a/quickadapter/user_data/strategies/Utils.py b/quickadapter/user_data/strategies/Utils.py index 2e3884e..60f965c 100644 --- a/quickadapter/user_data/strategies/Utils.py +++ b/quickadapter/user_data/strategies/Utils.py @@ -4088,12 +4088,12 @@ def _zigzag( natr_warmup_end_pos = int(finite_natr_positions[0]) if finite_natr_positions.size > 0 else n natr_values = natr.bfill().to_numpy() - # Pivots are tracked as POSITIONS, never as index labels. `compute_label_weights` - # consumes `ZigzagResult.indices` as row offsets (it casts to int and bounds them by - # `n_values`), and FreqAI hands the strategy a DatetimeIndex frame, so index labels - # would be Timestamps there: the weighting path raises TypeError on the first candle - # of every pair, and on any non-zero-based integer index it silently drops every pivot - # and returns all-zero weights. `LabelData.indices` documents the positional contract. + # Pivots are row positions, never index labels: LabelData.indices defines this contract. + # compute_label_weights consumes integer offsets in [0, n_values). + # Datetime labels are not integer row offsets and can raise conversion errors. + # Shifted integer labels can move weights to different rows; out-of-range labels + # are discarded, leaving all-zero weights only if no valid weighted pivots remain. + # This positional contract is independent of downstream index-alignment validation. thresholds: NDArray[np.floating] = natr_values * natr_multiplier closes = df.get("close").to_numpy(dtype=float) highs = df.get("high").to_numpy(dtype=float) -- 2.53.0