DUCG建模实战指南

1
┌─────────────────────────────────────────┐
2
│  Step 1: 问题定义与领域分析              │
3
│  - 明确建模目标                          │
4
│  - 识别关键问题                          │
5
│  - 确定应用场景                          │
6
└──────────────┬──────────────────────────┘
7
               │
8
┌──────────────▼──────────────────────────┐
9
│  Step 2: 变量识别与分类                  │
10
│  - 原因变量（Cause Variables）          │
11
│  - 结果变量（Effect Variables）         │
12
│  - 中间变量（Intermediate Variables）   │
13
└──────────────┬──────────────────────────┘
14
               │
15
┌──────────────▼──────────────────────────┐
16
│  Step 3: 因果关系挖掘                    │
17
│  - 专家知识提取                          │
18
│  - 数据驱动学习                          │
19
│  - 文献调研                              │
20
└──────────────┬──────────────────────────┘
21
               │
22
┌──────────────▼──────────────────────────┐
23
│  Step 4: 网络结构构建                    │
24
│  - 绘制因果图                            │
25
│  - 定义变量状态                          │
26
│  - 建立因果连接                          │
27
└──────────────┬──────────────────────────┘
28
               │
29
┌──────────────▼──────────────────────────┐
30
│  Step 5: 参数学习与估计                  │
31
│  - 因果强度学习                          │
32
│  - 条件概率估计                          │
33
│  - 权重参数优化                          │
34
└──────────────┬──────────────────────────┘
35
               │
36
┌──────────────▼──────────────────────────┐
37
│  Step 6: 模型验证与优化                  │
38
│  - 推理准确性测试                        │
39
│  - 敏感性分析                            │
40
│  - 模型迭代改进                          │
41
└──────────────┬──────────────────────────┘
42
               │
43
┌──────────────▼──────────────────────────┐
44
│  Step 7: 部署与应用                      │
45
│  - 实时推理系统                          │
46
│  - 可视化界面                            │
47
│  - 持续监控更新                          │
48
└──────────────────────────────────────────┘

1
# 建模目标定义
2
modeling_objective = {
3
    "problem_type": "诊断问题",
4
    "target": "根据软件测试结果和运行日志，诊断潜在缺陷类型",
5
    "inputs": [
6
        "测试用例执行结果",
7
        "代码覆盖率",
8
        "运行时异常日志",
9
        "性能指标"
10
    ],
11
    "outputs": [
12
        "缺陷类型（逻辑错误/内存泄漏/并发问题等）",
13
        "缺陷位置（模块/类/函数）",
14
        "严重程度评估"
15
    ],
16
    "constraints": [
17
        "实时性要求：< 5秒",
18
        "准确率要求：> 85%",
19
        "可解释性要求：输出诊断依据"
20
    ]
21
}

1
class ExpertKnowledgeExtraction:
2
    def __init__(self):
3
        self.knowledge_base = {}
4

5
    def structured_interview(self, expert_id):
6
        """结构化访谈模板"""
7
        questions = [
8
            "系统主要组件有哪些？",
9
            "各组件之间的依赖关系？",
10
            "常见故障模式及其表现？",
11
            "故障诊断的经验规则？",
12
            "关键监控指标有哪些？"
13
        ]
14

15
        answers = {}
16
        for q in questions:
17
            answers[q] = self.get_expert_response(expert_id, q)
18

19
        return self.extract_causal_relations(answers)
20

21
    def extract_causal_relations(self, answers):
22
        """从访谈中提取因果关系"""
23
        causal_relations = []
24

25
        # 示例：解析"如果A发生，则B可能出现"的语句
26
        for answer in answers.values():
27
            if "如果" in answer and "则" in answer:
28
                cause, effect = self.parse_if_then(answer)
29
                causal_relations.append({
30
                    "cause": cause,
31
                    "effect": effect,
32
                    "confidence": "high"  # 专家知识置信度高
33
                })
34

35
        return causal_relations

1
import pandas as pd
2

3
def literature_review(domain, keywords):
4
    """文献调研辅助工具"""
5
    # 搜索相关论文
6
    papers = search_papers(domain, keywords)
7

8
    # 提取因果关系
9
    extracted_relations = []
10
    for paper in papers:
11
        relations = extract_causal_statements(paper.text)
12
        extracted_relations.extend(relations)
13

14
    # 构建知识图谱
15
    knowledge_graph = build_knowledge_graph(extracted_relations)
