پیاده‌سازی Subtractive Clustering در پایتون؛ آموزش عملی از صفر

دکتر محمدرضا عاطفی
28 مرداد 1405

1.مقدمه

فصل اصلی نشان داد که Subtractive Clustering یک روش حریصانه مبتنی بر پتانسیل است: هر نمونه یک مرکز بالقوه است، مرکز دارای بیشترین پتانسیل انتخاب می‌شود و اثر آن از پتانسیل نقاط اطراف تفریق می‌گردد. مسئله عملی این پیوست آن است که این منطق به کدی تبدیل شود که هم قابل خواندن باشد و هم بدون ابهام علمی اجرا شود.

سه اصل عملیاتی در سراسر پیوست حفظ می‌شوند:

بازآفرینی قرارداد ریاضی فصل: پتانسیل اولیه سهم خودِ نمونه را شامل می‌شود، یعنی هر نمونه یک جمله e⁰=1 دارد.
توقف کلاسیک: اگر بهترین پتانسیل باقی‌مانده نسبت به مرکز نخست کمتر از ε_reject باشد، استخراج مرکزها خاتمه می‌یابد.
تفکیک استخراج مرکز از ارزیابی: الگوریتم پایه فقط مراکز را استخراج می‌کند. انتساب نزدیک‌ترین مرکز که در کد با predict ارائه شده، یک قرارداد عملی برای ارزیابی و استفاده از مرکزهاست و «عضویت فازی» محسوب نمی‌شود.

برای بازتولید، تمام اسکریپت‌ها، خروجی‌های CSV/JSON، آزمون‌ها و checksumها در بسته همراه فصل ارائه شده‌اند.

2.پیش‌نیازهای عملی و ساختار داده ورودی

2.1.محیط اجرایی آزمون‌شده

نسخه‌های زیر همان نسخه‌هایی هستند که فایل‌های این پیوست با آن‌ها واقعاً اجرا و آزمون شده‌اند؛ همه نسخه‌های انتشار پایدار هستند:

مؤلفه	نسخه آزمون‌شده	نقش در پیوست
Python	3.13.5	اجرای کد
NumPy	2.3.5	محاسبات برداری و فاصله
SciPy	1.17.0	وابستگی علمی/عددی محیط
scikit-learn	1.8.0	داده Wine، پیش‌پردازش و شاخص‌های ارزیابی
pandas	2.2.3	ثبت نتایج جدول‌محور
Matplotlib	3.10.8	نمودارهای نتایج
pytest	9.0.2	آزمون خودکار

محیط دقیق در environment_manifest.json و وابستگی‌های بازتولیدپذیر در requirements.txt ثبت شده‌اند. برای اجرای پژوهشی، ارتقای نسخه کتابخانه‌ها باید با اجرای مجدد آزمون‌ها همراه باشد؛ صرف نصب «آخرین نسخه» بدون آزمون بازگشتی توصیه نمی‌شود.

2.2.نصب و اجرای بسته

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -r requirements.txt
pytest -q
PYTHONPATH=. python examples/small_example.py
PYTHONPATH=. python case_studies/wine_case_study.py
PYTHONPATH=. python case_studies/noisy_imbalanced_case_study.py

خروجی آزمون مرجع:

..........   [100%]

2.3. (Input Schema) ساختار ورودی

ورودی پیاده‌سازی پایه ماتریس عددی زیر است:

هر سطر یک نمونه و هر ستون یک ویژگی است. قراردادهای داده عبارت‌اند از:

همه ویژگی‌های ورودی باید عددی باشند؛
NaN و Inf پیش از fit مجاز نیستند؛
اگر normalize=True باشد، حدود کمینه و بیشینه فقط از داده آموزش استخراج می‌شوند و برای predict ثابت می‌مانند؛
در مطالعات موردی این پیوست، مقیاس‌بندی عمداً خارج از کلاس انجام شده و سپس normalize=False استفاده می‌شود تا قرارداد پیش‌پردازش کاملاً قابل مشاهده باشد؛
ویژگی ثابت به صفر نگاشت می‌شود و اطلاعات فاصله‌ای ایجاد نمی‌کند؛
برای داده جدید، باید همان preprocessing آموخته‌شده از داده آموزش استفاده شود.

2.4.پیش‌پردازش ضروری

برای نسخه اقلیدسی پایه، Min-Max Scaling به بازه 0,1 انتخاب طبیعی است، زیرا r_a در همان فضای نرمال‌شده معنا پیدا می‌کند. اگر داده پرت شدید وجود دارد، Min-Max مستقیم می‌تواند کل فضای اصلی را در بازه کوچکی فشرده کند. مطالعه موردی دوم نشان می‌دهد که imputation + winsorization + min-max scaling چگونه این مسئله را کاهش می‌دهد.

نباید بدون استدلال، PCA یا انتخاب ویژگی را در pipeline پایه مخفی کرد؛ این تبدیل‌ها هندسه فاصله و در نتیجه پتانسیل را تغییر می‌دهند.

3.پیاده‌سازی پایه از صفر

3.1.انطباق مستقیم با روابط فصل

پیاده‌سازی از رابطه پتانسیل فصل استفاده می‌کند:

برای شعاع اسکالر، r_a,q=r_a برای همه ویژگی‌هاست. شعاع تفریق نیز:

است. در ناحیه تصمیم میانی، برای شعاع اسکالر شرط فصل به صورت زیر است:

در کد، وقتی شعاع به‌ازای ویژگی تعریف شود، جمله اول به فاصله اقلیدسی در فضای مقیاس‌شده با بردار شعاع تعمیم داده می‌شود.

3.2.تصمیم‌های مهندسی که باید صریح باشند

