Skip to content

fix: catch slippage and on-chain tx errors in trade functions #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
smypmsa merged 2 commits into from
Mar 18, 2026
Merged

fix: catch slippage and on-chain tx errors in trade functions #15

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
04367a3
fix: catch slippage and on-chain errors from send_tx in trade functions
smypmsa Mar 18, 2026
d538dd7
fix: stop gitignoring idl/ so protocol definitions stay versioned
smypmsa Mar 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,9 @@ build/
.surfpool/
.coverage
docs/*
idl/
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Outdated
.claude/skills/evals/
.claude/skills/review-pr/
.claude/worktrees/

*.local.*
60 changes: 41 additions & 19 deletions src/pumpfun_cli/core/pumpswap.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
from pumpfun_cli.core.validate import invalid_pubkey_error, parse_pubkey
from pumpfun_cli.crypto import decrypt_keypair
from pumpfun_cli.protocol.address import derive_amm_user_volume_accumulator
from pumpfun_cli.protocol.client import RpcClient
from pumpfun_cli.protocol.client import RpcClient, TransactionFailedError
from pumpfun_cli.protocol.contracts import (
ATA_RENT_LAMPORTS,
LAMPORTS_PER_SOL,
Expand All @@ -16,6 +16,7 @@
PUMPSWAP_BUY_COMPUTE_UNITS,
PUMPSWAP_PRIORITY_FEE,
PUMPSWAP_SELL_COMPUTE_UNITS,
PUMPSWAP_SLIPPAGE_ERROR_CODES,
SOL_RENT_EXEMPT_MIN,
TOKEN_DECIMALS,
)
Expand Down Expand Up @@ -43,6 +44,21 @@ def _estimate_buy_required_lamports(
return sol_lamports + fee_lamports + ATA_RENT_LAMPORTS + SOL_RENT_EXEMPT_MIN


def _handle_tx_error(exc: TransactionFailedError, slippage_codes: set[int]) -> dict:
"""Convert a TransactionFailedError into a structured error dict."""
if exc.error_code in slippage_codes:
return {
"error": "slippage",
"message": "Transaction failed: slippage tolerance exceeded.",
"error_code": exc.error_code,
}
return {
"error": "tx_error",
"message": f"Transaction failed on-chain: {exc.raw_error}",
"error_code": exc.error_code,
}


async def buy_pumpswap(
rpc_url: str,
keystore_path: str,
Expand Down Expand Up @@ -158,15 +174,18 @@ async def buy_pumpswap(
if not vol_resp.value:
ixs.insert(0, build_init_amm_user_volume_accumulator(keypair.pubkey()))

sig = await client.send_tx(
ixs,
[keypair],
compute_units=compute_units
if compute_units is not None
else PUMPSWAP_BUY_COMPUTE_UNITS,
priority_fee=priority_fee if priority_fee is not None else PUMPSWAP_PRIORITY_FEE,
confirm=confirm,
)
try:
sig = await client.send_tx(
ixs,
[keypair],
compute_units=compute_units
if compute_units is not None
else PUMPSWAP_BUY_COMPUTE_UNITS,
priority_fee=priority_fee if priority_fee is not None else PUMPSWAP_PRIORITY_FEE,
confirm=confirm,
)
except TransactionFailedError as exc:
return _handle_tx_error(exc, PUMPSWAP_SLIPPAGE_ERROR_CODES)
result = {
"action": "buy",
"venue": "pumpswap",
Expand Down Expand Up @@ -279,15 +298,18 @@ async def sell_pumpswap(
min_sol_out=min_sol_lamports,
)

sig = await client.send_tx(
ixs,
[keypair],
compute_units=compute_units
if compute_units is not None
else PUMPSWAP_SELL_COMPUTE_UNITS,
priority_fee=priority_fee if priority_fee is not None else PUMPSWAP_PRIORITY_FEE,
confirm=confirm,
)
try:
sig = await client.send_tx(
ixs,
[keypair],
compute_units=compute_units
if compute_units is not None
else PUMPSWAP_SELL_COMPUTE_UNITS,
priority_fee=priority_fee if priority_fee is not None else PUMPSWAP_PRIORITY_FEE,
confirm=confirm,
)
except TransactionFailedError as exc:
return _handle_tx_error(exc, PUMPSWAP_SLIPPAGE_ERROR_CODES)
result = {
"action": "sell",
"venue": "pumpswap",
Expand Down
52 changes: 37 additions & 15 deletions src/pumpfun_cli/core/trade.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,10 +10,11 @@
derive_associated_bonding_curve,
derive_bonding_curve,
)
from pumpfun_cli.protocol.client import RpcClient
from pumpfun_cli.protocol.client import RpcClient, TransactionFailedError
from pumpfun_cli.protocol.contracts import (
ATA_RENT_LAMPORTS,
LAMPORTS_PER_SOL,
PUMP_SLIPPAGE_ERROR_CODES,
SOL_RENT_EXEMPT_MIN,
TOKEN_DECIMALS,
)
Expand Down Expand Up @@ -54,6 +55,21 @@ def _estimate_buy_required_lamports(
return sol_lamports + fee_lamports + ATA_RENT_LAMPORTS + SOL_RENT_EXEMPT_MIN


def _handle_tx_error(exc: TransactionFailedError, slippage_codes: set[int]) -> dict:
"""Convert a TransactionFailedError into a structured error dict."""
if exc.error_code in slippage_codes:
return {
"error": "slippage",
"message": "Transaction failed: slippage tolerance exceeded.",
"error_code": exc.error_code,
}
return {
"error": "tx_error",
"message": f"Transaction failed on-chain: {exc.raw_error}",
"error_code": exc.error_code,
}


async def buy_token(
rpc_url: str,
keystore_path: str,
Expand Down Expand Up @@ -153,13 +169,16 @@ async def buy_token(
token_program=token_program,
)

sig = await client.send_tx(
ixs,
[keypair],
confirm=confirm,
priority_fee=priority_fee if priority_fee is not None else 200_000,
compute_units=compute_units if compute_units is not None else 100_000,
)
try:
sig = await client.send_tx(
ixs,
[keypair],
confirm=confirm,
priority_fee=priority_fee if priority_fee is not None else 200_000,
compute_units=compute_units if compute_units is not None else 100_000,
)
except TransactionFailedError as exc:
return _handle_tx_error(exc, PUMP_SLIPPAGE_ERROR_CODES)
result = {
"action": "buy",
"mint": mint_str,
Expand Down Expand Up @@ -274,13 +293,16 @@ async def sell_token(
token_program=token_program,
)

sig = await client.send_tx(
ixs,
[keypair],
confirm=confirm,
priority_fee=priority_fee if priority_fee is not None else 200_000,
compute_units=compute_units if compute_units is not None else 100_000,
)
try:
sig = await client.send_tx(
ixs,
[keypair],
confirm=confirm,
priority_fee=priority_fee if priority_fee is not None else 200_000,
compute_units=compute_units if compute_units is not None else 100_000,
)
except TransactionFailedError as exc:
return _handle_tx_error(exc, PUMP_SLIPPAGE_ERROR_CODES)
result = {
"action": "sell",
"mint": mint_str,
Expand Down
27 changes: 24 additions & 3 deletions src/pumpfun_cli/protocol/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,29 @@

DEFAULT_RPC_TIMEOUT = 30.0

_ERR_PATTERN = r"InstructionError\(\((\d+),.*InstructionErrorCustom\((\d+)\)"


class TransactionFailedError(RuntimeError):
"""Raised when a confirmed transaction fails on-chain."""

error_code: int | None
instruction_index: int | None
raw_error: str

def __init__(self, err_obj: object) -> None:
import re

self.raw_error = str(err_obj)
match = re.search(_ERR_PATTERN, self.raw_error)
if match:
self.instruction_index = int(match.group(1))
self.error_code = int(match.group(2))
else:
self.instruction_index = None
self.error_code = None
super().__init__(self.raw_error)


class RpcClient:
"""Simplified Solana RPC client — no background tasks, no lifecycle."""
Expand Down Expand Up @@ -104,9 +127,7 @@ async def send_tx(
resp.value, max_supported_transaction_version=0
)
if tx_resp.value and tx_resp.value.transaction.meta.err:
raise RuntimeError(
f"Transaction confirmed but failed: {tx_resp.value.transaction.meta.err}"
)
raise TransactionFailedError(tx_resp.value.transaction.meta.err)
return str(resp.value)

async def get_transaction(self, signature_str: str) -> dict | None:
Expand Down
4 changes: 4 additions & 0 deletions src/pumpfun_cli/protocol/contracts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,10 @@
ATA_RENT_LAMPORTS = 2_039_280
BASE_TX_FEE = 5_000

# On-chain slippage error codes
PUMP_SLIPPAGE_ERROR_CODES: set[int] = {6002, 6003, 6042}
PUMPSWAP_SLIPPAGE_ERROR_CODES: set[int] = {6004, 6040}
Comment thread
smypmsa marked this conversation as resolved.

# PumpSwap compute budgets
PUMPSWAP_BUY_COMPUTE_UNITS = 400_000
PUMPSWAP_SELL_COMPUTE_UNITS = 300_000
Expand Down
89 changes: 89 additions & 0 deletions tests/test_commands/test_trade_cmd.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -615,6 +615,95 @@ def test_sell_slippage_100(tmp_path, monkeypatch):
assert "Slippage must be between" not in result.output


# --- slippage / tx_error exit code tests ---


def test_buy_slippage_error_exit_code_3(tmp_path, monkeypatch):
"""Core returning slippage error results in exit code 3."""
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("PUMPFUN_PASSWORD", "testpass")

from solders.keypair import Keypair

from pumpfun_cli.crypto import encrypt_keypair

config_dir = tmp_path / "pumpfun-cli"
config_dir.mkdir()
encrypt_keypair(Keypair(), "testpass", config_dir / "wallet.enc")

with patch("pumpfun_cli.commands.trade.buy_token", new_callable=AsyncMock) as mock_buy:
mock_buy.return_value = {
"error": "slippage",
"message": "Transaction failed: slippage tolerance exceeded.",
"error_code": 6002,
}

result = runner.invoke(
app,
["--rpc", "http://rpc", "buy", _FAKE_MINT, "0.01"],
)

assert result.exit_code == 3
assert "slippage" in result.output.lower()


def test_sell_slippage_error_exit_code_3(tmp_path, monkeypatch):
"""Core returning slippage error on sell results in exit code 3."""
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("PUMPFUN_PASSWORD", "testpass")

from solders.keypair import Keypair

from pumpfun_cli.crypto import encrypt_keypair

config_dir = tmp_path / "pumpfun-cli"
config_dir.mkdir()
encrypt_keypair(Keypair(), "testpass", config_dir / "wallet.enc")

with patch("pumpfun_cli.commands.trade.sell_token", new_callable=AsyncMock) as mock_sell:
mock_sell.return_value = {
"error": "slippage",
"message": "Transaction failed: slippage tolerance exceeded.",
"error_code": 6003,
}

result = runner.invoke(
app,
["--rpc", "http://rpc", "sell", _FAKE_MINT, "all"],
)

assert result.exit_code == 3
assert "slippage" in result.output.lower()


def test_buy_tx_error_exit_code_1(tmp_path, monkeypatch):
"""Core returning tx_error results in exit code 1."""
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
monkeypatch.setenv("PUMPFUN_PASSWORD", "testpass")

from solders.keypair import Keypair

from pumpfun_cli.crypto import encrypt_keypair

config_dir = tmp_path / "pumpfun-cli"
config_dir.mkdir()
encrypt_keypair(Keypair(), "testpass", config_dir / "wallet.enc")

with patch("pumpfun_cli.commands.trade.buy_token", new_callable=AsyncMock) as mock_buy:
mock_buy.return_value = {
"error": "tx_error",
"message": "Transaction failed on-chain: error 6020",
"error_code": 6020,
}

result = runner.invoke(
app,
["--rpc", "http://rpc", "buy", _FAKE_MINT, "0.01"],
)

assert result.exit_code == 1


def test_buy_json_output_has_expected_keys(tmp_path, monkeypatch):
"""Verify JSON buy output has all expected keys."""
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path))
Expand Down
Loading
Loading