클라이언트 (client · api)¶

KiwoomClient¶

KiwoomApi와 KiwoomAuth를 통합한 사용자 진입점 파사드입니다. 초기화 시 접근토큰을 자동으로 발급합니다.

kiwoompy.client ¶

KiwoomClient — KiwoomApi, KiwoomAuth, KiwoomQuery, KiwoomOrder를 통합한 사용자 진입점 파사드.

KiwoomClient ¶

KiwoomClient(env: Env = 'demo', appkey: str = '', secretkey: str = '', rps: float | None = None)

키움증권 REST API 통합 클라이언트.

KiwoomApi(HTTP 통신·유량 제어), KiwoomAuth(토큰 발급·갱신), KiwoomQuery(계좌·잔고·손익 조회), KiwoomOrder(주식·신용 주문)를 하나로 묶은 파사드 클래스. 초기화 시 접근토큰을 자동으로 발급한다.

Parameters:

Name	Type	Description	Default
`env`	`Env`	환경 구분. `"real"` (운영) 또는 `"demo"` (모의투자). 기본값 `"demo"`.	`'demo'`
`appkey`	`str`	키움증권 앱 키.	`''`
`secretkey`	`str`	키움증권 시크릿 키.	`''`
`rps`	`float \| None`	초당 최대 요청 수. `None`이면 환경별 기본값 사용 (`demo` 2건/초, `real` 20건/초).	`None`

Examples:

기본 사용법:

from kiwoompy import KiwoomClient

client = KiwoomClient(
    env="demo",
    appkey="YOUR_APP_KEY",
    secretkey="YOUR_APP_SECRET",
)
balance = client.query.get_account_balance()
print(balance.tot_evlt_amt)  # 총평가금액

context manager:

with KiwoomClient(env="demo", appkey="...", secretkey="...") as client:
    unfilled = client.query.get_unfilled_orders(all_stock_type="0", trade_type="0")
    result = client.order.buy("005930", "1", trade_type="3")  # 시장가 매수

Source code in src/kiwoompy/client.py

def __init__(
    self,
    env: Env = "demo",
    appkey: str = "",
    secretkey: str = "",
    rps: float | None = None,
) -> None:
    self._api = KiwoomApi(env=env, rps=rps)
    self._auth = KiwoomAuth(self._api)
    self._query = KiwoomQuery(self._api)
    self._order = KiwoomOrder(self._api)
    self._auth.issue_token(appkey=appkey, secretkey=secretkey)

api `property` ¶

api: KiwoomApi

내부 KiwoomApi 인스턴스.

고급 사용자나 테스트에서 HTTP 클라이언트에 직접 접근할 때 사용한다.

auth `property` ¶

auth: KiwoomAuth

내부 KiwoomAuth 인스턴스.

토큰 유효성 확인이나 수동 갱신이 필요할 때 사용한다.

order `property` ¶

order: KiwoomOrder

내부 KiwoomOrder 인스턴스.

주식·신용 매수/매도/정정/취소 주문 메서드에 접근한다.

Examples:

result = client.order.buy("005930", "1", trade_type="3")
result = client.order.cancel(result.ord_no, "005930", "0")

query `property` ¶

query: KiwoomQuery

내부 KiwoomQuery 인스턴스.

계좌·잔고·손익·주문체결 조회 메서드에 접근한다.

Examples:

balance = client.query.get_account_balance()
unfilled = client.query.get_unfilled_orders(all_stock_type="0", trade_type="0")

close ¶

close() -> None

HTTP 클라이언트 세션을 닫는다.

Source code in src/kiwoompy/client.py

def close(self) -> None:
    """HTTP 클라이언트 세션을 닫는다."""
    self._api.close()

refresh_token ¶

refresh_token(appkey: str, secretkey: str) -> TokenResponse

접근토큰을 재발급한다.

토큰 만료 전후로 명시적으로 갱신이 필요할 때 호출한다.

Parameters:

Name	Type	Description	Default
`appkey`	`str`	키움증권 앱 키.	required
`secretkey`	`str`	키움증권 시크릿 키.	required

Returns:

Type	Description
`TokenResponse`	새로 발급된 `TokenResponse`.

Source code in src/kiwoompy/client.py

def refresh_token(self, appkey: str, secretkey: str) -> TokenResponse:
    """접근토큰을 재발급한다.

    토큰 만료 전후로 명시적으로 갱신이 필요할 때 호출한다.

    Args:
        appkey: 키움증권 앱 키.
        secretkey: 키움증권 시크릿 키.

    Returns:
        새로 발급된 ``TokenResponse``.
    """
    return self._auth.issue_token(appkey=appkey, secretkey=secretkey)

KiwoomApi¶

HTTP 통신, 유량 제어, 자동 재시도를 담당하는 저수준 클라이언트입니다. 테스트 모킹이나 세밀한 제어가 필요할 때 직접 사용합니다.

kiwoompy.api ¶

REST 통신 계층 — base URL 관리, 토큰 보관, HTTP get/post 단일 진입점.

KiwoomApi ¶

KiwoomApi(env: Env = 'demo', rps: float | None = None)

키움 REST API HTTP 클라이언트.

모든 HTTP 호출은 이 클래스를 통해서만 이루어진다. 발급된 접근토큰을 내부에 보관하고, 이후 요청 헤더에 자동으로 포함한다.

유량 제어: 환경별 기본 RPS를 자동 적용한다.

실전(real): 기본 20건/초
모의(demo): 기본 2건/초

rps 파라미터로 직접 조정할 수 있다.

재시도: 네트워크 오류·타임아웃·5xx 서버 오류는 지수 백오프로 최대 _MAX_ATTEMPTS회 재시도한다. 4xx 인증 오류는 재시도하지 않는다.

