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 }