cryptoHermes/cmd/backfill/main.go

package main

import (
	"context"
	"flag"
	"fmt"
	"log/slog"
	"os"
	"time"

	"cryptoHermes/config"
	"cryptoHermes/internal/entity"
	"cryptoHermes/internal/repo/persistent/postgres"
	"cryptoHermes/internal/repo/webapi/binance"
	pkglogger "cryptoHermes/pkg/logger"
	pgpkg "cryptoHermes/pkg/postgres"
)

func main() {
	var (
		cfgPath  string
		symbol   string
		interval string
		fromMs   int64
		toMs     int64
		limit    int
	)
	flag.StringVar(&cfgPath, "config", "config/config.yml", "config file path")
	flag.StringVar(&symbol, "symbol", "BTCUSDT", "trading pair, e.g. BTCUSDT")
	flag.StringVar(&interval, "interval", "1h", "kline interval (15m/1h/4h/1d/1w)")
	flag.Int64Var(&fromMs, "from", 0, "start time in ms (0 = pull most recent N bars)")
	flag.Int64Var(&toMs, "to", 0, "end time in ms (0 = now)")
	flag.IntVar(&limit, "limit", 1500, "bars per request, max 1500")
	flag.Parse()

	if env := os.Getenv("CONFIG_PATH"); env != "" {
		cfgPath = env
	}

	cfg, err := config.Load(cfgPath)
	if err != nil {
		slog.Default().Error("config_load_failed", "err", err)
		os.Exit(1)
	}
	log := pkglogger.New(cfg.App.Env)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
	defer cancel()

	pool, err := pgpkg.NewPool(ctx, pgpkg.Options{
		DSN:      cfg.Postgres.DSN,
		MaxConns: cfg.Postgres.MaxConns,
		MinConns: cfg.Postgres.MinConns,
	})
	if err != nil {
		log.Error("pool", "err", err)
		os.Exit(1)
	}
	defer pool.Close()

	client := binance.NewClient(binance.ClientOptions{
		BaseURL:    cfg.Binance.BaseURL,
		Timeout:    cfg.Binance.Timeout,
		RetryCount: cfg.Binance.RetryCount,
		RPS:        cfg.Binance.RPS,
		Burst:      cfg.Binance.Burst,
	})
	repo := postgres.NewKlineRepo(pool)

	if toMs == 0 {
		toMs = time.Now().UnixMilli()
	}

	if fromMs == 0 {
		ks, err := client.GetKlines(ctx, symbol, interval, limit)
		if err != nil {
			log.Error("fetch", "err", err)
			os.Exit(1)
		}
		closed := filterClosed(ks)
		if err := repo.UpsertMany(ctx, closed); err != nil {
			log.Error("upsert", "err", err)
			os.Exit(1)
		}
		log.Info("backfill_done", "symbol", symbol, "interval", interval, "rows", len(closed))
		return
	}

	cursor := fromMs
	totalRows := 0
	for cursor < toMs {
		ks, err := client.GetKlinesRange(ctx, symbol, interval, cursor, toMs, limit)
		if err != nil {
			log.Error("fetch_range", "cursor", cursor, "err", err)
			os.Exit(1)
		}
		if len(ks) == 0 {
			break
		}
		closed := filterClosed(ks)
		if err := repo.UpsertMany(ctx, closed); err != nil {
			log.Error("upsert", "err", err)
			os.Exit(1)
		}
		totalRows += len(closed)
		lastClose := ks[len(ks)-1].CloseTime
		if lastClose <= cursor {
			break
		}
		cursor = lastClose + 1
		fmt.Printf("\rbackfill: %s/%s rows=%d cursor=%d ", symbol, interval, totalRows, cursor)
	}
	fmt.Println()
	log.Info("backfill_done", "symbol", symbol, "interval", interval, "rows", totalRows)
}

func filterClosed(in []entity.Kline) []entity.Kline {
	out := make([]entity.Kline, 0, len(in))
	for _, k := range in {
		if k.IsClosed {
			out = append(out, k)
		}
	}
	return out
}