16

17
    return knowledge_graph
18

19
# 示例：软件可靠性领域
20
kg = literature_review(
21
    domain="software reliability",
22
    keywords=["fault", "failure", "defect", "causality"]
23
)

1
import pandas as pd
2
import seaborn as sns
3
import matplotlib.pyplot as plt
4

5
def exploratory_causal_analysis(data):
6
    """探索性因果分析"""
7
    # 1. 变量关联分析
8
    correlation_matrix = data.corr()
9
    plt.figure(figsize=(12, 10))
10
    sns.heatmap(correlation_matrix, annot=True, cmap='coolwarm')
11
    plt.title('变量相关性矩阵')
12
    plt.show()
13

14
    # 2. 时间序列因果发现（Granger因果）
15
    from statsmodels.tsa.stattools import grangercausalitytests
16

17
    granger_results = {}
18
    for col1 in data.columns:
19
        for col2 in data.columns:
20
            if col1 != col2:
21
                try:
22
                    result = grangercausalitytests(
23
                        data[[col2, col1]],
24
                        maxlag=5,
25
                        verbose=False
26
                    )
27
                    p_value = result[1][0]['ssr_ftest'][1]
28
                    if p_value < 0.05:
29
                        granger_results[(col1, col2)] = p_value
30
                except:
31
                    pass
32

33
    return granger_results
34

35
# 使用示例
36
data = pd.read_csv('software_metrics.csv')
37
causal_hints = exploratory_causal_analysis(data)
38
print("潜在因果关系:")
39
for (cause, effect), p_val in causal_hints.items():
40
    print(f"{cause} -> {effect} (p={p_val:.4f})")

1
from enum import Enum
2

3
class VariableType(Enum):
4
    """变量类型枚举"""
5
    ROOT_CAUSE = "root_cause"           # 根原因（无父节点）
6
    INTERMEDIATE = "intermediate"       # 中间节点
7
    OBSERVABLE = "observable"           # 可观测节点
8
    TARGET = "target"                   # 目标节点（诊断目标）
9

10
class VariableState(Enum):
11
    """变量状态枚举"""
12
    BINARY = "binary"                   # 二值（是/否）
13
    DISCRETE = "discrete"               # 离散（多个状态）
14
    CONTINUOUS = "continuous"           # 连续值
15

16
class DUCGVariable:
17
    """DUCG变量定义"""
18
    def __init__(self, name, var_type, state_type, states=None):
19
        self.name = name
20
        self.var_type = var_type
21
        self.state_type = state_type
22

23
        # 定义状态空间
24
        if state_type == VariableState.BINARY:
25
            self.states = ['False', 'True']
26
        elif state_type == VariableState.DISCRETE:
27
            self.states = states if states else []
28
        else:  # CONTINUOUS
29
            self.states = None  # 连续变量需要离散化
30

31
        self.parents = []      # 父节点
32
        self.children = []     # 子节点
33
        self.probability = {}  # 条件概率表
34
        self.strength = {}     # 因果强度
35

36
    def add_parent(self, parent_var):
37
        """添加父节点"""
38
        if parent_var not in self.parents:
39
            self.parents.append(parent_var)
40
            parent_var.children.append(self)
41

42
    def set_causal_strength(self, parent_var, strength):
43
        """设置因果强度"""
44
        self.strength[parent_var.name] = strength
45

46
    def __repr__(self):
47
        return f"DUCGVariable(name={self.name}, type={self.var_type.value})"

1
# 定义软件缺陷诊断问题的变量
2

3
# 1. 根原因变量（潜在缺陷类型）
4
defect_logic_error = DUCGVariable(
5
    name="逻辑错误",
6
    var_type=VariableType.ROOT_CAUSE,
7
    state_type=VariableState.BINARY
8
)
9

10
defect_memory_leak = DUCGVariable(
11
    name="内存泄漏",
12
    var_type=VariableType.ROOT_CAUSE,
13
    state_type=VariableState.BINARY
14
)
15

16
defect_concurrency = DUCGVariable(
17
    name="并发问题",
18
    var_type=VariableType.ROOT_CAUSE,
19
    state_type=VariableState.BINARY
20
)
21

22
# 2. 中间变量（症状/表现）
23
symptom_crash = DUCGVariable(
24
    name="程序崩溃",
25
    var_type=VariableType.INTERMEDIATE,
26
    state_type=VariableState.BINARY
27
)
28

