数据分析方法论：从数据到洞察的完整思维框架

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

分析类型矩阵:

  描述性分析 (Descriptive):
    问题: "发生了什么？"
    方法: 汇总统计、数据聚合
    工具: Excel、Tableau、Power BI
    示例:
      - 上月销售额是多少？
      - 网站流量趋势如何？
      - 用户增长曲线

  诊断性分析 (Diagnostic):
    问题: "为什么会发生？"
    方法: 下钻、相关性分析
    工具: SQL、Python Pandas
    示例:
      - 销售下降的原因是什么？
      - 哪些渠道转化率最高？
      - 用户流失的驱动因素

  预测性分析 (Predictive):
    问题: "可能发生什么？"
    方法: 回归、时间序列、ML
    工具: Python、R、MLflow
    示例:
      - 下季度销售额预测
      - 哪些用户可能流失？
      - 库存需求预测

  处方性分析 (Prescriptive):
    问题: "应该怎么做？"
    方法: 优化算法、模拟
    工具: 运筹优化、强化学习
    示例:
      - 最优定价策略
      - 库存补货决策
      - 营销预算分配

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

# 数据源分类框架
class DataSourceMapper:
    """企业数据源映射"""

    def __init__(self):
        self.data_sources = {
            # 第一方数据
            'first_party': {
                'web_analytics': {
                    'tools': ['Google Analytics', 'Mixpanel', 'Amplitude'],
                    'data_points': [
                        'page_views', 'sessions', 'users',
                        'events', 'conversions', 'funnel_steps'
                    ],
                    'value': 'High'
                },
                'crm_data': {
                    'tools': ['Salesforce', 'HubSpot', 'Pipedrive'],
                    'data_points': [
                        'leads', 'opportunities', 'customers',
                        'deals', 'activities', 'stage_history'
                    ],
                    'value': 'Critical'
                },
                'transaction_data': {
                    'tools': ['Database', 'Payment Gateway'],
                    'data_points': [
                        'orders', 'revenue', 'refunds',
                        'items', 'discounts', 'payments'
                    ],
                    'value': 'Critical'
                },
                'product_data': {
                    'tools': ['Product Analytics', 'Log Files'],
                    'data_points': [
                        'feature_usage', 'errors', 'performance',
                        'user_actions', 'sessions'
                    ],
                    'value': 'High'
                }
            },

            # 第二方数据
            'second_party': {
                'partners': {
                    'sources': ['Partner Platforms', 'Joint Ventures'],
                    'data_points': ['shared_customers', 'referrals'],
                    'access': 'Partnership Agreement'
                }
            },

            # 第三方数据
            'third_party': {
                'market_data': {
                    'sources': ['Industry Reports', 'Market Research'],
                    'data_points': ['market_size', 'trends', 'benchmarks'],
                    'cost': 'Medium to High'
                },
                'social_data': {
                    'sources': ['Social Media APIs', 'Listening Tools'],
                    'data_points': ['mentions', 'sentiment', 'engagement'],
                    'cost': 'Low to Medium'
                },
                'demographic': {
                    'sources': ['Census Data', 'Consumer Panels'],
                    'data_points': ['population', 'income', 'education'],
                    'cost': 'Low'
                }
            }
        }

    def recommend_priority(self, business_goal):
        """根据业务目标推荐数据优先级"""
        recommendations = {
            'user_growth': ['web_analytics', 'product_data', 'social_data'],
            'revenue_growth': ['transaction_data', 'crm_data', 'market_data'],
            'customer_retention': ['product_data', 'crm_data', 'web_analytics'],
            'market_expansion': ['market_data', 'demographic', 'partners']
        }
        return recommendations.get(business_goal, [])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

# R语言数据质量检查示例
library(dplyr)
library(tidyr)

