feat: simplify Claw402 autopilot trading flow

kernel/engine.go

@@ -10,11 +10,13 @@ import (
 	"nofx/market"
 	"nofx/provider/hyperliquid"
 	"nofx/provider/nofxos"
+	"nofx/provider/vergex"
 	"nofx/security"
 	"nofx/store"
 	"os"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -104,6 +106,7 @@ type Context struct {
 	MultiTFMarket      map[string]map[string]*market.Data `json:"-"`
 	OITopDataMap       map[string]*OITopData              `json:"-"`
 	QuantDataMap       map[string]*QuantData              `json:"-"`
+	VergexDataMap      map[string]*vergex.MarketAnalysis  `json:"-"`
 	OIRankingData      *nofxos.OIRankingData              `json:"-"` // Market-wide OI ranking data
 	NetFlowRankingData *nofxos.NetFlowRankingData         `json:"-"` // Market-wide fund flow ranking data
 	PriceRankingData   *nofxos.PriceRankingData           `json:"-"` // Market-wide price gainers/losers
@@ -183,8 +186,10 @@ type OIDeltaData struct {
 
 // StrategyEngine strategy execution engine
 type StrategyEngine struct {
-	config       *store.StrategyConfig
-	nofxosClient *nofxos.Client
+	config             *store.StrategyConfig
+	nofxosClient       *nofxos.Client
+	vergexClient       *vergex.Client
+	vergexRankingCache map[string]*vergex.SignalRankItem
 }
 
 // NewStrategyEngine creates strategy execution engine.
@@ -217,11 +222,25 @@ func NewStrategyEngine(config *store.StrategyConfig, claw402WalletKey ...string)
 		} else {
 			logger.Warnf("⚠️ Failed to init claw402 data client: %v (using direct nofxos.ai)", err)
 		}
+
+		vergexClient, err := vergex.NewClient(claw402URL, walletKey, &logger.MCPLogger{})
+		if err == nil {
+			logger.Infof("🔗 Vergex signals routed through claw402 (%s)", claw402URL)
+		} else {
+			logger.Warnf("⚠️ Failed to init Vergex claw402 client: %v", err)
+		}
+		return &StrategyEngine{
+			config:             config,
+			nofxosClient:       client,
+			vergexClient:       vergexClient,
+			vergexRankingCache: make(map[string]*vergex.SignalRankItem),
+		}
 	}
 
 	return &StrategyEngine{
-		config:       config,
-		nofxosClient: client,
+		config:             config,
+		nofxosClient:       client,
+		vergexRankingCache: make(map[string]*vergex.SignalRankItem),
 	}
 }
 
@@ -230,7 +249,7 @@ func (e *StrategyEngine) usesHyperliquidNativeUniverse() bool {
 		return false
 	}
 	source := e.config.CoinSource
