Python程序设计主要应用领域：从Web开发到人工智能

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

app = FastAPI(
    title="数据计算与业务分析接口服务",
    description="提供异步非阻塞的统计指标计算服务",
    version="1.0.0"
)

# 1. 定义强类型请求输入规范
class AnalysisRequest(BaseModel):
    batch_id: str = Field(..., description="批次唯一标识符", min_length=4)
    values: List[float] = Field(..., description="待计算的数据浮点数组")
    multiplier: Optional[float] = Field(1.0, description="放大权重因子", ge=0.0)

# 2. 定义规范化输出结果模型
class AnalysisResponse(BaseModel):
    batch_id: str
    total_count: int
    mean_value: float
    max_value: float
    scaled_sum: float

# 3. 异步路由处理函数
@app.post("/api/v1/analyze", response_model=AnalysisResponse, status_code=status.HTTP_200_OK)
async def process_analysis(request: AnalysisRequest):
    if not request.values:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="输入数值数组不能为空"
        )
    
    # 模拟非阻塞I/O操作（例如异步查询数据库或等待上游分布式缓存）
    await asyncio.sleep(0.05)
    
    total = sum(request.values)
    count = len(request.values)
    mean = total / count
    max_val = max(request.values)
    scaled_total = total * request.multiplier
    
    return AnalysisResponse(
        batch_id=request.batch_id,
        total_count=count,
        mean_value=round(mean, 4),
        max_value=round(max_val, 4),
        scaled_sum=round(scaled_total, 4)
    )

# 启动命令（示例终端指令，非直接执行）：
# uvicorn web_service:app --host 0.0.0.0 --port 8000 --workers 4

import pandas as pd
import numpy as np

# 1. 模拟电商销售明细样本数据
raw_data = {
    "order_id": ["ORD-101", "ORD-102", "ORD-103", "ORD-104", "ORD-105", "ORD-106", "ORD-107"],
    "region": ["华东", "华北", "华东", "华南", "华北", "华东", "华南"],
    "category": ["电子", "家居", "电子", "服装", "电子", "服装", "家居"],
    "sales_amount": [1280.0, 320.0, np.nan, 450.0, 2100.0, -50.0, 680.0],  # 包含缺失值与异常负值
    "quantity": [2, 1, 3, 2, 4, 1, 2]
}

df = pd.DataFrame(raw_data)

# 2. 数据清洗：滤除金额缺失与非法负数
clean_df = df.dropna(subset=["sales_amount"]).copy()
clean_df = clean_df[clean_df["sales_amount"] > 0]

# 3. 衍生计算：计算单件平均客单价
clean_df["unit_price"] = clean_df["sales_amount"] / clean_df["quantity"]

# 4. 多维分组聚合统计
summary = clean_df.groupby(["region", "category"]).agg(
    total_revenue=("sales_amount", "sum"),
    avg_price=("unit_price", "mean"),
    order_count=("order_id", "count")
).reset_index()

# 5. 排序与输出结果
summary = summary.sort_values(by="total_revenue", ascending=False)
print("=== 区域与品类销售聚合汇总表 ===")
print(summary.to_string(index=False))

=== 区域与品类销售聚合汇总表 ===
region category  total_revenue  avg_price  order_count
  华北       电子         2100.0     525.00            1
  华东       电子         1280.0     640.00            1
  华南       家居          680.0     340.00            1
  华南       服装          450.0     225.00            1
  华北       家居          320.0     320.00            1

import torch
import torch.nn as nn
import torch.nn.functional as F

# 1. 继承 nn.Module 构建前向分类网络
class RiskClassifier(nn.Module):
    def __init__(self, input_dim: int, hidden_dim: int, num_classes: int):
        super(RiskClassifier, self).__init__()
        # 全连接特征提取层与归一化
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.layer_norm = nn.LayerNorm(hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, 32)
        self.out_proj = nn.Linear(32, num_classes)
        self.dropout = nn.Dropout(p=0.2)
        
    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # 输入张量维度: [batch_size, input_dim]
        x = self.fc1(x)
        x = self.layer_norm(x)
        x = F.relu(x)
        x = self.dropout(x)
        
        x = F.relu(self.fc2(x))
        logits = self.out_proj(x)
        return logits

# 2. 模拟生产推理输入验证
input_features = 16
batch_samples = 4
model = RiskClassifier(input_dim=input_features, hidden_dim=64, num_classes=2)
model.eval()  # 切换为评估推理模式（关闭Dropout随机性）

# 模拟输入张量（4个待检测风控样本）
dummy_input = torch.randn(batch_samples, input_features, dtype=torch.float32)

with torch.no_grad():
    logits = model(dummy_input)
    probabilities = F.softmax(logits, dim=1)
    predictions = torch.argmax(probabilities, dim=1)

print("=== PyTorch 模型前向推理输出 ===")
print(f"预测类别结果: {predictions.tolist()}")
print(f"类别概率分布:\n{probabilities.numpy()}")

# 3. 固化保存模型权重（供服务化部署加载）
torch.save(model.state_dict(), "risk_model_weights.pth")
print("\n[成功] 模型权重已固化保存至 risk_model_weights.pth")

import time
import requests
from typing import Dict, Any

TARGET_URL = "https://api.example.com/healthz"
MAX_RETRIES = 3
TIMEOUT_SECONDS = 3.0

def probe_service_status(url: str, max_retries: int, timeout: float) -> Dict[str, Any]:
    # 带有退避重试机制的服务健康度探测函数
    backoff_delay = 1.0
    
    for attempt in range(1, max_retries + 1):
        try:
            start_time = time.perf_counter()
            response = requests.get(url, timeout=timeout)
            latency_ms = (time.perf_counter() - start_time) * 1000.0
            
            if response.status_code == 200:
                return {
                    "status": "HEALTHY",
                    "code": response.status_code,
                    "latency_ms": round(latency_ms, 2),
                    "attempt": attempt
                }
            else:
                print(f"[警告] 第 {attempt} 次探测返回非200状态码: {response.status_code}")
        except (requests.ConnectionError, requests.Timeout) as err:
            print(f"[错误] 第 {attempt} 次探测失败: {type(err).__name__}")
        
        # 指数退避休眠后重试
        if attempt < max_retries:
            time.sleep(backoff_delay)
            backoff_delay *= 2.0
            
    return {
        "status": "UNHEALTHY",
        "code": None,
        "latency_ms": None,
        "attempt": max_retries
    }

# 执行并打印模拟输出
result = probe_service_status("https://httpbin.org/status/200", MAX_RETRIES, TIMEOUT_SECONDS)
print(f"探测最终结果: {result}")