check_data_quality <- function(df) {
  # 创建数据质量报告
  quality_report <- list(

    # 1. 完整性检查
    completeness = df %>%
      summarise(
        total_rows = n(),
        total_cols = ncol(.),
        missing_values = sum(is.na(.)),
        missing_pct = round(sum(is.na(.)) / (n() * ncol(.)) * 100, 2)
      ),

    # 2. 唯一性检查
    uniqueness = df %>%
      summarise(
        duplicate_rows = sum(duplicated(.)),
        unique_pct = round(n_distinct(.) / n() * 100, 2)
      ),

    # 3. 一致性检查
    consistency = df %>%
      summarise(
        date_range = paste(min(date, na.rm = TRUE),
                          max(date, na.rm = TRUE),
                          sep = " to "),
        negative_values = sum(select_if(., is.numeric) < 0, na.rm = TRUE)
      ),

    # 4. 准确性检查
    accuracy = list(
      outliers = detect_outliers(df),
      invalid_formats = check_formats(df)
    )
  )

  return(quality_report)
}

# 数据清洗函数
clean_data <- function(df) {
  cleaned_df <- df %>%

    # 处理缺失值
    mutate(
      across(where(is.numeric), ~ifelse(is.na(.), median(., na.rm = TRUE), .)),
      across(where(is.character), ~ifelse(is.na(.), "Unknown", .))
    ) %>%

    # 去除重复值
    distinct() %>%

    # 处理异常值
    mutate(
      across(where(is.numeric), ~ifelse(. > quantile(., 0.99, na.rm = TRUE),
                                        quantile(., 0.99, na.rm = TRUE), .))
    ) %>%

    # 标准化格式
    mutate(
      date = as.Date(date),
      email = tolower(email),
      phone = gsub("[^0-9]", "", phone)
    )

  return(cleaned_df)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

数据清洗检查清单:

  缺失值处理:
    - [ ] 识别缺失模式（MCAR/MAR/MNAR）
    - [ ] 删除缺失比例>30%的变量
    - [ ] 填充数值型变量（均值/中位数/预测）
    - [ ] 填充分类型变量（众数/新类别）
    - [ ] 记录缺失值处理方式

  异常值处理:
    - [ ] 可视化检测（箱线图/散点图）
    - [ ] 统计方法检测（Z-score/IQR）
    - [ ] 判断异常性质（错误/真实）
    - [ ] 决定处理方式（删除/修正/保留）
    - [ ] 记录异常值处理决策

  重复值处理:
    - [ ] 识别完全重复记录
    - [ ] 识别关键字段重复
    - [ ] 确定保留规则（最新/最全）
    - [ ] 记录删除数量

  一致性检查:
    - [ ] 数值范围合理性
    - [ ] 日期格式统一
    - [ ] 类别编码一致
    - [ ] 单位标准化
    - [ ] 业务逻辑验证

  格式标准化:
    - [ ] 日期格式（YYYY-MM-DD）
    - [ ] 电话号码格式
    - [ ] 邮箱格式（小写）
    - [ ] 货币数值精度
    - [ ] 文本编码（UTF-8）

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112

# Python EDA单变量分析
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from scipy import stats

class UnivariateAnalysis:
    """单变量探索分析"""

    def __init__(self, df):
        self.df = df

    def analyze_numeric(self, column):
        """数值型变量分析"""
        data = self.df[column].dropna()

        analysis = {
            # 描述性统计
            'descriptive': {
                'count': len(data),
                'mean': data.mean(),
                'median': data.median(),
                'mode': data.mode().values[0] if len(data.mode()) > 0 else None,
                'std': data.std(),
                'variance': data.var(),
                'min': data.min(),
                'max': data.max(),
                'range': data.max() - data.min(),
                'q1': data.quantile(0.25),
                'q3': data.quantile(0.75),
                'iqr': data.quantile(0.75) - data.quantile(0.25),
                'skewness': stats.skew(data),
                'kurtosis': stats.kurtosis(data)
            },

            # 可视化
            'visualizations': self._create_numeric_plots(data, column)
        }

        return analysis

    def analyze_categorical(self, column):
        """分类型变量分析"""
        data = self.df[column].dropna()

        analysis = {
            # 频次统计
            'frequency': data.value_counts().to_dict(),

            # 百分比
            'percentage': (data.value_counts(normalize=True) * 100).round(2).to_dict(),

            # 基础统计
            'stats': {
                'unique': data.nunique(),
                'most_common': data.mode().values[0] if len(data.mode()) > 0 else None,
                'entropy': stats.entropy(data.value_counts())
            },

            # 可视化
            'visualizations': self._create_categorical_plots(data, column)
        }

        return analysis

    def _create_numeric_plots(self, data, column):
        """创建数值型变量可视化"""
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))

        # 直方图
        axes[0, 0].hist(data, bins=30, edgecolor='black')
        axes[0, 0].set_title(f'Distribution of {column}')
        axes[0, 0].set_xlabel(column)
        axes[0, 0].set_ylabel('Frequency')

        # 箱线图
        axes[0, 1].boxplot(data)
        axes[0, 1].set_title(f'Box Plot of {column}')
        axes[0, 1].set_ylabel(column)

        # Q-Q图
        stats.probplot(data, dist="norm", plot=axes[1, 0])
        axes[1, 0].set_title('Q-Q Plot')

        # 密度图
        data.plot.kde(ax=axes[1, 1])
        axes[1, 1].set_title(f'Density Plot of {column}')
        axes[1, 1].set_xlabel(column)

        plt.tight_layout()
        return fig

    def _create_categorical_plots(self, data, column):
        """创建分类型变量可视化"""
        fig, axes = plt.subplots(1, 2, figsize=(12, 5))

        # 条形图
        value_counts = data.value_counts()
        axes[0].bar(range(len(value_counts)), value_counts.values)
        axes[0].set_xticks(range(len(value_counts)))
        axes[0].set_xticklabels(value_counts.index, rotation=45)
        axes[0].set_title(f'Count Plot of {column}')
        axes[0].set_ylabel('Count')

        # 饼图
        axes[1].pie(value_counts.values, labels=value_counts.index,
                   autopct='%1.1f%%')
        axes[1].set_title(f'Percentage of {column}')

        plt.tight_layout()
        return fig

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

