package profitcalc import ( "context" "fmt" "math/big" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/ethclient" "github.com/fraktal/mev-beta/internal/logger" ) // PriceFeed provides real-time price data from multiple DEXs type PriceFeed struct { logger *logger.Logger client *ethclient.Client priceCache map[string]*PriceData priceMutex sync.RWMutex updateTicker *time.Ticker stopChan chan struct{} // DEX addresses for price queries uniswapV3Factory common.Address uniswapV2Factory common.Address sushiswapFactory common.Address camelotFactory common.Address traderJoeFactory common.Address } // PriceData represents price information from a DEX type PriceData struct { TokenA common.Address TokenB common.Address Price *big.Float // Token B per Token A InversePrice *big.Float // Token A per Token B Liquidity *big.Float // Total liquidity in pool DEX string // DEX name PoolAddress common.Address LastUpdated time.Time IsValid bool } // MultiDEXPriceData aggregates prices from multiple DEXs type MultiDEXPriceData struct { TokenA common.Address TokenB common.Address Prices []*PriceData BestBuyDEX *PriceData // Best DEX to buy Token A (lowest price) BestSellDEX *PriceData // Best DEX to sell Token A (highest price) PriceSpread *big.Float // Price difference between best buy/sell SpreadBps int64 // Spread in basis points LastUpdated time.Time } // NewPriceFeed creates a new price feed manager func NewPriceFeed(logger *logger.Logger, client *ethclient.Client) *PriceFeed { return &PriceFeed{ logger: logger, client: client, priceCache: make(map[string]*PriceData), stopChan: make(chan struct{}), // Arbitrum DEX factory addresses uniswapV3Factory: common.HexToAddress("0x1F98431c8aD98523631AE4a59f267346ea31F984"), uniswapV2Factory: common.HexToAddress("0xc35DADB65012eC5796536bD9864eD8773aBc74C4"), // SushiSwap on Arbitrum sushiswapFactory: common.HexToAddress("0xc35DADB65012eC5796536bD9864eD8773aBc74C4"), camelotFactory: common.HexToAddress("0x6EcCab422D763aC031210895C81787E87B82A80f"), traderJoeFactory: common.HexToAddress("0xaE4EC9901c3076D0DdBe76A520F9E90a6227aCB7"), } } // Start begins the price feed updates func (pf *PriceFeed) Start() { pf.updateTicker = time.NewTicker(15 * time.Second) // Update every 15 seconds go pf.priceUpdateLoop() pf.logger.Info("Price feed started with 15-second update interval") } // Stop halts the price feed updates func (pf *PriceFeed) Stop() { if pf.updateTicker != nil { pf.updateTicker.Stop() } close(pf.stopChan) pf.logger.Info("Price feed stopped") } // GetMultiDEXPrice gets aggregated price data from multiple DEXs func (pf *PriceFeed) GetMultiDEXPrice(tokenA, tokenB common.Address) *MultiDEXPriceData { pf.priceMutex.RLock() defer pf.priceMutex.RUnlock() var prices []*PriceData var bestBuy, bestSell *PriceData // Collect prices from all DEXs for _, price := range pf.priceCache { if (price.TokenA == tokenA && price.TokenB == tokenB) || (price.TokenA == tokenB && price.TokenB == tokenA) { if price.IsValid && time.Since(price.LastUpdated) < 5*time.Minute { prices = append(prices, price) // Find best buy price (lowest price to buy tokenA) if bestBuy == nil || price.Price.Cmp(bestBuy.Price) < 0 { bestBuy = price } // Find best sell price (highest price to sell tokenA) if bestSell == nil || price.Price.Cmp(bestSell.Price) > 0 { bestSell = price } } } } if len(prices) == 0 { return nil } // Calculate price spread var priceSpread *big.Float var spreadBps int64 if bestBuy != nil && bestSell != nil && bestBuy != bestSell { priceSpread = new(big.Float).Sub(bestSell.Price, bestBuy.Price) // Calculate spread in basis points spreadRatio := new(big.Float).Quo(priceSpread, bestBuy.Price) spreadFloat, _ := spreadRatio.Float64() spreadBps = int64(spreadFloat * 10000) // Convert to basis points } return &MultiDEXPriceData{ TokenA: tokenA, TokenB: tokenB, Prices: prices, BestBuyDEX: bestBuy, BestSellDEX: bestSell, PriceSpread: priceSpread, SpreadBps: spreadBps, LastUpdated: time.Now(), } } // GetBestArbitrageOpportunity finds the best arbitrage opportunity for a token pair func (pf *PriceFeed) GetBestArbitrageOpportunity(tokenA, tokenB common.Address, tradeAmount *big.Float) *ArbitrageRoute { multiPrice := pf.GetMultiDEXPrice(tokenA, tokenB) if multiPrice == nil || multiPrice.BestBuyDEX == nil || multiPrice.BestSellDEX == nil { return nil } // Skip if same DEX or insufficient spread if multiPrice.BestBuyDEX.DEX == multiPrice.BestSellDEX.DEX || multiPrice.SpreadBps < 50 { return nil } // Calculate potential profit buyPrice := multiPrice.BestBuyDEX.Price sellPrice := multiPrice.BestSellDEX.Price // Amount out when buying tokenA amountOut := new(big.Float).Quo(tradeAmount, buyPrice) // Revenue when selling tokenA revenue := new(big.Float).Mul(amountOut, sellPrice) // Gross profit grossProfit := new(big.Float).Sub(revenue, tradeAmount) // Validate that the profit calculation is reasonable grossProfitFloat, _ := grossProfit.Float64() tradeAmountFloat, _ := tradeAmount.Float64() if grossProfitFloat > tradeAmountFloat*100 { // If profit is more than 100x the trade amount, it's unrealistic pf.logger.Debug(fmt.Sprintf("Unrealistic arbitrage opportunity detected: tradeAmount=%s, grossProfit=%s", tradeAmount.String(), grossProfit.String())) return nil // Reject this opportunity as unrealistic } return &ArbitrageRoute{ TokenA: tokenA, TokenB: tokenB, BuyDEX: multiPrice.BestBuyDEX.DEX, SellDEX: multiPrice.BestSellDEX.DEX, BuyPrice: buyPrice, SellPrice: sellPrice, TradeAmount: tradeAmount, AmountOut: amountOut, GrossProfit: grossProfit, SpreadBps: multiPrice.SpreadBps, Timestamp: time.Now(), } } // ArbitrageRoute represents a complete arbitrage route type ArbitrageRoute struct { TokenA common.Address TokenB common.Address BuyDEX string SellDEX string BuyPrice *big.Float SellPrice *big.Float TradeAmount *big.Float AmountOut *big.Float GrossProfit *big.Float SpreadBps int64 Timestamp time.Time } // priceUpdateLoop runs the background price update process func (pf *PriceFeed) priceUpdateLoop() { defer pf.updateTicker.Stop() // Major trading pairs on Arbitrum tradingPairs := []TokenPair{ { TokenA: common.HexToAddress("0x82af49447d8a07e3bd95bd0d56f35241523fbab1"), // WETH TokenB: common.HexToAddress("0xaf88d065e77c8cc2239327c5edb3a432268e5831"), // USDC }, { TokenA: common.HexToAddress("0x82af49447d8a07e3bd95bd0d56f35241523fbab1"), // WETH TokenB: common.HexToAddress("0x912ce59144191c1204e64559fe8253a0e49e6548"), // ARB }, { TokenA: common.HexToAddress("0xaf88d065e77c8cc2239327c5edb3a432268e5831"), // USDC TokenB: common.HexToAddress("0xfd086bc7cd5c481dcc9c85ebe478a1c0b69fcbb9"), // USDT }, { TokenA: common.HexToAddress("0x82af49447d8a07e3bd95bd0d56f35241523fbab1"), // WETH TokenB: common.HexToAddress("0x2f2a2543b76a4166549f7aab2e75bef0aefc5b0f"), // WBTC }, } for { select { case <-pf.stopChan: return case <-pf.updateTicker.C: pf.updatePricesForPairs(tradingPairs) } } } // TokenPair represents a trading pair type TokenPair struct { TokenA common.Address TokenB common.Address } // updatePricesForPairs updates prices for specified trading pairs func (pf *PriceFeed) updatePricesForPairs(pairs []TokenPair) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() for _, pair := range pairs { // Update prices from multiple DEXs go pf.updatePriceFromDEX(ctx, pair.TokenA, pair.TokenB, "UniswapV3", pf.uniswapV3Factory) go pf.updatePriceFromDEX(ctx, pair.TokenA, pair.TokenB, "SushiSwap", pf.sushiswapFactory) go pf.updatePriceFromDEX(ctx, pair.TokenA, pair.TokenB, "Camelot", pf.camelotFactory) go pf.updatePriceFromDEX(ctx, pair.TokenA, pair.TokenB, "TraderJoe", pf.traderJoeFactory) } } // updatePriceFromDEX updates price data from a specific DEX func (pf *PriceFeed) updatePriceFromDEX(ctx context.Context, tokenA, tokenB common.Address, dexName string, factory common.Address) { // This is a simplified implementation // In a real implementation, you would: // 1. Query the factory for the pool address // 2. Call the pool contract to get reserves/prices // 3. Calculate the current price // For now, simulate price updates with mock data pf.priceMutex.Lock() defer pf.priceMutex.Unlock() key := fmt.Sprintf("%s_%s_%s", tokenA.Hex(), tokenB.Hex(), dexName) // Mock price data (in a real implementation, fetch from contracts) mockPrice := big.NewFloat(2000.0) // 1 ETH = 2000 USDC example if dexName == "SushiSwap" { mockPrice = big.NewFloat(2001.0) // Slightly different price } else if dexName == "Camelot" { mockPrice = big.NewFloat(1999.5) } // Validate that the price is reasonable (not extremely high or low) if mockPrice.Cmp(big.NewFloat(0.000001)) < 0 || mockPrice.Cmp(big.NewFloat(10000000)) > 0 { pf.logger.Debug(fmt.Sprintf("Invalid price detected for %s: %s, marking as invalid", dexName, mockPrice.String())) mockPrice = big.NewFloat(1000.0) // Default to reasonable price } pf.priceCache[key] = &PriceData{ TokenA: tokenA, TokenB: tokenB, Price: mockPrice, InversePrice: new(big.Float).Quo(big.NewFloat(1), mockPrice), Liquidity: big.NewFloat(1000000), // Mock liquidity DEX: dexName, PoolAddress: common.HexToAddress("0x1234567890123456789012345678901234567890"), // Mock address LastUpdated: time.Now(), IsValid: true, } pf.logger.Debug(fmt.Sprintf("Updated %s price for %s/%s: %s", dexName, tokenA.Hex()[:8], tokenB.Hex()[:8], mockPrice.String())) } // GetPriceStats returns statistics about tracked prices func (pf *PriceFeed) GetPriceStats() map[string]interface{} { pf.priceMutex.RLock() defer pf.priceMutex.RUnlock() totalPrices := len(pf.priceCache) validPrices := 0 stalePrices := 0 dexCounts := make(map[string]int) now := time.Now() for _, price := range pf.priceCache { if price.IsValid { validPrices++ } if now.Sub(price.LastUpdated) > 5*time.Minute { stalePrices++ } dexCounts[price.DEX]++ } return map[string]interface{}{ "totalPrices": totalPrices, "validPrices": validPrices, "stalePrices": stalePrices, "dexBreakdown": dexCounts, "lastUpdated": time.Now(), } }