Python 数据科学与 AI：Pandas/Polars、PyTorch 推理与 ONNX 部署

import numpy as np

# 向量化替代循环（1000x 加速）
# ❌ 慢
result = []
for x in range(1_000_000):
    result.append(x ** 2 + 2 * x + 1)

# ✅ 快
x = np.arange(1_000_000)
result = x ** 2 + 2 * x + 1  # 广播，无 Python 循环

# 高级索引
arr = np.random.randn(1000, 1000)
mask = arr > 2  # 布尔掩码
outliers = arr[mask]  # 只取大于 2 的元素

# 内存视图（零拷贝）
sub = arr[:100, :100]  # 不复制数据

import pandas as pd

# 读取大文件的 4 个技巧
df = pd.read_csv(
    "large.csv",
    usecols=["user_id", "amount", "timestamp"],  # 只读需要的列
    dtype={"user_id": "int32", "amount": "float32"},  # 降精度
    parse_dates=["timestamp"],
    chunksize=100_000  # 分块处理
)

# 高效分组聚合
# ❌ 慢：apply 调用 Python 函数
df.groupby("category").apply(lambda x: x["amount"].sum())

# ✅ 快：向量化聚合
agg = df.groupby("category").agg({
    "amount": ["sum", "mean", "count"],
    "user_id": "nunique"
})

# 时间序列重采样
# 按小时统计订单量
df.set_index("timestamp").resample("H")["order_id"].count()

import polars as pl

# 惰性查询（查询优化类似 SQL 执行计划）
df = pl.scan_csv("huge.csv")  # 不立即读数据

result = (
    df
    .filter(pl.col("amount") > 100)
    .group_by("category")
    .agg([
        pl.col("amount").sum().alias("total"),
        pl.col("user_id").n_unique().alias("unique_users")
    ])
    .sort("total", descending=True)
    .limit(10)
    .collect()  # 这里才真正执行
)

# 与 Pandas 互操作
pandas_df = result.to_pandas()
polars_df = pl.from_pandas(pandas_df)

import torch
from torchvision import models, transforms
from PIL import Image

# 加载预训练模型（ResNet50）
model = models.resnet50(weights=models.ResNet50_Weights.IMAGENET1K_V2)
model.eval()

# 推理模式（禁用 dropout、batchnorm 更新）
with torch.no_grad():
    # 预处理管道
    preprocess = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize(
            mean=[0.485, 0.456, 0.406],
            std=[0.229, 0.224, 0.225]
        )
    ])
    
    img = Image.open("cat.jpg")
    input_tensor = preprocess(img).unsqueeze(0)
    
    # 推理
    output = model(input_tensor)
    probabilities = torch.nn.functional.softmax(output[0], dim=0)
    
    # Top 5 预测
    top5_prob, top5_idx = torch.topk(probabilities, 5)

# 1. 半精度加速（FP16）
model.half()  # 权重转 FP16
input_tensor = input_tensor.half()

# 2. 批处理推理（摊平启动开销）
def batch_predict(model, images: list[Image.Image], batch_size: int = 32):
    results = []
    for i in range(0, len(images), batch_size):
        batch = images[i:i + batch_size]
        tensors = torch.stack([preprocess(img) for img in batch])
        with torch.no_grad():
            outputs = model(tensors)
        results.extend(outputs)
    return results

# 3. TorchScript 编译（图优化）
scripted = torch.jit.script(model)
# 或 traced = torch.jit.trace(model, example_input)

# 4. 推理服务封装
class InferenceService:
    def __init__(self, model_path: str, device: str = "cpu"):
        self.device = torch.device(device)
        self.model = torch.load(model_path, map_location=self.device)
        self.model.eval().to(self.device)
    
    @torch.inference_mode()
    def predict(self, image: Image.Image) -> dict:
        tensor = preprocess(image).unsqueeze(0).to(self.device)
        output = self.model(tensor)
        return {" logits": output.cpu().numpy().tolist()}

from transformers import pipeline

# 文本分类
classifier = pipeline(
    "sentiment-analysis",
    model="distilbert-base-uncased-finetuned-sst-2-english"
)
result = classifier("I love FastAPI!")
# [{'label': 'POSITIVE', 'score': 0.9998}]

# 命名实体识别
ner = pipeline("ner", model="dslim/bert-base-NER")
ner("Apple Inc. was founded by Steve Jobs in Cupertino.")

