1. 数据收集
   ↓
2. 数据预处理（清洗、特征工程）
   ↓
3. 选择模型
   ↓
4. 训练模型（学习参数）
   ↓
5. 模型评估
   ↓
6. 模型优化/调参
   ↓
7. 模型部署应用

from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

# 用户RFM数据
rfm = df[['recency', 'frequency', 'monetary']]

# 标准化
scaler = StandardScaler()
rfm_scaled = scaler.fit_transform(rfm)

# K-Means聚类
kmeans = KMeans(n_clusters=4, random_state=42)
df['cluster'] = kmeans.fit_predict(rfm_scaled)

# 分析每个簇的特征
df.groupby('cluster')[['recency', 'frequency', 'monetary']].mean()

           [年龄 < 30?]
          /            \
        是              否
        /                \
  [收入 < 5万?]      [资产 > 100万?]
    /      \           /        \
  拒绝    通过       拒绝      通过

from sklearn.tree import DecisionTreeClassifier
from sklearn.model_selection import train_test_split
from sklearn import tree
import matplotlib.pyplot as plt

# 准备数据
X = df[['age', 'income', 'balance']]
y = df['churn']

# 划分训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# 训练决策树
clf = DecisionTreeClassifier(
    max_depth=3,  # 最大深度（预剪枝）
    min_samples_split=20,  # 分裂所需最小样本数
    min_samples_leaf=10,  # 叶节点最小样本数
    random_state=42
)
clf.fit(X_train, y_train)

# 预测
y_pred = clf.predict(X_test)

# 可视化决策树
plt.figure(figsize=(12, 8))
tree.plot_tree(clf, feature_names=X.columns, 
               class_names=['Not Churn', 'Churn'],
               filled=True, rounded=True)
plt.show()

# 特征重要性
feature_importance = pd.DataFrame({
    'feature': X.columns,
    'importance': clf.feature_importances_
}).sort_values('importance', ascending=False)

学习曲线分析：

欠拟合：
训练误差 ─────────  高且平
测试误差 ─────────  高且平

良好拟合：
训练误差 ╲        低
测试误差  ─────   低且稳定

过拟合：
训练误差 ╲╲       很低
测试误差  ╱─     较高且不稳定

from sklearn.model_selection import learning_curve
import numpy as np
import matplotlib.pyplot as plt

# 绘制学习曲线
def plot_learning_curve(model, X, y):
    train_sizes, train_scores, test_scores = learning_curve(
        model, X, y, cv=5,
        train_sizes=np.linspace(0.1, 1.0, 10)
    )
    
    train_mean = np.mean(train_scores, axis=1)
    train_std = np.std(train_scores, axis=1)
    test_mean = np.mean(test_scores, axis=1)
    test_std = np.std(test_scores, axis=1)
    
    plt.figure(figsize=(10, 6))
    plt.plot(train_sizes, train_mean, label='Training score')
    plt.plot(train_sizes, test_mean, label='Validation score')
    plt.fill_between(train_sizes, train_mean - train_std,
                     train_mean + train_std, alpha=0.1)
    plt.fill_between(train_sizes, test_mean - test_std,
                     test_mean + test_std, alpha=0.1)
    plt.xlabel('Training Set Size')
    plt.ylabel('Score')
    plt.legend()
    plt.title('Learning Curve')
    plt.show()

# 使用
plot_learning_curve(model, X_train, y_train)

样本A：
簇1: 0 (不属于)
簇2: 1 (属于)
簇3: 0 (不属于)

样本A：
簇1: 0.1 (10%概率属于)
簇2: 0.7 (70%概率属于)
簇3: 0.2 (20%概率属于)

概率和 = 1.0

import skfuzzy as fuzz
import numpy as np

# 数据
data = np.array([[1, 2], [1.5, 1.8], [5, 8], [8, 8], [1, 0.6], [9, 11]])

# Fuzzy C-Means聚类
cntr, u, u0, d, jm, p, fpc = fuzz.cluster.cmeans(
    data.T,  # 转置
    c=2,  # 簇数量
    m=2,  # 模糊度参数
    error=0.005,
    maxiter=1000
)

# u 是隶属度矩阵（每行是一个样本，每列是一个簇）
# u[0, :] 第一个样本对各个簇的隶属度
print("样本1的隶属度:", u[:, 0])  # [0.95, 0.05] 表示95%属于簇1，5%属于簇2

                预测
              正例  负例
实际 正例      TP    FN
    负例      FP    TN

