本文目录导读：
的模糊粗糙决策，我将从数据预处理、模糊集构建、粗糙集近似和决策规则生成几个方面来说明。
整体架构设计
import numpy as np
import pandas as pd
from typing import List, Dict, Tuple
import re
class FuzzyRoughDecisionSystem:
def __init__(self, file_path: str):
self.data = self.load_data(file_path)
self.fuzzy_relations = {}
self.lower_approximations = {}
self.upper_approximations = {}
self.decision_rules = []
数据加载与预处理
def load_data(self, file_path: str) -> pd.DataFrame:
"""加载并预处理文件数据"""
# 支持多种文件格式
if file_path.endswith('.csv'):
df = pd.read_csv(file_path)
elif file_path.endswith('.xlsx') or file_path.endswith('.xls'):
df = pd.read_excel(file_path)
elif file_path.endswith('.txt'):
# 文本文件处理
with open(file_path, 'r', encoding='utf-8') as f:
lines = f.readlines()
# 解析为结构化数据
data = []
for line in lines:
parts = line.strip().split('\t')
if len(parts) >= 2:
data.append(parts)
df = pd.DataFrame(data[1:], columns=data[0])
else:
raise ValueError(f"不支持的文件格式: {file_path}")
# 数据清洗
df = df.dropna()
df = df.reset_index(drop=True)
return df
def preprocess_data(self, continuous_columns: List[str]) -> pd.DataFrame:
"""预处理连续数据，离散化和标准化"""
processed = self.data.copy()
for col in continuous_columns:
if col in processed.columns:
# 标准化
processed[col] = (processed[col] - processed[col].mean()) / processed[col].std()
return processed
模糊关系构建
def build_fuzzy_relation(self, attributes: List[str], method: str = 'gaussian'):
"""构建模糊相似关系"""
n = len(self.data)
fuzzy_matrix = np.zeros((n, n))
for i in range(n):
for j in range(i, n):
# 计算对象间的模糊相似度
similarity = self._calculate_similarity(i, j, attributes, method)
fuzzy_matrix[i][j] = similarity
fuzzy_matrix[j][i] = similarity
self.fuzzy_relations[tuple(attributes)] = fuzzy_matrix
return fuzzy_matrix
def _calculate_similarity(self, i: int, j: int, attributes: List[str], method: str) -> float:
"""计算两个对象的相似度"""
if method == 'gaussian':
# 高斯相似度
distance = 0
for attr in attributes:
if attr in self.data.columns:
diff = float(self.data[attr].iloc[i]) - float(self.data[attr].iloc[j])
distance += diff ** 2
similarity = np.exp(-np.sqrt(distance) / 2)
elif method == 'triangular':
# 三角模糊相似度
distances = []
for attr in attributes:
if attr in self.data.columns:
diff = abs(float(self.data[attr].iloc[i]) - float(self.data[attr].iloc[j]))
distances.append(1 - min(diff, 1))
similarity = np.mean(distances) if distances else 0
elif method == 'min_max':
# 最小-最大相似度
distances = []
for attr in attributes:
if attr in self.data.columns:
val_i, val_j = float(self.data[attr].iloc[i]), float(self.data[attr].iloc[j])
min_val = min(val_i, val_j)
max_val = max(val_i, val_j)
distances.append(min_val / max_val if max_val > 0 else 0)
similarity = np.mean(distances) if distances else 0
else:
raise ValueError(f"不支持的模糊方法: {method}")
return similarity
粗糙集近似计算
def compute_lower_approximation(self, decision_attr: str, threshold: float = 0.5):
"""计算下近似"""
decisions = self.data[decision_attr].unique()
self.lower_approximations = {}
for decision in decisions:
# 获取该决策类别的对象索引
class_indices = self.data[self.data[decision_attr] == decision].index.tolist()
lower_approx = []
# 对每个对象计算下近似
for i in range(len(self.data)):
fuzzy_relation = self.fuzzy_relations.get(tuple([decision_attr]))
if fuzzy_relation is not None:
# 检查对象是否肯定属于该决策类别
membership = np.mean([fuzzy_relation[i][j] for j in class_indices])
if membership >= threshold:
lower_approx.append(i)
self.lower_approximations[decision] = lower_approx
return self.lower_approximations
def compute_upper_approximation(self, decision_attr: str, threshold: float = 0.5):
"""计算上近似"""
decisions = self.data[decision_attr].unique()
self.upper_approximations = {}
for decision in decisions:
class_indices = self.data[self.data[decision_attr] == decision].index.tolist()
upper_approx = []
# 对每个对象计算上近似
for i in range(len(self.data)):
fuzzy_relation = self.fuzzy_relations.get(tuple([decision_attr]))
if fuzzy_relation is not None:
# 检查对象是否可能属于该决策类别
membership = np.max([fuzzy_relation[i][j] for j in class_indices])
if membership > 0:
upper_approx.append(i)
self.upper_approximations[decision] = upper_approx
return self.upper_approximations
决策规则生成
def generate_decision_rules(self, condition_attrs: List[str], decision_attr: str,
min_confidence: float = 0.6, min_support: float = 0.1):
"""生成模糊粗糙决策规则"""
rules = []
# 构建条件属性的模糊关系
condition_relation = self.build_fuzzy_relation(condition_attrs)
# 获取决策类别
decisions = self.data[decision_attr].unique()
for decision in decisions:
class_indices = self.data[self.data[decision_attr] == decision].index.tolist()
# 为每个条件组合生成规则
for i in range(len(self.data)):
rule_antecedent = []
for attr in condition_attrs:
value = self.data[attr].iloc[i]
# 创建模糊规则前件
fuzzy_value = self._fuzzify_value(attr, value)
rule_antecedent.append(f"{attr} IS {fuzzy_value}")
# 计算规则置信度
confidence = self._calculate_rule_confidence(i, class_indices, condition_relation)
# 计算支持度
support = len(class_indices) / len(self.data)
if confidence >= min_confidence and support >= min_support:
rule = {
'antecedent': ' AND '.join(rule_antecedent),
'consequent': f"{decision_attr} IS {decision}",
'confidence': confidence,
'support': support
}
rules.append(rule)
self.decision_rules = rules
return rules
def _fuzzify_value(self, attr: str, value: float) -> str:
"""模糊化值"""
if isinstance(value, (int, float)):
if value < -0.5:
return "Low"
elif value < 0.5:
return "Medium"
else:
return "High"
else:
return str(value)
def _calculate_rule_confidence(self, obj_idx: int, class_indices: List[int],
relation_matrix: np.ndarray) -> float:
"""计算规则的置信度"""
similarity_scores = [relation_matrix[obj_idx][j] for j in class_indices]
if similarity_scores:
return np.mean(similarity_scores)
return 0
完整使用示例
def fuzzy_rough_decision_analysis(file_path: str, condition_attrs: List[str],
decision_attr: str, threshold: float = 0.5):
"""完整的模糊粗糙决策分析流程"""
# 1. 加载数据
system = FuzzyRoughDecisionSystem(file_path)
print(f"成功加载 {len(system.data)} 条数据记录")
# 2. 数据预处理
processed_data = system.preprocess_data(condition_attrs)
print("数据预处理完成")
# 3. 构建模糊关系
system.build_fuzzy_relation(condition_attrs, method='gaussian')
system.build_fuzzy_relation([decision_attr], method='gaussian')
# 4. 计算粗糙近似
lower_approx = system.compute_lower_approximation(decision_attr, threshold)
upper_approx = system.compute_upper_approximation(decision_attr, threshold)
print("\n=== 粗糙集近似结果 ===")
for decision in system.data[decision_attr].unique():
print(f"决策 '{decision}':")
print(f" 下近似对象数: {len(lower_approx.get(decision, []))}")
print(f" 上近似对象数: {len(upper_approx.get(decision, []))}")
print(f" 边界区域对象数: {len(upper_approx.get(decision, [])) - len(lower_approx.get(decision, []))}")
# 5. 生成决策规则
rules = system.generate_decision_rules(condition_attrs, decision_attr)
print("\n=== 决策规则 ===")
for i, rule in enumerate(rules[:10]): # 只显示前10条规则
print(f"规则 {i+1}:")
print(f" IF {rule['antecedent']}")
print(f" THEN {rule['consequent']}")
print(f" 置信度: {rule['confidence']:.3f}")
print(f" 支持度: {rule['support']:.3f}\n")
# 6. 返回结果
return {
'lower_approximations': lower_approx,
'upper_approximations': upper_approx,
'decision_rules': rules,
'fuzzy_relations': system.fuzzy_relations
}
# 使用示例
if __name__ == "__main__":
# 示例用法
results = fuzzy_rough_decision_analysis(
file_path="data.csv",
condition_attrs=["Age", "Income", "Education"],
decision_attr="Risk_Level",
threshold=0.5
)
# 导出规则到文件
with open("decision_rules.txt", "w", encoding="utf-8") as f:
for rule in results['decision_rules']:
f.write(f"IF {rule['antecedent']} THEN {rule['consequent']}\n")
f.write(f"置信度: {rule['confidence']:.3f}, 支持度: {rule['support']:.3f}\n\n")
高级功能扩展
def attribute_reduction(self, decision_attr: str, threshold: float = 0.5):
"""属性约简 - 删除冗余属性"""
attributes = [col for col in self.data.columns if col != decision_attr]
# 计算属性重要性
importance_scores = {}
for attr in attributes:
# 计算包含该属性时的分类质量
system_with_attr = FuzzyRoughDecisionSystem.__new__(FuzzyRoughDecisionSystem)
system_with_attr.data = self.data
system_with_attr.build_fuzzy_relation([attr, decision_attr])
lower_approx = system_with_attr.compute_lower_approximation(decision_attr, threshold)
# 计算分类质量
quality = sum(len(approx) for approx in lower_approx.values()) / len(self.data)
importance_scores[attr] = quality
# 按重要性排序
sorted_attributes = sorted(importance_scores.items(), key=lambda x: x[1], reverse=True)
# 选择重要属性子集
reduced_set = []
for attr, score in sorted_attributes:
if score >= threshold:
reduced_set.append(attr)
return reduced_set, importance_scores
def evaluate_rules(self, test_data: pd.DataFrame, decision_attr: str):
"""评估决策规则在测试数据上的表现"""
correct = 0
total = len(test_data)
for idx, row in test_data.iterrows():
predicted = self._predict_with_rules(row)
actual = row[decision_attr]
if predicted == actual:
correct += 1
accuracy = correct / total if total > 0 else 0
return accuracy, correct, total
这个实现提供了一个完整的文件内容模糊粗糙决策系统，包括数据加载、预处理、模糊关系构建、粗糙近似计算和决策规则生成等核心功能,可以根据具体需求调整参数和扩展功能。