# 文本生成（LLM）
generator = pipeline(
    "text-generation",
    model="microsoft/Phi-3-mini-4k-instruct",
    torch_dtype="auto",
    device_map="auto"  # 自动分配到 GPU/CPU
)

prompt = "Write a Python function to calculate fibonacci:"
output = generator(prompt, max_new_tokens=100, temperature=0.7)

from transformers import AutoModelForCausalLM, BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,  # 4-bit 量化
    bnb_4bit_compute_dtype=torch.float16,
    bnb_4bit_quant_type="nf4",  # Normal Float 4
    bnb_4bit_use_double_quant=True,
)

model = AutoModelForCausalLM.from_pretrained(
    "meta-llama/Llama-2-7b",
    quantization_config=quantization_config,
    device_map="auto",
)
# 7B 模型从 14GB 降至 4GB

from vllm import LLM

# 比 Transformers 快 10-20x 的批处理推理
llm = LLM(model="meta-llama/Llama-2-7b", tensor_parallel_size=1)

prompts = [
    "What is Python used for?",
    "Explain async/await in one sentence.",
]
outputs = llm.generate(prompts)

for output in outputs:
    print(output.outputs[0].text)

import torch

dummy_input = torch.randn(1, 3, 224, 224)

torch.onnx.export(
    model,
    dummy_input,
    "resnet50.onnx",
    input_names=["input"],
    output_names=["output"],
    dynamic_axes={"input": {0: "batch_size"}, "output": {0: "batch_size"}},
    opset_version=17,
)

import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("resnet50.onnx")

# 输入准备
input_name = session.get_inputs()[0].name
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)

# 推理
outputs = session.run(None, {input_name: input_data})

# 不同执行提供器
providers = [
    "CUDAExecutionProvider",  # GPU
    "TensorRTExecutionProvider",  # NVIDIA TensorRT
    "CPUExecutionProvider",  # 回退
]
session = ort.InferenceSession("model.onnx", providers=providers)

// Rust 代码（lib.rs）
use pyo3::prelude::*;

#[pyfunction]
fn fast_sum(numbers: Vec<f64>) -> f64 {
    numbers.iter().sum()
}

#[pymodule]
fn my_rust_lib(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fast_sum, m)?)?;
    Ok(())
}

# Python 调用
import maturin  # pip install maturin
# maturin develop  # 编译并安装

from my_rust_lib import fast_sum

# 比 Python sum() 快 100x
result = fast_sum([1.0] * 10_000_000)

from pathlib import Path
import polars as pl
from transformers import pipeline
import torch

class DataPipeline:
    def __init__(self):
        self.classifier = pipeline(
            "sentiment-analysis",
            model="nlptown/bert-base-multilingual-uncased-sentiment",
            device=0 if torch.cuda.is_available() else -1
        )
    
    def process_csv(self, path: Path) -> pl.DataFrame:
        # 1. 读取（惰性）
        df = pl.scan_csv(path)
        
        # 2. 清洗
        df = df.filter(
            pl.col("text").is_not_null() &
            (pl.col("text").str.len_chars() > 10)
        )
        
        # 3. 采样（开发时）
        # df = df.limit(1000)
        
        # 4. 收集为 Pandas（兼容性）
        return df.collect().to_pandas()
    
    def analyze_sentiment(self, df: pd.DataFrame) -> pd.DataFrame:
        texts = df["text"].tolist()
        
        # 批处理推理
        results = self.classifier(texts, batch_size=32)
        
        df["sentiment"] = [r["label"] for r in results]
        df["confidence"] = [r["score"] for r in results]
        return df
    
    def run(self, input_path: Path, output_path: Path):
        df = self.process_csv(input_path)
        df = self.analyze_sentiment(df)
        df.to_parquet(output_path)  # 列式存储，比 CSV 快 10x

# 执行
pipeline = DataPipeline()
pipeline.run(Path("reviews.csv"), Path("reviews_analyzed.parquet"))

from sklearn.preprocessing import StandardScaler, RobustScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
import numpy as np

# 数值列：对异常值鲁棒的缩放
numeric_features = ["age", "income", "account_age_days"]
numeric_transformer = RobustScaler(quantile_range=(5, 95))

# 类别列：高频截断 + One-Hot
categorical_features = ["country", "device_type"]
categorical_transformer = OneHotEncoder(
    handle_unknown="ignore",
    max_categories=50,  # 只保留 top 50 类别
    sparse_output=False
)

# 组合预处理管道
preprocessor = ColumnTransformer([
    ("num", numeric_transformer, numeric_features),
    ("cat", categorical_transformer, categorical_features),
])

