arbitrage/as.py

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

from checker.logger_config import get_logger
logger = get_logger('as')
logger.info('\n\n----------------------------------------------as啓動--------------------------------------------')


import threading
import uuid # 用于生成唯一的流程ID
import time
import logging
import erc20_to_mexc
import web3_py_client
import traceback
import copy
import sys

from decimal import Decimal, ROUND_DOWN
from flask import Flask, request, jsonify
from flask_cors import CORS # 导入
from as_utils import get_formatted_timestamp
from as_utils import add_state_flow_entry
from config import wallet
from binance.client import Client # 用于获取ETH价格
from checker import ok_chain_client
from pprint import pprint
from pprint import pformat

ok_chain_client.api_config = {
  "api_key": 'a05643ab-fb17-402b-94a8-a886bd343301',  # 请替换为您的真实 API Key
  "secret_key": '9D59B53EB1E60B1B5F290D3698A8C9DA', # 请替换为您的真实 Secret Key
  "passphrase": 'Qwe123123.', # 请替换为您的真实 Passphrase
}

# 配置日志
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)

web3 = web3_py_client.EthClient() 
w3 = web3.w3

USER_WALLET = wallet["user_wallet"]
USER_EXCHANGE_WALLET = wallet["user_exchange_wallet"]

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

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

# --- 一些核心數據和鎖 ---
core_data = {
    "nonce": 0,                                     # 全局 Nonce
    "block_number": 0,                              # 全局 區塊號
    "eth_price": 0,                                 # eth價格
}
core_lock = threading.Lock()                        # 核心數據的锁

# --- pending數據和鎖 ---
pending_data = {
    # # 數據結構的demo
    # "0xaf181bbbf5bf56d9204bd18cd25abd90e51890e848525e7788b410689c0c26a4": {
    #     "block_number": 22570370,                   # 提交pending時的區塊，隔幾個區塊去獲取數據會更準確
    #     "tx_details": None,                         # okapi解析的數據, None就是還沒有獲取到
    #     "reponse": None,                            # okapi最後一次獲取到的數據
    # },
}
pending_lock = threading.Lock()
PENDING_CONFIRM_BLOCK = 3                           # 需要幾個區塊才進行確認

CHAIN_ID = -1
try:
    if w3.provider:
        CHAIN_ID = w3.eth.chain_id
        logger.info(f"Web3 已连接。chain_id={CHAIN_ID}")
    else:
        logger.info("Web3 未连接。")
except Exception as e:
    logger.info(f"初始化 {USER_WALLET} 的全局 nonce 时出错: {e}")

# Binance 客户端 (无需API Key/Secret即可获取公开行情数据)
try:
    binance_client = Client()
    # 测试连接 (可选，但建议)
    binance_client.ping()
    logger.info("成功连接到 Binance API。")
except Exception as e:
    logger.error(f"初始化 Binance Client 时发生错误: {e}")
    binance_client = None

# --- 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
            logger.info(f"已将 process_id: {process_id_to_move} 从 processing_list 移动到 history_process_list。")
            moved_successfully = True
        else:
            logger.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 = Decimal(process_item['profit'])
    profitLimit = Decimal(process_item['profitLimit'])
    USER_EXCHANGE_WALLET = process_item['userExchangeWallet']
    USER_WALLET = process_item['userWallet']
    SYMBOL = process_item['symbol']
    EXCHANGE_OUT_AMOUNT = process_item['exchangeOutAmount']


    gas_limit_multiplier = 1
    gas_price_multiplier = 1
    if profit > Decimal(2) * profitLimit:
        gas_price_multiplier = 1.2
    elif profit > Decimal(5) * profitLimit:
        gas_price_multiplier = 1.5
    elif profit > Decimal(10) * profitLimit:
        gas_price_multiplier = 2

    global core_data
    global core_lock
    global pending_data
    global pending_lock
    ap = erc20_to_mexc.ArbitrageProcess(tx, gas_limit_multiplier, gas_price_multiplier, process_item, 
        core_data, core_lock,
        pending_data, pending_lock,
    )

    # 一般都是从这个流程开始，测试时可以稍作修改、测试后续流程
    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)
    
