Python问卷数据清洗实战：直线作答秒填规律异常检测代

脚本专家

在问卷研究中，回收的数据难免混入大量无效样本：直线作答、秒填、规律选择、前后矛盾等。这些数据若不清理，会严重污染分析结果。本文提供一套完整的Python清洗流程，覆盖最常见的无效样本类型，并封装为可复用的类。

## 环境准备
需要 Python 3.8+，以及 pandas、numpy、matplotlib。安装命令：

2. pip install pandas numpy matplotlib openpyxl

复制代码

问卷星导出的数据通常结构为：一行一份问卷，一列一个题目，包含序号、提交时间、各题目答案。

## 数据加载与列识别
使用 pandas 读取 Excel 或 CSV 文件：

2. import pandas as pd
3. import numpy as np
4. df = pd.read_excel('问卷数据.xlsx')
5. print(f"原始数据量: {len(df)} 份")

复制代码

自动识别题目列（包含 'Q'、'第'、'题' 等关键词）和时间列：

2. question_cols = [c for c in df.columns if c.startswith('Q') or '第' in str(c)]
3. time_col = [c for c in df.columns if '时间' in str(c) or 'time' in str(c).lower()]
4. print(f"题目列: {question_cols}")
5. print(f"时间列: {time_col}")

复制代码

## 检测直线作答（Straight-lining）
所有题目选择完全相同的答案，是最常见的无效模式。以下函数检测完全一致的样本：

2. def detect_straight_lining(df, question_cols):
3.     invalid_indices = []
4.     for idx, row in df.iterrows():
5.         answers = row[question_cols].dropna().tolist()
6.         if len(answers) < 3:
7.             continue
8.         if len(set(answers)) == 1:
9.             invalid_indices.append(idx)
10.     return set(invalid_indices)

复制代码

为了避免误判（例如全部选“满意”可能是真实），增加接近直线检测：超过80%的答案相同。

2. from collections import Counter
3. def detect_near_straight_lining(df, question_cols, threshold=0.8):
4.     invalid_indices = []
5.     for idx, row in df.iterrows():
6.         answers = row[question_cols].dropna().tolist()
7.         if len(answers) < 3:
8.             continue
9.         counts = Counter(answers)
10.         most_common_ratio = counts.most_common(1)[0][1] / len(answers)
11.         if most_common_ratio >= threshold:
12.             invalid_indices.append(idx)
13.     return set(invalid_indices)

复制代码

## 检测秒填（Speeders）
如果问卷包含答题时长列，直接按阈值过滤。常见规则：10题以内<10秒，10-20题<20秒，20题以上<30秒。

2. def detect_speeders(df, time_col, min_seconds=15):
3.     if not time_col:
4.         return set()
5.     duration_col = [c for c in df.columns if '时长' in str(c) or '用时' in str(c)]
6.     if duration_col:
7.         durations = pd.to_numeric(df[duration_col[0]], errors='coerce')
8.         return set(df[durations < min_seconds].index)
9.     return set()

复制代码

## 检测规律作答（Pattern Response）
如 1-2-3-4-5 的递增/递减序列，或 A-B-A-B 的交替模式。

2. def detect_pattern_response(df, question_cols):
3.     invalid_indices = []
4.     for idx, row in df.iterrows():
5.         answers = row[question_cols].dropna().tolist()
6.         if len(answers) < 5:
7.             continue
8.         try:
9.             numeric = pd.to_numeric(answers)
10.         except:
11.             if len(answers) >= 6:
12.                 half = len(answers) // 2
13.                 if answers[:half] == answers[half:half*2]:
14.                     invalid_indices.append(idx)
15.             continue
16.         diffs = np.diff(numeric)
17.         if np.all(diffs == diffs[0]) and diffs[0] != 0:
18.             invalid_indices.append(idx)
19.             continue
20.         if len(numeric) >= 6:
21.             even_vals = set(numeric[::2])
22.             odd_vals = set(numeric[1::2])
23.             if len(even_vals) == 1 and len(odd_vals) == 1:
24.                 invalid_indices.append(idx)
25.     return set(invalid_indices)

复制代码

## 检测矛盾回答
针对跳题逻辑，例如 Q1选“否”但后续题仍有内容。需要根据具体问卷编写：

2. def detect_contradictions(df, filter_col, filter_value, check_col):
3.     mask = df[filter_col] == filter_value
4.     should_be_empty = df.loc[mask, check_col]
5.     contradictions = should_be_empty[should_be_empty.notna() & (should_be_empty != '')]
6.     return set(contradictions.index)

复制代码

## 检测异常值（Z-score）
对量表题（1-5分或1-7分）计算每个样本的均值Z分数，超过阈值（通常3.0）视为异常：