from sklearn.metrics import (
    confusion_matrix, classification_report,
    roc_auc_score, roc_curve
)
import matplotlib.pyplot as plt

# 混淆矩阵
cm = confusion_matrix(y_test, y_pred)
print("混淆矩阵:\n", cm)

# 分类报告（包含precision, recall, f1-score）
print(classification_report(y_test, y_pred))

# AUC
auc = roc_auc_score(y_test, y_pred_proba)
print(f"AUC: {auc:.3f}")

# ROC曲线
fpr, tpr, thresholds = roc_curve(y_test, y_pred_proba)
plt.figure(figsize=(8, 6))
plt.plot(fpr, tpr, label=f'ROC (AUC = {auc:.3f})')
plt.plot([0, 1], [0, 1], 'k--', label='Random')
plt.xlabel('False Positive Rate')
plt.ylabel('True Positive Rate')
plt.title('ROC Curve')
plt.legend()
plt.show()

# 三级标签体系
tags_hierarchy = {
    '用户价值': {
        '高价值用户': ['VIP用户', '超级会员', '企业客户'],
        '中价值用户': ['普通会员', '活跃用户'],
        '低价值用户': ['新用户', '沉睡用户']
    },
    '消费特征': {
        '高频购买': ['每周购买', '月均3单以上'],
        '价格敏感': ['爱用优惠券', '专买促销品']
    }
}

def calculate_user_tags(user_data):
    """计算用户标签"""
    tags = []
    
    # RFM评分
    rfm_score = (user_data['recency_score'] +
                 user_data['frequency_score'] +
                 user_data['monetary_score'])
    
    if rfm_score >= 12:
        tags.append('高价值用户')
    elif rfm_score >= 8:
        tags.append('中价值用户')
    else:
        tags.append('低价值用户')
    
    # 品类偏好
    if user_data['electronics_ratio'] > 0.5:
        tags.append('电子产品爱好者')
    
    # 价格敏感度
    if user_data['avg_discount_usage'] > 0.7:
        tags.append('价格敏感')
    
    return tags

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

def extract_user_data(**context):
    """提取用户数据"""
    date = context['ds']
    
    # 从多个数据源提取
    basic_info = extract_from_crm(date)
    behavior_log = extract_from_clickhouse(date)
    order_data = extract_from_mysql(date)
    
    return {
        'basic': basic_info,
        'behavior': behavior_log,
        'orders': order_data
    }

def transform_user_profile(**context):
    """转换为用户画像"""
    data = context['task_instance'].xcom_pull(task_ids='extract')
    
    profiles = []
    for user_id in data['basic']['user_id'].unique():
        user_behaviors = data['behavior'][
            data['behavior']['user_id'] == user_id
        ]
        user_orders = data['orders'][
            data['orders']['user_id'] == user_id
        ]
        
        profile = {
            'user_id': user_id,
            # 行为特征
            'pv_7d': len(user_behaviors),
            'active_days_30d': user_behaviors['date'].nunique(),
            # 交易特征
            'order_count_30d': len(user_orders),
            'total_amount_30d': user_orders['amount'].sum(),
            # 标签
            'tags': calculate_user_tags(...)
        }
        
        profiles.append(profile)
    
    return profiles

def load_to_database(**context):
    """加载到数据库"""
    profiles = context['task_instance'].xcom_pull(task_ids='transform')
    
    # 存储到MySQL
    pd.DataFrame(profiles).to_sql(
        'user_profiles',
        mysql_engine,
        if_exists='replace'
    )
    
    # 同步到Redis（快速查询）
    for profile in profiles:
        redis_client.hset(
            f"user:{profile['user_id']}",
            mapping=profile
        )

# Airflow DAG
dag = DAG(
    'user_profile_etl',
    default_args={'owner': 'data_team'},
    schedule_interval='@daily',  # 每天执行
    start_date=datetime(2024, 1, 1)
)

extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_user_data,
    dag=dag
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_user_profile,
    dag=dag
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load_to_database,
    dag=dag
)

extract_task >> transform_task >> load_task

# 筛选目标用户发送优惠券
target_users = session.query(UserProfile).filter(
    UserProfile.tags.contains('高价值用户'),
    UserProfile.tags.contains('电子产品爱好者'),
    UserProfile.recency > 30  # 30天未购买
).all()

for user in target_users:
    send_coupon(user.user_id, category='electronics', discount=0.15)