# 双变量关系分析
class BivariateAnalysis:
    """双变量关系分析"""

    def numeric_vs_numeric(self, x, y):
        """数值型 vs 数值型"""
        correlation = self.df[[x, y]].corr().iloc[0, 1]

        analysis = {
            'correlation': correlation,
            'correlation_type': self._interpret_correlation(correlation),
            'scatter_plot': self._create_scatter(x, y),
            'trend_line': self._fit_trend_line(x, y)
        }

        return analysis

    def numeric_vs_categorical(self, numeric, categorical):
        """数值型 vs 分类型"""
        analysis = {
            'group_stats': self.df.groupby(categorical)[numeric].agg([
                ('mean', 'mean'),
                ('median', 'median'),
                ('std', 'std'),
                ('count', 'count')
            ]),
            'anova_test': self._perform_anova(numeric, categorical),
            'box_plot': self._create_boxplot(numeric, categorical)
        }

        return analysis

    def categorical_vs_categorical(self, x, y):
        """分类 vs 分类"""
        crosstab = pd.crosstab(self.df[x], self.df[y])

        analysis = {
            'crosstab': crosstab,
            'chi_square_test': self._chi_square_test(x, y),
            'stacked_bar': self._create_stacked_bar(x, y),
            'heatmap': self._create_crosstab_heatmap(crosstab)
        }

        return analysis

    def _interpret_correlation(self, r):
        """解释相关系数"""
        abs_r = abs(r)
        if abs_r >= 0.8:
            strength = "强"
        elif abs_r >= 0.5:
            strength = "中等"
        elif abs_r >= 0.3:
            strength = "弱"
        else:
            strength = "极弱"

        direction = "正" if r > 0 else "负"
        return f"{direction}相关 ({strength})"

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

