在金融科技快速发展的今天机器学习与量化交易的结合正成为投资领域的重要趋势。佐治亚理工学院等顶尖学府的研究表明通过算法模型分析市场数据能够有效识别潜在收益机会。本文将基于实际案例完整拆解如何利用机器学习技术构建股票交易策略从数据准备到模型部署的全流程实践。无论你是金融从业者希望提升技术能力还是开发者想要进入量化领域本文都将提供可直接复用的代码示例和工程经验。我们将重点介绍特征工程、模型选择、回溯测试等核心环节帮助你理解机器学习如何赋能股票交易决策。1. 量化交易与机器学习基础概念1.1 什么是量化交易量化交易是通过数学模型和计算机程序来执行投资决策的交易方式。与传统的主观交易不同量化交易依赖于数据分析和算法模型力求消除情绪波动对投资决策的影响。核心思想是利用历史数据发现规律并基于这些规律制定交易策略。量化交易的优势主要体现在三个方面首先它能够处理海量数据发现人眼难以识别的复杂模式其次交易执行速度快能够在毫秒级别响应市场变化最后策略回测方便可以在实盘前验证策略的有效性。1.2 机器学习在量化交易中的应用机器学习为量化交易提供了强大的预测能力。在股票交易场景中机器学习模型主要用于价格预测、趋势判断、风险控制等方面。与传统的技术分析相比机器学习能够处理更多维度的数据并自动学习特征之间的非线性关系。常用的机器学习算法包括回归模型用于预测连续值如股票价格分类模型用于判断涨跌方向时间序列模型处理具有时间相关性的金融数据强化学习优化交易策略的长期收益1.3 量化交易的基本流程一个完整的量化交易系统通常包含以下步骤数据获取与清洗收集历史价格、成交量、财务数据等特征工程从原始数据中提取有预测能力的特征模型训练使用历史数据训练机器学习模型回测验证在历史数据上模拟交易评估策略性能实盘部署将验证通过的策略投入实际交易2. 环境准备与数据说明2.1 开发环境配置构建机器学习量化交易系统需要以下环境准备# 创建Python虚拟环境 python -m venv quant_env source quant_env/bin/activate # Linux/Mac # quant_env\Scripts\activate # Windows # 安装必要依赖包 pip install scikit-learn pandas lightgbm numpy keras tensorflow关键库的作用说明pandas数据处理和分析numpy数值计算scikit-learn机器学习算法lightgbm梯度提升决策树框架tensorflow/keras深度学习框架2.2 数据来源与结构本案例使用A股中证500指数成分股从2012年到2018年的历史数据包含以下字段# 数据字段示例 import pandas as pd data_structure { date: 交易日期, code: 股票代码, open: 开盘价, close: 收盘价, high: 最高价, low: 最低价, volume: 成交量 } # 数据预处理要点 def data_quality_check(df): 数据质量检查函数 # 检查缺失值 missing_ratio df.isnull().sum() / len(df) print(缺失值比例:) print(missing_ratio) # 检查重复值 duplicates df.duplicated().sum() print(f重复记录数: {duplicates}) # 检查价格合理性 price_check df[(df[high] df[low]) | (df[close] df[high]) | (df[close] df[low])] print(f价格异常记录数: {len(price_check)})2.3 数据预处理流程金融数据预处理是建模成功的关键需要特别注意以下几点class DataPreprocessor: def __init__(self): self.scalers {} def handle_missing_values(self, df): 处理缺失值 # 前向填充 df.fillna(methodffill, inplaceTrue) # 如果仍有缺失使用后向填充 df.fillna(methodbfill, inplaceTrue) return df def remove_anomalies(self, df): 去除异常值 # 去除价格为零或负值的记录 df df[(df[open] 0) (df[close] 0) (df[high] 0) (df[low] 0)] # 去除成交量异常大的记录超过3倍标准差 volume_mean df[volume].mean() volume_std df[volume].std() df df[df[volume] volume_mean 3 * volume_std] return df def calculate_returns(self, df): 计算收益率 df[daily_return] df[close].pct_change() df[log_return] np.log(df[close] / df[close].shift(1)) return df3. 特征工程与数据标准化3.1 时间序列特征构建基于60天历史数据构建特征是本案例的核心def create_time_series_features(df, window_size60): 创建时间序列特征 features [] for i in range(1, window_size 1): # 价格相对变化特征 df[fopen_ratio_{i}] df[open].shift(i) / df[close] df[fclose_ratio_{i}] df[close].shift(i) / df[close] df[fhigh_ratio_{i}] df[high].shift(i) / df[close] df[flow_ratio_{i}] df[low].shift(i) / df[close] # 成交量特征对数变换 df[fvolume_ratio_{i}] np.log(df[volume].shift(i) / df[volume]) # 技术指标特征 if i 5: # 需要足够的数据计算移动平均 df[fsma_5_{i}] df[close].shift(i).rolling(5).mean() / df[close] df[fsma_10_{i}] df[close].shift(i).rolling(10).mean() / df[close] # 添加波动率特征 df[volatility_20] df[daily_return].rolling(20).std() return df def prepare_features(raw_data): 特征工程主函数 print(开始特征工程...) # 数据清洗 cleaned_data raw_data.dropna() # 创建特征 featured_data create_time_series_features(cleaned_data) # 去除包含NaN的行由于shift操作产生 final_data featured_data.dropna() print(f特征工程完成最终数据形状: {final_data.shape}) return final_data3.2 特征标准化处理金融数据标准化是模型训练的重要环节from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split class FeatureStandardizer: def __init__(self): self.scaler StandardScaler() self.feature_columns [] def prepare_features_for_training(self, df): 准备训练特征 # 选择特征列排除目标变量和标识列 feature_columns [col for col in df.columns if col not in [date, code, target]] self.feature_columns feature_columns X df[feature_columns].values y df[target].values if target in df.columns else None return X, y, feature_columns def split_dataset(self, df, test_size0.2): 划分训练集和测试集 # 按时间顺序划分避免未来信息泄露 split_point int(len(df) * (1 - test_size)) train_data df.iloc[:split_point] test_data df.iloc[split_point:] print(f训练集大小: {len(train_data)}) print(f测试集大小: {len(test_data)}) return train_data, test_data3.3 目标变量定义在量化交易中目标变量的定义直接影响模型效果def define_target_variable(df, market_index_df): 定义目标变量相对大盘的超额收益 # 计算个股收益率 df[stock_return] df[close].pct_change().shift(-1) # 下一日收益率 # 计算大盘收益率需要大盘指数数据 market_index_df[market_return] market_index_df[close].pct_change().shift(-1) # 合并数据 merged_df pd.merge(df, market_index_df[[date, market_return]], ondate, howleft) # 计算超额收益 merged_df[excess_return] merged_df[stock_return] - merged_df[market_return] # 定义分类目标是否跑赢大盘 merged_df[target_class] (merged_df[excess_return] 0).astype(int) # 定义回归目标超额收益的具体数值 merged_df[target_regression] merged_df[excess_return] return merged_df4. 机器学习模型构建与训练4.1 梯度提升决策树模型LightGBM是处理表格数据的强大工具import lightgbm as lgb from sklearn.metrics import mean_squared_error, accuracy_score class GBDTModel: def __init__(self): self.model None self.feature_importance None def train(self, X_train, y_train, X_valNone, y_valNone): 训练GBDT模型 # 创建数据集 train_data lgb.Dataset(X_train, labely_train) if X_val is not None and y_val is not None: val_data lgb.Dataset(X_val, labely_val, referencetrain_data) # 参数设置 params { objective: regression, metric: rmse, num_leaves: 31, learning_rate: 0.05, feature_fraction: 0.9, bagging_fraction: 0.8, bagging_freq: 5, verbose: 0 } # 训练模型 self.model lgb.train( params, train_data, num_boost_round1000, valid_sets[train_data, val_data] if X_val is not None else [train_data], callbacks[lgb.early_stopping(50), lgb.log_evaluation(50)] ) return self.model def predict(self, X): 模型预测 if self.model is None: raise ValueError(模型尚未训练) return self.model.predict(X) def evaluate(self, X_test, y_test): 模型评估 predictions self.predict(X_test) mse mean_squared_error(y_test, predictions) rmse np.sqrt(mse) print(f测试集MSE: {mse:.6f}) print(f测试集RMSE: {rmse:.6f}) return predictions, rmse # 使用示例 def train_gbdt_model(): 完整的GBDT训练流程 # 数据准备 df load_and_preprocess_data() X, y, feature_names prepare_features_for_training(df) # 数据划分 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42 ) # 模型训练 gbdt_model GBDTModel() gbdt_model.train(X_train, y_train, X_test, y_test) # 模型评估 predictions, rmse gbdt_model.evaluate(X_test, y_test) return gbdt_model, predictions, rmse4.2 神经网络模型对于复杂模式识别神经网络可能提供更好的效果import tensorflow as tf from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Dense, Dropout, BatchNormalization from tensorflow.keras.optimizers import Adam from tensorflow.keras.callbacks import EarlyStopping, ReduceLROnPlateau class NeuralNetworkModel: def __init__(self, input_dim): self.model self._build_model(input_dim) self.history None def _build_model(self, input_dim): 构建神经网络结构 model Sequential([ Dense(512, activationrelu, input_shape(input_dim,)), BatchNormalization(), Dropout(0.3), Dense(256, activationrelu), BatchNormalization(), Dropout(0.3), Dense(128, activationrelu), BatchNormalization(), Dropout(0.2), Dense(64, activationrelu), Dropout(0.2), Dense(1, activationlinear) # 回归任务 ]) optimizer Adam(learning_rate0.001) model.compile(optimizeroptimizer, lossmse, metrics[mae]) return model def train(self, X_train, y_train, X_val, y_val, epochs100, batch_size32): 训练神经网络 callbacks [ EarlyStopping(patience15, restore_best_weightsTrue), ReduceLROnPlateau(factor0.5, patience10) ] self.history self.model.fit( X_train, y_train, validation_data(X_val, y_val), epochsepochs, batch_sizebatch_size, callbackscallbacks, verbose1 ) return self.history def predict(self, X): 模型预测 return self.model.predict(X).flatten() def plot_training_history(self): 绘制训练历史 import matplotlib.pyplot as plt plt.figure(figsize(12, 4)) plt.subplot(1, 2, 1) plt.plot(self.history.history[loss], label训练损失) plt.plot(self.history.history[val_loss], label验证损失) plt.title(模型损失) plt.xlabel(Epoch) plt.ylabel(Loss) plt.legend() plt.subplot(1, 2, 2) plt.plot(self.history.history[mae], label训练MAE) plt.plot(self.history.history[val_mae], label验证MAE) plt.title(平均绝对误差) plt.xlabel(Epoch) plt.ylabel(MAE) plt.legend() plt.tight_layout() plt.show() # 神经网络训练示例 def train_nn_model(X_train, y_train, X_val, y_val): 神经网络训练流程 nn_model NeuralNetworkModel(input_dimX_train.shape[1]) print(开始训练神经网络...) history nn_model.train(X_train, y_train, X_val, y_val) # 绘制训练过程 nn_model.plot_training_history() return nn_model4.3 模型集成与优化结合多个模型可以提升预测稳定性from sklearn.ensemble import VotingRegressor from sklearn.linear_model import LinearRegression class EnsembleModel: def __init__(self): self.models {} self.ensemble_model None def add_model(self, name, model): 添加基础模型 self.models[name] model def train_ensemble(self, X_train, y_train): 训练集成模型 # 创建投票回归器 estimators [(name, model) for name, model in self.models.items()] self.ensemble_model VotingRegressor(estimatorsestimators) self.ensemble_model.fit(X_train, y_train) return self.ensemble_model def predict(self, X): 集成预测 if self.ensemble_model is None: raise ValueError(集成模型尚未训练) return self.ensemble_model.predict(X) def get_model_weights(self, X_val, y_val): 根据验证集性能计算模型权重 predictions {} scores {} for name, model in self.models.items(): pred model.predict(X_val) score -mean_squared_error(y_val, pred) # 负MSE越大越好 predictions[name] pred scores[name] score # 标准化权重 total_score sum(scores.values()) weights {name: score/total_score for name, score in scores.items()} return weights # 模型集成示例 def create_ensemble_predictor(): 创建集成预测器 ensemble EnsembleModel() # 添加不同模型 ensemble.add_model(gbdt, gbdt_model) ensemble.add_model(nn, nn_model) ensemble.add_model(linear, LinearRegression()) return ensemble5. 回溯测试与策略评估5.1 回溯测试框架实现回溯测试是量化交易策略验证的核心环节class BacktestEngine: def __init__(self, initial_capital100000, transaction_cost0.001): self.initial_capital initial_capital self.transaction_cost transaction_cost # 交易成本率 self.portfolio_history [] def run_backtest(self, predictions, actual_prices, dates): 运行回溯测试 current_cash self.initial_capital current_positions {} # 股票代码: 持仓数量 portfolio_value current_cash daily_returns [] # 按日期循环 unique_dates sorted(set(dates)) for i, date in enumerate(unique_dates): # 获取当日数据 date_mask dates date date_predictions predictions[date_mask] date_actual_prices actual_prices[date_mask] date_stock_codes stock_codes[date_mask] # 交易逻辑买入预测涨幅前50的股票 top_indices np.argsort(date_predictions)[-50:] # 选择预测值最高的50只 stocks_to_buy date_stock_codes[top_indices] buy_prices date_actual_prices[top_indices] # 卖出不在买入列表中的持仓 positions_to_sell [code for code in current_positions.keys() if code not in stocks_to_buy] # 执行卖出 for stock_code in positions_to_sell: sell_price date_actual_prices[date_stock_codes stock_code][0] current_cash current_positions[stock_code] * sell_price * (1 - self.transaction_cost) del current_positions[stock_code] # 计算可用资金平均分配到要买入的股票 available_cash current_cash cash_per_stock available_cash / len(stocks_to_buy) if stocks_to_buy else 0 # 执行买入 for stock_code, buy_price in zip(stocks_to_buy, buy_prices): if stock_code not in current_positions: # 计算可买数量 shares_to_buy cash_per_stock // (buy_price * (1 self.transaction_cost)) if shares_to_buy 0: current_positions[stock_code] shares_to_buy current_cash - shares_to_buy * buy_price * (1 self.transaction_cost) # 计算当日 portfolio 价值 position_value 0 for stock_code, shares in current_positions.items(): current_price date_actual_prices[date_stock_codes stock_code][0] position_value shares * current_price daily_portfolio_value current_cash position_value # 计算日收益率 if i 0: daily_return (daily_portfolio_value - prev_portfolio_value) / prev_portfolio_value daily_returns.append(daily_return) prev_portfolio_value daily_portfolio_value # 记录历史 self.portfolio_history.append({ date: date, portfolio_value: daily_portfolio_value, cash: current_cash, positions: current_positions.copy() }) return np.array(daily_returns) def calculate_performance_metrics(self, daily_returns, risk_free_rate0.02/252): 计算性能指标 # 基本统计量 mean_return np.mean(daily_returns) * 100 # 转换为百分比 std_return np.std(daily_returns) * 100 total_return (self.portfolio_history[-1][portfolio_value] / self.initial_capital - 1) * 100 # 年化收益 annual_return (1 mean_return/100) ** 252 - 1 # 夏普比率 excess_returns daily_returns - risk_free_rate sharpe_ratio np.mean(excess_returns) / np.std(excess_returns) * np.sqrt(252) # 最大回撤 portfolio_values [day[portfolio_value] for day in self.portfolio_history] peak np.maximum.accumulate(portfolio_values) drawdown (portfolio_values - peak) / peak max_drawdown np.min(drawdown) * 100 metrics { 日均收益率(%): mean_return, 收益波动率(%): std_return, 总收益率(%): total_return, 年化收益率(%): annual_return * 100, 夏普比率: sharpe_ratio, 最大回撤(%): max_drawdown } return metrics5.2 策略性能可视化可视化帮助理解策略表现import matplotlib.pyplot as plt import seaborn as sns def plot_strategy_performance(backtest_engine, benchmark_returnsNone): 绘制策略性能图表 portfolio_values [day[portfolio_value] for day in backtest_engine.portfolio_history] dates [day[date] for day in backtest_engine.portfolio_history] plt.figure(figsize(15, 10)) # 1. 资产净值曲线 plt.subplot(2, 2, 1) plt.plot(dates, portfolio_values, label策略净值, linewidth2) if benchmark_returns is not None: benchmark_values [backtest_engine.initial_capital] for ret in benchmark_returns: benchmark_values.append(benchmark_values[-1] * (1 ret)) plt.plot(dates, benchmark_values[1:], label基准净值, linestyle--) plt.title(策略净值曲线) plt.xlabel(日期) plt.ylabel(净值) plt.legend() plt.grid(True) # 2. 每日收益率分布 plt.subplot(2, 2, 2) daily_returns np.diff(portfolio_values) / portfolio_values[:-1] plt.hist(daily_returns, bins50, alpha0.7, edgecolorblack) plt.title(每日收益率分布) plt.xlabel(收益率) plt.ylabel(频数) # 3. 回撤曲线 plt.subplot(2, 2, 3) peak np.maximum.accumulate(portfolio_values) drawdown (portfolio_values - peak) / peak plt.fill_between(dates, drawdown, 0, alpha0.3, colorred) plt.plot(dates, drawdown, colorred, linewidth1) plt.title(回撤曲线) plt.xlabel(日期) plt.ylabel(回撤) plt.grid(True) # 4. 月度收益热力图 plt.subplot(2, 2, 4) # 将日期转换为月份 monthly_returns [] current_month dates[0].month month_return 0 for i, date in enumerate(dates): if date.month ! current_month: monthly_returns.append(month_return) current_month date.month month_return 0 if i 0: daily_ret (portfolio_values[i] - portfolio_values[i-1]) / portfolio_values[i-1] month_return (1 month_return) * (1 daily_ret) - 1 # 创建热力图数据 years sorted(set([d.year for d in dates])) months range(1, 13) heatmap_data np.full((len(years), len(months)), np.nan) # 填充数据 for i, year in enumerate(years): for j, month in enumerate(months): try: idx [d.year year and d.month month for d in dates].index(True) heatmap_data[i, j] monthly_returns[idx] * 100 except ValueError: continue sns.heatmap(heatmap_data, annotTrue, fmt.1f, cmapRdYlGn, center0, xticklabelsmonths, yticklabelsyears) plt.title(月度收益率热力图(%)) plt.tight_layout() plt.show() # 完整的回溯测试流程 def complete_backtest_analysis(model, test_data): 完整的回溯测试分析 # 模型预测 predictions model.predict(test_data[feature_columns]) # 运行回溯测试 backtester BacktestEngine(initial_capital100000) daily_returns backtester.run_backtest( predictions, test_data[close].values, test_data[date].values ) # 计算性能指标 metrics backtester.calculate_performance_metrics(daily_returns) # 可视化结果 plot_strategy_performance(backtester) print(策略性能指标:) for metric, value in metrics.items(): print(f{metric}: {value:.4f}) return metrics, backtester6. 风险控制与模型优化6.1 风险管理策略有效的风险控制是量化交易成功的关键class RiskManager: def __init__(self, max_position_size0.1, max_drawdown0.2, stop_loss0.05): self.max_position_size max_position_size # 单票最大仓位 self.max_drawdown max_drawdown # 最大回撤阈值 self.stop_loss stop_loss # 单票止损比例 def position_sizing(self, predictions, current_portfolio_value): 头寸规模管理 # 根据预测置信度调整仓位 confidence_scores self.calculate_confidence(predictions) # 计算每只股票的仓位上限 max_position_value current_portfolio_value * self.max_position_size # 根据置信度分配仓位 position_weights confidence_scores / np.sum(confidence_scores) position_sizes position_weights * max_position_value * len(predictions) return position_sizes def calculate_confidence(self, predictions): 计算预测置信度 # 基于预测值的分布计算置信度 prediction_std np.std(predictions) if prediction_std 0: return np.ones_like(predictions) # 使用z-score的绝对值作为置信度基础 z_scores np.abs((predictions - np.mean(predictions)) / prediction_std) confidence 1 / (1 np.exp(-z_scores)) # Sigmoid变换 return confidence def check_stop_loss(self, current_positions, purchase_prices, current_prices): 检查止损条件 actions {} for stock_code, purchase_price in purchase_prices.items(): current_price current_prices.get(stock_code, purchase_price) drawdown (current_price - purchase_price) / purchase_price if drawdown -self.stop_loss: actions[stock_code] SELL # 触发止损 return actions def monitor_portfolio_risk(self, portfolio_history): 监控组合风险 current_value portfolio_history[-1][portfolio_value] peak_value max([day[portfolio_value] for day in portfolio_history]) current_drawdown (current_value - peak_value) / peak_value if current_drawdown -self.max_drawdown: return REDUCE_RISK # 需要降低风险暴露 return NORMAL # 集成风险管理的交易引擎 class RiskAwareBacktestEngine(BacktestEngine): def __init__(self, risk_manager, **kwargs): super().__init__(**kwargs) self.risk_manager risk_manager self.purchase_prices {} # 记录买入价格 def execute_trades_with_risk_control(self, predictions, prices, dates): 带风险控制的交易执行 # 检查组合层面风险 risk_status self.risk_manager.monitor_portfolio_risk(self.portfolio_history) if risk_status REDUCE_RISK: # 减少风险暴露降低仓位 predictions self.adjust_predictions_for_risk(predictions) # 检查个股止损 stop_loss_actions self.risk_manager.check_stop_loss( self.current_positions, self.purchase_prices, current_prices ) # 执行止损 for stock_code, action in stop_loss_actions.items(): if action SELL: self.execute_sell(stock_code, current_prices[stock_code]) # 继续正常交易逻辑 return super().execute_trades(predictions, prices, dates)6.2 模型性能优化持续优化模型提升预测能力class ModelOptimizer: def __init__(self, base_model, param_grid): self.base_model base_model self.param_grid param_grid self.best_params None self.best_score None def optimize_hyperparameters(self, X_train, y_train, X_val, y_val, cv5): 超参数优化 from sklearn.model_selection import RandomizedSearchCV search RandomizedSearchCV( self.base_model, self.param_grid, n_iter50, cvcv, scoringneg_mean_squared_error, n_jobs-1, random_state42 ) search.fit(X_train, y_train) self.best_params search.best_params_ self.best_score search.best_score_ print(f最佳参数: {self.best_params}) print(f最佳分数: {-self.best_score:.6f}) return search.best_estimator_ def feature_importance_analysis(self, model, feature_names, top_n20): 特征重要性分析 if hasattr(model, feature_importances_): importances model.feature_importances_ else: # 对于神经网络等模型使用排列重要性 importances self.calculate_permutation_importance(model, X_val, y_val) # 排序并选择最重要的特征 indices np.argsort(importances)[::-1] plt.figure(figsize(10, 8)) plt.title(特征重要性) plt.barh(range(top_n), importances[indices[:top_n]][::-1]) plt.yticks(range(top_n), [feature_names[i] for i in indices[:top_n]][::-1]) plt.tight_layout() plt.show() return indices, importances def calculate_permutation_importance(self, model, X, y, n_repeats10): 计算排列重要性 from sklearn.inspection import permutation_importance result permutation_importance( model, X, y, n_repeatsn_repeats, random_state42 ) return result.importances_mean # 模型优化示例 def optimize_trading_model(): 模型优化流程 # 定义参数网格 param_grid { n_estimators: [100, 200, 500], learning_rate: [0.01, 0.05, 0.1], max_depth: [3, 5, 7], subsample: [0.8, 0.9, 1.0] } # 创建优化器 optimizer ModelOptimizer( base_modellgb.LGBMRegressor(), param_gridparam_grid ) # 执行优化 best_model optimizer.optimize_hyperparameters(X_train, y_train, X_val, y_val) # 分析特征重要性 feature_indices, importances optimizer.feature_importance_analysis( best_model, feature_names ) return best_model, feature_indices, importances7. 实盘部署考虑与生产环境建议7.1 系统架构设计生产环境中的量化交易系统需要更高的可靠性和性能class ProductionTradingSystem: def __init__(self, model, data_fetcher, risk_manager): self.model model self.data_fetcher data_fetcher self.risk_manager risk_manager self.portfolio_manager PortfolioManager() self.order_executor OrderExecutor() def run_daily_trading(self): 每日交易流程 try: # 1. 获取最新数据 latest_data self.data_fetcher.get_latest_market_data() # 2. 数据预处理 processed_data self.preprocess_data(latest_data) # 3. 模型预测 predictions self.model.predict(processed_data) # 4. 风险检查 if not self.risk_manager.approve_trading(predictions): print(风险检查未通过暂停今日交易) return # 5. 生成交易信号 trading_signals self.generate_signals(predictions, processed_data) # 6. 执行交易 execution_results self.order_executor.execute_orders(trading_signals) # 7. 更新组合记录 self.portfolio_manager.update_portfolio(execution_results) # 8. 记录日志 self.log_trading_activity(execution_results) except Exception as e: print(f交易执行错误: {e}) self.handle_trading_error(e) def preprocess_data(self, raw_data): 生产环境数据预处理 # 确保与训练时相同的预处理流程 processed raw_data.copy() # 数据清洗 processed processed.dropna() # 特征工程