def recommend_for_user(user_id):
    """基于画像推荐商品"""
    profile = get_user_profile(user_id)
    
    # 根据偏好推荐
    if '电子产品爱好者' in profile.tags:
        products = get_new_electronics()
    
    # 根据价格敏感度
    if '价格敏感' in profile.tags:
        products = filter_by_discount(products)
    
    return products

def calculate_sample_size(baseline=0.15, mde=0.05, alpha=0.05, power=0.8):
    """
    计算所需样本量
    baseline: 基线转化率15%
    mde: 最小可检测效应5%
    alpha: 显著性水平
    power: 统计功效
    """
    import scipy.stats as stats
    import numpy as np
    
    effect_size = mde / np.sqrt(baseline * (1 - baseline))
    z_alpha = stats.norm.ppf(1 - alpha/2)
    z_beta = stats.norm.ppf(power)
    
    n = ((z_alpha + z_beta) / effect_size) ** 2
    
    return int(np.ceil(n))

n_per_group = calculate_sample_size()
print(f"每组需要: {n_per_group}个样本")
# 输出: 每组需要: 6200个样本

import hashlib

def assign_group(user_id, experiment_id):
    """确定性分组"""
    hash_value = int(hashlib.md5(
        f"{user_id}_{experiment_id}".encode()
    ).hexdigest(), 16)
    
    return 'treatment' if hash_value % 2 == 0 else 'control'

// 前端埋点
const group = getExperimentGroup(userId, 'detail_page_v2');

// 页面加载
track('page_view', {
    experiment_id: 'detail_page_v2',
    group: group,
    version: group === 'control' ? 'v1' : 'v2'
});

// 加购行为
document.querySelector('.add-cart').onclick = () => {
    track('add_to_cart', {
        experiment_id: 'detail_page_v2',
        group: group
    });
};

def analyze_ab_test(experiment_id):
    """分析AB测试结果"""
    # 提取数据
    df = get_experiment_data(experiment_id)
    
    control = df[df['group'] == 'control']['converted']
    treatment = df[df['group'] == 'treatment']['converted']
    
    # t检验
    from scipy import stats
    t_stat, p_value = stats.ttest_ind(control, treatment)
    
    # 计算提升
    control_rate = control.mean()
    treatment_rate = treatment.mean()
    lift = (treatment_rate - control_rate) / control_rate
    
    print(f"""
    对照组转化率: {control_rate:.2%}
    实验组转化率: {treatment_rate:.2%}
    相对提升: {lift:.2%}
    P值: {p_value:.4f}
    是否显著: {'是' if p_value < 0.05 else '否'}
    """)
    
    # 决策
    if p_value < 0.05 and lift > 0:
        return "建议全量上线"
    else:
        return "建议继续实验或放弃"

from pyspark.sql.functions import rand, concat, lit

# 原代码（倾斜）
result = orders.groupBy('merchant_id').agg(
    F.sum('amount').alias('total')
)

# 加盐打散
salt_num = 100
orders_salted = orders.withColumn(
    'merchant_id_salted',
    concat(col('merchant_id'), lit('_'), (rand() * salt_num).cast('int'))
)

# 两阶段聚合
partial = orders_salted.groupBy('merchant_id_salted').agg(
    F.sum('amount').alias('partial_sum')
)

final = partial.withColumn(
    'merchant_id',
    F.split(col('merchant_id_salted'), '_')[0]
).groupBy('merchant_id').agg(
    F.sum('partial_sum').alias('total')
)

# 识别倾斜Key
skewed_threshold = 10000
counts = orders.groupBy('merchant_id').count()
skewed_ids = counts.filter(col('count') > skewed_threshold)

# 分离处理
normal_orders = orders.filter(~col('merchant_id').isin(skewed_ids))
skewed_orders = orders.filter(col('merchant_id').isin(skewed_ids))

# 正常数据正常处理
normal_result = normal_orders.groupBy('merchant_id').agg(...)

# 倾斜数据加盐处理
skewed_result = process_with_salt(skewed_orders)

# 合并
final = normal_result.union(skewed_result)

# 小表Join大表
from pyspark.sql.functions import broadcast

# 原代码
result = large_orders.join(merchants, 'merchant_id')

# 广播小表
result = large_orders.join(
    broadcast(merchants),  # merchants表小于2GB
    'merchant_id'
)

# 空列表
lst = []
lst = list()

# 带初始值
lst = [1, 2, 3, 4, 5]
lst = list(range(1, 6))  # [1, 2, 3, 4, 5]

# 列表推导式
lst = [x**2 for x in range(5)]  # [0, 1, 4, 9, 16]

lst = [1, 2, 3]

