收藏！GitHub爆火的ML Python工具库大盘点，90%数据科学家都在用

- Python 3.8 或更高版本
- pip 包管理器
- 至少 8GB RAM（训练深度学习模型建议 16GB+）
- 50GB+ 可用磁盘空间

# 首先创建项目目录
mkdir ml-projects
cd ml-projects

# 使用 venv 创建虚拟环境（Python 3.3+ 内置）
python -m venv ml-venv

# 激活虚拟环境
# Linux/Mac:
source ml-venv/bin/activate
# Windows:
ml-venv\Scripts\activate

# 验证激活成功
python --version

# 安装数据处理核心库
pip install numpy pandas matplotlib seaborn

# 安装机器学习框架
pip install scikit-learn

# 安装深度学习框架（按需）
pip install torch torchvision torchaudio
# 或
pip install tensorflow

# 安装Jupyter Notebook用于交互式编程
pip install jupyter notebook

# 常用工具库
pip install tqdm requests beautifulsoup4

# 方式1: 直接访问GitHub仓库
# https://github.com/lukasmasuch/best-of-ml-python

# 方式2: 访问生成的榜单页面
# https://ml-tooling.github.io/best-of-ml-python/

# 方式3: 通过Python代码获取最新数据
import requests
import json

# 获取项目的raw数据
url = "https://raw.githubusercontent.com/lukasmasuch/best-of-ml-python/main/best-of-ml-python.json"
response = requests.get(url)
tools_data = response.json()

print(f"共收录 {len(tools_data)} 个工具")

# 代码格式化工具
pip install black isort

# 类型检查工具
pip install mypy pyright

# 代码质量检查
pip install pylint flake8

# Git hooks（防止提交未格式化的代码）
pip install pre-commit

import pandas as pd
import numpy as np

# 创建DataFrame
data = {
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
    "salary": [50000, 60000, 70000],
    "department": ["Engineering", "Sales", "Engineering"]
}
df = pd.DataFrame(data)

# 数据探索
print(df.head())           # 查看前几行
print(df.describe())       # 数值列统计摘要
print(df.info())           # 数据类型和缺失值信息

# 缺失值处理
df["age"].fillna(df["age"].mean(), inplace=True)

# 分组聚合
dept_stats = df.groupby("department")["salary"].agg(["mean", "max", "min"])
print(dept_stats)

import polars as pl

# Polars的API设计更加函数式
df_pl = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie"],
    "age": [25, 30, 35],
})

# 链式调用是Polars的特色
result = (
    df_pl
    .filter(pl.col("age") > 25)
    .with_columns([
        (pl.col("age") * 2).alias("age_doubled")
    ])
    .select(["name", "age_doubled"])
)
print(result)

from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer

# 定义数值和类别特征
numeric_features = ["age", "salary"]
categorical_features = ["department"]

# 创建预处理器
numeric_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

categorical_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore"))
])

preprocessor = ColumnTransformer(
    transformers=[
        ("num", numeric_transformer, numeric_features),
        ("cat", categorical_transformer, categorical_features)
    ]
)

# 完整的预处理管道
pipeline = Pipeline(steps=[
    ("preprocessor", preprocessor),
])

from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.ensemble import RandomForestClassifier, GradientBoostingRegressor
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

# 准备数据
X = df[["age", "salary"]]
y = (df["department"] == "Engineering").astype(int)

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

# 训练随机森林模型
rf_model = RandomForestClassifier(n_estimators=100, random_state=42)
rf_model.fit(X_train, y_train)

# 预测与评估
y_pred = rf_model.predict(X_test)
print(f"准确率: {accuracy_score(y_test, y_pred):.4f}")
print(f"\n分类报告:\n{classification_report(y_test, y_pred)}")

# 交叉验证评估模型稳定性
cv_scores = cross_val_score(rf_model, X, y, cv=5)
print(f"交叉验证分数: {cv_scores}")
print(f"平均分数: {cv_scores.mean():.4f} ± {cv_scores.std():.4f}")

import xgboost as xgb
import lightgbm as lgb

# XGBoost示例
xgb_model = xgb.XGBClassifier(
    n_estimators=200,
    max_depth=6,
    learning_rate=0.1,
    objective="binary:logistic",
    use_label_encoder=False,
    eval_metric="logloss"
)
xgb_model.fit(X_train, y_train)

# LightGBM示例 - 训练速度更快
lgb_model = lgb.LGBMClassifier(
    n_estimators=200,
    max_depth=6,
    learning_rate=0.1,
    verbose=-1  # 关闭日志输出
)
lgb_model.fit(X_train, y_train)

# 特征重要性比较
print("XGBoost特征重要性:", xgb_model.feature_importances_)
print("LightGBM特征重要性:", lgb_model.feature_importances_)

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

# 检查GPU是否可用
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"使用设备: {device}")

# 定义一个简单的神经网络
class SimpleNN(nn.Module):
    def __init__(self, input_size, hidden_size, num_classes):
        super(SimpleNN, self).__init__()
        self.layer1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(hidden_size, hidden_size)
        self.layer3 = nn.Linear(hidden_size, num_classes)

    def forward(self, x):
        x = self.layer1(x)
        x = self.relu(x)
        x = self.layer2(x)
        x = self.relu(x)
        x = self.layer3(x)
        return x

# 准备数据
X_tensor = torch.FloatTensor(X.values)
y_tensor = torch.LongTensor(y.values)

dataset = TensorDataset(X_tensor, y_tensor)
dataloader = DataLoader(dataset, batch_size=16, shuffle=True)

