Bithumb/scripts/00_download_candles.py

#!/usr/bin/env python3
"""사전: 빗썸 캔들 수집 — 기본: 전체 인터벌 증분 갱신, --full: 전체 인터벌 풀 다운."""

from __future__ import annotations

import argparse
import logging
import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
SRC = ROOT / "src"
if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))

from dataclasses import replace

from bithumb.config import load_settings
from bithumb.data.candle_store import CandleStore
from bithumb.data.downloader import CandleDownloader
from bithumb.data.intervals import INTERVAL_1MIN, estimate_download_requests, interval_label


def _configure_logging(verbose: bool) -> None:
    """로깅 레벨을 설정한다."""
    level = logging.DEBUG if verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )


def main() -> int:
    """CLI 진입점."""
    parser = argparse.ArgumentParser(
        description="빗썸 캔들 데이터 수집 (DOWNLOAD_INTERVALS 전체, 1분봉 포함)",
    )
    parser.add_argument(
        "--full",
        action="store_true",
        help="전체 인터벌을 DOWNLOAD_DAYS 구간만큼 역방향 풀 다운 (최초 1회·재구축)",
    )
    parser.add_argument(
        "--days",
        type=int,
        default=None,
        help="풀 다운(--full) 또는 DB 비어 있을 때 목표 일수 (기본: DOWNLOAD_DAYS)",
    )
    parser.add_argument(
        "--intervals",
        type=str,
        default=None,
        help="(고급) 쉼표 구분 인터벌만 수집. 기본: .env DOWNLOAD_INTERVALS 전체",
    )
    parser.add_argument(
        "--include-1min",
        action="store_true",
        help="1분봉(1)을 기존 DOWNLOAD_INTERVALS에 추가하여 수집",
    )
    parser.add_argument("-v", "--verbose", action="store_true", help="디버그 로그")
    args = parser.parse_args()

    _configure_logging(args.verbose)
    settings = load_settings()

    if args.intervals:
        settings = replace(
            settings,
            download_intervals=[
                int(x.strip()) for x in args.intervals.split(",") if x.strip()
            ],
        )
    elif args.include_1min and INTERVAL_1MIN not in settings.download_intervals:
        settings = replace(
            settings,
            download_intervals=sorted({*settings.download_intervals, INTERVAL_1MIN}),
        )

    days = args.days or settings.download_days
    mode_label = "full" if args.full else "incremental"
    log = logging.getLogger(__name__)
    log.info(
        "대상=%s DB=%s mode=%s days=%s intervals=%s",
        settings.market,
        settings.db_path,
        mode_label,
        days,
        settings.download_intervals,
    )
    for interval in settings.download_intervals:
        est = estimate_download_requests(interval, days, batch_size=settings.candle_count)
        log.info(
            "예상 API 요청: %s ≈ %s회 (sleep %.2fs)",
            interval_label(interval),
            est,
            settings.request_sleep_sec,
        )

    store = CandleStore(settings.db_path)
    try:
        for interval in settings.download_intervals:
            if args.full:
                est = estimate_download_requests(interval, days, batch_size=settings.candle_count)
                log.info(
                    "예상 API 요청: %s ≈ %s회 (풀 다운, sleep %.2fs)",
                    interval_label(interval),
                    est,
                    settings.request_sleep_sec,
                )
            else:
                _, _, db_max = store.get_range(settings.symbol, interval)
                if db_max is None:
                    est = estimate_download_requests(interval, days, batch_size=settings.candle_count)
                    log.info(
                        "예상 API 요청: %s ≈ %s회 (DB 없음 → 풀 다운)",
                        interval_label(interval),
                        est,
                    )
                else:
                    gap_days = max(1, (datetime.now() - db_max).days + 1)
                    est = estimate_download_requests(interval, gap_days, batch_size=settings.candle_count)
                    log.info(
                        "예상 API 요청: %s ≈ %s회 (증분, DB=%s, 갭≈%s일)",
                        interval_label(interval),
                        est,
                        db_max.strftime("%Y-%m-%d %H:%M:%S"),
                        gap_days,
                    )

        downloader = CandleDownloader(settings)
        results = downloader.download_all(store, days=days, full=args.full)

        print(f"\n=== 수집 완료 ({mode_label}) ===")
        for result in results:
            count, min_dt, max_dt = store.get_range(settings.symbol, result.interval_min)
            min_s = min_dt.strftime("%Y-%m-%d %H:%M:%S") if min_dt else "-"
            max_s = max_dt.strftime("%Y-%m-%d %H:%M:%S") if max_dt else "-"
            if result.mode == "uptodate":
                flag = "UPTODATE"
            elif result.reached_target:
                flag = "OK"
            else:
                flag = "PARTIAL"
            label = interval_label(result.interval_min)
            print(
                f"[{flag}] {label} ({result.interval_min}) mode={result.mode} | "
                f"requests={result.requests} upsert={result.saved_rows} "
                f"db_rows={count} range={min_s} ~ {max_s}"
            )
    finally:
        store.close()

    return 0


if __name__ == "__main__":
    raise SystemExit(main())