机器人焊接气孔缺陷挖掘——用关联规则与分类模型找出"谁在制造气孔"

 

一、实际应用场景（真实痛点）

 

时间：周四下午 14:30，焊接车间质量复盘。

 

地点：某压力容器制造厂焊接车间，6 台 6 轴弧焊机器人工作站一字排开，防护帘半拉着，里面闪着弧光。质检台上堆着 23 个刚焊完的法兰接管接头，X 光底片挂在灯箱上，密密麻麻的白色斑点像星空。

 

焊接班长老曹把一张 X 光评片报告甩在桌上："这周第三批了，气孔还是没压住。23 个接头，9 个有气孔超标，返修率 39%。一个接头返修成本 420 块，加上耽误的交期——老板让我写报告解释为什么。"

 

"你之前怎么调的？"工艺员小杨问。

 

"我调了啊！"老曹翻出焊接参数记录，"电流从 180A 降到 160A，焊接速度从 35 cm/min 提到 42 cm/min，保护气流量从 15 L/min 加到 18——按道理气孔应该少，结果这批反而多了。我怀疑是焊丝受潮，但烘了还是一样。"

 

"参数都记了？"我问。

 

"机器人焊接日志全有。"老曹打开 MES 导出表，"每台机器人每道焊缝都记了：电流、电压、焊接速度、送丝速度、保护气流量、层间温度、预热温度、焊丝批次、气体纯度、环境湿度、接头坡口形式……一共 14 个字段，这周 6 台机器人焊了 312 道焊缝，每道都有参数记录。气孔检测结果也有——X 光评片，合格/不合格。问题是：312 行数据，14 个参数，哪个跟气孔有关系？关系是什么？我拿眼睛看 Excel，看了两小时，越看越糊涂。**"

 

"你缺的不是数据，是挖掘。"我说，"你需要一个程序：导入这 312 条焊接记录（参数+气孔结果），用统计方法找出哪些参数跟气孔强相关、关联方向是什么（速度快了气孔多还是少了）、参数之间有没有交互效应（比如'电流大+速度慢'组合特别容易出气孔）、能不能建一个分类模型，给定一组新参数就预测气孔概率。这样下次调参不用靠猜，直接看模型告诉你'这个速度配这个电流，气孔风险 78%'。"

 

"对！"老曹一拍桌子，"最好还能告诉我：如果我把速度从 42 降回 38，气孔率能降几个点？也就是说——参数敏感度排序。"

 

"这活儿我干过。"我说，"pandas 加载焊接日志，numpy 做矩阵运算，scipy 做卡方检验和 ANOVA 分析类别/连续变量与气孔的关联，scikit-learn 的 LogisticRegression + RandomForestClassifier 双模型做气孔风险分类，用关联规则（mlxtend 的 Apriori）挖'参数组合→气孔'的频繁项集，networkx 建'参数→气孔'影响网络，matplotlib 画参数分布对比箱线图、ROC 曲线、关联规则网络图。纯 Python，不依赖任何商业焊接质量系统。"

 

二、痛点分析（映射到滨州职业学院课程模型）

 

《先进制造技术》课程模块 本篇痛点对应

工业机器人技术基础 弧焊机器人工艺参数设定、焊接过程控制

先进制造技术基础 焊接工艺学：气孔形成机理、工艺参数优化

智能制造与数字孪生 焊接质量数据驱动分析、工艺参数虚拟验证

柔性制造系统 FMS 与先进生产管理 过程质量控制、统计过程控制（SPC）

先进制造新模式 数据驱动质量优化、预测性工艺调整

 

一句话总结：本篇解决一个"给定机器人焊接多参数数据集（连续+类别变量）+ 气孔标签 → 统计检验筛选显著参数 → 分类模型量化气孔风险 → 关联规则挖掘危险参数组合 → 可视化影响网络与参数敏感度 → 指导工艺调参"的数据挖掘问题，映射到课程就是焊接工艺参数优化与缺陷成因分析。

 

三、核心逻辑讲解（大白话）

 

3.1 问题本质：把气孔想成"锅里的气泡"

 

焊接熔池就像一锅烧开的水。气泡要逃逸需要时间——如果水面（熔池表面）凝固太快，气泡就被封在里面变成气孔。影响气泡逃逸的因素：

 

- 焊接速度太快 → 熔池停留时间短 → 气泡来不及跑 → 气孔多

- 电流太小 → 热输入不够 → 熔池粘度大 → 气泡跑不动 → 气孔多