# 初始化模型
model = SimpleNN(input_size=2, hidden_size=32, num_classes=2).to(device)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 训练循环
model.train()
for epoch in range(100):
    total_loss = 0
    for batch_X, batch_y in dataloader:
        batch_X, batch_y = batch_X.to(device), batch_y.to(device)

        # 前向传播
        outputs = model(batch_X)
        loss = criterion(outputs, batch_y)

        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        total_loss += loss.item()

    if (epoch + 1) % 20 == 0:
        print(f"Epoch [{epoch+1}/100], Loss: {total_loss/len(dataloader):.4f}")

# 模型评估
model.eval()
with torch.no_grad():
    X_test_tensor = torch.FloatTensor(X_test.values).to(device)
    outputs = model(X_test_tensor)
    _, predicted = torch.max(outputs, 1)
    accuracy = (predicted.cpu().numpy() == y_test.values).mean()
    print(f"测试集准确率: {accuracy:.4f}")

from transformers import pipeline, AutoTokenizer, AutoModelForSequenceClassification

# 使用预训练模型进行情感分析
classifier = pipeline("sentiment-analysis")
result = classifier("I love using PyTorch for deep learning!")
print(result)

# 加载中文预训练模型
model_name = "bert-base-chinese"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForSequenceClassification.from_pretrained(model_name)

# 文本分类示例
text = "这个产品非常好用，推荐购买"
inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True)
outputs = model(**inputs)
predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
print(f"预测概率: {predictions}")

import shap
import lime
import lime.lime_tabular

# 继续使用之前的随机森林模型
explainer = shap.TreeExplainer(rf_model)

# 计算SHAP值
X_sample = X_test.iloc[:10]
shap_values = explainer.shap_values(X_sample)

# 可视化第一个样本的SHAP值
shap.initjs()
shap.force_plot(
    explainer.expected_value[1],
    shap_values[1][0],
    X_sample.iloc[0],
    matplotlib=True
)

# 使用LIME进行局部解释
explainer_lime = lime.lime_tabular.LimeTabularExplainer(
    training_data=X.values,
    feature_names=X.columns.tolist(),
    class_names=["Non-Engineering", "Engineering"],
    mode="classification"
)

# 解释单个预测
exp = explainer_lime.explain_instance(
    X_test.iloc[0].values,
    rf_model.predict_proba,
    num_features=2
)
exp.show_in_notebook()

# gradio_app.py
import gradio as gr
import numpy as np

def predict_sentiment(text, model_type):
    """模拟情感分析预测"""
    # 实际项目中这里会调用真实的模型
    positive_prob = hash(text) % 100 / 100
    return {
        "positive": positive_prob,
        "negative": 1 - positive_prob
    }

# 创建Gradio界面
interface = gr.Interface(
    fn=predict_sentiment,
    inputs=[
        gr.Textbox(label="输入文本", placeholder="请输入待分析的文本..."),
        gr.Radio(["BERT", "RoBERTa", "XLNet"], label="选择模型")
    ],
    outputs=gr.Label(num_top_classes=2),
    title="情感分析演示",
    description="体验不同预训练模型的情感分析效果",
    examples=[
        ["这个产品太棒了，我非常满意！", "BERT"],
        ["服务态度很差，不会再购买了。", "RoBERTa"]
    ]
)

# 启动服务
interface.launch(server_name="0.0.0.0", server_port=7860)

# streamlit_app.py
import streamlit as st
import pandas as pd
import numpy as np

st.set_page_config(page_title="机器学习仪表板", layout="wide")

st.title("📊 机器学习项目仪表板")

# 侧边栏配置
st.sidebar.header("配置选项")
model_choice = st.sidebar.selectbox("选择模型", ["Random Forest", "XGBoost", "LightGBM"])
threshold = st.sidebar.slider("预测阈值", 0.0, 1.0, 0.5)

# 主内容区
col1, col2 = st.columns(2)

with col1:
    st.subheader("模型性能")
    metrics = {"准确率": 0.92, "精确率": 0.89, "召回率": 0.91}
    st.bar_chart(metrics)

with col2:
    st.subheader("预测结果")
    if st.button("运行预测"):
        st.success("预测完成！")
        st.dataframe(pd.DataFrame({
            "样本": range(1, 6),
            "预测": np.random.choice([0, 1], 5),
            "概率": np.random.rand(5)
        }))

# 运行命令: streamlit run streamlit_app.py

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path