29
symptom_slow_performance = DUCGVariable(
30
    name="性能下降",
31
    var_type=VariableType.INTERMEDIATE,
32
    state_type=VariableState.DISCRETE,
33
    states=["正常", "轻微", "严重"]
34
)
35

36
symptom_memory_usage = DUCGVariable(
37
    name="内存占用异常",
38
    var_type=VariableType.INTERMEDIATE,
39
    state_type=VariableState.BINARY
40
)
41

42
# 3. 可观测变量（测试结果）
43
obs_test_failure = DUCGVariable(
44
    name="测试失败率",
45
    var_type=VariableType.OBSERVABLE,
46
    state_type=VariableState.DISCRETE,
47
    states=["低", "中", "高"]
48
)
49

50
obs_code_coverage = DUCGVariable(
51
    name="代码覆盖率",
52
    var_type=VariableType.OBSERVABLE,
53
    state_type=VariableState.DISCRETE,
54
    states=["低", "中", "高"]
55
)
56

57
obs_exception_count = DUCGVariable(
58
    name="异常数量",
59
    var_type=VariableType.OBSERVABLE,
60
    state_type=VariableState.DISCRETE,
61
    states=["少", "中", "多"]
62
)
63

64
# 4. 目标变量（综合诊断结果）
65
target_defect_severity = DUCGVariable(
66
    name="缺陷严重程度",
67
    var_type=VariableType.TARGET,
68
    state_type=VariableState.DISCRETE,
69
    states=["低", "中", "高", "严重"]
70
)

1
def determine_variable_granularity(system_complexity, available_data, time_constraint):
2
    """自动确定变量粒度"""
3
    if system_complexity == "low" and time_constraint == "strict":
4
        return "coarse"
5
    elif system_complexity == "high" and available_data == "abundant":
6
        return "fine"
7
    else:
8
        return "multi_level"
9

10
# 示例
11
granularity = determine_variable_granularity(
12
    system_complexity="medium",
13
    available_data="moderate",
14
    time_constraint="moderate"
15
)
16
print(f"推荐变量粒度: {granularity}")

1
class CausalRelationshipBuilder:
2
    def __init__(self):
3
        self.relationships = []
4

5
    def add_expert_knowledge(self, cause, effect, strength, confidence):
6
        """添加专家给出的因果关系"""
7
        self.relationships.append({
8
            'cause': cause,
9
            'effect': effect,
10
            'strength': strength,       # 因果强度 [0, 1]
11
            'confidence': confidence,   # 置信度 [0, 1]
12
            'source': 'expert'
13
        })
14

15
    def add_data_driven(self, cause, effect, strength, p_value):
16
        """添加数据驱动发现的因果关系"""
17
        self.relationships.append({
18
            'cause': cause,
19
            'effect': effect,
20
            'strength': strength,
21
            'p_value': p_value,
22
            'source': 'data'
23
        })
24

25
    def merge_relationships(self):
26
        """融合专家知识与数据发现"""
27
        merged = {}
28

29
        for rel in self.relationships:
30
            key = (rel['cause'], rel['effect'])
31

32
            if key not in merged:
33
                merged[key] = rel
34
            else:
35
                # 融合策略：加权平均
36
                existing = merged[key]
37
                if existing['source'] == 'expert' and rel['source'] == 'data':
38
                    # 专家知识权重0.6，数据权重0.4
39
                    merged[key]['strength'] = (
40
                        0.6 * existing['strength'] +
41
                        0.4 * rel['strength']
42
                    )
43
                    merged[key]['source'] = 'hybrid'
44

45
        return list(merged.values())
46

47
# 使用示例
48
builder = CausalRelationshipBuilder()
49

50
# 专家知识
51
builder.add_expert_knowledge(
52
    cause=defect_memory_leak,
53
    effect=symptom_memory_usage,
54
    strength=0.8,
55
    confidence=0.9
56
)
57

58
# 数据驱动
59
builder.add_data_driven(
60
    cause=defect_logic_error,
61
    effect=symptom_crash,
62
    strength=0.75,
63
    p_value=0.001
64
)
65

66
# 融合
67
final_relationships = builder.merge_relationships()

1
from causallearn.search.ConstraintBased.PC import pc
2
import numpy as np
3

4
def discover_causal_structure(data, alpha=0.05):
5
    """使用PC算法发现因果结构"""
6
    # 运行PC算法
