arbitrage/as.py

# -*- coding: utf-8 -*-

import decimal
import threading
import uuid # 用于生成唯一的流程ID
import time
import logging
import erc20_to_mexc_first_sell

# 配置日志
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

from flask import Flask, request, jsonify
from flask_cors import CORS # 导入
from web3_py_client import EthClient # 你特定的客户端
from as_utils import get_formatted_timestamp
from as_utils import add_state_flow_entry


web3 = EthClient() 

USER_WALLET = '0xb1f33026Db86a86372493a3B124d7123e9045Bb4' # 用户钱包地址
USER_EXCHANGE_WALLET = '0xc71835a042F4d870B0F4296cc89cAeb921a9f3DA' # 用户在交易所的钱包地址 (用于充值)

# 该代币最后一次执行套利的区块信息 (如果需要防止过于频繁的同类套利，不然变成砸盘、拉盘的了)
last_process_info = {} # 示例: {"RATO_USDT": 0}
MIN_BLOCKS_BETWEEN_ARB = decimal.Decimal(5) # 在重试相同交易对之前等待几个区块

# --- 全局状态和锁 ---
processing_list = [] # 正在处理的任务列表
history_process_list = [] # 已完成的任务历史列表
list_lock = threading.Lock() # 用于修改 processing_list 和 history_process_list 结构的锁

# USER_WALLET 的 Nonce 管理，前提是此服务器为该钱包发起交易。
# 如果传入的 'tx' 已预签名并包含 nonce，则此全局 nonce 对该特定 tx 不太重要。
# 但如果此服务器要从 USER_WALLET 发起*其他*交易（例如代币批准），则此 nonce 很重要。
global_nonce_USER_WALLET = 0 # 从 USER_WALLET 发送交易的全局 Nonce
nonce_lock_USER_WALLET = threading.Lock() # USER_WALLET Nonce 的锁

try:
    if web3.w3.provider: # 检查 web3 是否已某种程度初始化
        # 这个全局 nonce 应该小心初始化。
        # 如果 price_checker 发送交易，它应该管理 USER_WALLET 的 tx 的 nonce。
        # 这个服务器的 global_nonce 是针对它自己可能创建的 tx。
        # 暂时假设传入的 TX 具有其 nonce 或 price_checker 处理了它。
        global_nonce_USER_WALLET = web3.w3.eth.get_transaction_count(USER_WALLET, 'latest')
        logging.info(f"如果服务器要创建交易，{USER_WALLET} 的初始 nonce 将在此处获取。")
    else:
        logging.info("Web3 提供者未连接, USER_WALLET 的全局 nonce 未初始化。")
except Exception as e:
    logging.info(f"初始化 {USER_WALLET} 的全局 nonce 时出错: {e}")

# --- Flask 应用 ---
app = Flask(__name__)
CORS(app) # 在创建 app 实例后启用 CORS

def move_completed_process_to_history(process_id_to_move: str) -> bool:
    """
    将一个完成的 process_item 从 processing_list 移动到 history_process_list。
    此操作是线程安全的。

    Args:
        process_id_to_move (str): 要移动的 process_item 的 ID。

    Returns:
        bool: 如果成功找到并移动了 item，则返回 True，否则返回 False。
    """
    global processing_list, history_process_list # 因为我们要修改这两个列表
    
    item_to_move = None
    moved_successfully = False

    with list_lock:
        # 查找并从 processing_list 中移除
        found_index = -1
        for i, item in enumerate(processing_list):
            if item.get('id') == process_id_to_move:
                found_index = i
                break
        
        if found_index != -1:
            item_to_move = processing_list.pop(found_index) # 从 processing_list 中移除并获取它
            
            # 假设在 item_to_move 中，其 currentState 已经被 arbitrage_process_flow 更新为 COMPLETED 或 FAILED
            # arbitrage_process.add_state_flow_entry(item_to_move, "MOVED_TO_HISTORY", f"流程处理完毕，移至历史记录。最终状态: {item_to_move.get('currentState', 'N/A')}", "info")
            
            history_process_list.append(item_to_move) # 添加到 history_process_list
            logging.info(f"已将 process_id: {process_id_to_move} 从 processing_list 移动到 history_process_list。")
            moved_successfully = True
        else:
            logging.warning(f"尝试移动到 history_list 时，在 processing_list 中未找到 process_id: {process_id_to_move}")
            
    return moved_successfully

def arbitrage_process_flow(process_item):
    """
    在单独线程中执行的实际套利逻辑。
    会直接修改 'process_item' 字典。
    """
    process_id = process_item['id']
    SYMBOL = process_item['symbol']
    tx = process_item['tx'] # 预期包含 'rawTransaction' (原始交易)
    FROM_TOKEN = process_item['fromToken']
    TO_TOKEN = process_item['toToken']
    FROM_TOKEN_AMOUNT_HUMAM = process_item['fromTokenAmountHuman']
    TO_TOKEN_AMOUNT_HUMAM = process_item['toTokenAmountHuman']
    profit = float(process_item['profit'])
    USER_EXCHANGE_WALLET = process_item['userExchangeWallet']
    USER_WALLET = process_item['userWallet']
    SYMBOL = process_item['symbol']
    EXCHANGE_OUT_AMOUNT = process_item['exchangeOutAmount']


    gas_price_multiplier = 1
    if profit > 2:
        gas_price_multiplier = 1.1
    elif profit > 5:
        gas_price_multiplier = 1.5
    elif profit > 10:
        gas_price_multiplier = 2
    gas_limit_multiplier = 1.2

    ap = erc20_to_mexc_first_sell.ArbitrageProcess(tx, gas_limit_multiplier, gas_price_multiplier, 
                                                           FROM_TOKEN, TO_TOKEN, 
                                                           FROM_TOKEN_AMOUNT_HUMAM, EXCHANGE_OUT_AMOUNT,
                                                           USER_EXCHANGE_WALLET, USER_WALLET,
                                                           SYMBOL, process_item)

    # 一般都是从这个流程开始，测试时可以稍作修改、测试后续流程
    ap._set_state(ap.STATE_CHECK)

    # 在主循环中周期性调用 run_arbitrage_step
    while ap.current_state != ap.STATE_COMPLETED and ap.current_state != ap.STATE_FAILED and ap.current_state != ap.STATE_REJECT:
        ap.run_arbitrage_step()

        if ap.current_state == ap.STATE_WAITING_TRANSFER_ARRIVE or ap.current_state == ap.STATE_WAITING_WITHDRAWAL_CONFIRM:
            time.sleep(10)

    ap.run_arbitrage_step()

    move_completed_process_to_history(process_id)
    

