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.
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
import re
import subprocess
import unittest
+from pathlib import Path
from qa_support import COVERAGERC, REPO_ROOT, QaTestCase, temporary_directory
# 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:")
"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(
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()],
# 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:
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
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.
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
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])
# ------------------------------------------------------------------ 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])
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"):
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)
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(
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,
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
)
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):
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 (
get_distance,
get_ma_fn,
get_price_fn,
- get_zl_ma_fn,
midpoint,
price_retracement_percent,
smma,
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]
)
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,
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()
(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()
ZigzagResult,
_adapt_label_generator,
_zigzag,
+ compose_sample_weights,
+ compute_label_weights,
generate_label_data,
register_label_generator,
soft_extremum,
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(
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)
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 (
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)
repositories / freqai-strategies.git / commitdiff
