1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_anomaly.py",
4  "start_line": 160,
5  "end_line": 212,
6  "content": "class CustomAnomalyDetector:\n    \"\"\"Hybrid Copula-Isolation Forest (HCIF): adaptive combination of copula-based and isolation-based detection.\n    \n    Combines COPOD (copula tail probabilities) and Isolation Forest (path length)\n    with adaptive weighting based on feature dependency and skewness.\n    \n    Steps:\n    1. Compute COPOD and IForest scores on training data.\n    2. Normalize each score to zero mean, unit variance using training statistics.\n    3. Compute weight alpha = average absolute pairwise correlation of features\n       (higher correlation favors COPOD, lower favors IForest).\n    4. Optionally adjust weight based on average skewness magnitude.\n    5. Final score = alpha * copod_score + (1 - alpha) * iforest_score.\n    \"\"\"\n\n    def __init__(self):\n        from pyod.models.copod import COPOD\n        from pyod.models.iforest import IForest\n        import numpy as np\n        \n        self.copod = COPOD(contamination=0.1)\n        self.iforest = IForest(random_state=42, contamination=0.1)\n        self.copod_mean = None\n        self.copod_std = None\n        self.iforest_mean = None\n        self.iforest_std = None\n        self.weight = None\n        \n    def fit(self, X):\n        import numpy as np\n        \n        # Fit both detectors\n        self.copod.fit(X)\n        self.iforest.fit(X)\n        \n        # Get training scores for normalization\n        copod_scores = self.copod.decision_function(X)\n        iforest_scores = self.iforest.decision_function(X)\n        \n        # Normalize to zero mean, unit variance\n        self.copod_mean = np.mean(copod_scores)\n        self.copod_std = np.std(copod_scores) + 1e-8\n        self.iforest_mean = np.mean(iforest_scores)\n        self.iforest_std = np.std(iforest_scores) + 1e-8\n        \n        # Compute adaptive weight based on feature dependency\n        # Average absolute pairwise correlation\n        corr_matrix = np.corrcoef(X, rowvar=False)\n        np.fill_diagonal(corr_matrix, 0)\n        avg_abs_corr = np.mean(np.abs(corr_matrix))\n        \n        # Average absolute skewness\n        from scipy.stats import skew\n        sk = skew(X, axis=0)\n        avg_abs_skew = np.mean(np.abs(sk))\n        \n        # Weight: higher correlation favors COPOD, higher skewness favors IForest?\n        # Actually, COPOD handles skewness via left/right tails, but we can still adjust.\n        # Simple: weight = avg_abs_corr (0-1)\n        self.weight = np.clip(avg_abs_corr, 0.0, 1.0)\n        \n        # Optional: adjust weight by skewness (if skewness high, maybe give more weight to COPOD?)\n        # COPOD handles skewness well, so keep weight as correlation.\n        # For extreme skewness, maybe increase weight for COPOD.\n        # Let's combine: weight = 0.5 * avg_abs_corr + 0.5 * (avg_abs_skew / max_skew)\n        # where max_skew is heuristic, say 5.\n        # But keep simple for now.\n        \n        return self\n        \n    def decision_function(self, X):\n        import numpy as np\n        \n        # Get raw scores\n        copod_scores = self.copod.decision_function(X)\n        iforest_scores = self.iforest.decision_function(X)\n        \n        # Normalize using training statistics\n        copod_norm = (copod_scores - self.copod_mean) / self.copod_std\n        iforest_norm = (iforest_scores - self.iforest_mean) / self.iforest_std\n        \n        # Combine with adaptive weight\n        scores = self.weight * copod_norm + (1 - self.weight) * iforest_norm\n        \n        # Return scores (higher = more anomalous)\n        return scores"
7}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_anomaly.py",
4  "start_line": 160,
5  "end_line": 245,
6  "content": "class CustomAnomalyDetector:\n    \"\"\"Dimensionality-Adaptive Hybrid (DAH): combines ECOD (inductive) and Isolation Forest.\n    \n    Uses feature count to weight the two detectors:\n    - For high-dimensional data (n_features > 30), weight leans toward Isolation Forest.\n    - For low-dimensional data, weight leans toward ECOD.\n    \n    ECOD is implemented inductively (no transductive leak) using empirical CDF from training data.\n    Isolation Forest from pyod.\n    \n    Final score = w * ecod_score + (1 - w) * iforest_score, where w = min(1, n_features / 30).\n    \"\"\"\n\n    def __init__(self):\n        from pyod.models.iforest import IForest\n        import numpy as np\n        \n        self.iforest = IForest(random_state=42, contamination=0.1)\n        # ECOD inductive component (self-implemented)\n        self.ecod_sorted = None\n        self.ecod_skew_sign = None\n        self.ecod_n_train = None\n        \n        self.weight = None\n        self.iforest_mean = None\n        self.iforest_std = None\n        self.ecod_mean = None\n        self.ecod_std = None\n        \n    def _ecod_fit(self, X):\n        \"\"\"Inductive ECOD training: store sorted training values per feature.\"\"\"\n        import numpy as np\n        from scipy.stats import skew\n        \n        self.ecod_n_train = X.shape[0]\n        self.ecod_sorted = np.sort(X, axis=0)\n        self.ecod_skew_sign = np.sign(skew(X, axis=0))\n        \n    def _ecod_decision_function(self, X):\n        \"\"\"Compute ECOD scores inductively.\"\"\"\n        import numpy as np\n        \n        n = self.ecod_n_train\n        d = X.shape[1]\n        scores = np.zeros(X.shape[0])\n        for j in range(d):\n            ranks = np.searchsorted(self.ecod_sorted[:, j], X[:, j], side='right')\n            ecdf_left = ranks / (n + 1)\n            ecdf_right = 1.0 - ecdf_left\n            ecdf_left = np.clip(ecdf_left, 1e-10, 1.0)\n            ecdf_right = np.clip(ecdf_right, 1e-10, 1.0)\n            u_left = -np.log(ecdf_left)\n            u_right = -np.log(ecdf_right)\n            # Use max of left/right tail (original ECOD uses max for both skew signs)\n            # Actually ECOD uses left tail for negative skew, right tail for positive skew.\n            # We'll follow original: if skew negative, use max(u_left, u_right); else same.\n            # The baseline ECOD uses max for both. We'll keep simple.\n            u_j = np.maximum(u_left, u_right)\n            scores += u_j\n        return scores\n        \n    def fit(self, X):\n        import numpy as np\n        \n        n_features = X.shape[1]\n        # weight based on dimensionality\n        self.weight = min(1.0, n_features / 30.0)  # if n_features > 30, weight=1 (full weight to ECOD?)\n        # Actually we want high-dim -> more weight to IForest. So weight for ECOD should decrease.\n        # Let's set weight_ecod = 1 - min(1, n_features / 30). So high-dim weight_ecod small.\n        self.weight = 1.0 - min(1.0, n_features / 30.0)\n        \n        # Fit IForest\n        self.iforest.fit(X)\n        iforest_scores = self.iforest.decision_function(X)\n        self.iforest_mean = np.mean(iforest_scores)\n        self.iforest_std = np.std(iforest_scores) + 1e-8\n        \n        # Fit ECOD inductive\n        self._ecod_fit(X)\n        ecod_scores = self._ecod_decision_function(X)\n        self.ecod_mean = np.mean(ecod_scores)\n        self.ecod_std = np.std(ecod_scores) + 1e-8\n        \n        return self\n        \n    def decision_function(self, X):\n        import numpy as np\n        \n        iforest_scores = self.iforest.decision_function(X)\n        ecod_scores = self._ecod_decision_function(X)\n        \n        # Normalize using training statistics\n        iforest_norm = (iforest_scores - self.iforest_mean) / self.iforest_std\n        ecod_norm = (ecod_scores - self.ecod_mean) / self.ecod_std\n        \n        # Combine with weight\n        scores = self.weight * ecod_norm + (1 - self.weight) * iforest_norm\n        \n        return scores"
7}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_anomaly.py",
4  "start_line": 160,
5  "end_line": 258,
6  "content": "class CustomAnomalyDetector:\n    \"\"\"Maximum Ensemble of Diverse Detectors (MEDD): combines Isolation Forest, COPOD, and k-NN.\n    \n    Uses three diverse detection paradigms:\n    1. Isolation Forest (tree-based isolation)\n    2. COPOD (copula tail probabilities)\n    3. k-NN (distance to k-th nearest neighbor)\n    \n    Each detector's scores are normalized using robust scaling (median, IQR).\n    Final anomaly score = maximum of the three normalized scores.\n    This captures anomalies detectable by any of the three mechanisms.\n    \"\"\"\n\n    def __init__(self):\n        from pyod.models.iforest import IForest\n        from pyod.models.copod import COPOD\n        from pyod.models.knn import KNN\n        import numpy as np\n        \n        self.iforest = IForest(random_state=42, contamination=0.1)\n        self.copod = COPOD(contamination=0.1)\n        self.knn = KNN(n_neighbors=10, contamination=0.1)\n        \n        # Storage for normalization parameters\n        self.iforest_median = None\n        self.iforest_iqr = None\n        self.copod_median = None\n        self.copod_iqr = None\n        self.knn_median = None\n        self.knn_iqr = None\n        \n    def fit(self, X):\n        import numpy as np\n        \n        # Fit all detectors\n        self.iforest.fit(X)\n        self.copod.fit(X)\n        self.knn.fit(X)\n        \n        # Get training scores for robust normalization\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Compute median and IQR (75th - 25th percentile)\n        def robust_params(scores):\n            q75, q25 = np.percentile(scores, [75, 25])\n            iqr = q75 - q25\n            if iqr == 0:\n                iqr = 1.0\n            median = np.median(scores)\n            return median, iqr\n        \n        self.iforest_median, self.iforest_iqr = robust_params(iforest_scores)\n        self.copod_median, self.copod_iqr = robust_params(copod_scores)\n        self.knn_median, self.knn_iqr = robust_params(knn_scores)\n        \n        return self\n        \n    def decision_function(self, X):\n        import numpy as np\n        \n        # Get raw scores\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Robust normalization\n        iforest_norm = (iforest_scores - self.iforest_median) / self.iforest_iqr\n        copod_norm = (copod_scores - self.copod_median) / self.copod_iqr\n        knn_norm = (knn_scores - self.knn_median) / self.knn_iqr\n        \n        # Maximum across detectors\n        scores = np.maximum.reduce([iforest_norm, copod_norm, knn_norm])\n        \n        return scores"
7}

