📖 Python 3.12/3.13新特性与Polars¶

# === Python 3.11 旧写法 ===
from typing import TypeVar, Generic

T = TypeVar('T')
U = TypeVar('U')

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()

def first(items: list[T]) -> T:
    return items[0]

# === Python 3.12 新写法 ===
class Stack[T]:            # 直接在类名后声明类型参数
    def __init__(self) -> None:
        self._items: list[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()

def first[T](items: list[T]) -> T:  # 函数泛型也简化了
    return items[0]

# 类型别名新语法
type Point = tuple[float, float]            # 简单别名
type Matrix[T] = list[list[T]]             # 泛型别名
type RecursiveList[T] = T | list["RecursiveList[T]"]  # 递归类型

# Python 3.12: f-string支持任意表达式，包括嵌套引号
names = ["Alice", "Bob", "Charlie"]

# 3.12之前：f-string内不能用相同引号
# 3.12：可以了！
msg = f"Names: {", ".join(names)}"
print(msg)  # Names: Alice, Bob, Charlie

# 多行表达式
result = f"""
{
    sum(
        x**2
        for x in range(10)
    )
}
"""

# 嵌套f-string
data = {"name": "Alice", "score": 95}
report = f"{'='*20}\n{f'{data["name"]}: {data["score"]}'}\n{'='*20}"
print(report)

# Python 3.12: 支持每个子解释器有独立的GIL
# 这是未来移除GIL的第一步

# 目前通过C API使用（Python 3.13将暴露Python接口）
# 概念示意：
"""
主解释器 (GIL-1)
  └── 子解释器A (GIL-2)  ← 独立GIL，真正并行
  └── 子解释器B (GIL-3)  ← 独立GIL，真正并行

vs 传统多线程:
主解释器 (GIL)
  └── 线程A ← 共享GIL，无法真正并行
  └── 线程B ← 共享GIL，无法真正并行
"""

# Python 3.12 性能亮点:
# 1. 推导式内联（Comprehension Inlining）- 不再创建额外函数帧
# 2. 更小的对象内存占用
# 3. asyncio性能提升

import sys
print(f"Python版本: {sys.version}")

# 性能测试示例
import time

def benchmark_comprehension(n=1_000_000):
    """推导式性能（3.12比3.11快约2倍）"""
    start = time.perf_counter()
    result = [x * x for x in range(n)]
    elapsed = time.perf_counter() - start
    print(f"推导式 {n}项: {elapsed*1000:.1f}ms")

benchmark_comprehension()

# Python 3.13: 实验性移除GIL！
# 安装: python3.13t（带t后缀表示free-threaded版本）

# 传统Python: GIL限制CPU密集型多线程
# Free-threaded: 真正的多线程并行

import threading
import time

def cpu_bound(n):
    """CPU密集型任务"""
    total = 0
    for i in range(n):
        total += i * i
    return total

# 多线程测试
def test_multithreading():
    n = 10_000_000

    # 单线程
    start = time.perf_counter()
    cpu_bound(n)
    cpu_bound(n)
    single = time.perf_counter() - start

    # 双线程
    start = time.perf_counter()
    t1 = threading.Thread(target=cpu_bound, args=(n,))  # threading.Thread创建新线程执行并发任务
    t2 = threading.Thread(target=cpu_bound, args=(n,))
    t1.start(); t2.start()
    t1.join(); t2.join()
    multi = time.perf_counter() - start

    print(f"单线程: {single:.3f}s")
    print(f"双线程: {multi:.3f}s")
    print(f"加速比: {single/multi:.2f}x")
    # 传统Python: ~1.0x (GIL限制)
    # Free-threaded: ~1.8-2.0x (真正并行!)

test_multithreading()

# Python 3.13: 全新的交互式REPL
# 特性:
# - 多行编辑（上下键浏览历史时保持完整代码块）
# - 语法高亮（关键字/字符串/数字不同颜色）
# - 更好的错误提示
# - Tab自动补全增强
# - 支持paste mode（粘贴多行代码自动检测）

# 错误信息改进示例:
"""
>>> x = [1, 2, 3
... ]
>>> x[5]
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
    x[5]
    ~^^^
IndexError: list index out of range

# 3.13的错误提示更精确，用 ^^^ 箭头指向问题位置
"""

# Python 3.13引入实验性JIT编译器(copy-and-patch JIT)
# 需要编译时开启: --enable-experimental-jit

# JIT性能现状（实验性基础设施）:
# - 3.13 JIT 当前性能尚不如解释器（copy-and-patch 初始实现）
# - 3.15+ 预期仅个位数的百分比提升
# - 与PyPy不同，CPython JIT更轻量，保持C扩展兼容性
# - 重点是基础设施搭建，为未来优化铺路

# 检查JIT是否启用
import sys
print(f"Python {sys.version}")
# 3.13+: sys.flags中会包含jit相关flag

import asyncio

async def fetch_data(url: str) -> dict:  # async def定义协程函数，await暂停等待异步结果
    """模拟异步请求"""
    await asyncio.sleep(0.1)
    return {"url": url, "status": 200}

async def main():
    # Python 3.11+ TaskGroup（3.13进一步优化）
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("https://api1.com"))
        task2 = tg.create_task(fetch_data("https://api2.com"))
        task3 = tg.create_task(fetch_data("https://api3.com"))

    # 所有任务完成后才到这里
    results = [task1.result(), task2.result(), task3.result()]
    print(f"获取 {len(results)} 个结果")

    # 3.13改进: 更好的异常传播和取消处理
    # 任何任务失败会自动取消其他任务（结构化并发）