# 设置中文显示
plt.rcParams["font.sans-serif"] = ["SimHei", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False

# 加载数据（使用sklearn内置数据集模拟）
from sklearn.datasets import fetch_openml

# 尝试加载真实数据集
try:
    # churn_data = fetch_openml(name=" churn", version=1, as_frame=True)
    # df = churn_data.frame

    # 为了演示，我们创建模拟数据
    np.random.seed(42)
    n_samples = 1000

    df = pd.DataFrame({
        "customer_id": range(1, n_samples + 1),
        "tenure_months": np.random.randint(1, 72, n_samples),
        "monthly_charges": np.random.uniform(30, 120, n_samples),
        "total_charges": np.random.uniform(100, 8000, n_samples),
        "contract_type": np.random.choice(["Month-to-month", "One year", "Two year"], n_samples),
        "payment_method": np.random.choice(["Electronic check", "Mailed check", 
                                            "Bank transfer", "Credit card"], n_samples),
        "num_support_tickets": np.random.poisson(1, n_samples),
        "num_products": np.random.randint(1, 5, n_samples),
        "has_partner": np.random.choice([0, 1], n_samples),
        "has_dependents": np.random.choice([0, 1], n_samples),
        "internet_service": np.random.choice(["DSL", "Fiber optic", "No"], n_samples),
    })

    # 生成流失标签（与特征有一定关联）
    df["churned"] = (
        (df["contract_type"] == "Month-to-month") & 
        (df["num_support_tickets"] > 2) & 
        (df["tenure_months"] < 12)
    ).astype(int)

    print("数据集加载成功！")
    print(f"样本数: {len(df)}")
    print(f"特征数: {len(df.columns) - 2}")  # 减去customer_id和churned

except Exception as e:
    print(f"数据加载失败: {e}")
    raise

# 数据探索
print("\n" + "=" * 60)
print("数据集基本信息")
print("=" * 60)
print(df.info())

print("\n" + "=" * 60)
print("数值特征统计")
print("=" * 60)
print(df.describe())

print("\n" + "=" * 60)
print("流失率分析")
print("=" * 60)
churn_rate = df["churned"].mean()
print(f"总体流失率: {churn_rate:.2%}")
print(f"\n按合同类型:")
print(df.groupby("contract_type")["churned"].agg(["mean", "count"]))

# 可视化分析
fig, axes = plt.subplots(2, 2, figsize=(14, 10))

# 流失率与合同类型
ax1 = axes[0, 0]
contract_churn = df.groupby("contract_type")["churned"].mean().sort_values(ascending=False)
contract_churn.plot(kind="bar", ax=ax1, color=["#ff6b6b" if x > 0.3 else "#4ecdc4" for x in contract_churn])
ax1.set_title("流失率 vs 合同类型", fontsize=12)
ax1.set_ylabel("流失率")
ax1.tick_params(axis="x", rotation=45)

# 月费与流失关系
ax2 = axes[0, 1]
df[df["churned"] == 0]["monthly_charges"].hist(alpha=0.7, label="未流失", bins=30, ax=ax2)
df[df["churned"] == 1]["monthly_charges"].hist(alpha=0.7, label="已流失", bins=30, ax=ax2)
ax2.set_title("月费分布与流失关系", fontsize=12)
ax2.legend()

# 流失率与在网时长
ax3 = axes[1, 0]
df["tenure_bin"] = pd.cut(df["tenure_months"], bins=[0, 12, 24, 48, 72], labels=["0-12月", "12-24月", "24-48月", "48-72月"])
tenure_churn = df.groupby("tenure_bin")["churned"].mean()
tenure_churn.plot(kind="line", ax=ax3, marker="o", linewidth=2)
ax3.set_title("流失率 vs 在网时长", fontsize=12)
ax3.set_ylabel("流失率")

# 流失率与工单数量
ax4 = axes[1, 1]
ticket_churn = df.groupby("num_support_tickets")["churned"].mean()
ticket_churn.plot(kind="bar", ax=ax4, color="steelblue")
ax4.set_title("流失率 vs 支持工单数", fontsize=12)
ax4.set_xlabel("工单数量")
ax4.set_ylabel("流失率")

plt.tight_layout()
plt.savefig("churn_analysis.png", dpi=150)
plt.show()

print("\n分析图表已保存至 churn_analysis.png")

from sklearn.preprocessing import StandardScaler, LabelEncoder, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer

# 删除临时分组列
df = df.drop("tenure_bin", axis=1)

# 分离特征和目标变量
X = df.drop(["customer_id", "churned"], axis=1)
y = df["churned"]

# 定义特征类型
numeric_features = ["tenure_months", "monthly_charges", "total_charges", 
                   "num_support_tickets", "num_products"]

categorical_features = ["contract_type", "payment_method", "internet_service",
                       "has_partner", "has_dependents"]

print("=" * 60)
print("特征工程")
print("=" * 60)

# 创建预处理管道
numeric_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

categorical_transformer = Pipeline(steps=[
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
])

preprocessor = ColumnTransformer(
    transformers=[
        ("num", numeric_transformer, numeric_features),
        ("cat", categorical_transformer, categorical_features)
    ]
)

# 特征工程：添加衍生特征
def add_derived_features(X):
    """添加有业务意义的衍生特征"""
    X_new = X.copy()

    # 月均消费
    X_new["avg_monthly_charge"] = X["total_charges"] / (X["tenure_months"] + 1)

    # 消费波动（模拟）
    X_new["charge_volatility"] = X["monthly_charges"] * 0.1

    # 客户价值评分
    X_new["customer_value"] = (
        X["total_charges"] * 0.5 + 
        X["tenure_months"] * 10 + 
        X["num_products"] * 100
    )

    # 是否高风险客户
    X_new["high_risk_indicator"] = (
        (X["num_support_tickets"] > 2) & 
        (X["tenure_months"] < 12)
    ).astype(int)

    return X_new

X_enhanced = add_derived_features(X)
print(f"原始特征数: {X.shape[1]}")
print(f"增强后特征数: {X_enhanced.shape[1]}")
print(f"新增特征: avg_monthly_charge, charge_volatility, customer_value, high_risk_indicator")

# 更新特征列表
numeric_features_extended = numeric_features + ["avg_monthly_charge", "charge_volatility", "customer_value"]
categorical_features_extended = categorical_features + ["high_risk_indicator"]

# 创建完整的预处理管道
preprocessor_extended = ColumnTransformer(
    transformers=[
        ("num", Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler())
        ]), numeric_features_extended),
        ("cat", Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("onehot", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
        ]), categorical_features_extended)
    ]
)

print("\n预处理管道创建完成！")

from sklearn.model_selection import train_test_split, cross_val_score, StratifiedKFold
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (accuracy_score, precision_score, recall_score, 
                            f1_score, roc_auc_score, confusion_matrix,
                            classification_report, roc_curve)
import xgboost as xgb
import lightgbm as lgb

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

print("=" * 60)
print("模型训练与评估")
print("=" * 60)
print(f"训练集大小: {len(X_train)}")
print(f"测试集大小: {len(X_test)}")
print(f"训练集流失率: {y_train.mean():.2%}")
print(f"测试集流失率: {y_test.mean():.2%}")