1{
2  "n": 3
3}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_anomaly.py",
4  "start_line": 160,
5  "end_line": 235,
6  "content": "class CustomAnomalyDetector:\n    \"\"\"Maximum Ensemble of Diverse Detectors (MEDD): combines Isolation Forest, COPOD, and k-NN.\n    \n    Novel contributions:\n    1. Combines three diverse detection paradigms: isolation-based (IForest), \n       copula-based (COPOD), and distance-based (k-NN) for complementary coverage.\n    2. Uses robust normalization (median, interquartile range) instead of mean/std \n       to handle skewed score distributions and outliers in the scores themselves.\n    3. Employs maximum pooling (element-wise max) across normalized scores, \n       implementing an \"OR\" logic: anomalies detectable by ANY detector are highlighted.\n    4. Adaptive k-neighbors selection: k = min(10, sqrt(n_samples)) for k-NN.\n    \n    This ensemble is designed to capture different anomaly types:\n    - IForest: anomalies with few splits (isolated points)\n    - COPOD: anomalies in tail dependencies across features\n    - k-NN: anomalies far from their neighbors in Euclidean space\n    \"\"\"\n\n    def __init__(self):\n        from pyod.models.iforest import IForest\n        from pyod.models.copod import COPOD\n        from pyod.models.knn import KNN\n        import numpy as np\n        \n        self.iforest = IForest(random_state=42, contamination=0.1)\n        self.copod = COPOD(contamination=0.1)\n        # k-NN with adaptive neighbors based on data size (set later)\n        self.knn = None  # will be initialized in fit with appropriate k\n        \n        # Storage for robust normalization parameters\n        self.iforest_median = None\n        self.iforest_iqr = None\n        self.copod_median = None\n        self.copod_iqr = None\n        self.knn_median = None\n        self.knn_iqr = None\n        \n    def fit(self, X):\n        import numpy as np\n        from pyod.models.knn import KNN\n        \n        # Adaptive k for k-NN: min(10, sqrt(n_samples))\n        n_samples = X.shape[0]\n        k = min(10, int(np.sqrt(n_samples)))\n        self.knn = KNN(n_neighbors=k, contamination=0.1)\n        \n        # Fit all detectors\n        self.iforest.fit(X)\n        self.copod.fit(X)\n        self.knn.fit(X)\n        \n        # Get training scores for robust normalization\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Compute median and IQR (75th - 25th percentile)\n        def robust_params(scores):\n            q75, q25 = np.percentile(scores, [75, 25])\n            iqr = q75 - q25\n            if iqr == 0:\n                iqr = 1.0\n            median = np.median(scores)\n            return median, iqr\n        \n        self.iforest_median, self.iforest_iqr = robust_params(iforest_scores)\n        self.copod_median, self.copod_iqr = robust_params(copod_scores)\n        self.knn_median, self.knn_iqr = robust_params(knn_scores)\n        \n        return self\n        \n    def decision_function(self, X):\n        import numpy as np\n        \n        # Get raw scores\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Robust normalization: (score - median) / IQR\n        iforest_norm = (iforest_scores - self.iforest_median) / self.iforest_iqr\n        copod_norm = (copod_scores - self.copod_median) / self.copod_iqr\n        knn_norm = (knn_scores - self.knn_median) / self.knn_iqr\n        \n        # Maximum across detectors: anomaly if ANY detector flags it\n        scores = np.maximum.reduce([iforest_norm, copod_norm, knn_norm])\n        \n        return scores"
7}

