Bithumb/scripts/3_run_rsi_dca.py

#!/usr/bin/env python3
"""RSI DCA 현물 정액 매수 tick (15m RSI 30/35 상향 돌파, 매도 없음).

  python scripts/3_run_rsi_dca.py                 # RSI_DCA_MODE(.env, 기본 paper) 1회 tick
  python scripts/3_run_rsi_dca.py --mode paper --loop 60
  python scripts/3_run_rsi_dca.py --mode live     # 실제 주문 — 사용자 직접 실행
  python scripts/3_run_rsi_dca.py --status        # 상태만 출력
"""

from __future__ import annotations

import argparse
import json
import logging
import sys
import time
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 bithumb.config import load_settings  # noqa: E402
from bithumb.operations.rsi_dca_engine import load_state  # noqa: E402
from bithumb.operations.rsi_dca_runner import RsiDcaRunner  # noqa: E402


def _configure_logging(verbose: bool) -> None:
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )


def _print_status(settings, mode: str) -> None:
    st = load_state(settings.rsi_dca_state_json, mode)
    d = st.get("daily") or {}
    t = st.get("totals") or {}
    print(f"strategy={st.get('strategy')} mode={st.get('mode')} last_run={st.get('last_run_at')}")
    print(f"daily {d.get('date')} spent={float(d.get('spent_krw') or 0):,.0f}/{settings.rsi_dca_daily_max_krw:,.0f} count={d.get('count')}")
    print(f"totals spent={float(t.get('spent_krw') or 0):,.0f} count={t.get('count')} trades={len(st.get('trades') or [])}")
    for sym, s in (st.get("symbols") or {}).items():
        rsi = s.get("last_rsi")
        print(f"  {sym:4s} init={s.get('initialized')} cursor={s.get('last_confirm_time')} "
              f"rsi={rsi if rsi is None else round(rsi, 1)} buys={s.get('buy_count')} spent={float(s.get('spent_krw') or 0):,.0f}")


def main() -> int:
    parser = argparse.ArgumentParser(description="Bithumb RSI DCA 정액 매수 tick")
    parser.add_argument("--mode", choices=("paper", "live"), default=None,
                        help="기본: .env RSI_DCA_MODE (paper)")
    parser.add_argument("--loop", type=int, default=0, metavar="SEC")
    parser.add_argument("--status", action="store_true", help="상태 출력만")
    parser.add_argument("-v", "--verbose", action="store_true")
    args = parser.parse_args()
    _configure_logging(args.verbose)

    settings = load_settings()
    mode = (args.mode or settings.rsi_dca_mode or "paper").lower()
    if args.status:
        _print_status(settings, mode)
        return 0

    if mode == "live":
        print("경고: live — 실제 주문 발생 (RSI 정액 매수, 매도 없음)")
    levels = ", ".join(f"RSI {lv:g}↑ {krw:,.0f}원" for lv, krw in settings.rsi_dca_levels)
    print(
        f"rsi_dca {mode} | symbols={settings.rsi_dca_symbols} | {settings.rsi_dca_interval_min}m "
        f"RSI({settings.rsi_dca_period}) | {levels} | 일 상한 {settings.rsi_dca_daily_max_krw:,.0f}원"
    )

    def _once() -> dict:
        runner = RsiDcaRunner(settings, mode=mode)
        report = runner.tick()
        if not report.get("ok"):
            print(f"  skip: {report.get('note')}")
            return report
        d = report.get("daily") or {}
        for row in report.get("results") or []:
            if "error" in row:
                print(f"  {row['symbol']}: ERROR {row['error']}")
                continue
            rsi = row.get("rsi")
            print(f"  {row['symbol']:4s} rsi={'-' if rsi is None else f'{rsi:.1f}':>5s} fills={row['fills']} {row['note']}")
        print(f"  daily {d.get('date')} spent={float(d.get('spent_krw') or 0):,.0f} "
              f"remaining={float(d.get('remaining_krw') or 0):,.0f} kill_switch={report.get('kill_switch')}")
        return report

    if args.loop <= 0:
        _once()
        return 0
    while True:
        try:
            _once()
        except KeyboardInterrupt:
            print("\n종료")
            return 0
        except Exception:  # noqa: BLE001
            logging.exception("rsi_dca loop tick failed")
        time.sleep(args.loop)


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