113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
||
"""整点后5秒抓取 BTC 1h K线（至少512根，且包含最新时间）"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
|
||
import pandas as pd
|
||
import pytz
|
||
|
||
# 直接运行脚本时，确保可以从项目根目录导入 data 包
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from data.fetcher import BinanceDataFetcher
|
||
|
||
|
||
def get_next_run_time(delay_seconds: int = 5) -> datetime:
|
||
"""计算下一次执行时间：下一个整点 + delay_seconds。"""
|
||
now_utc = datetime.now(pytz.UTC)
|
||
next_hour = (now_utc + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
|
||
return next_hour + timedelta(seconds=delay_seconds)
|
||
|
||
|
||
def fetch_min_klines_with_latest(
|
||
fetcher: BinanceDataFetcher,
|
||
min_klines: int = 512,
|
||
latest_check_retries: int = 8,
|
||
latest_check_interval: int = 2,
|
||
) -> pd.DataFrame:
|
||
"""
|
||
拉取 BTC 1h K线，确保：
|
||
1) 数量不少于 min_klines
|
||
2) 包含当前小时的最新时间（UTC 整点）
|
||
"""
|
||
now_utc = datetime.now(pytz.UTC)
|
||
expected_latest_ts = now_utc.replace(minute=0, second=0, microsecond=0).replace(tzinfo=None)
|
||
|
||
lookback_days = 35
|
||
for attempt in range(latest_check_retries):
|
||
start_date = (datetime.now(pytz.UTC) - timedelta(days=lookback_days)).strftime("%Y-%m-%d")
|
||
|
||
# 使用 WebSocket 拉取，避免仅命中缓存导致最新K线不更新。
|
||
df = fetcher.fetch_ohlcv(
|
||
symbol="BTC",
|
||
timeframe="1h",
|
||
start_date=start_date,
|
||
end_date=None,
|
||
use_cache=False,
|
||
max_retries=3,
|
||
)
|
||
|
||
if len(df) < min_klines:
|
||
lookback_days += 7
|
||
print(
|
||
f"[WARN] 当前仅获取 {len(df)} 根K线，小于 {min_klines}，扩大回看窗口到 {lookback_days} 天后重试。"
|
||
)
|
||
continue
|
||
|
||
latest_ts = df["timestamp"].max().to_pydatetime().replace(tzinfo=None)
|
||
if latest_ts >= expected_latest_ts:
|
||
result = df.tail(min_klines).copy().reset_index(drop=True)
|
||
print(
|
||
f"[OK] 获取成功: {len(result)} 根 | 最新时间: {latest_ts} UTC | 期望最新时间: {expected_latest_ts} UTC"
|
||
)
|
||
return result
|
||
|
||
if attempt < latest_check_retries - 1:
|
||
print(
|
||
f"[WAIT] 最新K线未到位，当前最新: {latest_ts} UTC，期望: {expected_latest_ts} UTC，"
|
||
f"{latest_check_interval}秒后重试({attempt + 1}/{latest_check_retries})。"
|
||
)
|
||
time.sleep(latest_check_interval)
|
||
|
||
raise RuntimeError("多次重试后仍未拿到包含最新时间的BTC 1h K线数据")
|
||
|
||
|
||
def run_once(min_klines: int = 512) -> None:
|
||
"""执行一次抓取任务。"""
|
||
fetcher = BinanceDataFetcher()
|
||
data = fetch_min_klines_with_latest(fetcher=fetcher, min_klines=min_klines)
|
||
|
||
print("-" * 80)
|
||
print(f"抓取完成，返回K线数量: {len(data)}")
|
||
print(f"时间范围: {data['timestamp'].min()} -> {data['timestamp'].max()} (UTC)")
|
||
print(data.tail(3)[["timestamp", "open", "high", "low", "close", "volume"]])
|
||
print("-" * 80)
|
||
|
||
|
||
def run_forever(min_klines: int = 512, delay_seconds: int = 5) -> None:
|
||
"""常驻运行：每个整点延迟 delay_seconds 后执行抓取。"""
|
||
while True:
|
||
next_run = get_next_run_time(delay_seconds=delay_seconds)
|
||
sleep_seconds = max(0.0, (next_run - datetime.now(pytz.UTC)).total_seconds())
|
||
|
||
print(f"[SCHEDULE] 当前UTC: {datetime.now(pytz.UTC)}，下次执行: {next_run}，等待 {sleep_seconds:.1f} 秒")
|
||
time.sleep(sleep_seconds)
|
||
|
||
try:
|
||
run_once(min_klines=min_klines)
|
||
except Exception as exc:
|
||
print(f"[ERROR] 本轮抓取失败: {exc}")
|
||
|
||
|
||
def main() -> None:
|
||
"""脚本入口。"""
|
||
# 默认常驻运行，如需只执行一次，可把这里改成 run_once(min_klines=512)
|
||
run_forever(min_klines=512, delay_seconds=5)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|