fix pump parser

SLIPPAGE_MAPPING.md

@@ -68,7 +68,7 @@ Interpretation:
 - Positive: execution is better than the user limit
 - Zero: execution lands exactly on the user limit
 - `10000`: user limit is effectively unbounded on the constrained side (for example `min_out = 0`)
-- Negative: this usually indicates an incorrect parser-side mapping or inconsistent source data
+- Negative raw headroom is clamped to `0` because successful-swap storage uses a non-negative bounded metric
 
 This definition makes `SlippageBps` a bounded "remaining headroom to the user's limit" metric for successful swaps:
 

enum.go

@@ -126,4 +126,17 @@ const (
 	TxEventBuyFailed  = "buy_failed"
 	TxEventSellFailed = "sell_failed"
 	TxEventBurn       = "burn"
+	TxEventCreate     = "create"
+	TxEventComplete   = "complete"
+	TxEventMigrate    = "migrate"
+	TxEventDeposit    = "deposit"
+	TxEventWithdraw   = "withdraw"
+	TxEventOpen       = "open"
+	TxEventClose      = "close"
+	TxEventClaimFee   = "claim_fee"
+
+	TxEventAddLiquidity           = "add_liquidity"
+	TxEventAddLiquidityOneSide    = "add_liquidity_one_side"
+	TxEventRemoveLiquidity        = "remove_liquidity"
+	TxEventRemoveLiquidityOneSide = "remove_liquidity_one_side"
 )

internal/example/cmd/main.go

