Bot example source
For setup and commands, see Bot examples. These complete snapshots are included in the full docs for AI.
Source revision: 75c2d66. All scripts require scripts/bot_common.py.
Copytrader
Read copytrader_bot.py
copytrader_bot.py
"""Copy configured wallets with capped buys and proportional exits."""
from decimal import Decimal, ROUND_DOWN, localcontext
import logging
import aiohttp
from bot_common import QuoteSizing, StopBot, Trader, USDC, positive, run, run_stream, setup, trusted_pool
class Copytrader:
def __init__(self, cfg, trader):
self.cfg, self.trader = cfg, trader
self.positions = {}
self.quotes = QuoteSizing()
def watched_wallet(self, event):
watched = self.cfg['copytrader']['wallets']
signer = event.get('txSigner')
involved = event.get('tradersInvolved')
if isinstance(involved, dict):
# Explicit traders take precedence over a fee payer in postBalances.
owner = self.positions.get(event.get('mint'), {}).get('wallet')
if owner in involved:
return owner
if signer in watched and signer in involved:
return signer
return next((wallet for wallet in watched if wallet in involved), None)
if signer in watched:
return signer
# Older events without tradersInvolved retain the original fallback.
balances = event.get('postBalances') or {}
return next((wallet for wallet in watched if wallet in balances), None)
@staticmethod
def wallet_quantity(event, wallet):
breakdown = event.get('breakdown')
if isinstance(breakdown, list):
return sum(
positive(trade.get('tokenAmount'))
for trade in breakdown
if isinstance(trade, dict) and trade.get('trader') == wallet and trade.get('action') == event['action']
)
if len(event.get('tradersInvolved') or {}) > 1:
return None # An aggregate amount cannot identify this wallet's exit.
return positive(event.get('tokenAmount'))
async def on_event(self, event):
# Tracked queues also carry confirmations and non-trade market updates.
if event.get('action') not in ('buy', 'sell') or not trusted_pool(event):
return
wallet = self.watched_wallet(event)
mint, quote = event.get('mint'), event.get('quoteMint')
price, quantity = positive(event.get('price')), positive(event.get('tokenAmount'))
if not wallet or not mint or not price or not quantity or not self.quotes.supports(quote):
return
allowed = self.cfg['copytrader']['token_mints']
if allowed and mint not in allowed:
return
if event['action'] == 'buy':
if mint in self.positions:
if wallet == self.positions[mint]['wallet']:
owned_quantity = self.wallet_quantity(event, wallet)
if owned_quantity is not None:
self.positions[mint]['copied_remaining'] += owned_quantity
return
balances = (event.get('postBalances') or {}).get(wallet, {})
balance = positive(balances.get(mint))
# Preserve upstream's first-buy gate across ALL transaction wallets.
post_total = sum(positive(values.get(mint)) for values in (event.get('postBalances') or {}).values())
if abs(post_total - quantity) > max(0.000001, quantity * 0.000001):
return
amount = min(
positive(event.get('quoteAmount')) * self.cfg['copytrader']['buy_fraction'],
self.quotes.amount(quote, self.cfg['copytrader']['max_buy_amount']),
)
if amount <= 0:
return
result = await self.trader.order('buy', mint, quote, amount, price)
self.positions[mint] = {
'wallet': wallet,
'quote': quote,
'tokens': positive(result['tokenAmount']),
'copied_remaining': balance or quantity,
'decimals': event.get('decimals'),
}
elif mint in self.positions:
position = self.positions[mint]
if wallet != position['wallet']:
return # A different watched wallet must not close this position.
quantity = self.wallet_quantity(event, wallet)
if quantity is None:
logging.warning('Skipped ambiguous bundled exit mint=%s; no per-wallet breakdown', mint)
return
if quantity <= 0:
return
fraction = min(1.0, quantity / position['copied_remaining'])
percent = min(100.0, fraction * 100)
amount = '100%' if fraction >= 1 else position['tokens'] * fraction
if self.trader.live and fraction < 1:
decimals = position['decimals']
if not isinstance(decimals, int) or isinstance(decimals, bool) or not 0 <= decimals <= 255:
raise StopBot('Token decimals unavailable; cannot size a partial copy exit. Check the wallet')
with localcontext() as context:
context.prec = max(28, decimals + 20)
exact = (
Decimal(str(position['tokens']))
* Decimal(str(quantity))
/ Decimal(str(position['copied_remaining']))
)
amount = float(exact.quantize(Decimal(1).scaleb(-decimals), rounding=ROUND_DOWN))
if amount == 0:
position['copied_remaining'] = max(0, position['copied_remaining'] - quantity)
logging.info('Skipped copy exit below one token unit mint=%s', mint)
return
if quote != position['quote']:
price = price / self.quotes.sol_price if quote == USDC else price * self.quotes.sol_price
result = await self.trader.order('sell', mint, position['quote'], amount, price, position['tokens'])
position['tokens'] = max(0, position['tokens'] - positive(result['tokenAmount']))
position['copied_remaining'] = max(0, position['copied_remaining'] - quantity)
if fraction >= 1 or position['tokens'] == 0:
self.positions.pop(mint)
logging.info('Copied wallet exit %s mint=%s fraction=%.2f%%', wallet, mint, percent)
def accept_event(self, event):
if self.quotes.observe(event):
return False
return event.get('action') in ('buy', 'sell') and trusted_pool(event) and bool(self.watched_wallet(event))
async def on_tick(self, mint=None):
return mint in self.positions
async def main():
cfg, args = setup('copytrader')
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
trader = Trader(cfg, session, args.live)
strategy = Copytrader(cfg, trader)
await run_stream(cfg, trader, strategy.on_event, strategy.on_tick, strategy.accept_event)
if __name__ == '__main__':
run(main)
Sniper
Read live_sniper_bot.py
live_sniper_bot.py
"""Buy a selected token or snipe Pump launches, then manage TP/SL/idle exits."""
import logging
import time
import aiohttp
from bot_common import QuoteSizing, Trader, WSOL, positive, run, run_stream, setup, trusted_pool
class Sniper:
def __init__(self, cfg, trader, clock=time.monotonic):
self.cfg, self.trader, self.clock = cfg, trader, clock
self.positions = {}
self.pending = {}
self.finished = set()
self.quotes = QuoteSizing()
async def on_event(self, event):
mint, quote = event.get('mint'), event.get('quoteMint')
price = positive(event.get('price'))
if not mint:
return
selected = self.cfg['sniper']['token_mint']
if selected:
# Direct-token mode works across supported venues. Restrict to real
# trade events and buy once at the first observed market price.
eligible = mint == selected and event.get('action') in ('buy', 'sell')
else:
eligible = (
event.get('action') == 'create'
and event.get('pool') == 'pump'
and trusted_pool(event)
and (
positive(event.get('quoteAmount'))
> self.quotes.amount(quote, self.cfg['sniper']['min_initial_buy'])
)
)
if mint in self.pending:
pending = self.pending[mint]
if (
event.get('action') not in ('buy', 'sell')
or not price
or not self.quotes.supports(quote)
or (not selected and not trusted_pool(event))
):
return
if quote != pending['quote']:
price = price / self.quotes.sol_price if quote != WSOL else price * self.quotes.sol_price
self.pending.pop(mint)
ceiling = pending['decision'] / (1 - self.cfg['trade']['buy_slippage'] / 100)
if price > ceiling:
self.trader.paper_missed += 1
logging.info(
'PAPER MISS mint=%s price=%s ceiling=%s missed=%s', mint, price, ceiling, self.trader.paper_missed
)
if selected:
self.finished.add(mint)
return
await self.buy(mint, pending['quote'], price)
return # The fill event cannot also trigger a take-profit.
if mint not in self.positions:
if not self.quotes.supports(quote) or not price:
return
if not eligible or mint in self.finished:
return
if not self.trader.live:
self.pending[mint] = {'decision': price, 'quote': quote, 'last': self.clock()}
logging.info('PAPER PENDING BUY mint=%s; waiting for the next trade price', mint)
else:
await self.buy(mint, quote, price)
return
# As upstream, any token event resets idle; only these actions drive TP/SL.
self.positions[mint]['last'] = self.clock()
if event.get('action') not in ('buy', 'sell', 'add', 'remove'):
return
if not price or not self.quotes.supports(quote):
return
if not selected and not trusted_pool(event):
return
position = self.positions[mint]
# Compare prices in the position's entry quote, even across SOL/USDC pools.
if quote != position['quote']:
price = price / self.quotes.sol_price if quote != WSOL else price * self.quotes.sol_price
position.update(price=price, last=self.clock())
change = (price - position['entry']) / position['entry'] * 100
if change > self.cfg['sniper']['take_profit'] or change < -self.cfg['sniper']['stop_loss']:
await self.exit(mint, 'take profit' if change > self.cfg['sniper']['take_profit'] else 'stop loss')
async def buy(self, mint, quote, price):
result = await self.trader.order(
'buy',
mint,
quote,
self.quotes.amount(quote, self.cfg['trade']['buy_amount']),
price,
)
self.positions[mint] = {
'entry': positive(result['price']),
'price': positive(result['price']),
'tokens': positive(result['tokenAmount']),
'last': self.clock(),
'quote': quote,
}
logging.info(
'Opened %s; TP=%s%% SL=%s%%', mint, self.cfg['sniper']['take_profit'], self.cfg['sniper']['stop_loss']
)
def accept_event(self, event):
if self.quotes.observe(event):
return False
mint = event.get('mint')
return bool(mint) and (
mint in self.positions
or mint in self.pending
or mint == self.cfg['sniper']['token_mint']
or (event.get('action') == 'create' and event.get('pool') == 'pump' and trusted_pool(event))
)
async def exit(self, mint, reason):
position = self.positions[mint]
await self.trader.order('sell', mint, position['quote'], '100%', position['price'], position['tokens'])
self.positions.pop(mint)
if self.cfg['sniper']['token_mint']:
self.finished.add(mint)
logging.info('Closed %s (%s)', mint, reason)
async def on_tick(self, mint=None):
for token, pending in list(self.pending.items()):
if (mint is None or mint == token) and self.clock() - pending['last'] > self.cfg['sniper']['idle_seconds']:
self.pending.pop(token)
self.trader.paper_missed += 1
if self.cfg['sniper']['token_mint']:
self.finished.add(token)
logging.info('PAPER MISS mint=%s; no subsequent trade before idle timeout', token)
positions = (
list(self.positions.items())
if mint is None
else ([(mint, self.positions[mint])] if mint in self.positions else [])
)
for token, position in positions:
if self.clock() - position['last'] > self.cfg['sniper']['idle_seconds']:
await self.exit(token, 'idle timeout')
return mint in self.positions or mint in self.pending
async def main():
cfg, args = setup('sniper')
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
trader = Trader(cfg, session, args.live)
strategy = Sniper(cfg, trader)
await run_stream(cfg, trader, strategy.on_event, strategy.on_tick, strategy.accept_event)
if __name__ == '__main__':
run(main)
Backtester
Read backtest_sniper_strategy.py
backtest_sniper_strategy.py
"""Replay launch-sniper rules without credentials, signing, or broadcasting."""
import asyncio
from collections import Counter
from datetime import datetime, timedelta, timezone
import io
import logging
import math
from pathlib import Path
import tempfile
import aiohttp
import orjson
import zstandard
from bot_common import QuoteSizing, WSOL, USDC, run, setup, StopBot
class Backtest:
"""Upstream replay ordering, fills, fees and SOL-equivalent accounting."""
def __init__(self, cfg):
self.cfg = cfg
self.quotes = QuoteSizing()
self.positions = {}
self.trades = []
self.last_ts = 0
self.first_ts = self.window_end = None
self.seen = self.skipped = self.missed = self.buys = self.creates = 0
def fill_buy(self, mint, pos):
decision, fill = pos['decision_price'], pos['cur_price']
ceiling = decision / (1 - min(self.cfg['trade']['buy_slippage'], 99.9) / 100)
if fill > ceiling:
self.missed += 1
self.positions.pop(mint, None)
return
amount, quote = pos['amount'], pos['quote_mint']
pos.update(
state='held',
entry_price=fill,
quote_spent=amount,
tokens=amount
* (1 - self.cfg['backtest']['service_fee'] - self.cfg['backtest']['extra_fee'] - pos['entry_pool_fee'])
/ fill,
sol_price=self.quotes.sol_price,
spent_sol=amount if quote == WSOL else amount / self.quotes.sol_price,
buy_slip=(fill / decision - 1) * 100,
last_ts=pos['buy_fill_ts'],
)
self.buys += 1
def trigger_sell(self, pos, ts, reason):
pos.update(state='pending_sell', sell_reason=reason, sell_fill_ts=ts + self.cfg['backtest']['sell_latency_ms'])
def close(self, mint, pos, exit_price, reason):
gross = pos['tokens'] * exit_price
fee = self.cfg['backtest']['service_fee'] + self.cfg['backtest']['extra_fee']
proceeds = gross * (1 - (fee + pos['last_pool_fee']))
pnl = proceeds - pos['quote_spent']
fee_quote = pos['quote_spent'] * (fee + pos['entry_pool_fee']) + gross * (fee + pos['last_pool_fee'])
is_sol = pos['quote_mint'] == WSOL
self.trades.append(
{
'mint': mint,
'quote_mint': pos['quote_mint'],
'reason': reason,
'pct': (exit_price - pos['entry_price']) / pos['entry_price'] * 100,
'pnl': pnl,
'pnl_sol': pnl if is_sol else pnl / pos['sol_price'],
'spent_sol': pos['spent_sol'],
'fee_sol': fee_quote if is_sol else fee_quote / pos['sol_price'],
'buy_slip': pos['buy_slip'],
}
)
self.positions.pop(mint, None)
def resolve(self, ts):
for mint, pos in list(self.positions.items()):
if pos['state'] == 'pending_buy' and pos['buy_fill_ts'] <= ts:
self.fill_buy(mint, pos)
elif pos['state'] == 'pending_sell' and pos['sell_fill_ts'] <= ts:
self.close(mint, pos, pos['cur_price'], pos['sell_reason'])
elif pos['state'] == 'held' and ts - pos['last_ts'] > self.cfg['sniper']['idle_seconds'] * 1000:
self.trigger_sell(pos, pos['last_ts'] + self.cfg['sniper']['idle_seconds'] * 1000, 'idle')
def handle_event(self, event):
self.seen += 1
ts = event['timestamp']
self.last_ts = ts
self.first_ts = ts if self.first_ts is None else min(self.first_ts, ts)
self.window_end = ts if self.window_end is None else max(self.window_end, ts)
# Preserve upstream ordering: resolve BEFORE applying this event's price,
# including timestamps that arrive out of order in the archive.
self.resolve(ts)
if event['action'] not in ('buy', 'sell', 'add', 'remove', 'create', 'migrate', 'createPool'):
return
mint = event['mint']
if self.quotes.observe(event):
return
if event['action'] == 'create' and event['pool'] == 'pump' and not event.get('mayhemMode'):
self.creates += 1
quote = event['quoteMint']
threshold = self.quotes.amount(quote, self.cfg['sniper']['min_initial_buy'])
if self.quotes.supports(quote) and event['quoteAmount'] > threshold and mint not in self.positions:
pool_fee = event['poolFeeRate']
self.positions[mint] = {
'state': 'pending_buy',
'quote_mint': quote,
'amount': self.quotes.amount(quote, self.cfg['trade']['buy_amount']),
'decision_price': event['price'],
'cur_price': event['price'],
'entry_pool_fee': pool_fee,
'last_pool_fee': pool_fee,
'buy_fill_ts': ts + self.cfg['backtest']['buy_latency_ms'],
'buy_slip': 0.0,
}
elif mint in self.positions:
pos = self.positions[mint]
if event['action'] in ('buy', 'sell', 'add', 'remove', 'migrate') and (
event['pool'] == 'pump' or (event['pool'] == 'pump-amm' and event.get('poolCreatedBy') == 'pump')
):
price = event['price']
pos.update(cur_price=price, last_pool_fee=event['poolFeeRate'])
if pos['state'] == 'held':
change = (price - pos['entry_price']) / pos['entry_price'] * 100
if change > self.cfg['sniper']['take_profit'] or change < -self.cfg['sniper']['stop_loss']:
self.trigger_sell(pos, ts, 'tp' if change > self.cfg['sniper']['take_profit'] else 'sl')
if pos['state'] == 'held':
pos['last_ts'] = ts
def finish(self):
# Upstream forces in-flight buys to fill, then closes at last-known price.
for mint, pos in list(self.positions.items()):
if pos['state'] == 'pending_buy':
self.fill_buy(mint, pos)
for mint, pos in list(self.positions.items()):
reason = (
pos['sell_reason']
if pos['state'] == 'pending_sell'
else ('idle' if self.last_ts - pos['last_ts'] > self.cfg['sniper']['idle_seconds'] * 1000 else 'end')
)
self.close(mint, pos, pos['cur_price'], reason)
pnl_sol = sum(trade['pnl_sol'] for trade in self.trades)
volume = sum(trade['spent_sol'] for trade in self.trades)
wins = sum(trade['pnl_sol'] > 0 for trade in self.trades)
logging.info(
'Events=%s creates=%s buys=%s skipped=%s closed=%s missed=%s',
self.seen,
self.creates,
self.buys,
self.skipped,
len(self.trades),
self.missed,
)
logging.info(
'Modeled PnL=%+.8f SOL-equivalent; WSOL=%+.8f; USDC=%+.8f; wins=%s/%s',
pnl_sol,
sum(t['pnl'] for t in self.trades if t['quote_mint'] == WSOL),
sum(t['pnl'] for t in self.trades if t['quote_mint'] == USDC),
wins,
len(self.trades),
)
logging.info(
'Volume=%s SOL-equivalent; ROI=%+.2f%%; fees=%s SOL-equivalent',
volume,
pnl_sol / volume * 100 if volume else 0,
sum(t['fee_sol'] for t in self.trades),
)
count = len(self.trades)
if not count:
logging.info('No trades filled in this window; check missed entries and your entry/slippage settings.')
reasons = Counter(t['reason'] for t in self.trades)
logging.info(
'Exits: tp=%s sl=%s idle=%s end=%s; win_rate=%.2f%%',
reasons['tp'],
reasons['sl'],
reasons['idle'],
reasons['end'],
wins / count * 100 if count else 0,
)
logging.info(
'Average move=%+.2f%%; average entry slippage=%+.2f%%',
sum(t['pct'] for t in self.trades) / count if count else 0,
sum(t['buy_slip'] for t in self.trades) / count if count else 0,
)
if self.trades:
for label, trade in [
('Best', max(self.trades, key=lambda t: t['pnl_sol'])),
('Worst', min(self.trades, key=lambda t: t['pnl_sol'])),
]:
logging.info(
'%s trade: mint=%s pnl=%+.8f SOL-equivalent move=%+.2f%% reason=%s',
label,
trade['mint'],
trade['pnl_sol'],
trade['pct'],
trade['reason'],
)
if self.first_ts is not None:
start = datetime.fromtimestamp(self.first_ts / 1000, timezone.utc)
end = datetime.fromtimestamp(self.window_end / 1000, timezone.utc)
logging.info('Replay window: %s to %s UTC', start.isoformat(), end.isoformat())
logging.info(
'Original replay model: last-known-price fills, end-window forced fills; not live execution guarantees.'
)
def replay_file(path, strategy, compressed=None):
path = Path(path)
if compressed is None:
compressed = path.suffix == '.zst'
with path.open('rb') as file:
if compressed:
with zstandard.ZstdDecompressor().stream_reader(file) as reader:
read_lines(io.BufferedReader(reader), strategy)
else:
read_lines(file, strategy)
def read_lines(reader, strategy):
for line in reader:
if not line.strip():
continue
try:
event = orjson.loads(line)
except orjson.JSONDecodeError:
strategy.skipped += 1
continue
if not isinstance(event, dict):
strategy.skipped += 1
continue
try:
# Validate only inputs that can poison time or create/update positions.
ts = event.get('timestamp')
if not finite(ts) or ts < 0:
raise ValueError('Invalid timestamp')
launch = (
event.get('action') == 'create'
and event.get('pool') == 'pump'
and not event.get('mayhemMode')
and strategy.quotes.supports(event.get('quoteMint'))
)
amount = event.get('quoteAmount')
entry = (
launch
and finite(amount)
and amount > strategy.quotes.amount(event['quoteMint'], strategy.cfg['sniper']['min_initial_buy'])
)
update = (
event.get('mint') in strategy.positions
and event.get('action') in ('buy', 'sell', 'add', 'remove', 'migrate')
and (
event.get('pool') == 'pump'
or (event.get('pool') == 'pump-amm' and event.get('poolCreatedBy') == 'pump')
)
)
if entry or update:
price, fee = event.get('price'), event.get('poolFeeRate')
if not finite(price) or price <= 0 or not finite(fee) or not 0 <= fee < 1:
raise ValueError('Invalid price or pool fee')
strategy.handle_event(event)
except (KeyError, TypeError, ValueError, ZeroDivisionError):
strategy.skipped += 1
def finite(value):
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
STORAGE_ERROR = (
'Cannot write archive to temporary storage. Free space, set TMPDIR to a writable folder with space, or use --file'
)
class ArchiveFile:
"""Translate only file I/O errors; preserve a failing download during close."""
def __init__(self, destination):
self.destination = destination
@staticmethod
def storage_call(method, *args):
try:
return method(*args)
except OSError as exc:
raise StopBot(STORAGE_ERROR) from exc
def __enter__(self):
self.file = self.storage_call(Path(self.destination).open, 'wb')
return self
def write(self, chunk):
self.storage_call(self.file.write, chunk)
def __exit__(self, error_type, error, traceback):
try:
self.storage_call(self.file.close)
except StopBot:
if error_type is None:
raise
logging.warning('Archive close also failed; preserving the original error')
async def download_hour(session, cfg, hour, destination):
url = f'{cfg["network"]["replay_url"].rstrip("/")}/{hour:%Y/%m/%d/%H}.jsonl.zst'
logging.info('Downloading archive %s UTC', hour.strftime('%Y-%m-%d %H:00'))
try:
async with session.get(url) as response:
if response.status == 404 and cfg['backtest']['allow_gaps']:
logging.warning('Skipping missing archive hour')
return False
if response.status != 200:
hint = (
('; the newest completed hour may still be uploading. Wait 90 seconds and retry, or use --file')
if response.status == 404
else '; use --file'
)
raise StopBot(f'Archive download returned HTTP {response.status}{hint}')
with ArchiveFile(destination) as file:
async for chunk in response.content.iter_chunked(1 << 20):
file.write(chunk)
except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as exc:
raise StopBot(f'Archive network failure: {type(exc).__name__}: {exc}; retry or use --file') from exc
return True
async def main():
cfg, args = setup('backtest')
strategy = Backtest(cfg)
if args.file:
replay_file(args.file, strategy)
else:
now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0)
async with aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=None, sock_connect=30, sock_read=120)
) as session:
with tempfile.TemporaryDirectory(prefix='zeroslip-replay-') as folder:
path = Path(folder) / 'hour.jsonl.zst'
for offset in range(cfg['backtest']['hours'], 0, -1):
if await download_hour(session, cfg, now - timedelta(hours=offset), path):
await asyncio.to_thread(replay_file, path, strategy, True)
strategy.finish()
if __name__ == '__main__':
run(main)
Sell balances
Read sell_all_tokens.py
sell_all_tokens.py
"""Preview or sell selected token balances; --all explicitly selects every token."""
import logging
import aiohttp
from bot_common import WSOL, USDC, api_post, credentials, positive, run, setup, StopBot
TOKEN_PROGRAMS = (
'TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA',
'TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb',
)
async def balances(cfg, session, live):
if live:
# Defaults to the credential's wallet; never scan one wallet then sell
# from a different one. This documented read incurs an API balance fee.
data = await api_post(session, cfg['network']['trade_url'], {**credentials(), 'action': 'getBalances'})
result = data.get('tokenBalances')
if not isinstance(result, dict):
raise StopBot('API did not return token balances')
return result
result = {}
for program in TOKEN_PROGRAMS:
payload = {
'jsonrpc': '2.0',
'id': 1,
'method': 'getTokenAccountsByOwner',
'params': [
cfg['wallet']['public_key'],
{'programId': program},
{'encoding': 'jsonParsed', 'commitment': 'confirmed'},
],
}
data = await api_post(session, cfg['network']['rpc_url'], payload, rpc=True)
accounts = data.get('result', {}).get('value')
if not isinstance(accounts, list):
raise StopBot('RPC did not return token accounts')
for account in accounts:
info = account['account']['data']['parsed']['info']
amount = info['tokenAmount']
quantity = positive(amount.get('uiAmountString'))
mint = info['mint']
entry = result.setdefault(mint, {'balance': 0, 'frozen': False})
entry['balance'] += quantity
entry['frozen'] = entry['frozen'] or info.get('state') == 'frozen'
return result
def select_balances(cfg, holdings, all_tokens):
wanted = cfg['sell']['token_mints']
# Preserve both trading quote assets, including explicit mint selections.
excluded = {WSOL, USDC}
return {
mint: info
for mint, info in holdings.items()
if mint not in excluded and (all_tokens or mint in wanted) and positive(info.get('balance'))
}
async def sell_selected(cfg, session, live, all_tokens):
selected = select_balances(cfg, await balances(cfg, session, live), all_tokens)
if not selected:
logging.info('No matching token balances to sell.')
return
for mint, info in selected.items():
if info.get('frozen'):
logging.warning('Skipping frozen balance mint=%s; no burn is attempted', mint)
continue
logging.info('%s SELL 100%% mint=%s balance=%s', 'LIVE' if live else 'PREVIEW', mint, info['balance'])
if not live:
continue
data = await api_post(
session,
cfg['network']['trade_url'],
{
**credentials(),
'action': 'sell',
'mint': mint,
'amount': '100%',
'denominatedInQuote': 'false',
'slippage': cfg['sell']['slippage'],
'priorityFee': cfg['trade']['priority_fee'],
'guaranteedDelivery': 'true',
},
)
if data.get('confirmed') is not True or not data.get('signature'):
raise StopBot('Sale not confirmed; check the wallet before retrying. No burn was attempted')
quote = next(
(trade.get('quoteMint') for trade in data.get('trades', []) if isinstance(trade, dict)), 'auto-selected'
)
logging.info('Sale confirmed mint=%s quote=%s signature=%s', mint, quote, data['signature'])
async def main():
cfg, args = setup('sell')
async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30)) as session:
await sell_selected(cfg, session, args.live, args.all)
if __name__ == '__main__':
run(main)
Shared runtime
Read bot_common.py
bot_common.py
"""Shared standalone configuration, execution, and stream handling (Python 3.11+)."""
import argparse
import asyncio
import copy
import logging
import math
import os
from pathlib import Path
import tomllib
from urllib.parse import urlparse
import aiohttp
import orjson
from cachetools import TTLCache
from dotenv import load_dotenv
import websockets
BASE58_ALPHABET = '123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz'
ROOT = Path(__file__).resolve().parent.parent
WSOL = 'So11111111111111111111111111111111111111112'
USDC = 'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v'
SOL_USDC_POOL = 'Gf7sXMoP8iRw4iiXmJ1nq4vxcRycbGXy5RL8a8LnTd3v'
DEFAULTS = {
'network': {
'trade_url': 'https://api.zeroslip.ai',
'stream_url': 'wss://stream.zeroslip.ai/',
'replay_url': 'https://replay.pumpapi.io',
'rpc_url': 'https://api.mainnet-beta.solana.com',
},
'trade': {
'buy_amount': 0.001,
'buy_slippage': 20.0,
'sell_slippage': 99.0,
'priority_fee': 0.0001,
'confirmation_seconds': 3.0,
},
'wallet': {'public_key': ''},
'copytrader': {'wallets': [], 'token_mints': [], 'buy_fraction': 0.1, 'max_buy_amount': 0.01},
'sniper': {
'token_mint': '',
'min_initial_buy': 20.0,
'take_profit': 50.0,
'stop_loss': 20.0,
'idle_seconds': 300.0,
},
'backtest': {
'hours': 10,
'allow_gaps': False,
'buy_latency_ms': 400,
'sell_latency_ms': 400,
'service_fee': 0.0025,
'extra_fee': 0.0005,
},
'sell': {'token_mints': [], 'slippage': 100.0},
}
class ConfigError(ValueError):
pass
class StopBot(RuntimeError):
"""Execution is rejected or uncertain: stop without sending another order."""
def address(value, label):
if not isinstance(value, str) or not 32 <= len(value) <= 44 or any(c not in BASE58_ALPHABET for c in value):
raise ConfigError(f'{label} must be a Solana base58 address')
# Verify decoded length, not just the alphabet.
number = 0
for char in value:
number = number * 58 + BASE58_ALPHABET.index(char)
size = (number.bit_length() + 7) // 8 + len(value) - len(value.lstrip('1'))
if size != 32:
raise ConfigError(f'{label} must decode to 32 bytes')
return value
def load_config(path):
cfg = copy.deepcopy(DEFAULTS)
try:
with Path(path).open('rb') as file:
overrides = tomllib.load(file)
except (OSError, tomllib.TOMLDecodeError) as exc:
raise ConfigError('Cannot read config; copy config.example.toml to config.toml and check TOML syntax') from exc
for section, values in overrides.items():
if section not in cfg or not isinstance(values, dict):
raise ConfigError(f'Unknown or invalid configuration section: {section}')
for key, value in values.items():
if section == 'trade' and key == 'quote_mint':
logging.getLogger(__name__).warning(
'Ignoring deprecated trade.quote_mint; SOL/USDC quotes come from each event. '
'Remove this key from your config'
)
continue
if key not in cfg[section]:
raise ConfigError(f'Unknown configuration option: {section}.{key}')
default = cfg[section][key]
if isinstance(default, bool):
valid = isinstance(value, bool)
elif isinstance(default, (float, int)):
valid = isinstance(value, (float, int)) and not isinstance(value, bool) and math.isfinite(value)
else:
valid = isinstance(value, type(default))
if not valid:
raise ConfigError(f'Invalid value type: {section}.{key}')
cfg[section][key] = value
for section, key in [
('trade', 'buy_amount'),
('trade', 'confirmation_seconds'),
('copytrader', 'max_buy_amount'),
('sniper', 'idle_seconds'),
('sniper', 'take_profit'),
('sniper', 'stop_loss'),
]:
if cfg[section][key] <= 0:
raise ConfigError(f'{section}.{key} must be positive')
for section, key in [('backtest', 'hours')]:
if not isinstance(cfg[section][key], int) or cfg[section][key] < 1:
raise ConfigError(f'{section}.{key} must be a positive integer')
for key in ('buy_slippage', 'sell_slippage'):
if not 0 <= cfg['trade'][key] < 100:
raise ConfigError(f'trade.{key} must be between 0 and 100 (exclusive)')
if not 0 <= cfg['sell']['slippage'] <= 100:
raise ConfigError('sell.slippage must be between 0 and 100 (inclusive)')
if not 0 < cfg['copytrader']['buy_fraction'] <= 1:
raise ConfigError('copytrader.buy_fraction must be between 0 and 1')
if not 0 < cfg['sniper']['stop_loss'] < 100:
raise ConfigError('sniper.stop_loss must be between 0 and 100 (exclusive)')
for section, key in [
('trade', 'priority_fee'),
('sniper', 'min_initial_buy'),
('backtest', 'buy_latency_ms'),
('backtest', 'sell_latency_ms'),
('backtest', 'service_fee'),
('backtest', 'extra_fee'),
]:
if cfg[section][key] < 0:
raise ConfigError(f'{section}.{key} cannot be negative')
if cfg['backtest']['service_fee'] + cfg['backtest']['extra_fee'] >= 1:
raise ConfigError('Backtest fee fractions must total less than 1')
for key, url in cfg['network'].items():
parsed = urlparse(url)
if (
parsed.scheme != ('wss' if key == 'stream_url' else 'https')
or not parsed.hostname
or parsed.username
or parsed.password
or parsed.query
):
raise ConfigError(f'network.{key} must be a secure URL without credentials or query parameters')
for section, key in [('copytrader', 'wallets'), ('copytrader', 'token_mints'), ('sell', 'token_mints')]:
for item in cfg[section][key]:
address(item, f'{section}.{key}')
if cfg['sniper']['token_mint']:
address(cfg['sniper']['token_mint'], 'sniper.token_mint')
return cfg
def credentials():
api_key = os.getenv('ZEROSLIP_API_KEY', '').strip()
private_key = os.getenv('ZEROSLIP_PRIVATE_KEY', '').strip()
if not api_key and not private_key:
raise ConfigError('Live mode needs ZEROSLIP_API_KEY or ZEROSLIP_PRIVATE_KEY in .env or your environment')
return {'apiKey': api_key} if api_key else {'privateKey': private_key}
def setup(kind):
parser = argparse.ArgumentParser(description=f'ZeroSlip standalone {kind}; paper mode unless --live')
parser.add_argument('--config', type=Path, help='Settings file; defaults to config.toml or config.example.toml')
parser.add_argument('--check-config', action='store_true', help='Validate settings and exit without network calls')
if kind != 'backtest':
parser.add_argument('--live', action='store_true', help='Enable real Lightning trades using local credentials')
if kind in ('sniper', 'sell'):
parser.add_argument('--mint', help='Token mint to buy/watch or sell (overrides config)')
if kind == 'copytrader':
parser.add_argument('--wallet', action='append', help='Wallet to copy; repeat for several wallets')
if kind == 'sell':
parser.add_argument('--wallet', help='Public wallet address for previews; live sells use the credential wallet')
parser.add_argument('--all', action='store_true', help='Select every non-quote token balance')
if kind == 'backtest':
parser.add_argument('--hours', type=int, help='Number of completed UTC archive hours')
parser.add_argument('--file', type=Path, help='Replay a local JSONL or JSONL.zst file instead of downloading')
args = parser.parse_args()
if args.config is None:
args.config = ROOT / ('config.toml' if (ROOT / 'config.toml').exists() else 'config.example.toml')
load_dotenv(args.config.resolve().parent / '.env', override=False)
try:
cfg = load_config(args.config)
if getattr(args, 'mint', None):
address(args.mint, '--mint')
if kind == 'sniper':
cfg['sniper']['token_mint'] = args.mint
else:
cfg['sell']['token_mints'] = [args.mint]
if kind == 'copytrader' and args.wallet:
cfg['copytrader']['wallets'] = [address(w, '--wallet') for w in args.wallet]
if kind == 'sell' and args.wallet:
cfg['wallet']['public_key'] = address(args.wallet, '--wallet')
if getattr(args, 'hours', None) is not None:
if args.hours < 1:
raise ConfigError('--hours must be positive')
cfg['backtest']['hours'] = args.hours
if not (kind == 'sell' and args.wallet):
cfg['wallet']['public_key'] = (
os.getenv('ZEROSLIP_WALLET_PUBLIC_KEY', '').strip() or cfg['wallet']['public_key']
)
if cfg['wallet']['public_key']:
address(cfg['wallet']['public_key'], 'wallet.public_key')
if kind == 'copytrader' and not cfg['copytrader']['wallets']:
raise ConfigError('Set copytrader.wallets or pass --wallet ADDRESS')
if kind == 'copytrader' and cfg['wallet']['public_key'] in cfg['copytrader']['wallets']:
raise ConfigError('Do not copy your own trading wallet')
if kind == 'sell':
if not args.live and not cfg['wallet']['public_key']:
raise ConfigError(
'Sell preview needs --wallet ADDRESS, wallet.public_key, or ZEROSLIP_WALLET_PUBLIC_KEY'
)
if args.all and (args.mint or cfg['sell']['token_mints']):
raise ConfigError('Choose --all or specific token mints, not both')
if not args.all and not cfg['sell']['token_mints']:
raise ConfigError('Choose --mint ADDRESS, sell.token_mints, or --all')
if getattr(args, 'live', False):
credentials()
except ConfigError as exc:
parser.error(str(exc))
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
if args.check_config:
print('Configuration valid. No network calls or transactions made.')
raise SystemExit(0)
return cfg, args
def trusted_pool(event):
return not event.get('mayhemMode') and (
event.get('pool') == 'pump' or (event.get('pool') == 'pump-amm' and event.get('poolCreatedBy') == 'pump')
)
def positive(value):
try:
value = float(value)
return value if math.isfinite(value) and value > 0 else 0.0
except (ValueError, TypeError):
return 0.0
class QuoteSizing:
"""Original two-quote strategy: SOL sizes, converted to USDC at stream price."""
def __init__(self):
self.sol_price = 80.0 # Original fallback until the trusted pool updates.
def observe(self, event):
if event.get('poolId') != SOL_USDC_POOL:
return False
price = positive(event.get('price'))
if price:
self.sol_price = 1 / price
return True
def amount(self, quote, sol_amount):
return sol_amount if quote == WSOL else sol_amount * self.sol_price
@staticmethod
def supports(quote):
return quote in (WSOL, USDC)
async def api_post(session, url, payload, *, rpc=False):
# Never log the credential-bearing payload or a server body that may echo it.
try:
async with session.post(url, json=payload, allow_redirects=False) as response:
if response.status != 200:
if rpc:
raise StopBot(f'RPC returned HTTP {response.status}; retry the balance preview')
raise StopBot(f'API returned HTTP {response.status}; check the order outcome before retrying')
data = await response.json()
if not isinstance(data, dict) or data.get('err') or data.get('error'):
if rpc:
error = data.get('error') if isinstance(data, dict) else 'Invalid response'
raise StopBot(f'RPC balance query failed: {error}')
raise StopBot('API rejected the request; check the order outcome before retrying')
return data
except (aiohttp.ClientError, asyncio.TimeoutError, ValueError) as exc:
if rpc:
raise StopBot(f'RPC balance query unavailable: {type(exc).__name__}: {exc}') from exc
raise StopBot('API response unavailable; order outcome is uncertain. Check the wallet before retrying') from exc
class Trader:
def __init__(self, cfg, session, live=False):
self.cfg, self.session, self.live = cfg, session, live
self.events = TTLCache(maxsize=10000, ttl=300)
self.paper_positions = {}
self.paper_totals = {WSOL: 0.0, USDC: 0.0}
self.paper_pnl_sol = 0.0
self.paper_missed = self.paper_buys = self.paper_sells = 0
self.quotes = QuoteSizing()
def observe(self, event):
self.quotes.observe(event)
signature = event.get('signature')
if signature:
self.events.setdefault(signature, []).append(event)
async def order(self, action, mint, quote, amount, price, tokens=0):
if not self.live:
logging.info('PAPER %s mint=%s amount=%s quote=%s', action.upper(), mint, amount, quote)
if action == 'buy':
quantity = float(amount) / price
self.paper_positions[mint] = {
'tokens': quantity,
'cost': float(amount),
'sol_price': self.quotes.sol_price,
}
self.paper_buys += 1
else:
quantity = (
tokens * float(amount[:-1]) / 100
if isinstance(amount, str) and amount.endswith('%')
else float(amount)
)
quantity = min(tokens, quantity)
position = self.paper_positions[mint]
cost = position['cost'] * quantity / position['tokens']
pnl = quantity * price - cost
self.paper_totals[quote] += pnl
self.paper_pnl_sol += pnl if quote == WSOL else pnl / position['sol_price']
self.paper_sells += 1
logging.info(
'PAPER RESULT mint=%s move=%+.2f%% pnl=%+.8f quote=%s total_sol=%+.8f',
mint,
(quantity * price / cost - 1) * 100,
pnl,
quote,
self.paper_pnl_sol,
)
position['tokens'] -= quantity
position['cost'] -= cost
if position['tokens'] <= 0:
self.paper_positions.pop(mint)
return {'price': price, 'tokenAmount': quantity, 'quoteAmount': quantity * price}
settings = self.cfg['trade']
payload = {
**credentials(),
'action': action,
'mint': mint,
'quoteMint': quote,
'amount': amount,
'denominatedInQuote': 'true' if action == 'buy' else 'false',
'slippage': settings['buy_slippage' if action == 'buy' else 'sell_slippage'],
'priorityFee': settings['priority_fee'],
}
response = await api_post(self.session, self.cfg['network']['trade_url'], payload)
signature = response.get('signature')
if not isinstance(signature, str) or not signature:
raise StopBot('API did not return a signature. Check the wallet before retrying')
logging.info('Submitted %s mint=%s signature=%s', action, mint, signature)
deadline = asyncio.get_running_loop().time() + settings['confirmation_seconds']
while asyncio.get_running_loop().time() < deadline:
for event in self.events.get(signature, []):
if (
event.get('action') == action
and event.get('mint') == mint
and positive(event.get('price'))
and positive(event.get('tokenAmount'))
):
logging.info('Confirmed %s mint=%s signature=%s', action, mint, signature)
return event
await asyncio.sleep(0.1)
raise StopBot(
f'No trade event observed for {signature}. Check the wallet before restarting; no order was retried'
)
def report(self):
if not self.live:
logging.info(
'Paper summary: buys=%s sells=%s missed=%s open=%s '
'realized_pnl=%+.8f SOL-equivalent; WSOL=%+.8f USDC=%+.8f (fees excluded)',
self.paper_buys,
self.paper_sells,
self.paper_missed,
len(self.paper_positions),
self.paper_pnl_sol,
self.paper_totals[WSOL],
self.paper_totals[USDC],
)
async def run_stream(cfg, trader, on_event, on_tick, accept_event=None):
# Independent token workers restore the original cross-token concurrency.
# A token's exits stay in stream order so proportional bookkeeping is stable.
queues = {}
workers = set()
failure = asyncio.get_running_loop().create_future()
def completed(task):
workers.discard(task)
if not task.cancelled() and task.exception() is not None and not failure.done():
failure.set_exception(task.exception())
async def token_worker(mint, queue):
while True:
try:
event = await asyncio.wait_for(queue.get(), timeout=1)
await on_event(event)
except asyncio.TimeoutError:
pass
tracking = await on_tick(mint)
if not tracking and queue.empty():
# No await between checking/removing: reader cannot enqueue into
# an orphaned queue. Untracked tokens don't retain idle workers.
queues.pop(mint, None)
return
async def reader():
delay = 0.4
while True:
connected_at = None
try:
async with websockets.connect(cfg['network']['stream_url'], max_size=8 << 20) as socket:
connected_at = asyncio.get_running_loop().time()
logging.info('Connected to market stream (%s)', 'LIVE' if trader.live else 'PAPER')
async for message in socket:
try:
event = orjson.loads(message)
except (ValueError, TypeError):
continue
if not isinstance(event, dict):
continue
trader.observe(event)
mint = event.get('mint') or '__market__'
# Once a token worker exists, retain its market updates
# even while its buy hasn't registered a position yet.
if accept_event is not None and not accept_event(event) and mint not in queues:
continue
if mint not in queues:
queues[mint] = asyncio.Queue(maxsize=10000)
task = asyncio.create_task(token_worker(mint, queues[mint]))
workers.add(task)
task.add_done_callback(completed)
try:
queues[mint].put_nowait(event)
except asyncio.QueueFull as exc:
raise StopBot('Token event queue overflowed; reconcile wallet positions') from exc
reason = f'close={socket.close_code} {socket.close_reason}'
except (OSError, websockets.exceptions.WebSocketException) as exc:
reason = f'{type(exc).__name__}: {exc}'
if connected_at is not None and asyncio.get_running_loop().time() - connected_at >= 30:
delay = 0.4
logging.warning('Stream disconnected (%s); reconnecting in %.1f seconds', reason, delay)
await asyncio.sleep(delay)
delay = min(delay * 2, 30)
stream = asyncio.create_task(reader())
try:
done, _ = await asyncio.wait([stream, failure], return_when=asyncio.FIRST_COMPLETED)
for task in done:
task.result()
finally:
stream.cancel()
for task in workers:
task.cancel()
await asyncio.gather(stream, *workers, return_exceptions=True)
if not failure.done():
failure.cancel()
trader.report()
def run(main):
try:
asyncio.run(main())
except KeyboardInterrupt:
logging.info('Stopped. Existing live positions remain in the wallet.')
except (StopBot, ConfigError) as exc:
logging.error('%s', exc)
raise SystemExit(1) from exc
except Exception as exc:
logging.exception('Unexpected runtime failure (%s); stopped', type(exc).__name__)
raise SystemExit(1) from exc
config.example.toml
Read config.example.toml
config.example.toml
# Copy to config.toml. Secrets belong in .env, never in this file.
[wallet]
public_key = "" # Your wallet address; required for sell previews; live sells use the credential wallet.
[trade]
buy_amount = 0.001 # SOL-equivalent; USDC amounts convert using the trusted SOL/USDC stream pool.
buy_slippage = 20
sell_slippage = 99 # Original streaming sell tolerance
priority_fee = 0.0001 # SOL network priority fee
confirmation_seconds = 3
[sniper]
token_mint = "" # Set a mint for one entry attempt; blank scans new Pump launches.
min_initial_buy = 20 # SOL-equivalent; applies only to launch mode.
take_profit = 50 # Percent price rise from entry
stop_loss = 20 # Percent price fall from entry
idle_seconds = 300
[copytrader]
wallets = [] # Add wallet addresses to copy, or pass --wallet ADDRESS.
token_mints = [] # Empty allows any token within the Pump pool filters.
buy_fraction = 0.1
max_buy_amount = 0.01 # SOL-equivalent cap; converts to USDC automatically
[backtest]
hours = 10 # Original completed-UTC-hour default; archives can be hundreds of MB each.
allow_gaps = false
buy_latency_ms = 400
sell_latency_ms = 400
service_fee = 0.0025 # Fraction per side, not percent
extra_fee = 0.0005 # Additional modeling assumption per side
[sell]
slippage = 100 # Original sell-all tolerance
token_mints = [] # Or pass --mint ADDRESS; use --all to select all non-quote balances.
[network]
trade_url = "https://api.zeroslip.ai"
stream_url = "wss://stream.zeroslip.ai/"
replay_url = "https://replay.pumpapi.io" # Replay retains the documented PumpAPI hostname.
rpc_url = "https://api.mainnet-beta.solana.com" # Override with your own RPC if rate limited.
.env.example
Read .env.example
.env.example
# Needed only for --live. Set ONE of these two Lightning credentials.
ZEROSLIP_API_KEY=
# ZEROSLIP_PRIVATE_KEY=
# Optional alternative to wallet.public_key in config.toml:
ZEROSLIP_WALLET_PUBLIC_KEY=
requirements.txt
Read requirements.txt
requirements.txt
aiohttp>=3.12,<4
websockets>=15,<16
cachetools>=5.5,<8
python-dotenv>=1.1,<2
orjson>=3.10,<4
zstandard>=0.23,<1
pyproject.toml
Read pyproject.toml
pyproject.toml
[project]
name = "zeroslip-bot-examples"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
"aiohttp>=3.12,<4",
"websockets>=15,<16",
"cachetools>=5.5,<8",
"python-dotenv>=1.1,<2",
"orjson>=3.10,<4",
"zstandard>=0.23,<1",
]
[tool.ruff]
line-length = 120
extend-exclude = ["tests/fixtures"]
[tool.ruff.lint]
select = ["E9", "F", "B904", "B023", "E501"]
[tool.ruff.format]
quote-style = "single"
Strategy notes
Read strategy-notes.md
strategy-notes.md
# Strategy notes
Use these notes when changing the bots; setup and commands are in the [README](../README.md).
## Streaming strategies
Launch sniping and copy trading follow Pump.fun and Pump-created PumpSwap pools, excluding Mayhem mode. Selected-token sniping follows SOL/USDC trades across supported venues and buys once per process. Launch and copy modes have no position cap.
Amounts and launch thresholds are SOL-equivalent. USDC sizing uses a trusted SOL/USDC stream pool, with an $80/SOL fallback until an update arrives. Sniper exits compare prices in the entry quote; other quote assets are ignored. Paper sniper entries fill at a subsequent trade price, subject to the same buy-slippage ceiling as replay. Migration events cannot fill an entry, and a rejected fill is counted as a miss. Copytrader paper entries use the observed buy price. Paper exits and running realized PnL exclude depth and fees; launch-mode paper results are not profitability estimates. Paper and replay use different fill timing and can disagree.
## Copy trading
The first-buy gate requires token balances summed across transaction wallets to match the purchased amount within tolerance. Entry sizing uses the aggregate event amount, including bundled trades.
The wallet that opens a position controls its exits. Sells apply its unrounded fraction of remaining tokens to your remaining position. Additional followed buys update the exit baseline without another buy in your wallet. Bundled exits use per-wallet breakdowns; ambiguous exits are skipped. Partial live exits round down to the token decimals reported at entry and send absolute token amounts with `denominatedInQuote: "false"`; final exits use `100%`. Missing entry decimals stops a live partial exit. Fractional percentage support is not assumed. Live bookkeeping changes after confirmation. Subunit partial exits send no order; a full exit still sells `100%`.
## Replay
Replay models qualifying launch entries, not `sniper.token_mint`. Missing archive hours fail unless `backtest.allow_gaps = true`.
Pending fills resolve before the incoming price update using the last known price. Events remain in recorded order. At the end, pending buys are filled and open positions close at their last price. These assumptions can be optimistic with sparse events or incomplete windows. Depth, network fees, and atomic Jito execution are not fully modeled.
## Sell balances
Sell previews use read-only RPC queries and report RPC errors directly. Both WSOL and USDC are always kept, even when explicitly selected. When upgrading an older configuration, `trade.quote_mint` is ignored with a warning; remove it when editing the file.