package scanner import ( "context" "fmt" "math/big" "os" "strings" "sync" "time" "github.com/ethereum/go-ethereum/common" etypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" "github.com/fraktal/mev-beta/internal/config" "github.com/fraktal/mev-beta/internal/logger" "github.com/fraktal/mev-beta/internal/tokens" "github.com/fraktal/mev-beta/pkg/circuit" "github.com/fraktal/mev-beta/pkg/contracts" "github.com/fraktal/mev-beta/pkg/database" "github.com/fraktal/mev-beta/pkg/events" "github.com/fraktal/mev-beta/pkg/marketdata" "github.com/fraktal/mev-beta/pkg/pools" "github.com/fraktal/mev-beta/pkg/profitcalc" "github.com/fraktal/mev-beta/pkg/trading" stypes "github.com/fraktal/mev-beta/pkg/types" "github.com/fraktal/mev-beta/pkg/uniswap" "github.com/holiman/uint256" "golang.org/x/sync/singleflight" ) // MarketScanner scans markets for price movement opportunities with concurrency type MarketScanner struct { config *config.BotConfig logger *logger.Logger workerPool chan chan events.Event workers []*EventWorker wg sync.WaitGroup cacheGroup singleflight.Group cache map[string]*CachedData cacheMutex sync.RWMutex cacheTTL time.Duration slippageProtector *trading.SlippageProtection circuitBreaker *circuit.CircuitBreaker contractExecutor *contracts.ContractExecutor create2Calculator *pools.CREATE2Calculator database *database.Database profitCalculator *profitcalc.ProfitCalculator opportunityRanker *profitcalc.OpportunityRanker marketDataLogger *marketdata.MarketDataLogger // Enhanced market data logging system } // EventWorker represents a worker that processes event details type EventWorker struct { ID int WorkerPool chan chan events.Event JobChannel chan events.Event QuitChan chan bool scanner *MarketScanner } // NewMarketScanner creates a new market scanner with concurrency support func NewMarketScanner(cfg *config.BotConfig, logger *logger.Logger, contractExecutor *contracts.ContractExecutor, db *database.Database) *MarketScanner { scanner := &MarketScanner{ config: cfg, logger: logger, workerPool: make(chan chan events.Event, cfg.MaxWorkers), workers: make([]*EventWorker, 0, cfg.MaxWorkers), cache: make(map[string]*CachedData), cacheTTL: time.Duration(cfg.RPCTimeout) * time.Second, slippageProtector: trading.NewSlippageProtection(contractExecutor.GetClient(), logger), circuitBreaker: circuit.NewCircuitBreaker(&circuit.Config{ Logger: logger, Name: "market_scanner", MaxFailures: 10, ResetTimeout: time.Minute * 5, MaxRequests: 3, SuccessThreshold: 2, }), contractExecutor: contractExecutor, create2Calculator: pools.NewCREATE2Calculator(logger, contractExecutor.GetClient()), database: db, profitCalculator: profitcalc.NewProfitCalculatorWithClient(logger, contractExecutor.GetClient()), opportunityRanker: profitcalc.NewOpportunityRanker(logger), marketDataLogger: marketdata.NewMarketDataLogger(logger, db), // Initialize market data logger } // Initialize market data logger ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if err := scanner.marketDataLogger.Initialize(ctx); err != nil { logger.Warn(fmt.Sprintf("Failed to initialize market data logger: %v", err)) } // Create workers for i := 0; i < cfg.MaxWorkers; i++ { worker := NewEventWorker(i, scanner.workerPool, scanner) scanner.workers = append(scanner.workers, worker) worker.Start() } // Start cache cleanup routine go scanner.cleanupCache() return scanner } // NewEventWorker creates a new event worker func NewEventWorker(id int, workerPool chan chan events.Event, scanner *MarketScanner) *EventWorker { return &EventWorker{ ID: id, WorkerPool: workerPool, JobChannel: make(chan events.Event), QuitChan: make(chan bool), scanner: scanner, } } // Start begins the worker func (w *EventWorker) Start() { go func() { for { // Register the worker in the worker pool w.WorkerPool <- w.JobChannel select { case job := <-w.JobChannel: // Process the job w.Process(job) case <-w.QuitChan: // Stop the worker return } } }() } // Stop terminates the worker func (w *EventWorker) Stop() { go func() { w.QuitChan <- true }() } // Process handles an event detail func (w *EventWorker) Process(event events.Event) { // Analyze the event in a separate goroutine to maintain throughput go func() { defer w.scanner.wg.Done() // Log the processing w.scanner.logger.Debug(fmt.Sprintf("Worker %d processing %s event in pool %s from protocol %s", w.ID, event.Type.String(), event.PoolAddress, event.Protocol)) // Analyze based on event type switch event.Type { case events.Swap: w.scanner.analyzeSwapEvent(event) case events.AddLiquidity: w.scanner.analyzeLiquidityEvent(event, true) case events.RemoveLiquidity: w.scanner.analyzeLiquidityEvent(event, false) case events.NewPool: w.scanner.analyzeNewPoolEvent(event) default: w.scanner.logger.Debug(fmt.Sprintf("Worker %d received unknown event type: %d", w.ID, event.Type)) } }() } // SubmitEvent submits an event for processing by the worker pool func (s *MarketScanner) SubmitEvent(event events.Event) { s.wg.Add(1) // Get an available worker job channel jobChannel := <-s.workerPool // Send the job to the worker jobChannel <- event } // analyzeSwapEvent analyzes a swap event for arbitrage opportunities func (s *MarketScanner) analyzeSwapEvent(event events.Event) { s.logger.Debug(fmt.Sprintf("Analyzing swap event in pool %s", event.PoolAddress)) // Get comprehensive pool data to determine factory and fee poolInfo, poolExists := s.marketDataLogger.GetPoolInfo(event.PoolAddress) factory := common.Address{} fee := uint32(3000) // Default 0.3% if poolExists { factory = poolInfo.Factory fee = poolInfo.Fee } else { // Determine factory from known DEX protocols factory = s.getFactoryForProtocol(event.Protocol) } // Create comprehensive swap event data for market data logger swapData := &marketdata.SwapEventData{ TxHash: event.TransactionHash, BlockNumber: event.BlockNumber, LogIndex: uint(0), // Default log index (would need to be extracted from receipt) Timestamp: time.Now(), PoolAddress: event.PoolAddress, Factory: factory, Protocol: event.Protocol, Token0: event.Token0, Token1: event.Token1, Sender: common.Address{}, // Default sender (would need to be extracted from transaction) Recipient: common.Address{}, // Default recipient (would need to be extracted from transaction) SqrtPriceX96: event.SqrtPriceX96, Liquidity: event.Liquidity, Tick: int32(event.Tick), } // Extract swap amounts from event (handle signed amounts correctly) if event.Amount0 != nil && event.Amount1 != nil { amount0Float := new(big.Float).SetInt(event.Amount0) amount1Float := new(big.Float).SetInt(event.Amount1) // Determine input/output based on sign (negative means token was removed from pool = output) if amount0Float.Sign() < 0 { // Token0 out, Token1 in swapData.Amount0Out = new(big.Int).Abs(event.Amount0) swapData.Amount1In = event.Amount1 swapData.Amount0In = big.NewInt(0) swapData.Amount1Out = big.NewInt(0) } else if amount1Float.Sign() < 0 { // Token0 in, Token1 out swapData.Amount0In = event.Amount0 swapData.Amount1Out = new(big.Int).Abs(event.Amount1) swapData.Amount0Out = big.NewInt(0) swapData.Amount1In = big.NewInt(0) } else { // Both positive (shouldn't happen in normal swaps, but handle gracefully) swapData.Amount0In = event.Amount0 swapData.Amount1In = event.Amount1 swapData.Amount0Out = big.NewInt(0) swapData.Amount1Out = big.NewInt(0) } } // Calculate USD values using profit calculator's price oracle swapData.AmountInUSD, swapData.AmountOutUSD, swapData.FeeUSD = s.calculateSwapUSDValues(swapData, fee) // Calculate price impact based on pool liquidity and swap amounts swapData.PriceImpact = s.calculateSwapPriceImpact(event, swapData) // Log comprehensive swap event to market data logger ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := s.marketDataLogger.LogSwapEvent(ctx, event, swapData); err != nil { s.logger.Debug(fmt.Sprintf("Failed to log swap event to market data logger: %v", err)) } // Log the swap event to database (legacy) s.logSwapEvent(event) // Get pool data with caching poolData, err := s.getPoolData(event.PoolAddress.Hex()) if err != nil { s.logger.Error(fmt.Sprintf("Error getting pool data for %s: %v", event.PoolAddress, err)) return } // Calculate price impact priceMovement, err := s.calculatePriceMovement(event, poolData) if err != nil { s.logger.Error(fmt.Sprintf("Error calculating price movement for pool %s: %v", event.PoolAddress, err)) return } // Log opportunity with actual swap amounts from event (legacy) s.logSwapOpportunity(event, poolData, priceMovement) // Check if the movement is significant if s.isSignificantMovement(priceMovement, s.config.MinProfitThreshold) { s.logger.Info(fmt.Sprintf("Significant price movement detected in pool %s: %+v", event.PoolAddress, priceMovement)) // Look for arbitrage opportunities opportunities := s.findArbitrageOpportunities(event, priceMovement) if len(opportunities) > 0 { s.logger.Info(fmt.Sprintf("Found %d arbitrage opportunities for pool %s", len(opportunities), event.PoolAddress)) for _, opp := range opportunities { s.logger.Info(fmt.Sprintf("Arbitrage opportunity: %+v", opp)) // Execute the arbitrage opportunity s.executeArbitrageOpportunity(opp) } } } else { s.logger.Debug(fmt.Sprintf("Price movement in pool %s is not significant: %f", event.PoolAddress, priceMovement.PriceImpact)) } } // logSwapOpportunity logs swap opportunities using actual amounts from events func (s *MarketScanner) logSwapOpportunity(event events.Event, poolData interface{}, priceMovement *PriceMovement) { // Convert amounts from big.Int to big.Float for profit calculation amountInFloat := big.NewFloat(0) amountOutFloat := big.NewFloat(0) amountInDisplay := float64(0) amountOutDisplay := float64(0) // For swap events, Amount0 and Amount1 represent the actual swap amounts // The sign indicates direction (positive = token added to pool, negative = token removed from pool) if event.Amount0 != nil { amount0Float := new(big.Float).SetInt(event.Amount0) if event.Amount1 != nil { amount1Float := new(big.Float).SetInt(event.Amount1) // Determine input/output based on sign (negative means token was removed from pool = output) if amount0Float.Sign() < 0 { // Token0 out, Token1 in amountOutFloat = new(big.Float).Abs(amount0Float) amountInFloat = amount1Float amountOutDisplay, _ = new(big.Float).Quo(amountOutFloat, big.NewFloat(1e18)).Float64() amountInDisplay, _ = new(big.Float).Quo(amountInFloat, big.NewFloat(1e18)).Float64() } else { // Token0 in, Token1 out amountInFloat = amount0Float amountOutFloat = new(big.Float).Abs(amount1Float) amountInDisplay, _ = new(big.Float).Quo(amountInFloat, big.NewFloat(1e18)).Float64() amountOutDisplay, _ = new(big.Float).Quo(amountOutFloat, big.NewFloat(1e18)).Float64() } } } // Analyze arbitrage opportunity using the profit calculator var estimatedProfitUSD float64 = 0.0 var profitData map[string]interface{} if amountInFloat.Sign() > 0 && amountOutFloat.Sign() > 0 { opportunity := s.profitCalculator.AnalyzeSwapOpportunity( context.Background(), event.Token0, event.Token1, new(big.Float).Quo(amountInFloat, big.NewFloat(1e18)), // Convert to ETH units new(big.Float).Quo(amountOutFloat, big.NewFloat(1e18)), // Convert to ETH units event.Protocol, ) if opportunity != nil { // Add opportunity to ranking system rankedOpp := s.opportunityRanker.AddOpportunity(opportunity) // Use the calculated profit for logging if opportunity.NetProfit != nil { estimatedProfitFloat, _ := opportunity.NetProfit.Float64() estimatedProfitUSD = estimatedProfitFloat * 2000 // Assume 1 ETH = $2000 for USD conversion } // Add detailed profit analysis to additional data profitData = map[string]interface{}{ "arbitrageId": opportunity.ID, "isExecutable": opportunity.IsExecutable, "rejectReason": opportunity.RejectReason, "confidence": opportunity.Confidence, "profitMargin": opportunity.ProfitMargin, "netProfitETH": s.profitCalculator.FormatEther(opportunity.NetProfit), "gasCostETH": s.profitCalculator.FormatEther(opportunity.GasCost), "estimatedProfitETH": s.profitCalculator.FormatEther(opportunity.EstimatedProfit), } // Add ranking data if available if rankedOpp != nil { profitData["opportunityScore"] = rankedOpp.Score profitData["opportunityRank"] = rankedOpp.Rank profitData["competitionRisk"] = rankedOpp.CompetitionRisk profitData["updateCount"] = rankedOpp.UpdateCount } } } else if priceMovement != nil { // Fallback to simple price impact calculation estimatedProfitUSD = priceMovement.PriceImpact * 100 } // Resolve token symbols tokenIn := s.resolveTokenSymbol(event.Token0.Hex()) tokenOut := s.resolveTokenSymbol(event.Token1.Hex()) // Create additional data with profit analysis additionalData := map[string]interface{}{ "poolAddress": event.PoolAddress.Hex(), "protocol": event.Protocol, "token0": event.Token0.Hex(), "token1": event.Token1.Hex(), "tokenIn": tokenIn, "tokenOut": tokenOut, "blockNumber": event.BlockNumber, } // Add price impact if available if priceMovement != nil { additionalData["priceImpact"] = priceMovement.PriceImpact } // Merge profit analysis data if profitData != nil { for k, v := range profitData { additionalData[k] = v } } // Log the opportunity using actual swap amounts and profit analysis s.logger.Opportunity(event.TransactionHash.Hex(), "", event.PoolAddress.Hex(), "Swap", event.Protocol, amountInDisplay, amountOutDisplay, 0.0, estimatedProfitUSD, additionalData) } // resolveTokenSymbol converts token address to human-readable symbol func (s *MarketScanner) resolveTokenSymbol(tokenAddress string) string { // Convert to lowercase for consistent lookup addr := strings.ToLower(tokenAddress) // Known Arbitrum token mappings (same as in L2 parser) tokenMap := map[string]string{ "0x82af49447d8a07e3bd95bd0d56f35241523fbab1": "WETH", "0xaf88d065e77c8cc2239327c5edb3a432268e5831": "USDC", "0xff970a61a04b1ca14834a43f5de4533ebddb5cc8": "USDC.e", "0xfd086bc7cd5c481dcc9c85ebe478a1c0b69fcbb9": "USDT", "0x2f2a2543b76a4166549f7aab2e75bef0aefc5b0f": "WBTC", "0x912ce59144191c1204e64559fe8253a0e49e6548": "ARB", "0xfc5a1a6eb076a2c7ad06ed22c90d7e710e35ad0a": "GMX", "0xf97f4df75117a78c1a5a0dbb814af92458539fb4": "LINK", "0xfa7f8980b0f1e64a2062791cc3b0871572f1f7f0": "UNI", "0xba5ddd1f9d7f570dc94a51479a000e3bce967196": "AAVE", "0x0de59c86c306b9fead9fb67e65551e2b6897c3f6": "KUMA", "0x6efa9b8883dfb78fd75cd89d8474c44c3cbda469": "DIA", "0x440017a1b021006d556d7fc06a54c32e42eb745b": "G@ARB", "0x11cdb42b0eb46d95f990bedd4695a6e3fa034978": "CRV", "0x040d1edc9569d4bab2d15287dc5a4f10f56a56b8": "BAL", "0x354a6da3fcde098f8389cad84b0182725c6c91de": "COMP", "0x2e9a6df78e42c50b0cefcf9000d0c3a4d34e1dd5": "MKR", } if symbol, exists := tokenMap[addr]; exists { return symbol } // Return truncated address if not in mapping if len(tokenAddress) > 10 { return tokenAddress[:6] + "..." + tokenAddress[len(tokenAddress)-4:] } return tokenAddress } // GetTopOpportunities returns the top ranked arbitrage opportunities func (s *MarketScanner) GetTopOpportunities(limit int) []*profitcalc.RankedOpportunity { return s.opportunityRanker.GetTopOpportunities(limit) } // GetExecutableOpportunities returns executable arbitrage opportunities func (s *MarketScanner) GetExecutableOpportunities(limit int) []*profitcalc.RankedOpportunity { return s.opportunityRanker.GetExecutableOpportunities(limit) } // GetOpportunityStats returns statistics about tracked opportunities func (s *MarketScanner) GetOpportunityStats() map[string]interface{} { return s.opportunityRanker.GetStats() } // GetMarketDataStats returns comprehensive market data statistics func (s *MarketScanner) GetMarketDataStats() map[string]interface{} { if s.marketDataLogger != nil { return s.marketDataLogger.GetStatistics() } return map[string]interface{}{ "status": "market data logger not available", } } // GetCachedTokenInfo returns information about a cached token func (s *MarketScanner) GetCachedTokenInfo(tokenAddr common.Address) (*marketdata.TokenInfo, bool) { if s.marketDataLogger != nil { return s.marketDataLogger.GetTokenInfo(tokenAddr) } return nil, false } // GetCachedPoolInfo returns information about a cached pool func (s *MarketScanner) GetCachedPoolInfo(poolAddr common.Address) (*marketdata.PoolInfo, bool) { if s.marketDataLogger != nil { return s.marketDataLogger.GetPoolInfo(poolAddr) } return nil, false } // GetPoolsForTokenPair returns all cached pools for a token pair func (s *MarketScanner) GetPoolsForTokenPair(token0, token1 common.Address) []*marketdata.PoolInfo { if s.marketDataLogger != nil { return s.marketDataLogger.GetPoolsForTokenPair(token0, token1) } return nil } // GetActiveFactories returns all active DEX factories func (s *MarketScanner) GetActiveFactories() []*marketdata.FactoryInfo { if s.marketDataLogger != nil { return s.marketDataLogger.GetActiveFactories() } return nil } // analyzeLiquidityEvent analyzes liquidity events (add/remove) func (s *MarketScanner) analyzeLiquidityEvent(event events.Event, isAdd bool) { action := "adding" eventType := "mint" if !isAdd { action = "removing" eventType = "burn" } s.logger.Debug(fmt.Sprintf("Analyzing liquidity event (%s) in pool %s", action, event.PoolAddress)) // Get comprehensive pool data to determine factory poolInfo, poolExists := s.marketDataLogger.GetPoolInfo(event.PoolAddress) factory := common.Address{} if poolExists { factory = poolInfo.Factory } else { // Determine factory from known DEX protocols factory = s.getFactoryForProtocol(event.Protocol) } // Create comprehensive liquidity event data for market data logger liquidityData := &marketdata.LiquidityEventData{ TxHash: event.TransactionHash, BlockNumber: event.BlockNumber, LogIndex: uint(0), // Default log index (would need to be extracted from receipt) Timestamp: time.Now(), EventType: eventType, PoolAddress: event.PoolAddress, Factory: factory, Protocol: event.Protocol, Token0: event.Token0, Token1: event.Token1, Amount0: event.Amount0, Amount1: event.Amount1, Liquidity: event.Liquidity, Owner: common.Address{}, // Default owner (would need to be extracted from transaction) Recipient: common.Address{}, // Default recipient (would need to be extracted from transaction) } // Calculate USD values for liquidity amounts liquidityData.Amount0USD, liquidityData.Amount1USD, liquidityData.TotalUSD = s.calculateLiquidityUSDValues(liquidityData) // Log comprehensive liquidity event to market data logger ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := s.marketDataLogger.LogLiquidityEvent(ctx, event, liquidityData); err != nil { s.logger.Debug(fmt.Sprintf("Failed to log liquidity event to market data logger: %v", err)) } // Log the liquidity event to database (legacy) s.logLiquidityEvent(event, eventType) // Update cached pool data s.updatePoolData(event) s.logger.Info(fmt.Sprintf("Liquidity %s event processed for pool %s", action, event.PoolAddress)) } // analyzeNewPoolEvent analyzes new pool creation events func (s *MarketScanner) analyzeNewPoolEvent(event events.Event) { s.logger.Info(fmt.Sprintf("New pool created: %s (protocol: %s)", event.PoolAddress, event.Protocol)) // Add to known pools by fetching and caching the pool data s.logger.Debug(fmt.Sprintf("Adding new pool %s to monitoring", event.PoolAddress)) // Fetch pool data to validate it's a real pool poolData, err := s.getPoolData(event.PoolAddress.Hex()) if err != nil { s.logger.Error(fmt.Sprintf("Failed to fetch data for new pool %s: %v", event.PoolAddress, err)) return } // Validate that this is a real pool contract if poolData.Address == (common.Address{}) { s.logger.Warn(fmt.Sprintf("Invalid pool contract at address %s", event.PoolAddress.Hex())) return } // Log pool data to database s.logPoolData(poolData) s.logger.Info(fmt.Sprintf("Successfully added new pool %s to monitoring (tokens: %s-%s, fee: %d)", event.PoolAddress.Hex(), poolData.Token0.Hex(), poolData.Token1.Hex(), poolData.Fee)) } // isSignificantMovement determines if a price movement is significant enough to exploit func (s *MarketScanner) isSignificantMovement(movement *PriceMovement, threshold float64) bool { // Check if the price impact is above our threshold if movement.PriceImpact > threshold { return true } // Also check if the absolute amount is significant if movement.AmountIn != nil && movement.AmountIn.Cmp(big.NewInt(1000000000000000000)) > 0 { // 1 ETH return true } // For smaller amounts, we need a higher price impact to be significant if movement.AmountIn != nil && movement.AmountIn.Cmp(big.NewInt(100000000000000000)) > 0 { // 0.1 ETH return movement.PriceImpact > threshold/2 } return false } // findRelatedPools finds pools that trade the same token pair func (s *MarketScanner) findRelatedPools(token0, token1 common.Address) []*CachedData { s.logger.Debug(fmt.Sprintf("Finding related pools for token pair %s-%s", token0.Hex(), token1.Hex())) relatedPools := make([]*CachedData, 0) // Use dynamic pool discovery by checking known DEX factories poolAddresses := s.discoverPoolsForPair(token0, token1) s.logger.Debug(fmt.Sprintf("Found %d potential pools for pair %s-%s", len(poolAddresses), token0.Hex(), token1.Hex())) for _, poolAddr := range poolAddresses { poolData, err := s.getPoolData(poolAddr) if err != nil { s.logger.Debug(fmt.Sprintf("No data for pool %s: %v", poolAddr, err)) continue } // Check if this pool trades the same token pair (in either direction) if (poolData.Token0 == token0 && poolData.Token1 == token1) || (poolData.Token0 == token1 && poolData.Token1 == token0) { relatedPools = append(relatedPools, poolData) } } s.logger.Debug(fmt.Sprintf("Found %d related pools", len(relatedPools))) return relatedPools } // discoverPoolsForPair discovers pools for a specific token pair using real factory contracts func (s *MarketScanner) discoverPoolsForPair(token0, token1 common.Address) []string { poolAddresses := make([]string, 0) // Use the CREATE2 calculator to find all possible pools pools, err := s.create2Calculator.FindPoolsForTokenPair(token0, token1) if err != nil { s.logger.Error(fmt.Sprintf("Failed to discover pools for pair %s/%s: %v", token0.Hex(), token1.Hex(), err)) return poolAddresses } // Convert to string addresses for _, pool := range pools { poolAddresses = append(poolAddresses, pool.PoolAddr.Hex()) } s.logger.Debug(fmt.Sprintf("Discovered %d potential pools for pair %s/%s", len(poolAddresses), token0.Hex(), token1.Hex())) return poolAddresses } // estimateProfit estimates the potential profit from an arbitrage opportunity using real slippage protection func (s *MarketScanner) estimateProfit(event events.Event, pool *CachedData, priceDiff float64) *big.Int { // Use comprehensive slippage analysis instead of simplified calculation if s.slippageProtector != nil { return s.calculateProfitWithSlippageProtection(event, pool, priceDiff) } // Fallback to simplified calculation if slippage protection not available return s.calculateSophisticatedProfit(event, pool, priceDiff) } // calculateProfitWithSlippageProtection uses slippage protection for accurate profit estimation func (s *MarketScanner) calculateProfitWithSlippageProtection(event events.Event, pool *CachedData, priceDiff float64) *big.Int { // Create trade parameters from event data tradeParams := &trading.TradeParameters{ TokenIn: event.Token0, TokenOut: event.Token1, AmountIn: event.Amount0, MinAmountOut: new(big.Int).Div(event.Amount1, big.NewInt(100)), // Simplified min amount MaxSlippage: 3.0, // 3% max slippage Deadline: uint64(time.Now().Add(5 * time.Minute).Unix()), Pool: event.PoolAddress, ExpectedPrice: big.NewFloat(1.0), // Simplified expected price CurrentLiquidity: big.NewInt(1000000), // Simplified liquidity } // Analyze slippage protection slippageCheck, err := s.slippageProtector.ValidateTradeParameters(tradeParams) if err != nil { s.logger.Debug(fmt.Sprintf("Slippage analysis failed: %v", err)) return s.calculateSophisticatedProfit(event, pool, priceDiff) } // Don't proceed if trade is not safe if !slippageCheck.IsValid { s.logger.Debug("Trade rejected by slippage protection") return big.NewInt(0) } // Calculate profit considering slippage expectedAmountOut := event.Amount1 // Profit = (expected_out - amount_in) - gas_costs - slippage_buffer profit := new(big.Int).Sub(expectedAmountOut, event.Amount0) // REAL gas cost calculation for competitive MEV on Arbitrum // Base gas: 800k units, Price: 1.5 gwei, MEV premium: 15x = 0.018 ETH total baseGas := big.NewInt(800000) // 800k gas units for flash swap arbitrage gasPrice := big.NewInt(1500000000) // 1.5 gwei base price on Arbitrum mevPremium := big.NewInt(15) // 15x premium for MEV competition gasCostWei := new(big.Int).Mul(baseGas, gasPrice) totalGasCost := new(big.Int).Mul(gasCostWei, mevPremium) profit.Sub(profit, totalGasCost) // Apply safety margin for slippage if slippageCheck.CalculatedSlippage > 0 { slippageMarginFloat := slippageCheck.CalculatedSlippage / 100.0 slippageMargin := new(big.Float).Mul(new(big.Float).SetInt(expectedAmountOut), big.NewFloat(slippageMarginFloat)) slippageMarginInt, _ := slippageMargin.Int(nil) profit.Sub(profit, slippageMarginInt) } // Ensure profit is not negative if profit.Sign() < 0 { return big.NewInt(0) } return profit } // calculateSophisticatedProfit provides advanced profit calculation with MEV considerations func (s *MarketScanner) calculateSophisticatedProfit(event events.Event, pool *CachedData, priceDiff float64) *big.Int { amountIn := new(big.Int).Set(event.Amount0) // Use sophisticated pricing calculation based on Uniswap V3 concentrated liquidity var amountOut *big.Int var err error if pool.SqrtPriceX96 != nil && pool.Liquidity != nil { // Calculate output using proper Uniswap V3 math amountOut, err = s.calculateUniswapV3Output(amountIn, pool) if err != nil { s.logger.Debug(fmt.Sprintf("Failed to calculate V3 output, using fallback: %v", err)) amountOut = s.calculateFallbackOutput(amountIn, priceDiff) } } else { amountOut = s.calculateFallbackOutput(amountIn, priceDiff) } // Calculate arbitrage profit considering market impact marketImpact := s.calculateMarketImpact(amountIn, pool) adjustedAmountOut := new(big.Int).Sub(amountOut, marketImpact) // Calculate gross profit grossProfit := new(big.Int).Sub(adjustedAmountOut, amountIn) // Sophisticated gas cost calculation gasCost := s.calculateDynamicGasCost(event, pool) // MEV competition premium (front-running protection cost) mevPremium := s.calculateMEVPremium(grossProfit, priceDiff) // Calculate net profit after all costs netProfit := new(big.Int).Sub(grossProfit, gasCost) netProfit = netProfit.Sub(netProfit, mevPremium) // Apply slippage tolerance slippageTolerance := s.calculateSlippageTolerance(amountIn, pool) finalProfit := new(big.Int).Sub(netProfit, slippageTolerance) // Ensure profit is positive and meets minimum threshold minProfitThreshold := big.NewInt(1000000000000000000) // 1 ETH minimum if finalProfit.Cmp(minProfitThreshold) < 0 { return big.NewInt(0) } s.logger.Debug(fmt.Sprintf("Sophisticated profit calculation: gross=%s, gas=%s, mev=%s, slippage=%s, net=%s", grossProfit.String(), gasCost.String(), mevPremium.String(), slippageTolerance.String(), finalProfit.String())) return finalProfit } // calculatePriceMovement calculates the price movement from a swap event func (s *MarketScanner) calculatePriceMovement(event events.Event, poolData *CachedData) (*PriceMovement, error) { s.logger.Debug(fmt.Sprintf("Calculating price movement for pool %s", event.PoolAddress)) // Get current price from pool data currentPrice := uniswap.SqrtPriceX96ToPrice(poolData.SqrtPriceX96.ToBig()) if currentPrice == nil { return nil, fmt.Errorf("failed to calculate current price from sqrtPriceX96") } // Calculate price impact based on swap amounts var priceImpact float64 if event.Amount0.Sign() > 0 && event.Amount1.Sign() > 0 { // Both amounts are positive, calculate the impact amount0Float := new(big.Float).SetInt(event.Amount0) amount1Float := new(big.Float).SetInt(event.Amount1) // Price impact = |amount1 / amount0 - current_price| / current_price swapPrice := new(big.Float).Quo(amount1Float, amount0Float) priceDiff := new(big.Float).Sub(swapPrice, currentPrice) priceDiff.Abs(priceDiff) priceImpactFloat := new(big.Float).Quo(priceDiff, currentPrice) priceImpact, _ = priceImpactFloat.Float64() } movement := &PriceMovement{ Token0: event.Token0.Hex(), Token1: event.Token1.Hex(), Pool: event.PoolAddress.Hex(), Protocol: event.Protocol, AmountIn: event.Amount0, AmountOut: event.Amount1, PriceBefore: currentPrice, PriceAfter: currentPrice, // For now, assume same price (could be calculated based on swap) PriceImpact: priceImpact, TickBefore: poolData.Tick, TickAfter: poolData.Tick, // For now, assume same tick Timestamp: time.Now(), } s.logger.Debug(fmt.Sprintf("Price movement calculated: impact=%.6f%%, amount_in=%s", priceImpact*100, event.Amount0.String())) return movement, nil } // findTriangularArbitrageOpportunities looks for triangular arbitrage opportunities func (s *MarketScanner) findTriangularArbitrageOpportunities(event events.Event) []stypes.ArbitrageOpportunity { s.logger.Debug(fmt.Sprintf("Searching for triangular arbitrage opportunities involving pool %s", event.PoolAddress)) opportunities := make([]stypes.ArbitrageOpportunity, 0) // Define common triangular paths on Arbitrum // Get triangular arbitrage paths from token configuration triangularPaths := tokens.GetTriangularPaths() // Check if the event involves any tokens from our triangular paths eventInvolvesPaths := make([]int, 0) for i, path := range triangularPaths { for _, token := range path.Tokens { if token == event.Token0 || token == event.Token1 { eventInvolvesPaths = append(eventInvolvesPaths, i) break } } } // For each relevant triangular path, calculate potential profit for _, pathIdx := range eventInvolvesPaths { path := triangularPaths[pathIdx] // Define test amounts for arbitrage calculation testAmounts := []*big.Int{ big.NewInt(1000000), // 1 USDC (6 decimals) big.NewInt(100000000), // 0.1 WETH (18 decimals) big.NewInt(10000000), // 0.01 WETH (18 decimals) } for _, testAmount := range testAmounts { profit, gasEstimate, err := s.calculateTriangularProfit(path.Tokens, testAmount) if err != nil { s.logger.Debug(fmt.Sprintf("Error calculating triangular profit for %s: %v", path.Name, err)) continue } // Check if profitable after gas costs netProfit := new(big.Int).Sub(profit, gasEstimate) if netProfit.Sign() > 0 { // Calculate ROI roi := 0.0 if testAmount.Sign() > 0 { roiFloat := new(big.Float).Quo(new(big.Float).SetInt(netProfit), new(big.Float).SetInt(testAmount)) roi, _ = roiFloat.Float64() roi *= 100 // Convert to percentage } // Create arbitrage opportunity tokenPaths := make([]string, len(path.Tokens)) for i, token := range path.Tokens { tokenPaths[i] = token.Hex() } // Close the loop by adding the first token at the end tokenPaths = append(tokenPaths, path.Tokens[0].Hex()) opportunity := stypes.ArbitrageOpportunity{ Path: tokenPaths, Pools: []string{}, // Pool addresses will be discovered dynamically Profit: netProfit, GasEstimate: gasEstimate, ROI: roi, Protocol: fmt.Sprintf("Triangular_%s", path.Name), } opportunities = append(opportunities, opportunity) s.logger.Info(fmt.Sprintf("Found triangular arbitrage opportunity: %s, Profit: %s, ROI: %.2f%%", path.Name, netProfit.String(), roi)) } } } return opportunities } // calculateTriangularProfit calculates the profit from a triangular arbitrage path func (s *MarketScanner) calculateTriangularProfit(tokens []common.Address, initialAmount *big.Int) (*big.Int, *big.Int, error) { if len(tokens) < 3 { return nil, nil, fmt.Errorf("triangular arbitrage requires at least 3 tokens") } currentAmount := new(big.Int).Set(initialAmount) totalGasCost := big.NewInt(0) // Simulate trading through the triangular path for i := 0; i < len(tokens); i++ { nextIndex := (i + 1) % len(tokens) tokenIn := tokens[i] tokenOut := tokens[nextIndex] // Get pools that trade this token pair relatedPools := s.findRelatedPools(tokenIn, tokenOut) if len(relatedPools) == 0 { // No pools found for this pair, use estimation // Apply a 0.3% fee reduction as approximation currentAmount = new(big.Int).Mul(currentAmount, big.NewInt(997)) currentAmount = new(big.Int).Div(currentAmount, big.NewInt(1000)) } else { // Use the best pool for this trade bestPool := relatedPools[0] // Calculate swap output using current amount outputAmount, err := s.calculateSwapOutput(currentAmount, bestPool, tokenIn, tokenOut) if err != nil { s.logger.Debug(fmt.Sprintf("Error calculating swap output: %v", err)) // Fallback to simple fee calculation currentAmount = new(big.Int).Mul(currentAmount, big.NewInt(997)) currentAmount = new(big.Int).Div(currentAmount, big.NewInt(1000)) } else { currentAmount = outputAmount } } // Add gas cost for this hop (estimated) hopGas := big.NewInt(150000) // ~150k gas per swap totalGasCost.Add(totalGasCost, hopGas) } // Calculate profit (final amount - initial amount) profit := new(big.Int).Sub(currentAmount, initialAmount) return profit, totalGasCost, nil } // calculateSwapOutput calculates the output amount for a token swap func (s *MarketScanner) calculateSwapOutput(amountIn *big.Int, pool *CachedData, tokenIn, tokenOut common.Address) (*big.Int, error) { if pool.SqrtPriceX96 == nil || pool.Liquidity == nil { return nil, fmt.Errorf("missing pool price or liquidity data") } // Convert sqrtPriceX96 to price for calculation price := uniswap.SqrtPriceX96ToPrice(pool.SqrtPriceX96.ToBig()) // Use sophisticated Uniswap V3 concentrated liquidity calculation var amountOut *big.Int var err error // Try sophisticated V3 calculation first amountOut, err = s.calculateUniswapV3Output(amountIn, pool) if err != nil { s.logger.Debug(fmt.Sprintf("V3 calculation failed, using price-based fallback: %v", err)) // Fallback to price-based calculation with proper fee handling amountInFloat := new(big.Float).SetInt(amountIn) var amountOutFloat *big.Float if tokenIn == pool.Token0 { // Token0 -> Token1: multiply by price amountOutFloat = new(big.Float).Mul(amountInFloat, price) } else { // Token1 -> Token0: divide by price amountOutFloat = new(big.Float).Quo(amountInFloat, price) } // Apply dynamic fee based on pool configuration fee := pool.Fee if fee == 0 { fee = 3000 // Default 0.3% } // Calculate precise fee rate feeRateFloat := big.NewFloat(1.0) feeRateFloat.Sub(feeRateFloat, new(big.Float).Quo(big.NewFloat(float64(fee)), big.NewFloat(1000000))) amountOutFloat.Mul(amountOutFloat, feeRateFloat) // Convert back to big.Int amountOut = new(big.Int) amountOutFloat.Int(amountOut) } s.logger.Debug(fmt.Sprintf("Swap calculation: amountIn=%s, amountOut=%s, tokenIn=%s, tokenOut=%s", amountIn.String(), amountOut.String(), tokenIn.Hex(), tokenOut.Hex())) return amountOut, nil } // findArbitrageOpportunities looks for arbitrage opportunities based on price movements func (s *MarketScanner) findArbitrageOpportunities(event events.Event, movement *PriceMovement) []stypes.ArbitrageOpportunity { s.logger.Debug(fmt.Sprintf("Searching for arbitrage opportunities for pool %s", event.PoolAddress)) opportunities := make([]stypes.ArbitrageOpportunity, 0) // Get related pools for the same token pair relatedPools := s.findRelatedPools(event.Token0, event.Token1) // If we have related pools, compare prices if len(relatedPools) > 0 { // Get the current price in this pool currentPrice := movement.PriceBefore // Compare with prices in related pools for _, pool := range relatedPools { // Skip the same pool if pool.Address == event.PoolAddress { continue } // Get pool data poolData, err := s.getPoolData(pool.Address.Hex()) if err != nil { s.logger.Error(fmt.Sprintf("Error getting pool data for related pool %s: %v", pool.Address.Hex(), err)) continue } // Check if poolData.SqrtPriceX96 is nil to prevent panic if poolData.SqrtPriceX96 == nil { s.logger.Error(fmt.Sprintf("Pool data for %s has nil SqrtPriceX96", pool.Address.Hex())) continue } // Calculate price in the related pool relatedPrice := uniswap.SqrtPriceX96ToPrice(poolData.SqrtPriceX96.ToBig()) // Check if currentPrice or relatedPrice is nil to prevent panic if currentPrice == nil || relatedPrice == nil { s.logger.Error(fmt.Sprintf("Nil price detected for pool comparison")) continue } // Calculate price difference priceDiff := new(big.Float).Sub(currentPrice, relatedPrice) priceDiffRatio := new(big.Float).Quo(priceDiff, relatedPrice) // If there's a significant price difference, we might have an arbitrage opportunity priceDiffFloat, _ := priceDiffRatio.Float64() if priceDiffFloat > 0.005 { // 0.5% threshold // Estimate potential profit estimatedProfit := s.estimateProfit(event, pool, priceDiffFloat) if estimatedProfit != nil && estimatedProfit.Sign() > 0 { opp := stypes.ArbitrageOpportunity{ Path: []string{event.Token0.Hex(), event.Token1.Hex()}, Pools: []string{event.PoolAddress.Hex(), pool.Address.Hex()}, Profit: estimatedProfit, GasEstimate: big.NewInt(300000), // Estimated gas cost ROI: priceDiffFloat * 100, // Convert to percentage Protocol: fmt.Sprintf("%s->%s", event.Protocol, pool.Protocol), } opportunities = append(opportunities, opp) s.logger.Info(fmt.Sprintf("Found arbitrage opportunity: %+v", opp)) } } } } // Also look for triangular arbitrage opportunities triangularOpps := s.findTriangularArbitrageOpportunities(event) opportunities = append(opportunities, triangularOpps...) return opportunities } // executeArbitrageOpportunity executes an arbitrage opportunity using the smart contract func (s *MarketScanner) executeArbitrageOpportunity(opportunity stypes.ArbitrageOpportunity) { // Check if contract executor is available if s.contractExecutor == nil { s.logger.Warn("Contract executor not available, skipping arbitrage execution") return } // Only execute opportunities with sufficient profit minProfitThreshold := big.NewInt(10000000000000000) // 0.01 ETH minimum profit if opportunity.Profit.Cmp(minProfitThreshold) < 0 { s.logger.Debug(fmt.Sprintf("Arbitrage opportunity profit too low: %s < %s", opportunity.Profit.String(), minProfitThreshold.String())) return } s.logger.Info(fmt.Sprintf("Executing arbitrage opportunity with profit: %s", opportunity.Profit.String())) // Execute the arbitrage opportunity ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() var tx *etypes.Transaction var err error // Determine if this is a triangular arbitrage or standard arbitrage if len(opportunity.Path) == 3 && len(opportunity.Pools) == 3 { // Triangular arbitrage tx, err = s.contractExecutor.ExecuteTriangularArbitrage(ctx, opportunity) } else { // Standard arbitrage tx, err = s.contractExecutor.ExecuteArbitrage(ctx, opportunity) } if err != nil { s.logger.Error(fmt.Sprintf("Failed to execute arbitrage opportunity: %v", err)) return } s.logger.Info(fmt.Sprintf("Arbitrage transaction submitted: %s", tx.Hash().Hex())) } // logSwapEvent logs a swap event to the database func (s *MarketScanner) logSwapEvent(event events.Event) { if s.database == nil { return // Database not available } // Convert event to database record swapEvent := &database.SwapEvent{ Timestamp: time.Now(), BlockNumber: event.BlockNumber, TxHash: common.Hash{}, // TxHash not available in Event struct PoolAddress: event.PoolAddress, Token0: event.Token0, Token1: event.Token1, Amount0In: event.Amount0, Amount1In: event.Amount1, Amount0Out: big.NewInt(0), // Would need to calculate from event data Amount1Out: big.NewInt(0), // Would need to calculate from event data Sender: common.Address{}, // Would need to extract from transaction Recipient: common.Address{}, // Would need to extract from transaction Protocol: event.Protocol, } // Log the swap event asynchronously to avoid blocking go func() { if err := s.database.InsertSwapEvent(swapEvent); err != nil { s.logger.Debug(fmt.Sprintf("Failed to log swap event: %v", err)) } }() } // logLiquidityEvent logs a liquidity event to the database func (s *MarketScanner) logLiquidityEvent(event events.Event, eventType string) { if s.database == nil { return // Database not available } // Convert event to database record liquidityEvent := &database.LiquidityEvent{ Timestamp: time.Now(), BlockNumber: event.BlockNumber, TxHash: common.Hash{}, // TxHash not available in Event struct LogIndex: uint(0), // Default log index (would need to be extracted from receipt) PoolAddress: event.PoolAddress, Factory: s.getFactoryForProtocol(event.Protocol), Router: common.Address{}, // Would need router resolution based on transaction Token0: event.Token0, Token1: event.Token1, Liquidity: event.Liquidity.ToBig(), // Convert uint256 to big.Int Amount0: event.Amount0, Amount1: event.Amount1, TokenId: big.NewInt(0), // Default token ID for V3 positions TickLower: int32(0), // Default tick range TickUpper: int32(0), // Default tick range Owner: common.Address{}, // Would need to extract from transaction Recipient: common.Address{}, // Would need to extract from transaction EventType: eventType, Protocol: event.Protocol, Amount0USD: 0.0, // Will be calculated by market data logger Amount1USD: 0.0, // Will be calculated by market data logger TotalUSD: 0.0, // Will be calculated by market data logger } // Log the liquidity event asynchronously to avoid blocking go func() { if err := s.database.InsertLiquidityEvent(liquidityEvent); err != nil { s.logger.Debug(fmt.Sprintf("Failed to log liquidity event: %v", err)) } }() } // logPoolData logs pool data to the database func (s *MarketScanner) logPoolData(poolData *CachedData) { if s.database == nil { return // Database not available } // Convert cached data to database record dbPoolData := &database.PoolData{ Address: poolData.Address, Token0: poolData.Token0, Token1: poolData.Token1, Fee: poolData.Fee, Liquidity: poolData.Liquidity.ToBig(), SqrtPriceX96: poolData.SqrtPriceX96.ToBig(), Tick: int64(poolData.Tick), LastUpdated: time.Now(), Protocol: poolData.Protocol, } // Log the pool data asynchronously to avoid blocking go func() { if err := s.database.InsertPoolData(dbPoolData); err != nil { s.logger.Debug(fmt.Sprintf("Failed to log pool data: %v", err)) } }() } // PriceMovement represents a potential price movement type PriceMovement struct { Token0 string // Token address Token1 string // Token address Pool string // Pool address Protocol string // DEX protocol AmountIn *big.Int // Amount of token being swapped in AmountOut *big.Int // Amount of token being swapped out PriceBefore *big.Float // Price before the swap PriceAfter *big.Float // Price after the swap (to be calculated) PriceImpact float64 // Calculated price impact TickBefore int // Tick before the swap TickAfter int // Tick after the swap (to be calculated) Timestamp time.Time // Event timestamp } // CachedData represents cached pool data type CachedData struct { Address common.Address Token0 common.Address Token1 common.Address Fee int64 Liquidity *uint256.Int SqrtPriceX96 *uint256.Int Tick int TickSpacing int LastUpdated time.Time Protocol string } // getPoolData retrieves pool data with caching func (s *MarketScanner) getPoolData(poolAddress string) (*CachedData, error) { // Check cache first cacheKey := fmt.Sprintf("pool_%s", poolAddress) s.cacheMutex.RLock() if data, exists := s.cache[cacheKey]; exists && time.Since(data.LastUpdated) < s.cacheTTL { s.cacheMutex.RUnlock() s.logger.Debug(fmt.Sprintf("Cache hit for pool %s", poolAddress)) return data, nil } s.cacheMutex.RUnlock() // Use singleflight to prevent duplicate requests result, err, _ := s.cacheGroup.Do(cacheKey, func() (interface{}, error) { return s.fetchPoolData(poolAddress) }) if err != nil { return nil, err } poolData := result.(*CachedData) // Update cache s.cacheMutex.Lock() s.cache[cacheKey] = poolData s.cacheMutex.Unlock() s.logger.Debug(fmt.Sprintf("Fetched and cached pool data for %s", poolAddress)) return poolData, nil } // fetchPoolData fetches pool data from the blockchain func (s *MarketScanner) fetchPoolData(poolAddress string) (*CachedData, error) { s.logger.Debug(fmt.Sprintf("Fetching pool data for %s", poolAddress)) address := common.HexToAddress(poolAddress) // In test environment, return mock data to avoid network calls if s.isTestEnvironment() { return s.getMockPoolData(poolAddress), nil } // Create RPC client connection // Get RPC endpoint from config or environment rpcEndpoint := os.Getenv("ARBITRUM_RPC_ENDPOINT") if rpcEndpoint == "" { rpcEndpoint = "wss://arbitrum-mainnet.core.chainstack.com/f69d14406bc00700da9b936504e1a870" // fallback } client, err := ethclient.Dial(rpcEndpoint) if err != nil { return nil, fmt.Errorf("failed to connect to Ethereum node: %w", err) } defer client.Close() // Create Uniswap V3 pool interface pool := uniswap.NewUniswapV3Pool(address, client) // Validate that this is a real pool contract if !uniswap.IsValidPool(context.Background(), client, address) { return nil, fmt.Errorf("invalid pool contract at address %s", address.Hex()) } // Fetch real pool state from the blockchain ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() poolState, err := pool.GetPoolState(ctx) if err != nil { s.logger.Warn(fmt.Sprintf("Failed to fetch real pool state for %s: %v", address.Hex(), err)) return nil, fmt.Errorf("failed to fetch pool state: %w", err) } // Determine tick spacing based on fee tier tickSpacing := 60 // Default for 0.3% fee switch poolState.Fee { case 100: // 0.01% tickSpacing = 1 case 500: // 0.05% tickSpacing = 10 case 3000: // 0.3% tickSpacing = 60 case 10000: // 1% tickSpacing = 200 } // Determine protocol (assume UniswapV3 for now, could be enhanced to detect protocol) protocol := "UniswapV3" // Create pool data from real blockchain state poolData := &CachedData{ Address: address, Token0: poolState.Token0, Token1: poolState.Token1, Fee: poolState.Fee, Liquidity: poolState.Liquidity, SqrtPriceX96: poolState.SqrtPriceX96, Tick: poolState.Tick, TickSpacing: tickSpacing, Protocol: protocol, LastUpdated: time.Now(), } s.logger.Info(fmt.Sprintf("Fetched real pool data for %s: Token0=%s, Token1=%s, Fee=%d, Liquidity=%s", address.Hex(), poolState.Token0.Hex(), poolState.Token1.Hex(), poolState.Fee, poolState.Liquidity.String())) return poolData, nil } // updatePoolData updates cached pool data from an event func (s *MarketScanner) updatePoolData(event events.Event) { poolKey := event.PoolAddress.Hex() s.cacheMutex.Lock() defer s.cacheMutex.Unlock() // Update existing cache entry or create new one if pool, exists := s.cache[poolKey]; exists { // Update liquidity if provided if event.Liquidity != nil { pool.Liquidity = event.Liquidity } // Update sqrtPriceX96 if provided if event.SqrtPriceX96 != nil { pool.SqrtPriceX96 = event.SqrtPriceX96 } // Update tick if provided if event.Tick != 0 { pool.Tick = event.Tick } // Update last updated time pool.LastUpdated = time.Now() // Log updated pool data to database s.logPoolData(pool) } else { // Create new pool entry pool := &CachedData{ Address: event.PoolAddress, Token0: event.Token0, Token1: event.Token1, Fee: 3000, // Default fee since not available in Event struct Liquidity: event.Liquidity, SqrtPriceX96: event.SqrtPriceX96, Tick: event.Tick, TickSpacing: getTickSpacing(3000), // Default fee Protocol: event.Protocol, LastUpdated: time.Now(), } s.cache[poolKey] = pool // Log new pool data to database s.logPoolData(pool) } s.logger.Debug(fmt.Sprintf("Updated cache for pool %s", event.PoolAddress.Hex())) } // cleanupCache removes expired cache entries func (s *MarketScanner) cleanupCache() { ticker := time.NewTicker(10 * time.Minute) defer ticker.Stop() for { select { case <-ticker.C: s.cacheMutex.Lock() for key, data := range s.cache { if time.Since(data.LastUpdated) > s.cacheTTL { delete(s.cache, key) s.logger.Debug(fmt.Sprintf("Removed expired cache entry: %s", key)) } } s.cacheMutex.Unlock() } } } // isTestEnvironment checks if we're running in a test environment func (s *MarketScanner) isTestEnvironment() bool { // Check for explicit test environment variable if os.Getenv("GO_TEST") == "true" { return true } // Check for testing framework flags for _, arg := range os.Args { if strings.HasPrefix(arg, "-test.") || arg == "test" { return true } } // Check if the program name is from 'go test' progName := os.Args[0] if strings.Contains(progName, ".test") || strings.HasSuffix(progName, ".test") { return true } // Check if running under go test command if strings.Contains(progName, "go_build_") && strings.Contains(progName, "_test") { return true } // Default to production mode - NEVER return true by default return false } // getMockPoolData returns mock pool data for testing func (s *MarketScanner) getMockPoolData(poolAddress string) *CachedData { // Create deterministic mock data based on pool address mockTokens := tokens.GetArbitrumTokens() // Use different token pairs based on pool address var token0, token1 common.Address switch poolAddress { case "0x88e6A0c2dDD26FEEb64F039a2c41296FcB3f5640": token0 = mockTokens.USDC token1 = mockTokens.WETH case "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc": token0 = mockTokens.USDC token1 = mockTokens.WETH default: token0 = mockTokens.USDC token1 = mockTokens.WETH } // Convert big.Int to uint256.Int for compatibility liquidity := uint256.NewInt(1000000000000000000) // 1 ETH equivalent // Create a reasonable sqrtPriceX96 value for ~2000 USDC per ETH sqrtPrice, _ := uint256.FromHex("0x668F0BD9C5DB9D2F2DF6A0E4C") // Reasonable value return &CachedData{ Address: common.HexToAddress(poolAddress), Token0: token0, Token1: token1, Fee: 3000, // 0.3% TickSpacing: 60, Liquidity: liquidity, SqrtPriceX96: sqrtPrice, Tick: -74959, // Corresponds to the sqrt price above Protocol: "UniswapV3", LastUpdated: time.Now(), } } // getTickSpacing returns tick spacing based on fee tier func getTickSpacing(fee int64) int { switch fee { case 100: // 0.01% return 1 case 500: // 0.05% return 10 case 3000: // 0.3% return 60 case 10000: // 1% return 200 default: return 60 // Default to 0.3% fee spacing } } // calculateUniswapV3Output calculates swap output using proper Uniswap V3 concentrated liquidity math func (s *MarketScanner) calculateUniswapV3Output(amountIn *big.Int, pool *CachedData) (*big.Int, error) { // Calculate the new sqrt price after the swap using Uniswap V3 formula // Δ√P = (ΔY * √P) / (L + ΔY * √P) sqrtPrice := pool.SqrtPriceX96.ToBig() liquidity := pool.Liquidity.ToBig() // Validate amount size for calculations if amountIn.BitLen() > 256 { return nil, fmt.Errorf("amountIn too large for calculations") } // Calculate new sqrtPrice using concentrated liquidity formula numerator := new(big.Int).Mul(amountIn, sqrtPrice) denominator := new(big.Int).Add(liquidity, numerator) newSqrtPrice := new(big.Int).Div(new(big.Int).Mul(liquidity, sqrtPrice), denominator) // Calculate output amount: ΔY = L * (√P₀ - √P₁) priceDiff := new(big.Int).Sub(sqrtPrice, newSqrtPrice) amountOut := new(big.Int).Mul(liquidity, priceDiff) // Apply fee (get fee from pool or default to 3000 = 0.3%) fee := pool.Fee if fee == 0 { fee = 3000 // Default 0.3% } // Calculate fee amount feeAmount := new(big.Int).Mul(amountOut, big.NewInt(int64(fee))) feeAmount = feeAmount.Div(feeAmount, big.NewInt(1000000)) // Subtract fee from output finalAmountOut := new(big.Int).Sub(amountOut, feeAmount) s.logger.Debug(fmt.Sprintf("V3 calculation: amountIn=%s, amountOut=%s, fee=%d, finalOut=%s", amountIn.String(), amountOut.String(), fee, finalAmountOut.String())) return finalAmountOut, nil } // calculateFallbackOutput provides fallback calculation when V3 math fails func (s *MarketScanner) calculateFallbackOutput(amountIn *big.Int, priceDiff float64) *big.Int { // Simple linear approximation based on price difference priceDiffInt := big.NewInt(int64(priceDiff * 1000000)) amountOut := new(big.Int).Mul(amountIn, priceDiffInt) amountOut = amountOut.Div(amountOut, big.NewInt(1000000)) // Apply standard 0.3% fee fee := new(big.Int).Mul(amountOut, big.NewInt(3000)) fee = fee.Div(fee, big.NewInt(1000000)) return new(big.Int).Sub(amountOut, fee) } // calculateMarketImpact estimates the market impact of a large trade func (s *MarketScanner) calculateMarketImpact(amountIn *big.Int, pool *CachedData) *big.Int { if pool.Liquidity == nil { return big.NewInt(0) } // Market impact increases with trade size relative to liquidity liquidity := pool.Liquidity.ToBig() // Calculate impact ratio: amountIn / liquidity impactRatio := new(big.Float).Quo(new(big.Float).SetInt(amountIn), new(big.Float).SetInt(liquidity)) // Impact increases quadratically for large trades impactSquared := new(big.Float).Mul(impactRatio, impactRatio) // Convert back to wei amount (impact as percentage of trade) impact := new(big.Float).Mul(new(big.Float).SetInt(amountIn), impactSquared) result := new(big.Int) impact.Int(result) // Cap maximum impact at 10% of trade size maxImpact := new(big.Int).Div(amountIn, big.NewInt(10)) if result.Cmp(maxImpact) > 0 { result = maxImpact } return result } // calculateDynamicGasCost calculates gas cost based on current network conditions func (s *MarketScanner) calculateDynamicGasCost(event events.Event, pool *CachedData) *big.Int { // Base gas costs for different operation types baseGas := big.NewInt(200000) // Simple swap // Increase gas for complex operations if pool.Fee == 500 { // V3 concentrated position baseGas = big.NewInt(350000) } else if event.Protocol == "UniswapV3" { // V3 operations generally more expensive baseGas = big.NewInt(300000) } // Get current gas price (simplified - in production would fetch from network) gasPrice := big.NewInt(2000000000) // 2 gwei base // Add priority fee for MEV transactions priorityFee := big.NewInt(5000000000) // 5 gwei priority totalGasPrice := new(big.Int).Add(gasPrice, priorityFee) // Calculate total gas cost gasCost := new(big.Int).Mul(baseGas, totalGasPrice) s.logger.Debug(fmt.Sprintf("Gas calculation: baseGas=%s, gasPrice=%s, totalCost=%s", baseGas.String(), totalGasPrice.String(), gasCost.String())) return gasCost } // calculateMEVPremium calculates the premium needed to compete with other MEV bots func (s *MarketScanner) calculateMEVPremium(grossProfit *big.Int, priceDiff float64) *big.Int { // MEV premium increases with profit potential profitFloat := new(big.Float).SetInt(grossProfit) // Base premium: 5% of gross profit basePremium := new(big.Float).Mul(profitFloat, big.NewFloat(0.05)) // Increase premium for highly profitable opportunities (more competition) if priceDiff > 0.02 { // > 2% price difference competitionMultiplier := big.NewFloat(1.5 + priceDiff*10) // Scale with opportunity basePremium.Mul(basePremium, competitionMultiplier) } // Convert to big.Int premium := new(big.Int) basePremium.Int(premium) // Cap premium at 30% of gross profit maxPremium := new(big.Int).Div(grossProfit, big.NewInt(3)) if premium.Cmp(maxPremium) > 0 { premium = maxPremium } return premium } // calculateSlippageTolerance calculates acceptable slippage for the trade func (s *MarketScanner) calculateSlippageTolerance(amountIn *big.Int, pool *CachedData) *big.Int { // Base slippage tolerance: 0.5% baseSlippage := new(big.Float).Mul(new(big.Float).SetInt(amountIn), big.NewFloat(0.005)) // Increase slippage tolerance for larger trades relative to liquidity if pool.Liquidity != nil { liquidity := pool.Liquidity.ToBig() tradeRatio := new(big.Float).Quo(new(big.Float).SetInt(amountIn), new(big.Float).SetInt(liquidity)) // If trade is > 1% of liquidity, increase slippage tolerance if ratio, _ := tradeRatio.Float64(); ratio > 0.01 { multiplier := big.NewFloat(1 + ratio*5) // Scale slippage with trade size baseSlippage.Mul(baseSlippage, multiplier) } } // Convert to big.Int slippage := new(big.Int) baseSlippage.Int(slippage) // Cap maximum slippage at 2% of trade amount maxSlippage := new(big.Int).Div(amountIn, big.NewInt(50)) // 2% if slippage.Cmp(maxSlippage) > 0 { slippage = maxSlippage } return slippage } // getFactoryForProtocol returns the factory address for a known DEX protocol func (s *MarketScanner) getFactoryForProtocol(protocol string) common.Address { // Known factory addresses on Arbitrum knownFactories := map[string]common.Address{ "UniswapV3": common.HexToAddress("0x1F98431c8aD98523631AE4a59f267346ea31F984"), "UniswapV2": common.HexToAddress("0xc35DADB65012eC5796536bD9864eD8773aBc74C4"), // SushiSwap V2 factory "SushiSwap": common.HexToAddress("0xc35DADB65012eC5796536bD9864eD8773aBc74C4"), "Camelot": common.HexToAddress("0x6EcCab422D763aC031210895C81787E87B82A80f"), "TraderJoe": common.HexToAddress("0xaE4EC9901c3076D0DdBe76A520F9E90a6227aCB7"), "Balancer": common.HexToAddress("0xBA12222222228d8Ba445958a75a0704d566BF2C8"), "Curve": common.HexToAddress("0x445FE580eF8d70FF569aB36e80c647af338db351"), } if factory, exists := knownFactories[protocol]; exists { return factory } // Default to UniswapV3 if unknown return knownFactories["UniswapV3"] } // calculateSwapUSDValues calculates USD values for swap amounts using the profit calculator's price oracle func (s *MarketScanner) calculateSwapUSDValues(swapData *marketdata.SwapEventData, fee uint32) (amountInUSD, amountOutUSD, feeUSD float64) { if s.profitCalculator == nil { return 0, 0, 0 } // Get token prices in USD token0Price := s.getTokenPriceUSD(swapData.Token0) token1Price := s.getTokenPriceUSD(swapData.Token1) // Calculate decimals for proper conversion token0Decimals := s.getTokenDecimals(swapData.Token0) token1Decimals := s.getTokenDecimals(swapData.Token1) // Calculate amount in USD if swapData.Amount0In != nil && swapData.Amount0In.Sign() > 0 { amount0InFloat := s.bigIntToFloat(swapData.Amount0In, token0Decimals) amountInUSD = amount0InFloat * token0Price } else if swapData.Amount1In != nil && swapData.Amount1In.Sign() > 0 { amount1InFloat := s.bigIntToFloat(swapData.Amount1In, token1Decimals) amountInUSD = amount1InFloat * token1Price } // Calculate amount out USD if swapData.Amount0Out != nil && swapData.Amount0Out.Sign() > 0 { amount0OutFloat := s.bigIntToFloat(swapData.Amount0Out, token0Decimals) amountOutUSD = amount0OutFloat * token0Price } else if swapData.Amount1Out != nil && swapData.Amount1Out.Sign() > 0 { amount1OutFloat := s.bigIntToFloat(swapData.Amount1Out, token1Decimals) amountOutUSD = amount1OutFloat * token1Price } // Calculate fee USD (fee tier as percentage of input amount) feePercent := float64(fee) / 1000000.0 // Convert from basis points feeUSD = amountInUSD * feePercent return amountInUSD, amountOutUSD, feeUSD } // calculateSwapPriceImpact calculates the price impact of a swap based on pool liquidity and amounts func (s *MarketScanner) calculateSwapPriceImpact(event events.Event, swapData *marketdata.SwapEventData) float64 { if event.SqrtPriceX96 == nil || event.Liquidity == nil { return 0.0 } // Get pre-swap price from sqrtPriceX96 prePrice := s.sqrtPriceX96ToPrice(event.SqrtPriceX96) if prePrice == 0 { return 0.0 } // Calculate effective swap size in token0 terms var swapSize *big.Int if swapData.Amount0In != nil && swapData.Amount0In.Sign() > 0 { swapSize = swapData.Amount0In } else if swapData.Amount0Out != nil && swapData.Amount0Out.Sign() > 0 { swapSize = swapData.Amount0Out } else { return 0.0 } // Calculate price impact as percentage of pool liquidity liquidity := event.Liquidity.ToBig() if liquidity.Sign() == 0 { return 0.0 } // Simplified price impact calculation: impact = (swapSize^2) / (2 * liquidity) // This approximates the quadratic price impact in AMMs swapSizeFloat := new(big.Float).SetInt(swapSize) liquidityFloat := new(big.Float).SetInt(liquidity) // swapSize^2 swapSizeSquared := new(big.Float).Mul(swapSizeFloat, swapSizeFloat) // 2 * liquidity twoLiquidity := new(big.Float).Mul(liquidityFloat, big.NewFloat(2.0)) // price impact = swapSize^2 / (2 * liquidity) priceImpact := new(big.Float).Quo(swapSizeSquared, twoLiquidity) // Convert to percentage priceImpactPercent, _ := priceImpact.Float64() return priceImpactPercent * 100.0 } // getTokenPriceUSD gets the USD price of a token using various price sources func (s *MarketScanner) getTokenPriceUSD(tokenAddr common.Address) float64 { // Known token prices (in a production system, this would query price oracles) knownPrices := map[common.Address]float64{ common.HexToAddress("0x82af49447d8a07e3bd95bd0d56f35241523fbab1"): 2000.0, // WETH common.HexToAddress("0xaf88d065e77c8cc2239327c5edb3a432268e5831"): 1.0, // USDC common.HexToAddress("0xff970a61a04b1ca14834a43f5de4533ebddb5cc8"): 1.0, // USDC.e common.HexToAddress("0xfd086bc7cd5c481dcc9c85ebe478a1c0b69fcbb9"): 1.0, // USDT common.HexToAddress("0x2f2a2543b76a4166549f7aab2e75bef0aefc5b0f"): 43000.0, // WBTC common.HexToAddress("0x912ce59144191c1204e64559fe8253a0e49e6548"): 0.75, // ARB common.HexToAddress("0xfc5a1a6eb076a2c7ad06ed22c90d7e710e35ad0a"): 45.0, // GMX common.HexToAddress("0xf97f4df75117a78c1a5a0dbb814af92458539fb4"): 12.0, // LINK common.HexToAddress("0xfa7f8980b0f1e64a2062791cc3b0871572f1f7f0"): 8.0, // UNI common.HexToAddress("0xba5ddd1f9d7f570dc94a51479a000e3bce967196"): 85.0, // AAVE } if price, exists := knownPrices[tokenAddr]; exists { return price } // For unknown tokens, return 0 (in production, would query price oracle or DEX) return 0.0 } // getTokenDecimals returns the decimal places for a token func (s *MarketScanner) getTokenDecimals(tokenAddr common.Address) uint8 { // Known token decimals knownDecimals := map[common.Address]uint8{ common.HexToAddress("0x82af49447d8a07e3bd95bd0d56f35241523fbab1"): 18, // WETH common.HexToAddress("0xaf88d065e77c8cc2239327c5edb3a432268e5831"): 6, // USDC common.HexToAddress("0xff970a61a04b1ca14834a43f5de4533ebddb5cc8"): 6, // USDC.e common.HexToAddress("0xfd086bc7cd5c481dcc9c85ebe478a1c0b69fcbb9"): 6, // USDT common.HexToAddress("0x2f2a2543b76a4166549f7aab2e75bef0aefc5b0f"): 8, // WBTC common.HexToAddress("0x912ce59144191c1204e64559fe8253a0e49e6548"): 18, // ARB common.HexToAddress("0xfc5a1a6eb076a2c7ad06ed22c90d7e710e35ad0a"): 18, // GMX common.HexToAddress("0xf97f4df75117a78c1a5a0dbb814af92458539fb4"): 18, // LINK common.HexToAddress("0xfa7f8980b0f1e64a2062791cc3b0871572f1f7f0"): 18, // UNI common.HexToAddress("0xba5ddd1f9d7f570dc94a51479a000e3bce967196"): 18, // AAVE } if decimals, exists := knownDecimals[tokenAddr]; exists { return decimals } // Default to 18 for unknown tokens return 18 } // bigIntToFloat converts a big.Int amount to float64 accounting for token decimals func (s *MarketScanner) bigIntToFloat(amount *big.Int, decimals uint8) float64 { if amount == nil { return 0.0 } divisor := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(decimals)), nil) amountFloat := new(big.Float).SetInt(amount) divisorFloat := new(big.Float).SetInt(divisor) result := new(big.Float).Quo(amountFloat, divisorFloat) resultFloat, _ := result.Float64() return resultFloat } // sqrtPriceX96ToPrice converts sqrtPriceX96 to a regular price func (s *MarketScanner) sqrtPriceX96ToPrice(sqrtPriceX96 *uint256.Int) float64 { if sqrtPriceX96 == nil { return 0.0 } // Convert sqrtPriceX96 to price: price = (sqrtPriceX96 / 2^96)^2 sqrtPrice := new(big.Float).SetInt(sqrtPriceX96.ToBig()) q96 := new(big.Float).SetInt(new(big.Int).Lsh(big.NewInt(1), 96)) normalizedSqrt := new(big.Float).Quo(sqrtPrice, q96) price := new(big.Float).Mul(normalizedSqrt, normalizedSqrt) priceFloat, _ := price.Float64() return priceFloat } // calculateLiquidityUSDValues calculates USD values for liquidity event amounts func (s *MarketScanner) calculateLiquidityUSDValues(liquidityData *marketdata.LiquidityEventData) (amount0USD, amount1USD, totalUSD float64) { // Get token prices in USD token0Price := s.getTokenPriceUSD(liquidityData.Token0) token1Price := s.getTokenPriceUSD(liquidityData.Token1) // Calculate decimals for proper conversion token0Decimals := s.getTokenDecimals(liquidityData.Token0) token1Decimals := s.getTokenDecimals(liquidityData.Token1) // Calculate amount0 USD if liquidityData.Amount0 != nil { amount0Float := s.bigIntToFloat(liquidityData.Amount0, token0Decimals) amount0USD = amount0Float * token0Price } // Calculate amount1 USD if liquidityData.Amount1 != nil { amount1Float := s.bigIntToFloat(liquidityData.Amount1, token1Decimals) amount1USD = amount1Float * token1Price } // Total USD value totalUSD = amount0USD + amount1USD return amount0USD, amount1USD, totalUSD }