feat: 支持 jito #4

Author: mkdir700

example.config.toml

@@ -22,6 +22,9 @@ unit_limit = 81000
 unit_price = 3000000
 preflight_check = false
 tx_simulate = false
+use_jito = true
+# jito_api 可根据服务器地址选择，就近原则 https://docs.jito.wtf/lowlatencytxnsend/#api
+jito_api = "https://mainnet.block-engine.jito.wtf"
 
 [api]
 helius_api_base_url = "https://api.helius.xyz/v0"

src/common/config.py

@@ -11,6 +11,7 @@
     BaseModel,
     ConfigDict,
     Field,
+    HttpUrl,
     MySQLDsn,
     RedisDsn,
     field_validator,
@@ -152,6 +153,16 @@ class TradingConfig(BaseModel):
     unit_limit: int
     tx_simulate: bool = False
     preflight_check: bool = False
+    use_jito: bool = True
+    jito_api: str = "https://mainnet.block-engine.jito.wtf"
+
+    @field_validator("jito_api", mode="before")
+    def validate_jito_api(cls, value: str) -> str:
+        try:
+            HttpUrl(value)
+        except ValueError:
+            raise ValueError(f"Invalid Jito API URL: {value}")
+        return value
 
 
 class APIConfig(BaseModel):

src/common/utils/jito.py

@@ -0,0 +1,68 @@
+import base64
+from typing import Literal
+
+import base58
+import httpx
+from loguru import logger
+from solana.rpc.types import TxOpts
+from solders.signature import Signature  # type: ignore
+from solders.transaction import VersionedTransaction  # type: ignore
+
+from common.config import settings
+
+
+class JitoClient:
+    def __init__(self) -> None:
+        self.client = httpx.AsyncClient(
+            base_url=settings.trading.jito_api,
+        )
+
+    async def send_transaction(
+        self,
+        transaction: VersionedTransaction,
+        opts: TxOpts | None = None,
+        encoding: Literal["base64", "base58"] = "base64",
+    ) -> Signature:
+        """Send a signed transaction.
+
+        Args:
+            transaction (VersionedTransaction): The signed transaction to send
+            opts (TxOpts | None, optional): The transaction options. Defaults to None.
+
+        Returns:
+            Signature: The transaction signature
+        """
+        path = "/api/v1/transactions"
+        if opts is None:
+            opts = TxOpts(skip_preflight=not settings.trading.preflight_check)
+
+        txn = bytes(transaction)
+        if encoding == "base64":
+            txn_encoded = base64.b64encode(txn).decode("utf-8")
+        elif encoding == "base58":
+            txn_encoded = base58.b58encode(txn).decode("utf-8")
+        else:
+            raise ValueError("encoding must be either 'base64' or 'base58'")
+
+        resp = await self.client.post(
+            path,
+            json={
+                "id": 1,
+                "jsonrpc": "2.0",
+                "method": "sendTransaction",
+                "params": [
+                    txn_encoded,
+                    {"encoding": encoding},
+                ],
+            },
+        )
+        # {
+        #     "jsonrpc": "2.0",
+        #     "result": "2id3YC2jK9G5Wo2phDx4gJVAew8DcY5NAojnVuao8rkxwPYPe8cSwE5GzhEgJA2y8fVjDEo6iR6ykBvDxrTQrtpb",
+        #     "id": 1,
+        # }
+        js = resp.json()
+        bundle_id = resp.headers.get("x-bundle-id")
+        sig = Signature.from_string(js["result"])
+        logger.info(f"Bundle ID: {bundle_id}, Signature: {sig}")
+        return sig

src/trading/executor.py

@@ -6,19 +6,22 @@
 from common.log import logger
 from common.models.tg_bot.user import User
 from common.types.swap import SwapEvent
+from common.utils.jito import JitoClient
 from db.session import NEW_ASYNC_SESSION, provide_session
 from trading.swap import SwapDirection, SwapInType
 from common.utils.raydium import RaydiumAPI
 from cache.launch import LaunchCache
-from .swap_protocols import Gmgn, Pump
+from trading.transaction import TradingRoute, TradingService
+from common.config import settings
 
 PUMP_FUN_PROGRAM_ID = str(PUMP_FUN_PROGRAM)
 RAY_V4_PROGRAM_ID = str(RAY_V4)
 
 
 class TradingExecutor:
     def __init__(self, client: AsyncClient):
-        self._client = client
+        self._rpc_client = client
+        self._jito_client = JitoClient()
         self._raydium_api = RaydiumAPI()
         self._launch_cache = LaunchCache()
 
@@ -84,39 +87,31 @@ async def exec(self, swap_event: SwapEvent):
 
         if should_use_pump:
             logger.info("Program ID is PUMP")
-            sig = await Pump(self._client).swap(
-                keypair=keypair,
-                token_address=token_address,
-                ui_amount=swap_event.ui_amount,
-                swap_direction=swap_direction,
-                slippage_bps=slippage_bps,
-                in_type=swap_in_type,
-                priority_fee=swap_event.priority_fee,
-            )
+            trade_route = TradingRoute.PUMP
         # NOTE: 测试下来不是很理想，暂时使用备选方案
         elif swap_event.program_id == RAY_V4_PROGRAM_ID:
             logger.info("Program ID is RayV4")