7
    cg = pc(
8
        data.values,
9
        alpha=alpha,
10
        indep_test='fisherz',  # 独立性检验方法
11
        stable=True,
12
        uc_rule=0,
13
        uc_priority=2
14
    )
15

16
    # 提取因果图
17
    causal_graph = cg.G.graph
18

19
    # 可视化
20
    import networkx as nx
21
    G = nx.DiGraph()
22

23
    for i in range(len(data.columns)):
24
        for j in range(len(data.columns)):
25
            if causal_graph[i, j] == 1:  # 有向边
26
                G.add_edge(data.columns[i], data.columns[j])
27

28
    return G
29

30
# 使用示例
31
import pandas as pd
32
data = pd.read_csv('software_metrics.csv')
33
causal_graph = discover_causal_structure(data)
34

35
# 可视化
36
import matplotlib.pyplot as plt
37
pos = nx.spring_layout(causal_graph)
38
nx.draw(causal_graph, pos, with_labels=True,
39
        node_color='lightblue',
40
        node_size=2000,
41
        font_size=10,
42
        arrows=True)
43
plt.title('发现的因果结构')
44
plt.show()

1
from pgmpy.estimators import HillClimbSearch, BicScore
2
from pgmpy.models import BayesianNetwork
3

4
def learn_causal_structure_greedy(data):
5
    """使用爬山搜索学习因果结构"""
6
    # 定义评分函数（BIC分数）
7
    scoring_method = BicScore(data)
8

9
    # 爬山搜索
10
    hc = HillClimbSearch(data)
11
    best_model = hc.estimate(
12
        scoring_method=scoring_method,
13
        max_indegree=4,        # 最大入度限制
14
        max_iter=100,
15
        show_progress=True
16
    )
17

18
    return best_model
19

20
# 使用示例
21
learned_structure = learn_causal_structure_greedy(data)
22
print("学习到的边:")
23
print(learned_structure.edges())

1
import dowhy
2
from dowhy import CausalModel
3

4
def discover_causal_effects(data, treatment, outcome, confounders):
5
    """使用DoWhy进行因果效应估计"""
6
    # 定义因果图
7
    causal_graph = f"""{% raw %}
8
    digraph {{
9
        {' -> '.join([f'"{c}" -> "{outcome}"' for c in confounders])}
10
        "{treatment}" -> "{outcome}";
11
        {' -> '.join([f'"{c}" -> "{treatment}"' for c in confounders])}
12
    }}{% endraw %}
13
    """
14

15
    # 创建因果模型
16
    model = CausalModel(
17
        data=data,
18
        treatment=treatment,
19
        outcome=outcome,
20
        graph=causal_graph
21
    )
22

23
    # 识别因果效应
24
    identified_estimand = model.identify_effect(proceed_when_unidentifiable=True)
25

26
    # 估计因果效应
27
    causal_estimate = model.estimate_effect(
28
        identified_estimand,
29
        method_name="backdoor.propensity_score_matching"
30
    )
31

32
    # 反驳分析（敏感性测试）
33
    refutation = model.refute_estimate(
34
        identified_estimand,
35
        causal_estimate,
36
        method_name="random_common_cause"
37
    )
38

39
    return causal_estimate, refutation
40

41
# 使用示例
42
treatment = "代码复杂度"
43
outcome = "缺陷数量"
44
confounders = ["开发经验", "项目规模", "时间压力"]
45

46
estimate, refute = discover_causal_effects(data, treatment, outcome, confounders)
47
print(f"因果效应估计: {estimate.value}")
48
print(f"反驳结果: {refute}")

1
import networkx as nx
2
from typing import Dict, List
3

4
class DUCGNetwork:
5
    """DUCG网络类"""
6

7
    def __init__(self, name):
8
        self.name = name
9
        self.variables = {}      # 变量字典
10
        self.graph = nx.DiGraph()  # 有向图
11
        self.time_slices = 1     # 时间片数量（动态网络）
12

13
    def add_variable(self, variable: DUCGVariable):
14
        """添加变量节点"""
15
        self.variables[variable.name] = variable
16
        self.graph.add_node(variable.name, var_obj=variable)
17

18
    def add_edge(self, parent_name, child_name, strength=None, probability=None):
19
        """添加因果边"""
20
        if parent_name not in self.variables or child_name not in self.variables:
21
            raise ValueError("变量不存在")
22

