量化开发者的 Python 技术栈：从数据处理到机器学习的完整工具链

import polars as pl
import pyarrow.parquet as pq

# 使用 Polars 读取 Parquet 格式（列式存储，查询效率高）
df = pl.read_parquet("data/stock_prices.parquet")

# 复杂的因子计算（比 Pandas 快一个数量级）
result = (
    df
    .filter(pl.col("trade_date") > "2023-01-01")
    .with_columns([
        pl.col("close").pct_change().alias("returns"),
        pl.col("close").rolling_mean(20).alias("ma20"),
        pl.col("close").rolling_std(20).alias("vol20"),
    ])
    .with_columns(
        (pl.col("close") / pl.col("close").rolling_mean(60) - 1).alias("price_to_ma60")
    )
)

import backtrader as bt

class MyStrategy(bt.Strategy):
    params = (
        ('sma_period', 20),
        ('rsi_period', 14),
        ('rsi_buy', 30),
        ('rsi_sell', 70),
    )

    def __init__(self):
        self.sma = bt.indicators.SimpleMovingAverage(
            self.data.close, period=self.params.sma_period
        )
        self.rsi = bt.indicators.RSI(
            self.data.close, period=self.params.rsi_period
        )

    def next(self):
        if not self.position:
            if self.rsi < self.params.rsi_buy:
                self.buy()
        else:
            if self.rsi > self.params.rsi_sell:
                self.sell()

cerebro = bt.Cerebro()
cerebro.addstrategy(MyStrategy)

data = bt.feeds.GenericCSVData(
    dataname='stock_data.csv',
    fromdate=datetime(2023, 1, 1),
    todate=datetime(2026, 6, 30),
    dtformat='%Y-%m-%d',
    datetime=0, open=1, high=2, low=3, close=4, volume=5
)
cerebro.adddata(data)
cerebro.run()
print(f'最终资金: {cerebro.broker.getvalue():.2f}')

from zipline import run_algorithm
from zipline.api import (
    schedule_function, date_rules, time_rules,
    symbol, order_target_percent, record
)

def initialize(context):
    context.asset = symbol('AAPL')
    schedule_function(rebalance, date_rules.every_day(), time_rules.market_open())

def rebalance(context, data):
    prices = data.current(context.asset, 'price')
    mavg = data.history(context.asset, 'price', 20, '1d').mean()
    if prices > mavg:
        order_target_percent(context.asset, 1.0)
    else:
        order_target_percent(context.asset, 0.0)

result = run_algorithm(
    start=pd.Timestamp('2023-01-01', tz='UTC'),
    end=pd.Timestamp('2026-06-30', tz='UTC'),
    initialize=initialize,
    capital_base=100000,
    bundle='csvdir'
)

import lightgbm as lgb
from sklearn.model_selection import TimeSeriesSplit

# 时序交叉验证（防止数据泄漏）
tscv = TimeSeriesSplit(n_splits=5)

params = {
    'objective': 'binary',
    'metric': 'auc',
    'boosting_type': 'gbdt',
    'num_leaves': 31,
    'learning_rate': 0.05,
    'feature_fraction': 0.8,
    'bagging_fraction': 0.8,
    'bagging_freq': 5,
    'verbose': -1
}

model = lgb.LGBMClassifier(**params, n_estimators=200)

# 训练时使用时间序列交叉验证
for train_idx, val_idx in tscv.split(X):
    X_train, X_val = X[train_idx], X[val_idx]
    y_train, y_val = y[train_idx], y[val_idx]
    
    model.fit(X_train, y_train, eval_set=[(X_val, y_val)])
    
    val_pred = model.predict_proba(X_val)[:, 1]
    ic = np.corrcoef(val_pred, y_val)[0, 1]
    print(f"Fold IC: {ic:.4f}")

import torch
import torch.nn as nn

class StockLSTM(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, dropout):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout
        )
        self.fc = nn.Linear(hidden_size, 1)
        self.sigmoid = nn.Sigmoid()
    
    def forward(self, x):
        lstm_out, _ = self.lstm(x)
        out = self.fc(lstm_out[:, -1, :])
        return self.sigmoid(out)