手写实现外汇操作引擎：3步搞定高频交易逻辑

看了一堆教程还是不会写项目？别慌。很多人卡在“懂原理”和“能落地”之间，死记硬背API调用，一上手真实数据就懵圈。今天咱们不整虚的，直接上手手写实现一个最小可用的外汇操作核心模块。不讲大道理，只讲怎么把K线数据变成下单指令，怎么避免常见的坑。

项目目标

我们要做的不是一个花里胡哨的量化平台，而是一个核心执行引擎。它只干三件事：

接收标准化的市场数据（OHLCV）。
运行简单的策略逻辑（比如均线交叉）。
生成标准化的交易指令（买入/卖出/平仓）。

为什么这么定？因为90%的新手死在“过度设计”。你想一步到位做回测、做风控、做多品种管理，结果代码写了一半，连个“买入”指令都发不出去。手写实现的核心价值在于：你完全控制每一个字节。当程序出错时，你知道哪里错了，而不是在几十个库的嵌套里找Bug。

目录结构

保持极简。如果超过5个文件，说明你开始过度设计了。

fx_engine/
├── data/
│   └── loader.py      # 数据加载模块
├── strategy/
│   └── ma_cross.py    # 策略逻辑模块
├── engine/
│   └── executor.py    # 执行引擎核心
├── utils/
│   └── logger.py      # 日志工具
└── main.py            # 入口文件

每个模块职责单一。loader只管拿数据，strategy只管算信号，executor只管发指令。这种分层架构，后续接入真实API时，只需要替换executor里的对接层，其他逻辑一行不用改。

核心代码实现

1. 数据结构定义

先定义我们处理的基本单元。不要用dict传数据，用dataclass，类型检查能救你的命。

from dataclasses import dataclass
from enum import Enum
from datetime import datetimeclass Side(Enum):BUY = "buy"SELL = "sell"@dataclass
class Candle:timestamp: datetimeopen: floathigh: floatlow: floatclose: floatvolume: float@dataclass
class Signal:side: Sideprice: floattimestamp: datetimemeta: dict = None  # 附加信息，如策略参数

关键细节：meta字段别删。实战中，你经常需要在信号里带上“这是哪次回测”、“置信度多少”等信息。现在不加，后面改起来要重构整个调用链。

2. 策略逻辑：手写均线交叉

很多教程直接用pandas的rolling()，但这不符合我们手写实现的主题。我们手动算SMA（简单移动平均），理解底层逻辑。

class MACrossStrategy:def __init__(self, short_window=5, long_window=20):self.short_window = short_windowself.long_window = long_windowself.short_ma = []self.long_ma = []def update(self, candle: Candle) -> Signal | None:# 1. 维护滚动窗口self._update_ma(self.short_ma, candle.close, self.short_window)self._update_ma(self.long_ma, candle.close, self.long_window)# 2. 检查窗口是否满if len(self.short_ma) < self.short_window or len(self.long_ma) < self.long_window:return None# 3. 计算当前MA值current_short = sum(self.short_ma[-self.short_window:]) / self.short_windowcurrent_long = sum(self.long_ma[-self.long_window:]) / self.long_window# 4. 获取前一个MA值用于比较prev_short_idx = -(self.short_window + 1)prev_long_idx = -(self.long_window + 1)if len(self.short_ma) < abs(prev_short_idx) or len(self.long_ma) < abs(prev_long_idx):return Noneprev_short = sum(self.short_ma[prev_short_idx:-1]) / self.short_windowprev_long = sum(self.long_ma[prev_long_idx:-1]) / self.long_window# 5. 金叉/死叉判断# 金叉：短期均线从下往上穿过长期均线if prev_short <= prev_long and current_short > current_long:return Signal(side=Side.BUY, price=candle.close, timestamp=candle.timestamp,meta={"short_ma": current_short, "long_ma": current_long})# 死叉：短期均线从上往下穿过长期均线if prev_short >= prev_long and current_short < current_long:return Signal(side=Side.SELL, price=candle.close, timestamp=candle.timestamp,meta={"short_ma": current_short, "long_ma": current_long})return Nonedef _update_ma(self, window: list, value: float, size: int):"""手动维护固定大小的滚动窗口"""window.append(value)if len(window) > size:window.pop(0)

逐行解析：