# 定义多个模型进行比较
models = {
    "Logistic Regression": LogisticRegression(max_iter=1000, random_state=42),
    "Random Forest": RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=-1),
    "Gradient Boosting": GradientBoostingClassifier(n_estimators=100, random_state=42),
    "XGBoost": xgb.XGBClassifier(
        n_estimators=100,
        max_depth=6,
        learning_rate=0.1,
        use_label_encoder=False,
        eval_metric="logloss",
        random_state=42
    ),
    "LightGBM": lgb.LGBMClassifier(
        n_estimators=100,
        max_depth=6,
        learning_rate=0.1,
        verbose=-1,
        random_state=42
    )
}

# 训练和评估每个模型
results = {}
roc_curves = {}

for name, model in models.items():
    print(f"\n训练 {name}...")

    # 创建完整管道
    pipeline = Pipeline([
        ("preprocessor", preprocessor_extended),
        ("classifier", model)
    ])

    # 训练模型
    pipeline.fit(X_train, y_train)

    # 预测
    y_pred = pipeline.predict(X_test)
    y_pred_proba = pipeline.predict_proba(X_test)[:, 1]

    # 计算各项指标
    results[name] = {
        "accuracy": accuracy_score(y_test, y_pred),
        "precision": precision_score(y_test, y_pred),
        "recall": recall_score(y_test, y_pred),
        "f1": f1_score(y_test, y_pred),
        "roc_auc": roc_auc_score(y_test, y_pred_proba)
    }

    # 保存ROC曲线数据
    fpr, tpr, _ = roc_curve(y_test, y_pred_proba)
    roc_curves[name] = {"fpr": fpr, "tpr": tpr, "auc": results[name]["roc_auc"]}

    print(f"  准确率: {results[name]['accuracy']:.4f}")
    print(f"  精确率: {results[name]['precision']:.4f}")
    print(f"  召回率: {results[name]['recall']:.4f}")
    print(f"  F1分数: {results[name]['f1']:.4f}")
    print(f"  ROC-AUC: {results[name]['roc_auc']:.4f}")

# 结果汇总
print("\n" + "=" * 60)
print("模型性能对比")
print("=" * 60)

results_df = pd.DataFrame(results).T
results_df = results_df.round(4)
print(results_df.to_string())

# 找出最佳模型
best_model_name = results_df["f1"].idxmax()
print(f"\n最佳模型（F1分数）: {best_model_name}")
print(f"F1分数: {results_df.loc[best_model_name, 'f1']:.4f}")

import shap

print("=" * 60)
print("模型解释性分析")
print("=" * 60)

# 选择LightGBM作为解释对象（训练速度快且效果好）
best_pipeline = Pipeline([
    ("preprocessor", preprocessor_extended),
    ("classifier", lgb.LGBMClassifier(
        n_estimators=100,
        max_depth=6,
        learning_rate=0.1,
        verbose=-1,
        random_state=42
    ))
])

best_pipeline.fit(X_train, y_train)

# 获取分类器
lgbm_model = best_pipeline.named_steps["classifier"]

# 获取预处理后的特征名
preprocessor_fitted = best_pipeline.named_steps["preprocessor"]
numeric_feature_names = numeric_features_extended
categorical_feature_names = list(
    preprocessor_fitted.named_transformers_["cat"]
    .named_steps["onehot"]
    .get_feature_names_out(categorical_features_extended)
)
all_feature_names = numeric_feature_names + categorical_feature_names

print(f"特征总数: {len(all_feature_names)}")

# 计算SHAP值
print("\n计算SHAP值...")
explainer = shap.TreeExplainer(lgbm_model)

# 对测试集的一个子集计算SHAP值
X_test_transformed = preprocessor_fitted.transform(X_test)
X_test_sample = X_test_transformed[:100]

shap_values = explainer.shap_values(X_test_sample)

# 如果是二分类，取正类的SHAP值
if isinstance(shap_values, list):
    shap_values_plot = shap_values[1]
else:
    shap_values_plot = shap_values

# SHAP Summary Plot
print("\n生成SHAP Summary Plot...")
fig, ax = plt.subplots(figsize=(12, 8))
shap.summary_plot(shap_values_plot, X_test_sample, feature_names=all_feature_names, show=False)
plt.title("特征重要性 - SHAP Summary Plot", fontsize=14)
plt.tight_layout()
plt.savefig("shap_summary.png", dpi=150)
plt.show()

# SHAP特征重要性
print("\n生成SHAP特征重要性图...")
shap_mean = np.abs(shap_values_plot).mean(axis=0)
shap_importance = pd.DataFrame({
    "feature": all_feature_names,
    "importance": shap_mean
}).sort_values("importance", ascending=False)

print("Top 10 重要特征:")
print(shap_importance.head(10).to_string(index=False))

fig, ax = plt.subplots(figsize=(10, 6))
top_n = 15
shap_importance.head(top_n).plot(
    kind="barh", x="feature", y="importance", ax=ax, legend=False, color="steelblue"
)
ax.set_title(f"Top {top_n} SHAP特征重要性", fontsize=14)
ax.set_xlabel("平均 |SHAP值|")
plt.tight_layout()
plt.savefig("shap_importance.png", dpi=150)
plt.show()

# 单个预测的解释
print("\n单个客户流失预测解释...")
sample_idx = 0
sample_data = X_test.iloc[[sample_idx]]
sample_transformed = preprocessor_fitted.transform(sample_data)

expected_value = explainer.expected_value
if isinstance(expected_value, list):
    expected_value = expected_value[1]

print(f"\n客户ID: {X_test.iloc[sample_idx]['customer_id'] if 'customer_id' in X_test.columns else sample_idx}")
print(f"预测流失概率: {lgbm_model.predict_proba(sample_transformed)[0][1]:.4f}")
print(f"实际标签: {y_test.iloc[sample_idx]}")

