AI量化交易系统实战：从源码解析买卖信号生成机制

1次阅读

没有评论

共计 1802 个字符，预计需要花费 5 分钟才能阅读完成。

传统技术指标如 MACD、RSI 在高频交易中存在明显的滞后性问题。这些指标通常基于历史价格计算，例如：

MACD 依赖 12 日和 26 日 EMA 的差值，信号延迟至少 26 个周期
RSI 需要 14 日周期计算涨跌幅比率，无法捕捉瞬时市场变化

这种滞后性导致在快速变动的市场中，信号发出时行情已转向，最终影响策略的盈利能力。

针对高频交易场景，我们对比了三种主流方案：

统计套利
优点：逻辑简单，计算量小
缺点：依赖严格均值回归假设，在趋势市场中表现差
LSTM 时间序列预测
优点：能捕捉非线性时序特征
缺点：训练成本高，推理延迟大（通常在 10ms 以上）
强化学习
优点：可优化长期收益
缺点：需要复杂的环境模拟，实部署难度大

综合来看，我们选择 XGBoost 作为核心模型，因其在速度与精度间取得了良好平衡。

采用 tick 级滑窗处理，关键代码如下：

def create_features(ticks: np.ndarray, window_size=50) -> pd.DataFrame:
    """
    从 tick 数据生成特征矩阵
    :param ticks: shape=(N, 4)的数组，列为[timestamp, bid, ask, volume]
    :return: 特征 DataFrame
    """
    features = []
    for i in range(window_size, len(ticks)):
        window = ticks[i-window_size:i]
        # 计算统计特征
        mid_price = (window[:,1] + window[:,2])/2
        features.append({'mean_diff': np.mean(np.diff(mid_price)),
            'volatility': np.std(mid_price),
            'volume_ratio': window[-1,3]/np.mean(window[:,3])
        })
    return pd.DataFrame(features)

使用 ONNX 优化推理速度：

import onnxruntime as ort

class SignalGenerator:
    def __init__(self, model_path: str):
        self.session = ort.InferenceSession(model_path)

    def predict(self, features: np.ndarray) -> int:
        """
        生成交易信号
        :return: 1(买), -1(卖), 0(持有)
        """
        try:
            inputs = {'input': features.astype(np.float32)}
            pred = self.session.run(None, inputs)[0]
            return int(np.argmax(pred)) - 1  # 将 [0,1,2] 映射为[-1,0,1]
        except Exception as e:
            print(f"预测异常: {e}")
            return 0

采用双重检查锁模式避免信号冲突：

from threading import Lock

class TradeExecutor:
    def __init__(self):
        self.lock = Lock()
        self.last_signal = 0

    def execute(self, signal: int):
        with self.lock:
            if signal != self.last_signal:
                self._send_order(signal)
                self.last_signal = signal

采用滚动窗口验证防止过拟合：