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 }