# Force Plot
print("\n生成Force Plot...")
shap.initjs()
fig = shap.force_plot(
    expected_value,
    shap_values_plot[sample_idx],
    sample_transformed[0],
    feature_names=all_feature_names,
    matplotlib=True,
    show=False
)
plt.savefig("shap_force_plot.png", dpi=150, bbox_inches="tight")
plt.show()

# deploy_api.py
import gradio as gr
import joblib
import pandas as pd
import numpy as np

# 假设模型已经保存
# joblib.dump(best_pipeline, "churn_model.pkl")
# model = joblib.load("churn_model.pkl")

# 为了演示，创建一个模拟预测函数
def predict_churn(tenure_months, monthly_charges, total_charges,
                  contract_type, payment_method, num_support_tickets,
                  num_products, has_partner, has_dependents, internet_service):
    """
    预测客户流失概率

    参数:
    - tenure_months: 在网月数 (1-72)
    - monthly_charges: 月消费 (30-120)
    - total_charges: 累计消费 (100-8000)
    - contract_type: 合同类型
    - payment_method: 支付方式
    - num_support_tickets: 支持工单数 (0-10)
    - num_products: 使用产品数 (1-4)
    - has_partner: 是否有伴侣 (0/1)
    - has_dependents: 是否有家属 (0/1)
    - internet_service: 互联网服务类型

    返回:
    - 流失概率和建议
    """

    # 模拟风险评分
    risk_score = 0.0

    # 在网时长越短风险越高
    if tenure_months < 12:
        risk_score += 0.3
    elif tenure_months < 24:
        risk_score += 0.15

    # 月消费越高风险越高
    if monthly_charges > 80:
        risk_score += 0.2

    # 工单数越多风险越高
    risk_score += min(num_support_tickets * 0.1, 0.3)

    # 月付合同风险最高
    if contract_type == "Month-to-month":
        risk_score += 0.25

    # 无互联网服务风险较低
    if internet_service == "No":
        risk_score -= 0.1

    # 有伴侣风险较低
    if has_partner == "Yes":
        risk_score -= 0.1

    # 限制在0-1范围内
    risk_score = max(0, min(1, risk_score))

    # 生成建议
    if risk_score > 0.6:
        recommendation = "⚠️ 高风险客户，建议主动联系挽留"
        color = "red"
    elif risk_score > 0.4:
        recommendation = "⚡ 中等风险，建议加强客户关怀"
        color = "orange"
    else:
        recommendation = "✅ 低风险客户，维持现状即可"
        color = "green"

    # 关键流失因素
    factors = []
    if tenure_months < 12:
        factors.append("在网时间较短")
    if num_support_tickets > 2:
        factors.append("支持工单较多")
    if contract_type == "Month-to-month":
        factors.append("采用月付合同")
    if monthly_charges > 80:
        factors.append("月消费较高")

    result = {
        "流失概率": f"{risk_score:.1%}",
        "风险等级": "高" if risk_score > 0.6 else "中" if risk_score > 0.4 else "低",
        "建议": recommendation,
        "关键风险因素": ", ".join(factors) if factors else "无明显风险因素"
    }

    return result

# 创建Gradio界面
interface = gr.Interface(
    fn=predict_churn,
    title="📱 客户流失预测系统",
    description="基于机器学习的客户流失风险评估工具，帮助企业识别高风险客户并采取针对性挽留措施",
    inputs=[
        gr.Slider(minimum=1, maximum=72, value=24, step=1, label="在网月数", info="客户使用服务的月数"),
        gr.Slider(minimum=30, maximum=120, value=70, step=1, label="月消费(元)", info="每月平均消费金额"),
        gr.Slider(minimum=100, maximum=8000, value=2000, step=100, label="累计消费(元)", info="历史累计消费"),
        gr.Radio(["Month-to-month", "One year", "Two year"], value="Month-to-month", 
                 label="合同类型", info="选择合同期限"),
        gr.Radio(["Electronic check", "Mailed check", "Bank transfer", "Credit card"],
                 value="Electronic check", label="支付方式"),
        gr.Slider(minimum=0, maximum=10, value=1, step=1, label="支持工单数", info="提交的支持工单数量"),
        gr.Slider(minimum=1, maximum=4, value=2, step=1, label="使用产品数", info="购买的产品数量"),
        gr.Radio(["Yes", "No"], value="No", label="是否有伴侣"),
        gr.Radio(["Yes", "No"], value="No", label="是否有家属"),
        gr.Radio(["DSL", "Fiber optic", "No"], value="DSL", label="互联网服务类型"),
    ],
    outputs=gr.Label(num_top_classes=3),
    examples=[
        [6, 95, 570, "Month-to-month", "Electronic check", 4, 1, "No", "No", "Fiber optic"],
        [48, 65, 3120, "Two year", "Credit card", 0, 3, "Yes", "Yes", "DSL"],
        [18, 85, 1530, "Month-to-month", "Bank transfer", 2, 2, "No", "No", "Fiber optic"],
    ],
    theme="default"
)

# 启动服务
if __name__ == "__main__":
    print("启动客户流失预测服务...")
    print("访问地址: http://localhost:7860")
    interface.launch(server_name="0.0.0.0", server_port=7860)

# 快速原型模板
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import joblib

def quick_prototype(data_path, target_column):
    """
    快速构建机器学习原型

    适用于：概念验证、比赛baseline、学术实验
    """
    # 加载数据
    df = pd.read_csv(data_path)
    X = df.drop(target_column, axis=1)
    y = df[target_column]

    # 快速预处理 + 模型管道
    pipeline = Pipeline([
        ("scaler", StandardScaler()),
        ("classifier", RandomForestClassifier(n_estimators=100, random_state=42))
    ])

    # 划分数据集
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

    # 训练并评估
    pipeline.fit(X_train, y_train)
    score = pipeline.score(X_test, y_test)

    print(f"原型模型准确率: {score:.4f}")

    return pipeline

