Arbitrage Detection

// SpotArbitrageCalculator compares prices across exchanges
type SpotArbitrageCalculator struct{}

func (calc *SpotArbitrageCalculator) CalculateArbitrageOpportunities(
    ctx context.Context,
    marketData map[string][]models.MarketData,
) ([]models.ArbitrageOpportunity, error) {
    var opportunities []models.ArbitrageOpportunity
    
    for _, exchangeData := range marketData {
        if len(exchangeData) < 2 {
            continue // Need at least 2 exchanges
        }
        
        // Find lowest and highest prices
        var lowestPrice, highestPrice decimal.Decimal
        var lowestExchange, highestExchange *models.Exchange
        
        // Calculate profit percentage
        profitPercentage := highestPrice.Sub(lowestPrice)
            .Div(lowestPrice)
            .Mul(decimal.NewFromInt(100))
        
        if profitPercentage.GreaterThan(decimal.NewFromFloat(0.1)) {
            opportunities = append(opportunities, opportunity)
        }
    }
    return opportunities, nil
}

type FuturesArbitrageCalculationInput struct {
    Symbol           string
    LongExchange     string
    ShortExchange    string
    LongFundingRate  decimal.Decimal
    ShortFundingRate decimal.Decimal
    LongMarkPrice    decimal.Decimal
    ShortMarkPrice   decimal.Decimal
    FundingInterval  int // Hours (typically 8)
}

// Net funding rate is the profit from the spread
netFundingRate := shortRate.Sub(longRate)

// APY calculation
hourlyRate := netFundingRate.Div(decimal.NewFromInt(fundingInterval))
dailyRate := hourlyRate.Mul(decimal.NewFromInt(24))
apy := dailyRate.Mul(decimal.NewFromInt(365))

func (s *FuturesArbitrageService) runOpportunityCalculator() {
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-s.ctx.Done():
            return
        case <-ticker.C:
            calcCtx, cancel := context.WithTimeout(s.ctx, 25*time.Second)
            
            err := s.errorRecoveryManager.ExecuteWithRetry(
                calcCtx,
                "calculate_opportunities",
                func() error {
                    return s.calculateAndStoreOpportunities(calcCtx)
                },
            )
            cancel()
        }
    }
}

ctx, span := observability.StartSpan(s.ctx, observability.SpanOpArbitrage, "arbitrage_calculation_cycle")
defer observability.FinishSpan(span, nil)

span.SetTag("service", "arbitrage")
span.SetData("min_profit_threshold", s.arbitrageConfig.MinProfit)

observability.AddBreadcrumb(ctx, "arbitrage", "Starting arbitrage calculation cycle", sentry.LevelInfo)

func (s *ArbitrageService) diagnoseNoMarketData() {
    var totalRows, freshRows, activeExchanges, activePairs int
    
    // Count total market_data rows
    s.db.QueryRow(s.ctx, "SELECT COUNT(*) FROM market_data").Scan(&totalRows)
    
    // Count fresh data (within 10 minutes)
    s.db.QueryRow(s.ctx, freshDataQuery).Scan(&freshRows)
    
    // Count active exchanges and trading pairs
    s.db.QueryRow(s.ctx, "SELECT COUNT(*) FROM exchanges WHERE status = 'active'").Scan(&activeExchanges)
    s.db.QueryRow(s.ctx, "SELECT COUNT(*) FROM trading_pairs WHERE is_active = true").Scan(&activePairs)
    
    s.logger.WithFields(map[string]interface{}{
        "total_market_data_rows": totalRows,
        "fresh_rows_10min":       freshRows,
        "active_exchanges":       activeExchanges,
        "active_trading_pairs":   activePairs,
        "recommendation":         s.getDiagnosticRecommendation(...),
    }).Warn("Market data diagnostic")
}