refactor(data): 重构交易所适配器,修复 Kline 实体复合主键

- 废弃旧 adapter 体系 (base/binance/types.ts),新增 base_rest/rest.ts
  基于 Binance 官方 SDK 实现 REST K 线拉取
- Kline 实体改为四列复合主键 (exchange/symbol/interval/time),
  修复单列 time PK 导致的跨 symbol 写入冲突
- 新增 filterConsecutive():K 线连续性过滤,首缺口截断策略
- 新增 service/kline.ts:批量 UPSERT K 线入库
- 新增 types/ 共享类型定义、example/ 示例、run/ 运行脚本
This commit is contained in:
Rekey
2026-06-08 18:18:16 +08:00
parent 85a0031a78
commit 5e385547c7
16 changed files with 829 additions and 1043 deletions
+73
View File
@@ -0,0 +1,73 @@
import { AppDataSource } from "../db/data-source";
import { Kline } from "../db/entities/kline.entity";
import type { Kline as KlineItem } from "../types";
import { logger } from "../utils/logger";
const repo = AppDataSource.getRepository(Kline);
/**
* 批量 UPSERT K 线数据到 TimescaleDB。
*
* 映射应用层 KlineItem → 数据库实体,通过 INSERT ... ON CONFLICT DO UPDATE
* 实现幂等写入。冲突列为 [exchange, symbol, interval, time](四列复合主键),
* 冲突时更新 OHLCV 及扩展字段。
*
* 适用场景:
* - 回补历史 K 线(幂等,重复拉取不产生重复行)
* - WebSocket 实时 K 线增量刷新(更新最新一根未闭合 K 线的 high/low/close/volume
*
* 注意:依赖 Kline 实体的四列复合主键 [exchange, symbol, interval, time]。
* 若实体 PK 结构变更,需同步更新 conflictPaths。
*
* @param KlineItems - 应用层标准化 K 线数组
*/
export async function upsertOrUpdateKlines(KlineItems: KlineItem[]) {
if (KlineItems.length === 0) {
return;
}
logger.debug({ count: KlineItems.length }, "开始批量 UPSERT K 线");
// 应用层 KlineItem → 数据库实体 Kline
// 注意类型转换:应用层价格为 string(兼容交易所 SDK),DB 层为 NUMERIC(number)
const entities = KlineItems.map((item) => {
const entity = new Kline();
entity.time = new Date(item.openTime); // Unix ms → Date
entity.exchange = item.exchange;
entity.symbol = item.symbol;
entity.interval = item.interval;
entity.open = Number(item.open);
entity.high = Number(item.high);
entity.low = Number(item.low);
entity.close = Number(item.close);
entity.volume = Number(item.volume);
entity.quote_volume = item.quoteVolume ? Number(item.quoteVolume) : undefined;
entity.taker_buy_base_vol = item.takerBuyBaseVol
? Number(item.takerBuyBaseVol)
: undefined;
entity.taker_buy_quote_vol = item.takerBuyQuoteVol
? Number(item.takerBuyQuoteVol)
: undefined;
entity.trade_count = item.tradeCount ? Number(item.tradeCount) : undefined;
entity.is_closed = item.isClosed;
return entity;
});
try {
// UPSERT: 冲突列匹配复合主键 [exchange, symbol, interval, time]
// 实体已改为四列复合 PK,ON CONFLICT 直接命中主键约束
// skipUpdateIfNoValuesChanged: 减少不必要的写操作
const result = await repo.upsert(entities, {
conflictPaths: ["exchange", "symbol", "interval", "time"],
skipUpdateIfNoValuesChanged: true,
});
logger.info(
{ count: KlineItems.length, generatedMaps: result.generatedMaps.length },
"K 线 UPSERT 完成",
);
} catch (err) {
logger.error({ err, count: KlineItems.length }, "K 线 UPSERT 失败");
throw err;
}
}