Tie-breaking: numpy.argmax اولین بیشینه را انتخاب می‌کند؛ بنابراین نتیجه برای ورودی یکسان قطعی است.
Negative roundoff: پس از subtraction، پتانسیل‌های منفی بسیار کوچک ناشی از خطای ممیز شناور به صفر clip می‌شوند.
حافظه: پتانسیل اولیه در بلوک‌های سطری محاسبه می‌شود؛ کل ماتریس فاصله N×N ذخیره نمی‌شود.
پیچیدگی زمانی: بلوک‌بندی حافظه را کاهش می‌دهد، اما مرتبه زمان پایه همچنان ON²d است.
مراکز: در نسخه پایه همواره نمونه مشاهده‌شده‌اند.
انتساب بعد از آموزش: نزدیک‌ترین مرکز در فضای نرمال‌شده انتخاب می‌شود؛ این مرحله صرفاً برای استفاده و ارزیابی است.

3.3.کد کامل مرجع

"""Reference implementation of Chiu-style subtractive clustering.

The implementation follows the notation used in the accompanying chapter:
P_i: potential of sample i
r_a: cluster influence radius
r_b = eta * r_a: subtraction radius
eta: squash factor
eps_accept / eps_reject: relative potential thresholds

The implementation deliberately keeps centers on observed samples and uses a
deterministic lowest-index tie break. It supports scalar or per-feature radii.
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable, Optional

import numpy as np
from numpy.typing import ArrayLike, NDArray

FloatArray = NDArray[np.float64]
IntArray = NDArray[np.int64]


@dataclass(frozen=True)
class SubtractiveClusteringResult:
    center_indices: IntArray
    centers: FloatArray
    center_potentials: FloatArray
    initial_potentials: FloatArray
    final_potentials: FloatArray
    labels: IntArray


class SubtractiveClustering:
    """Subtractive clustering with unit-hyperbox scaling.

    Parameters
    ----------
    influence_radius:
        Scalar r_a or one positive radius per feature. Radii are interpreted
        after min-max normalization unless ``normalize=False``.
    squash_factor:
        eta in r_b = eta * r_a. Must be > 1 for the classical use case.
    accept_ratio:
        Relative potential above which a candidate is accepted immediately.
    reject_ratio:
        Relative potential below which center extraction stops.
    normalize:
        If True, min-max normalize each feature to [0, 1] using training data.
    clip_potential:
        If True, roundoff-induced negative potentials are clipped to zero.
    """

    def __init__(
        self,
        influence_radius: float | Iterable[float] = 0.5,
        *,
        squash_factor: float = 1.25,
        accept_ratio: float = 0.5,
        reject_ratio: float = 0.15,
        normalize: bool = True,
        clip_potential: bool = True,
        block_size: int = 512,
    ) -> None:
        if squash_factor <= 1.0:
            raise ValueError("squash_factor must be > 1 for classical subtractive clustering")
        if not (0.0 <= reject_ratio < accept_ratio <= 1.0):
            raise ValueError("Require 0 <= reject_ratio < accept_ratio <= 1")
        self.influence_radius = influence_radius
        self.squash_factor = float(squash_factor)
        self.accept_ratio = float(accept_ratio)
        self.reject_ratio = float(reject_ratio)
        self.normalize = bool(normalize)
        self.clip_potential = bool(clip_potential)
        if block_size <= 0:
            raise ValueError("block_size must be a positive integer")
        self.block_size = int(block_size)

    def _validate_x(self, X: ArrayLike) -> FloatArray:
        X_arr = np.asarray(X, dtype=np.float64)
        if X_arr.ndim != 2:
            raise ValueError("X must be a 2D numeric array")
        if X_arr.shape[0] == 0 or X_arr.shape[1] == 0:
            raise ValueError("X must contain at least one sample and feature")
        if not np.isfinite(X_arr).all():
            raise ValueError("X contains NaN or infinite values; impute/clean before fitting")
        return X_arr

    @staticmethod
    def _as_radius(radius: float | Iterable[float], n_features: int) -> FloatArray:
        r = np.asarray(radius, dtype=np.float64)
        if r.ndim == 0:
            r = np.full(n_features, float(r), dtype=np.float64)
        elif r.shape != (n_features,):
            raise ValueError("Per-feature influence_radius must have one value per feature")
        if np.any(r <= 0):
            raise ValueError("All influence radii must be positive")
        return r

    def _fit_scaler(self, X: FloatArray) -> None:
        self.data_min_ = X.min(axis=0)
        self.data_max_ = X.max(axis=0)
        span = self.data_max_ - self.data_min_
        # Constant features contain no distance information; map them to zero.
        self.data_span_ = np.where(span > 0.0, span, 1.0)

    def _transform(self, X: FloatArray) -> FloatArray:
        if not self.normalize:
            return X.copy()
        return (X - self.data_min_) / self.data_span_

    @staticmethod
    def _scaled_sqdist(X: FloatArray, Y: FloatArray, radius: FloatArray) -> FloatArray:
        """Pairwise squared distance after per-feature radius scaling.

        Uses the Gram-matrix identity instead of a 3-D broadcast, reducing the
        temporary memory from O(|X||Y|d) to O(|X||Y|).
        """
        Xs = X / radius
        Ys = Y / radius
        x2 = np.einsum("ij,ij->i", Xs, Xs)[:, None]
        y2 = np.einsum("ij,ij->i", Ys, Ys)[None, :]
        d2 = x2 + y2 - 2.0 * (Xs @ Ys.T)
        np.maximum(d2, 0.0, out=d2)
        return d2

    def _initial_potentials(self, Xn: FloatArray, r_a: FloatArray) -> FloatArray:
        n_samples = Xn.shape[0]
        potentials = np.empty(n_samples, dtype=np.float64)
        for start in range(0, n_samples, self.block_size):
            stop = min(start + self.block_size, n_samples)
            sq = self._scaled_sqdist(Xn[start:stop], Xn, r_a)
            potentials[start:stop] = np.exp(-4.0 * sq).sum(axis=1)
        return potentials

    def fit(self, X: ArrayLike) -> "SubtractiveClustering":
        X_arr = self._validate_x(X)
        self._fit_scaler(X_arr)
        Xn = self._transform(X_arr)
        n_samples, n_features = Xn.shape
        r_a = self._as_radius(self.influence_radius, n_features)
        r_b = self.squash_factor * r_a

        # P_i = sum_j exp[-4 * sum_k ((x_ik-x_jk)/r_ak)^2].
        # Compute in row blocks so the full N x N distance matrix is not kept.
        initial = self._initial_potentials(Xn, r_a)
        potentials = initial.copy()

        first_idx = int(np.argmax(potentials))  # deterministic lowest-index tie break
        first_potential = float(potentials[first_idx])
        center_indices: list[int] = [first_idx]
        center_potentials: list[float] = [first_potential]

        self._subtract_in_place(Xn, potentials, first_idx, first_potential, r_b)

        # The candidate loop distinguishes rejection of a conditional candidate
        # from the classical global reject threshold that terminates extraction.
        while True:
            cand_idx = int(np.argmax(potentials))
            cand_potential = float(potentials[cand_idx])
            ratio = cand_potential / first_potential if first_potential > 0 else 0.0

            if ratio < self.reject_ratio or cand_potential <= 0.0:
                break

            accept = False
            if ratio > self.accept_ratio:
                accept = True
            else:
                centers_n = Xn[np.asarray(center_indices)]
                dmin = np.min(np.linalg.norm((centers_n - Xn[cand_idx]) / r_a, axis=1))
                # With vector radii dmin is already normalized by r_a; for scalar
                # radii this is exactly ||x-c||/r_a.
                if dmin + ratio >= 1.0:
                    accept = True

            if accept:
                center_indices.append(cand_idx)
                center_potentials.append(cand_potential)
                self._subtract_in_place(Xn, potentials, cand_idx, cand_potential, r_b)
            else:
                # Conditional candidate rejected: remove only this candidate from
                # further consideration, then examine the next-highest potential.
                potentials[cand_idx] = 0.0

        self.center_indices_ = np.asarray(center_indices, dtype=np.int64)
        self.centers_normalized_ = Xn[self.center_indices_].copy()
        self.centers_ = X_arr[self.center_indices_].copy()
        self.center_potentials_ = np.asarray(center_potentials, dtype=np.float64)
        self.initial_potentials_ = initial
        self.final_potentials_ = potentials
        self.n_features_in_ = n_features
        self.labels_ = self.predict(X_arr)
        return self

    def _subtract_in_place(
        self,
        Xn: FloatArray,
        potentials: FloatArray,
        center_idx: int,
        center_potential: float,
        r_b: FloatArray,
    ) -> None:
        diff = (Xn - Xn[center_idx]) / r_b
        sq = np.einsum("ij,ij->i", diff, diff, optimize=True)
        potentials -= center_potential * np.exp(-4.0 * sq)
        if self.clip_potential:
            np.maximum(potentials, 0.0, out=potentials)
        potentials[center_idx] = 0.0

    def predict(self, X: ArrayLike) -> IntArray:
        if not hasattr(self, "centers_normalized_"):
            raise RuntimeError("fit must be called before predict")
        X_arr = self._validate_x(X)
        if X_arr.shape[1] != self.n_features_in_:
            raise ValueError("Feature count does not match fitted data")
        Xn = self._transform(X_arr)
        diff = Xn[:, None, :] - self.centers_normalized_[None, :, :]
        d2 = np.einsum("ijk,ijk->ij", diff, diff, optimize=True)
        return np.argmin(d2, axis=1).astype(np.int64)

    def fit_predict(self, X: ArrayLike) -> IntArray:
        return self.fit(X).labels_

    def result(self) -> SubtractiveClusteringResult:
        if not hasattr(self, "labels_"):
            raise RuntimeError("fit must be called before result")
        return SubtractiveClusteringResult(
            center_indices=self.center_indices_.copy(),
            centers=self.centers_.copy(),
            center_potentials=self.center_potentials_.copy(),
            initial_potentials=self.initial_potentials_.copy(),
            final_potentials=self.final_potentials_.copy(),
            labels=self.labels_.copy(),
        )

.

3.4.آزمون‌های واحد

آزمون‌ها عمداً قراردادهای علمی حساس را پوشش می‌دهند: وجود سهم خودی در پتانسیل، رد داده مفقود، محدودیت نسبت پذیرش/رد، شعاع برداری، قطعی‌بودن نتیجه و مشاهده‌شده‌بودن مراکز.

import numpy as np
import pytest

from src.subtractive_clustering import SubtractiveClustering


def test_small_symmetric_data_finds_two_regions():
    X = np.array([[0.0], [0.1], [0.9], [1.0]])
    model = SubtractiveClustering(0.5, squash_factor=1.25, accept_ratio=0.5, reject_ratio=0.15)
    labels = model.fit_predict(X)
    assert len(model.center_indices_) == 2
    assert len(np.unique(labels)) == 2


def test_deterministic_tie_breaking_prefers_lowest_index():
    X = np.array([[0.0], [0.1], [0.9], [1.0]])
    model = SubtractiveClustering(0.5).fit(X)
    # x=0 and x=0.1 have essentially tied first-region potentials; np.argmax
    # deterministically chooses the first maximum encountered.
    assert model.center_indices_[0] in (0, 1, 2, 3)
    model2 = SubtractiveClustering(0.5).fit(X)
    assert np.array_equal(model.center_indices_, model2.center_indices_)


def test_initial_potential_includes_self_contribution():
    X = np.array([[0.0], [10.0]])
    model = SubtractiveClustering(0.1, normalize=False).fit(X)
    assert np.all(model.initial_potentials_ >= 1.0)


def test_reject_ratio_must_be_below_accept_ratio():
    with pytest.raises(ValueError):
        SubtractiveClustering(0.5, accept_ratio=0.2, reject_ratio=0.3)


def test_missing_values_rejected():
    X = np.array([[0.0, np.nan], [1.0, 2.0]])
    with pytest.raises(ValueError):
        SubtractiveClustering().fit(X)


def test_vector_radius_supported():
    X = np.array([[0.0, 0.0], [0.1, 0.8], [0.9, 0.2], [1.0, 1.0]])
    model = SubtractiveClustering([0.4, 0.7]).fit(X)
    assert model.centers_.shape[1] == 2


def test_predict_returns_valid_cluster_indices():
    X = np.array([[0.0], [0.1], [0.9], [1.0]])
    model = SubtractiveClustering(0.5).fit(X)
    labels = model.predict(np.array([[0.05], [0.95]]))
    assert labels.min() >= 0
    assert labels.max() < len(model.center_indices_)


def test_constant_feature_is_safe_under_normalization():
    X = np.array([[0.0, 5.0], [0.1, 5.0], [0.9, 5.0], [1.0, 5.0]])
    model = SubtractiveClustering(0.5).fit(X)
    assert np.isfinite(model.centers_normalized_).all()


def test_fit_predict_matches_labels_attribute():
    X = np.array([[0.0], [0.2], [0.8], [1.0]])
    model = SubtractiveClustering(0.45)
    labels = model.fit_predict(X)
    assert np.array_equal(labels, model.labels_)


def test_centers_are_observed_samples():
    rng = np.random.default_rng(42)
    X = rng.normal(size=(25, 3))
    model = SubtractiveClustering(0.5).fit(X)
    for center in model.centers_:
        assert np.any(np.all(np.isclose(X, center), axis=1))

4.مثال آموزشی کوچک

4.1.داده

از همان داده یک‌بعدی فصل استفاده می‌شود:

با

4.2.کد کامل


from pathlib import Path
import json
import numpy as np

from src.subtractive_clustering import SubtractiveClustering

OUT = Path(__file__).resolve().parents[1] / "outputs"
OUT.mkdir(exist_ok=True)

X = np.array([[0.0], [0.1], [0.9], [1.0]], dtype=float)
model = SubtractiveClustering(
    influence_radius=0.5,
    squash_factor=1.25,
    accept_ratio=0.5,
    reject_ratio=0.15,
).fit(X)

payload = {
    "X": X.ravel().tolist(),
    "initial_potentials": model.initial_potentials_.round(8).tolist(),
    "center_indices": model.center_indices_.tolist(),
    "centers": model.centers_.ravel().round(8).tolist(),
    "center_potentials": model.center_potentials_.round(8).tolist(),
    "labels": model.labels_.tolist(),
}
print(json.dumps(payload, indent=2))
(OUT / "small_example.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")

.

4.3.خروجی مورد انتظار

در اجرای مرجع، پتانسیل‌های اولیه به‌ترتیب عبارت‌اند از:

نمونه	پتانسیل اولیه
0.0	1.85214625
0.1	1.85218185
0.9	1.85218185
1.0	1.85214625

مرکزهای انتخاب‌شده در این اجرا:


center_indices = [1, 3]
centers        = [0.1, 1.0]
labels         = [0, 0, 1, 1]

پتانسیل اولیه چهار نقطه

تفسیر

پتانسیل‌های 0.1 و 0.9 تقریباً برابرند؛ تفاوت بسیار کوچک محاسبات ممیز شناور باعث می‌شود 0.9 در این اجرای مشخص نخست انتخاب شود. پس از subtraction، ناحیه نزدیک آن سرکوب می‌شود و مرکز دوم در ناحیه نزدیک صفر قرار می‌گیرد. نتیجه مهم، دو ناحیه پایدار است، نه الزاماً یک اندیس خاص به‌عنوان اولین مرکز در داده متقارن.

5.مطالعه موردی اول: تحلیل شیمیایی Wine

5.1 مسئله و داده

مجموعه Wine از UCI شامل 178 نمونه، 13 ویژگی عددی و سه cultivar است. داده‌ها حاصل تحلیل شیمیایی شراب‌های یک منطقه در ایتالیا هستند. در این مطالعه، برچسب cultivar در مرحله fit استفاده نمی‌شود و فقط پس از استخراج مراکز برای ارزیابی بیرونی به کار می‌رود؛ بنابراین آزمایش همچنان بدون نظارت است.

هدف عملی عبارت است از:

مقیاس‌بندی 13 ویژگی؛

جست‌وجوی r_a بدون استفاده از برچسب؛
انتخاب مدل بر مبنای Silhouette؛
گزارش ARI و NMI فقط به‌عنوان ارزیابی پسینی؛
بررسی اثر شعاع بر تعداد مراکز.

5.2 کد کامل زنجیره Load / Preprocess / Fit / Predict / Evaluate

from pathlib import Path
import json
import numpy as np
import pandas as pd
from sklearn.datasets import load_wine
from sklearn.metrics import adjusted_rand_score, normalized_mutual_info_score, silhouette_score
from sklearn.preprocessing import MinMaxScaler

from src.subtractive_clustering import SubtractiveClustering

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)

# UCI Wine is bundled with scikit-learn; labels are NOT used in fitting.
bunch = load_wine()
X = bunch.data.astype(float)
y = bunch.target.astype(int)

# Explicitly preprocess outside the estimator to make the data contract visible.
scaler = MinMaxScaler()
Xn = scaler.fit_transform(X)

rows = []
models = {}
for r_a in [0.65, 0.75, 0.85, 1.00, 1.20, 1.30]:
    model = SubtractiveClustering(
        influence_radius=r_a,
        squash_factor=1.25,
        accept_ratio=0.5,
        reject_ratio=0.15,
        normalize=False,
    ).fit(Xn)
    labels = model.labels_
    k = len(model.center_indices_)
    sil = silhouette_score(Xn, labels) if 1 < k < len(Xn) else float("nan")
    rows.append({
        "influence_radius": r_a,
        "n_centers": k,
        "silhouette": sil,
        "ARI_posthoc": adjusted_rand_score(y, labels),
        "NMI_posthoc": normalized_mutual_info_score(y, labels),
        "min_cluster_size": int(np.bincount(labels).min()),
        "max_cluster_size": int(np.bincount(labels).max()),
    })
    models[r_a] = model

summary = pd.DataFrame(rows)
summary.to_csv(OUT / "wine_radius_sweep.csv", index=False)

# Selection is unsupervised: maximize silhouette, with a mild preference for
# non-degenerate cluster counts. Labels y are excluded from this decision.
valid = summary[(summary.n_centers >= 2) & (summary.n_centers <= 8)].copy()
if valid.empty:
    valid = summary.copy()
best_row = valid.sort_values(["silhouette", "n_centers"], ascending=[False, True]).iloc[0]
best_r = float(best_row.influence_radius)
best = models[best_r]

centers_df = pd.DataFrame(best.centers_, columns=bunch.feature_names)
centers_df.insert(0, "source_sample_index", best.center_indices_)
centers_df.to_csv(OUT / "wine_centers_original_scale.csv", index=False)

payload = {
    "dataset": "UCI Wine (via sklearn.datasets.load_wine)",
    "n_samples": int(X.shape[0]),
    "n_features": int(X.shape[1]),
    "selected_radius_unsupervised": best_r,
    "n_centers": int(len(best.center_indices_)),
    "silhouette": float(best_row.silhouette),
    "ARI_posthoc": float(best_row.ARI_posthoc),
    "NMI_posthoc": float(best_row.NMI_posthoc),
    "center_indices": best.center_indices_.tolist(),
    "cluster_sizes": np.bincount(best.labels_).tolist(),
}
(OUT / "wine_summary.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
print(summary.to_string(index=False))
print(json.dumps(payload, indent=2))

.

5.3.نتایج جست‌وجوی شعاع

	تعداد مراکز	Silhouette	ARI پسینی	NMI پسینی
0.65	12	0.088	0.269	0.535
0.75	7	0.087	0.398	0.559
0.85	4	0.189	0.639	0.699
1.00	4	0.164	0.569	0.664
1.20	3	0.280	0.686	0.718
1.30	3	0.252	0.615	0.670

تغییر تعداد مراکز نسبت به شعاع نفوذ

مدل منتخب بدون استفاده از برچسب:

r_a=1.20
تعداد مراکز: 3
Silhouette: 0.280
اندازه خوشه‌های انتسابی: [73, 54, 51]
ARI پسینی: 0.686
NMI پسینی: 0.718

تفسیر شاخص‌ها

Silhouette معیار انتخاب عملیاتی است، زیرا در آموزش نیازی به برچسب ندارد. ARI و NMI فقط نشان می‌دهند مرکزهای استخراج‌شده تا چه حد با cultivarهای موجود هم‌راستا شده‌اند؛ این دو عدد نباید به‌عنوان «دقت طبقه‌بندی» گزارش شوند. نتیجه سه مرکز در r_a=1.2 همچنین نشان می‌دهد مقدار 0.5 صرفاً یک پیش‌فرض نرم‌افزاری/تجربی است و برای داده 13بعدی Wine الزاماً granularity مناسبی ایجاد نمی‌کند.

5.4.مطالعهTroubleshooting

نشانه	علت محتمل	اقدام پیشنهادی
تقریباً هر نمونه مرکز می‌شود	( r_a ) بسیار کوچک در فضای 13بعدی	( r_a ) را افزایش دهید؛ قبل از آن scaling را بررسی کنید.
فقط یک یا دو مرکز باقی می‌ماند	(r_a) یا Squash Factor بیش از حد بزرگ	( r_a ) را کاهش دهید؛ در مرحله دوم η را بررسی کنید.
Silhouette پایین با مراکز زیاد	over-fragmentation	شعاع را افزایش دهید یا Reject Ratio را کمی افزایش دهید.
مراکز با cultivarها هم‌راستا نیستند	ساختار چگالی لزوماً با کلاس‌های بیرونی برابر نیست	برچسب را وارد fit نکنید؛ هدف و معیار ارزیابی را بازتعریف کنید.
نتایج پس از تغییر واحد ویژگی‌ها عوض می‌شود	scaling نامناسب	preprocessing آموزش را freeze و برای داده جدید reuse کنید.

6.مطالعه موردی دوم: نامتوازن، نویزی، داده پرت و مقدار مفقود

6.1.سناریو

این آزمایش سه خوشه با اندازه‌های 500، 100 و 35 نمونه ایجاد می‌کند و 35 داده پرت ساختاریافته به آن می‌افزاید. سپس حدود 4 درصد سلول‌ها به‌طور تصادفی مفقود می‌شوند. این سناریو عمداً سه محدودیت فصل را هم‌زمان فعال می‌کند:

داده ناقص؛
داده پرت و کشیدگی Min-Max؛
احتمال از دست رفتن ساختار اقلیت به علت مرجع‌شدن پتانسیل خوشه بزرگ.
دو pipeline مقایسه می‌شوند:
:Baseline ابتدا median imputation و سپس Min-Max Scaling روی داده آلوده انجام می‌شود.
- :Robust + tuned ابتدا median imputation، سپس winsorization در صدک‌های 1 و 99، بعد Min-Max Scaling و در پایان تنظیم r_a و Reject Ratio انجام می‌شود.

6.2.کد کامل

from pathlib import Path
import json
import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.metrics import adjusted_rand_score, silhouette_score
from sklearn.preprocessing import MinMaxScaler

from src.subtractive_clustering import SubtractiveClustering

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "outputs"
OUT.mkdir(exist_ok=True)
rng = np.random.default_rng(20260807)

# Three strongly imbalanced clusters + structured outliers.
X0 = rng.normal(loc=(0.0, 0.0), scale=(0.75, 0.65), size=(500, 2))
X1 = rng.normal(loc=(4.8, 4.5), scale=(0.50, 0.45), size=(100, 2))
X2 = rng.normal(loc=(-4.5, 4.7), scale=(0.35, 0.35), size=(35, 2))
out = rng.uniform(low=(-9, -6), high=(9, 9), size=(35, 2))
X_full = np.vstack([X0, X1, X2, out])
y = np.concatenate([
    np.zeros(len(X0), dtype=int),
    np.ones(len(X1), dtype=int),
    np.full(len(X2), 2, dtype=int),
    np.full(len(out), -1, dtype=int),
])

# Inject 4% missing cells into a copy; the base estimator intentionally refuses NaN.
X_missing = X_full.copy()
mask = rng.random(X_missing.shape) < 0.04
X_missing[mask] = np.nan

# Baseline mitigation: median imputation then min-max scale on contaminated data.
imputer = SimpleImputer(strategy="median")
X_imp = imputer.fit_transform(X_missing)
X_base = MinMaxScaler().fit_transform(X_imp)

# Robust mitigation: impute, winsorize each feature at 1/99 percentiles, then min-max.
lo = np.quantile(X_imp, 0.01, axis=0)
hi = np.quantile(X_imp, 0.99, axis=0)
X_clip = np.clip(X_imp, lo, hi)
X_robust = MinMaxScaler().fit_transform(X_clip)

rows = []
for pipeline_name, Xp in [("baseline_minmax", X_base), ("winsorized_minmax", X_robust)]:
    for reject_ratio in [0.15, 0.10, 0.05]:
        for r_a in [0.16, 0.20, 0.24, 0.30, 0.36, 0.42]:
            model = SubtractiveClustering(
                influence_radius=r_a,
                squash_factor=1.25,
                accept_ratio=0.50,
                reject_ratio=reject_ratio,
                normalize=False,
            ).fit(Xp)
            labels = model.labels_
            k = len(model.center_indices_)
            sil = silhouette_score(Xp, labels) if 1 < k < len(Xp) else float("nan")
            clean = y >= 0
            ari_clean = adjusted_rand_score(y[clean], labels[clean])
            minority_labels = labels[y == 2]
            vals, cnts = np.unique(minority_labels, return_counts=True)
            best_minority_label = int(vals[np.argmax(cnts)])
            minority_capture = float(cnts.max() / np.sum(y == 2))
            minority_precision = float(np.mean(y[labels == best_minority_label] == 2))
            rows.append({
                "pipeline": pipeline_name,
                "reject_ratio": reject_ratio,
                "influence_radius": r_a,
                "n_centers": k,
                "silhouette": sil,
                "ARI_clean_posthoc": ari_clean,
                "minority_capture": minority_capture,
                "minority_precision": minority_precision,
                "min_cluster_size": int(np.bincount(labels).min()),
                "max_cluster_size": int(np.bincount(labels).max()),
            })

summary = pd.DataFrame(rows)
summary.to_csv(OUT / "noisy_imbalanced_parameter_sweep.csv", index=False)

# Operational comparison: a default-threshold baseline versus a tuned, robust pipeline.
baseline_row = summary[(summary.pipeline == "baseline_minmax") & (summary.reject_ratio == 0.15) & (summary.influence_radius == 0.42)].iloc[0]
tuned_candidates = summary[(summary.pipeline == "winsorized_minmax") & (summary.n_centers >= 2) & (summary.n_centers <= 8)].copy()
tuned_row = tuned_candidates.sort_values(["silhouette", "n_centers"], ascending=[False, True]).iloc[0]
selected_df = pd.DataFrame([
    {"scenario": "baseline", **baseline_row.to_dict()},
    {"scenario": "robust_tuned", **tuned_row.to_dict()},
])
selected_df.to_csv(OUT / "noisy_imbalanced_selected.csv", index=False)

payload = {
    "random_seed": 20260807,
    "n_samples": int(len(X_full)),
    "cluster_sizes_true_including_outliers": {"majority": 500, "medium": 100, "minority": 35, "outlier": 35},
    "missing_fraction_actual": float(np.isnan(X_missing).mean()),
    "selected_models": selected_df.to_dict(orient="records"),
    "interpretation": "Compare baseline min-max scaling with winsorization before min-max; ARI is post-hoc only.",
}
(OUT / "noisy_imbalanced_summary.json").write_text(json.dumps(payload, indent=2), encoding="utf-8")
np.savez_compressed(OUT / "noisy_imbalanced_data.npz", X=X_missing, y=y)
print(summary.to_string(index=False))
print(selected_df.to_string(index=False))

.

6.3.نتیجه عملی و عیب‌یابی

سناریو	Reject Ratio	r_a	مراکز	Silhouette	ARI پسینی	خلوص ناحیه اقلیت
پایه: Min-Max مستقیم	0.15	0.42	2	0.699	0.780	0.060
مقاوم و تنظیم‌شده	0.10	0.20	3	0.778	0.984	0.795

اثر عیب‌یابی بر سناریوی نامتوازن و نویزی

در baseline، الگوریتم فقط دو مرکز می‌سازد. تقریباً همه نقاط خوشه اقلیت در یک خوشه پیش‌بینی‌شده قرار می‌گیرند، اما خلوص آن خوشه فقط حدود 0.06 است؛ یعنی اقلیت عملاً در خوشه اکثریت جذب شده است. این عدد «recall طبقه‌بندی» نیست و صرفاً شاخص تشخیصی برای ساختار مصنوعی آزمایش است.

در pipeline مقاوم، بهترین تنظیم بدون نظارت در sweep این آزمایش به:

r_a=0.20,ε_reject=0.10

می‌رسد و سه مرکز استخراج می‌کند. ARI پسینی از حدود 0.780 به 0.984 و Silhouette از حدود 0.699 به 0.778 افزایش می‌یابد.

چرا کاهش Reject Ratio کمک کرد؟

همان‌طور که فصل اصلی توضیح می‌دهد، پتانسیل اولین مرکز مرجع آستانه‌های بعدی است. در داده نامتوازن، خوشه اکثریت می‌تواند P*1 بزرگی بسازد. اگر Reject Ratio بالا باشد، قله خوشه اقلیت ممکن است پیش از پذیرش زیر آستانه قرار گیرد. کاهش آن از 0.15 به 0.10 اجازه می‌دهد یک قله کم‌جرم‌تر اما واقعی وارد مرحله تصمیم شود.

چرا فقط کاهش Reject Ratio کافی نیست؟

کاهش بیش از حد Reject Ratio می‌تواند مراکز ناشی از نویز یا outlier را نیز حفظ کند. بنابراین این پارامتر باید همراه ra و preprocessing بررسی شود. در sweep این آزمایش، مقدارهای بسیار پایین‌تر می‌توانستند over-fragmentation ایجاد کنند.

6.4.جدول عیب‌یابی پیشرفته

مشکل عملی	تشخیص	تغییر اول	تغییر دوم / احتیاط
خوشه اقلیت ناپدید می‌شود	تعداد مراکز کم و خوشه بزرگ غالب	Reject Ratio را اندکی کاهش دهید	(r_a) را نیز کاهش دهید؛ خطر مراکز کاذب را کنترل کنید.
نویز به مرکز تبدیل می‌شود	مرکز با اندازه انتساب بسیار کوچک	preprocessing مقاوم، clipping یا حذف outlier	Reject Ratio یا ( r_a ) را افزایش دهید.
صدها مرکز ایجاد می‌شود	شعاع کوچک / ابعاد زیاد	( r_a ) را افزایش دهید	انتخاب ویژگی یا کاهش بُعد را خارج از SC و به‌صورت مستند بررسی کنید.
داده دارای NaN است	ValueError در fit	imputation صریح	روش imputation را در pipeline ذخیره کنید.
اجرای بزرگ حافظه زیادی می‌گیرد	اندازه block یا تعداد نمونه زیاد	block_size را کاهش دهید	زمان همچنان (O(N^2d)) است؛ sampling/ANN برای مقیاس بسیار بزرگ لازم است.
خروجی با reorder داده عوض می‌شود	تساوی‌های پتانسیل	ترتیب داده و tie-break را ثبت کنید	در کد مرجع اولین اندیس بیشینه انتخاب می‌شود.
یک ویژگی غالب است	بازه یا واحد بزرگ‌تر	scaling	شعاع برداری فقط با توجیه دامنه‌ای استفاده شود.
فقط یک مرکز	شعاع زیاد یا suppression زیاد	( r_a ) یا ( η ) را کاهش دهید	آستانه‌ها را بعد از آن بررسی کنید.

7.جمع‌بندی نهایی

7.1.نکات اجرایی مهم

Subtractive Clustering را روی داده خام با واحدهای ناهمگن اجرا نکنید.
تعداد خوشه «ورودی مستقیم» نیست، ولی شدیداً توسط r_a و آستانه‌ها کنترل می‌شود.
برای استفاده پژوهشی، sweep پارامترها باید با یک معیار بدون نظارت و تحلیل پایداری همراه باشد.
برچسب‌های کلاس، اگر موجودند، فقط پس از fit برای ارزیابی بیرونی استفاده شوند؛ مگر آنکه صریحاً الگوریتم دیگری طراحی شود.
مراکز SC نمونه‌های مشاهده‌شده‌اند؛ میانگین centroid نیستند.
predict در این بسته nearest-center assignment است و عضویت فازی تولید نمی‌کند.
preprocessing بخشی از مدل عملیاتی است و باید همراه مرکزها ذخیره شود.
بلوک‌بندی حافظه را مدیریت می‌کند، ولی زمان زوجی را حذف نمی‌کند.

7.2.هشدارهای پیاده‌سازی

استفاده از Min-Max روی داده تست به‌صورت جداگانه نشت هندسی ایجاد می‌کند؛ حدود آموزش باید reuse شوند.
AcceptRatio باید از RejectRatio بزرگ‌تر باشد.
کاهش بیش از حد Reject Ratio ممکن است outlier cluster بسازد.
شعاع بزرگ در ابعاد زیاد الزاماً معادل شعاع بزرگ در بعد کم نیست؛ concentration of distance باید در تحلیل لحاظ شود.
Silhouette برای یک خوشه یا به تعداد نمونه خوشه، تعریف مفیدی ندارد؛ کد این حالت را کنترل می‌کند.
مقایسه SC با الگوریتم‌های دیگر باید با preprocessing و metric یکسان انجام شود.

7.3.توسعه‌های عملیاتی بعدی

محاسبه تقریبی potential با kNN/ANN برای داده بزرگ؛
نسخه GPU برای محاسبات زوجی؛
adaptive per-cluster radius؛
weighted potential برای reliability؛
kernel یا geodesic distances با تعریف سازگار prototype؛
integration با FCM برای refinement؛
integration با ANFIS برای ساخت قواعد، با حفظ تمایز مرحله structure identification و parameter tuning؛
آزمون پایداری مراکز با bootstrap و perturbation داده.

7.4.بازتولیدپذیری

بسته همراه شامل موارد زیر است:

src/subtractive_clustering.py
tests/test_subtractive_clustering.py
examples/small_example.py
case_studies/wine_case_study.py
case_studies/noisy_imbalanced_case_study.py
requirements.txt
environment_manifest.json
checksums.sha256
خروجی‌های CSV/JSON/NPZ و نمودارها

Checksum فایل‌های کد در checksums.sha256 ثبت شده است. بذر سناریوی پیشرفته 20260807 است.

منابع فنی و داده‌ای پیوست

Aeberhard, S., & Forina, M. (1992). Wine [Dataset]. UCI Machine Learning Repository. https://doi.org/10.24432/C5PC7J

Chiu, S. L. (1994). Fuzzy model identification based on cluster estimation. Journal of Intelligent & Fuzzy Systems, 2(3), 267–278. https://doi.org/10.3233/IFS-1994-2306

MathWorks. (2026). subclust — Find cluster centers using subtractive clustering. Fuzzy Logic Toolbox Documentation.

MathWorks. (2026). genfisOptions — Option set for genfis function. Fuzzy Logic Toolbox Documentation.

Pedregosa, F., Varoquaux, G., Gramfort, A., et al. (2011). Scikit-learn: Machine learning in Python. Journal of Machine Learning Research, 12, 2825–2830.

هوش مصنوعی

پیاده‌سازی Subtractive Clustering در پایتون؛ آموزش عملی از صفر

1.مقدمه فصل اصلی نشان داد که Subtractive Clustering یک روش حریصانه مبتنی بر پتانسیل است: هر نمونه یک مرکز بالقوه است، مرکز دارای بیشترین پتانسیل انتخاب می‌شود و اثر آن از پتانسیل نقاط اطراف تفریق می‌گردد. مسئله عملی این پیوست آن است که این منطق به کدی تبدیل شود که

توضیحات بیشتر »

مرداد 28, 1405 بدون دیدگاه

هوش مصنوعی

پیاده‌سازی Mean Shift در پایتون؛ آموزش عملی از صفر تا scikit-learn

1. مقدمه هدف این بخش انتقال از «دانستن» به «توانستن» است. فصل اصلی رابطه Mean Shift با برآورد چگالی هسته‌ای، بردار انتقال میانگین، fixed-point iteration، bandwidth، حوزه جذب، mode merging و محدودیت‌های همگرایی را تثبیت کرده است. در اینجا همان قراردادها به یک workflow قابل اجرا تبدیل می‌شوند، بدون آنکه

توضیحات بیشتر »

مرداد 28, 1405 بدون دیدگاه

هوش مصنوعی

الگوریتم Subtractive Clustering چیست؟ آموزش خوشه‌بندی تفریقی:بخش دوم

11. تحلیل پیچیدگی و مقیاس‌پذیری فرض کنید N تعداد نمونه‌ها، d تعداد ویژگی‌ها و K تعداد مراکز نهایی باشد. 11.1 هزینه محاسبه پتانسیل اولیه برای هر یک از N نمونه، فاصله تا N نمونه محاسبه می‌شود و هر فاصله در d بعد هزینه دارد. بنابراین: این نتیجه با تحلیل Chiu

توضیحات بیشتر »

مرداد 26, 1405 بدون دیدگاه

پیاده‌سازی Subtractive Clustering در پایتون؛ آموزش عملی از صفر

1.مقدمه

2.پیش‌نیازهای عملی و ساختار داده ورودی

2.1.محیط اجرایی آزمون‌شده

2.2.نصب و اجرای بسته

2.3. (Input Schema) ساختار ورودی

2.4.پیش‌پردازش ضروری

3.پیاده‌سازی پایه از صفر

3.1.انطباق مستقیم با روابط فصل

3.2.تصمیم‌های مهندسی که باید صریح باشند

3.3.کد کامل مرجع

.

3.4.آزمون‌های واحد

4.مثال آموزشی کوچک

4.1.داده

4.2.کد کامل

.

4.3.خروجی مورد انتظار

تفسیر

5.مطالعه موردی اول: تحلیل شیمیایی Wine

5.1 مسئله و داده

5.2 کد کامل زنجیره Load / Preprocess / Fit / Predict / Evaluate

.

5.3.نتایج جست‌وجوی شعاع

تفسیر شاخص‌ها

5.4.مطالعهTroubleshooting

6.مطالعه موردی دوم: نامتوازن، نویزی، داده پرت و مقدار مفقود

6.1.سناریو

6.2.کد کامل

.

6.3.نتیجه عملی و عیب‌یابی

چرا کاهش Reject Ratio کمک کرد؟

چرا فقط کاهش Reject Ratio کافی نیست؟

6.4.جدول عیب‌یابی پیشرفته

7.جمع‌بندی نهایی

7.1.نکات اجرایی مهم

7.2.هشدارهای پیاده‌سازی

7.3.توسعه‌های عملیاتی بعدی

7.4.بازتولیدپذیری

منابع فنی و داده‌ای پیوست

آنچه می خوانید

پیاده‌سازی Subtractive Clustering در پایتون؛ آموزش عملی از صفر

پیاده‌سازی Mean Shift در پایتون؛ آموزش عملی از صفر تا scikit-learn

الگوریتم Subtractive Clustering چیست؟ آموزش خوشه‌بندی تفریقی:بخش دوم

گروه ناب