@@ -25,7 +25,7 @@ func main() {
 	// laserstream-mainnet-slc.helius-rpc.com:80
 
 	ch := make(chan example.SubscriptionMessage, 1)
-	go example.RunLoopWithReConnect(context.Background(), "127.0.0.1:10001", parser.SolProgramPump, ch)
+	go example.RunLoopWithReConnect(context.Background(), "", "", parser.SolProgramPump, ch)
 	// var tokenTxs = make(map[string]*types.Tx)
 	// currentBlock := uint64(0)
 	for msg := range ch {
@@ -51,9 +51,24 @@ func main() {
 		//}
 
 		// 处理交易
-		if len(ptx.Swaps) > 0 && (ptx.Swaps[0].Program == parser.SolProgramPump || ptx.Swaps[0].Program == parser.SolProgramPumpAMM) {
+		if len(ptx.Swaps) > 0 {
-			fmt.Printf("success tx : %s, program: %s, event: %s, block: %d, tx: %s, base: %s, quote: %s \n", time.Now().Format("2006-01-02 15:04:05"), ptx.Swaps[0].Program, ptx.Swaps[0].Event, ptx.Block, ptx.GetTxHash(),
+			for _, swap := range ptx.Swaps {
-				ptx.Swaps[0].BaseAmount.Div(decimal.NewFromInt(1e6)), ptx.Swaps[0].QuoteAmount.Div(decimal.NewFromInt(1e9)))
+				if swap.SlippageBps.LessThan(decimal.Zero) || swap.SlippageBps.GreaterThan(decimal.NewFromInt(10000)) {
+					fmt.Printf("success tx : %s, program: %s, event: %s, block: %d, tx: %s, base: %s, quote: %s \n", time.Now().Format("2006-01-02 15:04:05"), swap.Program, swap.Event, ptx.Block, ptx.GetTxHash(),
+						swap.BaseAmount.Div(decimal.NewFromInt(1e6)), swap.QuoteAmount.Div(decimal.NewFromInt(1e9)))
+				}
+				if swap.SlippageBps.Equal(decimal.Zero) && (swap.Event == "buy" || swap.Event == "sell") {
+					fmt.Printf("zero success tx : %s, program: %s, event: %s, block: %d, tx: %s, base: %s, quote: %s, fix: %s, limit: %s, \n", time.Now().Format("2006-01-02 15:04:05"), swap.Program, swap.Event, ptx.Block, ptx.GetTxHash(),
+						swap.BaseAmount.Div(decimal.NewFromInt(1e6)), swap.QuoteAmount.Div(decimal.NewFromInt(1e9)), swap.FixedAmount.String(), swap.LimitAmount.String())
+				}
+			}
+			if len(ptx.Swaps) > 0 {
+				_, err := parser.EncodeTxBinary(ptx)
+				if err != nil {
+					fmt.Printf("success tx : %s, , block: %d, tx: %s, err: %s \n", time.Now().Format("2006-01-02 15:04:05"), ptx.Block, ptx.GetTxHash(), err.Error())
+				}
+			}
+
 		}
 		// currentBlock = ptx.Block
 		//

internal/example/yellowstone.go

@@ -45,9 +45,11 @@ type Client struct {
 	firstMessage bool
 
 	handler Handler
+
+	xToken string
 }
 
-func NewClientWithPumpSwap(endpoint string, ch chan SubscriptionMessage) *Client {
+func NewClientWithPumpSwap(endpoint string, xtoken string, ch chan SubscriptionMessage) *Client {
 	var subscription pb.SubscribeRequest
 
 	//var failed = true
@@ -58,10 +60,10 @@ func NewClientWithPumpSwap(endpoint string, ch chan SubscriptionMessage) *Client
 		Vote: &vote,
 	}
 
-	subscription.Transactions["transactions_sub"].AccountInclude = []string{
+	//subscription.Transactions["transactions_sub"].AccountInclude = []string{
-		"pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA", //Pump AMM
+	//	"pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA", //Pump AMM
-		"6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P", //Pump
+	//	"6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P", //Pump
-	}
+	//}
 	subscription.BlocksMeta = make(map[string]*pb.SubscribeRequestFilterBlocksMeta)
 	subscription.BlocksMeta["block_meta"] = &pb.SubscribeRequestFilterBlocksMeta{}
 
@@ -72,6 +74,7 @@ func NewClientWithPumpSwap(endpoint string, ch chan SubscriptionMessage) *Client
 		lastReceiveTime: time.Now(),
 		subStatus:       false,
 		subscription:    &subscription,
+		xToken:          xtoken,
 	}
 	c.handler = NewPumpHandler(func(tx *types.Tx) {
 		c.sendTx(tx)
@@ -112,12 +115,12 @@ func NewClientWithLaunchLab(endpoint string, ch chan SubscriptionMessage) *Clien
 	return c
 }
 
-func RunLoopWithReConnect(ctx context.Context, endpoint, program string, ch chan SubscriptionMessage) {
+func RunLoopWithReConnect(ctx context.Context, endpoint, token, program string, ch chan SubscriptionMessage) {
 	var client *Client
 	if program == types.SolProgramRaydiumLaunchLab {
 		client = NewClientWithLaunchLab(endpoint, ch)
 	} else {
-		client = NewClientWithPumpSwap(endpoint, ch)
+		client = NewClientWithPumpSwap(endpoint, token, ch)
 	}
 	for {
 		select {
@@ -206,12 +209,13 @@ func (c *Client) grpcSubscribe(ctx context.Context, conn *grpc.ClientConn) error
 	log.Printf("Subscription request: %s", string(subscriptionJson))
 
 	// Set up the subscription request
-	//if *token != "" {
+	if c.xToken != "" {
-	//	md := metadata.New(map[string]string{"x-token": *token})
+		fmt.Println("xtoken", c.xToken)
-	//	ctx = metadata.NewOutgoingContext(ctx, md)
+		md := metadata.New(map[string]string{"x-token": c.xToken})
-	//}
-	md := metadata.New(map[string]string{"x-token": "5adcf1f9-5719-43d1-bf3f-c2d4e1e5f94d"})
 		ctx = metadata.NewOutgoingContext(ctx, md)
+	}
+	//md := metadata.New(map[string]string{"x-token": "5adcf1f9-5719-43d1-bf3f-c2d4e1e5f94d"})
+	//ctx = metadata.NewOutgoingContext(ctx, md)
 
 	stream, err := client.Subscribe(ctx)
 	if err != nil {

internal/test2/test.go

@@ -2,29 +2,18 @@ package main
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"log"
-	"log/slog"
-	"strings"
-	"time"
 
 	"github.com/gagliardetto/solana-go"
 	"github.com/gagliardetto/solana-go/rpc"
-	"github.com/jackc/pgtype"
-	"github.com/shopspring/decimal"
 	solana_parser "github.com/thloyi/pump-parser"
-	"gorm.io/driver/postgres"
-	"gorm.io/gorm"
-	"gorm.io/gorm/logger"
 )
 
 var ()
 
 func main() {
 
-	var slot uint64 = 403021435
+	var slot uint64 = 414432104
-	var data = NewBlockData(decimal.NewFromFloat(100.0))
 	client := rpc.New("https://staked.helius-rpc.com?api-key=5adcf1f9-5719-43d1-bf3f-c2d4e1e5f94d")
 	var rewards = false
 	var version uint64 = 0
@@ -42,7 +31,7 @@ func main() {
 	}
 	solana_parser.EnableAllParsers()
 
-	var txs []*solana_parser.Tx
+	var txs []solana_parser.Tx
 	for i, tx := range blocks.Transactions {
 		var blockTime uint64
 		if blocks.BlockTime != nil {
@@ -61,766 +50,11 @@ func main() {
 			fmt.Println("parse tx error:", i, rawTx.TxHash(), err)
 			break
 		}
-		txs = append(txs, parsedTx)
+		txs = append(txs, *parsedTx)
 	}
-	for _, result := range txs {
+	_, err = solana_parser.EncodeTxsBinary(txs)
-		swapsLen := len(result.Swaps)
-		for i := 0; i < swapsLen; i++ {
-			action := result.Swaps[i]
-			var actions []solana_parser.Swap = make([]solana_parser.Swap, 0, 2)
-			actions = append(actions, action)
-			if i+1 < swapsLen {
-				nextAction := result.Swaps[i+1]
-				if action.Event == "buy" && nextAction.Event == "complete" &&
-					action.Program == solana_parser.SolProgramPump &&
-					nextAction.Program == solana_parser.SolProgramPump &&
-					action.BaseMint == nextAction.BaseMint {
-					actions = append(actions, nextAction)
-					i++
-				}
-				if action.Event == "migrate" && nextAction.Event == "create" &&
-					action.Program == solana_parser.SolProgramPump &&
-					nextAction.Program == solana_parser.SolProgramPumpAMM &&
-					action.BaseMint == nextAction.BaseMint {
-					actions = append(actions, nextAction)
-					i++
-				}
-			}
-			if err = HandleAction(context.Background(), result, actions, data); err != nil {
-				//h.logger.Errorf("handle action error: %s - %v", result.RawTx.Transaction.Signatures[0].String(), err)
-				fmt.Println("parse action error:", "tx", result.GetTxHash(), "i", i, "err", err)
-			}
-		}
-
-	}
-	fmt.Println("slot", slot, "tx count: ", len(data.Txs))
-
-}
-
-var (
-	meteoraDammV2Program = solana.MustPublicKeyFromBase58("cpamdpZCGKUy5JxQXB4dcpGPiikHawvSWAd6mEn1sGG")
-	raydiumCPmmProgramID = solana.MustPublicKeyFromBase58("CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C")
-)
-
-func HandleAction(ctx context.Context, tx *solana_parser.Tx, swaps []solana_parser.Swap, data *BlockData) error {
-	swapLen := len(swaps)
-	if len(swaps) == 0 {
-		return nil
-	}
-	if swaps[0].BaseMint != solana_parser.WSOL && swaps[0].QuoteMint != solana_parser.WSOL && !swaps[0].QuoteMint.IsZero() {
-		return nil
-	}
-
-	if len(swaps) == 0 {
-		return nil
-	}
-	event := swaps[0].Event
-	swap := swaps[0]
-	action := SwapGetter{swap}
-	switch event {
-	case "buy", "sell":
-
-		data.AppendTx(action.GetTx(tx, uint64(swap.TxIndex), data.Price))
-		if swap.Program == solana_parser.SolProgramPump {
-			if swapLen == 2 && swaps[1].Event == "complete" {
-				t := pgtype.Timestamptz{}
-				t.Set(time.Unix(tx.BlockAt, 0))
-				data.AppendAction(Action{
-					Maker:   swaps[1].User.String(),
-					Token:   swaps[1].BaseMint.String(),
-					Pair:    swaps[1].Pool.String(),
-					Action:  "pump-migrate",
-					Block:   tx.Block,
-					BlockAt: t,
-					TxHash:  tx.GetTxHash(),
-				})
-			}
-		}
-		return data.SetPair(action, tx.Block, "")
-
-	case "create":
-		pair, err := action.GetPair(tx.Block, "")
 	if err != nil {
-			return err
+		fmt.Println("EncodeTxsBinary err", err)
-		}
-		data.AppendTx(action.GetTx(tx, uint64(swap.TxIndex), data.Price))
-		data.Pairs[pair.Address] = *pair
-	case "add_liquidity", "remove_liquidity", "deposit", "withdraw", "add", "remove":
-		liquidityTx, err := action.GetLiquidityTx(tx, uint64(swap.TxIndex))
-		if liquidityTx == nil {
-			return err
-		}
-		data.AppendTx(*liquidityTx)
-		return data.SetPair(action, tx.Block, "")
 	}
 
-	if event != "migrate" {
-		return nil
-	}
-	if swap.Program == solana_parser.SolProgramPump {
-		t := pgtype.Timestamptz{}
-		t.Set(time.Unix(tx.BlockAt, 0))
-		if swapLen == 2 && swaps[1].Event == "create" && swaps[1].Program == solana_parser.SolProgramPumpAMM && swaps[1].BaseMint == swap.BaseMint {
-			tokenMint := swap.BaseMint.String()
-			data.AppendAction(Action{
-				Maker:   swap.User.String(),
-				Token:   tokenMint,
-				Pair:    swaps[1].Pool.String(),
-				Action:  "on-pumpswap",
-				Block:   tx.Block,
-				BlockAt: t,
-				TxHash:  tx.GetTxHash(),
-			})
-			data.NewRaydium = append(data.NewRaydium, tokenMint)
-		}
-	} else if swap.Program == solana_parser.SolProgramRaydiumLaunchLab || swap.Program == solana_parser.SolProgramRaydiumLaunchLabBonk {
-		t := pgtype.Timestamptz{}
-		t.Set(time.Unix(tx.BlockAt, 0))
-		var actionType string
-		if action.MigrateTopProgram == raydiumCPmmProgramID {
-			actionType = "on-raydium-cpmm"
-		} else {
-			actionType = "on-raydium-amm"
-		}
-		data.AppendAction(Action{
-			Maker:   action.User.String(),
-			Token:   action.BaseMint.String(),
-			Pair:    action.MigrateToPool.String(),
-			Action:  actionType,
-			Block:   tx.Block,
-			BlockAt: t,
-			TxHash:  tx.GetTxHash(),
-		})
-	} else if swap.Program == solana_parser.SolProgramMeteoraBondingCurve {
-		t := pgtype.Timestamptz{}
-		t.Set(time.Unix(tx.BlockAt, 0))
-		var actionType string
-		if swap.MigrateTopProgram == meteoraDammV2Program {
-			actionType = "on-meteora-amm-v2"
-		} else {
-			actionType = "on-meteora-amm-v1"
-		}
-		data.AppendAction(Action{
-			Maker:   action.User.String(),
-			Token:   action.BaseMint.String(),
-			Pair:    action.MigrateToPool.String(),
-			Action:  actionType,
-			Block:   uint64(tx.Block),
-			BlockAt: t,
-			TxHash:  tx.GetTxHash(),
-		})
-	}
-
-	return nil
-}
-
-type Pair struct {
-	Id        string `gorm:"column:id;primaryKey;default:uuid_generate_v4()"`
-	Address   string
-	Name      string
-	Token0    string
-	Token1    string
-	LpToken   string
-	ChainId   int64
-	Reserve0  decimal.Decimal
-	Reserve1  decimal.Decimal
-	Block     uint64
-	BlockAt   *pgtype.Timestamptz `gorm:"column:block_at;default:NULL" json:"block_at,omitempty"`
-	CreatedAt *pgtype.Timestamptz `gorm:"autoCreateTime" json:"created_at,omitempty"`
-	SortId    uint64
-	Program   string
-
-	IsCreate bool `gorm:"-"`
-	//TokenObj   *Token `gorm:"-" json:"token_obj,omitempty"`
-	UpdateSlot uint64 `gorm:"-"`
-	InDB       bool   `gorm:"-"`
-}
-
-type Tx struct {
-	Id              pgtype.UUID         `gorm:"column:id;primaryKey;default:uuid_generate_v4()" json:"-"`
-	PairAddress     string              `json:"pair_address"`
-	Maker           string              `json:"maker"`
-	Token0Address   string              `json:"token0_address"`
-	Token1Address   string              `json:"token1_address"`
-	Token0Amount    decimal.Decimal     `json:"token0Amount" gorm:"column:token0_amount;type:numeric"`
-	Token1Amount    decimal.Decimal     `json:"token1Amount" gorm:"column:token1_amount;type:numeric"`
-	PriceUsd        decimal.Decimal     `json:"price_usd" gorm:"column:price_usd;type:numeric"`
-	AmountUsd       decimal.Decimal     `json:"amount_usd"  gorm:"column:amount_usd;type:numeric"`
-	Block           uint64              `json:"block"`
-	BlockIndex      uint64              `json:"index"`
-	Event           string              `json:"event"`
-	TxHash          string              `json:"tx_hash"`
-	TxIndex         uint64              `json:"topic_index"`
-	Program         string              `json:"program"`
-	BlockAt         pgtype.Timestamptz  `gorm:"column:block_at;default:NULL" json:"block_at"`
-	CreatedAt       *pgtype.Timestamptz `gorm:"autoCreateTime" json:"-"`
-	TotalSupply     string              `gorm:"total_supply"`
-	AfterReserve0   string              `gorm:"after_reserve0"`
-	AfterReserve1   string              `gorm:"after_reserve1"`
-	PositionChange  int64               `gorm:"position_change"`
-	Platform        string              `gorm:"column:tx_platform;type:platform;default:'none'" json:"tx_platform"`
-	PlatformFee     decimal.Decimal     `gorm:"-" json:"-"` // TODO: save to db
-	CUPrice         decimal.Decimal     `gorm:"column:tx_cu_price;type:numeric" json:"tx_cu_price"`
-	MevAgent        string              `gorm:"column:tx_mev_agent;type:mev_agent;default:'none'" json:"tx_mev_agent"`
-	MevAgentFee     decimal.Decimal     `gorm:"column:tx_mev_agent_fee;type:numeric" json:"tx_mev_agent_fee"`
-	AfterSOLBalance decimal.Decimal     `gorm:"column:after_sol_balance;type:numeric" json:"after_sol_balance"`
-	EntryContract   string              `gorm:"column:tx_entry_contract;type:entry_contract;default:'none'" json:"tx_entry_contract"`
-}
-
-type Action struct {
-	Id        pgtype.UUID         `gorm:"column:id;primaryKey;default:uuid_generate_v4()" json:"-"`
-	Maker     string              `json:"maker"`
-	Token     string              `json:"token"`
-	Pair      string              `json:"pair"`
-	Action    string              `json:"action"`
-	Block     uint64              `json:"block"`
-	BlockAt   pgtype.Timestamptz  `json:"block_at"`
-	TxHash    string              `json:"tx_hash"`
-	CreatedAt *pgtype.Timestamptz `gorm:"autoCreateTime" json:"-"`
-}
-
-type BlockData struct {
-	Pairs      map[string]Pair
-	Txs        []Tx
-	Actions    []Action
-	Price      decimal.Decimal
-	NewRaydium []string
-}
-
-func NewBlockData(price decimal.Decimal) *BlockData {
-	return &BlockData{
-		Pairs:      make(map[string]Pair),
-		Txs:        make([]Tx, 0),
-		Actions:    make([]Action, 0),
-		Price:      price,
-		NewRaydium: make([]string, 0),
-	}
-}
-
-func (bd *BlockData) AppendTx(tx Tx) {
-	bd.Txs = append(bd.Txs, tx)
-}
-
-func (bd *BlockData) AppendAction(action Action) {
-	bd.Actions = append(bd.Actions, action)
-}
-
-func (bd *BlockData) SetPair(action SwapGetter, block uint64, _ string) error {
-	pair, err := action.GetPair(block, "")
-	if err != nil {
-		return err
-	}
-	bd.Pairs[pair.Address] = *pair
-	return nil
-}
-
-type SwapGetter struct {
-	solana_parser.Swap
-}
-
-const (
-	PositionChangeNone = int64(iota)
-	PositionChangeNewBuy
-	PositionChangeBuyMore
-	PositionChangeSellPart
-	PositionChangeSellAll
-)
-
-func (spg SwapGetter) GetLiquidityTx(tx *solana_parser.Tx, index uint64) (*Tx, error) {
-	if spg.BaseMint != solana.WrappedSol && spg.QuoteMint != solana.WrappedSol {
-		return nil, nil
-	}
-
-	var (
-		token0  string
-		amount0 decimal.Decimal
-		amount1 decimal.Decimal
-		pool0   decimal.Decimal
-		pool1   decimal.Decimal
-
-		event string
-	)
-
-	if spg.BaseMint == solana.WrappedSol {
-		amount0 = spg.QuoteAmount.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		amount1 = spg.BaseAmount.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		token0 = spg.QuoteMint.String()
-		pool0 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		pool1 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-	} else {
-		amount0 = spg.BaseAmount.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		amount1 = spg.QuoteAmount.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		token0 = spg.BaseMint.String()
-		pool0 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		pool1 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-	}
-	if spg.Event == "deposit" || spg.Event == "add" || spg.Event == "add_liquidity" || spg.Event == "add_liquidity_one_side" {
-		event = "add"
-	} else if spg.Event == "withdraw" || spg.Event == "remove" || spg.Event == "remove_liquidity" || spg.Event == "remove_liquidity_one_side" {
-		event = "remove"
-	}
-	if event == "" {
-		return nil, nil
-	}
-
-	mevName, mevFee := tx.CheckMevAgent()
-	platformName, platformFee := tx.CheckPlatform(spg.Swap)
-
-	pairString := ""
-	if spg.Program == solana_parser.SolProgramPump {
-		pairString = spg.BaseMint.String()
-	} else {
-		pairString = spg.Pool.String()
-	}
-	t := pgtype.Timestamptz{}
-	_ = t.Set(time.Unix(tx.BlockAt, 0))
-	return &Tx{
-		PairAddress:     pairString,
-		Maker:           spg.User.String(),
-		Token0Address:   token0,
-		Token1Address:   "So11111111111111111111111111111111111111112",
-		Token0Amount:    amount0,
-		Token1Amount:    amount1,
-		Block:           tx.Block,
-		BlockIndex:      tx.BlockIndex,
-		Event:           event,
-		TxHash:          tx.GetTxHash(),
-		TxIndex:         index,
-		BlockAt:         t,
-		Program:         spg.Program,
-		AfterReserve0:   pool0.String(),
-		AfterReserve1:   pool1.String(),
-		Platform:        platformName,
-		PlatformFee:     platformFee,
-		CUPrice:         tx.CUPrice,
-		MevAgent:        mevName,
-		MevAgentFee:     mevFee,
-		AfterSOLBalance: spg.AfterSOLBalance,
-		EntryContract:   spg.CheckEntryContract(),
-	}, nil
-}
-
-func (spg SwapGetter) GetTx(tx *solana_parser.Tx, index uint64, price decimal.Decimal) Tx {
-	var (
-		token0  string
-		amount0 decimal.Decimal
-		amount1 decimal.Decimal
-		pool0   decimal.Decimal
-		pool1   decimal.Decimal
-
-		event string
-	)
-
-	if spg.BaseMint == solana.WrappedSol {
-		amount0 = spg.QuoteAmount.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		amount1 = spg.BaseAmount.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		token0 = spg.QuoteMint.String()
-		pool0 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		pool1 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		if spg.Event == "buy" {
-			event = "sell"
-		} else if spg.Event == "sell" {
-			event = "buy"
-		}
-	} else {
-		amount0 = spg.BaseAmount.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		amount1 = spg.QuoteAmount.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		token0 = spg.BaseMint.String()
-		pool0 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-		pool1 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-		event = spg.Event
-	}
-
-	priceUsd := decimal.Zero
-	if amount0.GreaterThan(priceUsd) {
-		priceUsd = amount1.Div(amount0).Mul(price)
-	}
-	pc := PositionChangeNone
-	if event == "buy" {
-		pc = PositionChangeNewBuy
-		if spg.BaseMint == solana.WrappedSol {
-			if spg.UserQuoteBalance.GreaterThan(spg.QuoteAmount) {
-				pc = PositionChangeBuyMore
-			}
-		} else {
-			if spg.UserBaseBalance.GreaterThan(spg.BaseAmount) {
-				pc = PositionChangeBuyMore
-			}
-		}
-	} else if event == "sell" {
-		pc = PositionChangeSellPart
-		if spg.BaseMint == solana.WrappedSol {
-			if spg.UserQuoteBalance.Div(decimal.New(1, int32(spg.QuoteMintDecimals))).LessThan(decimal.NewFromFloat(0.01)) {
-				pc = PositionChangeSellAll
-			}
-		} else {
-			if spg.UserBaseBalance.Div(decimal.New(1, int32(spg.BaseMintDecimals))).LessThan(decimal.NewFromFloat(0.01)) {
-				pc = PositionChangeSellAll
-			}
-		}
-	}
-
-	mevName, mevFee := tx.CheckMevAgent()
-	platformName, platformFee := tx.CheckPlatform(spg.Swap)
-
-	if mevName == "" {
-		mevName = "none"
-	}
-	if mevName == "unknown" {
-		mevName = "none"
-		mevFee = decimal.Zero
-	}
-	pairString := ""
-	if spg.Program == solana_parser.SolProgramPump {
-		pairString = spg.BaseMint.String()
-	} else {
-		pairString = spg.Pool.String()
-	}
-	t := pgtype.Timestamptz{}
-	_ = t.Set(time.Unix(tx.BlockAt, 0))
-
-	return Tx{
-		PairAddress:     pairString,
-		Maker:           spg.User.String(),
-		Token0Address:   token0,
-		Token1Address:   "So11111111111111111111111111111111111111112",
-		Token0Amount:    amount0,
-		Token1Amount:    amount1,
-		PriceUsd:        priceUsd,
-		AmountUsd:       amount1.Mul(price),
-		Block:           tx.Block,
-		BlockIndex:      tx.BlockIndex,
-		Event:           event,
-		TxHash:          tx.GetTxHash(),
-		TxIndex:         index,
-		BlockAt:         t,
-		Program:         spg.Program,
-		AfterReserve0:   pool0.String(),
-		AfterReserve1:   pool1.String(),
-		PositionChange:  pc,
-		Platform:        platformName,
-		PlatformFee:     platformFee,
-		CUPrice:         tx.CUPrice,
-		MevAgent:        mevName,
-		MevAgentFee:     mevFee,
-		AfterSOLBalance: spg.AfterSOLBalance,
-		EntryContract:   spg.CheckEntryContract(),
-	}
-}
-
-func (spg SwapGetter) GetPair(slot uint64, _ string) (*Pair, error) {
-	//pump amm
-	if spg.Program == solana_parser.SolProgramPump {
-		tokenMint := spg.BaseMint.String()
-		return &Pair{
-			Address:    tokenMint,
-			Token0:     tokenMint,
-			Token1:     "So11111111111111111111111111111111111111112",
-			ChainId:    900,
-			Reserve0:   spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals))),
-			Reserve1:   spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals))),
-			IsCreate:   spg.Event == "create",
-			Program:    spg.Program,
-			UpdateSlot: slot,
-		}, nil
-	} else {
-		var (
-			token0  string
-			amount0 decimal.Decimal
-			amount1 decimal.Decimal
-		)
-		if spg.BaseMint.IsZero() || spg.QuoteMint.IsZero() {
-			return nil, errors.New("base mint or quote mint is empty")
-		}
-
-		if spg.BaseMint == solana.WrappedSol {
-			amount0 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-			amount1 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-			//decimal0 = spg.QuoteMintDecimals
-			token0 = spg.QuoteMint.String()
-		} else {
-			amount0 = spg.BaseReserve.Div(decimal.New(1, int32(spg.BaseMintDecimals)))
-			amount1 = spg.QuoteReserve.Div(decimal.New(1, int32(spg.QuoteMintDecimals)))
-			//decimal0 = a.BaseDecimals
-			token0 = spg.BaseMint.String()
-		}
-
-		return &Pair{
-			Address:    spg.Pool.String(),
-			LpToken:    spg.LpMint.String(),
-			Token0:     token0,
-			Token1:     "So11111111111111111111111111111111111111112",
-			ChainId:    900,
-			Reserve0:   amount0,
-			Reserve1:   amount1,
-			IsCreate:   false,
-			Program:    spg.Program,
-			UpdateSlot: slot,
-		}, nil
-	}
-}
-
-func getBlockTxsFromDb(db *gorm.DB, block uint64) ([]Tx, error) {
-	var txs []Tx
-	result := db.Table("tx").Where("block = ?", block).Find(&txs)
-	return txs, result.Error
-}
-
-func getBlockActionsFromDb(db *gorm.DB, block uint64) ([]Action, error) {
-	var txs []Action
-	result := db.Table("action").Where("block = ?", block).Find(&txs)
-	return txs, result.Error
-}
-
-type dbLog struct {
-	logger *slog.Logger
-}
-
-func (l *dbLog) Printf(format string, args ...interface{}) {
-	l.logger.Info(fmt.Sprintf(format, args...))
-}
-
-func newDbLog() *dbLog {
-	return &dbLog{logger: slog.Default()}
-}
-
-func NewGorm(dsn string) *gorm.DB {
-	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{
-		Logger: logger.New(newDbLog(), logger.Config{
-			Colorful:                  false,
-			LogLevel:                  logger.Warn,
-			SlowThreshold:             time.Second * 10,
-			IgnoreRecordNotFoundError: true,
-		}),
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	return db
-}
-
-func compareTxs(dbTxs []Tx, dataTxs []Tx) (diff int, missing int) {
-	dataByHash := make(map[string][]Tx, len(dataTxs))
-	for _, tx := range dataTxs {
-		dataByHash[fmt.Sprintf("%s-%d", tx.TxHash, tx.TxIndex)] = append(dataByHash[fmt.Sprintf("%s-%d", tx.TxHash, tx.TxIndex)], tx)
-	}
-
-	for _, dbTx := range dbTxs {
-		candidates := dataByHash[fmt.Sprintf("%s-%d", dbTx.TxHash, dbTx.TxIndex)]
-		if len(candidates) == 0 {
-			missing++
-			log.Printf("missing tx: %s", txCompareString(dbTx))
-			continue
-		}
-		matched := false
-		for _, dataTx := range candidates {
-			if txEqualWithoutHash(dbTx, dataTx) {
-				matched = true
-				break
-			}
-		}
-		if !matched {
-			diff++
-			log.Printf("tx diff hash=%s, program=%s, event:%s: %s, ", dbTx.TxHash, dbTx.Program, dbTx.Event, txCompareDiffString(dbTx, candidates[0]))
-		}
-	}
-	log.Printf("compare txs done: db=%d parsed=%d missing=%d diff=%d", len(dbTxs), len(dataTxs), missing, diff)
-	return diff, missing
-}
-
-func withinOnePercentDecimal(a decimal.Decimal, b decimal.Decimal) bool {
-	if a.IsZero() {
-		return b.IsZero()
-	}
-	diff := a.Sub(b).Abs()
-	threshold := a.Abs().Mul(decimal.NewFromFloat(0.03))
-	return diff.LessThanOrEqual(threshold)
-}
-
-func withinOnePercentStringDecimal(a string, b string) bool {
-	ad, errA := decimal.NewFromString(a)
-	bd, errB := decimal.NewFromString(b)
-	if errA != nil || errB != nil {
-		return a == b
-	}
-	return withinOnePercentDecimal(ad, bd)
-}
-
-func txEqualWithoutHash(a Tx, b Tx) bool {
-	//mevMatch := a.MevAgent == b.MevAgent || (a.MevAgent == "none" && b.MevAgent == "unknown") || (a.MevAgent == "unknown" && b.MevAgent == "none")
-	//mevNone := a.MevAgent == "none" || a.MevAgent == "unknown"
-
-	return a.PairAddress == b.PairAddress &&
-		a.Token1Address == b.Token1Address &&
-		(a.Token0Address == "" || a.Token0Address == b.Token0Address) &&
-		//a.Maker == b.Maker &&
-		(a.Token0Address == "" || withinOnePercentDecimal(a.Token0Amount, b.Token0Amount)) &&
-		withinOnePercentDecimal(a.Token1Amount, b.Token1Amount) &&
-		a.Block == b.Block &&
-		a.BlockIndex == b.BlockIndex &&
-		a.Event == b.Event &&
-		a.TxIndex == b.TxIndex &&
-		a.Program == b.Program &&
-		(a.Program == solana_parser.SolProgramPumpAMM || a.Program == solana_parser.SolProgramPump || a.Program == solana_parser.SolProgramMeteoraPools || (a.Event == "create") || withinOnePercentStringDecimal(a.AfterReserve0, b.AfterReserve0)) &&
-		(a.Program == solana_parser.SolProgramPumpAMM || a.Program == solana_parser.SolProgramPump || a.Program == solana_parser.SolProgramMeteoraPools || (a.Event == "create") || withinOnePercentStringDecimal(a.AfterReserve1, b.AfterReserve1)) &&
-		// a.PositionChange == b.PositionChange &&
-		(a.Platform == b.Platform || (a.Platform == "photon" && b.Platform == "fake") || (a.Platform == "trojan" && b.Platform == "fake")) &&
-		a.CUPrice.String() == b.CUPrice.String() // &&
-	//mevMatch &&
-	//(mevNone || a.MevAgentFee.String() == b.MevAgentFee.String()) &&
-	//(a.Program == solana_parser.SolProgramRaydiumV4 || a.AfterSOLBalance.String() == b.AfterSOLBalance.String())
-	//&&
-	//	a.EntryContract == b.EntryContract
-}
-
-func txCompareDiffString(a Tx, b Tx) string {
-	var diffs []string
-	if a.PairAddress != b.PairAddress {
-		diffs = append(diffs, fmt.Sprintf("PairAddress db=%s data=%s", a.PairAddress, b.PairAddress))
-	}
-	//if a.Maker != b.Maker {
-	//	diffs = append(diffs, fmt.Sprintf("Maker db=%s, data=%s", a.Maker, b.Maker))
-	//}
-	if a.Token1Address != b.Token1Address {
-		diffs = append(diffs, fmt.Sprintf("Token1Address db=%s data=%s", a.Token1Address, b.Token1Address))
-	}
-	if a.Token0Address != b.Token0Address {
-		diffs = append(diffs, fmt.Sprintf("Token0Address db=%s data=%s", a.Token0Address, b.Token0Address))
-	}
-	if !withinOnePercentDecimal(a.Token0Amount, b.Token0Amount) {
-		diffs = append(diffs, fmt.Sprintf("Token0Amount db=%s data=%s", a.Token0Amount.String(), b.Token0Amount.String()))
-	}
-	if !withinOnePercentDecimal(a.Token1Amount, b.Token1Amount) {
-		diffs = append(diffs, fmt.Sprintf("Token1Amount db=%s data=%s", a.Token1Amount.String(), b.Token1Amount.String()))
-	}
-	if a.Block != b.Block {
-		diffs = append(diffs, fmt.Sprintf("Block db=%d data=%d", a.Block, b.Block))
-	}
-	if a.BlockIndex != b.BlockIndex {
-		diffs = append(diffs, fmt.Sprintf("BlockIndex db=%d data=%d", a.BlockIndex, b.BlockIndex))
-	}
-	if a.Event != b.Event {
-		diffs = append(diffs, fmt.Sprintf("Event db=%s data=%s", a.Event, b.Event))
-	}
-	if a.TxIndex != b.TxIndex {
-		diffs = append(diffs, fmt.Sprintf("TxIndex db=%d data=%d", a.TxIndex, b.TxIndex))
-	}
-	if a.Program != b.Program {
-		diffs = append(diffs, fmt.Sprintf("Program db=%s data=%s", a.Program, b.Program))
-	}
-	if !withinOnePercentStringDecimal(a.AfterReserve0, b.AfterReserve0) {
-		diffs = append(diffs, fmt.Sprintf("AfterReserve0 db=%s data=%s", a.AfterReserve0, b.AfterReserve0))
-	}
-	if !withinOnePercentStringDecimal(a.AfterReserve1, b.AfterReserve1) {
-		diffs = append(diffs, fmt.Sprintf("AfterReserve1 db=%s data=%s", a.AfterReserve1, b.AfterReserve1))
-	}
-	//if a.PositionChange != b.PositionChange {
-	//	diffs = append(diffs, fmt.Sprintf("PositionChange db=%d data=%d", a.PositionChange, b.PositionChange))
-	//}
-	if a.Platform != b.Platform {
-		diffs = append(diffs, fmt.Sprintf("Platform db=%s data=%s", a.Platform, b.Platform))
-	}
-	if a.CUPrice.String() != b.CUPrice.String() {
-		diffs = append(diffs, fmt.Sprintf("CUPrice db=%s data=%s", a.CUPrice.String(), b.CUPrice.String()))
-	}
-	//if a.MevAgent != b.MevAgent {
-	//	diffs = append(diffs, fmt.Sprintf("MevAgent db=%s data=%s", a.MevAgent, b.MevAgent))
-	//}
-	//if a.MevAgentFee.String() != b.MevAgentFee.String() {
-	//	diffs = append(diffs, fmt.Sprintf("MevAgentFee db=%s data=%s", a.MevAgentFee.String(), b.MevAgentFee.String()))
-	//}
-	//if a.AfterSOLBalance.String() != b.AfterSOLBalance.String() {
-	//	diffs = append(diffs, fmt.Sprintf("AfterSOLBalance db=%s data=%s", a.AfterSOLBalance.String(), b.AfterSOLBalance.String()))
-	//}
-	//if a.EntryContract != b.EntryContract {
-	//	diffs = append(diffs, fmt.Sprintf("EntryContract db=%s data=%s", a.EntryContract, b.EntryContract))
-	//}
-	return strings.Join(diffs, "; ")
-}
-
-func compareActions(dbActions []Action, dataActions []Action) (diff, missing int) {
-	dataByHash := make(map[string][]Action, len(dataActions))
-	for _, action := range dataActions {
-		dataByHash[action.TxHash] = append(dataByHash[action.TxHash], action)
-	}
-
-	for _, dbAction := range dbActions {
-		candidates := dataByHash[dbAction.TxHash]
-		if len(candidates) == 0 {
-			missing++
-			log.Printf("missing action: %s", actionCompareString(dbAction))
-			continue
-		}
-		matched := false
-		for _, dataAction := range candidates {
-			if actionEqualWithoutHash(dbAction, dataAction) {
-				matched = true
-				break
-			}
-		}
-		if !matched {
-			diff++
-			log.Printf("action diff hash=%s: %s", dbAction.TxHash, actionCompareDiffString(dbAction, candidates[0]))
-		}
-	}
-	log.Printf("compare actions done: db=%d parsed=%d missing=%d diff=%d", len(dbActions), len(dataActions), missing, diff)
-	return diff, missing
-}
-
-func actionEqualWithoutHash(a Action, b Action) bool {
-	return a.Maker == b.Maker &&
-		a.Token == b.Token &&
-		a.Pair == b.Pair &&
-		a.Action == b.Action &&
-		a.Block == b.Block
-}
-
-func actionCompareDiffString(a Action, b Action) string {
-	var diffs []string
-	if a.Maker != b.Maker {
-		diffs = append(diffs, fmt.Sprintf("Maker db=%s data=%s", a.Maker, b.Maker))
-	}
-	if a.Token != b.Token {
-		diffs = append(diffs, fmt.Sprintf("Token db=%s data=%s", a.Token, b.Token))
-	}
-	if a.Pair != b.Pair {
-		diffs = append(diffs, fmt.Sprintf("Pair db=%s data=%s", a.Pair, b.Pair))
-	}
-	if a.Action != b.Action {
-		diffs = append(diffs, fmt.Sprintf("Action db=%s data=%s", a.Action, b.Action))
-	}
-	if a.Block != b.Block {
-		diffs = append(diffs, fmt.Sprintf("Block db=%d data=%d", a.Block, b.Block))
-	}
-	return strings.Join(diffs, "; ")
-}
-
-func actionCompareString(action Action) string {
-	return fmt.Sprintf("Maker=%s Token=%s Pair=%s Action=%s Block=%d TxHash=%s", action.Maker, action.Token, action.Pair, action.Action, action.Block, action.TxHash)
-}
-
-func txCompareString(tx Tx) string {
-	return fmt.Sprintf(
-		"tx.Program=%s TxHash=%s PairAddress=%s Token1Address=%s Token0Amount=%s Token1Amount=%s Block=%d BlockIndex=%d Event=%s  TxIndex=%d AfterReserve0=%s AfterReserve1=%s PositionChange=%d Platform=%s CUPrice=%s MevAgent=%s MevAgentFee=%s AfterSOLBalance=%s EntryContract=%s",
-		tx.Program,
-		tx.TxHash,
-		tx.PairAddress,
-		tx.Token1Address,
-		tx.Token0Amount.String(),
-		tx.Token1Amount.String(),
-		tx.Block,
-		tx.BlockIndex,
-		tx.Event,
-		tx.TxIndex,
-		tx.AfterReserve0,
-		tx.AfterReserve1,
-		tx.PositionChange,
-		tx.Platform,
-		tx.CUPrice.String(),
-		tx.MevAgent,
-		tx.MevAgentFee.String(),
-		tx.AfterSOLBalance.String(),
-		tx.EntryContract,
-	)
 }

meteoradamm.go

@@ -193,6 +193,7 @@ func meteoraDammSwapAmountInfo(event string, params *struct {
 	Amount1  uint64
 	SwapMode uint8
 }) (swapMode SwapMode, fixedAmount decimal.Decimal, limitAmount decimal.Decimal, ok bool) {
+	_ = event
 	if params == nil {
 		return SwapModeUnknown, decimal.Zero, decimal.Zero, false
 	}
@@ -203,21 +204,14 @@ func meteoraDammSwapAmountInfo(event string, params *struct {
 	//   - ExactIn / PartialFill: amount0=amount_in,  amount1=minimum_amount_out
 	//   - ExactOut:              amount0=amount_out, amount1=maximum_amount_in
 	//
-	// The emitted event is normalized as token A <-> token B:
+	// `SetSwapAmountInfo` derives sides from the normalized buy/sell event, so
-	// - `sell` means A -> B, so A is the input side and B is the output side
+	// the instruction parameters should stay in raw IDL order here.
-	// - `buy`  means B -> A, so B is the input side and A is the output side
 	switch params.SwapMode {
 	case 0, 1: // ExactIn / PartialFill
 		swapMode = SwapModeExactIn
-		if event == TxEventSell {
 		return swapMode, decimal.NewFromUint64(params.Amount0), decimal.NewFromUint64(params.Amount1), true
-		}
-		return swapMode, decimal.NewFromUint64(params.Amount1), decimal.NewFromUint64(params.Amount0), true
 	case 2: // ExactOut
 		swapMode = SwapModeExactOut
-		if event == TxEventSell {
-			return swapMode, decimal.NewFromUint64(params.Amount1), decimal.NewFromUint64(params.Amount0), true
-		}
 		return swapMode, decimal.NewFromUint64(params.Amount0), decimal.NewFromUint64(params.Amount1), true
 	default:
 		return SwapModeUnknown, decimal.Zero, decimal.Zero, false

orcawhirpool.go

@@ -281,7 +281,7 @@ func orcaWhirPoolLiquidityParser(tx *Tx, instruction Instruction, innerInstructi
 		return nil, increaseOffset(offset), InstructionIgnoredError
 	}
 	if baseAmount.Equal(decimal.Zero) || quoteAmount.Equal(decimal.Zero) {
-		instructionName += "_on_side"
+		instructionName += "_one_side"
 	}
 	if (baseTokenBalance == nil && !baseAmount.Equal(decimal.Zero)) || (quoteTokenBalance == nil && !quoteAmount.Equal(decimal.Zero)) {
 		return nil, offset, fmt.Errorf("token balance is nil but amount is not zero")
@@ -388,7 +388,7 @@ func orcaWhirPoolLiquidityV2Parser(tx *Tx, instruction Instruction, innerInstruc
 		return nil, offset, InstructionIgnoredError
 	}
 	if baseAmount.Equal(decimal.Zero) || quoteAmount.Equal(decimal.Zero) {
-		instructionName += "_on_side"
+		instructionName += "_one_side"
 	}
 	if (baseTokenBalance == nil && !baseAmount.Equal(decimal.Zero)) || (quoteTokenBalance == nil && !quoteAmount.Equal(decimal.Zero)) {
 		return nil, offset, fmt.Errorf("token balance is nil but amount is not zero")
@@ -493,7 +493,7 @@ func orcaWhirPoolCollectFeeParser(tx *Tx, instruction Instruction, innerInstruct
 		return nil, offset, InstructionIgnoredError
 	}
 	if baseAmount.Equal(decimal.Zero) || quoteAmount.Equal(decimal.Zero) {
-		instructionName += "_on_side"
+		instructionName += "_one_side"
 	}
 	if (baseTokenBalance == nil && !baseAmount.Equal(decimal.Zero)) || (quoteTokenBalance == nil && !quoteAmount.Equal(decimal.Zero)) {
 		return nil, offset, fmt.Errorf("token balance is nil but amount is not zero")
@@ -595,7 +595,7 @@ func orcaWhirPoolCollectFeeV2Parser(tx *Tx, instruction Instruction, innerInstru
 		return nil, offset, InstructionIgnoredError
 	}
 	if baseAmount.Equal(decimal.Zero) || quoteAmount.Equal(decimal.Zero) {
-		instructionName += "_on_side"
+		instructionName += "_one_side"
 	}
 	if (baseTokenBalance == nil && !baseAmount.Equal(decimal.Zero)) || (quoteTokenBalance == nil && !quoteAmount.Equal(decimal.Zero)) {
 		return nil, offset, fmt.Errorf("token balance is nil but amount is not zero")
@@ -697,7 +697,7 @@ func orcaWhirPoolCollectProtocolFeeV2Parser(tx *Tx, instruction Instruction, inn
 		return nil, offset, InstructionIgnoredError
 	}
 	if baseAmount.Equal(decimal.Zero) || quoteAmount.Equal(decimal.Zero) {
-		instructionName += "_on_side"
+		instructionName += "_one_side"
 	}
 	if (baseTokenBalance == nil && !baseAmount.Equal(decimal.Zero)) || (quoteTokenBalance == nil && !quoteAmount.Equal(decimal.Zero)) {
 		return nil, offset, fmt.Errorf("token balance is nil but amount is not zero")

pump.go

@@ -231,6 +231,19 @@ func pumpTradeAmountInfoFromArgs(args PumpTradeArgs) (swapMode SwapMode, fixedAm
 	}
 }
 
+func pumpCompleteMatchesTradeEvent(completeEvent CompleteEvent, tradeEvent PumpTradeEvent, bondingCurve solana.PublicKey) bool {
+	if completeEvent.Mint != tradeEvent.Mint {
+		return false
+	}
+	if completeEvent.User != tradeEvent.User {
+		return false
+	}
+	if completeEvent.BondingCurve != bondingCurve {
+		return false
+	}
+	return true
+}
+
 func normalizePumpQuoteSideMint(s *Swap) {
 	if s.FixedAmountSide == SwapAmountSideQuote && s.FixedMint.IsZero() {
 		s.FixedMint = wSolMint
@@ -366,6 +379,7 @@ func BuyOrSellParser(tx *Tx, instruction Instruction, innerInstructions InnerIns
 		completeEvent CompleteEvent
 		completed     bool
 		newoffset     [2]uint
+		tradeFound    bool
 	)
 
 	var prefixLen = offset[1]
@@ -394,6 +408,9 @@ func BuyOrSellParser(tx *Tx, instruction Instruction, innerInstructions InnerIns
 		}
 		if innerInstr.ProgramIDIndex == programIndex && bytes.Equal(innerInstr.Data[:8], pumpEventDiscriminator[:]) {
 			if bytes.Equal(innerInstr.Data[8:16], pumpTradeEventDiscriminator[8:16]) {
+				if tradeFound {
+					break
+				}
 				err = agbinary.NewBorshDecoder(innerInstr.Data[16:]).Decode(&tradeEvent)
 				if offset[1] == 0 {
 					newoffset = [2]uint{offset[0] + 1, offset[1]}
@@ -403,19 +420,31 @@ func BuyOrSellParser(tx *Tx, instruction Instruction, innerInstructions InnerIns
 				if err != nil {
 					return nil, newoffset, fmt.Errorf("pump buy event decode error: %v, offset, %d, %d", err, offset[0], offset[1])
 				}
+				expectedIsBuy := !bytes.Equal(instruction.Data[:8], pumpSellDiscriminator[:])
+				if tradeEvent.IsBuy != expectedIsBuy {
+					tradeEvent = PumpTradeEvent{}
+					continue
+				}
+				tradeFound = true
 				if !tradeEvent.IsBuy {
 					break
 				}
 			} else if bytes.Equal(innerInstr.Data[8:16], pumpCompleteEventDiscriminator[:]) {
+				if !tradeFound {
+					continue
+				}
 				err = agbinary.NewBorshDecoder(innerInstr.Data[16:]).Decode(&completeEvent)
+				if err != nil {
+					return nil, increaseOffset(offset), fmt.Errorf("pump completeEvent event decode error: %v, offset, %d, %d", err, offset[0], offset[1])
+				}
+				if !pumpCompleteMatchesTradeEvent(completeEvent, tradeEvent, result.accountList[instruction.Accounts[3]]) {
+					break
+				}
 				if offset[1] == 0 {
 					newoffset = [2]uint{offset[0] + 1, offset[1]}
 				} else {
 					newoffset = [2]uint{offset[0], prefixLen + uint(innerIndex) + 1}
 				}
-				if err != nil {
-					return nil, newoffset, fmt.Errorf("pump completeEvent event decode error: %v, offset, %d, %d", err, offset[0], offset[1])
-				}
 				completed = true
 				break
 			}
@@ -428,6 +457,11 @@ func BuyOrSellParser(tx *Tx, instruction Instruction, innerInstructions InnerIns
 
 	offset = [2]uint{newoffset[0], newoffset[1]}
 
+	var args PumpTradeArgs
+	if err := agbinary.NewBorshDecoder(instruction.Data[:]).Decode(&args); err != nil {
+		return nil, increaseOffset(offset), fmt.Errorf("failed tx pump buy/sell decode error: %v, offset, %d, %d", err, offset[0], offset[1])
+	}
+
 	event := ""
 	baseTokenProgram := solana.TokenProgramID
 	if tradeEvent.IsBuy {
@@ -495,13 +529,10 @@ func BuyOrSellParser(tx *Tx, instruction Instruction, innerInstructions InnerIns
 			Cashback:          isCashbackCoin,
 		},
 	}
-	var args PumpTradeArgs
-	if err := agbinary.NewBorshDecoder(instruction.Data[:]).Decode(&args); err == nil {
 	if swapMode, fixedAmount, limitAmount, ok := pumpTradeAmountInfoFromArgs(args); ok {
 		swaps[0].SetSwapAmountInfo(swapMode, fixedAmount, limitAmount)
 		normalizePumpQuoteSideMint(&swaps[0])
 	}
-	}
 	if completed {
 		swaps = append(swaps, Swap{
 			Program:           SolProgramPump,

pump_test.go

@@ -76,3 +76,27 @@ func TestCal(t *testing.T) {
 
 	fmt.Println(solana.MustPublicKeyFromBase58("BM9CcyErJcu2mjrFvUsRRrD3snGeHDDVirJLvL6EjvMN").IsOnCurve())
 }
+
+func TestPumpCompleteMatchesTradeEvent(t *testing.T) {
+	mint := solana.MustPublicKeyFromBase58("8GNGkNnfBuoTP3QRnmdNzSYuuE15M8tvcNvxNsV4pump")
+	user := solana.MustPublicKeyFromBase58("DS95KxqUCCjwQaXhD7fhKatXbivwWDNrJdNV5ZcubGdz")
+	bondingCurve := solana.MustPublicKeyFromBase58("Gz5EX3X7kUDS48baijJKubQDKy3BBKpnMJQ3f3W1e9jA")
+
+	tradeEvent := PumpTradeEvent{
+		Mint: mint,
+		User: user,
+	}
+	completeEvent := CompleteEvent{
+		Mint:         mint,
+		User:         user,
+		BondingCurve: bondingCurve,
+	}
+	if !pumpCompleteMatchesTradeEvent(completeEvent, tradeEvent, bondingCurve) {
+		t.Fatal("pumpCompleteMatchesTradeEvent() = false, want true")
+	}
+
+	completeEvent.User = solana.MustPublicKeyFromBase58("3g89wLRwJ5P22fkCdPJBAP7iiYAo6yY96geQvMYj6tYm")
+	if pumpCompleteMatchesTradeEvent(completeEvent, tradeEvent, bondingCurve) {
+		t.Fatal("pumpCompleteMatchesTradeEvent() = true for mismatched user")
+	}
+}

swap_amounts.go

@@ -9,6 +9,16 @@ import (
 
 var maxSlippageBps = decimal.NewFromInt(10000)
 
+func normalizeSlippageBps(value decimal.Decimal) decimal.Decimal {
+	//if value.IsNegative() {
+	//	return decimal.Zero
+	//}
+	//if value.GreaterThan(maxSlippageBps) {
+	//	return maxSlippageBps
+	//}
+	return value
+}
+
 type SwapMode uint8
 type SwapAmountSide uint8
 type SwapLimitType uint8
@@ -141,29 +151,36 @@ func limitSwapAmountType(swapMode SwapMode) SwapLimitType {
 }
 
 func calculateLimitSlippageBps(limitType SwapLimitType, limitAmount, actualAmount decimal.Decimal) decimal.Decimal {
+	var value decimal.Decimal
 	switch limitType {
 	case SwapLimitTypeMinOut:
 		if !actualAmount.IsPositive() {
 			if !limitAmount.IsPositive() {
-				return maxSlippageBps
+				value = maxSlippageBps
+				break
 			}
-			return maxSlippageBps.Neg()
+			value = maxSlippageBps.Neg()
+			break
 		}
 		if !limitAmount.IsPositive() {
-			return maxSlippageBps
+			value = maxSlippageBps
+			break
 		}
-		return actualAmount.Sub(limitAmount).Mul(maxSlippageBps).Div(actualAmount)
+		value = actualAmount.Sub(limitAmount).Mul(maxSlippageBps).Div(actualAmount)
 	case SwapLimitTypeMaxIn:
 		if !limitAmount.IsPositive() {
 			if !actualAmount.IsPositive() {
-				return maxSlippageBps
+				value = maxSlippageBps
+				break
 			}
-			return maxSlippageBps.Neg()
+			value = maxSlippageBps.Neg()
+			break
 		}
-		return limitAmount.Sub(actualAmount).Mul(maxSlippageBps).Div(limitAmount)
+		value = limitAmount.Sub(actualAmount).Mul(maxSlippageBps).Div(limitAmount)
 	default:
-		return decimal.Zero
+		value = decimal.Zero
 	}
+	return normalizeSlippageBps(value)
 }
 
 func (s *Swap) SetSwapAmountInfoDetailed(

swap_amounts_test.go

@@ -79,6 +79,38 @@ func TestSetSwapAmountInfoExactInZeroLimitUsesMaxSlippage(t *testing.T) {
 	}
 }
 
+func TestSetSwapAmountInfoExactInNegativeHeadroomClampsToZero(t *testing.T) {
+	swap := Swap{
+		Event:       TxEventBuy,
+		BaseMint:    solana.MustPublicKeyFromBase58("11111111111111111111111111111111"),
+		QuoteMint:   solana.MustPublicKeyFromBase58("So11111111111111111111111111111111111111112"),
+		BaseAmount:  decimal.NewFromInt(90),
+		QuoteAmount: decimal.NewFromInt(100),
+	}
+
+	swap.SetSwapAmountInfo(SwapModeExactIn, decimal.NewFromInt(100), decimal.NewFromInt(110))
+
+	if got := swap.SlippageBps.String(); got != "0" {
+		t.Fatalf("slippage bps = %s, want 0", got)
+	}
+}
+
+func TestSetSwapAmountInfoExactOutNegativeHeadroomClampsToZero(t *testing.T) {
+	swap := Swap{
+		Event:       TxEventSell,
+		BaseMint:    solana.MustPublicKeyFromBase58("11111111111111111111111111111111"),
+		QuoteMint:   solana.MustPublicKeyFromBase58("So11111111111111111111111111111111111111112"),
+		BaseAmount:  decimal.NewFromInt(120),
+		QuoteAmount: decimal.NewFromInt(100),
+	}
+
+	swap.SetSwapAmountInfo(SwapModeExactOut, decimal.NewFromInt(100), decimal.NewFromInt(105))
+
+	if got := swap.SlippageBps.String(); got != "0" {
+		t.Fatalf("slippage bps = %s, want 0", got)
+	}
+}
+
 func TestMeteoraDammSwapAmountInfo(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -116,6 +148,18 @@ func TestMeteoraDammSwapAmountInfo(t *testing.T) {
 			wantFixed: 101,
 			wantLimit: 96,
 		},
+		{
+			name:  "buy exact in keeps amount0 as input and amount1 as min out",
+			event: TxEventBuy,
+			params: &struct {
+				Amount0  uint64
+				Amount1  uint64
+				SwapMode uint8
+			}{Amount0: 130, Amount1: 120, SwapMode: 0},
+			wantMode:  SwapModeExactIn,
+			wantFixed: 130,
+			wantLimit: 120,
+		},
 		{
 			name:  "buy exact out uses amount0 as target output and amount1 as max input",
 			event: TxEventBuy,
@@ -128,6 +172,18 @@ func TestMeteoraDammSwapAmountInfo(t *testing.T) {
 			wantFixed: 120,
 			wantLimit: 130,
 		},
+		{
+			name:  "sell exact out keeps amount0 as target output and amount1 as max input",
+			event: TxEventSell,
+			params: &struct {
+				Amount0  uint64
+				Amount1  uint64
+				SwapMode uint8
+			}{Amount0: 140, Amount1: 150, SwapMode: 2},
+			wantMode:  SwapModeExactOut,
+			wantFixed: 140,
+			wantLimit: 150,
+		},
 	}
 
 	for _, tt := range tests {

tx_binary.go

@@ -1,6 +1,7 @@
 package pump_parser
 
 import (
+	"bufio"
 	"bytes"
 	"encoding/binary"
 	"fmt"
@@ -11,6 +12,7 @@ import (
 	"strconv"
 
 	"github.com/gagliardetto/solana-go"
+	"github.com/mr-tron/base58"
 	"github.com/shopspring/decimal"
 )
 
@@ -79,8 +81,8 @@ type SwapBinary struct {
 	ActualLimitAmountSide SwapAmountSide
 	SlippageBps           uint64
 
-	BaseReserve  uint64
+	BaseReserve  float64
-	QuoteReserve uint64
+	QuoteReserve float64
 	Mayhem       bool
 	Cashback     bool
 
@@ -107,6 +109,18 @@ type TxsBinaryReaderSource interface {
 	OpenTxsBinaryReader() (io.ReadCloser, error)
 }
 
+type TxsBinaryBatchHeaderContext struct {
+	SourceIndex int
+	BatchIndex  int
+	Reader      *bufio.Reader
+}
+
+type TxsBinaryBatchHeaderFunc func(ctx *TxsBinaryBatchHeaderContext) (skip bool, err error)
+
+type TxsBinaryMergeOptions struct {
+	BatchHeaderFunc TxsBinaryBatchHeaderFunc
+}
+
 type PlatformBinary struct {
 	Platform    string
 	PlatformFee uint64
@@ -173,7 +187,7 @@ func NewTxsBinary(txs []Tx) (*TxsBinary, error) {
 	for i, tx := range txPtrs {
 		binaryTx, err := newTxBinaryWithAddressTable(tx, addressTable, addressIndex)
 		if err != nil {
-			return nil, fmt.Errorf("tx[%d]: %w", i, err)
+			return nil, fmt.Errorf("tx[%d], %s: %w", i, base58.Encode(tx.TxHash[:]), err)
 		}
 		out.Txs = append(out.Txs, *binaryTx)
 	}
@@ -307,24 +321,32 @@ func DecodeTxsBinaryReader(r io.Reader) iter.Seq2[*Tx, error] {
 }
 
 func MergeTxsBinaryBytes(encodedBatches [][]byte) ([]byte, error) {
+	return MergeTxsBinaryBytesWithOptions(encodedBatches, TxsBinaryMergeOptions{})
+}
+
+func MergeTxsBinaryBytesWithOptions(encodedBatches [][]byte, opts TxsBinaryMergeOptions) ([]byte, error) {
 	sources := make([]TxsBinaryReaderSource, 0, len(encodedBatches))
 	for _, encoded := range encodedBatches {
 		sources = append(sources, txBinaryBytesSource{data: encoded})
 	}
 
 	var out bytes.Buffer
-	if err := MergeTxsBinarySourcesToWriter(sources, &out); err != nil {
+	if err := MergeTxsBinarySourcesToWriterWithOptions(sources, &out, opts); err != nil {
 		return nil, err
 	}
 	return out.Bytes(), nil
 }
 
 func MergeTxsBinarySourcesToWriter(sources []TxsBinaryReaderSource, w io.Writer) error {
+	return MergeTxsBinarySourcesToWriterWithOptions(sources, w, TxsBinaryMergeOptions{})
+}
+
+func MergeTxsBinarySourcesToWriterWithOptions(sources []TxsBinaryReaderSource, w io.Writer, opts TxsBinaryMergeOptions) error {
 	if w == nil {
 		return fmt.Errorf("txs binary writer is nil")
 	}
 
-	plan, err := txBinaryBuildMergePlan(sources)
+	plan, err := txBinaryBuildMergePlan(sources, opts)
 	if err != nil {
 		return err
 	}
@@ -343,9 +365,22 @@ func MergeTxsBinarySourcesToWriter(sources []TxsBinaryReaderSource, w io.Writer)
 			return fmt.Errorf("source[%d]: open reader: %w", sourceIndex, err)
 		}
 
-		dec := txBinaryStreamDecoder{reader: reader}
+		bufferedReader := bufio.NewReader(reader)
+		dec := txBinaryStreamDecoder{reader: bufferedReader}
 		batchIndex := 0
 		for {
+			skipBatch, err := txBinaryApplyMergeBatchHeader(bufferedReader, opts, sourceIndex, batchIndex)
+			if err != nil {
+				closeErr := reader.Close()
+				if err == io.EOF {
+					if closeErr != nil {
+						return fmt.Errorf("source[%d]: close reader: %w", sourceIndex, closeErr)
+					}
+					break
+				}
+				return fmt.Errorf("source[%d].batch[%d]: %w", sourceIndex, batchIndex, err)
+			}
+
 			header, err := dec.readTxsBinaryHeaderOrEOF()
 			if err != nil {
 				closeErr := reader.Close()
@@ -368,6 +403,9 @@ func MergeTxsBinarySourcesToWriter(sources []TxsBinaryReaderSource, w io.Writer)
 					reader.Close()
 					return fmt.Errorf("source[%d].batch[%d].tx[%d]: %w", sourceIndex, batchIndex, txIndex, err)
 				}
+				if skipBatch {
+					continue
+				}
 				if err := txBinaryRemapTxAddressTable(&tx, header.addressTable, plan.addressTable, plan.addressIndex); err != nil {
 					reader.Close()
 					return fmt.Errorf("source[%d].batch[%d].tx[%d]: %w", sourceIndex, batchIndex, txIndex, err)
@@ -441,7 +479,7 @@ func (txs *TxsBinary) MarshalBinary() ([]byte, error) {
 	enc.writeUint32(uint32(len(txs.Txs)))
 	for i := range txs.Txs {
 		if err := enc.writeTxBinaryBody(&txs.Txs[i], enumTable); err != nil {
-			return nil, fmt.Errorf("tx[%d]: %w", i, err)
+			return nil, fmt.Errorf("tx[%d], %s: %w", i, base58.Encode(txs.Txs[i].TxHash[:]), err)
 		}
 	}
 	return enc.bytes(), nil
@@ -691,7 +729,7 @@ func newSwapBinary(swap Swap, index int, addressIndex *txBinaryAddressIndex) (Sw
 
 	out := SwapBinary{
 		Program:               swap.Program,
-		Event:                 swap.Event,
+		Event:                 txBinaryCanonicalEvent(swap.Event),
 		TxIndex:               int32(swap.TxIndex),
 		InstrIdx:              swap.InstrIdx,
 		InnerIdx:              swap.InnerIdx,
@@ -740,10 +778,10 @@ func newSwapBinary(swap Swap, index int, addressIndex *txBinaryAddressIndex) (Sw
 	if out.SlippageBps, err = txBinaryRoundedDecimalToUint64(swap.SlippageBps, fmt.Sprintf("swap[%d].slippage_bps", index)); err != nil {
 		return SwapBinary{}, err
 	}
-	if out.BaseReserve, err = txBinaryDecimalToUint64(swap.BaseReserve, fmt.Sprintf("swap[%d].base_reserve", index)); err != nil {
+	if out.BaseReserve, err = txBinaryDecimalToFloat64Raw(swap.BaseReserve, fmt.Sprintf("swap[%d].base_reserve", index)); err != nil {
 		return SwapBinary{}, err
 	}
-	if out.QuoteReserve, err = txBinaryDecimalToUint64(swap.QuoteReserve, fmt.Sprintf("swap[%d].quote_reserve", index)); err != nil {
+	if out.QuoteReserve, err = txBinaryDecimalToFloat64Raw(swap.QuoteReserve, fmt.Sprintf("swap[%d].quote_reserve", index)); err != nil {
 		return SwapBinary{}, err
 	}
 	if out.UserBaseBalance, err = txBinaryDecimalToUint64(swap.UserBaseBalance, fmt.Sprintf("swap[%d].user_base_balance", index)); err != nil {
@@ -841,8 +879,8 @@ func (swap SwapBinary) toSwap(addressTable []solana.PublicKey, index int) (Swap,
 		ActualLimitAmount:     decimal.NewFromUint64(swap.ActualLimitAmount),
 		ActualLimitAmountSide: swap.ActualLimitAmountSide,
 		SlippageBps:           decimal.NewFromUint64(swap.SlippageBps),
-		BaseReserve:           decimal.NewFromUint64(swap.BaseReserve),
+		BaseReserve:           txBinaryFloat64ToDecimalRaw(swap.BaseReserve),
-		QuoteReserve:          decimal.NewFromUint64(swap.QuoteReserve),
+		QuoteReserve:          txBinaryFloat64ToDecimalRaw(swap.QuoteReserve),
 		Mayhem:                swap.Mayhem,
 		Cashback:              swap.Cashback,
 		UserBaseBalance:       decimal.NewFromUint64(swap.UserBaseBalance),
@@ -881,6 +919,17 @@ func txBinaryPlatformsFromTx(platforms map[string]platformInfo) ([]PlatformBinar
 	return out, nil
 }
 
+func txBinaryCanonicalEvent(event string) string {
+	switch event {
+	case "add_liquidity_on_side":
+		return TxEventAddLiquidityOneSide
+	case "remove_liquidity_on_side":
+		return TxEventRemoveLiquidityOneSide
+	default:
+		return event
+	}
+}
+
 func txBinaryMevAgentsFromTx(mevAgents map[string]mevInfo) ([]MevAgentBinary, error) {
 	if len(mevAgents) == 0 {
 		return nil, nil
@@ -1026,6 +1075,14 @@ func txBinaryDecimalToFloat64(value decimal.Decimal, scale int32, field string)
 	return f, nil
 }
 
+func txBinaryDecimalToFloat64Raw(value decimal.Decimal, field string) (float64, error) {
+	f, exact := value.Float64()
+	if !exact && math.IsInf(f, 0) {
+		return 0, fmt.Errorf("%s cannot be represented as float64: %s", field, value.String())
+	}
+	return f, nil
+}
+
 func txBinaryFloat64ToDecimal(value float64, scale int32) decimal.Decimal {
 	formatted := strconv.FormatFloat(value, 'f', int(scale), 64)
 	out, err := decimal.NewFromString(formatted)
@@ -1035,6 +1092,15 @@ func txBinaryFloat64ToDecimal(value float64, scale int32) decimal.Decimal {
 	return out
 }
 
+func txBinaryFloat64ToDecimalRaw(value float64) decimal.Decimal {
+	formatted := strconv.FormatFloat(value, 'f', -1, 64)
+	out, err := decimal.NewFromString(formatted)
+	if err != nil {
+		return decimal.Zero
+	}
+	return out
+}
+
 type txBinaryEncoder struct {
 	buf bytes.Buffer
 }
@@ -1187,8 +1253,8 @@ func (enc *txBinaryEncoder) writeSwaps(swaps []SwapBinary, enumTable *txBinaryEn
 		enc.writeUint64(swap.ActualLimitAmount)
 		enc.writeUint8(uint8(swap.ActualLimitAmountSide))
 		enc.writeUint64(swap.SlippageBps)
-		enc.writeUint64(swap.BaseReserve)
+		enc.writeFloat64(swap.BaseReserve)
-		enc.writeUint64(swap.QuoteReserve)
+		enc.writeFloat64(swap.QuoteReserve)
 		enc.writeBool(swap.Mayhem)
 		enc.writeBool(swap.Cashback)
 		enc.writeUint64(swap.UserBaseBalance)
@@ -1683,10 +1749,10 @@ func txBinaryReadSwaps(dec txBinaryBodyReader, enumTable *txBinaryEnumTable) ([]
 		if swap.SlippageBps, err = dec.readUint64(); err != nil {
 			return nil, err
 		}
-		if swap.BaseReserve, err = dec.readUint64(); err != nil {
+		if swap.BaseReserve, err = dec.readFloat64(); err != nil {
 			return nil, err
 		}
-		if swap.QuoteReserve, err = dec.readUint64(); err != nil {
+		if swap.QuoteReserve, err = dec.readFloat64(); err != nil {
 			return nil, err
 		}
 		if swap.Mayhem, err = dec.readBool(); err != nil {
@@ -1780,7 +1846,7 @@ func txBinaryReadTxBody(dec txBinaryBodyReader, tx *TxBinary, enumTable *txBinar
 	return nil
 }
 
-func txBinaryBuildMergePlan(sources []TxsBinaryReaderSource) (*txsBinaryMergePlan, error) {
+func txBinaryBuildMergePlan(sources []TxsBinaryReaderSource, opts TxsBinaryMergeOptions) (*txsBinaryMergePlan, error) {
 	if len(sources) == 0 {
 		return nil, fmt.Errorf("txs binary sources are empty")
 	}
@@ -1801,9 +1867,22 @@ func txBinaryBuildMergePlan(sources []TxsBinaryReaderSource) (*txsBinaryMergePla
 			return nil, fmt.Errorf("source[%d]: open reader: %w", sourceIndex, err)
 		}
 
-		dec := txBinaryStreamDecoder{reader: reader}
+		bufferedReader := bufio.NewReader(reader)
+		dec := txBinaryStreamDecoder{reader: bufferedReader}
 		batchIndex := 0
 		for {
+			skipBatch, err := txBinaryApplyMergeBatchHeader(bufferedReader, opts, sourceIndex, batchIndex)
+			if err != nil {
+				closeErr := reader.Close()
+				if err == io.EOF {
+					if closeErr != nil {
+						return nil, fmt.Errorf("source[%d]: close reader: %w", sourceIndex, closeErr)
+					}
+					break
+				}
+				return nil, fmt.Errorf("source[%d].batch[%d]: %w", sourceIndex, batchIndex, err)
+			}
+
 			header, err := dec.readTxsBinaryHeaderOrEOF()
 			if err != nil {
 				closeErr := reader.Close()
@@ -1833,17 +1912,21 @@ func txBinaryBuildMergePlan(sources []TxsBinaryReaderSource) (*txsBinaryMergePla
 			}
 
 			for addressIndex, address := range header.addressTable {
+				if !skipBatch {
 					if err := builder.add(address); err != nil {
 						reader.Close()
 						return nil, fmt.Errorf("source[%d].batch[%d].address[%d]: %w", sourceIndex, batchIndex, addressIndex, err)
 					}
 				}
+			}
 
+			if !skipBatch {
 				if uint64(plan.txCount)+uint64(header.count) > uint64(math.MaxUint32) {
 					reader.Close()
 					return nil, fmt.Errorf("merged tx count exceeds uint32 capacity")
 				}
 				plan.txCount += header.count
+			}
 
 			for txIndex := uint32(0); txIndex < header.count; txIndex++ {
 				tx := TxBinary{
@@ -1947,6 +2030,17 @@ func txBinaryWriteAll(w io.Writer, data []byte) error {
 	return nil
 }
 
+func txBinaryApplyMergeBatchHeader(reader *bufio.Reader, opts TxsBinaryMergeOptions, sourceIndex int, batchIndex int) (bool, error) {
+	if opts.BatchHeaderFunc == nil {
+		return false, nil
+	}
+	return opts.BatchHeaderFunc(&TxsBinaryBatchHeaderContext{
+		SourceIndex: sourceIndex,
+		BatchIndex:  batchIndex,
+		Reader:      reader,
+	})
+}
+
 type txBinaryEnumTable struct {
 	version   uint16
 	programs  txBinaryEnumSet
@@ -1990,6 +2084,18 @@ var txBinaryEnumTables = map[uint16]*txBinaryEnumTable{
 			TxEventBuyFailed,
 			TxEventSellFailed,
 			TxEventBurn,
+			TxEventCreate,
+			TxEventComplete,
+			TxEventMigrate,
+			TxEventDeposit,
+			TxEventWithdraw,
+			TxEventOpen,
+			TxEventClose,
+			TxEventClaimFee,
+			TxEventAddLiquidity,
+			TxEventAddLiquidityOneSide,
+			TxEventRemoveLiquidity,
+			TxEventRemoveLiquidityOneSide,
 		},
 		"platform",
 		[]string{

tx_binary_test.go

@@ -225,6 +225,191 @@ func TestTxBinaryRejectsUnknownProgramEnum(t *testing.T) {
 	}
 }
 
+func TestTxBinaryAcceptsKnownEventEnums(t *testing.T) {
+	events := []string{
+		TxEventAddLP,
+		TxEventRemoveLP,
+		TxEventBuy,
+		TxEventSell,
+		TxEventBuyFailed,
+		TxEventSellFailed,
+		TxEventBurn,
+		TxEventCreate,
+		TxEventComplete,
+		TxEventMigrate,
+		TxEventDeposit,
+		TxEventWithdraw,
+		TxEventOpen,
+		TxEventClose,
+		TxEventClaimFee,
+		TxEventAddLiquidity,
+		TxEventAddLiquidityOneSide,
+		TxEventRemoveLiquidity,
+		TxEventRemoveLiquidityOneSide,
+	}
+
+	for _, event := range events {
+		t.Run(event, func(t *testing.T) {
+			txBinary := &TxBinary{
+				SchemaVersion: txBinarySchemaVersionCurrent,
+				EnumVersion:   txBinaryEnumVersionV1,
+				AddressTable: []solana.PublicKey{
+					mustPubKey("11111111111111111111111111111111"),
+					mustPubKey("So11111111111111111111111111111111111111112"),
+					solana.TokenProgramID,
+					mustPubKey("BPFLoader1111111111111111111111111111111111"),
+					mustPubKey("SysvarRent111111111111111111111111111111111"),
+				},
+				Swaps: []SwapBinary{
+					{
+						Program:           SolProgramPump,
+						Event:             event,
+						Pool:              0,
+						BaseMint:          1,
+						QuoteMint:         1,
+						BaseTokenProgram:  2,
+						QuoteTokenProgram: 2,
+						Creator:           3,
+						User:              4,
+						FixedMint:         1,
+						LimitMint:         1,
+					},
+				},
+			}
+
+			encoded, err := txBinary.MarshalBinary()
+			if err != nil {
+				t.Fatalf("MarshalBinary() error = %v", err)
+			}
+
+			var decoded TxBinary
+			if err := decoded.UnmarshalBinary(encoded); err != nil {
+				t.Fatalf("UnmarshalBinary() error = %v", err)
+			}
+			if got := decoded.Swaps[0].Event; got != event {
+				t.Fatalf("decoded event = %q, want %q", got, event)
+			}
+		})
+	}
+}
+
+func TestTxBinaryPreservesFractionalReserves(t *testing.T) {
+	tx := &Tx{
+		Signer:               mustPubKey("So11111111111111111111111111111111111111112"),
+		Block:                1,
+		BlockIndex:           1,
+		CuFee:                decimal.NewFromInt(1),
+		CUPrice:              decimal.RequireFromString("0.000001"),
+		BeforeSolBalance:     decimal.RequireFromString("1.000000000"),
+		AfterSOLBalance:      decimal.RequireFromString("0.900000000"),
+		ComputeUnitsConsumed: 1,
+		CuLimit:              1,
+		Swaps: []Swap{
+			{
+				Program:               SolProgramMeteoraPools,
+				Event:                 TxEventAddLiquidity,
+				Pool:                  mustPubKey("11111111111111111111111111111111"),
+				BaseMint:              mustPubKey("3wyAj7RtG72wM1Wv9DkYfL7RAx9X3Jx1sC6E6mN4jWeL"),
+				QuoteMint:             solana.WrappedSol,
+				BaseTokenProgram:      solana.TokenProgramID,
+				QuoteTokenProgram:     solana.TokenProgramID,
+				Creator:               mustPubKey("BPFLoader1111111111111111111111111111111111"),
+				BaseMintDecimals:      6,
+				QuoteMintDecimals:     9,
+				User:                  mustPubKey("SysvarRent111111111111111111111111111111111"),
+				BaseAmount:            decimal.NewFromInt(10),
+				QuoteAmount:           decimal.NewFromInt(20),
+				SwapMode:              SwapModeExactIn,
+				FixedAmount:           decimal.NewFromInt(20),
+				FixedAmountSide:       SwapAmountSideQuote,
+				FixedMint:             solana.WrappedSol,
+				LimitAmountType:       SwapLimitTypeMinOut,
+				LimitAmount:           decimal.NewFromInt(9),
+				LimitAmountSide:       SwapAmountSideBase,
+				LimitMint:             mustPubKey("3wyAj7RtG72wM1Wv9DkYfL7RAx9X3Jx1sC6E6mN4jWeL"),
+				ActualLimitAmount:     decimal.NewFromInt(10),
+				ActualLimitAmountSide: SwapAmountSideBase,
+				BaseReserve:           decimal.RequireFromString("123.4"),
+				QuoteReserve:          decimal.RequireFromString("710079483.625409498"),
+				AfterSOLBalance:       decimal.RequireFromString("0.800000000"),
+			},
+		},
+	}
+
+	encoded, err := EncodeTxBinary(tx)
+	if err != nil {
+		t.Fatalf("EncodeTxBinary() error = %v", err)
+	}
+
+	decoded, err := DecodeTxBinary(encoded)
+	if err != nil {
+		t.Fatalf("DecodeTxBinary() error = %v", err)
+	}
+	if got := decoded.Swaps[0].BaseReserve.String(); got != "123.4" {
+		t.Fatalf("BaseReserve = %s, want 123.4", got)
+	}
+	diff := decoded.Swaps[0].QuoteReserve.Sub(decimal.RequireFromString("710079483.625409498")).Abs()
+	if diff.GreaterThan(decimal.RequireFromString("0.0000001")) {
+		t.Fatalf("QuoteReserve diff = %s, want <= 0.0000001", diff)
+	}
+}
+
+func TestTxBinaryCanonicalizesOnSideEventAlias(t *testing.T) {
+	tx := &Tx{
+		Signer:               mustPubKey("So11111111111111111111111111111111111111112"),
+		Block:                1,
+		BlockIndex:           1,
+		CuFee:                decimal.NewFromInt(1),
+		CUPrice:              decimal.RequireFromString("0.000001"),
+		BeforeSolBalance:     decimal.RequireFromString("1.000000000"),
+		AfterSOLBalance:      decimal.RequireFromString("0.900000000"),
+		ComputeUnitsConsumed: 1,
+		CuLimit:              1,
+		Swaps: []Swap{
+			{
+				Program:               SolProgramOrcaWhirPool,
+				Event:                 "remove_liquidity_on_side",
+				Pool:                  mustPubKey("11111111111111111111111111111111"),
+				BaseMint:              mustPubKey("3wyAj7RtG72wM1Wv9DkYfL7RAx9X3Jx1sC6E6mN4jWeL"),
+				QuoteMint:             solana.WrappedSol,
+				BaseTokenProgram:      solana.TokenProgramID,
+				QuoteTokenProgram:     solana.TokenProgramID,
+				Creator:               mustPubKey("BPFLoader1111111111111111111111111111111111"),
+				BaseMintDecimals:      6,
+				QuoteMintDecimals:     9,
+				User:                  mustPubKey("SysvarRent111111111111111111111111111111111"),
+				BaseAmount:            decimal.NewFromInt(10),
+				QuoteAmount:           decimal.Zero,
+				SwapMode:              SwapModeExactIn,
+				FixedAmount:           decimal.NewFromInt(10),
+				FixedAmountSide:       SwapAmountSideBase,
+				FixedMint:             mustPubKey("3wyAj7RtG72wM1Wv9DkYfL7RAx9X3Jx1sC6E6mN4jWeL"),
+				LimitAmountType:       SwapLimitTypeMinOut,
+				LimitAmount:           decimal.Zero,
+				LimitAmountSide:       SwapAmountSideQuote,
+				ActualLimitAmount:     decimal.Zero,
+				ActualLimitAmountSide: SwapAmountSideQuote,
+				BaseReserve:           decimal.RequireFromString("123.4"),
+				QuoteReserve:          decimal.RequireFromString("456.7"),
+				AfterSOLBalance:       decimal.RequireFromString("0.800000000"),
+			},
+		},
+	}
+
+	encoded, err := EncodeTxBinary(tx)
+	if err != nil {
+		t.Fatalf("EncodeTxBinary() error = %v", err)
+	}
+
+	decoded, err := DecodeTxBinary(encoded)
+	if err != nil {
+		t.Fatalf("DecodeTxBinary() error = %v", err)
+	}
+	if got := decoded.Swaps[0].Event; got != TxEventRemoveLiquidityOneSide {
+		t.Fatalf("Event = %q, want %q", got, TxEventRemoveLiquidityOneSide)
+	}
+}
+
 func TestTxsBinaryRoundTripWithSharedAddressTable(t *testing.T) {
 	tx1 := Tx{
 		Signer:               mustPubKey("So11111111111111111111111111111111111111112"),
@@ -602,6 +787,89 @@ func TestMergeTxsBinarySourcesToWriterWithConcatenatedBatches(t *testing.T) {
 	}
 }
 
+func TestMergeTxsBinarySourcesToWriterWithBatchHeaderFuncSkip(t *testing.T) {
+	tx1 := Tx{
+		Signer:               mustPubKey("So11111111111111111111111111111111111111112"),
+		Block:                31,
+		BlockIndex:           1,
+		CuFee:                decimal.NewFromInt(1),
+		CUPrice:              decimal.RequireFromString("0.000001"),
+		BeforeSolBalance:     decimal.RequireFromString("1.000000000"),
+		AfterSOLBalance:      decimal.RequireFromString("0.900000000"),
+		ComputeUnitsConsumed: 11,
+		CuLimit:              111,
+	}
+	tx2 := tx1
+	tx2.Block = 32
+	tx2.BlockIndex = 2
+	tx2.Signer = mustPubKey("SysvarRent111111111111111111111111111111111")
+	tx3 := tx1
+	tx3.Block = 33
+	tx3.BlockIndex = 3
+	tx3.Signer = mustPubKey("ComputeBudget111111111111111111111111111111")
+
+	batch1, err := EncodeTxsBinary([]Tx{tx1})
+	if err != nil {
+		t.Fatalf("EncodeTxsBinary(batch1) error = %v", err)
+	}
+	batch2, err := EncodeTxsBinary([]Tx{tx2})
+	if err != nil {
+		t.Fatalf("EncodeTxsBinary(batch2) error = %v", err)
+	}
+	batch3, err := EncodeTxsBinary([]Tx{tx3})
+	if err != nil {
+		t.Fatalf("EncodeTxsBinary(batch3) error = %v", err)
+	}
+
+	source := &testTxsBinarySource{
+		data: append(
+			append(
+				append([]byte{}, testBatchHeader(false)...),
+				batch1...,
+			),
+			append(
+				append(testBatchHeader(true), batch2...),
+				append(testBatchHeader(false), batch3...)...,
+			)...,
+		),
+	}
+
+	var out bytes.Buffer
+	err = MergeTxsBinarySourcesToWriterWithOptions(
+		[]TxsBinaryReaderSource{source},
+		&out,
+		TxsBinaryMergeOptions{
+			BatchHeaderFunc: func(ctx *TxsBinaryBatchHeaderContext) (bool, error) {
+				header := make([]byte, 5)
+				if _, err := io.ReadFull(ctx.Reader, header); err != nil {
+					return false, err
+				}
+				if !bytes.Equal(header[:4], []byte("BHDR")) {
+					return false, io.ErrUnexpectedEOF
+				}
+				return header[4] == 1, nil
+			},
+		},
+	)
+	if err != nil {
+		t.Fatalf("MergeTxsBinarySourcesToWriterWithOptions() error = %v", err)
+	}
+
+	decoded, err := DecodeTxsBinary(out.Bytes())
+	if err != nil {
+		t.Fatalf("DecodeTxsBinary(merged) error = %v", err)
+	}
+	if len(decoded) != 2 {
+		t.Fatalf("decoded len = %d, want 2", len(decoded))
+	}
+	if decoded[0].Block != tx1.Block || decoded[1].Block != tx3.Block {
+		t.Fatalf("decoded block order mismatch after skip")
+	}
+	if source.opens != 2 {
+		t.Fatalf("source.opens = %d, want 2", source.opens)
+	}
+}
+
 func mustPubKey(value string) solana.PublicKey {
 	return solana.MustPublicKeyFromBase58(value)
 }
@@ -625,3 +893,11 @@ func (s *testTxsBinarySource) OpenTxsBinaryReader() (io.ReadCloser, error) {
 	s.opens++
 	return io.NopCloser(bytes.NewReader(s.data)), nil
 }
+
+func testBatchHeader(skip bool) []byte {
+	header := []byte("BHDR\x00")
+	if skip {
+		header[4] = 1
+	}
+	return header
+}