QuantEngine 집합 분류 알고리즘 Python 구현 — 키움 API 종목교체 자동화 [Claude Code · 키움리밸런싱 · 8편]

Python 알고리즘 퀀트전략 집합연산 asyncio

QuantEngine — 종목 분류 알고리즘 Python 구현

A-B / A∩B / B-A 집합 연산으로 매도·매수 계획을 자동 생성하는 핵심 코드 공개

QuantEngine 설계 목표

QuantEngine은 V2 퀀트 리밸런싱의 두뇌입니다. 현재 보유 종목(A)과 목표 종목(B)을 입력받아 두 가지 결과를 만듭니다.

• classify() — A-B / A∩B / B-A 분류 결과 (미리보기용)
• build_plan() — 실제 분할 주문 계획 생성 (실행용)

V1에서 사용하던 _compute_sell_rounds(), _compute_buy_rounds() 함수를 그대로 재활용했습니다. 분할 주문 누산기 알고리즘은 검증된 코드를 바꿀 이유가 없었습니다.

매도 완료 후 계좌 재조회로 실체결 금액 기준 매수

classify() — 미리보기용 분류 함수

실행 전 "어떤 종목을 팔고 어떤 종목을 살 것인지"를 미리 보여주는 함수입니다. 계좌 잔고를 변경하지 않고 분류 결과만 반환합니다.

set 자료구조로 A와 B의 교집합/차집합을 O(1) 복잡도로 판별합니다. 종목 수가 많아도 빠릅니다.

def classify(self, account_info, target_stocks) -> QuantClassification:
    target_codes = {s["code"] for s in target_stocks}  # B 집합
    pos_map = {p.code: p for p in account_info.positions}  # A 집합

    total_eval = sum(p.eval_amount for p in account_info.positions) + account_info.deposit
    target_per = total_eval // len(target_stocks)  # 목표 균등금액

    # A - B: 기존에만 있는 종목 → 전량 매도 예정
    sell_only = [
        {"code": p.code, "name": p.name, "eval_amount": p.eval_amount, "quantity": p.quantity}
        for p in account_info.positions if p.code not in target_codes
    ]

    # A ∩ B: 중복 종목 → 초과/부족/균형 판별
    overlap = []
    for p in account_info.positions:
        if p.code in target_codes:
            diff = p.eval_amount - target_per
            if diff > 0 and p.current_price > 0:
                sell_qty = math.floor(diff / p.current_price)
                action = "reduce" if sell_qty > 0 else "hold"
            elif diff < 0:
                action = "supplement"
            else:
                action = "hold"
            overlap.append(OverlapStock(..., action=action))

    # B - A: 신규 종목 → 목표금액 전체 매수 예정
    existing_codes = {p.code for p in account_info.positions}
    buy_only = [
        {"code": s["code"], "target_amount": target_per,
          "est_qty": math.floor(target_per / s["price"])}
        for s in target_stocks if s["code"] not in existing_codes
    ]

    return QuantClassification(sell_only=sell_only, overlap=overlap, buy_only=buy_only, ...)

build_plan() — 실행용 분할 주문 계획

classify()와 같은 분류 로직을 거치되, 각 종목에 분할 주문 수량 리스트를 붙여 반환합니다.

매도 수량을 누산기 방식으로 분할하는 이유: 단순 나눗셈은 소수점 버림으로 총 수량보다 적게 주문됩니다. 누산기가 오차를 다음 회차에 누적해 마지막 회차에서 정확히 맞춥니다.

def _compute_sell_rounds(total_qty: int, total_rounds: int) -> list[int]:
    step = total_qty / total_rounds
    acc = 0.0
    qtys = []
    for i in range(total_rounds):
        acc += step
        q = int(acc)       # 소수점 버림
        acc -= q            # 버린 소수점을 다음 회차로 누산
        qtys.append(q)
    remainder = total_qty - sum(qtys)
    if remainder > 0:
        qtys[-1] += remainder  # 마지막 회차에 나머지 합산
    return qtys

# 예시: 17주를 10회차로 분할
_compute_sell_rounds(17, 10)
# → [1, 2, 1, 2, 1, 2, 1, 2, 1, 4]  (합계 17 보장)

매수 분할은 금액 기준입니다. 매 회차 현재가로 수량을 재계산하므로 가격 변동에 자동 대응합니다.

def _compute_buy_rounds(total_amount: int, total_rounds: int, price: int) -> list[int]:
    step = total_amount / total_rounds
    for i in range(total_rounds):
        if i == total_rounds - 1:
            amount = ((total_amount - total_spent) // price) * price  # 마지막 회차
        else:
            shares = int(acc / price)
            amount = shares * price  # 1주 단위 절삭
        amounts.append(amount)

2단계 실행 — 매도 후 계좌 재조회가 핵심

실행은 QuantScheduleManager._run_inner()가 담당합니다. 매도 단계와 매수 단계 사이에 계좌를 반드시 재조회합니다.

왜 재조회가 필요한가: 매도 체결 시 수수료·슬리피지로 예수금이 계획과 다릅니다. 재조회 없이 매수하면 잔액 부족 오류가 납니다.

# 매도 단계
account_info = await self.am.fetch_account_info(account_no)
self.plan = self.engine.build_plan(account_info, targets, sell_split, buy_split,
                                     sell_enabled=True, buy_enabled=False)
for i in range(cfg.sell_split):
    await self.om.execute_sell_round(self.plan.sell_items, round_no=i+1)
    await asyncio.sleep(interval)

# 매도 미체결 완료 대기
await self.om.wait_sell_complete(cfg.unfilled_poll_minutes, self.log_fn)

# 계좌 재조회 (실체결 반영)
account_info = await self.am.fetch_account_info(account_no)

# 매수 계획 재계산 + 신규 종목 현재가 갱신
self.plan = self.engine.build_plan(account_info, targets, sell_split, buy_split,
                                     sell_enabled=False, buy_enabled=True)
for i in range(cfg.buy_split):
    await self.om.execute_buy_round(self.plan.buy_items, round_no=i+1)

겪은 문제들

문제 1: KOSDAQ 종목코드 앞에 'Q'가 붙어 집합 비교 불일치
키움 REST API가 계좌 잔고를 조회하면 KOSPI는 "A357870", KOSDAQ은 "Q357870" 형태로 반환합니다. 사용자가 입력한 "357870"과 집합 비교 시 항상 불일치해 모든 종목이 매도 대상이 되는 버그였습니다.

해결: re.sub()로 선행 알파벳 전체 제거
lstrip("A")는 KOSPI만 처리합니다. re.sub(r'^[A-Za-z]+', '', raw_code)로 교체해 모든 시장 구분자를 제거했습니다.

문제 2: Mock 모드에서 중복 종목 0개
config.json의 "is_mock": true 상태에서 테스트하면 Mock API가 삼성전자, SK하이닉스 등 고정 종목을 반환합니다. 실제 보유 ETF 코드와 전혀 달라 A∩B = 0이었습니다.

해결: /api/debug/compare 진단 엔드포인트 추가
계좌 보유 코드와 목표 종목 코드를 직접 비교해 출력하는 진단 API를 추가했습니다. Mock 모드임을 즉시 확인할 수 있었습니다.

---

다음 편에서는 실제 웹 UI 화면을 공개합니다. 종목교체 설정, 미리보기, 스케줄 진단 기능을 스크린샷과 함께 살펴봅니다.

시리즈 전체 목차
1~6편 — V1 균등배분 리밸런싱
7편 — V2 기획 & 시스템 설계
8편 — QuantEngine 집합 분류 알고리즘
9편 — 웹 UI & 실제 화면 공개