23
        parent = self.variables[parent_name]
24
        child = self.variables[child_name]
25

26
        # 更新变量关系
27
        child.add_parent(parent)
28

29
        # 添加边到图
30
        self.graph.add_edge(
31
            parent_name,
32
            child_name,
33
            strength=strength,
34
            probability=probability
35
        )
36

37
    def set_causal_strength(self, parent_name, child_name, strength):
38
        """设置因果强度"""
39
        self.graph[parent_name][child_name]['strength'] = strength
40
        self.variables[child_name].set_causal_strength(
41
            self.variables[parent_name],
42
            strength
43
        )
44

45
    def set_conditional_probability(self, variable_name, parent_states, probability):
46
        """设置条件概率"""
47
        var = self.variables[variable_name]
48
        var.probability[tuple(parent_states)] = probability
49

50
    def visualize(self, layout='spring', figsize=(12, 8)):
51
        """可视化DUCG网络"""
52
        import matplotlib.pyplot as plt
53

54
        plt.figure(figsize=figsize)
55

56
        # 选择布局
57
        if layout == 'spring':
58
            pos = nx.spring_layout(self.graph, k=2, iterations=50)
59
        elif layout == 'hierarchical':
60
            pos = nx.nx_agraph.graphviz_layout(self.graph, prog='dot')
61
        else:
62
            pos = nx.circular_layout(self.graph)
63

64
        # 节点颜色映射
65
        color_map = {
66
            VariableType.ROOT_CAUSE: 'lightcoral',
67
            VariableType.INTERMEDIATE: 'lightblue',
68
            VariableType.OBSERVABLE: 'lightgreen',
69
            VariableType.TARGET: 'gold'
70
        }
71

72
        node_colors = [
73
            color_map[self.variables[node].var_type]
74
            for node in self.graph.nodes()
75
        ]
76

77
        # 绘制网络
78
        nx.draw(
79
            self.graph,
80
            pos,
81
            with_labels=True,
82
            node_color=node_colors,
83
            node_size=3000,
84
            font_size=10,
85
            font_weight='bold',
86
            arrows=True,
87
            arrowsize=20,
88
            edge_color='gray',
89
            width=2
90
        )
91

92
        # 添加边标签（因果强度）
93
        edge_labels = {}
94
        for u, v, data in self.graph.edges(data=True):
95
            if 'strength' in data and data['strength'] is not None:
96
                edge_labels[(u, v)] = f"{data['strength']:.2f}"
97

98
        nx.draw_networkx_edge_labels(
99
            self.graph,
100
            pos,
101
            edge_labels,
102
            font_size=8
103
        )
104

105
        plt.title(f'DUCG Network: {self.name}', fontsize=16, fontweight='bold')
106
        plt.axis('off')
107
        plt.tight_layout()
108
        plt.show()
109

110
    def get_markov_blanket(self, variable_name):
111
        """获取变量的马尔可夫毯"""
112
        var = self.variables[variable_name]
113

114
        # 马尔可夫毯 = 父节点 + 子节点 + 子节点的其他父节点
115
        mb = set(var.parents)
116
        mb.update(var.children)
117

118
        for child in var.children:
119
            mb.update(child.parents)
120

121
        mb.discard(var)  # 移除自己
122
        return list(mb)
123

124
    def to_dict(self):
125
        """导出为字典格式"""
126
        network_dict = {
127
            'name': self.name,
128
            'variables': [],
129
            'edges': []
130
        }
131

132
        for var_name, var in self.variables.items():
133
            network_dict['variables'].append({
134
                'name': var.name,
135
                'type': var.var_type.value,
136
                'state_type': var.state_type.value,
137
                'states': var.states
138
            })
139

140
        for u, v, data in self.graph.edges(data=True):
141
            network_dict['edges'].append({
142
                'from': u,
143
                'to': v,
144
                'strength': data.get('strength'),
145
                'probability': data.get('probability')
146
            })
147

148
        return network_dict
149

150
    def save_to_json(self, filepath):
151
        """保存为JSON文件"""
152
        import json
153
        with open(filepath, 'w', encoding='utf-8') as f:
154
            json.dump(self.to_dict(), f, ensure_ascii=False, indent=2)
155

156
    @classmethod
157
    def load_from_json(cls, filepath):
158
        """从JSON文件加载"""
159
        import json
160
        with open(filepath, 'r', encoding='utf-8') as f:
161
            data = json.load(f)