# 特征工程最佳实践
class FeatureEngineering:
    """特征工程处理"""

    def __init__(self, df):
        self.df = df

    def create_temporal_features(self, date_column):
        """创建时间特征"""
        df = self.df.copy()
        df[date_column] = pd.to_datetime(df[date_column])

        temporal_features = pd.DataFrame({
            f'{date_column}_year': df[date_column].dt.year,
            f'{date_column}_month': df[date_column].dt.month,
            f'{date_column}_day': df[date_column].dt.day,
            f'{date_column}_dayofweek': df[date_column].dt.dayofweek,
            f'{date_column}_dayofyear': df[date_column].dt.dayofyear,
            f'{date_column}_week': df[date_column].dt.isocalendar().week,
            f'{date_column}_quarter': df[date_column].dt.quarter,
            f'{date_column}_is_weekend': df[date_column].dt.dayofweek >= 5,
            f'{date_column}_is_month_start': df[date_column].dt.is_month_start,
            f'{date_column}_is_month_end': df[date_column].dt.is_month_end
        })

        return temporal_features

    def create_aggregation_features(self, group_column, agg_column, aggs):
        """创建聚合特征"""
        # 常用聚合函数
        agg_funcs = {
            'mean': '平均值',
            'median': '中位数',
            'std': '标准差',
            'min': '最小值',
            'max': '最大值',
            'count': '计数',
            'nunique': '唯一值数量'
        }

        features = self.df.groupby(group_column)[agg_column].agg(aggs)

        # 重命名列
        feature_names = [f'{group_column}_{agg}_{agg_column}'
                        for agg in aggs]
        features.columns = feature_names

        return features

    def create_interaction_features(self, features):
        """创建交互特征"""
        from sklearn.preprocessing import PolynomialFeatures

        poly = PolynomialFeatures(
            degree=2,
            interaction_only=True,
            include_bias=False
        )

        interaction_array = poly.fit_transform(
            self.df[features]
        )

        feature_names = poly.get_feature_names_out(features)

        interaction_df = pd.DataFrame(
            interaction_array,
            columns=feature_names
        )

        return interaction_df

    def encode_categorical(self, column, method='target'):
        """分类编码"""
        if method == 'onehot':
            # One-Hot编码
            encoded = pd.get_dummies(self.df[column], prefix=column)
        elif method == 'label':
            # Label编码
            from sklearn.preprocessing import LabelEncoder
            le = LabelEncoder()
            encoded = pd.Series(le.fit_transform(self.df[column]),
                               name=f'{column}_encoded')
        elif method == 'target':
            # Target编码（需要目标变量）
            # 使用目标变量的均值替换类别
            encoded = self._target_encode(column)
        else:
            raise ValueError(f"Unknown encoding method: {method}")

        return encoded

    def create_bin_features(self, column, bins=5, labels=None):
        """分箱特征"""
        binned = pd.cut(
            self.df[column],
            bins=bins,
            labels=labels
        )

        return pd.Series(binned, name=f'{column}_binned')

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104

# 客户细分聚类分析
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import silhouette_score

class CustomerSegmentation:
    """客户细分分析"""

    def __init__(self, data):
        self.data = data
        self.scaler = StandardScaler()

    def prepare_features(self, feature_columns):
        """准备特征"""
        X = self.data[feature_columns].dropna()

        # 标准化
        X_scaled = self.scaler.fit_transform(X)

        return X, X_scaled

    def find_optimal_clusters(self, X, max_k=10):
        """寻找最优聚类数"""
        inertias = []
        silhouette_scores = []
        K_range = range(2, max_k + 1)

        for k in K_range:
            kmeans = KMeans(n_clusters=k, random_state=42, n_init=10)
            kmeans.fit(X)

            inertias.append(kmeans.inertia_)
            silhouette_scores.append(silhouette_score(X, kmeans.labels_))

        # 肘部法则和轮廓系数
        optimal_k_elbow = self._elbow_method(K_range, inertias)
        optimal_k_silhouette = K_range[np.argmax(silhouette_scores)]

        return {
            'elbow_k': optimal_k_elbow,
            'silhouette_k': optimal_k_silhouette,
            'inertias': inertias,
            'silhouette_scores': silhouette_scores
        }

    def perform_clustering(self, X, n_clusters):
        """执行聚类"""
        kmeans = KMeans(
            n_clusters=n_clusters,
            random_state=42,
            n_init=10
        )

        clusters = kmeans.fit_predict(X)

        return {
            'labels': clusters,
            'centroids': kmeans.cluster_centers_,
            'model': kmeans
        }

    def analyze_segments(self, feature_columns, labels):
        """分析细分群体"""
        df_with_clusters = self.data[feature_columns].copy()
        df_with_clusters['cluster'] = labels

        segment_analysis = df_with_clusters.groupby('cluster').agg([
            ('count', 'count'),
            ('mean', 'mean'),
            ('median', 'median'),
            ('std', 'std')
        ])

        return segment_analysis

    def profile_segments(self, n_clusters, feature_columns):
        """细分画像"""
        X, X_scaled = self.prepare_features(feature_columns)
        result = self.perform_clustering(X_scaled, n_clusters)

        df_with_clusters = self.data.copy()
        df_with_clusters['cluster'] = result['labels']

        profiles = {}
        for cluster_id in range(n_clusters):
            cluster_data = df_with_clusters[
                df_with_clusters['cluster'] == cluster_id
            ]

            profile = {
                'size': len(cluster_data),
                'percentage': len(cluster_data) / len(df_with_clusters) * 100,
                'characteristics': {}
            }

            for col in feature_columns:
                profile['characteristics'][col] = {
                    'mean': cluster_data[col].mean(),
                    'median': cluster_data[col].median()
                }

            profiles[f'cluster_{cluster_id}'] = profile

        return profiles

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101

