从理论到实践：我的金融机器学习项目开发笔记

# 爬取东方财富网个股历史数据（需替换实际URL和Cookie）
import requests
import pandas as pd
from lxml import etree

def crawl_stock_data(code, pages=5):
    data = []
    for i in range(1, pages+1):
        url = f'http://quote.eastmoney.com/{code}.html?page={i}'
        headers = {'User-Agent': 'Mozilla/5.0', 'Cookie': 'your_cookie'}
        response = requests.get(url, headers=headers)
        html = etree.HTML(response.text)
        # 解析开盘价、收盘价、成交量等字段
        # ...（具体解析逻辑）
        data.append(row)
    return pd.DataFrame(data)

# 示例：获取贵州茅台数据
df = crawl_stock_data('SH600519')

# 处理缺失值与异常值
df.fillna(method='ffill', inplace=True)  # 前向填充
df = df[(df['换手率'] > 0) & (df['成交量'] < df['成交量'].quantile(0.95))]  # 剔除异常交易

# 构建标签：未来5日涨幅是否超过3%
df['label'] = (df['收盘价'].shift(-5) / df['收盘价'] - 1 > 0.03).astype(int)
df.dropna(inplace=True)

import talib

# 计算MACD
df['macd'], df['macdsignal'], df['macdhist'] = talib.MACD(df['收盘价'])

# 计算布林带
df['upper'], df['middle'], df['lower'] = talib.BBANDS(df['收盘价'], timeperiod=20)

# 标准化处理（避免量纲差异）
from sklearn.preprocessing import StandardScaler
scaler = StandardScaler()
X = scaler.fit_transform(df[['macd', 'upper', '成交量']])

import lightgbm as lgb
from sklearn.model_selection import train_test_split

# 数据集划分
X_train, X_test, y_train, y_test = train_test_split(X, df['label'], test_size=0.2)

# 定义模型参数
params = {
    'objective': 'binary',
    'metric': 'auc',
    'num_leaves': 31,
    'learning_rate': 0.05,
    'feature_fraction': 0.8
}

# 训练与评估
model = lgb.train(params, 
                 lgb.Dataset(X_train, label=y_train),
                 valid_sets=[lgb.Dataset(X_test, label=y_test)],
                 num_boost_round=1000,
                 early_stopping_rounds=50)

from sklearn.model_selection import GridSearchCV

param_grid = {
    'num_leaves': [15, 31, 63],
    'min_data_in_leaf': [20, 50, 100]
}

gbm = lgb.LGBMClassifier()
grid = GridSearchCV(gbm, param_grid, cv=3)
grid.fit(X_train, y_train)
print(f'最优参数：{grid.best_params_}')

# 简易回测框架（参考Backtrader设计思想）[47](@ref)
class Backtest:
    def __init__(self, data, model):
        self.data = data
        self.model = model
        self.cash = 1000000  # 初始资金
    
    def run(self):
        for i in range(5, len(self.data)):
            features = self._extract_features(i)
            prob = self.model.predict_proba([features])[0][1]
            if prob > 0.7:  # 置信度高时买入
                self._execute_order('buy', i)
            elif prob < 0.3:  # 置信度低时卖出
                self._execute_order('sell', i)
    
    def _extract_features(self, idx):
        # 提取当前时间步的特征
        return self.data.iloc[idx-5:idx][['macd', 'RSI', '成交量']].mean()