162

163
        network = cls(data['name'])
164

165
        # 重建变量
166
        for var_data in data['variables']:
167
            var = DUCGVariable(
168
                name=var_data['name'],
169
                var_type=VariableType(var_data['type']),
170
                state_type=VariableState(var_data['state_type']),
171
                states=var_data.get('states')
172
            )
173
            network.add_variable(var)
174

175
        # 重建边
176
        for edge_data in data['edges']:
177
            network.add_edge(
178
                edge_data['from'],
179
                edge_data['to'],
180
                strength=edge_data.get('strength'),
181
                probability=edge_data.get('probability')
182
            )
183

184
        return network

1
# 创建DUCG网络
2
ducg_net = DUCGNetwork("软件缺陷诊断系统")
3

4
# 添加所有变量
5
ducg_net.add_variable(defect_logic_error)
6
ducg_net.add_variable(defect_memory_leak)
7
ducg_net.add_variable(defect_concurrency)
8
ducg_net.add_variable(symptom_crash)
9
ducg_net.add_variable(symptom_slow_performance)
10
ducg_net.add_variable(symptom_memory_usage)
11
ducg_net.add_variable(obs_test_failure)
12
ducg_net.add_variable(obs_code_coverage)
13
ducg_net.add_variable(obs_exception_count)
14
ducg_net.add_variable(target_defect_severity)
15

16
# 建立因果边及强度
17
# 逻辑错误 -> 症状
18
ducg_net.add_edge("逻辑错误", "程序崩溃", strength=0.7)
19
ducg_net.add_edge("逻辑错误", "测试失败率", strength=0.8)
20

21
# 内存泄漏 -> 症状
22
ducg_net.add_edge("内存泄漏", "内存占用异常", strength=0.9)
23
ducg_net.add_edge("内存泄漏", "性能下降", strength=0.6)
24

25
# 并发问题 -> 症状
26
ducg_net.add_edge("并发问题", "程序崩溃", strength=0.5)
27
ducg_net.add_edge("并发问题", "性能下降", strength=0.7)
28

29
# 症状 -> 可观测
30
ducg_net.add_edge("程序崩溃", "异常数量", strength=0.85)
31
ducg_net.add_edge("性能下降", "代码覆盖率", strength=0.4)
32
ducg_net.add_edge("内存占用异常", "性能下降", strength=0.6)
33

34
# 汇聚到目标
35
ducg_net.add_edge("测试失败率", "缺陷严重程度", strength=0.7)
36
ducg_net.add_edge("异常数量", "缺陷严重程度", strength=0.8)
37

38
# 可视化网络
39
ducg_net.visualize(layout='hierarchical')

1
import numpy as np
2
from sklearn.linear_model import LogisticRegression
3

4
def learn_causal_strength(data, cause_var, effect_var):
5
    """
6
    从数据中学习因果强度
7
    使用逻辑回归估计因果效应
8
    """
9
    X = data[[cause_var]].values
10
    y = data[effect_var].values
11

12
    # 训练逻辑回归模型
13
    model = LogisticRegression()
14
    model.fit(X, y)
15

16
    # 因果强度 = 回归系数的归一化值
17
    coefficient = model.coef_[0][0]
18
    strength = 1 / (1 + np.exp(-coefficient))  # Sigmoid归一化到[0,1]
19

20
    return strength
21

22
# 批量学习所有边的强度
23
def learn_all_strengths(ducg_network, data):
24
    """学习网络中所有边的因果强度"""
25
    for parent, child in ducg_network.graph.edges():
26
        strength = learn_causal_strength(data, parent, child)
27
        ducg_network.set_causal_strength(parent, child, strength)
28
        print(f"{parent} -> {child}: 强度 = {strength:.3f}")
29

30
# 使用示例
31
data = pd.read_csv('defect_data.csv')
32
learn_all_strengths(ducg_net, data)

1
def calibrate_with_expert_knowledge(learned_strength, expert_strength,
2
                                   data_confidence, expert_confidence):
3
    """
4
    融合数据学习的强度和专家给出的强度
5
    使用贝叶斯加权平均
6
    """
7
    total_confidence = data_confidence + expert_confidence
8

9
    calibrated_strength = (
10
        (learned_strength * data_confidence +
11
         expert_strength * expert_confidence) /
12
        total_confidence
13
    )
14

15
    return calibrated_strength
16