asyncio.run(main())  # asyncio.run()启动异步事件循环并执行协程

import polars as pl

# 创建DataFrame
df = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie", "Diana", "Eve"],
    "age": [25, 30, 35, 28, 32],
    "department": ["AI", "Backend", "AI", "Frontend", "AI"],
    "salary": [50000, 60000, 75000, 55000, 70000]
})

print(df)
print(f"Schema: {df.schema}")

# 基础筛选和选择
result = df.filter(
    pl.col("department") == "AI"
).select(
    "name", "salary"
).sort("salary", descending=True)

print("\nAI部门（按薪资降序）:")
print(result)

import polars as pl
import numpy as np

# 生成大数据集
np.random.seed(42)
n = 1_000_000
df = pl.DataFrame({
    "user_id": np.random.randint(1, 10001, n),
    "product_id": np.random.randint(1, 1001, n),
    "amount": np.random.uniform(10, 1000, n).round(2),
    "category": np.random.choice(["电子", "服装", "食品", "家居"], n),
    "date": pl.date_range(pl.date(2024, 1, 1), pl.date(2024, 12, 31),
                          eager=True).sample(n, with_replacement=True)
})

# 链式表达式（声明式，可自动优化）
result = (
    df
    .group_by("category")
    .agg(
        pl.col("amount").sum().alias("总销售额"),
        pl.col("amount").mean().alias("平均客单价"),
        pl.col("user_id").n_unique().alias("独立用户数"),
        pl.col("amount").quantile(0.95).alias("P95金额"),
        (pl.col("amount") > 500).sum().alias("大额订单数"),  # 布尔表达式求和：True计为1，统计满足条件的行数
    )
    .sort("总销售额", descending=True)
    .with_columns(
        (pl.col("总销售额") / pl.col("总销售额").sum() * 100)  # .sum()在表达式内计算该列总和，实现"每行值/总和*100"求占比
        .round(2)
        .alias("销售占比%")
    )
)

print("按品类统计:")
print(result)

import polars as pl

# 惰性模式：构建执行计划，优化后再执行
lazy_result = (
    pl.scan_csv("sales_data.csv")  # 不立即读取
    .filter(pl.col("amount") > 100)
    .group_by("category")
    .agg(pl.col("amount").sum())
    .sort("amount", descending=True)
    .head(10)
)

# 查看执行计划（Polars会自动优化）
print("执行计划:")
print(lazy_result.explain())

# 实际执行
# result = lazy_result.collect()

# 优化示例:
# - 谓词下推: filter在group_by之前执行
# - 投影下推: 只读取需要的列
# - 公共子表达式消除

import polars as pl

df = pl.DataFrame({
    "name": ["Alice", "Bob", "Charlie", "Diana", "Eve", "Frank"],
    "department": ["AI", "AI", "AI", "Backend", "Backend", "Backend"],
    "performance": [95, 88, 92, 85, 90, 78]
})