# 使用示例
# model = quick_prototype("your_data.csv", "target")

# production_template.py
import numpy as np
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.model_selection import cross_val_score, StratifiedKFold
import logging
from datetime import datetime
import json

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("model_training.log"),
        logging.StreamHandler()
    ]
)

class DataValidator(BaseEstimator, TransformerMixin):
    """数据验证器"""

    def __init__(self, expected_columns, null_threshold=0.1):
        self.expected_columns = expected_columns
        self.null_threshold = null_threshold

    def fit(self, X, y=None):
        # 验证列名
        missing_cols = set(self.expected_columns) - set(X.columns)
        if missing_cols:
            raise ValueError(f"缺少必需列: {missing_cols}")

        # 验证缺失值
        null_ratio = X.isnull().sum().sum() / (X.shape[0] * X.shape[1])
        if null_ratio > self.null_threshold:
            logging.warning(f"缺失值比例较高: {null_ratio:.2%}")

        return self

    def transform(self, X):
        return X[self.expected_columns].copy()


class ModelMonitor:
    """模型性能监控"""

    def __init__(self, model_name):
        self.model_name = model_name
        self.metrics_history = []

    def log_metrics(self, metrics, metadata=None):
        """记录模型指标"""
        record = {
            "timestamp": datetime.now().isoformat(),
            "model": self.model_name,
            "metrics": metrics,
            "metadata": metadata or {}
        }
        self.metrics_history.append(record)

        # 保存到文件
        with open(f"{self.model_name}_metrics.json", "a") as f:
            f.write(json.dumps(record) + "\n")

        logging.info(f"记录指标: {metrics}")

    def get_drift_report(self):
        """生成数据漂移报告"""
        if len(self.metrics_history) < 2:
            return "数据不足，无法生成漂移报告"

        # 简化示例
        recent = self.metrics_history[-1]["metrics"]
        baseline = self.metrics_history[0]["metrics"]

        drift = {k: recent[k] - baseline.get(k, 0) for k in recent.keys()}

        return pd.DataFrame([drift]).T.rename(columns={0: "变化量"})


def train_production_model(X, y, model, cv_folds=5):
    """
    生产级模型训练流程
    """
    monitor = ModelMonitor(type(model).__name__)

    # 交叉验证评估
    skf = StratifiedKFold(n_splits=cv_folds, shuffle=True, random_state=42)
    cv_scores = cross_val_score(model, X, y, cv=skf, scoring="f1")

    metrics = {
        "cv_mean": cv_scores.mean(),
        "cv_std": cv_scores.std(),
        "cv_scores": cv_scores.tolist()
    }

    monitor.log_metrics(metrics, {"dataset_size": len(X)})

    # 训练最终模型
    model.fit(X, y)
    monitor.log_metrics({"status": "training_completed"})

    return model, monitor


# 使用示例
# from sklearn.ensemble import GradientBoostingClassifier
# model = GradientBoostingClassifier(n_estimators=100)
# trained_model, monitor = train_production_model(X_train, y_train, model)

# automl_example.py
from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
import xgboost as xgb
import numpy as np

def automl_search(X_train, y_train, time_budget_minutes=10):
    """
    简化的AutoML搜索流程

    在实际项目中，可以使用:
    - auto-sklearn
    - FLAML
    - Optuna
    - Hyperopt
    """

    # 定义搜索空间
    search_space = [
        {
            "name": "LogisticRegression",
            "model": LogisticRegression(max_iter=1000),
            "params": {
                "classifier__C": [0.01, 0.1, 1, 10],
                "classifier__penalty": ["l1", "l2"],
            }
        },
        {
            "name": "RandomForest",
            "model": RandomForestClassifier(random_state=42),
            "params": {
                "classifier__n_estimators": [50, 100, 200],
                "classifier__max_depth": [5, 10, 15, None],
                "classifier__min_samples_split": [2, 5, 10],
            }
        },
        {
            "name": "XGBoost",
            "model": xgb.XGBClassifier(
                use_label_encoder=False,
                eval_metric="logloss",
                random_state=42
            ),
            "params": {
                "classifier__n_estimators": [50, 100, 200],
                "classifier__max_depth": [3, 5, 7],
                "classifier__learning_rate": [0.01, 0.1, 0.2],
            }
        },
    ]

    results = []

    for config in search_space:
        print(f"搜索 {config['name']}...")

        pipeline = Pipeline([
            ("scaler", StandardScaler()),
            ("classifier", config["model"])
        ])

        # 使用随机搜索（更快）
        search = RandomizedSearchCV(
            pipeline,
            param_distributions=config["params"],
            n_iter=10,
            cv=3,
            scoring="f1",
            random_state=42,
            n_jobs=-1
        )

        search.fit(X_train, y_train)

        results.append({
            "model": config["name"],
            "best_score": search.best_score_,
            "best_params": search.best_params_,
            "best_estimator": search.best_estimator_
        })

        print(f"  最佳分数: {search.best_score_:.4f}")

    # 返回最佳模型
    best_result = max(results, key=lambda x: x["best_score"])
    print(f"\n最佳模型: {best_result['model']}")
    print(f"最佳分数: {best_result['best_score']:.4f}")

    return best_result