- 保护气不好/有风 → 空气混进去 → 氮气孔/一氧化碳孔

- 焊丝受潮 → 水分分解出氢气 → 氢气孔

 

但这些都是"理论上"。实际现场是 14 个参数同时变，你调了速度也调了电流也换了焊丝——到底谁在制造气孔？这就是多维数据中的因果挖掘问题。

 

3.2 业务逻辑 → 代码映射

 

输入: 机器人焊接日志 CSV

  字段: 焊缝ID, 电流(A), 电压(V), 焊接速度(cm/min), 送丝速度(m/min),

        保护气流量(L/min), 层间温度(℃), 预热温度(℃), 焊丝批次,

        气体纯度(%), 环境湿度(%), 坡口形式, 板厚(mm), 气孔结果(0/1)

   │

   ▼ models.py (dataclass)

数据模型:

  WeldRecord(焊接记录) / PorosityResult(气孔结果)

  ParameterInfluence(参数影响度)

   │

   ▼ data_loader.py (pandas)

数据加载器:

  读 CSV → 校验标签列 → 编码类别变量 → 处理缺失值

   │

   ▼ statistical_tester.py

统计检验器 (scipy):

  1. 连续变量 vs 气孔: t-test / Mann-Whitney U (非正态)

  2. 类别变量 vs 气孔: Chi-square / Cramér's V

  3. ANOVA (多类别)

  输出: 每个参数的 p-value + 效应量

   │

   ▼ classifier.py

气孔分类模型 (sklearn):

  1. LogisticRegression (线性可解释)

  2. RandomForestClassifier (捕捉非线性交互)

  3. 交叉验证 + ROC/AUC

   │

   ▼ association_miner.py

关联规则挖掘:

  参数离散化 → 频繁项集 → 置信度/提升度

  挖掘 "速度>40 AND 电流<170 → 气孔=1" 规则

   │

   ▼ network_builder.py (networkx)

影响网络:

  节点 = 参数 + 气孔

  边权重 = 标准化影响度 (统计显著性 + 模型重要性)

   │

   ▼ visualizer.py (matplotlib)

可视化:

  1. 有/无气孔的参数分布箱线图

  2. ROC 曲线对比

  3. 参数重要性条形图

  4. 关联规则网络图

   │

   ▼ 输出

  显著参数列表 + 分类模型 + 危险参数组合 + 调参建议

 

3.3 为什么"凭经验调参"不行

 

维度 经验调参 数据驱动挖掘

参数筛选 "都调调看" 统计检验 p<0.05 的才关注

交互效应 看不到 随机森林 + 关联规则捕捉

调参方向 凭感觉 模型给出"速度每降 1 cm/min，气孔概率降 2.3%"

风险预测 "应该没问题" "这组参数气孔概率 78%"

返修成本 420 元/件 × 批量 提前规避

 

四、OOP 代码实现

 

4.1 项目结构

 

weld_porosity_mining/

├── weld_porosity_mining/

│ ├── __init__.py

│ ├── models.py # 数据模型

│ ├── data_loader.py # CSV加载与校验

│ ├── statistical_tester.py # 统计检验

│ ├── classifier.py # 分类模型

│ ├── association_miner.py # 关联规则挖掘

│ ├── network_builder.py # 影响网络

│ ├── visualizer.py # 可视化

│ └── report_generator.py # 分析报告

├── tests/

│ ├── __init__.py

│ └── test_weld.py

├── sample_data/

│ └── weld_log.csv

├── results/

│ ├── boxplot_params.png

│ ├── roc_curve.png

│ ├── param_importance.png

│ ├── association_network.png

│ └── porosity_report.txt

├── run_weld_mining.py

└── README.md

 

4.2 核心源码

 

<details><summary></summary>

 

"""机器人焊接气孔缺陷数据模型 (dataclass)。"""

 

from dataclasses import dataclass, field

from typing import List, Optional, Dict

import numpy as np

 

 

@dataclass

