130 lines
5.1 KiB
Python
130 lines
5.1 KiB
Python
"""KIS adapter tests (REQ-012).
|
|
|
|
Unit tests run always; the sandbox integration test runs only when KIS
|
|
모의투자 credentials are present (TREX_KIS_BROKER_* env / .env).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from trex_core.adapters.broker.kis import KISBroker
|
|
from trex_core.application.broker_port import BrokerBalanceError, OrderRequest
|
|
|
|
_MOCK_URL = "https://openapivts.koreainvestment.com:29443"
|
|
|
|
|
|
def test_account_split_and_mock_tr_ids():
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL)
|
|
assert b._account() == ("12345678", "01")
|
|
assert b._tr_submit("buy") == "VTTC0012U"
|
|
assert b._tr_submit("sell") == "VTTC0011U"
|
|
assert b.name() == "kis-mock"
|
|
|
|
|
|
def test_live_base_url_requires_allow_live():
|
|
with pytest.raises(RuntimeError):
|
|
KISBroker("12345678-01", app_key="k", app_secret="s",
|
|
base_url="https://openapi.koreainvestment.com:9443")
|
|
# gated on explicitly
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s",
|
|
base_url="https://openapi.koreainvestment.com:9443", allow_live=True)
|
|
assert b.name() == "kis-live" and b._tr_submit("buy") == "TTTC0012U"
|
|
|
|
|
|
def _kis_env():
|
|
keys = ("TREX_KIS_BROKER_BASE_URL", "TREX_KIS_BROKER_APP_KEY",
|
|
"TREX_KIS_BROKER_APP_SECRET", "TREX_KIS_BROKER_ACCOUNT_REF")
|
|
if not all(os.environ.get(k) for k in keys):
|
|
# try .env via dotenv
|
|
try:
|
|
from dotenv import load_dotenv, find_dotenv
|
|
load_dotenv(find_dotenv(usecwd=True), override=False)
|
|
except ImportError:
|
|
pass
|
|
if not all(os.environ.get(k) for k in keys):
|
|
pytest.skip("KIS 모의투자 credentials not set; skipping sandbox integration test")
|
|
return {k: os.environ[k] for k in keys}
|
|
|
|
|
|
def test_sandbox_token_and_submit():
|
|
"""Single sandbox round-trip (one token issuance — KIS rate-limits token requests)."""
|
|
import httpx
|
|
|
|
env = _kis_env()
|
|
assert "openapivts" in env["TREX_KIS_BROKER_BASE_URL"], "expect 모의투자 sandbox URL"
|
|
b = KISBroker(
|
|
env["TREX_KIS_BROKER_ACCOUNT_REF"],
|
|
app_key=env["TREX_KIS_BROKER_APP_KEY"],
|
|
app_secret=env["TREX_KIS_BROKER_APP_SECRET"],
|
|
base_url=env["TREX_KIS_BROKER_BASE_URL"],
|
|
timeout=float(os.environ.get("TREX_KIS_BROKER_TIMEOUT", "10")),
|
|
)
|
|
try:
|
|
token = b._ensure_token()
|
|
except httpx.HTTPStatusError as exc: # KIS EGW00133: token issuance rate-limited
|
|
pytest.skip(f"KIS token rate-limited/unavailable: {exc}")
|
|
assert isinstance(token, str) and len(token) > 0
|
|
|
|
# limit buy 1 share at a low price (sandbox, no real money). Reuses the cached token.
|
|
ack = b.submit_order(OrderRequest(slot_id="t", side="buy", symbol="005930", qty=1,
|
|
limit_price=10000, order_type="00"))
|
|
assert ack.raw, "expected a KIS response payload"
|
|
if ack.accepted:
|
|
assert ack.broker_order_id
|
|
else:
|
|
assert ack.reason # rejection carries a reason (e.g., 장 미개장/잔고부족)
|
|
|
|
|
|
def test_get_balance_rejects_failure_instead_of_false_empty(monkeypatch):
|
|
"""BAT-057: a rejected KIS response is not a successful empty balance."""
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL)
|
|
monkeypatch.setattr(b, "_headers", lambda _tr: {})
|
|
|
|
class Response:
|
|
def raise_for_status(self): pass
|
|
def json(self): return {"rt_cd": "1", "output1": []}
|
|
|
|
monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response())
|
|
with pytest.raises(BrokerBalanceError, match="broker_rejected"):
|
|
b.get_balance()
|
|
|
|
|
|
def test_get_balance_accepts_validated_empty_balance(monkeypatch):
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL)
|
|
monkeypatch.setattr(b, "_headers", lambda _tr: {})
|
|
|
|
class Response:
|
|
def raise_for_status(self): pass
|
|
def json(self): return {"rt_cd": "0", "output1": []}
|
|
|
|
monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response())
|
|
assert b.get_balance() == []
|
|
|
|
|
|
def test_get_balance_sanitizes_token_issuance_failure(monkeypatch):
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL)
|
|
monkeypatch.setattr(b, "_headers", lambda _tr: (_ for _ in ()).throw(KeyError("access_token-secret")))
|
|
with pytest.raises(BrokerBalanceError) as raised:
|
|
b.get_balance()
|
|
assert raised.value.code == "token_issuance_failed"
|
|
assert "secret" not in str(raised.value)
|
|
|
|
|
|
def test_get_available_cash_uses_cash_only_amount(monkeypatch):
|
|
b = KISBroker("12345678-01", app_key="k", app_secret="s", base_url=_MOCK_URL)
|
|
monkeypatch.setattr(b, "_headers", lambda _tr: {})
|
|
monkeypatch.setattr(b, "get_current_price", lambda _symbol: None)
|
|
|
|
class Response:
|
|
def raise_for_status(self): pass
|
|
def json(self):
|
|
return {
|
|
"rt_cd": "0",
|
|
"output": {"nrcvb_buy_amt": "695591", "max_buy_amt": "999999"},
|
|
}
|
|
|
|
monkeypatch.setattr("trex_core.adapters.broker.kis.httpx.get", lambda *a, **kw: Response())
|
|
assert b.get_available_cash() == 695_591
|