Coverage for agentos/marketplace/skills/data-analysis/data-analysis.py: 2%

1""" 

2data-analysis — 基本数据分析：描述性统计、相关性、频率分布。 

3 

4Category: data 

5""" 

6 

7 

8def run( 

9 action: str, 

10 file_path: str = "", 

11 column: str = "", 

12 group_by: str = "", 

13 delimiter: str = ",", 

14 encoding: str = "utf-8", 

15) -> str: 

16 """数据分析工具。action: describe/correlation/freq/top_n。输入为 CSV 文件。""" 

17 import collections 

18 import csv 

19 import math 

20 import os 

21 

22 if not file_path or not os.path.isfile(file_path): 

23 return f"[data-analysis] 文件不存在: {file_path}" 

24 

25 try: 

26 with open(file_path, encoding=encoding, newline="") as f: 

27 reader = csv.DictReader(f, delimiter=delimiter) 

28 if reader.fieldnames is None: 

29 return "[data-analysis] 无法解析表头" 

30 headers = list(reader.fieldnames) 

31 rows = list(reader) 

32 

33 def _numeric(col): 

34 vals = [] 

35 for row in rows: 

36 v = row.get(col, "") 

37 try: 

38 vals.append(float(v)) 

39 except ValueError: 

40 pass 

41 return vals 

42 

43 if action == "describe": 

44 target = [column] if column and column in headers else headers 

45 result = [f"文件: {file_path}", f"行数: {len(rows)}", ""] 

46 for col in target: 

47 vals = _numeric(col) 

48 if not vals: 

49 result.append(f"{col}: 无数值数据") 

50 continue 

51 n = len(vals) 

52 mean = sum(vals) / n 

53 srt = sorted(vals) 

54 median = srt[n // 2] if n % 2 else (srt[n // 2 - 1] + srt[n // 2]) / 2 

55 var = sum((x - mean) ** 2 for x in vals) / n 

56 std = math.sqrt(var) 

57 result.append( 

58 f"{col}: count={n}, mean={mean:.4f}, std={std:.4f}, " 

59 f"min={min(vals):.4f}, 25%={srt[n//4]:.4f}, median={median:.4f}, " 

60 f"75%={srt[3*n//4]:.4f}, max={max(vals):.4f}" 

61 ) 

62 return "\n".join(result) 

63 

64 if action == "correlation": 

65 numeric_cols = [h for h in headers if _numeric(h)] 

66 if len(numeric_cols) < 2: 

67 return "[data-analysis] 需要至少 2 个数值列" 

68 if column: 

69 targets = [column] if column in numeric_cols else numeric_cols[:2] 

70 else: 

71 targets = numeric_cols[: min(5, len(numeric_cols))] 

72 

73 def _pearson(xs, ys): 

74 n = min(len(xs), len(ys)) 

75 mx, my = sum(xs) / n, sum(ys) / n 

76 num = sum((xs[i] - mx) * (ys[i] - my) for i in range(n)) 

77 dx = math.sqrt(sum((x - mx) ** 2 for x in xs)) 

78 dy = math.sqrt(sum((y - my) ** 2 for y in ys)) 

79 return num / (dx * dy) if dx and dy else 0 

80 

81 result = ["相关性矩阵:", ""] 

82 result.append(" " + " ".join(f"{t:>8}" for t in targets)) 

83 for t1 in targets: 

84 row_vals = [f"{t1:<8}"] 

85 vals1 = _numeric(t1) 

86 for t2 in targets: 

87 if t1 == t2: 

88 row_vals.append(f"{1.0:>8.3f}") 

89 elif t1 < t2: 

90 row_vals.append(f"{_pearson(vals1,_numeric(t2)):>8.3f}") 

91 else: 

92 row_vals.append(" ") 

93 result.append(" ".join(row_vals)) 

94 return "\n".join(result) 

95 

96 if action == "freq": 

97 if not column or column not in headers: 

98 return f"[data-analysis] 请指定有效列名。可用: {', '.join(headers)}" 

99 counter = collections.Counter(row.get(column, "") for row in rows) 

100 lines = [f"{column} 频率分布 (共{len(counter)}个不同值):"] 

101 for val, cnt in counter.most_common(20): 

102 pct = cnt / len(rows) * 100 

103 lines.append(f" {val}: {cnt} ({pct:.1f}%)") 

104 return "\n".join(lines) 

105 

106 if action == "top_n": 

107 if not column or column not in headers: 

108 return f"[data-analysis] 请指定有效列名。可用: {', '.join(headers)}" 

109 vals = [] 

110 for row in rows: 

111 v = row.get(column, "") 

112 try: 

113 vals.append((float(v), row)) 

114 except ValueError: 

115 pass 

116 vals.sort(key=lambda x: x[0], reverse=True) 

117 lines = [f"{column} Top 10:"] 

118 for i, (val, row) in enumerate(vals[:10]): 

119 lines.append(f" {i+1}. {column}={val} | {dict(row)}") 

120 return "\n".join(lines) 

121 

122 return f"[data-analysis] 未知操作: {action}, 支持: describe/correlation/freq/top_n" 

123 except Exception as e: 

124 return f"[data-analysis] 失败: {e}" 

125 

126 

127__all__ = ["run"]