1{
2  "op": "replace",
3  "filename": "scikit-learn/custom_anomaly.py",
4  "start_line": 160,
5  "end_line": 247,
6  "content": "class CustomAnomalyDetector:\n    \"\"\"Diversity-Enhanced Maximum Ensemble (DEME): combines Isolation Forest, COPOD, and k-NN with consensus weighting.\n    \n    Novel algorithmic contributions:\n    1. **Three diverse detection paradigms**: isolation-based (IForest), copula-based (COPOD), \n       and distance-based (k-NN) for complementary anomaly detection.\n    2. **Robust normalization**: Uses median and interquartile range (IQR) instead of mean/std \n       to handle skewed score distributions and remain outlier-resistant.\n    3. **Consensus-enhanced maximum pooling**: Instead of simple element-wise max, we compute:\n          score = max(s1, s2, s3) + α * (s1 * s2 * s3)^(1/3)\n       where α = agreement_weight (tunable). The product term boosts scores when all detectors agree.\n    4. **Adaptive k-selection**: k = min(10, sqrt(n_samples)) for k-NN to adapt to dataset size.\n    5. **Feature-aware initialization**: Optionally, weight detectors based on feature characteristics\n       (correlation, skewness), though baseline uses equal consideration.\n    \n    The combination captures:\n    - IForest: anomalies isolated in feature space\n    - COPOD: anomalies in multivariate tail dependencies  \n    - k-NN: anomalies far from local neighbors\n    - Consensus term: anomalies flagged by multiple distinct mechanisms\n    \"\"\"\n\n    def __init__(self, agreement_weight=0.3):\n        from pyod.models.iforest import IForest\n        from pyod.models.copod import COPOD\n        import numpy as np\n        \n        self.iforest = IForest(random_state=42, contamination=0.1)\n        self.copod = COPOD(contamination=0.1)\n        self.knn = None  # initialized in fit with adaptive k\n        \n        # Consensus enhancement parameter\n        self.agreement_weight = agreement_weight\n        \n        # Robust normalization parameters\n        self.iforest_median = None\n        self.iforest_iqr = None\n        self.copod_median = None\n        self.copod_iqr = None\n        self.knn_median = None\n        self.knn_iqr = None\n        \n    def fit(self, X):\n        import numpy as np\n        from pyod.models.knn import KNN\n        \n        # Adaptive k for k-NN based on data size\n        n_samples = X.shape[0]\n        k = min(10, int(np.sqrt(n_samples)))\n        self.knn = KNN(n_neighbors=k, contamination=0.1)\n        \n        # Fit all detectors\n        self.iforest.fit(X)\n        self.copod.fit(X)\n        self.knn.fit(X)\n        \n        # Get training scores for robust normalization\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Compute median and IQR (75th - 25th percentile)\n        def robust_params(scores):\n            q75, q25 = np.percentile(scores, [75, 25])\n            iqr = q75 - q25\n            if iqr == 0:\n                iqr = 1.0\n            median = np.median(scores)\n            return median, iqr\n        \n        self.iforest_median, self.iforest_iqr = robust_params(iforest_scores)\n        self.copod_median, self.copod_iqr = robust_params(copod_scores)\n        self.knn_median, self.knn_iqr = robust_params(knn_scores)\n        \n        return self\n        \n    def decision_function(self, X):\n        import numpy as np\n        \n        # Get raw scores from each detector\n        iforest_scores = self.iforest.decision_function(X)\n        copod_scores = self.copod.decision_function(X)\n        knn_scores = self.knn.decision_function(X)\n        \n        # Robust normalization: (score - median) / IQR\n        iforest_norm = (iforest_scores - self.iforest_median) / self.iforest_iqr\n        copod_norm = (copod_scores - self.copod_median) / self.copod_iqr\n        knn_norm = (knn_scores - self.knn_median) / self.knn_iqr\n        \n        # Ensure non-negative for geometric mean (clip at small positive value)\n        iforest_pos = np.clip(iforest_norm, 1e-8, None)\n        copod_pos = np.clip(copod_norm, 1e-8, None)\n        knn_pos = np.clip(knn_norm, 1e-8, None)\n        \n        # Maximum component (any detector flags it)\n        max_component = np.maximum.reduce([iforest_norm, copod_norm, knn_norm])\n        \n        # Geometric mean component (consensus among detectors)\n        # Use cube root of product for geometric mean\n        geom_component = (iforest_pos * copod_pos * knn_pos) ** (1/3)\n        \n        # Combined score: max + weighted geometric mean\n        scores = max_component + self.agreement_weight * geom_component\n        \n        return scores"
7}