-            sig = await Gmgn(self._client).swap(
-                keypair=keypair,
-                token_address=token_address,
-                ui_amount=swap_event.ui_amount,
-                swap_direction=swap_direction,
-                slippage_bps=slippage_bps,
-                in_type=swap_in_type,
-                priority_fee=swap_event.priority_fee,
-            )
+            trade_route = TradingRoute.GMGN
         elif program_id is None or program_id == RAY_V4_PROGRAM_ID:
             logger.warning("Program ID is Unknown, So We use thrid party to trade")
-            sig = await Gmgn(self._client).swap(
-                keypair=keypair,
-                token_address=token_address,
-                ui_amount=swap_event.ui_amount,
-                swap_direction=swap_direction,
-                slippage_bps=slippage_bps,
-                in_type=swap_in_type,
-                priority_fee=swap_event.priority_fee,
-            )
+            trade_route = TradingRoute.GMGN
         else:
             raise ValueError(f"Program ID is not supported, {swap_event.program_id}")
 
+        sig = await TradingService.create(
+            trade_route,
+            self._rpc_client,
+            use_jito=settings.trading.use_jito,
+            jito_client=self._jito_client,
+        ).swap(
+            keypair,
+            token_address,
+            swap_event.ui_amount,
+            swap_direction,
+            slippage_bps,
+            swap_in_type,
+            use_jito=settings.trading.use_jito,
+            priority_fee=swap_event.priority_fee,
+        )
+
         return sig

src/trading/swap_protocols/__init__.py

@@ -0,0 +1,15 @@
+from trading.transaction.base import TransactionSender
+from trading.transaction.builders.base import TransactionBuilder
+from trading.transaction.factory import TransactionFactory, TradingService
+from trading.transaction.protocol import TradingRoute
+from trading.transaction.sender import DefaultTransactionSender, JitoTransactionSender
+
+__all__ = [
+    "TransactionBuilder",
+    "TransactionSender",
+    "TransactionFactory",
+    "TradingService",
+    "TradingRoute",
+    "DefaultTransactionSender",
+    "JitoTransactionSender",
+]

src/trading/swap_protocols/proto.py

@@ -0,0 +1,48 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from solana.rpc.async_api import AsyncClient
+from solana.rpc.types import TxOpts
+from solders.signature import Signature
+from solders.transaction import VersionedTransaction
+
+from common.utils.jito import JitoClient
+
+
+class TransactionSender(ABC):
+    """交易发送器的抽象基类"""
+
+    def __init__(self, client: AsyncClient | JitoClient):
+        self.client = client
+
+    @abstractmethod
+    async def send_transaction(
+        self,
+        transaction: VersionedTransaction,
+        opts: Optional[TxOpts] = None,
+    ) -> Signature:
+        """发送交易
+
+        Args:
+            transaction (VersionedTransaction): 要发送的交易
+            opts (Optional[TxOpts], optional): 交易选项. Defaults to None.
+
+        Returns:
+            Signature: 交易签名
+        """
+        pass
+
+    @abstractmethod
+    async def simulate_transaction(
+        self,
+        transaction: VersionedTransaction,
+    ) -> bool:
+        """模拟交易
+
+        Args:
+            transaction (VersionedTransaction): 要模拟的交易
+
+        Returns:
+            bool: 模拟是否成功
+        """
+        pass

src/trading/transaction/__init__.py

@@ -0,0 +1,44 @@
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from solana.rpc.async_api import AsyncClient
+from solders.keypair import Keypair
+from solders.transaction import VersionedTransaction
+
+from trading.swap import SwapDirection, SwapInType
+
+
+class TransactionBuilder(ABC):
+    """交易构建器的抽象基类"""
+
+    def __init__(self, rpc_client: AsyncClient):
+        self.rpc_client = rpc_client
+
+    @abstractmethod
+    async def build_swap_transaction(
+        self,
+        keypair: Keypair,
+        token_address: str,
+        ui_amount: float,
+        swap_direction: SwapDirection,
+        slippage_bps: int,
+        in_type: SwapInType | None = None,
+        use_jito: bool = False,
+        priority_fee: Optional[float] = None,
+    ) -> VersionedTransaction:
+        """构建交易
+
+        Args:
+            keypair (Keypair): 钱包密钥对
+            token_address (str): 代币地址
+            ui_amount (float): 交易数量
+            swap_direction (SwapDirection): 交易方向
+            slippage_bps (int): 滑点，以 bps 为单位
+            in_type (SwapInType | None, optional): 输入类型. Defaults to None.
+            use_jito (bool, optional): 是否使用 Jito. Defaults to False.
+            priority_fee (Optional[float], optional): 优先费用. Defaults to None.
+
+        Returns:
+            VersionedTransaction: 构建好的交易
+        """
+        pass