@app.route('/submit_process', methods=['POST'])
def handle_submit_process():
    data = request.get_json()
    if not data:
        return jsonify({"error": "无效的 JSON 请求体"}), 400

    required_fields = ['tx', 'profit', 'profitLimit', 'symbol', 'fromToken', 'fromTokenAmountHuman', 'fromTokenDecimal', 'toToken', 'toTokenAmountHuman', 'exchangeOutAmount']
    for field in required_fields:
        if field not in data:
            return jsonify({"error": f"缺少字段: {field}"}), 400

    try:
        profit = decimal.Decimal(str(data['profit']))                                   # 利润
        profit_limit = decimal.Decimal(str(data['profitLimit']))                        # 利润阈值
        from_token_amount_human = decimal.Decimal(str(data['fromTokenAmountHuman']))    # fromToken 的人类可读数量
        from_token_decimal = decimal.Decimal(str(data['fromTokenDecimal']))             # fromToken 的小数位数
        to_token_amount_human = decimal.Decimal(str(data['toTokenAmountHuman']))        # toToken 的人类可读数量
        exchange_out_amount = decimal.Decimal(str(data['exchangeOutAmount']))           # 交易所需要卖出的数量
    except (decimal.InvalidOperation, ValueError) as e:
        return jsonify({"error": f"请求体中包含无效的小数/整数值: {e}"}), 400

    symbol = data['symbol'] # 交易对符号

    # 检查此交易对此区块是否处理过
    last_trade_block = last_process_info.get(symbol)
    current_block = web3.w3.eth.block_number
    if last_trade_block:
        if current_block - last_trade_block < MIN_BLOCKS_BETWEEN_ARB:
            return jsonify({"message": f"已跳过: {symbol} 最近已处理 (区块 {last_trade_block}). 当前区块 {current_block}."}), 200

    if profit >= profit_limit:
        process_id = str(uuid.uuid4()) # 生成唯一流程ID
        process_item = {
            "id": process_id,
            "creationTime": get_formatted_timestamp(), # 创建时间
            "tx": data['tx'], # 交易详情，应包含 rawTransaction
            "profit": str(profit), # 利润 (字符串存储)
            "profitLimit": str(profit_limit), # 利润阈值 (字符串存储)
            "symbol": symbol, # 交易对
            "userWallet": USER_WALLET,
            "userExchangeWallet": USER_EXCHANGE_WALLET,
            "fromToken": data['fromToken'], # 起始代币
            "fromTokenAmountHuman": str(from_token_amount_human), # 起始代币数量 (人类可读, 字符串存储)
            "fromTokenDecimal": from_token_decimal, # 起始代币小数位数
            "toTokenAmountHuman": str(to_token_amount_human),
            "exchangeOutAmount": str(exchange_out_amount),
            "toToken": data['toToken'], # 目标代币
            "stateFlow": [], # 状态流转记录
            "currentState": "PENDING_START", # 当前状态
        }
        # 初始状态更新
        add_state_flow_entry(process_item, "RECEIVED", f"流程已接收。利润 {profit} >= 利润阈值 {profit_limit}。开始套利。", "success")

        with list_lock:
            processing_list.append(process_item)

        last_process_info[symbol] = current_block
        logging.info(f"已更新 {symbol} 的最后处理信息至区块 {current_block}")

        # 在新线程中开始套利过程
        arb_thread = threading.Thread(target=arbitrage_process_flow, args=(process_item,), daemon=True)
        arb_thread.start()

        return jsonify({"message": "套利过程已启动", "process_id": process_id}), 201
    else:
        return jsonify({"message": f"利润 {profit} 小于利润阈值 {profit_limit}。不处理。"}), 200

@app.route('/processing', methods=['GET'])
def get_processing_list():
    """获取正在处理的任务列表"""
    with list_lock:
        # 返回一个副本，以避免在迭代生成 JSON 响应时列表被修改的问题
        return jsonify(list(processing_list))

@app.route('/history', methods=['GET'])
def get_history_list():
    """获取已完成的任务历史列表"""
    with list_lock:
        return jsonify(list(history_process_list))

@app.route('/status', methods=['GET'])
def get_status():
    """获取系统状态概览"""
    with list_lock:
        return jsonify({
            "processing_count": len(processing_list), # 正在处理的任务数量
            "history_count": len(history_process_list), # 历史任务数量
            # "current_nonce_USER_WALLET_if_managed_here": global_nonce_USER_WALLET, # 示例：如果服务器管理此nonce
            "last_process_info": last_process_info # 最后处理信息 (如果使用)
        })

if __name__ == "__main__":
    # 如果此服务器为其自身的交易管理 global_nonce_USER_WALLET，则在此处初始化
    logging.info("启动 Flask 套利执行服务器...")
    app.run(host='0.0.0.0', port=188, debug=False) # 使用与 price_checker 不同的端口