-	if source.SourceType == "hyper_all" || source.SourceType == "hyper_main" || source.SourceType == "hyper_rank" || source.UseHyperAll || source.UseHyperMain {
+	if source.SourceType == "hyper_all" || source.SourceType == "hyper_main" || source.SourceType == "hyper_rank" || source.SourceType == "vergex_signal" || source.UseHyperAll || source.UseHyperMain {
 		return true
 	}
 	for _, symbol := range source.StaticCoins {
@@ -392,6 +411,20 @@ func (e *StrategyEngine) GetCandidateCoins() ([]CandidateCoin, error) {
 		}
 		return e.filterExcludedCoins(coins), nil
 
+	case "vergex_signal":
+		coins, err := e.getVergexSignalCoins(
+			coinSource.VergexLimit,
+			coinSource.VergexMarketType,
+			coinSource.VergexChain,
+			coinSource.VergexLiqBand,
+			coinSource.HyperRankCategory,
+			coinSource.StaticCoins,
+		)
+		if err != nil {
+			return nil, err
+		}
+		return e.filterExcludedCoins(coins), nil
+
 	case "mixed":
 		if coinSource.UseAI500 {
 			poolCoins, err := e.getAI500Coins(coinSource.AI500Limit)
@@ -694,6 +727,126 @@ func (e *StrategyEngine) getHyperRankCoins(category, direction string, limit int
 	return candidates, nil
 }
 
+func (e *StrategyEngine) getVergexSignalCoins(limit int, marketType, chain, liqBand, category string, selectedSymbols []string) ([]CandidateCoin, error) {
+	if e.vergexClient == nil {
+		return nil, fmt.Errorf("vergex signal source requires a configured claw402 wallet")
+	}
+	if marketType == "" {
+		marketType = vergex.DefaultMarketType
+	}
+	chain = vergex.QueryChain(chain)
+	if limit <= 0 {
+		limit = 5
+	}
+	if limit > store.MaxCandidateCoins {
+		limit = store.MaxCandidateCoins
+	}
+	category = strings.ToLower(strings.TrimSpace(category))
+
+	ranking, err := e.vergexClient.GetSignalRanking(context.Background(), vergex.Query{
+		Chain:   chain,
+		LiqBand: liqBand,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch Vergex signal ranking: %w", err)
+	}
+
+	rankedItems := vergex.FilterSignalRankingItems(ranking.Items, marketType, store.MaxCandidateCoins)
+	if len(rankedItems) == 0 && strings.TrimSpace(chain) != "" {
+		fallbackRanking, fallbackErr := e.vergexClient.GetSignalRanking(context.Background(), vergex.Query{
+			LiqBand: liqBand,
+		})
+		if fallbackErr == nil {
+			fallbackItems := vergex.FilterSignalRankingItems(fallbackRanking.Items, marketType, store.MaxCandidateCoins)
+			if len(fallbackItems) > 0 {
+				logger.Infof("✅ Vergex signal ranking returned TradeFi items after retrying without chain filter (chain=%s)", chain)
+				ranking = fallbackRanking
+				rankedItems = fallbackItems
+			}
+		} else {
+			logger.Warnf("⚠️ Vergex signal ranking retry without chain failed: %v", fallbackErr)
+		}
+	}
+	e.vergexRankingCache = make(map[string]*vergex.SignalRankItem, len(rankedItems))
+	for _, item := range rankedItems {
+		itemCopy := item
+		if symbol := vergex.TradableSymbolForMarket(item.MarketType, item.Symbol); symbol != "" {
+			e.vergexRankingCache[symbol] = &itemCopy
+		}
+	}
+
+	if len(selectedSymbols) > 0 {
+		candidates := make([]CandidateCoin, 0, minInt(len(selectedSymbols), limit))
+		seen := make(map[string]bool)
+		for _, raw := range selectedSymbols {
+			symbol := vergex.TradableSymbolForMarket(marketType, raw)
+			if symbol == "" || seen[symbol] {
+				continue
+			}
+			candidates = append(candidates, CandidateCoin{
+				Symbol:  symbol,
+				Sources: []string{"vergex_signal"},
+			})
+			seen[symbol] = true
+			if len(candidates) >= limit {
+				break
+			}
+		}
+		if len(candidates) == 0 {
+			return nil, fmt.Errorf("selected Claw402 symbols are not tradable %s items", marketType)
+		}
+		logger.Infof("✅ Loaded %d selected Vergex candidates (%s)", len(candidates), marketType)
+		return candidates, nil
+	}
+
+	items := make([]vergex.SignalRankItem, 0, limit)
+	for _, item := range rankedItems {
+		if category != "" && category != "all" && item.Category != category {
+			continue
+		}
+		items = append(items, item)
+		if len(items) >= limit {
+			break
+		}
+	}
+	if len(items) == 0 {
+		if category != "" && category != "all" {
+			return nil, fmt.Errorf("vergex signal ranking returned no tradable %s items in category %s", marketType, category)
+		}
+		return nil, fmt.Errorf("vergex signal ranking returned no tradable %s items", marketType)
+	}
+
+	candidates := make([]CandidateCoin, 0, len(items))
+	for _, item := range items {
+		itemCopy := item
+		symbol := vergex.TradableSymbolForMarket(item.MarketType, item.Symbol)
+		if symbol == "" {
+			continue
+		}
+		e.vergexRankingCache[symbol] = &itemCopy
+		candidates = append(candidates, CandidateCoin{
+			Symbol:  symbol,
+			Sources: []string{"vergex_signal"},
+		})
+	}
+	logger.Infof("✅ Loaded %d Vergex signal candidates (%s/%s, capped at %d)", len(candidates), marketType, withDefaultText(category, "all"), limit)
+	return candidates, nil
+}
+
+func minInt(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func withDefaultText(value, fallback string) string {
+	if strings.TrimSpace(value) == "" {
+		return fallback
+	}
+	return value
+}
+
 // ============================================================================
 // External & Quant Data
 // ============================================================================
@@ -879,6 +1032,282 @@ func (e *StrategyEngine) FetchQuantDataBatch(symbols []string) map[string]*Quant
 	return result
 }
 
+func (e *StrategyEngine) FetchVergexDataBatch(ctx context.Context, symbols []string) map[string]*vergex.MarketAnalysis {
+	result := make(map[string]*vergex.MarketAnalysis)
+	if e == nil || e.config == nil || e.config.CoinSource.SourceType != "vergex_signal" {
+		return result
+	}
+	if e.vergexClient == nil {
+		logger.Warnf("⚠️ Vergex signal data skipped: claw402 wallet is not configured")
+		return result
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	source := e.config.CoinSource
+	marketType := source.VergexMarketType
+	if marketType == "" {
+		marketType = vergex.DefaultMarketType
+	}
+	chain := source.VergexChain
+	chain = vergex.QueryChain(chain)
+
+	seen := make(map[string]bool)
+	limited := make([]string, 0, store.MaxCandidateCoins)
+	for _, symbol := range symbols {
+		symbol = vergexDetailSymbolForLookup(marketType, symbol)
+		if symbol == "" {
+			continue
+		}
+		if seen[symbol] {
+			continue
+		}
+		seen[symbol] = true
+		limited = append(limited, symbol)
+		if len(limited) >= store.MaxCandidateCoins+store.MaxPositions {
+			break
+		}
+	}
+
+	type vergexAnalysisResult struct {
+		symbol   string
+		analysis *vergex.MarketAnalysis
+	}
+
+	resultCh := make(chan vergexAnalysisResult, len(limited))
+	var wg sync.WaitGroup
+	sem := make(chan struct{}, vergexDetailSymbolConcurrency)
+	for _, symbol := range limited {
+		symbol := symbol
+		querySymbol := vergex.QuerySymbol(symbol)
+		if querySymbol == "" {
+			continue
+		}
+		itemMarketType := marketType
+		itemCategory := ""
+		var ranking *vergex.SignalRankItem
+		if cached, ok := e.vergexRankingCache[symbol]; ok && cached != nil {
+			ranking = cached
+			if cached.MarketType != "" {
+				itemMarketType = cached.MarketType
+			}
+			itemCategory = cached.Category
+		}
+
+		analysis := &vergex.MarketAnalysis{
+			Symbol:      symbol,
+			QuerySymbol: querySymbol,
+			MarketType:  itemMarketType,
+			Ranking:     ranking,
+		}
+		query := vergex.Query{
+			MarketType: itemMarketType,
+			Symbol:     symbol,
+			Chain:      chain,
+			LiqBand:    source.VergexLiqBand,
+			Category:   itemCategory,
+		}
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			select {
+			case sem <- struct{}{}:
+				defer func() { <-sem }()
+			case <-ctx.Done():
+				analysis.SignalLabError = ctx.Err().Error()
+				analysis.HeatmapError = ctx.Err().Error()
+				resultCh <- vergexAnalysisResult{symbol: symbol, analysis: analysis}
+				return
+			}
+			e.populateVergexDetailData(ctx, analysis, query)
+			if len(analysis.SignalLab) > 0 || len(analysis.Heatmap) > 0 ||
+				analysis.SignalLabError != "" || analysis.HeatmapError != "" || analysis.Ranking != nil {
+				resultCh <- vergexAnalysisResult{symbol: symbol, analysis: analysis}
+			}
+		}()
+	}
+
+	wg.Wait()
+	close(resultCh)
+	for item := range resultCh {
+		result[item.symbol] = item.analysis
+	}
+
+	logger.Infof("📊 Vergex detail data ready for %d symbols", len(result))
+	return result
+}
+
+func vergexDetailSymbolForLookup(marketType, symbol string) string {
+	return vergex.TradableSymbolForMarket(marketType, symbol)
+}
+
+const (
+	vergexDetailRequestTimeout    = 45 * time.Second
+	vergexDetailSymbolConcurrency = 2
+)
+
+func (e *StrategyEngine) populateVergexDetailData(ctx context.Context, analysis *vergex.MarketAnalysis, query vergex.Query) {
+	type endpointResult struct {
+		name string
+		body json.RawMessage
+		err  error
+	}
+
+	run := func(name string, fetch func(context.Context, vergex.Query) (json.RawMessage, error), out chan<- endpointResult) {
+		requestCtx, cancel := context.WithTimeout(ctx, vergexDetailRequestTimeout)
+		defer cancel()
+		body, err := fetch(requestCtx, query)
+		out <- endpointResult{name: name, body: body, err: err}
+	}
+
+	out := make(chan endpointResult, 2)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		run("signal-lab", e.fetchVergexSignalLabWithFallback, out)
+	}()
+	go func() {
+		defer wg.Done()
+		run("heatmap", e.fetchVergexHeatmapWithFallback, out)
+	}()
+	wg.Wait()
+	close(out)
+
+	for item := range out {
+		switch item.name {
+		case "signal-lab":
+			if item.err != nil {
+				logger.Warnf("⚠️ Failed to fetch Vergex signal-lab for %s: %v", analysis.Symbol, item.err)
+				analysis.SignalLabError = item.err.Error()
+			} else {
+				analysis.SignalLab = item.body
+			}
+		case "heatmap":
+			if item.err != nil {
+				logger.Warnf("⚠️ Failed to fetch Vergex heatmap for %s: %v", analysis.Symbol, item.err)
+				analysis.HeatmapError = item.err.Error()
+			} else {
+				analysis.Heatmap = item.body
+			}
+		}
+	}
+}
+
+func (e *StrategyEngine) fetchVergexSignalLabWithFallback(ctx context.Context, query vergex.Query) (json.RawMessage, error) {
+	var lastErr error
+	for idx, candidate := range vergexDetailQueryCandidates(query) {
+		body, err := e.vergexClient.GetSignalLab(ctx, candidate)
+		if err == nil {
+			if idx > 0 {
+				logger.Infof("✅ Vergex signal-lab succeeded with fallback marketType=%s chain=%s", candidate.MarketType, withDefaultText(candidate.Chain, "default"))
+			}
+			return body, nil
+		}
+		lastErr = err
+		if !isRetryableVergexDetailError(err) {
+			break
+		}
+	}
+	return nil, lastErr
+}
+
+func (e *StrategyEngine) fetchVergexHeatmapWithFallback(ctx context.Context, query vergex.Query) (json.RawMessage, error) {
+	var lastErr error
+	for idx, candidate := range vergexDetailQueryCandidates(query) {
+		body, err := e.vergexClient.GetCostLiquidationHeatmap(ctx, candidate)
+		if err == nil {
+			if idx > 0 {
+				logger.Infof("✅ Vergex heatmap succeeded with fallback marketType=%s chain=%s", candidate.MarketType, withDefaultText(candidate.Chain, "default"))
+			}
+			return body, nil
+		}
+		lastErr = err
+		if !isRetryableVergexDetailError(err) {
+			break
+		}
+	}
+	return nil, lastErr
+}
+
+func vergexDetailQueryCandidates(query vergex.Query) []vergex.Query {
+	marketTypes := vergexDetailMarketTypeCandidates(query)
+	chains := uniqueValues(query.Chain, "mainnet", "")
+
+	candidates := make([]vergex.Query, 0, len(marketTypes)*len(chains))
+	for _, marketType := range marketTypes {
+		for _, chain := range chains {
+			candidate := query
+			candidate.MarketType = marketType
+			candidate.Chain = chain
+			candidates = append(candidates, candidate)
+		}
+	}
+	return candidates
+}
+
+func vergexDetailMarketTypeCandidates(query vergex.Query) []string {
+	if isVergexAllMarketType(query.MarketType) {
+		if market.IsXyzDexAsset(query.Symbol) {
+			return uniqueNonEmpty(vergex.DefaultMarketType, "hip3-perp", "hip3Perp", "core_perp")
+		}
+		return uniqueNonEmpty("core_perp", vergex.DefaultMarketType, "hip3-perp", "hip3Perp")
+	}
+	values := []string{query.MarketType, vergex.DefaultMarketType, "hip3-perp", "hip3Perp", "core_perp"}
+	return uniqueNonEmpty(values...)
+}
+
+func isVergexAllMarketType(marketType string) bool {
+	switch strings.ToLower(strings.TrimSpace(marketType)) {
+	case "", "all", "any", "ranking", "signal-ranking", "signal_ranking", "claw402", "vergex":
+		return true
+	default:
+		return false
+	}
+}
+
+func isRetryableVergexDetailError(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "invalid markettype") ||
+		strings.Contains(msg, "invalid_request") ||
+		strings.Contains(msg, "invalid chain") ||
+		strings.Contains(msg, "market not found") ||
+		strings.Contains(msg, "not_found")
+}
+
+func uniqueNonEmpty(values ...string) []string {
+	out := make([]string, 0, len(values))
+	seen := make(map[string]bool, len(values))
+	for _, value := range values {
+		value = strings.TrimSpace(value)
+		if value == "" || seen[value] {
+			continue
+		}
+		seen[value] = true
+		out = append(out, value)
+	}
+	return out
+}
+
+func uniqueValues(values ...string) []string {
+	out := make([]string, 0, len(values))
+	seen := make(map[string]bool, len(values))
+	for _, value := range values {
+		value = strings.TrimSpace(value)
+		if seen[value] {
+			continue
+		}
+		seen[value] = true
+		out = append(out, value)
+	}
+	return out
+}
+
 // FetchOIRankingData fetches market-wide OI ranking data
 func (e *StrategyEngine) FetchOIRankingData() *nofxos.OIRankingData {
 	indicators := e.config.Indicators