Python量化金融：从数据到策略的工程实现

# 使用Polars进行高性能数据处理
import polars as pl

def clean_tick_data(df: pl.DataFrame) -> pl.DataFrame:
    return (
        df.lazy()
        .filter(pl.col("volume") > 0)
        .with_columns([
            (pl.col("ask") - pl.col("bid")).alias("spread"),
            pl.col("amount").log().alias("log_amount")
        ])
        .collect(streaming=True)
    )

class RiskEngine:
    def __init__(self, max_drawdown=0.2):
        self.portfolio = {}
        self.max_drawdown = max_drawdown
    
    async def check_order(self, order):
        position = await get_current_position()
        if position.unrealized_pnl < -self.max_drawdown:
            raise RiskException("触发最大回撤限制")

from backtesting import Strategy

class MeanReversion(Strategy):
    def init(self):
        self.sma = self.I(SMA, self.data.Close, 20)
        self.upper, self.lower = bollinger_bands(self.data.Close)
    
    def next(self):
        if crossover(self.data.Close, self.lower):
            self.buy()
        elif crossunder(self.data.Close, self.upper):
            self.sell()