import { logger } from "../utils/logger"; import { getAllPairs, updatePairLastBackfillTime } from '../service/pair'; import { upsertOrUpdateKlines } from "../service/kline"; import { fetchKlines } from '../exchanges'; import { AppDataSource } from '../db/data-source'; async function exit() { AppDataSource.destroy().finally(() => { process.exit(0); }); } function getNowMinuteMS() { const minuteMS = 1000 * 60; return Math.floor(Date.now() / minuteMS) * minuteMS } const allPairs = await getAllPairs(); for (const pair of allPairs) { let lastBackfillTime = pair.last_backfill_time.getTime(); while (lastBackfillTime < getNowMinuteMS()) { const timer = setTimeout(exit, 10000); try { logger.info({ lastBackfillTime }, '回补进度'); const klines = await fetchKlines({ exchange: 'binance', type: pair.type, symbol: pair.symbol, startTime: lastBackfillTime, limit: 1000, }); clearTimeout(timer); logger.info(`拉取到 ${klines.length} 条 K 线`); if (klines.length > 0) { await upsertOrUpdateKlines(klines); const lastK = klines[klines.length - 1]; if (lastK) { await updatePairLastBackfillTime(lastK?.symbol, new Date(lastK.openTime), pair.type); if (lastBackfillTime === lastK.openTime) { break; } lastBackfillTime = lastK.openTime; } } } catch (err) { clearTimeout(timer); logger.error({ err }, "拉取失败"); } await new Promise((resolve) => { setTimeout(resolve, Math.random() * 1000); }); } } // 所有交易对均已完成回补,等待 10~40 秒再退出, // 避免外部进程管理立即重启导致高频空查触发 API 限流。 await new Promise((resolve) => { setTimeout(resolve, Math.random() * 30 * 1000 + 10000); }); exit();