2. def detect_outliers_zscore(df, question_cols, threshold=3.0):
3.     numeric_df = df[question_cols].apply(pd.to_numeric, errors='coerce')
4.     row_means = numeric_df.mean(axis=1)
5.     z_scores = (row_means - row_means.mean()) / row_means.std()
6.     return set(df[abs(z_scores) > threshold].index)

复制代码

## 汇总无效样本与标记
合并所有检测结果，并在DataFrame中添加标记列，导出清洗后的有效数据和无效明细：

2. all_invalid = straight_lining | near_straight | speeders | patterns | outliers
3. df['is_invalid'] = df.index.isin(all_invalid)
4. df['invalid_reason'] = ''
5. df.loc[df.index.isin(straight_lining), 'invalid_reason'] += '直线作答;'
6. df.loc[df.index.isin(near_straight), 'invalid_reason'] += '接近直线;'
7. df.loc[df.index.isin(speeders), 'invalid_reason'] += '秒填;'
8. df.loc[df.index.isin(patterns), 'invalid_reason'] += '规律作答;'
9. df.loc[df.index.isin(outliers), 'invalid_reason'] += '统计异常;'
10. clean_df = df[~df['is_invalid']].copy()
11. clean_df.to_excel('问卷数据_已清洗.xlsx', index=False)
12. invalid_df = df[df['is_invalid']].copy()
13. invalid_df.to_excel('无效样本明细.xlsx', index=False)

复制代码

## 可视化清洗结果
使用matplotlib绘制无效类型柱状图和有效率饼图（需安装中文字体）：

2. import matplotlib.pyplot as plt
3. plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']
4. plt.rcParams['axes.unicode_minus'] = False
5. reasons = {
6.     '直线作答': len(straight_lining),
7.     '接近直线': len(near_straight - straight_lining),
8.     '秒填': len(speeders),
9.     '规律作答': len(patterns),
10.     '统计异常': len(outliers),
11. }
12. fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
13. ax1.bar(reasons.keys(), reasons.values(), color='#ff6b6b')
14. ax1.set_title('无效样本类型分布')
15. ax2.pie([len(clean_df), len(all_invalid)], labels=['有效样本', '无效样本'], autopct='%1.1f%%', colors=['#51cf66', '#ff6b6b'])
16. ax2.set_title(f'样本有效率 (总计 {len(df)} 份)')
17. plt.tight_layout()
18. plt.show()

复制代码

## 封装为可复用的类
将上述函数整合成 SurveyDataCleaner 类，支持一键运行：

