package market import ( "context" "errors" "fmt" "math/big" "os" "strings" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" "github.com/holiman/uint256" "golang.org/x/sync/singleflight" "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/internal/validation" "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" ) // safeConvertInt64ToUint64 safely converts an int64 to uint64, ensuring no negative values func safeConvertInt64ToUint64(v int64) uint64 { if v < 0 { return 0 } return uint64(v) } // 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 addressValidator *validation.AddressValidator poolBlacklist map[common.Address]BlacklistReason // Pools that consistently fail RPC calls blacklistMutex sync.RWMutex } // BlacklistReason contains information about why a pool was blacklisted type BlacklistReason struct { Reason string FailCount int LastFailure time.Time AddedAt time.Time } // ErrInvalidPoolCandidate is returned when a pool address fails pre-validation // checks and should not be fetched from the RPC endpoint. var ErrInvalidPoolCandidate = errors.New("invalid pool candidate") // 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 { var ethClient *ethclient.Client if contractExecutor != nil { ethClient = contractExecutor.GetClient() } var slippageProtector *trading.SlippageProtection if ethClient != nil { slippageProtector = trading.NewSlippageProtection(ethClient, logger) } var profitCalculator *profitcalc.ProfitCalculator if ethClient != nil { profitCalculator = profitcalc.NewProfitCalculatorWithClient(logger, ethClient) } else { profitCalculator = profitcalc.NewProfitCalculator(logger) } create2Calculator := pools.NewCREATE2Calculator(logger, ethClient) if ethClient == nil { logger.Debug("CREATE2 calculator initialized without Ethereum client; live pool discovery limited") } marketDataLogger := marketdata.NewMarketDataLogger(logger, db) addressValidator := validation.NewAddressValidator() addressValidator.InitializeKnownContracts() 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: slippageProtector, circuitBreaker: circuit.NewCircuitBreaker(&circuit.Config{ Logger: logger, Name: "market_scanner", MaxFailures: 10, ResetTimeout: time.Minute * 5, MaxRequests: 3, SuccessThreshold: 2, }), contractExecutor: contractExecutor, create2Calculator: create2Calculator, database: db, profitCalculator: profitCalculator, opportunityRanker: profitcalc.NewOpportunityRanker(logger), marketDataLogger: marketDataLogger, addressValidator: addressValidator, poolBlacklist: make(map[common.Address]BlacklistReason), } // Initialize pool blacklist with known failing pools scanner.initializePoolBlacklist() // Initialize market data logger ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() if scanner.marketDataLogger != nil { if err := scanner.marketDataLogger.Initialize(ctx); err != nil { logger.Warn(fmt.Sprintf("Failed to initialize market data logger: %v", err)) } } else { logger.Warn("Market data logger disabled: database not configured") } // 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.LogSwapEvent(event) case events.AddLiquidity: w.scanner.LogLiquidityEvent(event, "add") case events.RemoveLiquidity: w.scanner.LogLiquidityEvent(event, "remove") case events.NewPool: w.scanner.logger.Info(fmt.Sprintf("Worker %d detected new pool: %s", w.ID, event.PoolAddress.Hex())) 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) // CRITICAL FIX: Populate token addresses if they're missing (zero addresses) // This fixes the issue where events are logged with 0x00000000 token addresses zeroAddr := common.Address{} if event.Token0 == zeroAddr || event.Token1 == zeroAddr { // Try to get token addresses from cache first s.cacheMutex.RLock() poolKey := event.PoolAddress.Hex() if cachedPool, exists := s.cache[poolKey]; exists { event.Token0 = cachedPool.Token0 event.Token1 = cachedPool.Token1 s.logger.Debug(fmt.Sprintf("✅ Enriched event with cached tokens: %s ↔ %s for pool %s", event.Token0.Hex()[:10], event.Token1.Hex()[:10], poolKey[:10])) } else { s.cacheMutex.RUnlock() // Cache miss - fetch pool data to get token addresses poolData, err := s.fetchPoolData(poolKey) if err == nil { event.Token0 = poolData.Token0 event.Token1 = poolData.Token1 s.logger.Debug(fmt.Sprintf("✅ Enriched event with fetched tokens: %s ↔ %s for pool %s", event.Token0.Hex()[:10], event.Token1.Hex()[:10], poolKey[:10])) } else { s.logger.Debug(fmt.Sprintf("⚠️ Could not fetch tokens for pool %s: %v", poolKey[:10], err)) } s.cacheMutex.RLock() } s.cacheMutex.RUnlock() } // Get an available worker job channel jobChannel := <-s.workerPool // Send the job to the worker jobChannel <- event } // 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 } // 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) if s.create2Calculator == nil { s.logger.Debug("CREATE2 calculator unavailable; skipping pool discovery") return poolAddresses } // 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: safeConvertInt64ToUint64(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 } // 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()) // Properly initialize all required fields now := time.Now() opportunity := stypes.ArbitrageOpportunity{ ID: fmt.Sprintf("arb_%d_%s", now.Unix(), path.Tokens[0].Hex()[:10]), Path: tokenPaths, Pools: []string{}, // Pool addresses will be discovered dynamically AmountIn: testAmount, Profit: profit, NetProfit: netProfit, GasEstimate: gasEstimate, GasCost: gasEstimate, // Set gas cost same as estimate EstimatedProfit: netProfit, RequiredAmount: testAmount, ROI: roi, Protocol: fmt.Sprintf("Triangular_%s", path.Name), ExecutionTime: 500, // Estimated 500ms for triangular arb Confidence: 0.5, // Medium confidence for triangular PriceImpact: 0.0, // Will be calculated dynamically MaxSlippage: 2.0, // 2% max slippage TokenIn: path.Tokens[0], TokenOut: path.Tokens[0], // Circular, starts and ends with same token Timestamp: now.Unix(), DetectedAt: now, ExpiresAt: now.Add(5 * time.Second), // 5 second expiry Urgency: 5, // Medium urgency Risk: 0.3, // Low-medium risk Profitable: netProfit.Sign() > 0, } 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) hopGasUnits := big.NewInt(150000) // ~150k gas units per swap totalGasCost.Add(totalGasCost, hopGasUnits) } // FIXED: Convert gas units to wei (gas units * gas price) // Use 0.1 gwei (100000000 wei) as conservative gas price estimate gasPrice := big.NewInt(100000000) // 0.1 gwei in wei totalGasCostWei := new(big.Int).Mul(totalGasCost, gasPrice) // Calculate profit (final amount - initial amount) profit := new(big.Int).Sub(currentAmount, initialAmount) return profit, totalGasCostWei, 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.SqrtPriceX96ToPriceCached(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 } // 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 *types.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 } // normalizeAndValidatePoolAddress returns a canonical representation of the // pool candidate and a validation result. It rejects obviously corrupted // addresses and known non-pool contracts before any RPC work is attempted. func (s *MarketScanner) normalizeAndValidatePoolAddress(candidate string) (string, *validation.AddressValidationResult, error) { if s.addressValidator == nil { trimmed := strings.TrimSpace(candidate) return trimmed, nil, nil } trimmed := strings.TrimSpace(candidate) if trimmed == "" { return "", nil, fmt.Errorf("%w: empty address", ErrInvalidPoolCandidate) } if !strings.HasPrefix(trimmed, "0x") && len(trimmed) == 40 { trimmed = "0x" + trimmed } // Lowercase for consistent cache keys and validation while preserving the // ability to distinguish later via validation result if needed. normalized := strings.ToLower(trimmed) result := s.addressValidator.ValidateAddress(normalized) if !result.IsValid || result.CorruptionScore >= 30 { return "", result, fmt.Errorf("%w: validation_failed", ErrInvalidPoolCandidate) } switch result.ContractType { case validation.ContractTypeERC20Token, validation.ContractTypeRouter, validation.ContractTypeFactory: return "", result, fmt.Errorf("%w: non_pool_contract", ErrInvalidPoolCandidate) } return normalized, result, nil } // getPoolData retrieves pool data with caching func (s *MarketScanner) getPoolData(poolAddress string) (*CachedData, error) { normalized, validationResult, err := s.normalizeAndValidatePoolAddress(poolAddress) if err != nil { if errors.Is(err, ErrInvalidPoolCandidate) { corruptionScore := -1 if validationResult != nil { corruptionScore = validationResult.CorruptionScore } contractType := "" if validationResult != nil { contractType = validationResult.ContractType.String() } s.logger.Debug("Pool candidate rejected before fetch", "address", poolAddress, "normalized", normalized, "error", err, "corruption_score", corruptionScore, "contract_type", contractType, ) } return nil, err } cacheKey := fmt.Sprintf("pool_%s", normalized) 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", normalized)) return data, nil } s.cacheMutex.RUnlock() // Use singleflight to prevent duplicate requests result, err, _ := s.cacheGroup.Do(cacheKey, func() (interface{}, error) { return s.fetchPoolData(normalized) }) 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", normalized)) return poolData, nil } // initializePoolBlacklist sets up the initial pool blacklist func (s *MarketScanner) initializePoolBlacklist() { // Known failing pools on Arbitrum that consistently revert on slot0() calls knownFailingPools := []struct { address common.Address reason string }{ { address: common.HexToAddress("0xB1026b8e7276e7AC75410F1fcbbe21796e8f7526"), reason: "slot0() consistently reverts - invalid pool contract", }, // Add more known failing pools here as discovered } s.blacklistMutex.Lock() defer s.blacklistMutex.Unlock() for _, pool := range knownFailingPools { s.poolBlacklist[pool.address] = BlacklistReason{ Reason: pool.reason, FailCount: 0, LastFailure: time.Time{}, AddedAt: time.Now(), } s.logger.Info(fmt.Sprintf("🚫 Blacklisted pool %s: %s", pool.address.Hex(), pool.reason)) } } // isPoolBlacklisted checks if a pool is in the blacklist func (s *MarketScanner) isPoolBlacklisted(poolAddr common.Address) (bool, string) { s.blacklistMutex.RLock() defer s.blacklistMutex.RUnlock() if reason, exists := s.poolBlacklist[poolAddr]; exists { return true, reason.Reason } return false, "" } // addToPoolBlacklist adds a pool to the blacklist after repeated failures func (s *MarketScanner) addToPoolBlacklist(poolAddr common.Address, reason string) { s.blacklistMutex.Lock() defer s.blacklistMutex.Unlock() if existing, exists := s.poolBlacklist[poolAddr]; exists { // Increment fail count existing.FailCount++ existing.LastFailure = time.Now() s.poolBlacklist[poolAddr] = existing s.logger.Warn(fmt.Sprintf("🚫 Pool %s blacklist updated (fail count: %d): %s", poolAddr.Hex(), existing.FailCount, reason)) } else { // New blacklist entry s.poolBlacklist[poolAddr] = BlacklistReason{ Reason: reason, FailCount: 1, LastFailure: time.Now(), AddedAt: time.Now(), } s.logger.Warn(fmt.Sprintf("🚫 Pool %s added to blacklist: %s", poolAddr.Hex(), reason)) } } // recordPoolFailure records a pool failure and blacklists after threshold func (s *MarketScanner) recordPoolFailure(poolAddr common.Address, errorMsg string) { const failureThreshold = 5 // Blacklist after 5 consecutive failures s.blacklistMutex.Lock() defer s.blacklistMutex.Unlock() if existing, exists := s.poolBlacklist[poolAddr]; exists { // Already blacklisted, just increment counter existing.FailCount++ existing.LastFailure = time.Now() s.poolBlacklist[poolAddr] = existing } else { // Check if we should blacklist this pool // Create temporary entry to track failures tempEntry := BlacklistReason{ Reason: errorMsg, FailCount: 1, LastFailure: time.Now(), AddedAt: time.Now(), } // If we've seen this pool fail before (would be in cache), increment // For now, blacklist after first failure of specific error types if strings.Contains(errorMsg, "execution reverted") || strings.Contains(errorMsg, "invalid pool contract") { s.poolBlacklist[poolAddr] = tempEntry s.logger.Warn(fmt.Sprintf("🚫 Pool %s blacklisted after critical error: %s", poolAddr.Hex(), errorMsg)) } } } // 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) // Check blacklist before attempting expensive RPC calls if blacklisted, reason := s.isPoolBlacklisted(address); blacklisted { s.logger.Debug(fmt.Sprintf("Skipping blacklisted pool %s: %s", poolAddress, reason)) return nil, fmt.Errorf("pool is blacklisted: %s", reason) } // In test environment, return mock data to avoid network calls if s.isTestEnvironment() { return s.getMockPoolData(poolAddress), nil } // Use shared RPC client from contract executor to respect rate limits // Creating new clients bypasses rate limiting and causes 429 errors var client *ethclient.Client if s.contractExecutor != nil { client = s.contractExecutor.GetClient() } if client == nil { // Fallback: create new client only if no shared client available rpcEndpoint := os.Getenv("ARBITRUM_RPC_ENDPOINT") if rpcEndpoint == "" { rpcEndpoint = "wss://arbitrum-mainnet.core.chainstack.com/53c30e7a941160679fdcc396c894fc57" } var err error 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)) // Record failure for potential blacklisting s.recordPoolFailure(address, err.Error()) 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(), } liquidityStr := "0" if poolState.Liquidity != nil { liquidityStr = poolState.Liquidity.String() } 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, liquidityStr)) 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") } // FIXED: Properly scale calculations to avoid overflow // Use big.Float for intermediate calculations to handle X96 scaling sqrtPriceFloat := new(big.Float).SetInt(sqrtPrice) liquidityFloat := new(big.Float).SetInt(liquidity) amountInFloat := new(big.Float).SetInt(amountIn) // Q96 constant = 2^96 Q96 := new(big.Float).SetInt(new(big.Int).Lsh(big.NewInt(1), 96)) // Calculate new sqrtPrice: newSqrtPrice = (L * sqrtPrice) / (L + amountIn) // Note: This is simplified - proper V3 calculation would account for tick ranges numerator := new(big.Float).Mul(liquidityFloat, sqrtPriceFloat) denominator := new(big.Float).Add(liquidityFloat, amountInFloat) newSqrtPriceFloat := new(big.Float).Quo(numerator, denominator) // Calculate output: amountOut = L * (sqrtPrice - newSqrtPrice) / Q96 priceDiffFloat := new(big.Float).Sub(sqrtPriceFloat, newSqrtPriceFloat) amountOutFloat := new(big.Float).Mul(liquidityFloat, priceDiffFloat) amountOutFloat.Quo(amountOutFloat, Q96) // Divide by 2^96 to un-scale // Convert back to big.Int amountOut := new(big.Int) amountOutFloat.Int(amountOut) // Sanity check: if amountOut is still massive or negative, return error if amountOut.BitLen() > 128 || amountOut.Sign() < 0 { return nil, fmt.Errorf("calculated amountOut is invalid: %s", amountOut.String()) } // 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) amountInStr := "0" if amountIn != nil { amountInStr = amountIn.String() } amountOutStr := "0" if amountOut != nil { amountOutStr = amountOut.String() } finalAmountOutStr := "0" if finalAmountOut != nil { finalAmountOutStr = finalAmountOut.String() } s.logger.Debug(fmt.Sprintf("V3 calculation: amountIn=%s, amountOut=%s, fee=%d, finalOut=%s", amountInStr, amountOutStr, fee, finalAmountOutStr)) 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"] } // GetMarketDataLogger returns the market data logger func (s *MarketScanner) GetMarketDataLogger() *marketdata.MarketDataLogger { return s.marketDataLogger } // GetProfitCalculator returns the profit calculator func (s *MarketScanner) GetProfitCalculator() *profitcalc.ProfitCalculator { return s.profitCalculator } // GetOpportunityRanker returns the opportunity ranker func (s *MarketScanner) GetOpportunityRanker() *profitcalc.OpportunityRanker { return s.opportunityRanker } // GetPoolData retrieves pool data with caching func (s *MarketScanner) GetPoolData(poolAddress string) (*CachedData, error) { return s.getPoolData(poolAddress) } // UpdatePoolData updates cached pool data from an event func (s *MarketScanner) UpdatePoolData(event events.Event) { s.updatePoolData(event) } // IsSignificantMovement determines if a price movement is significant enough to exploit func (s *MarketScanner) IsSignificantMovement(movement *PriceMovement, threshold float64) bool { return s.isSignificantMovement(movement, threshold) } // FindRelatedPools finds pools that trade the same token pair func (s *MarketScanner) FindRelatedPools(token0, token1 common.Address) []*CachedData { return s.findRelatedPools(token0, token1) } // 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 { return s.estimateProfit(event, pool, priceDiff) } // FindTriangularArbitrageOpportunities looks for triangular arbitrage opportunities func (s *MarketScanner) FindTriangularArbitrageOpportunities(event events.Event) []stypes.ArbitrageOpportunity { return s.findTriangularArbitrageOpportunities(event) } // ExecuteArbitrageOpportunity executes an arbitrage opportunity using the smart contract func (s *MarketScanner) ExecuteArbitrageOpportunity(opportunity stypes.ArbitrageOpportunity) { s.executeArbitrageOpportunity(opportunity) } // LogSwapEvent logs a swap event to the database func (s *MarketScanner) LogSwapEvent(event events.Event) { s.logSwapEvent(event) } // LogLiquidityEvent logs a liquidity event to the database func (s *MarketScanner) LogLiquidityEvent(event events.Event, eventType string) { s.logLiquidityEvent(event, eventType) } // LogPoolData logs pool data to the database func (s *MarketScanner) LogPoolData(poolData *CachedData) { s.logPoolData(poolData) } // GetFactoryForProtocol returns the factory address for a known DEX protocol func (s *MarketScanner) GetFactoryForProtocol(protocol string) common.Address { return s.getFactoryForProtocol(protocol) } // Config returns the scanner's configuration func (s *MarketScanner) Config() *config.BotConfig { return s.config }