"""频数、描述统计及列联表；明确记录各类百分比的分母。""" import numpy as np import pandas as pd from scipy import stats from flask_babel import gettext as _ from .contracts import Column, PaperTable, StatisticalResult, columns, numeric_frame, enough, choice def frequency(data, options): selected = columns(data, options.get('variables')) rows, details = [], {} enough(data, 1) for name in selected: series = data[name] valid_n = int(series.notna().sum()) details[name] = {'total_n': len(series), 'valid_n': valid_n, 'missing_n': len(series)-valid_n} for value, count in series.value_counts(sort=False).items(): rows.append({'variable': name, 'category': str(value), 'n': int(count), 'percent': 100*count/len(series), 'valid_percent': 100*count/valid_n}) if series.isna().any(): rows.append({'variable': name, 'category': _('缺失'), 'n': len(series)-valid_n, 'percent': 100*(len(series)-valid_n)/len(series), 'valid_percent': None}) table = PaperTable(_('频数分析'), [Column('variable', _('变量'), 'text', merge=True), Column('category', _('类别'), 'text'), Column('n', _('频数'), 'integer'), Column('percent', _('百分比'), 'percent'), Column('valid_percent', _('有效百分比'), 'percent')], rows) return StatisticalResult([table], details) def _same_code(value, expected): """允许界面以文本传入数值编码，但不混淆普通类别文本。""" if value == expected: return True try: return float(value) == float(expected) except (TypeError, ValueError): return str(value) == str(expected) def multi_response(data, options): selected = columns(data, options.get('variables')) mode = choice(options, 'response_mode', 'binary', ('binary', 'slots')) frame = data[selected] # 整行缺失不进入分母；填写了全部0的个案仍属于有效个案。 frame = enough(frame.loc[frame.notna().any(axis=1)], 1) counts = {} if mode == 'binary': yes, no = options.get('selected_code', 1), options.get('unselected_code', 0) if _same_code(yes, no): raise ValueError(_('选中与未选中的编码必须不同')) for name in selected: values = frame[name].dropna() if any(not _same_code(v, yes) and not _same_code(v, no) for v in values): raise ValueError(_('多选题变量“%(name)s”包含未定义的编码', name=name)) counts[name] = sum(_same_code(v, yes) for v in values) else: for values in frame.itertuples(index=False, name=None): # 同一个受访者重复填写同一选项只计一次。 chosen = {str(v) for v in values if not pd.isna(v) and str(v).strip()} for category in sorted(chosen): counts[category] = counts.get(category, 0) + 1 responses = sum(counts.values()) if not responses: raise ValueError(_('有效个案没有选择任何选项，无法计算响应百分比')) rows = [{'category': key, 'n': count, 'response_percent': 100*count/responses, 'case_percent': 100*count/len(frame)} for key, count in counts.items()] table = PaperTable(_('多重响应分析'), [Column('category', _('选项'), 'text'), Column('n', _('响应数'), 'integer'), Column('response_percent', _('响应百分比'), 'percent'), Column('case_percent', _('个案百分比'), 'percent')], rows, _('有效个案N=%(n)s；响应总数=%(responses)s。', n=len(frame), responses=responses)) return StatisticalResult([table], {'valid_cases': len(frame), 'total_responses': responses, 'excluded_cases': len(data)-len(frame), 'coding': mode}) def descriptive(data, options, grouped=False): selected = columns(data, options.get('variables')) numeric = numeric_frame(data, selected) available = { 'n': (_('N'), 'integer'), 'mean': (_('均值'), 'number'), 'sd': (_('标准差'), 'number'), 'median': (_('中位数'), 'number'), 'q1': (_('下四分位数'), 'number'), 'q3': (_('上四分位数'), 'number'), 'min': (_('最小值'), 'number'), 'max': (_('最大值'), 'number'), 'skewness': (_('偏度'), 'number'), 'kurtosis': (_('峰度'), 'number'), } metrics = options.get('metrics', ['n', 'mean', 'sd', 'median', 'min', 'max']) if not isinstance(metrics, list) or not metrics or len(set(metrics)) != len(metrics) or any(k not in available for k in metrics): raise ValueError(_('请选择有效的描述统计项目')) parts = [(None, numeric)] if grouped: group = columns(data, [options.get('group')])[0] if group in selected: raise ValueError(_('分组变量不能同时作为分析变量')) parts = [(str(key), numeric.loc[index]) for key, index in data.groupby(group, sort=False, observed=True).groups.items()] if not parts: raise ValueError(_('分组变量没有有效类别')) rows = [] for label, part in parts: for name in selected: values = part[name].dropna() row = {'variable': name, 'group': label, 'n': len(values), 'mean': values.mean(), 'sd': values.std(ddof=1), 'median': values.median(), 'q1': values.quantile(.25), 'q3': values.quantile(.75), 'min': values.min(), 'max': values.max(), 'skewness': values.skew(), 'kurtosis': values.kurt()} rows.append(row) head = [Column('variable', _('变量'), 'text')] if grouped: head.append(Column('group', _('组别'), 'text')) head += [Column(key, *available[key]) for key in metrics] return StatisticalResult([PaperTable(_('分组描述性统计') if grouped else _('描述性统计'), head, rows)], {'missing_policy': 'available_per_variable', 'total_n': len(data), 'kurtosis_definition': 'Fisher_excess_unbiased', 'descriptive': rows}) def crosstab(data, options): selected = columns(data, options.get('variables'), 2, 2) sample = enough(data[selected].dropna(), 2) table = pd.crosstab(sample[selected[0]], sample[selected[1]], dropna=True) if min(table.shape) < 2 or max(table.shape) > 50: raise ValueError(_('列联表的行、列变量都需要2至50个有效类别')) test = choice(options, 'test', 'chi2', ('chi2', 'fisher')) percent = choice(options, 'percent', 'row', ('row', 'column', 'total', 'none')) chi2, p_value, df, expected = stats.chi2_contingency(table, correction=False) association = float(np.sqrt(chi2/(table.values.sum()*min(table.shape[0]-1, table.shape[1]-1)))) raw = {'n': len(sample), 'observed': table, 'expected': expected, 'chi2': chi2, 'df': df, 'chi2_p': p_value, 'cramers_v': association, 'expected_below_five': int((expected < 5).sum())} if test == 'fisher': if table.shape != (2, 2): raise ValueError(_('Fisher精确检验首版适用于2×2列联表')) odds, p_value = stats.fisher_exact(table.values, alternative='two-sided') raw.update(odds_ratio=odds, fisher_p=p_value) rows, headers = [], [Column('category', selected[0], 'text')] for i, label in enumerate(table.columns): headers.append(Column('c'+str(i), str(label), 'text')) headers.append(Column('total', _('合计'), 'integer')) for label, cells in table.iterrows(): row = {'category': str(label), 'total': int(cells.sum())} for i, count in enumerate(cells): denominator = cells.sum() if percent == 'row' else table.iloc[:, i].sum() if percent == 'column' else len(sample) row['c'+str(i)] = str(count) if percent == 'none' else '%d (%.1f%%)' % (count, 100*count/denominator) rows.append(row) p_text = '<0.001' if p_value < .001 else '=%.3f' % p_value test_text = 'Fisher' if test == 'fisher' else 'χ²(%d)=%.3f' % (df, chi2) note = _('N=%(n)s；%(test)s，p%(p)s；Cramér’s V=%(v)s。', n=len(sample), test=test_text, p=p_text, v='%.3f' % association) return StatisticalResult([PaperTable(_('列联表分析'), headers, rows, note)], raw)