class WeldRecord:

    """单道焊缝焊接记录。"""

    weld_id: str

    # 连续参数

    current_a: float = 0.0 # 焊接电流 (A)

    voltage_v: float = 0.0 # 电弧电压 (V)

    weld_speed_cm_min: float = 0.0 # 焊接速度 (cm/min)

    wire_feed_m_min: float = 0.0 # 送丝速度 (m/min)

    gas_flow_l_min: float = 0.0 # 保护气流量 (L/min)

    interpass_temp_c: float = 0.0 # 层间温度 (℃)

    preheat_temp_c: float = 0.0 # 预热温度 (℃)

    gas_purity_pct: float = 99.9 # 气体纯度 (%)

    humidity_pct: float = 50.0 # 环境湿度 (%)

    plate_thickness_mm: float = 0.0 # 板厚 (mm)

    # 类别参数

    wire_batch: str = "" # 焊丝批次

    groove_type: str = "" # 坡口形式

    # 结果

    porosity: int = 0 # 气孔 (0=无, 1=有)

 

    def to_feature_array(self, continuous_only: bool = False) -> np.ndarray:

        """转换为特征向量。"""

        continuous = np.array([

            self.current_a, self.voltage_v, self.weld_speed_cm_min,

            self.wire_feed_m_min, self.gas_flow_l_min,

            self.interpass_temp_c, self.preheat_temp_c,

            self.gas_purity_pct, self.humidity_pct, self.plate_thickness_mm

        ])

        return continuous

 

    def to_dict(self) -> dict:

        return {

            "weld_id": self.weld_id,

            "current_a": self.current_a,

            "voltage_v": self.voltage_v,

            "weld_speed_cm_min": self.weld_speed_cm_min,

            "wire_feed_m_min": self.wire_feed_m_min,

            "gas_flow_l_min": self.gas_flow_l_min,

            "interpass_temp_c": self.interpass_temp_c,

            "preheat_temp_c": self.preheat_temp_c,

            "gas_purity_pct": self.gas_purity_pct,

            "humidity_pct": self.humidity_pct,

            "plate_thickness_mm": self.plate_thickness_mm,

            "wire_batch": self.wire_batch,

            "groove_type": self.groove_type,

            "porosity": self.porosity,

        }

 

 

@dataclass

class ParameterInfluence:

    """参数对气孔的影响度。"""

    param_name: str

    p_value: float = 1.0

    effect_size: float = 0.0

    direction: str = "unknown" # "positive"/"negative"/"unknown"

    model_importance: float = 0.0

    test_method: str = ""

 

    @property

    def significance(self) -> str:

        if self.p_value < 0.001:

            return "***"

        elif self.p_value < 0.01:

            return "**"

        elif self.p_value < 0.05:

            return "*"

        else:

            return "n.s."

 

    @property

    def composite_score(self) -> float:

        """综合影响评分。"""

        sig_score = max(0, 1 - self.p_value * 10) # p<0.1 → 正分

        return 0.4 * sig_score + 0.4 * self.model_importance + 0.2 * abs(self.effect_size)

 

</details>

 

<details><summary></summary>

 

"""焊接日志数据加载器 (pandas)。"""

 

import pandas as pd

import numpy as np

from pathlib import Path

from typing import List, Tuple, Dict

from .models import WeldRecord

 

 