Parameters:

Name	Type	Description	Default
`env`	`Env`	환경 구분. `"real"` (운영) 또는 `"demo"` (모의투자).	`'demo'`
`rps`	`float \| None`	초당 최대 요청 수. `None`이면 환경별 기본값 사용.	`None`

Source code in src/kiwoompy/api.py

def __init__(self, env: Env = "demo", rps: float | None = None) -> None:
    self._base_url: str = _BASE_URLS[env]
    self._token: str | None = None
    self._rate_limiter = _RateLimiter(rps if rps is not None else _DEFAULT_RPS[env])
    self._client = httpx.Client(
        base_url=self._base_url,
        headers={"Content-Type": "application/json;charset=UTF-8"},
        timeout=_TIMEOUT,
    )

close ¶

close() -> None

HTTP 클라이언트 세션을 닫는다.

Source code in src/kiwoompy/api.py

def close(self) -> None:
    """HTTP 클라이언트 세션을 닫는다."""
    self._client.close()

get_auth_header ¶

get_auth_header() -> dict[str, str]

현재 저장된 접근토큰으로 Authorization 헤더를 반환한다.

Returns:

Type	Description
`dict[str, str]`	`{"Authorization": "Bearer <token>"}` 형태의 딕셔너리.

Raises:

Type	Description
`KiwoomAuthError`	토큰이 아직 발급되지 않은 경우.

Source code in src/kiwoompy/api.py

def get_auth_header(self) -> dict[str, str]:
    """현재 저장된 접근토큰으로 Authorization 헤더를 반환한다.

    Returns:
        ``{"Authorization": "Bearer <token>"}`` 형태의 딕셔너리.

    Raises:
        KiwoomAuthError: 토큰이 아직 발급되지 않은 경우.
    """
    if self._token is None:
        raise KiwoomAuthError("접근토큰이 없습니다. issue_token()을 먼저 호출하세요.")
    return {"Authorization": f"Bearer {self._token}"}

post ¶

post(path: str, body: dict, headers: dict[str, str] | None = None) -> dict

JSON POST 요청을 보내고 응답 JSON을 반환한다.

유량 제어 후 요청을 전송한다. 네트워크 오류·타임아웃·5xx는 지수 백오프로 재시도한다. 4xx 응답(KiwoomAuthError)은 재시도하지 않고 즉시 raise한다.

Parameters:

Name	Type	Description	Default
`path`	`str`	엔드포인트 경로 (예: `"/oauth2/token"`).	required
`body`	`dict`	요청 본문 딕셔너리.	required
`headers`	`dict[str, str] \| None`	추가 요청 헤더. `None`이면 기본 헤더만 사용.	`None`

Returns:

Type	Description
`dict`	응답 JSON을 파싱한 딕셔너리.

Raises:

Type	Description
`KiwoomAuthError`	HTTP 4xx 응답 (인증 실패 등). 재시도 없음.
`KiwoomApiError`	최대 재시도 후에도 5xx·네트워크·파싱 오류가 지속되는 경우.

Source code in src/kiwoompy/api.py

@_retry
def post(self, path: str, body: dict, headers: dict[str, str] | None = None) -> dict:
    """JSON POST 요청을 보내고 응답 JSON을 반환한다.

    유량 제어 후 요청을 전송한다. 네트워크 오류·타임아웃·5xx는 지수 백오프로
    재시도한다. 4xx 응답(``KiwoomAuthError``)은 재시도하지 않고 즉시 raise한다.

    Args:
        path: 엔드포인트 경로 (예: ``"/oauth2/token"``).
        body: 요청 본문 딕셔너리.
        headers: 추가 요청 헤더. ``None``이면 기본 헤더만 사용.

    Returns:
        응답 JSON을 파싱한 딕셔너리.

    Raises:
        KiwoomAuthError: HTTP 4xx 응답 (인증 실패 등). 재시도 없음.
        KiwoomApiError: 최대 재시도 후에도 5xx·네트워크·파싱 오류가 지속되는 경우.
    """
    self._rate_limiter.acquire()

    try:
        response = self._client.post(path, json=body, headers=headers)
    except httpx.TimeoutException as exc:
        raise KiwoomApiError(f"요청 타임아웃: {path}") from exc
    except httpx.RequestError as exc:
        raise KiwoomApiError(f"네트워크 오류: {exc}") from exc

    if 400 <= response.status_code < 500:
        raise KiwoomAuthError(
            f"인증 오류: {response.text}",
            status_code=response.status_code,
        )
    if response.status_code >= 500:
        raise KiwoomApiError(
            f"서버 오류: {response.text}",
            status_code=response.status_code,
        )

    try:
        return response.json()
    except Exception as exc:
        raise KiwoomApiError(f"응답 파싱 실패: {response.text}") from exc

set_token ¶

set_token(token: str) -> None

발급된 접근토큰을 저장한다. 이후 모든 요청에 자동 포함된다.

Parameters:

Name	Type	Description	Default
`token`	`str`	접근토큰 문자열.	required

Source code in src/kiwoompy/api.py

def set_token(self, token: str) -> None:
    """발급된 접근토큰을 저장한다. 이후 모든 요청에 자동 포함된다.

    Args:
        token: 접근토큰 문자열.
    """
    self._token = token

클라이언트 (client · api)¶

KiwoomClient¶

kiwoompy.client ¶

KiwoomClient ¶

api property ¶

auth property ¶

order property ¶

query property ¶

close ¶

refresh_token ¶

KiwoomApi¶

kiwoompy.api ¶

KiwoomApi ¶

close ¶

get_auth_header ¶

post ¶

set_token ¶

api `property` ¶

auth `property` ¶

order `property` ¶

query `property` ¶