# --- 核心數據更新綫程函數 ---
def update_core_data_periodically():
    """
    周期性更新 nonce 和 ETH 价格的线程函数。
    """
    global core_data # 明确表示我们要修改全局的 core_data

    if not USER_WALLET or USER_WALLET == "你的钱包地址":
        logger.error("USER_WALLET 未正确配置。nonce 更新将无法进行。")
        # 如果 USER_WALLET 未配置，可以考虑让线程不执行 nonce 更新，或者直接退出
        # 这里我们选择继续运行，但 nonce 不会被更新

    while True:
        try:
            new_eth_price = None
            new_nonce = None
            new_block_number = None

            # 1. 从 Binance 获取 ETH 价格
            if binance_client:
                try:
                    ticker = binance_client.get_symbol_ticker(symbol="ETHUSDT")
                    new_eth_price = float(ticker['price'])
                except Exception as e:
                    logger.error(f"从 Binance 获取 ETH 价格失败: {e}")
            else:
                logger.warning("Binance client 未初始化, 无法获取 ETH 价格。")

            # 2. 获取最新的 Nonce 和 最新的block_number
            # 确保 w3 已初始化且 USER_WALLET 已配置
            if w3 and w3.is_connected() and USER_WALLET and USER_WALLET != "你的钱包地址":
                try:
                    new_block_number = w3.eth.block_number
                    new_nonce = w3.eth.get_transaction_count(USER_WALLET, 'latest')
                except Exception as e:
                    logger.error(f"为 {USER_WALLET} 获取 Nonce 失败: {e}")
            elif not (w3 and w3.is_connected()):
                logger.warning("Web3 未连接, 无法获取 nonce。")
            elif not (USER_WALLET and USER_WALLET != "你的钱包地址"):
                logger.warning("USER_WALLET 配置不正确, 无法获取 nonce。")

            # 3. 更新共享数据 core_data (使用锁)
            # 只有当获取到新数据时才更新，避免不必要的写操作和日志
            with core_lock:
                if new_eth_price is not None and core_data["eth_price"] != new_eth_price:
                    eth_price = Decimal(new_eth_price)
                    eth_price = eth_price.quantize(Decimal('1e-2'), rounding=ROUND_DOWN)
                    core_data["eth_price"] = eth_price
                
                # 判斷block_number是否發生變化（升高）
                if new_block_number is not None and new_block_number > core_data["block_number"]:
                    core_data["block_number"] = new_block_number
                
                    # 區塊變了才刷新nonce，否則還是要靠本地的緩存維護
                    if new_nonce is not None and core_data["nonce"] != new_nonce:
                        core_data["nonce"] = new_nonce
            
            # logger.info(f"核心数据已更新: ETH Price = {core_data['eth_price']}, Nonce ({USER_WALLET}) = {core_data['nonce']}, BlockNumber = {core_data['block_number']}")

        except Exception as e:
            # 捕获线程循环中的其他潜在错误
            logger.error(f"数据更新线程发生未知错误: {e}")
            traceback.print_exc()
        
        # 等待 500ms
        time.sleep(0.5)


# --- tx pending數據獲取綫程函數 ---
def update_tx_data():
    """
    每一秒獲取一條tx數據
    """
    global pending_data # 明确表示我们要修改全局的 pending_data

    while True:
        # 等待1s
        time.sleep(1)

        try:
            # 使用拷貝后的數據，否則可能會出現綫程問題
            with pending_lock:
                pending_data_copy = copy.deepcopy(pending_data)

            # 核心數據同理
            with core_lock:
                core_data_copy = copy.deepcopy(core_data)

            block_number = core_data_copy['block_number']
            for tx in pending_data_copy:
                try:
                    # 已獲取的就不要再獲取了
                    if pending_data_copy[tx]['tx_details'] is not None:
                        continue

                    # PENDING_CONFIRM_BLOCK個區塊之後的才進行確認，防止回滾頻繁觸發
                    if block_number < pending_data_copy[tx]['block_number'] + PENDING_CONFIRM_BLOCK:
                        continue

                    # 調用ok的api，直接獲取詳細交易
                    ok_rst = ok_chain_client.history(CHAIN_ID, tx)
                    # 存儲最後一次獲取的細節
                    with pending_lock:
                        pending_data[tx]['response'] = ok_rst

                    # 錯誤響應
                    if ok_rst['code'] != '0':
                        raise RuntimeError("API 返回错误响应", ok_rst)
                    
                    # ok不一定那麽快獲取到
                    if ok_rst['data'] is None:
                        # 每一個之間等待1s
                        time.sleep(1)

                        continue
                    
                    details = ok_rst['data']
                    # 有時候不會馬上識別出成交數量
                    if details['fromTokenDetails'] is None or details['toTokenDetails']:
                        # 每一個之間等待1s
                        time.sleep(1)

                        continue

                    # 成功獲取之後直接調用更新
                    with pending_lock:
                        pending_data[tx]['tx_details'] = details

                        formated_data = pformat(ok_rst['data'], indent=2) # indent=2 让格式更整齐
                        logger.info(f"獲取成功: \n{formated_data}")
                except Exception as e:
                    logger.error(f"tx數據獲取失敗: {e}")
                    traceback.print_exc()
                
                # 每一個之間等待1s
                time.sleep(1)
        except Exception as e:
            logger.error(f"pending更新线程发生未知错误: {e}")
            traceback.print_exc()

@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(str(data['profit']))                                   # 利润
        profit_limit = Decimal(str(data['profitLimit']))                        # 利润阈值
        from_token_amount_human = Decimal(str(data['fromTokenAmountHuman']))    # fromToken 的人类可读数量
        from_token_decimal = Decimal(str(data['fromTokenDecimal']))             # fromToken 的小数位数
        to_token_amount_human = Decimal(str(data['toTokenAmountHuman']))        # toToken 的人类可读数量
        to_token_decimal = Decimal(str(data['toTokenDecimal']))             # fromToken 的小数位数
        exchange_out_amount = 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)
    with core_lock:
        current_block = core_data['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": str(from_token_decimal), # 起始代币小数位数
            "toTokenAmountHuman": str(to_token_amount_human),
            "toTokenDecimal": str(to_token_decimal),
            "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
        logger.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__":
    logger.info("启动核心数据更新线程...")
    updater_thread = threading.Thread(target=update_core_data_periodically, daemon=True)
    updater_thread.start()

    logger.info("启动pending信息獲取线程...")
    pending_thread = threading.Thread(target=update_tx_data, daemon=True)
    pending_thread.start()

    logger.info("主线程继续执行，可以执行其他任务或保持运行以观察数据更新。")

    logger.info("启动 Flask 套利执行服务器...")
    app.run(host='0.0.0.0', port=188, debug=False) # 使用与 price_checker 不同的端口