# 转化漏斗分析
class FunnelAnalysis:
    """漏斗分析"""

    def __init__(self, df):
        self.df = df

    def calculate_funnel(self, steps, user_col='user_id',
                        date_col='event_date'):
        """
        计算漏斗各步骤
        steps: 漏斗步骤列表，如 ['visit', 'signup', 'purchase']
        """
        funnel_data = []

        for i, step in enumerate(steps):
            step_users = self.df[
                self.df['event'] == step
            ][user_col].nunique()

            if i == 0:
                conversion_rate = 1.0
            else:
                previous_step_users = funnel_data[i-1]['users']
                conversion_rate = step_users / previous_step_users

            funnel_data.append({
                'step': step,
                'step_number': i + 1,
                'users': step_users,
                'conversion_rate': conversion_rate,
                'overall_conversion': step_users / funnel_data[0]['users']
            })

        return pd.DataFrame(funnel_data)

    def analyze_time_to_convert(self, from_step, to_step,
                                user_col='user_id',
                                time_col='event_time'):
        """分析转化时间"""
        from_events = self.df[self.df['event'] == from_step]
        to_events = self.df[self.df['event'] == to_step]

        # 合并事件
        merged = from_events.merge(
            to_events,
            on=user_col,
            suffixes=('_from', '_to')
        )

        # 计算时间差
        merged['time_diff'] = (
            merged[f'{time_col}_to'] - merged[f'{time_col}_from']
        ).dt.total_seconds() / 3600  # 转换为小时

        analysis = {
            'median_hours': merged['time_diff'].median(),
            'mean_hours': merged['time_diff'].mean(),
            'distribution': merged['time_diff'].describe()
        }

        return analysis

    def cohort_analysis(self, user_col='user_id',
                       event_col='event',
                       date_col='event_date'):
        """同期群分析"""
        # 创建同期群（按首次活动日期）
        df_sorted = self.df.sort_values([user_col, date_col])
        cohort = df_sorted.groupby(user_col)[date_col].first()
        cohort.name = 'cohort_date'

        # 合并同期群信息
        df_with_cohort = self.df.merge(
            cohort,
            on=user_col
        )

        # 计算周期数
        df_with_cohort['period_num'] = (
            df_with_cohort[date_col].dt.to_period('M') -
            df_with_cohort['cohort_date'].dt.to_period('M')
        ).apply(lambda x: x.n)

        # 创建透视表
        cohort_data = df_with_cohort.pivot_table(
            index='cohort_date',
            columns='period_num',
            values=user_col,
            aggfunc='nunique'
        )

        # 计算留存率
        cohort_size = cohort_data.iloc[:, 0]
        retention = cohort_data.divide(cohort_size, axis=0) * 100

        return {
            'cohort_data': cohort_data,
            'retention': retention,
            'cohort_size': cohort_size
        }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