17
# 示例：校准"内存泄漏 -> 内存占用异常"的强度
18
learned = 0.85
19
expert = 0.90
20
calibrated = calibrate_with_expert_knowledge(
21
    learned, expert,
22
    data_confidence=0.7,
23
    expert_confidence=0.9
24
)
25
print(f"校准后的因果强度: {calibrated:.3f}")

1
def estimate_cpt(data, variable, parent_variables):
2
    """
3
    估计条件概率表
4
    P(Variable | Parents)
5
    """
6
    if not parent_variables:
7
        # 无父节点，估计先验概率
8
        return data[variable].value_counts(normalize=True).to_dict()
9

10
    # 有父节点，估计条件概率
11
    cpt = {}
12

13
    # 枚举所有父节点状态组合
14
    parent_states = [data[p].unique() for p in parent_variables]
15

16
    from itertools import product
17
    for states in product(*parent_states):
18
        # 筛选满足父节点状态的数据
19
        mask = True
20
        for p, state in zip(parent_variables, states):
21
            mask &= (data[p] == state)
22

23
        subset = data[mask]
24

25
        if len(subset) > 0:
26
            # 计算条件概率
27
            prob_dist = subset[variable].value_counts(normalize=True).to_dict()
28
            cpt[states] = prob_dist
29
        else:
30
            # 数据稀疏，使用均匀分布
31
            var_states = data[variable].unique()
32
            cpt[states] = {s: 1.0/len(var_states) for s in var_states}
33

34
    return cpt
35

36
# 示例：估计"缺陷严重程度"的CPT
37
parents = ["测试失败率", "异常数量"]
38
cpt = estimate_cpt(data, "缺陷严重程度", parents)
39

40
print("条件概率表:")
41
for parent_states, probs in cpt.items():
42
    print(f"  给定父状态 {parent_states}:")
43
    for var_state, prob in probs.items():
44
        print(f"    P(缺陷严重程度={var_state}) = {prob:.3f}")

1
def bayesian_cpt_estimation(data, variable, parent_variables, prior_strength=1.0):
2
    """
3
    使用贝叶斯估计处理数据稀疏问题
4
    添加伪计数（Laplace平滑）
5
    """
6
    cpt = {}
7
    var_states = data[variable].unique()
8
    n_states = len(var_states)
9

10
    parent_states = [data[p].unique() for p in parent_variables]
11

12
    from itertools import product
13
    for states in product(*parent_states):
14
        mask = True
15
        for p, state in zip(parent_variables, states):
16
            mask &= (data[p] == state)
17

18
        subset = data[mask]
19

20
        # 计算计数 + 伪计数
21
        counts = subset[variable].value_counts().to_dict()
22

23
        total_count = len(subset) + prior_strength * n_states
24

25
        prob_dist = {}
26
        for var_state in var_states:
27
            count = counts.get(var_state, 0)
28
            # 贝叶斯估计
29
            prob_dist[var_state] = (count + prior_strength) / total_count
30

31
        cpt[states] = prob_dist
32

33
    return cpt

1
from sklearn.model_selection import KFold
2

3
def cross_validate_ducg(ducg_network, data, n_folds=5):
4
    """
5
    交叉验证DUCG模型
6
    """
7
    kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
8

9
    accuracies = []
10

11
    for train_idx, test_idx in kf.split(data):
12
        train_data = data.iloc[train_idx]
13
        test_data = data.iloc[test_idx]
14

15
        # 在训练集上学习参数
16
        learn_all_strengths(ducg_network, train_data)
17

18
        # 在测试集上评估
19
        predictions = []
20
        actuals = []
21

22
        for _, row in test_data.iterrows():
23
            evidence = row.to_dict()
24
            prediction = ducg_network.infer(evidence, target="缺陷严重程度")
25
            predictions.append(prediction)
26
            actuals.append(row["缺陷严重程度"])
27

28
        # 计算准确率
29
        accuracy = np.mean(np.array(predictions) == np.array(actuals))
30
        accuracies.append(accuracy)
31

32
    return {
33
        'mean_accuracy': np.mean(accuracies),
34
        'std_accuracy': np.std(accuracies),
35
        'fold_accuracies': accuracies
36
    }
37

38
# 使用示例
39
cv_results = cross_validate_ducg(ducg_net, data)
40
print(f"交叉验证准确率: {cv_results['mean_accuracy']:.3f} ± {cv_results['std_accuracy']:.3f}")