class DataLoader:

    """CSV 数据加载与校验。"""

 

    CONTINUOUS_COLUMNS = [

        "current_a", "voltage_v", "weld_speed_cm_min",

        "wire_feed_m_min", "gas_flow_l_min",

        "interpass_temp_c", "preheat_temp_c",

        "gas_purity_pct", "humidity_pct", "plate_thickness_mm"

    ]

 

    CATEGORICAL_COLUMNS = ["wire_batch", "groove_type"]

 

    TARGET_COLUMN = "porosity"

 

    def load_csv(self, csv_path: str

                 ) -> Tuple[np.ndarray, np.ndarray, List[str], Dict]:

        """

        加载焊接日志 CSV。

        返回: X_cont, y, weld_ids, categorical_data

        """

        df = pd.read_csv(csv_path)

 

        # 校验

        required = self.CONTINUOUS_COLUMNS + [self.TARGET_COLUMN]

        missing = [c for c in required if c not in df.columns]

        if missing:

            raise ValueError(f"CSV 缺少列: {missing}")

 

        # 处理缺失值

        df = df.dropna(subset=[self.TARGET_COLUMN])

        df[self.CONTINUOUS_COLUMNS] = df[self.CONTINUOUS_COLUMNS].fillna(

            df[self.CONTINUOUS_COLUMNS].median())

 

        X_cont = df[self.CONTINUOUS_COLUMNS].values.astype(float)

        y = df[self.TARGET_COLUMN].values.astype(int)

        weld_ids = df.get("weld_id",

                         [f"W{i}" for i in range(len(df))]).tolist()

 

        # 类别变量

        cat_data = {}

        for col in self.CATEGORICAL_COLUMNS:

            if col in df.columns:

                cat_data[col] = df[col].fillna("unknown").tolist()

            else:

                cat_data[col] = ["unknown"] * len(df)

 

        return X_cont, y, weld_ids, cat_data

 

    def generate_synthetic_data(self, n_samples: int = 312

                               ) -> Tuple[np.ndarray, np.ndarray]:

        """

        生成合成焊接数据 (教学用)。

        物理模型:

          气孔概率 ∝ 速度过快 + 电流过小 + 湿度过大

                   + 保护气不足 + 交互项

        """

        np.random.seed(42)

 

        records = []

        for i in range(n_samples):

            # 连续参数 (正常范围 + 随机扰动)

            current = np.random.normal(170, 20) # 120~220 A

            voltage = current * 0.04 + np.random.normal(14, 1.5)

            speed = np.random.normal(38, 8) # 20~55 cm/min

            wire_feed = speed * 0.08 + np.random.normal(2, 0.5)

            gas_flow = np.random.normal(16, 3) # 10~25 L/min

            interpass = np.random.normal(120, 30) # 60~200 ℃

            preheat = np.random.normal(80, 20) # 20~150 ℃

            gas_purity = np.random.normal(99.5, 0.3) # 98~100%

            humidity = np.random.normal(55, 15) # 30~85%

            thickness = np.random.choice([6, 8, 10, 12, 16])

 

            # 类别参数

            wire_batch = np.random.choice(

                ["B2026-01", "B2026-02", "B2026-03", "B2025-12"])

            groove = np.random.choice(["V", "U", "X", "单V"])

 

            # 物理模型: 气孔概率 (logistic)

            # 速度过快 → 正贡献

            speed_risk = max(0, (speed - 35) / 20) * 1.2

            # 电流过小 → 正贡献

            current_risk = max(0, (160 - current) / 40) * 1.0

            # 湿度过大 → 正贡献

            humidity_risk = max(0, (humidity - 60) / 25) * 0.8

            # 保护气不足 → 正贡献

            gas_risk = max(0, (14 - gas_flow) / 8) * 0.9

            # 交互: 速度快 + 电流小 → 叠加

            interaction = 0.5 * speed_risk * current_risk

 

            logit = (speed_risk + current_risk + humidity_risk

                    + gas_risk + interaction

                    - 1.5 + np.random.normal(0, 0.5))

 

            prob = 1 / (1 + np.exp(-logit))

            porosity = int(np.random.rand() < prob)

 

            records.append({

                "weld_id": f"W-{i+1:04d}",

                "current_a": max(100, current),

                "voltage_v": max(12, voltage),

                "weld_speed_cm_min": max(15, speed),

                "wire_feed_m_min": max(1, wire_feed),

                "gas_flow_l_min": max(8, gas_flow),

                "interpass_temp_c": max(40, interpass),

                "preheat_temp_c": max(15, preheat),

                "gas_purity_pct": min(100, max(98, gas_purity)),

                "humidity_pct": min(90, max(25, humidity)),

                "plate_thickness_mm": thickness,

                "wire_batch": wire_batch,

                "groove_type": groove,

                "porosity": porosity,

            })

 

        # 保存

        sample_dir = Path("sample_data")

        sample_dir.mkdir(exist_ok=True)

        df = pd.DataFrame(records)

        df.to_csv(sample_dir / "weld_log.csv", index=False)

 

        X = df[self.CONTINUOUS_COLUMNS].values.astype(float)

        y = df[self.TARGET_COLUMN].values.astype(int)

 

        return X, y

 

</details>

 

<details><summary></summary>

 

"""焊接参数与气孔的统计检验 (scipy)。"""

 

import numpy as np

from typing import Dict, List

from scipy import stats

from .models import ParameterInfluence

 

 