result = df.with_columns(
    # 部门内排名
    pl.col("performance")
        .rank(descending=True)
        .over("department")
        .alias("部门排名"),

    # 部门平均分
    pl.col("performance")
        .mean()
        .over("department")
        .alias("部门平均"),

    # 与部门平均的差距
    (pl.col("performance") - pl.col("performance").mean().over("department"))
        .round(1)
        .alias("偏差")
)

print(result)

import polars as pl
import pandas as pd
import numpy as np
import time

def benchmark_comparison(n=5_000_000):
    """Polars vs Pandas性能对比"""
    # 生成数据
    np.random.seed(42)
    data = {
        "group": np.random.choice(["A", "B", "C", "D", "E"], n),
        "value1": np.random.randn(n),
        "value2": np.random.randn(n),
    }

    # Pandas
    df_pd = pd.DataFrame(data)
    start = time.perf_counter()
    result_pd = df_pd.groupby("group").agg({"value1": "mean", "value2": "sum"})
    pandas_time = time.perf_counter() - start

    # Polars
    df_pl = pl.DataFrame(data)
    start = time.perf_counter()
    result_pl = df_pl.group_by("group").agg(
        pl.col("value1").mean(),
        pl.col("value2").sum()
    )
    polars_time = time.perf_counter() - start

    print(f"数据量: {n:,}行")
    print(f"Pandas:  {pandas_time*1000:.1f}ms")
    print(f"Polars:  {polars_time*1000:.1f}ms")
    print(f"加速比:  {pandas_time/polars_time:.1f}x")

benchmark_comparison()

import polars as pl

# 特征工程场景（Polars擅长的领域）
def feature_engineering(df: pl.DataFrame) -> pl.DataFrame:
    """用Polars做ML特征工程"""
    return df.with_columns(
        # 滚动统计
        pl.col("amount").rolling_mean(window_size=7).alias("7日均值"),
        pl.col("amount").rolling_std(window_size=7).alias("7日标准差"),

        # 滞后特征
        pl.col("amount").shift(1).alias("前1天"),
        pl.col("amount").shift(7).alias("前7天"),

        # 分组统计编码
        pl.col("amount").mean().over("category").alias("品类均值编码"),

        # 时间特征
        pl.col("date").dt.weekday().alias("星期几"),
        pl.col("date").dt.month().alias("月份"),

        # 组合特征
        (pl.col("amount") / pl.col("amount").mean().over("category"))
        .alias("相对品类均值"),
    ).drop_nulls()  # 去除滚动计算产生的null

# Polars与sklearn配合
# df_features = feature_engineering(df)
# X = df_features.select(feature_cols).to_numpy()  # 零拷贝转numpy
# y = df_features["target"].to_numpy()

from dataclasses import dataclass

@dataclass  # @dataclass自动生成__init__、__repr__等常用方法
class Point:
    x: float
    y: float

@dataclass
class Circle:
    center: Point
    radius: float

@dataclass
class Rectangle:
    top_left: Point
    bottom_right: Point

def describe_shape(shape):
    """结构模式匹配 - 强大的解构能力"""
    match shape:  # match-case是Python 3.10+的结构化模式匹配语法
        # 解构dataclass
        case Circle(center=Point(x=0, y=0), radius=r):
            return f"原点处的圆，半径={r}"

        case Circle(center=Point(x=cx, y=cy), radius=r) if r > 10:
            return f"大圆在({cx},{cy})，半径={r}"

        case Rectangle(top_left=Point(x=x1, y=y1),
                       bottom_right=Point(x=x2, y=y2)):
            w, h = abs(x2 - x1), abs(y2 - y1)
            return f"矩形 {w}x{h}"

        case _:
            return "未知形状"

# 测试
shapes = [
    Circle(Point(0, 0), 5),
    Circle(Point(3, 4), 15),
    Rectangle(Point(0, 0), Point(10, 20))
]

for s in shapes:
    print(f"{s} → {describe_shape(s)}")

# Python 3.11+的精确错误定位
"""
示例1:
  x = {"key": "value"}
  y = x["missing_key"]
       ~~^^^^^^^^^^^^^^^
  KeyError: 'missing_key'

示例2:
  result = a + b * c / d
                    ~~^~~
  ZeroDivisionError: division by zero

示例3:
  obj.attr1.attr2.attr3
  ~~~~~~~~~~^^^^^
  AttributeError: 'NoneType' object has no attribute 'attr2'

# 3.13进一步改进:
# - 更好的建议 "did you mean ...?"
# - 导入错误的自动建议
"""