From e39716e93b9b1a401a2cf17c7a6d530151ef4dfe Mon Sep 17 00:00:00 2001 From: Heewon Oh Date: Thu, 4 Jun 2026 11:14:59 +0900 Subject: [PATCH] =?UTF-8?q?refactor(agent):=20extract=20RateLimiterControl?= =?UTF-8?q?Mixin=20from=20Agent=20(955=20=E2=86=92=20865=20LOC)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit agent.py에서 Rate Limiter 제어 메서드 4개를 별도 Mixin으로 분리. 분리 대상 (kis_agent/core/rate_limiter_mixin.py): - get_rate_limiter_status() — 상태 조회 - set_rate_limits() — 제한 값 동적 변경 - reset_rate_limiter() — 통계/백오프 초기화 - enable_adaptive_rate_limiting() — 적응형 속도 조절 토글 분리 근거: - 4개 메서드 모두 self.rate_limiter에만 의존, 다른 도메인 API와 결합 없음 - 모두 단순 위임 layer (예외 처리·비활성 경고 외 로직 없음) - 기존 TechnicalAnalysisMixin / MethodDiscoveryMixin과 동일한 컨벤션 - Agent 클래스에서 도메인 API 코드와 인프라 제어 코드 분리 공개 API 영향 없음: - agent.get_rate_limiter_status() 등 사용자 호출 방식 동일 - Agent.__mro__: [Agent, TechAnalysisMixin, MethodDiscoveryMixin, RateLimiterControlMixin, BaseExceptionHandler, object] 검증: - pytest tests/unit/test_agent.py tests/unit/test_agent_unit_comprehensive.py tests/unit/test_rate_limiter.py: 49/49 passed - ruff check: All checks passed --- kis_agent/core/agent.py | 106 ++-------------------- kis_agent/core/rate_limiter_mixin.py | 126 +++++++++++++++++++++++++++ 2 files changed, 134 insertions(+), 98 deletions(-) create mode 100644 kis_agent/core/rate_limiter_mixin.py diff --git a/kis_agent/core/agent.py b/kis_agent/core/agent.py index d5055ee..e290b62 100644 --- a/kis_agent/core/agent.py +++ b/kis_agent/core/agent.py @@ -28,10 +28,16 @@ from .constants import REAL_BASE_URL from .method_discovery import MethodDiscoveryMixin from .rate_limiter import RateLimiter, get_global_rate_limiter +from .rate_limiter_mixin import RateLimiterControlMixin from .technical_analysis import TechnicalAnalysisMixin -class Agent(TechnicalAnalysisMixin, MethodDiscoveryMixin, BaseExceptionHandler): +class Agent( + TechnicalAnalysisMixin, + MethodDiscoveryMixin, + RateLimiterControlMixin, + BaseExceptionHandler, +): """ 한국투자증권 API의 통합 인터페이스 @@ -685,105 +691,9 @@ def order_stock_credit( ) # ============================================================================ - # Rate Limiter 관리 메서드 + # Rate Limiter 관리 메서드 (kis_agent.core.rate_limiter_mixin.RateLimiterControlMixin 상속) # ============================================================================ - def get_rate_limiter_status(self) -> Optional[Dict[str, Any]]: - """ - Rate Limiter 상태 조회 - - Returns: - Dict: Rate Limiter 상태 정보 - - requests_per_second: 현재 초당 요청 수 - - requests_per_minute: 현재 분당 요청 수 - - limit_per_second: 초당 제한 - - limit_per_minute: 분당 제한 - - backoff_multiplier: 백오프 배수 - - total_requests: 총 요청 수 - - throttled_count: 제한된 요청 수 - - avg_wait_time: 평균 대기 시간 - None: Rate Limiter가 비활성화된 경우 - - Example: - >>> status = agent.get_rate_limiter_status() - >>> if status: - ... print(f"현재 요청률: {status['requests_per_second']}/초") - ... print(f"제한 도달 횟수: {status['throttled_count']}") - """ - if self.rate_limiter: - return self.rate_limiter.get_current_rate() - return None - - def set_rate_limits( - self, - requests_per_second: Optional[int] = None, - requests_per_minute: Optional[int] = None, - min_interval_ms: Optional[int] = None, - ) -> None: - """ - Rate Limiter 제한 값 동적 변경 - - Args: - requests_per_second: 초당 최대 요청 수 (None이면 변경 안 함) - requests_per_minute: 분당 최대 요청 수 (None이면 변경 안 함) - min_interval_ms: 최소 간격 (밀리초) (None이면 변경 안 함) - - Example: - >>> # 더 보수적인 설정으로 변경 - >>> agent.set_rate_limits( - ... requests_per_second=10, - ... requests_per_minute=500 - ... ) - >>> - >>> # 최소 간격만 변경 - >>> agent.set_rate_limits(min_interval_ms=100) - """ - if self.rate_limiter: - self.rate_limiter.set_limits( - requests_per_second=requests_per_second, - requests_per_minute=requests_per_minute, - min_interval_ms=min_interval_ms, - ) - # [변경 이유] 불필요한 f-string 사용으로 F541 경고 발생 -> 리터럴 문자열로 변경 - logging.info("Rate limits 업데이트 완료") - else: - logging.warning("Rate Limiter가 비활성화 상태입니다") - - def reset_rate_limiter(self) -> None: - """ - Rate Limiter 상태 초기화 - - 모든 요청 기록과 통계를 초기화합니다. - 백오프 배수도 1.0으로 리셋됩니다. - - Example: - >>> agent.reset_rate_limiter() - >>> print("Rate limiter 초기화 완료") - """ - if self.rate_limiter: - self.rate_limiter.reset() - logging.info("Rate limiter 초기화 완료") - else: - logging.warning("Rate Limiter가 비활성화 상태입니다") - - def enable_adaptive_rate_limiting(self, enable: bool = True) -> None: - """ - 적응형 속도 조절 활성화/비활성화 - - Args: - enable: True면 활성화, False면 비활성화 - - Example: - >>> # 적응형 속도 조절 비활성화 - >>> agent.enable_adaptive_rate_limiting(False) - """ - if self.rate_limiter: - self.rate_limiter.enable_adaptive = enable - status = "활성화" if enable else "비활성화" - logging.info(f"적응형 속도 조절 {status}") - else: - logging.warning("Rate Limiter가 비활성화 상태입니다") - # ============================================================================ # 관심종목 관련 메서드 # ============================================================================ diff --git a/kis_agent/core/rate_limiter_mixin.py b/kis_agent/core/rate_limiter_mixin.py new file mode 100644 index 0000000..464b99f --- /dev/null +++ b/kis_agent/core/rate_limiter_mixin.py @@ -0,0 +1,126 @@ +"""Agent의 Rate Limiter 제어 메서드를 분리한 Mixin. + +원래 ``Agent`` 클래스에 직접 정의되어 있던 4개의 rate-limiter 제어 메서드 +(``get_rate_limiter_status``, ``set_rate_limits``, ``reset_rate_limiter``, +``enable_adaptive_rate_limiting``)를 별도 Mixin으로 옮긴 것. + +분리 이유: +- 본 메서드들은 ``self.rate_limiter``에만 의존하며 다른 도메인 API와 결합이 없음 +- 모두 단순 위임 layer (예외 처리·비활성 경고 외에는 로직 없음) +- ``TechnicalAnalysisMixin``/``MethodDiscoveryMixin``과 동일한 컨벤션 +- ``Agent``의 LOC를 100줄 줄이고 도메인 API 코드와 인프라 제어 코드를 분리 + +사용자 입장의 API는 변경되지 않음 — ``agent.get_rate_limiter_status()`` +등은 동일하게 호출 가능. +""" + +import logging +from typing import Any, Dict, Optional + + +class RateLimiterControlMixin: + """Agent의 Rate Limiter를 외부에서 제어하기 위한 Mixin. + + 사용 측에서는 다음 속성이 셋업되어 있어야 한다: + + - ``self.rate_limiter``: ``RateLimiter`` 인스턴스 또는 ``None`` + (``Agent.__init__``에서 ``KISClient``의 rate limiter를 그대로 저장) + """ + + def get_rate_limiter_status(self) -> Optional[Dict[str, Any]]: + """ + Rate Limiter 상태 조회 + + Returns: + Dict: Rate Limiter 상태 정보 + - requests_per_second: 현재 초당 요청 수 + - requests_per_minute: 현재 분당 요청 수 + - limit_per_second: 초당 제한 + - limit_per_minute: 분당 제한 + - backoff_multiplier: 백오프 배수 + - total_requests: 총 요청 수 + - throttled_count: 제한된 요청 수 + - avg_wait_time: 평균 대기 시간 + None: Rate Limiter가 비활성화된 경우 + + Example: + >>> status = agent.get_rate_limiter_status() + >>> if status: + ... print(f"현재 요청률: {status['requests_per_second']}/초") + ... print(f"제한 도달 횟수: {status['throttled_count']}") + """ + if self.rate_limiter: + return self.rate_limiter.get_current_rate() + return None + + def set_rate_limits( + self, + requests_per_second: Optional[int] = None, + requests_per_minute: Optional[int] = None, + min_interval_ms: Optional[int] = None, + ) -> None: + """ + Rate Limiter 제한 값 동적 변경 + + Args: + requests_per_second: 초당 최대 요청 수 (None이면 변경 안 함) + requests_per_minute: 분당 최대 요청 수 (None이면 변경 안 함) + min_interval_ms: 최소 간격 (밀리초) (None이면 변경 안 함) + + Example: + >>> # 더 보수적인 설정으로 변경 + >>> agent.set_rate_limits( + ... requests_per_second=10, + ... requests_per_minute=500 + ... ) + >>> + >>> # 최소 간격만 변경 + >>> agent.set_rate_limits(min_interval_ms=100) + """ + if self.rate_limiter: + self.rate_limiter.set_limits( + requests_per_second=requests_per_second, + requests_per_minute=requests_per_minute, + min_interval_ms=min_interval_ms, + ) + logging.info("Rate limits 업데이트 완료") + else: + logging.warning("Rate Limiter가 비활성화 상태입니다") + + def reset_rate_limiter(self) -> None: + """ + Rate Limiter 상태 초기화 + + 모든 요청 기록과 통계를 초기화합니다. + 백오프 배수도 1.0으로 리셋됩니다. + + Example: + >>> agent.reset_rate_limiter() + >>> print("Rate limiter 초기화 완료") + """ + if self.rate_limiter: + self.rate_limiter.reset() + logging.info("Rate limiter 초기화 완료") + else: + logging.warning("Rate Limiter가 비활성화 상태입니다") + + def enable_adaptive_rate_limiting(self, enable: bool = True) -> None: + """ + 적응형 속도 조절 활성화/비활성화 + + Args: + enable: True면 활성화, False면 비활성화 + + Example: + >>> # 적응형 속도 조절 비활성화 + >>> agent.enable_adaptive_rate_limiting(False) + """ + if self.rate_limiter: + self.rate_limiter.enable_adaptive = enable + status = "활성화" if enable else "비활성화" + logging.info(f"적응형 속도 조절 {status}") + else: + logging.warning("Rate Limiter가 비활성화 상태입니다") + + +__all__ = ["RateLimiterControlMixin"]