# 使用Optuna进行更高级的超参数优化
try:
    import optuna
    optuna.logging.set_verbosity(optuna.logging.WARNING)

    def objective(trial, X, y):
        """Optuna优化目标函数"""
        classifier_name = trial.suggest_categorical("classifier", 
                                                     ["RandomForest", "XGBoost", "LogisticRegression"])

        if classifier_name == "RandomForest":
            params = {
                "n_estimators": trial.suggest_int("rf_n_estimators", 50, 200),
                "max_depth": trial.suggest_int("rf_max_depth", 3, 15),
                "min_samples_split": trial.suggest_int("rf_min_samples_split", 2, 10),
            }
            model = RandomForestClassifier(**params, random_state=42)
        elif classifier_name == "XGBoost":
            params = {
                "n_estimators": trial.suggest_int("xgb_n_estimators", 50, 200),
                "max_depth": trial.suggest_int("xgb_max_depth", 3, 10),
                "learning_rate": trial.suggest_float("xgb_learning_rate", 0.01, 0.3, log=True),
            }
            model = xgb.XGBClassifier(**params, use_label_encoder=False, 
                                      eval_metric="logloss", random_state=42)
        else:
            params = {
                "C": trial.suggest_float("lr_C", 0.01, 10, log=True),
                "penalty": trial.suggest_categorical("lr_penalty", ["l1", "l2"]),
            }
            model = LogisticRegression(**params, max_iter=1000)

        # 交叉验证评分
        from sklearn.model_selection import cross_val_score
        scores = cross_val_score(model, X, y, cv=3, scoring="f1")

        return scores.mean()

    def run_optuna_study(X, y, n_trials=50):
        """运行Optuna优化研究"""
        study = optuna.create_study(direction="maximize")
        study.optimize(lambda trial: objective(trial, X, y), n_trials=n_trials)

        print(f"最佳F1分数: {study.best_value:.4f}")
        print(f"最佳参数: {study.best_params}")

        return study.best_params

except ImportError:
    print("Optuna未安装，跳过高级优化示例")
    print("安装命令: pip install optuna")

# project_structure.py
"""
推荐的机器学习项目结构

project/
├── data/
│   ├── raw/           # 原始数据，不做修改
│   ├── processed/     # 处理后的数据
│   └── external/      # 外部数据源
├── models/            # 保存训练好的模型
├── notebooks/         # Jupyter notebooks
├── scripts/           # 可执行脚本
├── src/               # 源代码
│   ├── __init__.py
│   ├── data/          # 数据处理模块
│   ├── features/      # 特征工程
│   ├── models/        # 模型定义
│   └── evaluation/    # 评估指标
├── tests/             # 单元测试
├── configs/           # 配置文件
├── requirements.txt
└── README.md
"""

# 配置管理示例
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class ModelConfig:
    """模型配置类"""
    model_type: str = "xgboost"
    n_estimators: int = 100
    max_depth: int = 6
    learning_rate: float = 0.1
    early_stopping_rounds: Optional[int] = 10

@dataclass
class DataConfig:
    """数据配置类"""
    train_path: str = "data/processed/train.csv"
    test_path: str = "data/processed/test.csv"
    target_column: str = "target"
    test_size: float = 0.2
    random_state: int = 42

# 使用示例
model_config = ModelConfig(n_estimators=200, max_depth=8)
data_config = DataConfig()

# data_processing_tips.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import KBinsDiscretizer

# 技巧1: 链式方法调用让代码更清晰
df = (
    pd.read_csv("data.csv")
    .dropna(thresh=len(df.columns) * 0.5)  # 删除缺失率超过50%的列
    .assign(
        new_feature=lambda x: x["a"] / (x["b"] + 1),  # 创建新特征
        category=lambda x: pd.cut(x["value"], bins=5, labels=["A", "B", "C", "D", "E"])
    )
    .query("new_feature > 0")  # 过滤行
    .reset_index(drop=True)
)

# 技巧2: 高效处理缺失值
def smart_impute(df, strategy="auto"):
    """
    根据数据类型自动选择填充策略
    """
    df_filled = df.copy()

    # 数值列用中位数填充
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    for col in numeric_cols:
        if df[col].isnull().sum() > 0:
            if strategy == "auto":
                # 根据分布选择
                if df[col].skew() > 1:
                    df_filled[col].fillna(df[col].median(), inplace=True)
                else:
                    df_filled[col].fillna(df[col].mean(), inplace=True)
            else:
                df_filled[col].fillna(df[col].mean(), inplace=True)

    # 类别列用众数填充
    cat_cols = df.select_dtypes(include=["object", "category"]).columns
    for col in cat_cols:
        if df[col].isnull().sum() > 0:
            df_filled[col].fillna(df[col].mode()[0], inplace=True)

    return df_filled

# 技巧3: 高效的One-Hot编码（避免稀疏矩阵问题）
def memory_efficient_ohe(df, columns):
    """
    内存优化的独热编码
    """
    for col in columns:
        # 获取所有唯一值
        unique_vals = df[col].unique()

        # 逐列创建二进制列
        for val in unique_vals:
            df[f"{col}_{val}"] = (df[col] == val).astype(np.int8)

        # 删除原始列
        df.drop(col, axis=1, inplace=True)

    return df

# 技巧4: 分箱处理连续变量
def create_binned_features(df, numeric_col, n_bins=5, strategy="quantile"):
    """
    创建分箱特征
    """
    kbd = KBinsDiscretizer(n_bins=n_bins, encode="ordinal", strategy=strategy)
    df[f"{numeric_col}_binned"] = kbd.fit_transform(df[[numeric_col]])

    return df

# training_tips.py
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.metrics import get_scorer
import numpy as np

# 技巧1: 使用分层K折保持类别分布
skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

# 技巧2: 多指标评估
scoring = ["accuracy", "precision", "recall", "f1", "roc_auc"]

# 技巧3: 早停法防止过拟合
class EarlyStoppingCallback:
    """简单的早停回调"""

    def __init__(self, patience=10, min_delta=0.001):
        self.patience = patience
        self.min_delta = min_delta
        self.best_score = -np.inf
        self.counter = 0

    def __call__(self, score):
        if score > self.best_score + self.min_delta:
            self.best_score = score
            self.counter = 0
        else:
            self.counter += 1

        return self.counter >= self.patience