class StatisticalTester:

    """

    统计检验: 每个焊接参数与气孔的关联显著性。

    """

 

    PARAMETER_NAMES = [

        "电流", "电压", "焊接速度", "送丝速度", "保护气流量",

        "层间温度", "预热温度", "气体纯度", "环境湿度", "板厚"

    ]

 

    def __init__(self):

        self.results_: Dict[str, ParameterInfluence] = {}

 

    def test_all(self, X: np.ndarray, y: np.ndarray

                 ) -> Dict[str, ParameterInfluence]:

        """对所有连续参数进行统计检验。"""

        n_features = X.shape[1]

        self.results_ = {}

 

        # 按气孔分组

        group0 = [X[y == 0, i] for i in range(n_features)] # 无气孔

        group1 = [X[y == 1, i] for i in range(n_features)] # 有气孔

 

        for i in range(n_features):

            name = self.PARAMETER_NAMES[i] if i < len(self.PARAMETER_NAMES) else f"参数{i}"

 

            # Shapiro-Wilk 正态性检验

            _, p_norm0 = stats.shapiro(group0[i][:50]) if len(group0[i]) >= 3 else (0, 0)

            _, p_norm1 = stats.shapiro(group1[i][:50]) if len(group1[i]) >= 3 else (0, 0)

 

            # 选择检验方法

            if len(group0[i]) < 3 or len(group1[i]) < 3:

                p_value = 1.0

                test_method = "insufficient_data"

                effect_size = 0.0

            elif p_norm0 < 0.05 or p_norm1 < 0.05:

                # 非正态 → Mann-Whitney U

                stat, p_value = stats.mannwhitneyu(

                    group0[i], group1[i], alternative='two-sided')

                test_method = "Mann-Whitney U"

                # Cliff's delta 效应量

                effect_size = abs(stats.ranksums(group0[i], group1[i]).statistic) / np.sqrt(len(group0[i]) + len(group1[i]))

            else:

                # 正态 → 独立样本 t-test

                stat, p_value = stats.ttest_ind(group0[i], group1[i])

                test_method = "t-test"

                # Cohen's d

                pooled_std = np.sqrt(

                    ((len(group0[i])-1)*np.var(group0[i], ddof=1) +

                     (len(group1[i])-1)*np.var(group1[i], ddof=1)) /

                    (len(group0[i]) + len(group1[i]) - 2)

                )

                effect_size = abs(np.mean(group1[i]) - np.mean(group0[i])) / (pooled_std + 1e-8)

 

            # 方向

            mean_diff = np.mean(group1[i]) - np.mean(group0[i])

            direction = "positive" if mean_diff > 0 else "negative"

 

            self.results_[name] = ParameterInfluence(

                param_name=name,

                p_value=float(p_value),

                effect_size=float(effect_size),

                direction=direction,

                test_method=test_method,

            )

 

        return self.results_

 

    def get_significant(self, alpha: float = 0.05

                       ) -> List[tuple]:

        """获取显著参数 (p < alpha)。"""

        sig = [(name, inf) for name, inf in self.results_.items()

               if inf.p_value < alpha]

        sig.sort(key=lambda x: x[1].p_value)

        return sig

 

</details>

 

<details><summary></summary>

 

"""气孔风险分类模型 (sklearn)。"""

 

import numpy as np

from typing import Dict, List, Tuple

from sklearn.linear_model import LogisticRegression

from sklearn.ensemble import RandomForestClassifier

from sklearn.model_selection import cross_val_score, StratifiedKFold

from sklearn.metrics import (roc_auc_score, roc_curve,

                           accuracy_score, precision_score, recall_score)

from sklearn.preprocessing import StandardScaler

 

 

