金融机器学习工具包：我的常用库与使用技巧分享

# 金融时间序列滑窗处理（含未来数据防护）
import pandas as pd
import numpy as np

# 生成含停牌日的股票数据（2025-03-01为停牌日）
dates = pd.date_range('2025-02-25', periods=5, freq='D')
data = {'close': [100, 102, np.nan, 105, 108]}
df = pd.DataFrame(data, index=dates)

# 停牌日处理：前复权填充+成交量置零
df['close'] = df['close'].ffill()
df['volume'] = [1e6, 1.2e6, 0, 9e5, 1.1e6]

# 避免未来数据泄露的20日均线计算
df['ma20'] = df['close'].shift(1).rolling(20, min_periods=5).mean()

import talib
from sklearn.preprocessing import RobustScaler

# 计算MACD指标（处理极端值）
df['macd'], df['signal'], _ = talib.MACD(df['close'], 
                                      fastperiod=12, 
                                      slowperiod=26, 
                                      signalperiod=9)

# 使用鲁棒标准化（避免离群值影响）
scaler = RobustScaler()
df[['macd_scaled', 'signal_scaled']] = scaler.fit_transform(df[['macd', 'signal']])

from sklearn.pipeline import Pipeline
from sklearn.ensemble import StackingClassifier
from xgboost import XGBClassifier

# 构建特征工程流水线
feature_pipe = Pipeline([
    ('fillna', SimpleImputer(strategy='median')),
    ('zscore', StandardScaler())
])

# 堆叠模型架构（XGBoost+逻辑回归）
estimators = [
    ('xgb', XGBClassifier(max_depth=3, 
                         learning_rate=0.1, 
                         subsample=0.8)),
    ('lr', LogisticRegression(penalty='l1', 
                            solver='saga'))
]

model = StackingClassifier(estimators=estimators,
                          final_estimator=RandomForestClassifier())

import backtrader as bt

class MLStrategy(bt.Strategy):
    params = (
        ('lookback', 20),  # 特征计算窗口
        ('probability_threshold', 0.65)  # 模型预测阈值
    )

    def __init__(self):
        self.model = load_pretrained_model()  # 加载预训练模型
        self.dataclose = self.datas[0].close
        
    def next(self):
        # 生成特征向量（滞后数据）
        features = np.array([
            self.dataclose.get(size=self.p.lookback),
            bt.indicators.RSI(self.data, period=14).get(size=self.p.lookback)
        ]).T
        
        # 模型实时预测
        prob = self.model.predict_proba(features[-1].reshape(1, -1))[0][1]
        
        if prob > self.p.probability_threshold and not self.position:
            self.order = self.buy(size=1000)
        elif prob < 0.35 and self.position:
            self.order = self.sell(size=1000)

import QuantLib as ql

# 构建波动率曲面（2025年3月数据）
calendar = ql.China()
calculation_date = ql.Date(19, 3, 2025)
ql.Settings.instance().evaluationDate = calculation_date

vol_surface = ql.BlackVarianceSurface(
    calculation_date,
    calendar,
    [ql.Date(19,6,2025), ql.Date(19,12,2025)],
    [0.8, 1.0, 1.2],
    [[0.25, 0.23], [0.21, 0.19], [0.18, 0.16]],
    ql.Actual365Fixed()
)

# 计算欧式看涨期权价格
option = ql.EuropeanOption(
    ql.PlainVanillaPayoff(ql.Option.Call, 100),
    ql.EuropeanExercise(ql.Date(19,6,2025))
)

process = ql.BlackScholesProcess(
    ql.QuoteHandle(ql.SimpleQuote(105)),
    ql.YieldTermStructureHandle(ql.FlatForward(calculation_date, 0.03, ql.Actual365Fixed())),
    ql.BlackVolTermStructureHandle(vol_surface)
)

engine = ql.AnalyticEuropeanEngine(process)
option.setPricingEngine(engine)
print(f"期权理论价格: {option.NPV():.2f}")