# 添加元素
lst.append(4)        # [1, 2, 3, 4]（末尾添加）
lst.insert(1, 99)    # [1, 99, 2, 3, 4]（指定位置插入）
lst.extend([5, 6])   # [1, 99, 2, 3, 4, 5, 6]（批量添加）
lst += [7, 8]        # 同extend

# 删除元素
lst.remove(99)       # [1, 2, 3, 4, 5, 6, 7, 8]（删除首个值为99的元素）
item = lst.pop()     # 删除并返回最后一个元素
item = lst.pop(0)    # 删除并返回索引0的元素
del lst[0]           # 删除索引0的元素
lst.clear()          # 清空列表

# 修改元素
lst[0] = 100         # 单个修改
lst[1:3] = [200, 300]  # 批量修改

lst = [3, 1, 4, 1, 5, 9, 2, 6]

# 查找
lst.index(4)         # 2（返回首个值为4的索引）
lst.count(1)         # 2（统计1出现的次数）
4 in lst             # True（判断是否存在）

# 排序
lst.sort()           # [1, 1, 2, 3, 4, 5, 6, 9]（原地排序）
lst.sort(reverse=True)  # 降序
sorted_lst = sorted(lst)  # 返回新列表，不改变原列表

# 反转
lst.reverse()        # 原地反转
reversed_lst = lst[::-1]  # 返回新列表

# 其他
len(lst)             # 长度
max(lst)             # 最大值
min(lst)             # 最小值
sum(lst)             # 求和

# 基本形式
[表达式 for 变量 in 可迭代对象]

# 示例
[x**2 for x in range(5)]  # [0, 1, 4, 9, 16]

# 带条件
[x for x in range(10) if x % 2 == 0]  # [0, 2, 4, 6, 8]

# 多重循环
[(x, y) for x in range(3) for y in range(2)]
# [(0, 0), (0, 1), (1, 0), (1, 1), (2, 0), (2, 1)]

# 嵌套列表推导式
matrix = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]
[item for row in matrix for item in row]  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

# 空字典
d = {}
d = dict()

# 带初始值
d = {'name': 'Alice', 'age': 25, 'city': 'Beijing'}

# 使用dict()函数
d = dict(name='Alice', age=25)
d = dict([('name', 'Alice'), ('age', 25)])

# 字典推导式
d = {x: x**2 for x in range(5)}  # {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}

d = {'name': 'Alice', 'age': 25, 'city': 'Beijing'}

# 获取所有键
keys = d.keys()      # dict_keys(['name', 'age', 'city'])
keys_list = list(d.keys())

# 获取所有值
values = d.values()  # dict_values(['Alice', 25, 'Beijing'])

# 获取所有键值对
items = d.items()    # dict_items([('name', 'Alice'), ...])

# 遍历
for key in d:
    print(key, d[key])

for key, value in d.items():
    print(key, value)

# 判断键是否存在
'name' in d          # True
'email' in d         # False

# 复制
d2 = d.copy()        # 浅拷贝

# 基本形式
{key_expr: value_expr for 变量 in 可迭代对象}

# 示例
{x: x**2 for x in range(5)}
# {0: 0, 1: 1, 2: 4, 3: 9, 4: 16}

# 带条件
{x: x**2 for x in range(10) if x % 2 == 0}

# 交换键值
d = {'a': 1, 'b': 2, 'c': 3}
{v: k for k, v in d.items()}  # {1: 'a', 2: 'b', 3: 'c'}

# 从两个列表创建字典
keys = ['name', 'age', 'city']
values = ['Alice', 25, 'Beijing']
d = {k: v for k, v in zip(keys, values)}

# 1. 计数
from collections import Counter
lst = ['a', 'b', 'a', 'c', 'b', 'a']
count = Counter(lst)  # Counter({'a': 3, 'b': 2, 'c': 1})

# 手动实现
count = {}
for item in lst:
    count[item] = count.get(item, 0) + 1

# 2. 分组
from collections import defaultdict
students = [
    {'name': 'Alice', 'class': 'A'},
    {'name': 'Bob', 'class': 'B'},
    {'name': 'Charlie', 'class': 'A'}
]

groups = defaultdict(list)
for student in students:
    groups[student['class']].append(student['name'])
# {'A': ['Alice', 'Charlie'], 'B': ['Bob']}

# 3. 缓存/记忆化
cache = {}
def fib(n):
    if n in cache:
        return cache[n]
    if n <= 1:
        return n
    cache[n] = fib(n-1) + fib(n-2)
    return cache[n]

lambda 参数: 表达式