class PorosityClassifier:

    """

    气孔风险分类:

    1. LogisticRegression (可解释)

    2. RandomForestClassifier (非线性+交互)

    """

 

    def __init__(self, random_state: int = 42):

        self.random_state = random_state

        self.scaler_ = StandardScaler()

        self.lr_ = LogisticRegression(

            max_iter=1000, random_state=random_state)

        self.rf_ = RandomForestClassifier(

            n_estimators=100, max_depth=5,

            random_state=random_state, n_jobs=-1)

        self.is_fitted_ = False

 

    def fit(self, X: np.ndarray, y: np.ndarray):

        """训练双模型。"""

        X_scaled = self.scaler_.fit_transform(X)

        self.lr_.fit(X_scaled, y)

        self.rf_.fit(X_scaled, y)

        self.is_fitted_ = True

        return self

 

    def predict_proba(self, X: np.ndarray) -> np.ndarray:

        """预测气孔概率。"""

        if not self.is_fitted_:

            raise ValueError("模型尚未训练")

        X_scaled = self.scaler_.transform(X)

        return self.rf_.predict_proba(X_scaled)[:, 1]

 

    def evaluate(self, X: np.ndarray, y: np.ndarray

                 ) -> Dict[str, float]:

        """交叉验证评估。"""

        if not self.is_fitted_:

            X_scaled = self.scaler_.fit_transform(X)

            self.lr_.fit(X_scaled, y)

            self.rf_.fit(X_scaled, y)

            self.is_fitted_ = True

 

        X_scaled = self.scaler_.transform(X)

        cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

 

        lr_scores = cross_val_score(self.lr_, X_scaled, y, cv=cv, scoring='roc_auc')

        rf_scores = cross_val_score(self.rf_, X_scaled, y, cv=cv, scoring='roc_auc')

 

        y_pred_lr = self.lr_.predict(X_scaled)

        y_pred_rf = self.rf_.predict(X_scaled)

        y_proba_rf = self.rf_.predict_proba(X_scaled)[:, 1]

 

        return {

            "lr_auc": float(lr_scores.mean()),

            "rf_auc": float(rf_scores.mean()),

            "lr_accuracy": float(accuracy_score(y, y_pred_lr)),

            "rf_accuracy": float(accuracy_score(y, y_pred_rf)),

            "rf_precision": float(precision_score(y, y_pred_rf, zero_division=0)),

            "rf_recall": float(recall_score(y, y_pred_rf, zero_division=0)),

            "rf_auc_full": float(roc_auc_score(y, y_proba_rf)),

        }

 

    def get_roc_curve(self, X: np.ndarray, y: np.ndarray

                      ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:

        """获取 ROC 曲线数据。"""

        if not self.is_fitted_:

            self.fit(X, y)

        X_scaled = self.scaler_.transform(X)

        y_proba = self.rf_.predict_proba(X_scaled)[:, 1]

        fpr, tpr, thresholds = roc_curve(y, y_proba)

        return fpr, tpr, thresholds

 

    def get_feature_importance(self) -> Dict[str, float]:

        """随机森林特征重要性。"""

        if not self.is_fitted_:

            return {}

        names = StatisticalTester.PARAMETER_NAMES

        importances = self.rf_.feature_importances_

        return dict(zip(names[:len(importances)], importances))

 

    def get_lr_coefficients(self) -> Dict[str, float]:

        """Logistic 回归系数 (可解释性)。"""

        if not self.is_fitted_:

            return {}

        names = StatisticalTester.PARAMETER_NAMES

        coefs = self.lr_.coef_[0]

        return dict(zip(names[:len(coefs)], coefs))

 

</details>

 

<details><summary></summary>

 

"""关联规则挖掘: 参数组合 → 气孔。"""

 

import numpy as np

import pandas as pd

from typing import Dict, List, Tuple

from mlxtend.frequent_patterns import apriori, association_rules

 

 

class AssociationMiner:

    """

    焊接参数关联规则挖掘。

    将连续参数离散化 → 频繁项集 → 关联规则。

    """

 

    PARAMETER_NAMES = [

        "电流", "电压", "焊接速度", "送丝速度", "保护气流量",

        "层间温度", "预热温度", "气体纯度", "环境湿度", "板厚"

    ]

 

    def __init__(self, min_support: float = 0.05,

                 min_confidence: float = 0.6):

        self.min_support = min_support

        self.min_confidence = min_confidence

 

    def discretize(self, X: np.ndarray,

                   y: np.ndarray) -> pd.DataFrame:

        """连续参数离散化为类别。"""

        df = pd.DataFrame(X, columns=self.PARAMETER_NAMES[:X.shape[1]])

 

        # 对每个参数按三分位数离散化

        for col in df.columns:

            q1 = df[col].quantile(0.33)

            q2 = df[col].quantile(0.67)

            df[col] = pd.cut(df[col],

                           bins=[-np.inf, q1, q2, np.inf],

                           labels=[f"{col}_低", f"{col}_中", f"{col}_高"])

 

        # 添加气孔标签

        df["气孔_有"] = ["气孔_无" if v == 0 else "气孔_有" for v in y]

 

        return df

 

    def mine_rules(self, X: np.ndarray, y: np.ndarray

                   ) -> pd.DataFrame:

        """挖掘关联规则。"""

        df = self.discretize(X, y)

 

        # 转换为 one-hot 编码

        df_encoded = pd.get_dummies(df)

 

        # 频繁项集

        frequent_itemsets = apriori(df_encoded,

                                   min_support=self.min_support,

                                   use_colnames=True)

 

        if len(frequent_itemsets) == 0:

            return pd.DataFrame()

 

        # 关联规则

        rules = association_rules(frequent_itemsets,

                                 metric="confidence",

                                 min_threshold=self.min_confidence)

 

        # 筛选包含"气孔_有"的规则

        porosity_rules = rules[

利用AI解决实际问题，如果你觉得这个工具好用，欢迎关注长安牧笛！