2. import pandas as pd
3. import numpy as np
4. from collections import Counter
6. class SurveyDataCleaner:
7.     def __init__(self, filepath: str, question_cols: list = None):
8.         if filepath.endswith('.xlsx'):
9.             self.df = pd.read_excel(filepath)
10.         else:
11.             self.df = pd.read_csv(filepath)
12.         self.question_cols = question_cols or self._auto_detect_questions()
13.         self.invalid_sets = {}
15.     def _auto_detect_questions(self):
16.         return [c for c in self.df.columns if any(k in str(c) for k in ['Q', '第', '题'])]
18.     def run_all(self, min_seconds=15, z_threshold=3.0, straight_threshold=0.8):
19.         self.invalid_sets['直线作答'] = self._detect_straight()
20.         self.invalid_sets['接近直线'] = self._detect_near_straight(straight_threshold)
21.         self.invalid_sets['秒填'] = self._detect_speeders(min_seconds)
22.         self.invalid_sets['规律作答'] = self._detect_pattern()
23.         self.invalid_sets['统计异常'] = self._detect_outliers(z_threshold)
24.         return self
26.     def _detect_straight(self):
27.         invalid = set()
28.         for idx, row in self.df.iterrows():
29.             ans = row[self.question_cols].dropna().tolist()
30.             if len(ans) >= 3 and len(set(ans)) == 1:
31.                 invalid.add(idx)
32.         return invalid
34.     def _detect_near_straight(self, threshold):
35.         invalid = set()
36.         for idx, row in self.df.iterrows():
37.             ans = row[self.question_cols].dropna().tolist()
38.             if len(ans) >= 3:
39.                 counts = Counter(ans)
40.                 if counts.most_common(1)[0][1] / len(ans) >= threshold:
41.                     invalid.add(idx)
42.         return invalid
44.     def _detect_speeders(self, min_seconds):
45.         duration_col = [c for c in self.df.columns if '时长' in str(c) or '用时' in str(c)]
46.         if not duration_col:
47.             return set()
48.         durations = pd.to_numeric(self.df[duration_col[0]], errors='coerce')
49.         return set(self.df[durations < min_seconds].index)
51.     def _detect_pattern(self):
52.         invalid = set()
53.         for idx, row in self.df.iterrows():
54.             ans = row[self.question_cols].dropna().tolist()
55.             if len(ans) < 5:
56.                 continue
57.             try:
58.                 numeric = pd.array(ans, dtype='float')
59.                 diffs = np.diff(numeric)
60.                 if len(diffs) > 0 and np.all(diffs == diffs[0]) and diffs[0] != 0:
61.                     invalid.add(idx)
62.             except:
63.                 pass
64.         return invalid
66.     def _detect_outliers(self, threshold):
67.         numeric_df = self.df[self.question_cols].apply(pd.to_numeric, errors='coerce')
68.         row_means = numeric_df.mean(axis=1)
69.         z_scores = (row_means - row_means.mean()) / row_means.std()
70.         return set(self.df[abs(z_scores) > threshold].index)
72.     @property
73.     def all_invalid(self):
74.         result = set()
75.         for s in self.invalid_sets.values():
76.             result |= s
77.         return result
79.     @property
80.     def clean_data(self):
81.         return self.df[~self.df.index.isin(self.all_invalid)].copy()
83.     def report(self):
84.         total = len(self.df)
85.         invalid_count = len(self.all_invalid)
86.         print(f"\n{'='*50}")
87.         print(f"问卷数据清洗报告")
88.         print(f"{'='*50}")
89.         print(f"原始样本量: {total} 份")
90.         for reason, indices in self.invalid_sets.items():
91.             print(f"  {reason}: {len(indices)} 份")
92.         print(f"{'─'*50}")
93.         print(f"无效样本合计: {invalid_count} 份")
94.         print(f"有效样本: {total - invalid_count} 份")
95.         print(f"有效率: {(1 - invalid_count/total)*100:.1f}%")
96.         print(f"{'='*50}")
98.     def export(self, output_path='清洗结果'):
99.         clean = self.clean_data
100.         clean.to_excel(f'{output_path}_有效数据.xlsx', index=False)
101.         invalid_df = self.df[self.df.index.isin(self.all_invalid)].copy()
102.         invalid_df['无效原因'] = ''
103.         for reason, indices in self.invalid_sets.items():
104.             invalid_df.loc[invalid_df.index.isin(indices), '无效原因'] += f'{reason};'
105.         invalid_df.to_excel(f'{output_path}_无效样本.xlsx', index=False)
107. if __name__ == '__main__':
108.     cleaner = SurveyDataCleaner('问卷数据.xlsx')
109.     cleaner.run_all(min_seconds=15, z_threshold=3.0)
110.     cleaner.report()
111.     cleaner.export()

复制代码

## 关于样本质量的延伸
清洗后有效样本量可能不足。若无效率超过20%，建议检查数据来源。例如，相对于互填群，使用真实用户渠道（如问卷鸭等）的无效率通常更低。但无论来源如何，清洗流程不可或缺。以上代码覆盖了直线作答、秒填、规律模式、矛盾回答和统计异常等常见场景，可用于学术调研、市场研究等。如需进一步信效度分析，可结合 factor_analyzer 库。

热心网友5

感谢分享，这套清洗流程很实用，尤其是近直线检测和交替模式判断，考虑到了真实填写中可能出现的模糊情况。想请教一下，对于混合型问卷（既有评分题又有选择题），你是如何区分哪些列适用于这些检测逻辑的？另外，如果数据量很大（比如上万份），直接用循环逐行遍历会不会影响效率，有没有考虑用向量化操作来优化？

热心网友5

非常实用的分享！最近正好在处理一批问卷数据，直线作答和秒填确实是最头疼的问题。你提供的检测函数思路清晰，尤其是“接近直线检测”用占比阈值来避免误判，很合理。 想请教一下：在实际应用中，不同题目类型的取值范围差异较大（比如李克特5点量表 vs 多选题），你会建议针对不同的题目块分别设定阈值吗？另外，如果问卷中包含矩阵题（比如一行多题），题目列识别时是否需要特殊处理？期待继续补充逻辑矛盾检测的部分～

热心网友5

感谢分享！代码框架很清晰，直接复制就能跑起来，对处理问卷数据非常实用。我之前也遇到过直线作答和规律填答的问题，手动筛选太费时了，你这个封装成类的方式很棒。想请教一下，在实际项目中，如果问卷包含多选题（列名不规则），是否需要对question_cols的识别逻辑做额外处理？另外对于秒填检测，如果数据里没有单独的时长列，只靠提交时间计算间隔可行吗？希望多交流。