trade/data/run/exchange.ts

import { logger } from "../utils/logger";
import { getAllPairs, updatePairLastBackfillTime } from '../service/pair';
import { upsertOrUpdateKlines } from "../service/kline";
import { fetchKlines } from '../exchanges';

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();
    try {
        while (lastBackfillTime < getNowMinuteMS()) {
            console.log('lastBackfillTime', lastBackfillTime);
            const klines = await fetchKlines({
                exchange: 'binance',
                type: pair.type,
                symbol: pair.symbol,
                startTime: lastBackfillTime,
                limit: 500,
            });
            console.log(`拉取到 ${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;
                }
            }
            await new Promise((resolve) => {
                setTimeout(resolve, Math.random() * 1000);
            });
        }
    } catch (err) {
        console.error("拉取失败:", err);
    }
}

// 所有币种回补完成以后等待1秒关闭pgsql连接退出此进程
await new Promise((resolve) => setTimeout(resolve, 1000));
const { AppDataSource } = await import("../db/data-source");
if (AppDataSource.isInitialized) {
    await AppDataSource.destroy();
    console.log("pgsql 连接已关闭");
}
process.exit(0);