763 lines
30 KiB
Python
763 lines
30 KiB
Python
"""
|
||
BitMart 多周期K线数据抓取脚本
|
||
支持同时获取 1分钟、3分钟、5分钟、15分钟、30分钟、1小时 K线数据
|
||
支持秒级价格数据（通过成交记录API）
|
||
支持断点续传，从数据库最新/最早记录继续抓取
|
||
"""
|
||
|
||
import time
|
||
import datetime
|
||
from pathlib import Path
|
||
from loguru import logger
|
||
from peewee import *
|
||
from bitmart.api_contract import APIContract
|
||
|
||
# 数据库配置（使用脚本所在项目目录下的 models）
|
||
DB_PATH = Path(__file__).parent / 'models' / 'database.db'
|
||
db = SqliteDatabase(str(DB_PATH))
|
||
|
||
# K线周期配置：step值 -> 表名后缀
|
||
KLINE_CONFIGS = {
|
||
1: '1m', # 1分钟
|
||
3: '3m', # 3分钟
|
||
5: '5m', # 5分钟
|
||
15: '15m', # 15分钟
|
||
30: '30m', # 30分钟
|
||
60: '1h', # 1小时
|
||
}
|
||
|
||
|
||
class BitMartETHTrades(Model):
|
||
"""成交记录模型（秒级/毫秒级原始数据）"""
|
||
id = BigIntegerField(primary_key=True) # 成交ID
|
||
timestamp = BigIntegerField(index=True) # 成交时间戳（毫秒）
|
||
price = FloatField() # 成交价格
|
||
volume = FloatField() # 成交量
|
||
side = IntegerField() # 方向: 1=买, -1=卖
|
||
|
||
class Meta:
|
||
database = db
|
||
table_name = 'bitmart_eth_trades'
|
||
|
||
|
||
class BitMartETHSecond(Model):
|
||
"""秒级K线模型（由成交记录聚合而来）"""
|
||
id = BigIntegerField(primary_key=True) # 时间戳（毫秒，取整到秒）
|
||
open = FloatField(null=True)
|
||
high = FloatField(null=True)
|
||
low = FloatField(null=True)
|
||
close = FloatField(null=True)
|
||
volume = FloatField(null=True)
|
||
trade_count = IntegerField(null=True) # 该秒内成交笔数
|
||
|
||
class Meta:
|
||
database = db
|
||
table_name = 'bitmart_eth_1s'
|
||
|
||
|
||
def create_kline_model(step: int):
|
||
"""
|
||
动态创建K线数据模型
|
||
:param step: K线周期（分钟）
|
||
:return: Model类
|
||
"""
|
||
suffix = KLINE_CONFIGS.get(step, f'{step}m')
|
||
tbl_name = f'bitmart_eth_{suffix}'
|
||
|
||
# 使用 type() 动态创建类，避免闭包问题
|
||
attrs = {
|
||
'id': BigIntegerField(primary_key=True),
|
||
'open': FloatField(null=True),
|
||
'high': FloatField(null=True),
|
||
'low': FloatField(null=True),
|
||
'close': FloatField(null=True),
|
||
}
|
||
|
||
# 创建 Meta 类
|
||
meta_attrs = {
|
||
'database': db,
|
||
'table_name': tbl_name,
|
||
}
|
||
Meta = type('Meta', (), meta_attrs)
|
||
attrs['Meta'] = Meta
|
||
|
||
# 动态创建 Model 类
|
||
model_name = f'BitMartETH{suffix.upper()}'
|
||
KlineModel = type(model_name, (Model,), attrs)
|
||
|
||
return KlineModel
|
||
|
||
|
||
class BitMartMultiKlineCollector:
|
||
"""多周期K线数据抓取器"""
|
||
|
||
def __init__(self):
|
||
self.api_key = "a0fb7b98464fd9bcce67e7c519d58ec10d0c38a8"
|
||
self.secret_key = "4eaeba78e77aeaab1c2027f846a276d164f264a44c2c1bb1c5f3be50c8de1ca5"
|
||
self.memo = "数据抓取"
|
||
self.contract_symbol = "ETHUSDT"
|
||
self.contractAPI = APIContract(self.api_key, self.secret_key, self.memo, timeout=(5, 15))
|
||
|
||
# 存储各周期的模型
|
||
self.models = {}
|
||
|
||
# 初始化数据库连接和表
|
||
self._init_database()
|
||
|
||
def _init_database(self):
|
||
"""初始化数据库，创建所有周期的表"""
|
||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
db.connect(reuse_if_open=True)
|
||
|
||
for step in KLINE_CONFIGS.keys():
|
||
model = create_kline_model(step)
|
||
self.models[step] = model
|
||
# 创建表（如果不存在）
|
||
db.create_tables([model], safe=True)
|
||
logger.info(f"初始化表: {model._meta.table_name}")
|
||
|
||
# 创建成交记录表和秒级K线表
|
||
db.create_tables([BitMartETHTrades, BitMartETHSecond], safe=True)
|
||
logger.info(f"初始化表: bitmart_eth_trades (成交记录)")
|
||
logger.info(f"初始化表: bitmart_eth_1s (秒级K线)")
|
||
|
||
def get_db_time_range(self, step: int):
|
||
"""
|
||
获取数据库中已有数据的时间范围
|
||
:param step: K线周期
|
||
:return: (earliest_ts, latest_ts) 毫秒时间戳，无数据返回 (None, None)
|
||
"""
|
||
model = self.models.get(step)
|
||
if not model:
|
||
return None, None
|
||
|
||
try:
|
||
# 获取最早记录
|
||
earliest = model.select(fn.MIN(model.id)).scalar()
|
||
# 获取最新记录
|
||
latest = model.select(fn.MAX(model.id)).scalar()
|
||
return earliest, latest
|
||
except Exception as e:
|
||
logger.error(f"查询数据库时间范围异常: {e}")
|
||
return None, None
|
||
|
||
def get_klines(self, step: int, start_time: int, end_time: int, max_retries: int = 3):
|
||
"""
|
||
获取K线数据（带重试）
|
||
:param step: K线周期（分钟）
|
||
:param start_time: 开始时间戳（秒级）
|
||
:param end_time: 结束时间戳（秒级）
|
||
:param max_retries: 最大重试次数
|
||
:return: K线数据列表
|
||
"""
|
||
for attempt in range(max_retries):
|
||
try:
|
||
start_time = int(start_time)
|
||
end_time = int(end_time)
|
||
|
||
response = self.contractAPI.get_kline(
|
||
contract_symbol=self.contract_symbol,
|
||
step=step,
|
||
start_time=start_time,
|
||
end_time=end_time
|
||
)[0]
|
||
|
||
if response['code'] != 1000:
|
||
logger.warning(f"API返回错误 (尝试 {attempt+1}/{max_retries}): {response}")
|
||
if attempt < max_retries - 1:
|
||
time.sleep(1)
|
||
continue
|
||
return []
|
||
|
||
klines = response.get('data', [])
|
||
formatted = []
|
||
for k in klines:
|
||
timestamp_ms = int(k["timestamp"]) * 1000
|
||
formatted.append({
|
||
'id': timestamp_ms,
|
||
'open': float(k["open_price"]),
|
||
'high': float(k["high_price"]),
|
||
'low': float(k["low_price"]),
|
||
'close': float(k["close_price"])
|
||
})
|
||
|
||
formatted.sort(key=lambda x: x['id'])
|
||
return formatted
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取K线异常 (尝试 {attempt+1}/{max_retries}): {e}")
|
||
if attempt < max_retries - 1:
|
||
time.sleep(2)
|
||
continue
|
||
return []
|
||
|
||
return []
|
||
|
||
def save_klines(self, step: int, klines: list):
|
||
"""
|
||
保存K线数据到数据库
|
||
:param step: K线周期
|
||
:param klines: K线数据列表
|
||
:return: 新保存的数量
|
||
"""
|
||
model = self.models.get(step)
|
||
if not model:
|
||
logger.error(f"未找到 {step}分钟 的数据模型")
|
||
return 0
|
||
|
||
new_count = 0
|
||
for kline in klines:
|
||
try:
|
||
_, created = model.get_or_create(
|
||
id=kline['id'],
|
||
defaults={
|
||
'open': kline['open'],
|
||
'high': kline['high'],
|
||
'low': kline['low'],
|
||
'close': kline['close'],
|
||
}
|
||
)
|
||
if created:
|
||
new_count += 1
|
||
except Exception as e:
|
||
logger.error(f"保存K线数据失败 {kline['id']}: {e}")
|
||
|
||
return new_count
|
||
|
||
def get_batch_seconds(self, step: int):
|
||
"""根据周期获取合适的批次大小"""
|
||
if step == 1:
|
||
return 3600 * 4 # 1分钟: 每次4小时
|
||
elif step == 3:
|
||
return 3600 * 8 # 3分钟: 每次8小时
|
||
elif step == 5:
|
||
return 3600 * 12 # 5分钟: 每次12小时
|
||
elif step == 15:
|
||
return 3600 * 24 # 15分钟: 每次1天
|
||
elif step == 30:
|
||
return 3600 * 48 # 30分钟: 每次2天
|
||
else:
|
||
return 3600 * 72 # 1小时: 每次3天
|
||
|
||
def collect_period_range(self, step: int, target_start: int, target_end: int):
|
||
"""
|
||
抓取指定时间范围的K线数据（支持断点续传）
|
||
:param step: K线周期（分钟）
|
||
:param target_start: 目标开始时间戳（秒）
|
||
:param target_end: 目标结束时间戳（秒）
|
||
:return: 保存的总数量
|
||
"""
|
||
suffix = KLINE_CONFIGS.get(step, f'{step}m')
|
||
batch_seconds = self.get_batch_seconds(step)
|
||
|
||
# 获取数据库已有数据范围
|
||
db_earliest, db_latest = self.get_db_time_range(step)
|
||
|
||
if db_earliest and db_latest:
|
||
db_earliest_sec = db_earliest // 1000
|
||
db_latest_sec = db_latest // 1000
|
||
logger.info(f"[{suffix}] 数据库已有数据: "
|
||
f"{time.strftime('%Y-%m-%d %H:%M', time.localtime(db_earliest_sec))} ~ "
|
||
f"{time.strftime('%Y-%m-%d %H:%M', time.localtime(db_latest_sec))}")
|
||
else:
|
||
db_earliest_sec = None
|
||
db_latest_sec = None
|
||
logger.info(f"[{suffix}] 数据库暂无数据")
|
||
|
||
total_saved = 0
|
||
|
||
# === 第一阶段：向前抓取历史数据（从数据库最早记录向前，直到 target_start）===
|
||
if db_earliest_sec:
|
||
backward_end = db_earliest_sec
|
||
else:
|
||
backward_end = target_end
|
||
|
||
if backward_end > target_start:
|
||
logger.info(f"[{suffix}] === 开始向前抓取历史数据 ===")
|
||
total_backward = backward_end - target_start
|
||
|
||
current_end = backward_end
|
||
fail_count = 0
|
||
max_fail = 5
|
||
|
||
while current_end > target_start and fail_count < max_fail:
|
||
current_start = max(current_end - batch_seconds, target_start)
|
||
|
||
# 计算进度
|
||
progress = (backward_end - current_end) / total_backward * 100 if total_backward > 0 else 0
|
||
start_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(current_start))
|
||
end_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(current_end))
|
||
|
||
klines = self.get_klines(step, current_start, current_end)
|
||
if klines:
|
||
saved = self.save_klines(step, klines)
|
||
total_saved += saved
|
||
logger.info(f"[{suffix}] ← 历史 {start_str} ~ {end_str} | "
|
||
f"获取 {len(klines)} 条, 新增 {saved} 条 | 进度 {progress:.1f}%")
|
||
fail_count = 0
|
||
else:
|
||
fail_count += 1
|
||
logger.warning(f"[{suffix}] ← 历史 {start_str} 无数据 (连续失败 {fail_count}/{max_fail})")
|
||
if fail_count >= max_fail:
|
||
earliest_date = time.strftime('%Y-%m-%d', time.localtime(current_end))
|
||
logger.warning(f"[{suffix}] 已达到API历史数据限制，最早可获取: {earliest_date}")
|
||
break
|
||
|
||
current_end = current_start
|
||
time.sleep(0.3)
|
||
|
||
# === 第二阶段：向后抓取最新数据（从数据库最新记录向后，直到 target_end）===
|
||
if db_latest_sec:
|
||
forward_start = db_latest_sec
|
||
else:
|
||
# 如果没有数据，从第一阶段结束的地方开始
|
||
forward_start = target_start
|
||
|
||
if forward_start < target_end:
|
||
logger.info(f"[{suffix}] === 开始向后抓取最新数据 ===")
|
||
total_forward = target_end - forward_start
|
||
|
||
current_start = forward_start
|
||
fail_count = 0
|
||
max_fail = 3
|
||
|
||
while current_start < target_end and fail_count < max_fail:
|
||
current_end = min(current_start + batch_seconds, target_end)
|
||
|
||
# 计算进度
|
||
progress = (current_start - forward_start) / total_forward * 100 if total_forward > 0 else 0
|
||
start_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(current_start))
|
||
end_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(current_end))
|
||
|
||
klines = self.get_klines(step, current_start, current_end)
|
||
if klines:
|
||
saved = self.save_klines(step, klines)
|
||
total_saved += saved
|
||
logger.info(f"[{suffix}] → 最新 {start_str} ~ {end_str} | "
|
||
f"获取 {len(klines)} 条, 新增 {saved} 条 | 进度 {progress:.1f}%")
|
||
fail_count = 0
|
||
else:
|
||
fail_count += 1
|
||
logger.warning(f"[{suffix}] → 最新 {start_str} 无数据 (失败 {fail_count}/{max_fail})")
|
||
|
||
current_start = current_end
|
||
time.sleep(0.3)
|
||
|
||
# 统计最终数据范围
|
||
final_earliest, final_latest = self.get_db_time_range(step)
|
||
if final_earliest and final_latest:
|
||
logger.success(f"[{suffix}] 抓取完成！本次新增 {total_saved} 条 | 数据范围: "
|
||
f"{time.strftime('%Y-%m-%d', time.localtime(final_earliest//1000))} ~ "
|
||
f"{time.strftime('%Y-%m-%d', time.localtime(final_latest//1000))}")
|
||
else:
|
||
logger.success(f"[{suffix}] 抓取完成！本次新增 {total_saved} 条")
|
||
|
||
return total_saved
|
||
|
||
def collect_from_date(self, start_date: str, periods: list = None):
|
||
"""
|
||
从指定日期抓取到当前时间
|
||
:param start_date: 起始日期 'YYYY-MM-DD'
|
||
:param periods: 要抓取的周期列表，如 [1, 5, 15]，默认全部
|
||
"""
|
||
if periods is None:
|
||
periods = list(KLINE_CONFIGS.keys())
|
||
|
||
# 计算时间范围
|
||
start_dt = datetime.datetime.strptime(start_date, '%Y-%m-%d')
|
||
target_start = int(start_dt.timestamp())
|
||
target_end = int(time.time())
|
||
|
||
start_str = start_dt.strftime('%Y-%m-%d')
|
||
end_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M')
|
||
|
||
logger.info(f"{'='*60}")
|
||
logger.info(f"目标时间范围: {start_str} ~ {end_str}")
|
||
logger.info(f"抓取周期: {[KLINE_CONFIGS[p] for p in periods]}")
|
||
logger.info(f"{'='*60}")
|
||
|
||
results = {}
|
||
for step in periods:
|
||
if step not in KLINE_CONFIGS:
|
||
logger.warning(f"不支持的周期: {step}分钟，跳过")
|
||
continue
|
||
|
||
logger.info(f"\n{'='*60}")
|
||
logger.info(f"开始抓取 {KLINE_CONFIGS[step]} K线")
|
||
logger.info(f"{'='*60}")
|
||
|
||
saved = self.collect_period_range(step, target_start, target_end)
|
||
results[KLINE_CONFIGS[step]] = saved
|
||
|
||
time.sleep(1) # 不同周期之间间隔
|
||
|
||
# 打印总结
|
||
logger.info(f"\n{'='*60}")
|
||
logger.info("所有周期抓取完成！统计：")
|
||
for period, count in results.items():
|
||
logger.info(f" {period}: 新增 {count} 条")
|
||
logger.info(f"{'='*60}")
|
||
|
||
return results
|
||
|
||
def get_stats(self):
|
||
"""获取各周期数据统计"""
|
||
logger.info(f"\n{'='*60}")
|
||
logger.info("数据库统计:")
|
||
logger.info(f"{'='*60}")
|
||
|
||
for step, model in self.models.items():
|
||
suffix = KLINE_CONFIGS.get(step, f'{step}m')
|
||
try:
|
||
count = model.select().count()
|
||
earliest, latest = self.get_db_time_range(step)
|
||
if earliest and latest:
|
||
earliest_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(earliest//1000))
|
||
latest_str = time.strftime('%Y-%m-%d %H:%M', time.localtime(latest//1000))
|
||
logger.info(f" {suffix:>4}: {count:>8} 条 | {earliest_str} ~ {latest_str}")
|
||
else:
|
||
logger.info(f" {suffix:>4}: {count:>8} 条")
|
||
except Exception as e:
|
||
logger.error(f" {suffix}: 查询失败 - {e}")
|
||
|
||
# 成交记录统计
|
||
try:
|
||
trades_count = BitMartETHTrades.select().count()
|
||
if trades_count > 0:
|
||
earliest_trade = BitMartETHTrades.select(fn.MIN(BitMartETHTrades.timestamp)).scalar()
|
||
latest_trade = BitMartETHTrades.select(fn.MAX(BitMartETHTrades.timestamp)).scalar()
|
||
earliest_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(earliest_trade//1000))
|
||
latest_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(latest_trade//1000))
|
||
logger.info(f"trades: {trades_count:>8} 条 | {earliest_str} ~ {latest_str}")
|
||
else:
|
||
logger.info(f"trades: {trades_count:>8} 条")
|
||
except Exception as e:
|
||
logger.error(f"trades: 查询失败 - {e}")
|
||
|
||
# 秒级K线统计
|
||
try:
|
||
second_count = BitMartETHSecond.select().count()
|
||
if second_count > 0:
|
||
earliest_sec = BitMartETHSecond.select(fn.MIN(BitMartETHSecond.id)).scalar()
|
||
latest_sec = BitMartETHSecond.select(fn.MAX(BitMartETHSecond.id)).scalar()
|
||
earliest_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(earliest_sec//1000))
|
||
latest_str = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(latest_sec//1000))
|
||
logger.info(f" 1s: {second_count:>8} 条 | {earliest_str} ~ {latest_str}")
|
||
else:
|
||
logger.info(f" 1s: {second_count:>8} 条")
|
||
except Exception as e:
|
||
logger.error(f" 1s: 查询失败 - {e}")
|
||
|
||
logger.info(f"{'='*60}")
|
||
|
||
# ==================== 秒级数据相关方法 ====================
|
||
|
||
def get_trades(self, limit: int = 100):
|
||
"""
|
||
获取最近成交记录
|
||
:param limit: 获取条数
|
||
:return: 成交记录列表
|
||
"""
|
||
try:
|
||
response = self.contractAPI.get_trades(
|
||
contract_symbol=self.contract_symbol,
|
||
)[0]
|
||
|
||
if response['code'] != 1000:
|
||
logger.error(f"获取成交记录失败: {response}")
|
||
return []
|
||
|
||
trades = response.get('data', {}).get('trades', [])
|
||
formatted = []
|
||
for t in trades:
|
||
formatted.append({
|
||
'id': int(t.get('trade_id', 0)),
|
||
'timestamp': int(t.get('create_time', 0)),
|
||
'price': float(t.get('deal_price', 0)),
|
||
'volume': float(t.get('deal_vol', 0)),
|
||
'side': int(t.get('way', 0)),
|
||
})
|
||
|
||
return formatted
|
||
except Exception as e:
|
||
logger.error(f"获取成交记录异常: {e}")
|
||
return []
|
||
|
||
def save_trades(self, trades: list):
|
||
"""保存成交记录到数据库"""
|
||
new_count = 0
|
||
for trade in trades:
|
||
try:
|
||
_, created = BitMartETHTrades.get_or_create(
|
||
id=trade['id'],
|
||
defaults={
|
||
'timestamp': trade['timestamp'],
|
||
'price': trade['price'],
|
||
'volume': trade['volume'],
|
||
'side': trade['side'],
|
||
}
|
||
)
|
||
if created:
|
||
new_count += 1
|
||
except Exception as e:
|
||
pass # 忽略重复数据
|
||
return new_count
|
||
|
||
def collect_trades_realtime(self, duration_seconds: int = 3600, interval: float = 0.3):
|
||
"""
|
||
实时持续采集成交记录（秒级数据源）
|
||
:param duration_seconds: 采集时长（秒），默认1小时
|
||
:param interval: 采集间隔（秒），默认0.3秒
|
||
"""
|
||
logger.info(f"{'='*60}")
|
||
logger.info(f"开始实时采集成交记录")
|
||
logger.info(f"时长: {duration_seconds}秒 ({duration_seconds/3600:.1f}小时)")
|
||
logger.info(f"间隔: {interval}秒")
|
||
logger.info(f"{'='*60}")
|
||
|
||
start_time = time.time()
|
||
end_time = start_time + duration_seconds
|
||
total_saved = 0
|
||
batch_count = 0
|
||
|
||
while time.time() < end_time:
|
||
trades = self.get_trades(limit=100)
|
||
if trades:
|
||
saved = self.save_trades(trades)
|
||
total_saved += saved
|
||
batch_count += 1
|
||
|
||
# 每10批显示一次进度
|
||
if batch_count % 10 == 0:
|
||
elapsed = time.time() - start_time
|
||
remaining = end_time - time.time()
|
||
latest = trades[-1]
|
||
ts_str = datetime.datetime.fromtimestamp(
|
||
latest['timestamp']/1000
|
||
).strftime('%H:%M:%S')
|
||
logger.info(f"[{ts_str}] 价格: {latest['price']:.2f} | "
|
||
f"本批新增: {saved} | 累计: {total_saved} | "
|
||
f"剩余: {remaining/60:.1f}分钟")
|
||
|
||
time.sleep(interval)
|
||
|
||
logger.success(f"采集完成！共新增 {total_saved} 条成交记录")
|
||
|
||
# 自动聚合为秒级K线
|
||
logger.info("正在将成交记录聚合为秒级K线...")
|
||
self.aggregate_trades_to_seconds()
|
||
|
||
return total_saved
|
||
|
||
def aggregate_trades_to_seconds(self, start_ts: int = None, end_ts: int = None):
|
||
"""
|
||
将成交记录聚合为秒级K线数据
|
||
:param start_ts: 开始时间戳（毫秒），默认全部
|
||
:param end_ts: 结束时间戳（毫秒），默认全部
|
||
:return: 聚合的秒级K线数量
|
||
"""
|
||
# 构建查询
|
||
query = BitMartETHTrades.select().order_by(BitMartETHTrades.timestamp)
|
||
if start_ts:
|
||
query = query.where(BitMartETHTrades.timestamp >= start_ts)
|
||
if end_ts:
|
||
query = query.where(BitMartETHTrades.timestamp <= end_ts)
|
||
|
||
# 按秒聚合
|
||
second_data = {}
|
||
trade_count = 0
|
||
|
||
for trade in query:
|
||
trade_count += 1
|
||
# 取整到秒（毫秒时间戳）
|
||
second_ts = (trade.timestamp // 1000) * 1000
|
||
|
||
if second_ts not in second_data:
|
||
second_data[second_ts] = {
|
||
'open': trade.price,
|
||
'high': trade.price,
|
||
'low': trade.price,
|
||
'close': trade.price,
|
||
'volume': trade.volume,
|
||
'trade_count': 1
|
||
}
|
||
else:
|
||
second_data[second_ts]['high'] = max(second_data[second_ts]['high'], trade.price)
|
||
second_data[second_ts]['low'] = min(second_data[second_ts]['low'], trade.price)
|
||
second_data[second_ts]['close'] = trade.price
|
||
second_data[second_ts]['volume'] += trade.volume
|
||
second_data[second_ts]['trade_count'] += 1
|
||
|
||
# 保存到数据库
|
||
saved_count = 0
|
||
for ts, ohlc in second_data.items():
|
||
try:
|
||
BitMartETHSecond.insert(
|
||
id=ts,
|
||
open=ohlc['open'],
|
||
high=ohlc['high'],
|
||
low=ohlc['low'],
|
||
close=ohlc['close'],
|
||
volume=ohlc['volume'],
|
||
trade_count=ohlc['trade_count']
|
||
).on_conflict(
|
||
conflict_target=[BitMartETHSecond.id],
|
||
update={
|
||
BitMartETHSecond.open: ohlc['open'],
|
||
BitMartETHSecond.high: ohlc['high'],
|
||
BitMartETHSecond.low: ohlc['low'],
|
||
BitMartETHSecond.close: ohlc['close'],
|
||
BitMartETHSecond.volume: ohlc['volume'],
|
||
BitMartETHSecond.trade_count: ohlc['trade_count'],
|
||
}
|
||
).execute()
|
||
saved_count += 1
|
||
except Exception as e:
|
||
logger.error(f"保存秒级K线失败 {ts}: {e}")
|
||
|
||
logger.success(f"聚合完成！{trade_count} 条成交记录 → {saved_count} 条秒级K线")
|
||
return saved_count
|
||
|
||
def get_second_klines(self, start_ts: int = None, end_ts: int = None):
|
||
"""
|
||
获取秒级K线数据
|
||
:param start_ts: 开始时间戳（毫秒）
|
||
:param end_ts: 结束时间戳（毫秒）
|
||
:return: 秒级K线列表
|
||
"""
|
||
query = BitMartETHSecond.select().order_by(BitMartETHSecond.id)
|
||
if start_ts:
|
||
query = query.where(BitMartETHSecond.id >= start_ts)
|
||
if end_ts:
|
||
query = query.where(BitMartETHSecond.id <= end_ts)
|
||
|
||
return [{
|
||
'timestamp': k.id,
|
||
'open': k.open,
|
||
'high': k.high,
|
||
'low': k.low,
|
||
'close': k.close,
|
||
'volume': k.volume,
|
||
'trade_count': k.trade_count
|
||
} for k in query]
|
||
|
||
def aggregate_trades_custom(self, interval_ms: int = 100, start_ts: int = None, end_ts: int = None):
|
||
"""
|
||
将成交记录聚合为自定义毫秒级K线数据（不保存到数据库，直接返回）
|
||
:param interval_ms: 聚合周期（毫秒），如 100=100ms, 500=500ms, 1000=1秒
|
||
:param start_ts: 开始时间戳（毫秒）
|
||
:param end_ts: 结束时间戳（毫秒）
|
||
:return: K线列表 [{'timestamp', 'open', 'high', 'low', 'close', 'volume', 'trade_count'}, ...]
|
||
"""
|
||
# 构建查询
|
||
query = BitMartETHTrades.select().order_by(BitMartETHTrades.timestamp)
|
||
if start_ts:
|
||
query = query.where(BitMartETHTrades.timestamp >= start_ts)
|
||
if end_ts:
|
||
query = query.where(BitMartETHTrades.timestamp <= end_ts)
|
||
|
||
# 按指定间隔聚合
|
||
interval_data = {}
|
||
trade_count = 0
|
||
|
||
for trade in query:
|
||
trade_count += 1
|
||
# 取整到指定间隔
|
||
interval_ts = (trade.timestamp // interval_ms) * interval_ms
|
||
|
||
if interval_ts not in interval_data:
|
||
interval_data[interval_ts] = {
|
||
'open': trade.price,
|
||
'high': trade.price,
|
||
'low': trade.price,
|
||
'close': trade.price,
|
||
'volume': trade.volume,
|
||
'trade_count': 1
|
||
}
|
||
else:
|
||
interval_data[interval_ts]['high'] = max(interval_data[interval_ts]['high'], trade.price)
|
||
interval_data[interval_ts]['low'] = min(interval_data[interval_ts]['low'], trade.price)
|
||
interval_data[interval_ts]['close'] = trade.price
|
||
interval_data[interval_ts]['volume'] += trade.volume
|
||
interval_data[interval_ts]['trade_count'] += 1
|
||
|
||
# 转换为列表
|
||
result = []
|
||
for ts, ohlc in sorted(interval_data.items()):
|
||
result.append({
|
||
'timestamp': ts,
|
||
'datetime': datetime.datetime.fromtimestamp(ts/1000).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
|
||
'open': ohlc['open'],
|
||
'high': ohlc['high'],
|
||
'low': ohlc['low'],
|
||
'close': ohlc['close'],
|
||
'volume': ohlc['volume'],
|
||
'trade_count': ohlc['trade_count']
|
||
})
|
||
|
||
logger.info(f"聚合完成: {trade_count} 条成交记录 → {len(result)} 条 {interval_ms}ms K线")
|
||
return result
|
||
|
||
def get_raw_trades(self, start_ts: int = None, end_ts: int = None, limit: int = None):
|
||
"""
|
||
获取原始成交记录（逐笔数据，毫秒级）
|
||
:param start_ts: 开始时间戳（毫秒）
|
||
:param end_ts: 结束时间戳（毫秒）
|
||
:param limit: 最大返回条数
|
||
:return: 成交记录列表
|
||
"""
|
||
query = BitMartETHTrades.select().order_by(BitMartETHTrades.timestamp)
|
||
if start_ts:
|
||
query = query.where(BitMartETHTrades.timestamp >= start_ts)
|
||
if end_ts:
|
||
query = query.where(BitMartETHTrades.timestamp <= end_ts)
|
||
if limit:
|
||
query = query.limit(limit)
|
||
|
||
return [{
|
||
'id': t.id,
|
||
'timestamp': t.timestamp,
|
||
'datetime': datetime.datetime.fromtimestamp(t.timestamp/1000).strftime('%Y-%m-%d %H:%M:%S.%f')[:-3],
|
||
'price': t.price,
|
||
'volume': t.volume,
|
||
'side': '买' if t.side == 1 else '卖'
|
||
} for t in query]
|
||
|
||
def close(self):
|
||
"""关闭数据库连接"""
|
||
if not db.is_closed():
|
||
db.close()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
collector = BitMartMultiKlineCollector()
|
||
|
||
try:
|
||
# 查看当前数据统计
|
||
collector.get_stats()
|
||
|
||
# ============ 选择要执行的任务 ============
|
||
|
||
# 任务1: 抓取K线数据（1分钟~1小时周期）
|
||
# 从 2025-01-01 抓取到当前时间（支持断点续传）
|
||
collector.collect_from_date(
|
||
start_date='2010-01-01',
|
||
periods=[1, 3, 5, 15, 30, 60] # 所有周期
|
||
)
|
||
|
||
# 任务2: 实时采集秒级数据（成交记录）
|
||
# 注意: 秒级数据只能实时采集，无法获取历史
|
||
# collector.collect_trades_realtime(
|
||
# duration_seconds=3600, # 采集1小时
|
||
# interval=0.3 # 每0.3秒请求一次
|
||
# )
|
||
|
||
# 任务3: 将已采集的成交记录聚合为秒级K线
|
||
# collector.aggregate_trades_to_seconds()
|
||
|
||
# 再次查看统计
|
||
collector.get_stats()
|
||
|
||
finally:
|
||
collector.close()
|