如何解决Python币安交易机器人执行交易时的签名无效错误？

核心问题 #

你遇到的-1022错误是Binance API签名验证失败，根源是签名生成逻辑与请求参数不匹配，且签名传递方式不符合API规范。

具体错误点及修正 #

1. 签名传递方式错误 #

Binance要求签名作为请求参数的一部分，而非放在请求头X-MBX-SIGNATURE里，原代码的签名位置不符合规范。

2. 参数不一致 #

生成签名用的query_string包含了timestamp，但实际请求的params字典未加入该字段，导致Binance收到的参数与签名计算的参数不一致，验证失败。

3. 未定义变量引用 #

execute_trade函数里引用了未定义的data和current_time变量，会导致交易成功后触发额外报错。

4. 主逻辑重复冗余 #

代码同时存在独立无限循环和main函数，但main函数未包含交易判断逻辑，实际运行时无法触发交易。

完整修正后的代码 #

import talib
import pandas as pd
import time
import requests
import logging
import hmac
from datetime import datetime
import hashlib
from hashlib import sha256
import API

# 初始化日志
logging.basicConfig(filename='trading_bot.log', level=logging.INFO)

# Binance API端点
api_endpoint = "https://api.binance.com/api/v3/klines"
api_trade = "https://api.binance.com/api/v3/order"
api_balance = "https://api.binance.com/api/v3/account"

# API密钥
api_key = API.API_KEY
api_secret = API.API_SECRET

# 配置参数
interval = "1m"
symbol = "DOGEBUSD"
stop_loss = 0.5  # 止损百分比
amount = 10
timeframe = 30
bb_window = 20
bb_deviation = 2

def hashing(query_string, secret):
    return hmac.new(
        secret.encode("utf-8"), query_string.encode("utf-8"), hashlib.sha256
    ).hexdigest()

# 获取Binance历史数据
def fetch_binance_data():
    try:
        params = {"symbol": symbol, "interval": interval}
        data = requests.get(api_endpoint, params=params).json()
        df = pd.DataFrame(data, columns=["timestamp", "open", "high", "low", "close", "volume", "close_time", "quote_asset_volume", "number_of_trades", "taker_buy_base_asset_volume", "taker_buy_quote_asset_volume", "ignore"])
        df["timestamp"] = pd.to_datetime(df["timestamp"], unit='ms')
        df.set_index("timestamp", inplace=True)
        df = df.astype(float)
    except Exception as e:
        logging.error("Error fetching data from Binance: %s", e)
        df = pd.DataFrame()
    return df

# 检查账户余额
def check_balance():
    try:
        params = {"timestamp": int(time.time()*1000)}
        query_string = '&'.join([f'{k}={v}' for k, v in params.items()])
        signature = hmac.new(api_secret.encode(), query_string.encode(), sha256).hexdigest()
        params["signature"] = signature
        headers = {"X-MBX-APIKEY": api_key}
        response = requests.get(api_balance, params=params, headers=headers)
        if response.status_code == 200:
            balance = response.json()
            print("你的当前账户余额：", balance)
            return balance
        else:
            print("获取余额失败。")
            return {}
    except Exception as e:
        logging.error("获取余额错误: %s", e)
        return {}

# 计算技术指标
def calculate_indicators(df):
    try:
        # 布林带
        df["bb_upper"], df["bb_middle"], df["bb_lower"] = talib.BBANDS(df["close"], timeperiod=bb_window, nbdevup=bb_deviation, nbdevdn=bb_deviation)
        # RSI
        df["rsi"] = talib.RSI(df["close"], timeperiod=timeframe)
    except Exception as e:
        logging.error("计算指标错误: %s", e)
    return df

# 执行交易
def execute_trade(side, price, current_time):
    try:
        params = {
            "symbol": symbol, 
            "side": side, 
            "type": "MARKET", 
            "quantity": amount,
            "timestamp": int(time.time()*1000)  # 将timestamp加入请求参数
        }
        # 按字典序排序参数（Binance要求，避免因参数顺序导致签名错误）
        sorted_params = sorted(params.items())
        query_string = '&'.join([f'{k}={v}' for k, v in sorted_params])
        signature = hmac.new(api_secret.encode(), query_string.encode(), hashlib.sha256).hexdigest()
        params["signature"] = signature  # 将签名加入请求参数
        headers = {"X-MBX-APIKEY": api_key}
        
        response = requests.post(api_trade, params=params, headers=headers)
        if response.status_code != 200:
            logging.error("执行交易错误: %s", response.json())
            print("执行交易错误: ", response.json())
        else:
            logging.info("交易执行成功: %s", response.json())
            print(f"交易执行: {side}，价格 {price}，时间 {current_time}")
    except Exception as e:
        logging.error("执行交易异常: %s", e)

# 主交易逻辑
def main():
    trade_executed = False
    while True:
        now = datetime.now()
        current_time = now.strftime("%H:%M:%S")
        data = fetch_binance_data()

        # 检查数据量是否足够
        if len(data) < timeframe:
            print("数据量不足")
            time.sleep(5)
            continue

        # 计算指标
        data['RSI'] = talib.RSI(data['close'], timeperiod=14)
        data['upper'], data['middle'], data['lower'] = talib.BBANDS(data['close'], timeperiod=bb_window, nbdevup=bb_deviation, nbdevdn=bb_deviation, matype=talib.MA_Type.EMA)

        price = data['close'].iloc[-1]

        # 买入条件
        if (data['low'].iloc[-1] < data['middle'].iloc[-1] and data['RSI'].iloc[-1] < 50):
            if not trade_executed:
                execute_trade("BUY", price, current_time)
                trade_executed = True
                print(f"买入条件触发，时间 {current_time}，价格 {price}")
        
        # 卖出条件
        if (data['high'].iloc[-1] > data['middle'].iloc[-1] and data['RSI'].iloc[-1] > 60):
            if not trade_executed:
                execute_trade("SELL", price, current_time)
                trade_executed = True
                print(f"卖出条件触发，时间 {current_time}，价格 {price}")
        
        trade_executed = False
        time.sleep(5)

if __name__ == "__main__":
    main()

额外注意事项 #

• 确保API密钥已开启现货交易权限，若设置了IP白名单，当前运行机器的IP需在白名单内
• 建议先在Binance测试网调试代码，避免真实资金损失
• Binance API要求签名计算时参数按字典序排列，修正后的代码已处理该逻辑

内容的提问来源于stack exchange，提问作者gandtrain1000