app.py

@@ -18,6 +18,7 @@ from sklearn.cluster import DBSCAN
 from sklearn.decomposition import PCA
 from sklearn.preprocessing import StandardScaler
 from defect_analysis.data_quality import build_data_quality_report
+from defect_analysis.ml.key_factors import find_key_factors
 from app_utils import (
     apply_defect_filters,
     build_diagnostic_dashboard,
@@ -298,6 +299,7 @@ if _t:
         dashboard = build_diagnostic_dashboard(filtered_df)
         industry_diagnosis = generate_industry_diagnosis(filtered_df, dashboard)
         quality_report = build_data_quality_report(filtered_df)
+        key_factors = find_key_factors(filtered_df, target_defect_type=dashboard["top_defect_type"], top_n=10)
         level_colors = {
             "严重": ("#7f1d1d", "#fee2e2"),
             "关注": ("#92400e", "#fef3c7"),
@@ -559,6 +561,16 @@ if _t:
                 st.dataframe(extended_table, use_container_width=True, hide_index=True)
                 st.caption("覆盖治具、吸嘴、材料批次、清洗/绑定等维度，用于多前制程链路追溯。")
 
+            if not key_factors.empty:
+                st.subheader(f"关键因子分析：{dashboard['top_defect_type']}")
+                key_factor_table = key_factors.copy()
+                key_factor_table["目标占比"] = key_factor_table["目标占比"].map(lambda v: f"{v:.1%}")
+                key_factor_table["基线占比"] = key_factor_table["基线占比"].map(lambda v: f"{v:.1%}")
+                key_factor_table["异常倍数"] = key_factor_table["异常倍数"].map(lambda v: f"{v:.2f}x")
+                key_factor_table["支持度"] = key_factor_table["支持度"].map(lambda v: f"{v:.1%}")
+                st.dataframe(key_factor_table, use_container_width=True, hide_index=True)
+                st.caption("关键因子按目标缺陷占比、异常倍数、样本数和支持度综合排序。")
+
         trend_col, pareto_col = st.columns([1, 1])
         with trend_col:
             st.subheader("每日缺陷走势")

defect_analysis/ml/key_factors.py

@@ -0,0 +1,113 @@
+"""不良关键因子发现。"""
+
+import numpy as np
+import pandas as pd
+
+from defect_analysis.schemas import normalize_defect_schema
+
+
+DEFAULT_FACTOR_DIMENSIONS = [
+    "equipment_id",
+    "seat_id",
+    "lam_equipment_id",
+    "lam_seat_id",
+    "lam_fixture_id",
+    "lam_jig_id",
+    "lam_nozzle_id",
+    "material_lot_oca",
+    "material_lot_glass",
+    "material_lot_polarizer",
+    "clean_equipment_id",
+    "clean_slot_id",
+    "bond_equipment_id",
+    "bond_head_id",
+    "recipe_id",
+    "shift",
+    "defect_geometry_type",
+]
+
+
+def _build_target_mask(df, target_defect_type=None, target_severity=None):
+    if target_defect_type:
+        return df["defect_type"] == target_defect_type
+    if target_severity:
+        return df["severity"] == target_severity
+    return df["severity"] == "严重"
+
+
+def find_key_factors(
+    df,
+    *,
+    target_defect_type=None,
+    target_severity=None,
+    dimensions=None,
+    min_count=3,
+    min_lift=1.1,
+    top_n=20,
+):
+    """查找与目标不良显著相关的关键因子。
+
+    当前实现是可解释统计排序：按每个维度取值计算目标占比、异常倍数和支持度，
+    用综合得分排序。它适合生产早期作为 ML 特征与根因候选的基线。
+    """
+    normalized = normalize_defect_schema(df)
+    if normalized.empty:
+        return pd.DataFrame()
+
+    dimensions = DEFAULT_FACTOR_DIMENSIONS if dimensions is None else dimensions
+    target_mask = _build_target_mask(normalized, target_defect_type, target_severity)
+    baseline_rate = float(target_mask.mean())
+    if baseline_rate <= 0:
+        return pd.DataFrame(
+            columns=["维度", "因子值", "样本数", "目标数", "目标占比", "基线占比", "异常倍数", "支持度", "关键因子得分"]
+        )
+
+    rows = []
+    total = len(normalized)
+    for dimension in dimensions:
+        if dimension not in normalized.columns:
+            continue
+        values = normalized[dimension].fillna("").astype(str)
+        valid = normalized[values != ""].copy()
+        if valid.empty:
+            continue
+        valid_target = target_mask.loc[valid.index]
+        grouped = valid.assign(_target=valid_target.astype(int)).groupby(dimension)
+        for value, group in grouped:
+            count = len(group)
+            if count < min_count:
+                continue
+            target_count = int(group["_target"].sum())
+            if target_count == 0:
+                continue
+            target_rate = target_count / count
+            lift = target_rate / baseline_rate
+            if lift < min_lift:
+                continue
+            support = count / total
+            score = (max(lift - 1, 0) * 45) + (target_rate * 30) + (np.sqrt(target_count) * 8) + (support * 17)
+            rows.append(
+                {
+                    "维度": dimension,
+                    "因子值": str(value),
+                    "样本数": int(count),
+                    "目标数": target_count,
+                    "目标占比": round(float(target_rate), 4),
+                    "基线占比": round(float(baseline_rate), 4),
+                    "异常倍数": round(float(lift), 2),
+                    "支持度": round(float(support), 4),
+                    "关键因子得分": round(float(score), 2),
+                }
+            )
+
+    if not rows:
+        return pd.DataFrame(
+            columns=["维度", "因子值", "样本数", "目标数", "目标占比", "基线占比", "异常倍数", "支持度", "关键因子得分"]
+        )
+
+    result = pd.DataFrame(rows)
+    return (
+        result.sort_values(["关键因子得分", "目标数", "异常倍数"], ascending=False)
+        .head(top_n)
+        .reset_index(drop=True)
+    )

tests/test_app_utils.py

@@ -1,10 +1,6 @@
 import math
-import os
-import sys
 import unittest
 
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
 import pandas as pd
 
 from app_utils import (