cryptoHermes/internal/worker/collector.go

package worker

import (
	"context"
	"log/slog"

	"cryptoHermes/internal/entity"
	"cryptoHermes/internal/usecase"
)

type Deps struct {
	MarketData  usecase.MarketDataProvider
	Derivatives usecase.DerivativesProvider

	KlineRepo   usecase.KlineRepository
	FundingRepo usecase.FundingRepository
	OIRepo      usecase.OpenInterestRepository
	LSRepo      usecase.LongShortRatioRepository
	TakerRepo   usecase.TakerVolumeRepository

	Symbols   []string
	Intervals []string
	Limit     int
	Logger    *slog.Logger
}

type Collector struct {
	d Deps
}

func NewCollector(d Deps) *Collector {
	if d.Limit <= 0 {
		d.Limit = 500
	}
	return &Collector{d: d}
}

func (c *Collector) Symbols() []string   { return c.d.Symbols }
func (c *Collector) Intervals() []string { return c.d.Intervals }

func (c *Collector) CollectKlines(ctx context.Context, interval string) error {
	for _, sym := range c.d.Symbols {
		ks, err := c.d.MarketData.GetKlines(ctx, sym, interval, c.d.Limit)
		if err != nil {
			c.d.Logger.Error("collect_klines_failed", "symbol", sym, "interval", interval, "err", err)
			continue
		}
		closed := make([]entity.Kline, 0, len(ks))
		for _, k := range ks {
			if k.IsClosed {
				closed = append(closed, k)
			}
		}
		if err := c.d.KlineRepo.UpsertMany(ctx, closed); err != nil {
			c.d.Logger.Error("upsert_klines_failed", "symbol", sym, "interval", interval, "err", err)
			continue
		}
		c.d.Logger.Info("collect_klines_ok", "symbol", sym, "interval", interval, "rows", len(closed))
	}
	return nil
}

func (c *Collector) CollectAllKlines(ctx context.Context) error {
	for _, iv := range c.d.Intervals {
		_ = c.CollectKlines(ctx, iv)
	}
	return nil
}

func (c *Collector) CollectFunding(ctx context.Context) error {
	for _, sym := range c.d.Symbols {
		hist, err := c.d.Derivatives.GetFundingHistory(ctx, sym, 100)
		if err != nil {
			c.d.Logger.Error("collect_funding_failed", "symbol", sym, "err", err)
			continue
		}
		if err := c.d.FundingRepo.UpsertMany(ctx, hist); err != nil {
			c.d.Logger.Error("upsert_funding_failed", "symbol", sym, "err", err)
			continue
		}
		c.d.Logger.Info("collect_funding_ok", "symbol", sym, "rows", len(hist))
	}
	return nil
}

func (c *Collector) CollectOpenInterest(ctx context.Context) error {
	periods := []string{"5m", "15m", "1h", "4h", "1d"}
	for _, sym := range c.d.Symbols {
		for _, p := range periods {
			hist, err := c.d.Derivatives.GetOpenInterestHistory(ctx, sym, p, 500)
			if err != nil {
				c.d.Logger.Error("collect_oi_failed", "symbol", sym, "period", p, "err", err)
				continue
			}
			if err := c.d.OIRepo.UpsertMany(ctx, hist); err != nil {
				c.d.Logger.Error("upsert_oi_failed", "symbol", sym, "period", p, "err", err)
				continue
			}
			c.d.Logger.Info("collect_oi_ok", "symbol", sym, "period", p, "rows", len(hist))
		}
	}
	return nil
}

func (c *Collector) CollectLongShortRatio(ctx context.Context) error {
	periods := []string{"5m", "15m", "1h", "4h", "1d"}
	for _, sym := range c.d.Symbols {
		for _, p := range periods {
			g, err := c.d.Derivatives.GetGlobalLongShortRatio(ctx, sym, p, 500)
			if err == nil {
				if err := c.d.LSRepo.UpsertMany(ctx, g); err != nil {
					c.d.Logger.Error("upsert_global_ls_failed", "symbol", sym, "period", p, "err", err)
				} else {
					c.d.Logger.Info("collect_global_ls_ok", "symbol", sym, "period", p, "rows", len(g))
				}
			} else {
				c.d.Logger.Error("collect_global_ls_failed", "symbol", sym, "period", p, "err", err)
			}

			t, err := c.d.Derivatives.GetTopTraderPositionRatio(ctx, sym, p, 500)
			if err == nil {
				if err := c.d.LSRepo.UpsertMany(ctx, t); err != nil {
					c.d.Logger.Error("upsert_top_ls_failed", "symbol", sym, "period", p, "err", err)
				} else {
					c.d.Logger.Info("collect_top_ls_ok", "symbol", sym, "period", p, "rows", len(t))
				}
			} else {
				c.d.Logger.Error("collect_top_ls_failed", "symbol", sym, "period", p, "err", err)
			}
		}
	}
	return nil
}

func (c *Collector) CollectTakerVolume(ctx context.Context) error {
	periods := []string{"5m", "15m", "1h", "4h", "1d"}
	for _, sym := range c.d.Symbols {
		for _, p := range periods {
			items, err := c.d.Derivatives.GetTakerBuySellVolume(ctx, sym, p, 500)
			if err != nil {
				c.d.Logger.Error("collect_taker_failed", "symbol", sym, "period", p, "err", err)
				continue
			}
			if err := c.d.TakerRepo.UpsertMany(ctx, items); err != nil {
				c.d.Logger.Error("upsert_taker_failed", "symbol", sym, "period", p, "err", err)
				continue
			}
			c.d.Logger.Info("collect_taker_ok", "symbol", sym, "period", p, "rows", len(items))
		}
	}
	return nil
}