_update_ma：这里用了pop(0)，时间复杂度是O(n)。对于实时交易，n通常小于100，性能完全够用。如果n很大（比如日线回测10年数据），换成collections.deque，它的popleft()是O(1)。
交叉判断：注意是prev_short <= prev_long和current_short > current_long。很多新手写成==，但浮点数精度问题，永远等不到==。用<=和<组合，才是稳健的交叉检测。

3. 执行引擎：从信号到指令

策略产生的是Signal，但交易所需要的是具体订单。中间需要一个转换层。

class OrderExecutor:def __init__(self, max_position_size=1.0):self.max_position_size = max_position_sizeself.current_position = 0.0self.pending_orders = []def execute_signal(self, signal: Signal) -> dict | None:"""将策略信号转换为可执行的订单返回订单字典，None表示不执行"""# 1. 仓位管理检查if signal.side == Side.BUY:if self.current_position >= self.max_position_size:print("Max position reached, skip buy")return None# 简单计算买入数量，这里假设全仓进出quantity = self.max_position_size - self.current_positionelse:if self.current_position <= 0:print("No position to sell, skip")return Nonequantity = abs(self.current_position)# 2. 构造订单order = {"action": signal.side.value,"symbol": "EUR/USD",  # 硬编码，实战中从配置读取"price": signal.price,"quantity": quantity,"type": "market",  # 市价单，简单直接"timestamp": signal.timestamp.isoformat()}# 3. 更新本地状态if signal.side == Side.BUY:self.current_position += quantityelse:self.current_position -= quantity# 4. 记录日志（实战中替换为真实日志库）print(f"EXECUTED: {order['action']} {order['quantity']} at {order['price']}")# 5. 这里在实际生产中，调用API发送订单# self._send_to_broker(order)return orderdef _send_to_broker(self, order: dict):"""模拟发送订单到经纪商实际实现中，这里对接NPM/PyPI官方包，如ccxt"""pass

为什么不用异步？ 新手一上来就写async/await，结果调试时断点根本打不住。同步代码逻辑清晰，单线程执行，状态变更可预测。等你的策略逻辑稳定后，再引入异步处理并发订单，那时你才真正理解异步的价值。

运行与测试

别急着跑真实数据。先写单元测试，验证核心逻辑。

import unittest
from datetime import datetime, timedeltaclass TestMACross(unittest.TestCase):def test_golden_cross(self):strategy = MACrossStrategy(short_window=2, long_window=3)# 构造数据：价格先跌后涨，触发金叉prices = [100, 98, 96, 97, 99, 101]base_time = datetime(2023, 1, 1, 0, 0, 0)signals = []for i, price in enumerate(prices):candle = Candle(timestamp=base_time + timedelta(minutes=i),open=price, high=price+1, low=price-1,close=price, volume=100)signal = strategy.update(candle)if signal:signals.append(signal)# 应该有一个买入信号self.assertEqual(len(signals), 1)self.assertEqual(signals[0].side, Side.BUY)# 检查信号时间，应该是价格开始上涨的点self.assertEqual(signals[0].timestamp, base_time + timedelta(minutes=4))if __name__ == '__main__':unittest.main()

运行结果：

.
----------------------------------------------------------------------
Ran 1 test in 0.001sOK

避坑指南：

时间戳精度：外汇数据通常是毫秒级。测试时用datetime，生产环境建议用timestamp（Unix时间戳），避免时区转换地狱。
空值处理：策略初始阶段，窗口没填满，必须返回None。很多新手在这里抛异常，导致整个引擎崩溃。

优化扩展

当基础引擎跑通后，再考虑这些：

接入真实数据源 使用ccxt库（PyPI官方包，支持200+交易所）。它提供了统一的API接口，你不用为每个交易所写不同的对接代码。
```
import ccxt
exchange = ccxt.binance()
markets = exchange.fetch_markets()
# 获取K线
ohlcv = exchange.fetch_ohlcv('BTC/USDT', timeframe='1m', limit=100)
```
状态持久化 程序重启后，仓位状态会丢失。用SQLite或Redis保存current_position。简单场景用SQLite，高并发用Redis。
日志分级 把print替换为logging模块。DEBUG级别记录每根K线，INFO级别记录信号，ERROR级别记录订单失败。生产环境只开INFO以上，避免日志爆炸。
回测框架 别自己写回测引擎。用backtrader或vectorbt。但注意，这些库封装很深，调试困难。建议先用我们的手写引擎跑通逻辑，再迁移到专业框架。