package discovery import ( "context" "fmt" "math/big" "os" "sync" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/ethclient" "gopkg.in/yaml.v3" "github.com/fraktal/mev-beta/internal/logger" exchangeMath "github.com/fraktal/mev-beta/pkg/math" ) // MarketDiscovery manages pool discovery and market building type MarketDiscovery struct { client *ethclient.Client logger *logger.Logger config *MarketConfig mathCalc *exchangeMath.MathCalculator // Market state pools map[common.Address]*PoolInfoDetailed tokens map[common.Address]*TokenInfo factories map[common.Address]*FactoryInfo routers map[common.Address]*RouterInfo mu sync.RWMutex // Performance tracking poolsDiscovered uint64 arbitrageOpps uint64 lastScanTime time.Time totalScanTime time.Duration } // MarketConfig represents the configuration for market discovery type MarketConfig struct { Version string `yaml:"version"` Network string `yaml:"network"` ChainID int64 `yaml:"chain_id"` Tokens map[string]*TokenConfigInfo `yaml:"tokens"` Factories map[string]*FactoryConfig `yaml:"factories"` Routers map[string]*RouterConfig `yaml:"routers"` PriorityPools []PriorityPoolConfig `yaml:"priority_pools"` MarketScan MarketScanConfig `yaml:"market_scan"` Arbitrage ArbitrageConfig `yaml:"arbitrage"` Logging LoggingConfig `yaml:"logging"` Risk RiskConfig `yaml:"risk"` Monitoring MonitoringConfig `yaml:"monitoring"` } type TokenConfigInfo struct { Address string `yaml:"address"` Symbol string `yaml:"symbol"` Decimals int `yaml:"decimals"` Priority int `yaml:"priority"` } type FactoryConfig struct { Address string `yaml:"address"` Type string `yaml:"type"` InitCodeHash string `yaml:"init_code_hash"` FeeTiers []uint32 `yaml:"fee_tiers"` Priority int `yaml:"priority"` } type RouterConfig struct { Address string `yaml:"address"` Factory string `yaml:"factory"` Type string `yaml:"type"` Priority int `yaml:"priority"` } type PriorityPoolConfig struct { Pool string `yaml:"pool"` Factory string `yaml:"factory"` Token0 string `yaml:"token0"` Token1 string `yaml:"token1"` Fee uint32 `yaml:"fee"` Priority int `yaml:"priority"` } type MarketScanConfig struct { ScanInterval int `yaml:"scan_interval"` MaxPools int `yaml:"max_pools"` MinLiquidityUSD float64 `yaml:"min_liquidity_usd"` MinVolume24hUSD float64 `yaml:"min_volume_24h_usd"` Discovery PoolDiscoveryConfig `yaml:"discovery"` } type PoolDiscoveryConfig struct { MaxBlocksBack uint64 `yaml:"max_blocks_back"` MinPoolAge uint64 `yaml:"min_pool_age"` DiscoveryInterval uint64 `yaml:"discovery_interval"` } type ArbitrageConfig struct { MinProfitUSD float64 `yaml:"min_profit_usd"` MaxSlippage float64 `yaml:"max_slippage"` MaxGasPrice float64 `yaml:"max_gas_price"` ProfitMargins map[string]float64 `yaml:"profit_margins"` } type LoggingConfig struct { Level string `yaml:"level"` Files map[string]string `yaml:"files"` RealTime map[string]interface{} `yaml:"real_time"` } type RiskConfig struct { MaxPositionETH float64 `yaml:"max_position_eth"` MaxDailyLossETH float64 `yaml:"max_daily_loss_eth"` MaxConcurrentTxs int `yaml:"max_concurrent_txs"` CircuitBreaker map[string]interface{} `yaml:"circuit_breaker"` } type MonitoringConfig struct { Enabled bool `yaml:"enabled"` UpdateInterval int `yaml:"update_interval"` Metrics []string `yaml:"metrics"` } // PoolInfoDetailed represents detailed pool information for market discovery type PoolInfoDetailed struct { Address common.Address `json:"address"` Factory common.Address `json:"factory"` FactoryType string `json:"factory_type"` Token0 common.Address `json:"token0"` Token1 common.Address `json:"token1"` Fee uint32 `json:"fee"` Reserve0 *big.Int `json:"reserve0"` Reserve1 *big.Int `json:"reserve1"` Liquidity *big.Int `json:"liquidity"` SqrtPriceX96 *big.Int `json:"sqrt_price_x96,omitempty"` // For V3 pools Tick int32 `json:"tick,omitempty"` // For V3 pools LastUpdated time.Time `json:"last_updated"` Volume24h *big.Int `json:"volume_24h"` Priority int `json:"priority"` Active bool `json:"active"` } type TokenInfo struct { Address common.Address `json:"address"` Symbol string `json:"symbol"` Name string `json:"name"` Decimals uint8 `json:"decimals"` Priority int `json:"priority"` LastPrice *big.Int `json:"last_price"` Volume24h *big.Int `json:"volume_24h"` } type FactoryInfo struct { Address common.Address `json:"address"` Type string `json:"type"` InitCodeHash common.Hash `json:"init_code_hash"` FeeTiers []uint32 `json:"fee_tiers"` PoolCount uint64 `json:"pool_count"` Priority int `json:"priority"` } type RouterInfo struct { Address common.Address `json:"address"` Factory common.Address `json:"factory"` Type string `json:"type"` Priority int `json:"priority"` } // MarketScanResult represents the result of a market scan type MarketScanResult struct { Timestamp time.Time `json:"timestamp"` BlockNumber uint64 `json:"block_number"` PoolsScanned int `json:"pools_scanned"` NewPoolsFound int `json:"new_pools_found"` ArbitrageOpps []*ArbitrageOpportunityDetailed `json:"arbitrage_opportunities"` TopPools []*PoolInfoDetailed `json:"top_pools"` ScanDuration time.Duration `json:"scan_duration"` GasPrice *big.Int `json:"gas_price"` NetworkConditions map[string]interface{} `json:"network_conditions"` } type ArbitrageOpportunityDetailed struct { ID string `json:"id"` Type string `json:"type"` TokenIn common.Address `json:"token_in"` TokenOut common.Address `json:"token_out"` AmountIn *big.Int `json:"amount_in"` ExpectedAmountOut *big.Int `json:"expected_amount_out"` ActualAmountOut *big.Int `json:"actual_amount_out"` Profit *big.Int `json:"profit"` ProfitUSD float64 `json:"profit_usd"` ProfitMargin float64 `json:"profit_margin"` GasCost *big.Int `json:"gas_cost"` NetProfit *big.Int `json:"net_profit"` ExchangeA string `json:"exchange_a"` ExchangeB string `json:"exchange_b"` PoolA common.Address `json:"pool_a"` PoolB common.Address `json:"pool_b"` PriceA float64 `json:"price_a"` PriceB float64 `json:"price_b"` PriceImpactA float64 `json:"price_impact_a"` PriceImpactB float64 `json:"price_impact_b"` CapitalRequired float64 `json:"capital_required"` GasCostUSD float64 `json:"gas_cost_usd"` Confidence float64 `json:"confidence"` RiskScore float64 `json:"risk_score"` ExecutionTime time.Duration `json:"execution_time"` Timestamp time.Time `json:"timestamp"` } // PoolDiscoveryResult represents pool discovery results type PoolDiscoveryResult struct { Timestamp time.Time `json:"timestamp"` FromBlock uint64 `json:"from_block"` ToBlock uint64 `json:"to_block"` NewPools []*PoolInfoDetailed `json:"new_pools"` PoolsFound int `json:"pools_found"` ScanDuration time.Duration `json:"scan_duration"` } // NewMarketDiscovery creates a new market discovery instance func NewMarketDiscovery(client *ethclient.Client, logger *logger.Logger, configPath string) (*MarketDiscovery, error) { // Load configuration config, err := LoadMarketConfig(configPath) if err != nil { return nil, fmt.Errorf("failed to load config: %w", err) } // Initialize math calculator mathCalc := exchangeMath.NewMathCalculator() md := &MarketDiscovery{ client: client, logger: logger, config: config, mathCalc: mathCalc, pools: make(map[common.Address]*PoolInfoDetailed), tokens: make(map[common.Address]*TokenInfo), factories: make(map[common.Address]*FactoryInfo), routers: make(map[common.Address]*RouterInfo), } // Load initial configuration if err := md.loadInitialMarkets(); err != nil { return nil, fmt.Errorf("failed to load initial markets: %w", err) } logger.Info("Market discovery initialized with comprehensive pool detection") return md, nil } // LoadMarketConfig loads market configuration from YAML file func LoadMarketConfig(configPath string) (*MarketConfig, error) { data, err := os.ReadFile(configPath) if err != nil { return nil, fmt.Errorf("failed to read config file: %w", err) } var config MarketConfig if err := yaml.Unmarshal(data, &config); err != nil { return nil, fmt.Errorf("failed to parse config: %w", err) } return &config, nil } // loadInitialMarkets loads initial tokens, factories, and priority pools func (md *MarketDiscovery) loadInitialMarkets() error { md.mu.Lock() defer md.mu.Unlock() // Load tokens for _, token := range md.config.Tokens { tokenAddr := common.HexToAddress(token.Address) md.tokens[tokenAddr] = &TokenInfo{ Address: tokenAddr, Symbol: token.Symbol, Decimals: uint8(token.Decimals), Priority: token.Priority, } } // Load factories for _, factory := range md.config.Factories { factoryAddr := common.HexToAddress(factory.Address) md.factories[factoryAddr] = &FactoryInfo{ Address: factoryAddr, Type: factory.Type, InitCodeHash: common.HexToHash(factory.InitCodeHash), FeeTiers: factory.FeeTiers, Priority: factory.Priority, } } // Load routers for _, router := range md.config.Routers { routerAddr := common.HexToAddress(router.Address) factoryAddr := common.Address{} if router.Factory != "" { for _, f := range md.config.Factories { if f.Type == router.Factory { factoryAddr = common.HexToAddress(f.Address) break } } } md.routers[routerAddr] = &RouterInfo{ Address: routerAddr, Factory: factoryAddr, Type: router.Type, Priority: router.Priority, } } // Load priority pools for _, poolConfig := range md.config.PriorityPools { poolAddr := common.HexToAddress(poolConfig.Pool) token0 := common.HexToAddress(poolConfig.Token0) token1 := common.HexToAddress(poolConfig.Token1) // Find factory var factoryAddr common.Address var factoryType string for _, f := range md.config.Factories { if f.Type == poolConfig.Factory { factoryAddr = common.HexToAddress(f.Address) factoryType = f.Type break } } pool := &PoolInfoDetailed{ Address: poolAddr, Factory: factoryAddr, FactoryType: factoryType, Token0: token0, Token1: token1, Fee: poolConfig.Fee, Priority: poolConfig.Priority, Active: true, LastUpdated: time.Now(), } md.pools[poolAddr] = pool } md.logger.Info(fmt.Sprintf("Loaded initial markets: %d tokens, %d factories, %d routers, %d priority pools", len(md.tokens), len(md.factories), len(md.routers), len(md.pools))) return nil } // buildComprehensiveMarkets builds markets for all exchanges and top token pairs func (md *MarketDiscovery) buildComprehensiveMarkets() error { md.logger.Info("🏗️ Building comprehensive markets for all exchanges and top tokens") // Get top tokens (sorted by priority) topTokens := md.getTopTokens(10) // Reduced from 20 to 10 tokens to reduce load md.logger.Info(fmt.Sprintf("💼 Found %d top tokens for market building", len(topTokens))) // Build markets for each factory marketsBuilt := 0 for factoryAddr, factoryInfo := range md.factories { markets, err := md.buildFactoryMarkets(factoryAddr, factoryInfo, topTokens) if err != nil { md.logger.Error(fmt.Sprintf("Failed to build markets for factory %s: %v", factoryAddr.Hex(), err)) continue } marketsBuilt += len(markets) md.logger.Info(fmt.Sprintf("✅ Built %d markets for %s factory", len(markets), factoryInfo.Type)) } md.logger.Info(fmt.Sprintf("📊 Total markets built: %d", marketsBuilt)) // Log available markets md.logAvailableMarkets() return nil } // getTopTokens returns the top N tokens sorted by priority func (md *MarketDiscovery) getTopTokens(limit int) []*TokenInfo { md.mu.RLock() defer md.mu.RUnlock() // Convert map to slice tokens := make([]*TokenInfo, 0, len(md.tokens)) for _, token := range md.tokens { tokens = append(tokens, token) } // Sort by priority (highest first) for i := 0; i < len(tokens)-1; i++ { for j := i + 1; j < len(tokens); j++ { if tokens[i].Priority < tokens[j].Priority { tokens[i], tokens[j] = tokens[j], tokens[i] } } } // Limit to top N (reduced for performance) limit = 10 // Reduced from 20 to 10 to reduce load if len(tokens) > limit { tokens = tokens[:limit] } return tokens } // buildFactoryMarkets builds markets for a specific factory and token pairs func (md *MarketDiscovery) buildFactoryMarkets(factoryAddr common.Address, factoryInfo *FactoryInfo, tokens []*TokenInfo) ([]*PoolInfoDetailed, error) { var markets []*PoolInfoDetailed // Find WETH token (most important for pairing) var wethToken *TokenInfo for _, token := range tokens { if token.Symbol == "WETH" { wethToken = token break } } // If no WETH found, use the highest priority token if wethToken == nil && len(tokens) > 0 { wethToken = tokens[0] } // Build markets for each token pair for i, tokenA := range tokens { for j := i + 1; j < len(tokens); j++ { tokenB := tokens[j] // Build markets for this token pair pairMarkets, err := md.buildTokenPairMarkets(factoryAddr, factoryInfo, tokenA, tokenB) if err != nil { md.logger.Debug(fmt.Sprintf("Failed to build markets for %s-%s pair: %v", tokenA.Symbol, tokenB.Symbol, err)) continue } markets = append(markets, pairMarkets...) } // Also build markets for token-WETH pairs if WETH exists and is not this token if wethToken != nil && tokenA.Address != wethToken.Address { wethMarkets, err := md.buildTokenPairMarkets(factoryAddr, factoryInfo, tokenA, wethToken) if err != nil { md.logger.Debug(fmt.Sprintf("Failed to build markets for %s-WETH pair: %v", tokenA.Symbol, err)) continue } markets = append(markets, wethMarkets...) } } // Add built markets to tracking md.mu.Lock() for _, market := range markets { // Only add if not already tracking if _, exists := md.pools[market.Address]; !exists { md.pools[market.Address] = market } } md.mu.Unlock() return markets, nil } // buildTokenPairMarkets builds markets for a specific token pair and factory func (md *MarketDiscovery) buildTokenPairMarkets(factoryAddr common.Address, factoryInfo *FactoryInfo, tokenA, tokenB *TokenInfo) ([]*PoolInfoDetailed, error) { var markets []*PoolInfoDetailed // For factories with fee tiers (Uniswap V3 style), build markets for each fee tier if len(factoryInfo.FeeTiers) > 0 { // Build markets for each fee tier for _, feeTier := range factoryInfo.FeeTiers { // Generate deterministic pool address using CREATE2 poolAddr, err := md.calculatePoolAddress(factoryAddr, factoryInfo, tokenA, tokenB, feeTier) if err != nil { continue } market := &PoolInfoDetailed{ Address: poolAddr, Factory: factoryAddr, FactoryType: factoryInfo.Type, Token0: tokenA.Address, Token1: tokenB.Address, Fee: feeTier, Reserve0: big.NewInt(0), Reserve1: big.NewInt(0), Liquidity: big.NewInt(0), SqrtPriceX96: big.NewInt(0), Tick: 0, LastUpdated: time.Now(), Volume24h: big.NewInt(0), Priority: (tokenA.Priority + tokenB.Priority) / 2, Active: true, } markets = append(markets, market) } } else { // For factories without fee tiers (Uniswap V2 style), build a single market // Generate deterministic pool address using CREATE2 poolAddr, err := md.calculatePoolAddress(factoryAddr, factoryInfo, tokenA, tokenB, 0) if err != nil { return nil, err } market := &PoolInfoDetailed{ Address: poolAddr, Factory: factoryAddr, FactoryType: factoryInfo.Type, Token0: tokenA.Address, Token1: tokenB.Address, Reserve0: big.NewInt(0), Reserve1: big.NewInt(0), Liquidity: big.NewInt(0), LastUpdated: time.Now(), Volume24h: big.NewInt(0), Priority: (tokenA.Priority + tokenB.Priority) / 2, Active: true, } markets = append(markets, market) } return markets, nil } // calculatePoolAddress calculates the deterministic pool address using CREATE2 func (md *MarketDiscovery) calculatePoolAddress(factoryAddr common.Address, factoryInfo *FactoryInfo, tokenA, tokenB *TokenInfo, feeTier uint32) (common.Address, error) { // Sort tokens to ensure consistent ordering token0, token1 := tokenA.Address, tokenB.Address if token0.Big().Cmp(token1.Big()) > 0 { token0, token1 = token1, token0 } switch factoryInfo.Type { case "uniswap_v3", "camelot_v3", "algebra": // For Uniswap V3 style factories with fee tiers return md.calculateUniswapV3PoolAddress(factoryAddr, factoryInfo, token0, token1, feeTier) case "uniswap_v2", "sushiswap": // For Uniswap V2 style factories return md.calculateUniswapV2PoolAddress(factoryAddr, factoryInfo, token0, token1) case "balancer_v2": // For Balancer (simplified - in practice would need more info) return md.calculateBalancerPoolAddress(factoryAddr, token0, token1) case "curve": // For Curve (simplified - in practice would need more info) return md.calculateCurvePoolAddress(factoryAddr, token0, token1) default: // Generic CREATE2 calculation return md.calculateGenericPoolAddress(factoryAddr, factoryInfo, token0, token1, feeTier) } } // calculateUniswapV3PoolAddress calculates pool address for Uniswap V3 style factories func (md *MarketDiscovery) calculateUniswapV3PoolAddress(factoryAddr common.Address, factoryInfo *FactoryInfo, token0, token1 common.Address, feeTier uint32) (common.Address, error) { // Encode the pool key: keccak256(abi.encode(token0, token1, fee)) poolKey := crypto.Keccak256(append(append(token0.Bytes(), token1.Bytes()...), big.NewInt(int64(feeTier)).Bytes()...)) // Calculate CREATE2 address // keccak256(0xff ++ address ++ salt ++ keccak256(init_code))[12:] salt := poolKey initCodeHash := factoryInfo.InitCodeHash.Bytes() create2Input := append([]byte{0xff}, factoryAddr.Bytes()...) create2Input = append(create2Input, salt...) create2Input = append(create2Input, initCodeHash...) poolAddrBytes := crypto.Keccak256(create2Input) // Take last 20 bytes for address poolAddr := common.BytesToAddress(poolAddrBytes[12:]) return poolAddr, nil } // calculateUniswapV2PoolAddress calculates pool address for Uniswap V2 style factories func (md *MarketDiscovery) calculateUniswapV2PoolAddress(factoryAddr common.Address, factoryInfo *FactoryInfo, token0, token1 common.Address) (common.Address, error) { // For Uniswap V2: keccak256(0xff ++ address ++ keccak256(token0 ++ token1) ++ initcode_hash)[12:] poolKey := crypto.Keccak256(append(token0.Bytes(), token1.Bytes()...)) create2Input := append([]byte{0xff}, factoryAddr.Bytes()...) create2Input = append(create2Input, poolKey...) create2Input = append(create2Input, factoryInfo.InitCodeHash.Bytes()...) poolAddrBytes := crypto.Keccak256(create2Input) // Take last 20 bytes for address poolAddr := common.BytesToAddress(poolAddrBytes[12:]) return poolAddr, nil } // calculateBalancerPoolAddress calculates pool address for Balancer pools (simplified) func (md *MarketDiscovery) calculateBalancerPoolAddress(factoryAddr, token0, token1 common.Address) (common.Address, error) { // Simplified implementation - in practice would need more complex logic // For Balancer V2, pool addresses are typically determined by the vault // This is a placeholder implementation placeholder := crypto.Keccak256(append(append(factoryAddr.Bytes(), token0.Bytes()...), token1.Bytes()...)) return common.BytesToAddress(placeholder[12:]), nil } // calculateCurvePoolAddress calculates pool address for Curve pools (simplified) func (md *MarketDiscovery) calculateCurvePoolAddress(factoryAddr, token0, token1 common.Address) (common.Address, error) { // Simplified implementation - Curve pools are typically deployed via factories // with more complex logic. This is a placeholder implementation placeholder := crypto.Keccak256(append(append(factoryAddr.Bytes(), token0.Bytes()...), token1.Bytes()...)) return common.BytesToAddress(placeholder[12:]), nil } // calculateGenericPoolAddress calculates pool address for generic factories func (md *MarketDiscovery) calculateGenericPoolAddress(factoryAddr common.Address, factoryInfo *FactoryInfo, token0, token1 common.Address, feeTier uint32) (common.Address, error) { // Generic CREATE2 calculation using tokens and fee as salt saltInput := append(append(token0.Bytes(), token1.Bytes()...), big.NewInt(int64(feeTier)).Bytes()...) salt := crypto.Keccak256(saltInput) create2Input := append([]byte{0xff}, factoryAddr.Bytes()...) create2Input = append(create2Input, salt...) create2Input = append(create2Input, factoryInfo.InitCodeHash.Bytes()...) poolAddrBytes := crypto.Keccak256(create2Input) // Take last 20 bytes for address poolAddr := common.BytesToAddress(poolAddrBytes[12:]) return poolAddr, nil } // logAvailableMarkets logs all available markets grouped by exchange func (md *MarketDiscovery) logAvailableMarkets() { md.mu.RLock() defer md.mu.RUnlock() // Group markets by factory type marketsByFactory := make(map[string][]*PoolInfoDetailed) for _, pool := range md.pools { factoryType := pool.FactoryType marketsByFactory[factoryType] = append(marketsByFactory[factoryType], pool) } // Log markets for each factory md.logger.Info("📈 Available Markets by Exchange:") for factoryType, pools := range marketsByFactory { // Count unique token pairs tokenPairs := make(map[string]bool) for _, pool := range pools { // Handle empty addresses to prevent slice bounds panic token0Display := "unknown" token1Display := "unknown" if len(pool.Token0.Hex()) > 0 { if len(pool.Token0.Hex()) > 6 { token0Display = pool.Token0.Hex()[:6] } else { token0Display = pool.Token0.Hex() } } if len(pool.Token1.Hex()) > 0 { if len(pool.Token1.Hex()) > 6 { token1Display = pool.Token1.Hex()[:6] } else { token1Display = pool.Token1.Hex() } } pairKey := fmt.Sprintf("%s-%s", token0Display, token1Display) tokenPairs[pairKey] = true } md.logger.Info(fmt.Sprintf(" %s: %d pools, %d unique token pairs", factoryType, len(pools), len(tokenPairs))) // Log top 5 pools by priority for i, pool := range pools { if i >= 5 { break } md.logger.Debug(fmt.Sprintf(" 🏦 Pool %s (%s-%s, Fee: %d)", pool.Address.Hex()[:10], pool.Token0.Hex()[:6], pool.Token1.Hex()[:6], pool.Fee)) } if len(pools) > 5 { md.logger.Debug(fmt.Sprintf(" ... and %d more pools", len(pools)-5)) } } } // DiscoverPools discovers pools from factories within a block range func (md *MarketDiscovery) DiscoverPools(ctx context.Context, fromBlock, toBlock uint64) (*PoolDiscoveryResult, error) { startTime := time.Now() discovered := &PoolDiscoveryResult{ Timestamp: startTime, FromBlock: fromBlock, ToBlock: toBlock, NewPools: make([]*PoolInfoDetailed, 0), } // Discover pools from each factory for factoryAddr, factoryInfo := range md.factories { pools, err := md.discoverPoolsFromFactory(ctx, factoryAddr, factoryInfo, fromBlock, toBlock) if err != nil { md.logger.Error(fmt.Sprintf("Failed to discover pools from factory %s: %v", factoryAddr.Hex(), err)) continue } discovered.NewPools = append(discovered.NewPools, pools...) } discovered.PoolsFound = len(discovered.NewPools) discovered.ScanDuration = time.Since(startTime) md.poolsDiscovered += uint64(discovered.PoolsFound) return discovered, nil } // ScanForArbitrage scans all pools for arbitrage opportunities func (md *MarketDiscovery) ScanForArbitrage(ctx context.Context, blockNumber uint64) (*MarketScanResult, error) { startTime := time.Now() md.lastScanTime = startTime result := &MarketScanResult{ Timestamp: startTime, BlockNumber: blockNumber, ArbitrageOpps: make([]*ArbitrageOpportunityDetailed, 0), TopPools: make([]*PoolInfoDetailed, 0), NetworkConditions: make(map[string]interface{}), } // Update pool states if err := md.updatePoolStates(ctx); err != nil { return nil, fmt.Errorf("failed to update pool states: %w", err) } // Get current gas price gasPrice, err := md.client.SuggestGasPrice(ctx) if err != nil { gasPrice = big.NewInt(5000000000) // 5 gwei fallback } result.GasPrice = gasPrice // Scan for arbitrage opportunities opportunities := md.findArbitrageOpportunities(ctx, gasPrice) result.ArbitrageOpps = opportunities result.PoolsScanned = len(md.pools) // Get top pools by liquidity result.TopPools = md.getTopPoolsByLiquidity(10) result.ScanDuration = time.Since(startTime) md.totalScanTime += result.ScanDuration md.arbitrageOpps += uint64(len(opportunities)) return result, nil } // GetStatistics returns market discovery statistics func (md *MarketDiscovery) GetStatistics() map[string]interface{} { md.mu.RLock() defer md.mu.RUnlock() return map[string]interface{}{ "pools_tracked": len(md.pools), "tokens_tracked": len(md.tokens), "factories_tracked": len(md.factories), "pools_discovered": md.poolsDiscovered, "arbitrage_opportunities": md.arbitrageOpps, "last_scan_time": md.lastScanTime, "total_scan_time": md.totalScanTime.String(), } } // BuildComprehensiveMarkets builds comprehensive markets for all exchanges and top tokens // This should be called after initialization is complete to avoid deadlocks func (md *MarketDiscovery) BuildComprehensiveMarkets() error { return md.buildComprehensiveMarkets() } // getTopPoolsByLiquidity returns top pools sorted by liquidity func (md *MarketDiscovery) getTopPoolsByLiquidity(limit int) []*PoolInfoDetailed { md.mu.RLock() defer md.mu.RUnlock() pools := make([]*PoolInfoDetailed, 0, len(md.pools)) for _, pool := range md.pools { if pool.Active && pool.Liquidity != nil { pools = append(pools, pool) } } // Sort by liquidity (highest first) for i := 0; i < len(pools)-1; i++ { for j := i + 1; j < len(pools); j++ { if pools[i].Liquidity.Cmp(pools[j].Liquidity) < 0 { pools[i], pools[j] = pools[j], pools[i] } } } if len(pools) > limit { pools = pools[:limit] } return pools } // findArbitrageOpportunities finds arbitrage opportunities across all pools func (md *MarketDiscovery) findArbitrageOpportunities(ctx context.Context, gasPrice *big.Int) []*ArbitrageOpportunityDetailed { // Create arbitrage calculator calculator := NewArbitrageCalculator(md.logger, &md.config.Arbitrage, md.mathCalc) md.mu.RLock() pools := make(map[common.Address]*PoolInfoDetailed, len(md.pools)) for addr, pool := range md.pools { pools[addr] = pool } md.mu.RUnlock() return calculator.findArbitrageOpportunities(ctx, gasPrice, pools, md.logger, &md.config.Arbitrage, md.mathCalc) } // updatePoolStates updates the state of all tracked pools func (md *MarketDiscovery) updatePoolStates(ctx context.Context) error { // Create pool state manager manager := NewPoolStateManager(md.client, md.logger) md.mu.Lock() pools := make(map[common.Address]*PoolInfoDetailed, len(md.pools)) for addr, pool := range md.pools { pools[addr] = pool } md.mu.Unlock() return manager.updatePoolStates(ctx, pools, &md.mu, md.logger) } // discoverPoolsFromFactory discovers pools from a specific factory func (md *MarketDiscovery) discoverPoolsFromFactory(ctx context.Context, factoryAddr common.Address, factoryInfo *FactoryInfo, fromBlock, toBlock uint64) ([]*PoolInfoDetailed, error) { // Implementation would query factory events for pool creation return []*PoolInfoDetailed{}, nil }