# 与模型组合为完整管道
from sklearn.pipeline import Pipeline
from sklearn.ensemble import GradientBoostingClassifier

clf = Pipeline([
    ("preprocess", preprocessor),
    ("classifier", GradientBoostingClassifier(n_estimators=100)),
])

def extract_time_features(df: pd.DataFrame, col: str) -> pd.DataFrame:
    dt = pd.to_datetime(df[col])
    df[f"{col}_hour"] = dt.dt.hour
    df[f"{col}_dayofweek"] = dt.dt.dayofweek
    df[f"{col}_month"] = dt.dt.month
    df[f"{col}_is_weekend"] = dt.dt.dayofweek >= 5
    df[f"{col}_quarter"] = dt.dt.quarter
    # 周期性编码（正弦/余弦，保持周期连续性）
    df[f"{col}_hour_sin"] = np.sin(2 * np.pi * dt.dt.hour / 24)
    df[f"{col}_hour_cos"] = np.cos(2 * np.pi * dt.dt.hour / 24)
    return df

from sklearn.model_selection import StratifiedKFold, cross_val_score
from sklearn.metrics import roc_auc_score, classification_report

# 时序数据：TimeSeriesSplit；类别不平衡：StratifiedKFold
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

scores = cross_val_score(clf, X, y, cv=cv, scoring="roc_auc", n_jobs=-1)
print(f"AUC: {scores.mean():.4f} (+/- {scores.std() * 2:.4f})")

import optuna
from sklearn.model_selection import cross_val_score

def objective(trial):
    params = {
        "n_estimators": trial.suggest_int("n_estimators", 50, 500),
        "max_depth": trial.suggest_int("max_depth", 3, 10),
        "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.3, log=True),
        "subsample": trial.suggest_float("subsample", 0.6, 1.0),
        "colsample_bytree": trial.suggest_float("colsample_bytree", 0.6, 1.0),
    }
    model = GradientBoostingClassifier(**params, random_state=42)
    score = cross_val_score(model, X_train, y_train, cv=3, scoring="roc_auc").mean()
    return score

study = optuna.create_study(direction="maximize")
study.optimize(objective, n_trials=50, show_progress_bar=True)
print(f"Best score: {study.best_value:.4f}, params: {study.best_params}")

import mlflow
import mlflow.sklearn

mlflow.set_experiment("fraud-detection")

with mlflow.start_run():
    # 记录参数
    mlflow.log_param("model_type", "xgboost")
    mlflow.log_param("n_estimators", 200)

    # 记录指标
    mlflow.log_metric("auc", 0.943)
    mlflow.log_metric("f1", 0.891)

    # 记录模型 + 签名
    mlflow.sklearn.log_model(
        model,
        artifact_path="model",
        registered_model_name="fraud_detection_v1"
    )

from fastapi import FastAPI
import onnxruntime as ort
import numpy as np

app = FastAPI()
session = ort.InferenceSession("model.onnx")
input_name = session.get_inputs()[0].name

@app.post("/predict")
def predict(features: list[float]):
    arr = np.array([features], dtype=np.float32)
    outputs = session.run(None, {input_name: arr})
    return {"prediction": float(outputs[0][0])}

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset, TargetDriftPreset

# 比较参考数据集与当前生产数据
report = Report(metrics=[DataDriftPreset(), TargetDriftPreset()])
report.run(reference_data=df_ref, current_data=df_prod)

# 生成 HTML 报告
report.save_html("drift_report.html")

# 若 drift > 阈值，触发重训练管道
if report.as_dict()["metrics"][0]["result"]["dataset_drift"]:
    trigger_retraining_pipeline()

import torch

# 自动混合精度（AMP）：减少显存占用 + 加速
def train_step_amp(model, data, target, optimizer, scaler):
    optimizer.zero_grad()
    with torch.autocast(device_type="cuda", dtype=torch.float16):
        output = model(data)
        loss = criterion(output, target)

    scaler.scale(loss).backward()
    scaler.step(optimizer)
    scaler.update()

# Gradient Accumulation：小显存模拟大 batch
accumulation_steps = 4
for i, (data, target) in enumerate(dataloader):
    loss = model(data, target) / accumulation_steps
    loss.backward()
    if (i + 1) % accumulation_steps == 0:
        optimizer.step()
        optimizer.zero_grad()

# 实时监控显存使用
torch.cuda.memory_summary(device=0)
# 或
!nvidia-smi --query-gpu=memory.used,memory.total --format=csv