파일 시그널로 텔레그램 봇 일시정지 구현하기 | 토스 자동매매 개발기 #16

🌐 Read in English

지난 글에서 읽기전용 대시보드를 붙이고 나니, 자연스럽게 다음 욕심이 생겼다. “보기”만 되는 게 아니라 “멈추기”도 되면 좋겠다는 거였다. 그런데 이건 지금까지 해온 작업들과는 완전히 다른 문제였다.

왜 지금까지와 다른 문제였나

대시보드는 DB에 쌓인 데이터를 읽기만 하면 됐다. 반면 텔레그램 봇의 /pause 명령어는 이미 돌아가고 있는 auto_trader.py의 감시 루프를 외부에서 멈춰야 한다. 봇 프로세스와 감시 루프 프로세스는 서로 존재조차 몰랐다. 둘 사이에 뭔가 다리를 놓아야 했다.

세 가지 후보 중 파일 시그널을 고른 이유

고민한 방식은 세 가지였다.

1. 봇과 감시 루프를 아예 한 프로세스로 합치기
2. 파일 하나를 신호로 삼아 존재 여부만 확인하기
3. DB에 명령 큐 테이블을 만들어 폴링하기

1번은 구조를 너무 많이 뒤엎어야 했고, 3번은 지금 규모에 비해 과했다. 결국 파일 기반 신호 전달을 골랐다. 신호 파일이 있으면 일시정지, 없으면 정상 동작 — 이보다 단순할 수가 없었다.

여기서 하나 더 확실히 정해둔 게 있다. /pause가 멈추는 건 시세 조회와 매수 판단이 도는 감시 루프 전체이지, DRY_RUN이나 목표가 같은 3중 안전장치 값은 절대 건드리지 않는다는 원칙이다.

engine_control.py — 상태 판단은 최대한 단순하게

핵심은 아래처럼 신호 파일 하나의 존재 여부다.

class EngineControl:
    def __init__(self, signal_path="pause.signal"):
        self.signal_path = signal_path

    def pause(self):
        with open(self.signal_path, "w") as f:
            f.write("paused")

    def resume(self):
        if os.path.exists(self.signal_path):
            os.remove(self.signal_path)

    def is_paused(self):
        return os.path.exists(self.signal_path)

auto_trader.py — 일시정지 중에는 아예 건너뛰기

감시 루프 안에서는 매 주기 is_paused()만 확인하면 된다. 다만 매번 알림을 쏘면 스팸이 되니, 일시정지/재개로 상태가 바뀌는 그 순간에만 알림을 보내도록 했다.

if engine_control.is_paused():
    if not was_paused:
        notify("⏸ 감시 루프 일시정지됨 (000660)")
        was_paused = True
    time.sleep(WATCH_INTERVAL)
    continue

if was_paused:
    notify("▶ 감시 루프 재개됨")
    was_paused = False

price = get_current_price("000660")
# 이하 매수 판단 로직은 기존과 동일

bot_commands.py — API 호출 없는 순수 함수로

텔레그램 라이브러리를 아직 연결하지 않았기 때문에, /status·/holdings가 반환할 텍스트를 조립하는 부분만 먼저 순수 함수로 뗐다. 이렇게 하면 실제 봇 없이도 완전히 테스트할 수 있다.

def get_status_text(is_paused, trade_count, alert_count):
    mode = "일시정지" if is_paused else "정상 동작"
    return (
        f"상태: {mode}\n"
        f"누적 거래: {trade_count}건\n"
        f"누적 알림: {alert_count}건"
    )

def get_holdings_text(holdings):
    # holdings 예: [{"ticker": "005930", "qty": 3, "avg_price": 1}]
    lines = [f"{h['ticker']} {h['qty']}주 (평단 {h['avg_price']})" for h in holdings]
    return "\n".join(lines) if lines else "보유 종목 없음"

검증 — “정말로 멈췄는지”를 직접 확인하기

가장 신경 쓴 테스트는 test_auto_trader.py였다. 단순히 코드가 돌아가는지가 아니라, 일시정지 중엔 get_current_price()가 단 한 번도 호출되지 않는지를 확인해야 의미가 있었다.

def test_price_check_skipped_while_paused(mocker):
    mock_price = mocker.patch("auto_trader.get_current_price")
    mock_is_paused = mocker.patch(
        "engine_control.EngineControl.is_paused",
        side_effect=[True, True, False]
    )
    run_watch_cycle(times=3)
    mock_price.assert_not_called()

신호 파일 생성/삭제 기반 pause·resume 테스트 5개, /status·/holdings 텍스트 조립 테스트 6개, 감시 루프 연동 테스트 3개까지 신규 14개가 추가됐고, 기존 88개를 포함해 전체 pytest는 102개 전부 통과했다. 안전장치 값이 실수로라도 바뀌지 않았는지도 별도로 확인했다.

텔레그램 봇으로 자동매매 프로세스를 제어하려면, DB나 프로세스 통합 없이도 신호 파일 하나의 존재 여부만으로 pause/resume 상태를 판단하는 방식이 가장 단순하고 안전하다.

남은 것 — 진짜 봇 연결

이번 세션에서는 “무엇을 멈추고 무엇을 보여줄지”의 로직만 완성했다. python-telegram-bot 같은 실제 라이브러리를 붙여서, /status 수신 시 bot_commands.get_status_text()를, /pause 수신 시 engine_control.pause()를 그대로 호출하도록 명령어 핸들러를 연결하는 작업만 남았다. 이게 끝나면 Phase 5(알림/모니터링 고도화)가 전체 마무리되고, 다음은 Phase 6(운영 자동화)으로 넘어간다.

FAQ

Q1. 왜 DB 명령 큐 대신 파일 시그널을 선택했나요?
A1. 지금 규모에서는 DB 폴링이 과했고, 파일 존재 여부 확인만으로도 pause/resume 상태 전달이 충분히 안정적이었기 때문입니다.

Q2. /pause를 눌러도 안전장치 설정은 그대로 유지되나요?
A2. 네. DRY_RUN이나 목표가 같은 3중 안전장치 값은 감시 루프 정지와 무관하게 전혀 건드리지 않는다는 원칙을 코드 레벨에서 지켰습니다.

Q3. 실제 텔레그램 명령어는 이미 동작하나요?
A3. 아직입니다. 이번 세션은 제어 로직까지만 완성했고, 실제 라이브러리 연결과 명령어 수신부는 다음 세션에서 이어집니다.

Q4. 알림이 매 주기마다 쏟아지지는 않나요?
A4. 아니요. 일시정지/재개 상태가 바뀌는 전이 시점에만 알림이 1회씩 발송되도록 설계했습니다.