数据可视化设计原则:

  选择正确的图表类型:
    比较:
      - 柱状图 (2-7个类别)
      - 条形图 (长标签名称)
      - 雷达图 (多维比较)

    趋势:
      - 折线图 (时间序列)
      - 面积图 (累积变化)
      - 斜率图 (变化率)

    关系:
      - 散点图 (两个变量)
      - 气泡图 (三个变量)
      - 热力图 (矩阵关系)

    分布:
      - 直方图 (数值分布)
      - 箱线图 (统计分布)
      - 小提琴图 (密度分布)

    组成:
      - 饼图 (部分占比，<5个)
      - 堆叠柱状图 (分类组成)
      - 树状图 (层级结构)

  视觉优化:
    - 使用数据墨水比原则
    - 每个图表传达单一信息
    - 消除图表杂乱
    - 使用对比突出重点
    - 保持设计一致性

  颜色使用:
    - 色盲友好配色
    - 用颜色传达意义
    - 限制调色板数量(3-5色)
    - 避免红色/绿色表示好坏

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100

# 仪表盘配置示例
class DashboardConfig:
    """仪表盘配置"""

    # KPI指标卡片
    kpi_cards = [
        {
            'title': '总收入',
            'value': 1250000,
            'format': 'currency',
            'change': 12.5,
            'change_period': 'MoM',
            'trend': 'up',
            'target': 1000000,
            'progress': 125
        },
        {
            'title': '新增用户',
            'value': 5432,
            'format': 'number',
            'change': 8.3,
            'change_period': 'WoW',
            'trend': 'up',
            'target': 6000,
            'progress': 90
        },
        {
            'title': '转化率',
            'value': 3.8,
            'format': 'percentage',
            'change': -0.5,
            'change_period': 'MoM',
            'trend': 'down',
            'target': 4.0,
            'progress': 95
        },
        {
            'title': '客单价',
            'value': 230,
            'format': 'currency',
            'change': 5.2,
            'change_period': 'YoY',
            'trend': 'up',
            'target': 250,
            'progress': 92
        }
    ]

    # 图表配置
    charts = [
        {
            'type': 'line',
            'title': '收入趋势',
            'data_source': 'revenue_over_time',
            'x_axis': 'date',
            'y_axis': 'revenue',
            'group_by': 'channel',
            'comparison': 'previous_period'
        },
        {
            'type': 'bar',
            'title': '渠道表现',
            'data_source': 'channel_performance',
            'x_axis': 'channel',
            'y_axis': ['impressions', 'clicks', 'conversions'],
            'y_axis_format': 'logarithmic'
        },
        {
            'type': 'funnel',
            'title': '转化漏斗',
            'data_source': 'conversion_funnel',
            'steps': ['visit', 'signup', 'activation', 'purchase']
        },
        {
            'type': 'heatmap',
            'title': '用户活跃热力图',
            'data_source': 'user_activity',
            'x_axis': 'hour',
            'y_axis': 'day_of_week',
            'value': 'activity_count'
        }
    ]

    # 布局配置
    layout = {
        'rows': [
            {
                'height': 150,
                'components': ['kpi_cards'] * 4
            },
            {
                'height': 400,
                'components': ['revenue_trend_chart', 'channel_performance_chart']
            },
            {
                'height': 350,
                'components': ['funnel_chart', 'heatmap_chart']
            }
        ]
    }

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

数据展示技巧:

  数据简化:
    - 一次只传达一个主要观点
    - 去除不必要的数据
    - 使用渐进式披露
    - 提供可选详细信息

  上下文提供:
    - 与目标比较
    - 与历史比较
    - 与同行比较
    - 与行业基准比较

  行动导向:
    - 明确数据意味着什么
    - 提供具体的行动建议
    - 量化预期影响
    - 给出时间表

  可视化技巧:
    - 使用标题说明要点
    - 标注关键数据点
    - 添加趋势线/基准线
    - 使用注释解释异常