시스템 아키텍처

시스템 개요

HTS RPA는 Google Sheets에서 주문 데이터를 읽어와 메리츠증권 HTS를 자동으로 조작하여 주문을 실행하는 RPA 시스템입니다.

┌────────────────────────────────────────────────────────┐
│                      전략 모듈                           │
│  ┌────────────┐ ┌────────────┐ ┌────────────┐          │
│  │ 단순매매   │ │ 그리드     │ │ 무한매수법 │  ...     │
│  └─────┬──────┘ └─────┬──────┘ └─────┬──────┘          │
│        └──────────────┼──────────────┘                  │
│                       ▼                                  │
│  ┌──────────────────────────────────────────────────┐  │
│  │             공통 주문 실행 모듈                    │  │
│  │  - OrderRequest 생성                              │  │
│  │  - HTS 주문창 조작                                │  │
│  │  - 주문 결과 반환                                 │  │
│  └──────────────────────────────────────────────────┘  │
└────────────────────────────────────────────────────────┘

프로젝트 구조

hts_rpa/
├── main.py                     # 메인 진입점
├── requirements.txt
├── config/
│   ├── settings.py             # 설정 관리 (JSON 로드)
│   ├── settings.json           # 애플리케이션 설정 (gitignore)
│   ├── strategies.json         # 전략 설정 (gitignore)
│   └── credentials.json        # Google Service Account 인증 (gitignore)
├── core/                       # 공통 핵심 모듈
│   ├── order_executor.py       # 공통 주문 실행 (HTS 조작)
│   ├── order_request.py        # 주문 요청 데이터 클래스
│   └── order_result.py         # 주문 결과 데이터 클래스
├── modules/
│   ├── sheets_manager.py       # Google Sheets 연동 (Service Account)
│   ├── credential_manager.py   # 자격증명 관리 (keyring)
│   ├── notifier.py             # 통합 알림 모듈
│   └── hts/                    # HTS 자동화 모듈
│       ├── controller.py       # HTSController (통합 API)
│       ├── window_manager.py   # 창 관리
│       ├── login_handler.py    # 인증서 로그인
│       ├── order_handler.py    # 주문, 잔고/예수금 조회
│       └── control_finder.py   # win32gui 컨트롤 탐색 (DPI-aware)
├── strategies/                 # 전략별 모듈
│   ├── base.py                 # 전략 기본 추상 클래스
│   ├── registry.py             # 전략 레지스트리 (자동 등록)
│   ├── simple_order/           # 단순매매 전략
│   ├── grid/                   # 그리드 트레이딩 전략
│   └── infinite_buy/           # 무한매수법 V3.0 전략
├── gui/                        # CustomTkinter GUI
│   ├── app.py                  # MainApp (메인 윈도우)
│   ├── frames/                 # UI 프레임
│   └── dialogs/                # 설정 다이얼로그
├── utils/
│   ├── logger.py               # 로깅 유틸리티
│   └── input_blocker.py        # 사용자 입력 차단
├── scripts/                    # 유틸리티 스크립트
├── logs/                       # 로그 파일
└── tests/                      # 단위 테스트

핵심 데이터 구조

OrderRequest (주문 요청)

전략 모듈에서 생성하여 주문 실행 모듈로 전달하는 공통 데이터 구조입니다.

@dataclass
class OrderRequest:
    symbol: str                    # 종목코드 (예: "AAPL")
    side: OrderSide               # 매수/매도
    quantity: int                  # 수량
    order_type: OrderType          # 시장가/지정가/LOC/MOC
    price: Optional[float] = None  # 지정가일 때 가격
    row_index: Optional[int] = None
    strategy_data: Optional[dict] = None

OrderResult (주문 결과)

@dataclass
class OrderResult:
    status: OrderStatus           # 완료/실패/부분체결
    message: str                   # 결과 메시지
    executed_time: datetime        # 실행 시간
    request: OrderRequest          # 원본 요청
    executed_quantity: Optional[int] = None
    executed_price: Optional[float] = None

전략 시스템

BaseStrategy (추상 클래스)

모든 전략이 구현해야 하는 인터페이스입니다.

메서드	설명
`load_orders()`	시트에서 주문 데이터를 로드하여 OrderRequest 리스트 반환
`on_order_result()`	주문 실행 결과를 처리 (시트 업데이트 등)
`on_cycle_complete()`	사이클 완료 시 호출 (CONTINUOUS 모드용)

StrategyRegistry (자동 등록)

@StrategyRegistry.register
class GridStrategy(BaseStrategy):
    STRATEGY_ID = "grid_trading"
    STRATEGY_NAME = "그리드 트레이딩"
    EXECUTION_MODE = ExecutionMode.CONTINUOUS

@StrategyRegistry.register 데코레이터로 전략을 등록하면, 전략 ID로 자동 조회할 수 있습니다.

실행 플로우

Grid 전략 3-Phase Lock 구조

# Phase 1: Sheets 읽기 (락 없음)
strategy.prepare_cycle_data()

# Phase 2: HTS 조작 (락 보유)
with hts_execution_lock:
    while order_loop_count < max_order_loops:
        orders = strategy.load_orders()
        execute_orders(orders)
        filled = strategy.check_fills_and_save()
        if filled > 0: continue
        else: break

# Phase 3: Sheets 쓰기 + 알림 (락 없음)
strategy.flush_deferred_writes()

기본 전략 실행 루프

while not should_stop:
    orders = strategy.load_orders()
    for order in orders:
        result = executor.execute(order)
        strategy.on_order_result(result)
    strategy.on_cycle_complete()
    if strategy.EXECUTION_MODE == ExecutionMode.ONE_TIME:
        break
    time.sleep(strategy.POLLING_INTERVAL)

기술 스택

핵심 라이브러리

라이브러리	용도
`customtkinter`	GUI 프레임워크
`pywinauto`	Windows 앱 자동화 (인증서 창)
`pywin32`	win32gui/win32api (HTS 커스텀 컨트롤)
`gspread`	Google Sheets API
`google-auth`	Google Service Account 인증
`keyring`	Windows 자격증명관리자 연동
`requests`	Telegram/Discord 웹훅 호출
`colorlog`	컬러 로그 출력

개발 도구

도구	용도
`pytest`	테스트 프레임워크
`pyinstaller`	실행 파일 빌드
`black`	코드 포매터
`flake8`	린터