Python主要有哪些应用方向？Web、数据科学与AI领域解析

from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel, Field
from typing import List
import asyncio

app = FastAPI(title="订单微服务系统", version="1.0.0")

# 1. 声明数据传输模型 (DTO / Schema)
class OrderItem(BaseModel):
    item_id: int
    name: str = Field(..., min_length=1, max_length=50)
    price: float = Field(..., gt=0.0)

class CreateOrderRequest(BaseModel):
    user_id: int
    items: List[OrderItem]
    total_amount: float

class OrderResponse(BaseModel):
    order_id: str
    status: str
    total_amount: float

# 2. 模拟数据库异步连接依赖
async def get_db_session():
    # 模拟从异步连接池获取 session
    session = {"pool_id": "pg_conn_01", "active": True}
    yield session
    # 请求处理完成后自动执行清理释放

# 3. 异步业务接口实现
@app.post("/api/v1/orders", response_model=OrderResponse, status_code=status.HTTP_201_CREATED)
async def create_order(payload: CreateOrderRequest, db=Depends(get_db_session)):
    # 检查业务逻辑一致性
    calculated_sum = sum(item.price for item in payload.items)
    if abs(calculated_sum - payload.total_amount) > 0.01:
        raise HTTPException(status_code=400, detail="订单金额与商品总价不符")
    
    # 模拟异步写入数据库 (非阻塞 I/O)
    await asyncio.sleep(0.02) 
    generated_id = f"ORD-{payload.user_id}-9871"
    
    return OrderResponse(
        order_id=generated_id,
        status="CONFIRMED",
        total_amount=payload.total_amount
    )

import pandas as pd
import numpy as np

# 1. 模拟生成原始交易流水数据
np.random.seed(42)
records = 10000
raw_data = {
    "order_id": np.arange(10001, 10001 + records),
    "user_id": np.random.randint(100, 800, size=records),
    "category": np.random.choice(["Electronics", "Clothing", "Home", "Books"], size=records),
    "amount": np.random.exponential(scale=120, size=records).round(2),
    "is_returned": np.random.choice([0, 1], p=[0.92, 0.08], size=records)
}
df = pd.DataFrame(raw_data)

# 2. 数据清洗与特征列派生 (矢量化计算，耗时毫秒级)
valid_orders = df[df["is_returned"] == 0].copy()
valid_orders["is_high_value"] = valid_orders["amount"] >= 200

# 3. 多维度分组聚合 (用户级别行为画像)
user_summary = valid_orders.groupby("user_id").agg(
    total_spend=("amount", "sum"),
    order_count=("order_id", "count"),
    high_value_orders=("is_high_value", "sum")
)

# 4. 筛选核心优质客户群 (客单价 >= 100 且 订单量 >= 5)
user_summary["avg_order_value"] = (user_summary["total_spend"] / user_summary["order_count"]).round(2)
vip_users = user_summary[
    (user_summary["order_count"] >= 5) & 
    (user_summary["avg_order_value"] >= 100)
].sort_values(by="total_spend", ascending=False)

print(f"有效订单总数: {len(valid_orders)}")
print(f"核心 VIP 用户规模: {len(vip_users)} 人")
print("Top 3 客户指标示例:")
print(vip_users.head(3))

import torch
from transformers import AutoTokenizer, AutoModel

# 1. 检测硬件计算加速设备
device = torch.device("cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu"))
print(f"当前计算后端: {device}")

# 2. 加载预训练语义特征模型与分词器 (示例使用开源轻量向量模型)
model_name = "BAAI/bge-small-zh-v1.5"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModel.from_pretrained(model_name).to(device)
model.eval()  # 设置为评估模式，关闭 Dropout

# 3. 准备待编码的企业知识库文本
documents = [
    "Python Web 推荐使用 FastAPI 结合 Pydantic 处理异步 JSON 接口",
    "NumPy 和 Pandas 主要用于处理连续内存结构化表格数据分析",
    "PyTorch 是全球深度学习和通用人工智能大模型研发的核心框架"
]

# 4. 执行批量分词与张量前向传递 (无梯度推断节省显存)
with torch.no_grad():
    inputs = tokenizer(documents, padding=True, truncation=True, max_length=128, return_tensors="pt").to(device)
    outputs = model(**inputs)
    # 提取 [CLS] token 作为整句的高维语义表征 (768 维稠密向量)
    sentence_embeddings = outputs.last_hidden_state[:, 0, :]
    # L2 归一化，便于后续直接使用点积计算余弦相似度
    normalized_embeddings = torch.nn.functional.normalize(sentence_embeddings, p=2, dim=1)

print(f"输出特征矩阵形状: {normalized_embeddings.shape}")
# 示例：计算文档 0 与文档 1、2 之间的语义关联度
similarity_matrix = torch.matmul(normalized_embeddings, normalized_embeddings.T)
print("文档间余弦相似度矩阵 (示例输出):")
print(similarity_matrix.cpu().numpy().round(3))