# 技巧4: 处理不平衡数据集
def handle_imbalance(X, y, method="smote"):
    """
    处理不平衡数据集

    方法:
    - smote: Synthetic Minority Over-sampling Technique
    - undersample: 欠采样
    - class_weight: 类别权重
    """
    from sklearn.utils.class_weight import compute_class_weight

    if method == "class_weight":
        class_weights = compute_class_weight(
            "balanced",
            classes=np.unique(y),
            y=y
        )
        return dict(zip(np.unique(y), class_weights))

    elif method == "smote":
        try:
            from imblearn.over_sampling import SMOTE
            smote = SMOTE(random_state=42)
            X_resampled, y_resampled = smote.fit_resample(X, y)
            return X_resampled, y_resampled
        except ImportError:
            print("请安装 imbalanced-learn: pip install imbalanced-learn")
            return X, y

    return None

# 技巧5: 集成多个模型提升性能
class ModelEnsemble:
    """模型集成"""

    def __init__(self, models, weights=None):
        self.models = models
        self.weights = weights or [1/len(models)] * len(models)

    def fit(self, X, y):
        for model in self.models:
            model.fit(X, y)
        return self

    def predict_proba(self, X):
        predictions = []
        for model, weight in zip(self.models, self.weights):
            pred = model.predict_proba(X) * weight
            predictions.append(pred)

        return np.sum(predictions, axis=0)

    def predict(self, X):
        proba = self.predict_proba(X)
        return np.argmax(proba, axis=1)

# optimization_tips.py
import numpy as np
import pandas as pd
from time import time

# 技巧1: 使用向量化操作替代循环
def vectorized_vs_loop():
    """对比向量化与循环的性能"""

    # 创建测试数据
    n = 1_000_000
    data = np.random.randn(n)

    # 循环方式
    start = time()
    result_loop = np.zeros(n)
    for i in range(n):
        result_loop[i] = data[i] * 2 + 1
    loop_time = time() - start

    # 向量化方式
    start = time()
    result_vec = data * 2 + 1
    vec_time = time() - start

    print(f"循环耗时: {loop_time:.4f}秒")
    print(f"向量化耗时: {vec_time:.4f}秒")
    print(f"加速比: {loop_time/vec_time:.2f}x")

# 技巧2: 使用适当的数据类型节省内存
def optimize_dtypes(df):
    """优化DataFrame数据类型"""

    # 优化整数列
    for col in df.select_dtypes(include=["int"]).columns:
        col_min = df[col].min()
        col_max = df[col].max()

        if col_min >= 0:
            if col_max < 255:
                df[col] = df[col].astype(np.uint8)
            elif col_max < 65535:
                df[col] = df[col].astype(np.uint16)
            elif col_max < 4294967295:
                df[col] = df[col].astype(np.uint32)
        else:
            if col_min > -128 and col_max < 127:
                df[col] = df[col].astype(np.int8)
            # ... 其他类型

    # 优化浮点数列
    for col in df.select_dtypes(include=["float"]).columns:
        df[col] = df[col].astype(np.float32)

    return df

# 技巧3: 使用numba加速数值计算
try:
    from numba import jit

    @jit(nopython=True)
    def numba_accelerated_sum(arr):
        """Numba加速的求和函数"""
        total = 0.0
        for i in range(len(arr)):
            total += arr[i]
        return total

    @jit(nopython=True)
    def numba_monte_carlo_pi(n_samples):
        """Numba加速的蒙特卡洛Pi计算"""
        n_inside = 0
        for _ in range(n_samples):
            x = np.random.random()
            y = np.random.random()
            if x*x + y*y <= 1.0:
                n_inside += 1
        return 4.0 * n_inside / n_samples

except ImportError:
    print("Numba未安装: pip install numba")

# 技巧4: 分批处理大数据集
def batch_processing(data, batch_size=10000):
    """
    分批处理大数据集
    """
    n_samples = len(data)
    n_batches = (n_samples + batch_size - 1) // batch_size

    results = []
    for i in range(n_batches):
        start_idx = i * batch_size
        end_idx = min((i + 1) * batch_size, n_samples)

        batch = data[start_idx:end_idx]
        # 处理这个批次
        batch_result = process_batch(batch)
        results.append(batch_result)

        if (i + 1) % 10 == 0:
            print(f"已处理 {i + 1}/{n_batches} 批次")

    return combine_results(results)

def process_batch(batch):
    """处理单个批次"""
    # 实际处理逻辑
    return batch

def combine_results(results):
    """合并所有批次结果"""
    return pd.concat(results, ignore_index=True)

# experiment_tracking.py
import mlflow
from mlflow.tracking import MlflowClient

# 设置MLflow跟踪服务器
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("customer_churn_prediction")

def log_experiment(model, X_train, X_test, y_train, y_test, params):
    """
    使用MLflow记录实验
    """
    with mlflow.start_run():
        # 记录参数
        mlflow.log_params(params)

        # 训练和评估
        model.fit(X_train, y_train)
        train_score = model.score(X_train, y_train)
        test_score = model.score(X_test, y_test)

        # 记录指标
        mlflow.log_metrics({
            "train_accuracy": train_score,
            "test_accuracy": test_score,
            "test_train_gap": train_score - test_score
        })

        # 记录模型
        mlflow.sklearn.log_model(model, "model")

        # 记录特征重要性（如果是树模型）
        if hasattr(model, "feature_importances_"):
            importances = model.feature_importances_
            mlflow.log_dict(
                dict(zip(feature_names, importances)),
                "feature_importances.json"
            )

        return test_score

# 使用示例
# with mlflow.start_run(run_name="xgboost_baseline"):
#     log_experiment(xgb_model, X_train, X_test, y_train, y_test, params)