基于 AllTick API 的外汇量化交易策略

[Tempo Lam]

在这篇教程中，我将分享一个基于 AllTick API 获取实时外汇数据的量化交易策略。我们的交易逻辑基于 相对强弱指数 (RSI)，当 RSI 低于买入阈值时买入，高于卖出阈值时卖出，并且执行自动交易。

---

1. 代码实现概述

代码的核心结构包括：

1. 获取实时外汇数据：通过 AllTick API 获取外汇交易对的最新价格。
2. 计算 RSI 指标：基于最近 14 个价格计算 RSI，作为交易信号。
3. 执行交易决策：如果 RSI 低于买入阈值，则买入；如果 RSI 高于卖出阈值，则卖出。

---

2. 代码解析2.1 导入依赖和定义参数

import time
import requests
import json

# Replace 'your_token_here' with your actual AllTick API token
API_TOKEN = 'your_token_here'
FOREX_API_URL = 'https://quote.alltick.io/quote-b-ws-api'

# Trading parameters
TRADE_SYMBOL = "EUR/USD"
BUY_THRESHOLD = -0.2  # Example: Buy when RSI below this value
SELL_THRESHOLD = 0.8   # Example: Sell when RSI above this value

这里定义了 API 访问地址 和 交易参数：

• TRADE_SYMBOL: 交易的外汇对，例如 EUR/USD。
• BUY_THRESHOLD 和 SELL_THRESHOLD: RSI 指标的买入和卖出阈值。

---

2.2 获取实时外汇数据

def get_forex_data(symbol):
    query = json.dumps({"trace": "forex_query", "data": {"symbol_list": [{"code": symbol}]}})
    url = f'{FOREX_API_URL}/trade-tick?token={API_TOKEN}&query={query}'
   
    response = requests.get(url, headers={'Content-Type': 'application/json'})
   
    if response.status_code == 200:
        data = response.json()
        return data  # You need to parse the exact price from this response
    else:
        print(f"Error fetching data: {response.text}")
        return None

• 通过 HTTP GET 请求 向 AllTick API 发送请求，获取指定交易对的 最新价格数据。
• 如果请求成功，返回解析后的 JSON 数据；如果失败，则输出错误信息。

---

2.3 计算 RSI 指标

def calculate_rsi(prices, period=14):
    gains = [prices[i] - prices[i - 1] for i in range(1, len(prices)) if prices[i] > prices[i - 1]]
    losses = [-1 * (prices[i] - prices[i - 1]) for i in range(1, len(prices)) if prices[i] < prices[i - 1]]
   
    avg_gain = sum(gains) / period if gains else 0
    avg_loss = sum(losses) / period if losses else 0
   
    if avg_loss == 0:
        return 1  # Max RSI
   
    rs = avg_gain / avg_loss
    return 1 - (1 / (1 + rs))

RSI 指标用于衡量市场的超买或超卖情况：

1. 计算最近 14 根 K 线的 平均涨幅 和 平均跌幅。
2. 使用 RSI 公式 1 - (1 / (1 + rs)) 计算 RSI 值。
3. 当 RSI 接近 0 时，表示超卖；接近 1 时，表示超买。

---

2.4 执行交易

def execute_trade(action, symbol):
    print(f"Executing {action} order for {symbol}")
    # Here you would integrate with a broker API to place an order

这里的 execute_trade() 函数是一个 占位函数 (mock function)，用于模拟交易执行。实际使用时，可以连接 MT4/MT5、Binance、Interactive Brokers (IB) 等交易所 API 进行真实交易。

---

2.5 交易主循环

price_history = []
while True:
    forex_data = get_forex_data(TRADE_SYMBOL)
   
    if forex_data:
        price = forex_data["data"]["price"]  # Adjust based on actual response structure
        price_history.append(price)
       
        if len(price_history) > 14:
            price_history.pop(0)
            rsi = calculate_rsi(price_history)
           
            if rsi < BUY_THRESHOLD:
                execute_trade("BUY", TRADE_SYMBOL)
            elif rsi > SELL_THRESHOLD:
                execute_trade("SELL", TRADE_SYMBOL)
   
    time.sleep(10)  # Adjust frequency as needed

• 维护一个 price_history 列表存储最近 14 个价格。
• 每隔 10 秒 轮询一次最新的外汇价格。
• 计算 RSI 并根据 买入/卖出阈值 触发交易信号。

---

3. 代码改进方向

目前代码是一个 最小可运行版本，可以进一步优化：

1. 增加交易日志，记录交易执行情况。
2. 优化 RSI 计算，使用 Pandas 计算 RSI，更高效。
3. 连接真实交易账户，比如 Binance、MT4、IBKR 进行自动化交易。

[Tempo Lam]

2025年3月6日，对代码进一步做优化：

1. 新增日志记录功能

在execute_trade函数中添加日志记录逻辑，并修改函数参数以获取完整的交易上下文信息。

# 新增日志文件路径常量（添加在参数定义区）
LOG_FILE = "trading_log.csv"  # 使用CSV格式便于后续分析

def execute_trade(action, symbol, price, rsi, buy_thresh, sell_thresh):
    """执行交易并记录日志"""
    timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
   
    # 构造日志条目（CSV格式）
    log_entry = f"{timestamp},{action},{symbol},{price:.5f},{rsi:.4f},{buy_thresh},{sell_thresh}\n"
   
    # 如果是首次运行，写入CSV头部
    try:
        with open(LOG_FILE, "x") as f:
            f.write("timestamp,action,symbol,price,rsi,buy_threshold,sell_threshold\n")
    except FileExistsError:
        pass
   
    # 追加日志记录
    with open(LOG_FILE, "a") as f:
        f.write(log_entry)
   
    print(f"[{timestamp}] {action} {symbol} @ {price:.5f} | RSI: {rsi:.4f}")

2. 修改主循环的交易触发逻辑

将必要的交易参数传递给execute_trade函数：

# 修改后的主循环交易触发部分
if len(price_history) >= 14:  # 修改为>=14确保足够数据点
    current_prices = price_history[-14:]  # 始终取最近14个数据点
    rsi = calculate_rsi(current_prices)
   
    if rsi < BUY_THRESHOLD:
        execute_trade("BUY", TRADE_SYMBOL, price, rsi, BUY_THRESHOLD, SELL_THRESHOLD)
    elif rsi > SELL_THRESHOLD:
        execute_trade("SELL", TRADE_SYMBOL, price, rsi, BUY_THRESHOLD, SELL_THRESHOLD)

3. 优化建议说明

• 日志采用CSV格式存储，方便后续用Excel/Pandas进行分析

• 首次运行会自动创建带标题行的日志文件

• 记录完整的交易上下文信息（时间、操作、品种、成交价、RSI值、阈值参数）

• 价格显示精度调整为5位小数（适应外汇报价惯例）

• RSI值保留4位小数保证精度

4. 示例日志输出

timestamp,action,symbol,price,rsi,buy_threshold,sell_threshold
2023-10-05 14:30:00,BUY,EUR/USD,1.12345,0.1890,-0.2,0.8
2023-10-05 14:45:15,SELL,EUR/USD,1.12560,0.8105,-0.2,0.8

这些修改会生成结构化交易日志，包含交易决策的所有关键参数，方便后续进行：

1. 交易绩效分析

2. 策略参数优化

3. 异常交易排查

4. 市场环境回查