写回购部分逻辑

checker/c_erc20_to_mexc.py

@@ -215,13 +215,13 @@ def send_arb_msg(pct, chain_swap_data, cex_price, dex_price):
             print("Error decoding JSON from api.ipify.org")
             return None
     '''
-    ip = '127.0.0.1'
+    ip = 'http://127.0.0.1'
     url = '/table-data'
     query_price_endpoint = f'{ip}:{api_endpoint_port}{url}'
     arbitrage_data = {
         "pct": str(pct),
-        "openLimit": str(OPEN_LIMIT.quantize(decimal.Decimal('0.001'))),
-        "closeLimit": str(CLOSE_LIMIT.quantize(decimal.Decimal('0.001'))),
+        "openLimit": str(OPEN_LIMIT.quantize(decimal.Decimal('0.00001'))),
+        "closeLimit": str(CLOSE_LIMIT.quantize(decimal.Decimal('0.00001'))),
         "cexPrice": str(cex_price),
         "dexPrice": str(dex_price),
         "symbol": MEXC_TARGET_PAIR_USDT,
@@ -338,7 +338,7 @@ def update_data_for_plotly_and_table():
             historical_data_points.append(current_point)
             latest_values_for_table["dex_price"] = f"{dex_price:.8f}" if dex_price else "N/A"
             latest_values_for_table["cex_price"] = f"{cex_price:.8f}" if cex_price else "N/A"
-            latest_values_for_table["diff_dex_vs_cex_percentage"] = f"{diff_erc20_vs_mexc_pct:+.4%}" if diff_erc20_vs_mexc_pct is not None else "N/A" # 显示为百分比
+            latest_values_for_table["diff_dex_vs_cex_percentage"] = diff_erc20_vs_mexc_pct
             latest_values_for_table["profit_value_for_table"] = f"{actual_profit_usdt:.2f} {BASE_CURRENCY_SYMBOL}" if actual_profit_usdt is not None else "N/A" # 新增
             latest_values_for_table["dex_error"] = erc20_err
             latest_values_for_table["cex_error"] = mexc_err

s_erc20_to_mexc.py

@@ -2,12 +2,15 @@ import time
 import traceback
 import copy
 import os
+import requests
+import json
+
 from web3_py_client import EthClient
 from mexc_client import MexcClient
 from decimal import Decimal, ROUND_DOWN
 from as_utils import add_state_flow_entry
 from checker.logger_config import get_logger
-from pprint import pformat
+from plogger.error import pformat
 
 mexc = MexcClient()
 
@@ -178,6 +181,46 @@ class ArbitrageProcess:
             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):
         """
         前置检查,防止低能错误
@@ -220,7 +263,7 @@ class ArbitrageProcess:
             add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
             self._set_state(self.STATE_REJECT)
 
-            # traceback.print_exc()
+            # traceback.logger.error_exc()
 
     # 执行卖出，使用超价单
     def _execute_sell_on_exchange(self):
@@ -335,10 +378,190 @@ class ArbitrageProcess:
             add_state_flow_entry(self.process_item, self.current_state, msg, "fail")
             self._set_state(self.STATE_FAILED)
 
-            # traceback.print_exc()
+            # traceback.logger.error_exc()
    
     def _execute_wait_pct_cover(self):
-        pass
+        """
+        等待价差回归
+        """
+        msg = f"等待中心化交易所价差回归，目标：{self.close_limit}"
+        logger.info(msg)
+        add_state_flow_entry(self.process_item, self.current_state, msg, "pending")
+        try:
+            while True:
+                try:
+                    table_data = self.get_local_data_no_params(self.query_price_url)
+
+                    if table_data is None:
+                        continue
+
+                    # 处理价差信息等
+                    dex_vs_cex_percentage = Decimal(table_data['diff_dex_vs_cex_percentage'])
+
+                    if dex_vs_cex_percentage < self.close_limit:
+                        msg = f"价差已回归，目标：{self.close_limit}%, 当前: {dex_vs_cex_percentage}%"
+                        logger.info(msg)
+                        add_state_flow_entry(self.process_item, self.current_state, msg, "success")
+
+                        self._set_state(self.STATE_BUYING_ON_EXCHANGE)
+                        break
+                except Exception as e:
+                    exc_traceback = traceback.format_exc()
+                    msg = f"请求价格接口时出现错误\n{exc_traceback}"
+                    logger.error(msg)
+
+                time.sleep(0.5)
+        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)
 
     def _execute_buy_on_exchange(self):
-        pass
+        """
+        执行回购操作
+        """
+        msg = f"正在回购，目标回购数量：{self.already_sold_amount}, 金额: {self.buy_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
+            order_price = Decimal(0)
+            self.already_bought_amount = Decimal(0)
+
+            while order_error_times < 10:
+                time.sleep(0.5)
+
+                try:
+                    # 挂单价格获取
+                    params = {
+                        "symbol": self.symbol.replace('_', ''),
+                        "limit": 1
+                    }
+                    depth = mexc.market.get_depth(params)
+                    # 数据合法性
+                    if 'bids' not in depth or not depth['bids']:
+                        continue
+
+                    bid1_price = Decimal(depth['bids'][0][0])
+                    order_price = bid1_price
+
+                    # 准备购入的价值, 如果小于2u就不要提交了
+                    pseudo_value_to_buy = order_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}, 余额校验通过。"
+                                        logger.info(msg)
+                                        add_state_flow_entry(self.process_item, self.current_state, msg, "success")
+                                        break
+
+                        # 实际能购入的数量（可能会亏损导致买不回来, 所以要考虑实际有多少money）
+                        quantity_for_api = pseudo_value_to_buy / order_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)
+
+                        order_params = {
+                            "symbol": self.symbol.replace('_', ''),
+                            "side": "BUY",
+                            "type": "LIMIT",
+                            "price": order_price,
+                            "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:
+                            msg = f"交易所现货买入订单发送失败 \n params:{order_params_formated} \n rst: {exchange_buy_order_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
+                        
+                        self.exchange_buy_order_id = exchange_buy_order['orderId']
+                    
+                    # 有订单时的逻辑
+                    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'])
+                            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
+                        
+                        # 如果没有成交或取消则判断是否达到取消条件了，这里面不能置空
+                        else:
+                            params = {
+                                "symbol": self.symbol.replace('_', ''),
+                                "orderId": self.exchange_buy_order_id
+                            }
+                            mexc.trade.delete_order(params)
+                            # delete_order_formated = pformat(delete_order, indent=2)
+
+                            # msg = f"【WARNING】价格变化，重新挂单\n order: {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 = self.already_sold_amount - self.already_bought_amount
+            diff_value = diff * order_price
+
+            profit = (self.sell_value - self.buy_value) - diff_value
+            msg = f"套利流程完成, 最终利润{profit}, 卖值{self.sell_value}, 买值{self.buy_value}, 差量{diff}({diff_value})"
+            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)

toto.readme

@@ -8,8 +8,11 @@
 [ ] 前期给默认值，直接开啃策略
     [-] 架构整理
     [-] 参数解析
-    [ ] 写等待价差回归部分
+    [-] 写等待价差回归部分
     [ ] 写回购部分逻辑
+        [-] 没有订单时
+        [-] 有订单时
+        [-] 统计部分
 
 待定
 [ ]