From: Jérôme Benoit Date: Tue, 28 Jul 2026 23:57:36 +0000 (+0200) Subject: fix(quickadapter): recheck weight support after feature pipeline (#150) X-Git-Url: https://git.piment-noir.org/?a=commitdiff_plain;h=a175c4969f84355bec7942230c6eb91ba15e90af;p=freqai-strategies.git fix(quickadapter): recheck weight support after feature pipeline (#150) Label-weight support checks (min_pivot_equivalent_count, min_positive_label_weight_fraction, min_effective_sample_size Kish ESS) previously ran only before feature_pipeline.fit_transform, which can remove training rows and renormalize the surviving weights below those thresholds without re-invoking support_policy. Recheck support post-pipeline on the surviving rows: carry the raw base and label weight components through fit_transform as temporary object-keyed label columns (row-filtered in lockstep with X/y/sample_weight, since DataSieve resets indexes), pop them, restore the label schema, and re-gate via support_policy. _sanitize_pipeline_weights fails closed on any malformed post-pipeline weight shape (None/scalar/2-D/wrong-length). Base-only weights carry no label support and skip the re-gate. Selection training and final refit share the path; eval (val/test) weights still bypass support_policy but now fail closed on shape. No tunable, canonical default, or public API change. Fixes #93 --- diff --git a/README.md b/README.md index 07d369e..83d1424 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ docker compose up -d --build | freqai.label_weighting.fill_bandwidth | `fixed` | enum {`fixed`,`knn`} | Per-pivot Gaussian bandwidth selector. `fixed` applies a constant `fill_sigma_candles` to every pivot (legacy behavior). `knn` adapts each pivot's sigma to local pivot density via `sigma_p = clip(fill_bandwidth_alpha * d_k(p), fill_sigma_min_candles, fill_sigma_candles)` where `d_k(p)` is the index distance to the `k`-th nearest pivot neighbor (Loftsgaarden & Quesenberry 1965; Silverman 1986, §5.2). Mitigates the crushing of weaker pivots by stronger neighbors in dense clusters. Ignored when `fill_method` not in {`gaussian`,`epsilon_gaussian`}. | | freqai.label_weighting.fill_bandwidth_neighbors | 1 | int >= 1 | `k` for the k-nearest-neighbor bandwidth selector. Ignored when `fill_method` not in {`gaussian`,`epsilon_gaussian`} or `fill_bandwidth != "knn"`. | | freqai.label_weighting.fill_bandwidth_alpha | 0.5 | float > 0 | Multiplicative factor on the k-th neighbor distance. Smaller values produce sharper, more separated Gaussians; larger values approach the `fixed` behavior. Ignored when `fill_method` not in {`gaussian`,`epsilon_gaussian`} or `fill_bandwidth != "knn"`. | -| freqai.label_weighting.support_policy | `fallback` | enum {`fallback`,`raise`} | Policy when active label weighting fails support checks (after row filtering and causal split guards). `raise` aborts the fit with `ValueError`; `fallback` logs a `WARNING` and uses sanitized base sample weights for that fit. Eval (test/val) weights bypass this policy and always fall back on composition errors. | +| freqai.label_weighting.support_policy | `fallback` | enum {`fallback`,`raise`} | Policy when active label weighting fails support checks (after causal split guards and feature-pipeline row filtering). `raise` aborts the fit with `ValueError`; `fallback` logs a `WARNING` and uses sanitized base sample weights for that fit. Eval (test/val) weights bypass this policy and always fall back on composition errors. | | freqai.label_weighting.min_pivot_equivalent_count | 3 | int >= 1 | Minimum number of surviving pivot-equivalent label weights required after filtering. Pivot-equivalent rows are weights at least 10% of the surviving maximum label weight. | | freqai.label_weighting.min_positive_label_weight_fraction | 0.01 | float [0,1] | Minimum fraction of filtered training rows with finite positive label weights. | | freqai.label_weighting.min_effective_sample_size | 3.0 | float >= 1 | Minimum Kish effective sample size of the final composed training weights. | diff --git a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py index 49f5c35..99d6cf3 100644 --- a/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py +++ b/quickadapter/user_data/freqaimodels/QuickAdapterRegressorV3.py @@ -745,7 +745,41 @@ class QuickAdapterRegressorV3(BaseRegressionModel): reasons=[str(exc)], ) - summary = summarize_label_weight_support(label_weights, composed) + # Support is gated twice on purpose: here on the full split (fail-fast, + # esp. under support_policy='raise') and again post-pipeline in + # _fit_training_pipelines on the rows fed to model.fit. Outlier removal + # is non-monotone on support (pivot_equivalent_count/ESS use a + # max-relative threshold), so this pre-gate is not a conservative bound: + # keeping it changes some raise/fallback outcomes, and under 'raise' it + # can abort a split the post-pipeline gate would have passed. + return QuickAdapterRegressorV3._enforce_train_weight_support( + base_weights, + label_weights, + composed, + label_weighting_config, + context=context, + ) + + @staticmethod + def _enforce_train_weight_support( + base_weights: NDArray[np.floating], + label_weights: NDArray[np.floating], + sample_weights: NDArray[np.floating], + label_weighting_config: dict[str, Any], + *, + context: str, + ) -> NDArray[np.floating]: + """Enforce label-weight support on already-composed training weights. + + Kish ``effective_sample_size`` is measured on the ``sample_weights`` + passed in (full-split composed weights pre-pipeline, post-outlier + surviving weights post-pipeline); ``pivot_equivalent_count`` and + ``positive_label_weight_fraction`` derive from ``label_weights``. + """ + policy = cast( + LabelWeightSupportPolicy, label_weighting_config["support_policy"] + ) + summary = summarize_label_weight_support(label_weights, sample_weights) reasons: list[str] = [] min_pivot_equivalent_count = label_weighting_config[ "min_pivot_equivalent_count" @@ -772,7 +806,10 @@ class QuickAdapterRegressorV3(BaseRegressionModel): ) if reasons: return QuickAdapterRegressorV3._apply_support_policy( - base_weights, context=context, policy=policy, reasons=reasons + base_weights, + context=context, + policy=policy, + reasons=reasons, ) logger.debug( "%s: label weighting support passed " @@ -783,7 +820,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): summary.positive_label_weight_fraction, summary.effective_sample_size, ) - return composed + return sample_weights @staticmethod def _get_selection_category(method: str) -> Optional[str]: @@ -1888,8 +1925,9 @@ class QuickAdapterRegressorV3(BaseRegressionModel): before splitting. After split + causal-guard filtering, train weights compose through ``_compose_train_weights_with_support`` (gated by ``support_policy``) and eval weights through ``_compose_eval_weights`` - (bypasses ``support_policy``). ``_train_common`` then feeds them to - ``model.fit(sample_weight=...)``. + (bypasses ``support_policy``). Training support is rechecked after the + feature pipeline removes rows, before ``_train_common`` feeds the final + weights to ``model.fit(sample_weight=...)``. """ method = self.data_split_parameters.get( "method", QuickAdapterRegressorV3.DATA_SPLIT_METHOD_DEFAULT @@ -2219,15 +2257,22 @@ class QuickAdapterRegressorV3(BaseRegressionModel): unfiltered_df, pair, ) + train_positions = features_filtered.index.get_indexer( + dd["train_features"].index + ) + if (train_positions < 0).any(): + raise ValueError( + f"[{pair}] _train_common: unable to align training rows to " + f"sample weight inputs (missing={int((train_positions < 0).sum())})" + ) + train_weight_inputs = SampleWeightInputs( + base=weights.base[train_positions], + label=None if weights.label is None else weights.label[train_positions], + label_weighting_config=weights.label_weighting_config, + ) if not self.freqai_info.get("fit_live_predictions_candles", 0) or not self.live: dk.fit_labels() - dd = self._apply_pipelines(dd, dk, pair) - if len(dd["train_features"]) != len(dd["train_weights"]): - raise RuntimeError( - f"Pipeline broke shape invariant: " - f"len(train_features)={len(dd['train_features'])} != " - f"len(train_weights)={len(dd['train_weights'])}" - ) + dd = self._apply_pipelines(dd, train_weight_inputs, dk, pair) logger.info(f"Training model on {len(dd['train_features'].columns)} features") logger.info(f"Training model on {len(dd['train_features'])} data points") model = self.fit(dd, dk, **kwargs) @@ -2476,34 +2521,99 @@ class QuickAdapterRegressorV3(BaseRegressionModel): context=f"[{pair}] refit", ) ) + data_dictionary["refit_weight_inputs"] = SampleWeightInputs( + base=refit_base_weights, + label=refit_label_weights, + label_weighting_config=weights.label_weighting_config, + ) return data_dictionary + @staticmethod + def _sanitize_pipeline_weights( + features: pd.DataFrame, + weights: Any, + *, + pair: str, + context: str, + ) -> NDArray[np.floating]: + """Validate and normalize weights returned by the feature pipeline.""" + expected_shape = (len(features),) + actual_shape = np.shape(weights) + if actual_shape != expected_shape: + raise RuntimeError( + f"[{pair}] post_feature_pipeline:{context}: feature pipeline broke " + f"sample-weight shape invariant (expected {expected_shape}, " + f"got {actual_shape})" + ) + return sanitize_and_renormalize( + weights, + logger=logger, + context=f"[{pair}] post_feature_pipeline:{context}", + ) + def _fit_training_pipelines( self, features: pd.DataFrame, labels: pd.DataFrame, weights: NDArray[np.floating], + weight_inputs: SampleWeightInputs, dk: FreqaiDataKitchen, pair: str, context: str, ) -> tuple[pd.DataFrame, pd.DataFrame, NDArray[np.floating]]: - """Fit FreqAI pipelines and return their transformed training data.""" + """Fit FreqAI pipelines and enforce support on the surviving train rows.""" dk.feature_pipeline = self.define_data_pipeline(threads=dk.thread_count) dk.label_pipeline = self.define_label_pipeline(threads=dk.thread_count) - features, labels, weights = dk.feature_pipeline.fit_transform( - features, labels, weights - ) - weights = sanitize_and_renormalize( + pipeline_labels = labels + if weight_inputs.label is not None: + # Smuggle base/label weights as extra label columns so datasieve + # row-filters them in lockstep with the features. Relies on datasieve + # not altering y VALUES (X-only transforms + row drops) and restoring + # them via label_list. _sanitize_pipeline_weights guards row count, + # not a silent y-value transform: a future y-transforming step would + # corrupt these vectors undetected. + base_weight_column = object() + label_weight_column = object() + pipeline_labels = labels.copy() + pipeline_labels[base_weight_column] = weight_inputs.base + pipeline_labels[label_weight_column] = weight_inputs.label + features, pipeline_labels, weights = dk.feature_pipeline.fit_transform( + features, pipeline_labels, weights + ) + weights = QuickAdapterRegressorV3._sanitize_pipeline_weights( + features, weights, - logger=logger, - context=f"[{pair}] post_feature_pipeline:{context}", - ) + pair=pair, + context=context, + ) + # Label-only re-gate: base-only weights carry no pivot/fraction/ESS + # support to recheck (settled pre-pipeline), so they skip this stage. + if weight_inputs.label is not None: + post_pipeline_base_weights = pipeline_labels.pop( + base_weight_column + ).to_numpy(dtype=float) + post_pipeline_label_weights = pipeline_labels.pop( + label_weight_column + ).to_numpy(dtype=float) + # Load-bearing: fit_transform captured label_list WITH the smuggled + # columns; restore it to the real labels or the next validation/test + # transform rebuilds y with the wrong column count (ValueError). + dk.feature_pipeline.label_list = pipeline_labels.columns + weights = QuickAdapterRegressorV3._enforce_train_weight_support( + post_pipeline_base_weights, + post_pipeline_label_weights, + weights, + weight_inputs.label_weighting_config, + context=f"[{pair}] post_feature_pipeline:{context}", + ) + labels = pipeline_labels labels, _, _ = dk.label_pipeline.fit_transform(labels) return features, labels, weights def _apply_pipelines( self, dd: dict, + train_weight_inputs: SampleWeightInputs, dk: FreqaiDataKitchen, pair: str, ) -> dict: @@ -2513,6 +2623,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): dd["train_features"], dd["train_labels"], dd["train_weights"], + train_weight_inputs, dk, pair, "train", @@ -2535,10 +2646,13 @@ class QuickAdapterRegressorV3(BaseRegressionModel): f"transform (outlier removal); relax SVM/DBSCAN outlier " f"thresholds or increase test_size" ) - dd["validation_weights"] = sanitize_and_renormalize( - dd["validation_weights"], - logger=logger, - context=f"[{pair}] post_feature_pipeline:validation", + dd["validation_weights"] = ( + QuickAdapterRegressorV3._sanitize_pipeline_weights( + dd["validation_features"], + dd["validation_weights"], + pair=pair, + context="validation", + ) ) dd["validation_labels"], _, _ = dk.label_pipeline.transform( dd["validation_labels"] @@ -2592,10 +2706,11 @@ class QuickAdapterRegressorV3(BaseRegressionModel): dd["test_features"], dd["test_labels"], dd["test_weights"] ) ) - dd["test_weights"] = sanitize_and_renormalize( + dd["test_weights"] = QuickAdapterRegressorV3._sanitize_pipeline_weights( + dd["test_features"], dd["test_weights"], - logger=logger, - context=f"[{pair}] post_feature_pipeline:test", + pair=pair, + context="test", ) dd["test_labels"], _, _ = dk.label_pipeline.transform(dd["test_labels"]) @@ -2896,6 +3011,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): data_dictionary["train_features"] = data_dictionary.pop("refit_features") data_dictionary["train_labels"] = data_dictionary.pop("refit_labels") data_dictionary["train_weights"] = data_dictionary.pop("refit_weights") + refit_weight_inputs = data_dictionary.pop("refit_weight_inputs") if ( not self.freqai_info.get("fit_live_predictions_candles", 0) or not self.live @@ -2909,6 +3025,7 @@ class QuickAdapterRegressorV3(BaseRegressionModel): data_dictionary["train_features"], data_dictionary["train_labels"], data_dictionary["train_weights"], + refit_weight_inputs, dk, dk.pair, "refit",