1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_boosting.py",
4  "content": "class BoostingStrategy:\n    \"\"\"Scale-adaptive bandpass Newton boosting.\n\n    Core idea:\n    1. Use trust-region pseudo-targets instead of raw residuals/Newton steps.\n       This keeps updates stable on outliers and confident mistakes.\n    2. Reweight samples by *informative difficulty*: emphasize examples that are\n       still wrong/uncertain, but avoid exponential focus on extreme outliers.\n    3. Compute learner weights with a regularized weighted line search.\n\n    This gives a unified strategy for classification and regression that mixes\n    fast gradient/Newton-style correction with robust sample selection.\n    \"\"\"\n\n    def __init__(self, config):\n        self.config = config\n        self.task_type = config[\"task_type\"]\n        self.n_rounds = config[\"n_rounds\"]\n        self.learning_rate = config[\"learning_rate\"]\n\n        # Hard-coded algorithmic constants.\n        self.reg_lambda = 0.5\n        self.weight_floor = 0.15\n        self.weight_memory = 0.25\n        self.class_step_clip = 2.5\n        self.reg_alpha_clip = 2.5\n        self.cls_alpha_clip = 3.0\n\n        self._last_predictions = None\n\n    def init_weights(self, n_samples):\n        return np.ones(n_samples) / n_samples\n\n    def _sigmoid(self, x):\n        return 1.0 / (1.0 + np.exp(-np.clip(x, -30.0, 30.0)))\n\n    def _uniform_weights(self, n_samples):\n        return np.ones(n_samples) / n_samples\n\n    def _progress(self, round_idx):\n        return round_idx / max(1, self.n_rounds - 1)\n\n    def compute_targets(self, y, current_predictions, sample_weights, round_idx):\n        self._last_predictions = np.asarray(current_predictions).copy()\n        progress = self._progress(round_idx)\n\n        if self.task_type == \"regression\":\n            residual = y - current_predictions\n            scale = np.sqrt(np.average(residual ** 2, weights=sample_weights)) + 1e-8\n\n            # Pseudo-Huber style target: behaves like residuals near zero but\n            # smoothly clips very large residuals. The trust region widens over\n            # rounds so early updates are conservative and later updates refine.\n            delta = scale * (1.25 + 1.25 * progress)\n            return residual / np.sqrt(1.0 + (residual / delta) ** 2)\n\n        probs = self._sigmoid(current_predictions)\n        grad = y - probs\n        hess = probs * (1.0 - probs)\n\n        # Blend first-order and moderated Newton targets. Early rounds use more\n        # gradient signal; later rounds lean more on curvature for refinement.\n        newton_mix = 0.2 + 0.6 * progress\n        newton_step = grad / np.sqrt(hess + 0.05)\n        newton_step = np.clip(newton_step, -self.class_step_clip, self.class_step_clip)\n        targets = (1.0 - newton_mix) * grad + newton_mix * newton_step\n\n        # Mild focal amplification for samples with larger probability error.\n        targets *= (0.75 + 0.5 * np.sqrt(np.abs(grad) + 1e-12))\n        return targets\n\n    def compute_learner_weight(self, learner, X, y, pseudo_targets,\n                                sample_weights, round_idx):\n        preds = learner.predict(X)\n\n        if self.task_type == \"regression\":\n            numerator = np.sum(sample_weights * pseudo_targets * preds)\n            denominator = np.sum(sample_weights * preds ** 2) + self.reg_lambda\n            alpha = numerator / (denominator + 1e-12)\n            return float(np.clip(alpha, 0.0, self.reg_alpha_clip))\n\n        probs = self._sigmoid(self._last_predictions)\n        hess = probs * (1.0 - probs)\n\n        numerator = np.sum(sample_weights * pseudo_targets * preds)\n        denominator = np.sum(sample_weights * (hess + 0.05) * preds ** 2) + self.reg_lambda\n        alpha = numerator / (denominator + 1e-12)\n        return float(np.clip(alpha, 0.0, self.cls_alpha_clip))\n\n    def update_weights(self, sample_weights, learner, X, y, pseudo_targets,\n                       alpha, round_idx):\n        n_samples = len(sample_weights)\n        uniform = self._uniform_weights(n_samples)\n        old_weights = sample_weights / (sample_weights.sum() + 1e-12)\n        preds = learner.predict(X)\n        progress = self._progress(round_idx)\n\n        if self.task_type == \"regression\":\n            new_predictions = self._last_predictions + self.learning_rate * alpha * preds\n            residual = y - new_predictions\n            scale = np.sqrt(np.average(residual ** 2, weights=old_weights)) + 1e-8\n            z = np.abs(residual) / scale\n\n            # Informative-difficulty weighting: emphasize unresolved errors, but\n            # exponentially temper extremely large residuals to avoid chasing\n            # outliers. This is the bandpass component of the method.\n            informative = (z / (1.0 + z)) * np.exp(-0.15 * z)\n            proposal = self.weight_floor + informative\n        else:\n            new_scores = self._last_predictions + self.learning_rate * alpha * preds\n            y_signed = 2.0 * y - 1.0\n            margin = y_signed * new_scores\n            probs = self._sigmoid(new_scores)\n\n            entropy = 4.0 * probs * (1.0 - probs)  # high near the decision boundary\n            hard = 1.0 / (1.0 + np.exp(np.clip(margin, -10.0, 10.0)))\n            proposal = self.weight_floor + 0.6 * entropy + 0.4 * np.sqrt(hard)\n\n        proposal = proposal / (proposal.sum() + 1e-12)\n\n        # Curriculum schedule: start close to uniform, then gradually trust the\n        # learned informative-difficulty weights more strongly.\n        focus = 0.2 + 0.6 * progress\n        new_weights = (1.0 - focus) * uniform + focus * proposal\n        new_weights = (1.0 - self.weight_memory) * new_weights + self.weight_memory * old_weights\n        return new_weights\n",
5  "after_line": 0,
6  "start_line": 147,
7  "end_line": 255
8}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_boosting.py",
4  "content": "class BoostingStrategy:\n    \"\"\"Curriculum bandpass boosting.\n\n    Classification:\n      - Fit discrete label trees (strong on shallow trees).\n      - Update weights using post-round margins, emphasizing misclassified and\n        near-boundary samples without AdaBoost's exponential blow-up.\n\n    Regression:\n      - Fit a blend of raw residuals and robust pseudo-Huber residuals.\n      - Use informative-difficulty reweighting: focus on unresolved errors while\n        tempering extreme residuals that behave like outliers.\n    \"\"\"\n\n    def __init__(self, config):\n        self.config = config\n        self.task_type = config[\"task_type\"]\n        self.n_rounds = config[\"n_rounds\"]\n        self.learning_rate = config[\"learning_rate\"]\n        self.n_samples = config[\"n_samples\"]\n\n        self.reg_lambda = 0.5\n        self.weight_floor = 0.15\n        self.weight_memory = 0.30\n        self._last_predictions = None\n\n    def init_weights(self, n_samples):\n        return np.ones(n_samples) / n_samples\n\n    def _sigmoid(self, x):\n        return 1.0 / (1.0 + np.exp(-np.clip(x, -30.0, 30.0)))\n\n    def _uniform_weights(self, n_samples):\n        return np.ones(n_samples) / n_samples\n\n    def _progress(self, round_idx):\n        return round_idx / max(1, self.n_rounds - 1)\n\n    def compute_targets(self, y, current_predictions, sample_weights, round_idx):\n        self._last_predictions = np.asarray(current_predictions).copy()\n\n        if self.task_type == \"classification\":\n            # Use discrete trees for classification, and place the novelty in the\n            # margin-aware alpha/weight update rather than residual fitting.\n            return y.astype(int)\n\n        progress = self._progress(round_idx)\n        residual = y - current_predictions\n        scale = np.sqrt(np.average(residual ** 2, weights=sample_weights)) + 1e-8\n\n        # Robust residual with a widening trust region.\n        delta = scale * (1.5 + progress)\n        robust = residual / np.sqrt(1.0 + (residual / delta) ** 2)\n\n        # Large datasets benefit from staying closer to raw residual fitting,\n        # while smaller datasets benefit more from robustness.\n        raw_mix = 0.35 + 0.35 * progress + (0.20 if self.n_samples > 5000 else 0.0)\n        raw_mix = min(raw_mix, 0.95)\n        return raw_mix * residual + (1.0 - raw_mix) * robust\n\n    def compute_learner_weight(self, learner, X, y, pseudo_targets,\n                                sample_weights, round_idx):\n        preds = learner.predict(X)\n\n        if self.task_type == \"classification\":\n            incorrect = (preds != y).astype(float)\n            weighted_err = np.sum(sample_weights * incorrect) / (sample_weights.sum() + 1e-12)\n            weighted_err = np.clip(weighted_err, 1e-6, 1.0 - 1e-6)\n            progress = self._progress(round_idx)\n\n            # Slightly stronger than the reference AdaBoost step, but still\n            # shrinked because the ensemble runs for many rounds.\n            alpha = (0.75 + 0.50 * progress) * self.learning_rate\n            alpha *= 0.5 * np.log((1.0 - weighted_err) / weighted_err)\n            return float(max(alpha, 0.0))\n\n        numerator = np.sum(sample_weights * pseudo_targets * preds)\n        denominator = np.sum(sample_weights * preds ** 2) + self.reg_lambda\n        alpha = numerator / (denominator + 1e-12)\n        alpha_cap = 4.0 if self.n_samples > 5000 else 3.0\n        return float(np.clip(alpha, 0.0, alpha_cap))\n\n    def update_weights(self, sample_weights, learner, X, y, pseudo_targets,\n                       alpha, round_idx):\n        n_samples = len(sample_weights)\n        uniform = self._uniform_weights(n_samples)\n        old_weights = sample_weights / (sample_weights.sum() + 1e-12)\n        preds = learner.predict(X)\n        progress = self._progress(round_idx)\n\n        if self.task_type == \"classification\":\n            vote = 2.0 * preds - 1.0\n            new_scores = self._last_predictions + alpha * vote\n            y_signed = 2.0 * y - 1.0\n            margin = y_signed * new_scores\n            incorrect = (preds != y).astype(float)\n\n            hard = 1.0 / (1.0 + np.exp(np.clip(margin, -10.0, 10.0)))\n            boundary = np.exp(-np.abs(margin))\n            proposal = self.weight_floor + 0.55 * hard + 0.35 * boundary + 0.10 * incorrect\n\n            proposal = proposal / (proposal.sum() + 1e-12)\n            focus = 0.35 + 0.35 * progress\n            new_weights = (1.0 - focus) * uniform + focus * proposal\n        else:\n            new_predictions = self._last_predictions + self.learning_rate * alpha * preds\n            residual = y - new_predictions\n            scale = np.sqrt(np.average(residual ** 2, weights=old_weights)) + 1e-8\n            z = np.abs(residual) / scale\n\n            temper = 0.08 if self.n_samples > 5000 else 0.15\n            informative = (z / (1.0 + z)) * np.exp(-temper * z)\n            proposal = self.weight_floor + informative\n            proposal = proposal / (proposal.sum() + 1e-12)\n\n            focus = 0.12 + 0.38 * progress\n            if self.n_samples > 5000:\n                focus *= 0.35\n            new_weights = (1.0 - focus) * uniform + focus * proposal\n\n        new_weights = (1.0 - self.weight_memory) * new_weights + self.weight_memory * old_weights\n        return new_weights\n",
5  "after_line": 0,
6  "start_line": 147,
7  "end_line": 272
8}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_boosting.py",
4  "content": "class BoostingStrategy:\n    \"\"\"Task-adaptive bandpass boosting.\n\n    - Classification: AdaBoost-style voting with a margin-bandpass reweighting\n      that keeps attention on mistakes and low-margin samples.\n    - Regression (small data): robust trust-region residual boosting with\n      informative-difficulty sample reweighting.\n    - Regression (large data): fall back to stable Newton-style residual\n      boosting to avoid underfitting from aggressive reweighting.\n    \"\"\"\n\n    def __init__(self, config):\n        self.config = config\n        self.task_type = config[\"task_type\"]\n        self.n_rounds = config[\"n_rounds\"]\n        self.learning_rate = config[\"learning_rate\"]\n        self.n_samples = config[\"n_samples\"]\n\n        self.reg_lambda = 1.0\n        self.weight_floor = 0.15\n        self.weight_memory = 0.25\n        self.reg_alpha_clip = 2.5\n        self._last_predictions = None\n\n    def init_weights(self, n_samples):\n        return np.ones(n_samples) / n_samples\n\n    def _progress(self, round_idx):\n        return round_idx / max(1, self.n_rounds - 1)\n\n    def _sigmoid(self, x):\n        return 1.0 / (1.0 + np.exp(-np.clip(x, -30.0, 30.0)))\n\n    def _is_large_regression(self):\n        return self.task_type == \"regression\" and self.n_samples > 5000\n\n    def compute_targets(self, y, current_predictions, sample_weights, round_idx):\n        self._last_predictions = np.asarray(current_predictions).copy()\n\n        if self.task_type == \"classification\":\n            return y\n\n        residual = y - current_predictions\n        if self._is_large_regression():\n            # Stable residual boosting on large datasets.\n            return residual\n\n        progress = self._progress(round_idx)\n        scale = np.sqrt(np.average(residual ** 2, weights=sample_weights)) + 1e-8\n        delta = scale * (1.25 + 1.25 * progress)\n        return residual / np.sqrt(1.0 + (residual / delta) ** 2)\n\n    def compute_learner_weight(self, learner, X, y, pseudo_targets,\n                                sample_weights, round_idx):\n        preds = learner.predict(X)\n\n        if self.task_type == \"classification\":\n            incorrect = (preds != y).astype(float)\n            weighted_err = np.dot(sample_weights, incorrect) / (sample_weights.sum() + 1e-12)\n            weighted_err = np.clip(weighted_err, 1e-10, 1.0 - 1e-10)\n            alpha = self.learning_rate * 0.5 * np.log((1.0 - weighted_err) / weighted_err)\n            return float(max(alpha, 0.0))\n\n        numerator = np.sum(sample_weights * pseudo_targets * preds)\n        denominator = np.sum(sample_weights * preds ** 2) + self.reg_lambda\n        alpha = numerator / (denominator + 1e-12)\n        if self._is_large_regression():\n            return float(np.clip(alpha, 0.0, 4.0))\n        return float(np.clip(alpha, 0.0, self.reg_alpha_clip))\n\n    def update_weights(self, sample_weights, learner, X, y, pseudo_targets,\n                       alpha, round_idx):\n        preds = learner.predict(X)\n\n        if self.task_type == \"classification\":\n            incorrect = (preds != y).astype(float)\n            new_weights = sample_weights * np.exp(alpha * incorrect)\n\n            # Margin-bandpass refinement: give a mild bonus to low-margin points\n            # without overwhelming the classic AdaBoost update.\n            new_scores = self._last_predictions + alpha * (2.0 * preds - 1.0)\n            margin = (2.0 * y - 1.0) * new_scores\n            bandpass = 0.85 + 0.15 * np.exp(-np.abs(margin))\n            new_weights = new_weights * bandpass\n            return new_weights / (new_weights.sum() + 1e-12)\n\n        if self._is_large_regression():\n            # Keep large-scale regression stable and close to Newton boosting.\n            return sample_weights\n\n        old_weights = sample_weights / (sample_weights.sum() + 1e-12)\n        new_predictions = self._last_predictions + self.learning_rate * alpha * preds\n        residual = y - new_predictions\n        scale = np.sqrt(np.average(residual ** 2, weights=old_weights)) + 1e-8\n        z = np.abs(residual) / scale\n\n        informative = (z / (1.0 + z)) * np.exp(-0.15 * z)\n        proposal = self.weight_floor + informative\n        proposal = proposal / (proposal.sum() + 1e-12)\n\n        progress = self._progress(round_idx)\n        focus = 0.2 + 0.6 * progress\n        uniform = np.ones_like(old_weights) / len(old_weights)\n        new_weights = (1.0 - focus) * uniform + focus * proposal\n        new_weights = (1.0 - self.weight_memory) * new_weights + self.weight_memory * old_weights\n        return new_weights\n",
5  "after_line": 0,
6  "start_line": 147,
7  "end_line": 268
8}

1{
2  "n": 2
3}