arbitrage/s_erc20_to_mexc.py

import time
import traceback
import copy
import os
import requests
import json

from mexc_client import MexcClient
from decimal import Decimal, ROUND_DOWN
from as_utils import add_state_flow_entry
from decimal_utils import decimal_to_string_no_scientific
from checker.logger_config import get_logger
from pprint import pformat
from pprint import pprint

mexc = MexcClient()

# 配置日志
logger = get_logger('as')

class ArbitrageProcess:
    def __init__(self, process_item, 
        core_data, core_lock,
        mexc_data, mexc_lock,
    ):
        """
        初始化套利流程

        Args:
            process_item: 信號發送端傳入的原始參數
        """


        '''
        process_item
        {
            'cexPrice': '0.0104700000',
            'dexPrice': '0.01030220134289945035466016829',
            'pct': '0',
            'closeLimit': '-1.000',
            'openLimit': '0.000',
            'exchangeOutAmount': '3000.000000000000000000',
            'queryPriceUrl': '127.0.0.1:7777/table-data',
            'strategy': 'erc20_to_mexc',
            'symbol': 'APETH_USDT'
        }
        '''
        self.core_data = core_data
        self.core_lock = core_lock
        self.mexc_data = mexc_data
        self.mexc_lock = mexc_lock

        # symbol轉大寫
        self.symbol = process_item['symbol'].upper()
        self.coin = self.symbol.split('_')[0]
        self.base_coin = self.symbol.split('_')[1]
        
        # 其它参数
        self.cex_price = Decimal(process_item['cexPrice'])
        self.dex_price = Decimal(process_item['dexPrice'])
        self.pct = Decimal(process_item['pct'])
        self.open_limit = Decimal(process_item['openLimit'])
        self.close_limit = Decimal(process_item['closeLimit'])
        self.query_price_url = process_item['queryPriceUrl']

        # 留档
        self.process_item = process_item

        # 存储当前套利交易的细节信息,例如买入数量、价格等
        self.sell_amount = Decimal(process_item['exchangeOutAmount'])           # 交易所卖出量
        self.sell_price = Decimal(0)                                            # 实际卖出价格
        self.sell_value = Decimal(0)                                            # 实际卖出价值
        self.buy_price = Decimal(0)
        self.buy_amount = Decimal(0)
        self.buy_value = Decimal(0)
        self.actual_profit = Decimal(0)                                         # 實際利潤

        # 定义可能的状态
        self.STATE_IDLE = "IDLE"
        self.STATE_CHECK = "CHECK"                                      # 检查余额、估算gas等
        self.STATE_SELLING_ON_EXCHANGE = "SELLING_ON_EXCHANGE"          # 正在中心化交易所卖出现货
        self.STATE_WAITING_PCT_CONVER = "WAITING_PCT_CONVER"            # 等待价差回归
        self.STATE_BUYING_ON_EXCHANGE = "BUYING_ON_EXCHANGE"            # 正在中心化交易所买回现货
        self.STATE_COMPLETED = "COMPLETED"                              # 套利流程完成
        self.STATE_REJECT = "REJECT"                                    # 套利被程序拒绝
        self.STATE_FAILED = "FAILED"                                    # 套利流程失败

        self.STATES = [
            self.STATE_IDLE,
            self.STATE_CHECK,
            self.STATE_SELLING_ON_EXCHANGE,
            self.STATE_WAITING_PCT_CONVER,
            self.STATE_BUYING_ON_EXCHANGE,
            self.STATE_COMPLETED,
            self.STATE_REJECT,
            self.STATE_FAILED,
        ]

        # 所有前置信息获取都没有问题的话就等待开机信号
        self.current_state = self.STATE_IDLE

        # --------------------------------------- 获取交易规则 ---------------------------------------
        exchange_info_params = {
            "symbols": self.symbol.replace('_', '')
        }
        exchange_info_rst = mexc.market.get_exchangeInfo(exchange_info_params)
        # 返回值检查
        if 'symbols' not in exchange_info_rst or len(exchange_info_rst['symbols']) != 1:
            params_formated = pformat(exchange_info_params, indent=2)
            info_formated = pformat(exchange_info_rst, indent=2)
            msg = f'获取交易规则时出现错误\n{exchange_info_params}\n{info_formated}'
            logger.error(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

            self.current_state = self.STATE_FAILED
            return

        # 返回的交易对信息核对]
        exchange_info = exchange_info_rst['symbols'][0]
        if exchange_info['symbol'].upper() != self.symbol.replace('_', ''):
            info_formated = pformat(exchange_info, indent=2)
            msg = f'获取到的交易规则与交易币对无关\n{info_formated}'
            logger.error(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

            self.current_state = self.STATE_FAILED
            return
        
        # 精度取得, 假如是RATOUSDT这个交易对的话 
        self.coin_asset_precision = Decimal(f'1e-{exchange_info['baseAssetPrecision']}')            # 这是RATO的精度
        self.base_coin_asset_precision = Decimal(f'1e-{exchange_info['quoteAssetPrecision']}')      # 这是USDT的精度
        self.price_precision = Decimal(f'1e-{exchange_info['quotePrecision']}')                     # 这是价格的精度

        # 格式化价格
        self.cex_price = self.cex_price.quantize(self.price_precision, rounding=ROUND_DOWN)
        self.dex_price = self.dex_price.quantize(self.price_precision, rounding=ROUND_DOWN)

        # 数量
        self.sell_amount = self.sell_amount.quantize(self.coin_asset_precision, rounding=ROUND_DOWN)

    def _set_state(self, state):
        """
        设置系统状态,并打印日志
        """
        if state in self.STATES:
            logger.info(f"状态变更 {self.current_state} -> {state}")
            logger.info('')
            self.current_state = state
        else:
            logger.error(f"尝试设置无效状态 {state}")

    def run_arbitrage_step(self):
        """
        根据当前状态执行套利流程的下一步
        这是一个周期性调用的函数,例如在主循环中调用
        """

        if self.current_state == self.STATE_CHECK:
            self._execute_check()

        elif self.current_state == self.STATE_SELLING_ON_EXCHANGE:
            self._execute_sell_on_exchange()

        elif self.current_state == self.STATE_WAITING_PCT_CONVER:
            self._execute_wait_pct_cover()

        elif self.current_state == self.STATE_BUYING_ON_EXCHANGE:
            self._execute_buy_on_exchange()

        elif self.current_state == self.STATE_COMPLETED:
            msg = "套利流程成功完成！"
            logger.info(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "success")

        elif self.current_state == self.STATE_REJECT:
            msg = "套利流程被程序拒绝"
            logger.error(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

        elif self.current_state == self.STATE_FAILED:
            msg = "套利流程失败！"
            logger.error(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

    def get_local_data_no_params(self, url):
        """
        请求本地接口, 不携带参数, 并将返回值解析为 JSON。

        Args:
            url (str): 本地接口的 URL。

        Returns:
            dict or None: 如果请求成功且返回是有效的 JSON, 则返回 JSON 数据（Python 字典）。
                        否则, 返回 None。
        """
        try:
            # 发送 GET 请求到指定的 URL, 不携带参数
            response = requests.get(url)

            # 检查 HTTP 状态码, 200 表示成功
            response.raise_for_status()

            # 尝试将响应内容解析为 JSON
            # requests 库提供了一个方便的方法 .json() 来自动处理 JSON 解析和编码问题
            data = response.json()
            return data

        except requests.exceptions.HTTPError as err_http:
            logger.error(f"HTTP 错误发生: {err_http}")  # 例如 404 Not Found, 500 Internal Server Error
            return None
        except requests.exceptions.ConnectionError as err_conn:
            logger.error(f"连接错误发生: {err_conn}")  # 例如本地接口未运行
            return None
        except requests.exceptions.Timeout as err_timeout:
            logger.error(f"请求超时: {err_timeout}")
            return None
        except requests.exceptions.RequestException as err:
            logger.error(f"发生未知错误: {err}")
            return None
        except json.JSONDecodeError as err_json:
            logger.error(f"无法解析 JSON: {err_json}")
            logger.error(f"响应内容可能不是有效的 JSON: \n{response.text}")
            return None

    def _execute_check(self):
        """
        前置检查,防止低能错误
        """
        try:
            # step1,檢查交易所的餘額是否夠用
            # 处理精度
            pseudo_amount_to_sell = self.sell_amount

            msg = f"套利开始, dex {self.dex_price}, cex {self.cex_price}, pct {self.pct}"
            logger.info(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "success")

            # 交易所套保余额判断
            with self.mexc_lock:
                balances = self.mexc_data['account_info']['balances']

                for balance in balances:
                    if balance['asset'] == self.coin:
                        if Decimal(balance['free']) < pseudo_amount_to_sell:
                            msg = f"交易所剩余{self.coin}: {balance['free']}, 交易所准备卖出 {pseudo_amount_to_sell}, 不能触发交易。"
                            logger.info(msg)
                            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
                            self._set_state(self.STATE_REJECT)
                            return
                        else:
                            msg = f"交易所剩余{self.coin}: {balance['free']}, 交易所准备卖出 {pseudo_amount_to_sell}, 余额校验通过（可以交易）。"
                            logger.info(msg)
                            add_state_flow_entry(self.process_item, self.current_state, msg, "success")

                            break

            # final, 設定交易狀態,開始交易
            self._set_state(self.STATE_SELLING_ON_EXCHANGE)
        except Exception as e:
            exc_traceback = traceback.format_exc()
            msg = f"前置檢查未通過\n{exc_traceback}"
            logger.error(msg)

            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
            self._set_state(self.STATE_REJECT)

            # traceback.logger.error_exc()

    # 执行卖出, 使用超价单
    def _execute_sell_on_exchange(self):
        """
        在中心化交易所卖出现货
        """
        msg = "执行 中心化交易所卖出现货..."
        logger.info(msg)
        add_state_flow_entry(self.process_item, self.current_state, msg, "pending")
        try:
            self.already_sold_amount = Decimal(0)
            # 第一步直接卖出,这个数量用固定数量
            pseudo_amount_to_sell = self.sell_amount - self.already_sold_amount
            # 处理精度
            pseudo_amount_to_sell = pseudo_amount_to_sell.quantize(self.coin_asset_precision, rounding=ROUND_DOWN)
            price_for_api = self.dex_price.quantize(self.price_precision, rounding=ROUND_DOWN)
            price_for_api = decimal_to_string_no_scientific(price_for_api)
            # 初始化 quantity 变量
            quantity_for_api = None
            # 用求余法判断是否是整数
            if pseudo_amount_to_sell % 1 == 0:
                # 如果是整数, 转换为 int 类型。某些API可能只接受整数交易对的整数数量
                quantity_for_api = int(pseudo_amount_to_sell)
            else:
                # 如果是非整数, 转换为 float 类型。这是最常见的API数量类型
                quantity_for_api = float(pseudo_amount_to_sell)
            order_params = {
                "symbol": self.symbol.replace('_', ''),
                "side": "SELL",
                "type": "LIMIT",
                "price": price_for_api,
                "quantity": quantity_for_api,
            }
            
            order_params_formated = pformat(order_params, indent=2)
            
            exchange_sell_order = mexc.trade.post_order(order_params)
            exchange_sell_order_formated = pformat(exchange_sell_order, indent=2)

            msg = f"交易所现货卖出订单已发送 \n params:{order_params_formated} \n rst: {exchange_sell_order_formated}"
            logger.info(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "success")

            if 'orderId' not in exchange_sell_order:
                msg = '下单失败, 请检查参数及返回值'
                logger.error(msg)
                add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

                self._set_state(self.STATE_FAILED)

                return

            # 查询交易所订单状态
            self.exchange_sell_order_id = exchange_sell_order['orderId']
            waiting_times = 30
            order = None
            while waiting_times > 0:
                # 最后一次尝试就撤单了不搞了
                if waiting_times == 1:
                    params = {
                        "symbol": self.symbol.replace('_', ''),
                        "orderId": self.exchange_sell_order_id
                    }
                    delete_order = mexc.trade.delete_order(params)
                    delete_order_formated = pformat(delete_order, indent=2)
                    msg = f"【WARNING】交易所现货卖出未完全成交\n order: {delete_order_formated}"
                    logger.warning(msg)
                    add_state_flow_entry(self.process_item, self.current_state, msg, "success")

                    time.sleep(5)

                # 获取订单状态, 直到完全成交或超时
                params = {
                    "symbol": self.symbol.replace('_', ''),
                    "orderId": self.exchange_sell_order_id
                }
                order = mexc.trade.get_order(params)

                if order['status'] in ["FILLED", "PARTIALLY_CANCELED"]:
                    # 以实际成交数量为准
                    money = Decimal(order['cummulativeQuoteQty'])
                    self.already_sold_amount = self.already_sold_amount + Decimal(order['executedQty'])

                    self.sell_value = self.sell_value + money
                    self.sell_price = self.sell_value / self.already_sold_amount
                    self.sell_price = self.sell_price.quantize(self.price_precision, rounding=ROUND_DOWN)
                    
                    break
                else:
                    time.sleep(1)
                    waiting_times = waiting_times - 1

            order_formated = pformat(order, indent=2)
            msg = f"交易所现货卖出订单已完成, 价格 {self.sell_price}, sell_value: {self.sell_value}\n order: {order_formated}"
            logger.info(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "success")
            
            if self.sell_value > 2:
                msg = 'mexc现货卖出流程完成'
                logger.info(msg)
                add_state_flow_entry(self.process_item, self.current_state, msg, "success")

                self._set_state(self.STATE_BUYING_ON_EXCHANGE)
            else:
                msg = 'mexc现货卖出流程失败, 成交价值小于2'
                logger.info(msg)
                add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

                self._set_state(self.STATE_FAILED)
        except Exception as e:
            exc_traceback = traceback.format_exc()
            msg = f"交易所现货卖出下单失败\n{exc_traceback}"
            logger.error(msg)

            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
            self._set_state(self.STATE_FAILED)

            # traceback.logger.error_exc()、
    
    def _execute_buy_on_exchange(self):
        """
        执行回购操作
        """
        msg = f"正在回购, 目标回购数量 {self.already_sold_amount}, 金额: {self.sell_value}"
        logger.info(msg)
        add_state_flow_entry(self.process_item, self.current_state, msg, "pending")

        try:
            exchange_buy_order = None
            order_error_times = 0
            last_order_price = Decimal(0)
            dex_price = Decimal(0)
            self.already_bought_amount = Decimal(0)

            while order_error_times < 10:
                time.sleep(0.5)

                try:
                    table_data = self.get_local_data_no_params(self.query_price_url)

                    # 数据合法性
                    if table_data is None or 'dex_price' not in table_data:
                        continue

                    dex_price = Decimal(table_data['dex_price'])
                    dex_price = dex_price.quantize(self.price_precision, rounding=ROUND_DOWN)

                    ready_buy_price = dex_price * (Decimal(1) - self.close_limit)
                    ready_buy_price = ready_buy_price.quantize(self.price_precision, rounding=ROUND_DOWN)

                    # 准备购入的价值, 如果小于2u就不要提交了
                    pseudo_value_to_buy = ready_buy_price * (self.already_sold_amount - self.already_bought_amount)

                    if pseudo_value_to_buy < 2:
                        break

                    # 没有订单时的逻辑
                    if exchange_buy_order is None:
                        # 交易所U余额判断
                        with self.mexc_lock:
                            balances = self.mexc_data['account_info']['balances']
                            for balance in balances:
                                if balance['asset'] == self.base_coin:
                                    free_balance = Decimal(balance['free'])
                                    pseudo_value_to_buy = min(free_balance, pseudo_value_to_buy)

                                    if pseudo_value_to_buy < Decimal('2'):
                                        msg = f"交易所剩余{self.base_coin}: {free_balance}, 小于2, 不能触发回购交易。"
                                        logger.info(msg)
                                        add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
                                        self._set_state(self.STATE_FAILED)
                                        return
                                    else:
                                        msg = f"交易所剩余{self.base_coin}: {free_balance}, 准备使用 {pseudo_value_to_buy}, fp {dex_price}, 挂单价格{ready_buy_price},  余额校验通过。"
                                        logger.info(msg)
                                        # add_state_flow_entry(self.process_item, self.current_state, msg, "success")
                                        break

                        # 实际能购入的数量（可能会亏损导致买不回来, 所以要考虑实际有多少money）
                        quantity_for_api = pseudo_value_to_buy / ready_buy_price
                        quantity_for_api = quantity_for_api.quantize(self.coin_asset_precision, rounding=ROUND_DOWN)
                        # 用求余法判断是否是整数
                        if quantity_for_api % 1 == 0:
                            # 如果是整数, 转换为 int 类型。某些API可能只接受整数交易对的整数数量
                            quantity_for_api = int(quantity_for_api)
                        else:
                            # 如果是非整数, 转换为 float 类型。这是最常见的API数量类型
                            quantity_for_api = float(quantity_for_api)

                        price_for_api = decimal_to_string_no_scientific(ready_buy_price)
                        order_params = {
                            "symbol": self.symbol.replace('_', ''),
                            "side": "BUY",
                            "type": "LIMIT",
                            "price": price_for_api,
                            "quantity": quantity_for_api,
                        }
                        order_params_formated = pformat(order_params, indent=2)
                        exchange_buy_order = mexc.trade.post_order(order_params)
                        exchange_buy_order_formated = pformat(exchange_buy_order, indent=2)
                    
                        if 'orderId' not in exchange_buy_order:
                            table_data_formated = pformat(table_data, indent=2)
                            msg = f"交易所现货买入订单发送失败 \n params:{order_params_formated} \n rst: {exchange_buy_order_formated} \n table: {table_data_formated}"
                            logger.error(msg)
                            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")

                            exchange_buy_order = None
                            order_error_times = order_error_times + 1
                        else:
                            self.exchange_buy_order_id = exchange_buy_order['orderId']
                            last_order_price = ready_buy_price
                    # 有订单时的逻辑
                    else:
                        # 获取订单状态, 直到完全成交或超时
                        params = {
                            "symbol": self.symbol.replace('_', ''),
                            "orderId": self.exchange_buy_order_id
                        }
                        order = mexc.trade.get_order(params)

                        # 主要判断成交或取消了的订单
                        if order['status'] in ["FILLED", "PARTIALLY_CANCELED", "CANCELED"]:
                            # 以实际成交价值为准
                            money = Decimal(order['cummulativeQuoteQty'])

                            # 实际成交价值大于0才计算
                            if money > Decimal(0):
                                order_formated = pformat(order, indent=2)
                                logger.info(f"检测到有成交 \n {order_formated}")
                                self.already_bought_amount = self.already_bought_amount + Decimal(order['executedQty'])

                                self.buy_value = self.buy_value + money
                                self.buy_price = self.buy_value / self.already_bought_amount
                                self.buy_price = self.buy_price.quantize(self.price_precision, rounding=ROUND_DOWN)

                            exchange_buy_order = None
                        
                        # 如果没有成交或取消则判断是否达到取消条件了, 这里面不能置空
                        elif abs(Decimal(1) - last_order_price / ready_buy_price) > Decimal(0.0005):
                            params = {
                                "symbol": self.symbol.replace('_', ''),
                                "orderId": self.exchange_buy_order_id
                            }
                            _deleteed_order = mexc.trade.delete_order(params)
                            
                            # deleteed_order_formated = pformat(_deleteed_order, indent=2)
                            # msg = f"【WARNING】价格变化, 重新挂单, order price {last_order_price}, dex_price {dex_price} \n order: {deleteed_order_formated}"
                            # logger.warning(msg)
                            # add_state_flow_entry(self.process_item, self.current_state, msg, "success")
                except Exception as e:
                    exc_traceback = traceback.format_exc()
                    msg = f"请求价格接口时出现错误\n{exc_traceback}"
                    logger.error(msg)
            
            # diff 仍然代表未买回的数量, 非常重要, 需要记录
            diff = self.already_sold_amount - self.already_bought_amount
            unrealized_cost = diff * dex_price # 使用最后一次获取到的市价 dex_price 更合适

            # 已实现的利润 = 总卖出额 - 总买入额
            realized_profit = (self.sell_value - self.buy_value) - unrealized_cost
            realized_profit = realized_profit.quantize(Decimal('1e-4'), rounding=ROUND_DOWN)

            if diff > 0:
                # 如果有未买回的部分, 将其与最后一次的市价相乘, 作为 "浮动亏损" 或 "未平仓成本" 单独记录
                msg = f"套利流程完成, 但有 {diff} 的数量未回补。已实现利润: {realized_profit}, 未平仓成本估算: {unrealized_cost} (基于价格 {dex_price})"
            else:
                msg = f"套利流程完成, 全部回补。最终利润: {realized_profit}, 总卖值: {self.sell_value}, 总买值: {self.buy_value}"

            self.process_item['profit'] = realized_profit

            logger.info(msg)
            add_state_flow_entry(self.process_item, self.current_state, msg, "success")

            self._set_state(self.STATE_COMPLETED)
        except Exception as e:
            exc_traceback = traceback.format_exc()
            msg = f"等待价差回归失败\n{exc_traceback}"
            logger.error(msg)

            add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
            self._set_state(self.STATE_FAILED)