Coverage for common / types.py: 92%
345 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-04 15:08 +0000
1# common/types.py
2from typing import Optional, Generic, TypeVar, Type, Any, Dict
3from enum import Enum, auto
4from pydantic import BaseModel, Field, model_validator, ConfigDict
6T = TypeVar("T")
9class Exchange(str, Enum):
10 KRX = "KRX"
11 NXT = "NXT"
12 UN = "UN" # 통합시세 (KRX+NXT, 시세 조회 전용)
15# API 응답 결과의 성공/실패를 나타내는 Enum
16class ErrorCode(Enum):
17 SUCCESS = "0"
18 API_ERROR = "100" # 외부 API 호출 실패
19 PARSING_ERROR = "101" # 응답 파싱 실패
20 INVALID_INPUT = "102" # 유효성 오류
21 NETWORK_ERROR = "103" # 네트워크 오류
22 MISSING_KEY = "104" # MISSING_KEY
23 RETRY_LIMIT = "105" # RETRY_LIMIT
24 WRONG_RET_TYPE = "106" # WRONG_RET_TYPE
25 EMPTY_VALUES = "107" # 조회 결과 없음
26 MARKET_CLOSED = "108" # 장 마감
27 UNKNOWN_ERROR = "999" # 기타 오류
29 @property
30 def is_retriable(self) -> bool:
31 """서비스 레벨에서 재시도 가능한 오류인지 반환."""
32 return self in (ErrorCode.NETWORK_ERROR, ErrorCode.RETRY_LIMIT)
35# --- 전략 신호 ---
36class TradeSignal(BaseModel):
37 """전략에서 생성하는 표준 매수/매도 신호."""
38 code: str
39 name: str
40 action: str # "BUY" / "SELL"
41 price: int
42 qty: int = 1
43 reason: str = ""
44 strategy_name: str = ""
45 exchange: str = "KRX" # "KRX" 또는 "NXT"
47 def to_dict(self):
48 return self.model_dump()
51# --- 공통적으로 사용되는 데이터 응답 구조 ---
52class ResPriceSummary(BaseModel):
53 symbol: str
54 open: int
55 current: int
56 change_rate: float
57 prdy_ctrt: float
58 new_high_low_status: Optional[str] = None
60 def to_dict(self):
61 return self.model_dump()
63 @classmethod
64 def from_dict(cls, data: dict):
65 return cls.model_validate(data)
68class ResMomentumStock(BaseModel):
69 symbol: str
70 change_rate: float
71 prev_volume: int
72 current_volume: int
74 def to_dict(self):
75 return self.model_dump()
77 @classmethod
78 def from_dict(cls, data: dict):
79 return cls.model_validate(data)
82class ResMarketCapStockItem(BaseModel):
83 rank: Optional[str]
84 name: Optional[str]
85 code: str
86 current_price: Optional[str]
88 def to_dict(self):
89 return self.model_dump()
91 @classmethod
92 def from_dict(cls, data: dict):
93 return cls.model_validate(data)
96# --- 한국투자증권 API 특화 응답 구조 ---
98# --- 한국투자증권 API 특화 응답 구조 (종목 상세정보) ---
101class ResStockFullInfoApiOutput(BaseModel):
102 acml_tr_pbmn: str = "" # 누적 거래 대금 (원)
103 acml_vol: str = "" # 누적 거래량 (주)
104 aspr_unit: str = "" # 호가 단위
105 bps: str = "" # 주당순자산 (Book-value Per Share)
106 bstp_kor_isnm: str = "" # 업종명 (예: 일반서비스)
107 clpr_rang_cont_yn: str = "" # 종가 범위제 적용 여부
108 cpfn: str = "" # 자본금 (억원)
109 cpfn_cnnm: str = "" # 자본금 (단위표기 포함, 예: 468 억)
110 crdt_able_yn: str = "" # 신용거래 가능 여부 (Y/N)
111 d250_hgpr: str = "" # 250일 최고가
112 d250_hgpr_date: str = "" # 250일 최고가 기록일
113 d250_hgpr_vrss_prpr_rate: str = "" # 현재가 대비 250일 최고가 등락률 (%)
114 d250_lwpr: str = "" # 250일 최저가
115 d250_lwpr_date: str = "" # 250일 최저가 기록일
116 d250_lwpr_vrss_prpr_rate: str = "" # 현재가 대비 250일 최저가 등락률 (%)
117 dmrs_val: str = "" # 매도호가 1 (최우선)
118 dmsp_val: str = "" # 매수호가 1 (최우선)
119 dryy_hgpr_date: str = "" # 당해 연도 최고가 기록일
120 dryy_hgpr_vrss_prpr_rate: str = "" # 현재가 대비 연중 최고가 등락률 (%)
121 dryy_lwpr_date: str = "" # 당해 연도 최저가 기록일
122 dryy_lwpr_vrss_prpr_rate: str = "" # 현재가 대비 연중 최저가 등락률 (%)
123 elw_pblc_yn: str = "" # ELW 발행 여부
124 eps: str = "" # 주당순이익 (Earnings Per Share)
125 fcam_cnnm: str = "" # 액면가 (단위 포함, 예: 500 원)
126 frgn_hldn_qty: str = "" # 외국인 보유 수량
127 frgn_ntby_qty: str = "" # 외국인 순매매 수량
128 grmn_rate_cls_code: str = "" # 결산월 분류코드
129 hts_avls: str = "" # HTS 시가총액 (원)
130 hts_deal_qty_unit_val: str = "" # HTS 거래량 단위 값
131 hts_frgn_ehrt: str = "" # HTS 외국인 보유율 (%)
132 invt_caful_yn: str = "" # 투자주의 종목 여부
133 iscd_stat_cls_code: str = "" # 종목 상태 코드
134 last_ssts_cntg_qty: str = "" # 직전 체결 수량
135 lstn_stcn: str = "" # 상장 주식수 (주)
136 mang_issu_cls_code: str = "" # 관리종목 구분 코드
137 marg_rate: str = "" # 증거금율 (%)
138 mrkt_warn_cls_code: str = "" # 시장경고종목 분류코드
139 new_hgpr_lwpr_cls_code: Optional[str] = None # 신고가/신저가 구분 코드
140 oprc_rang_cont_yn: str = "" # 시가 범위제 적용 여부
141 ovtm_vi_cls_code: str = "" # 시간외 VI 발동 분류코드 (NXT 응답에 미포함)
142 pbr: str = "" # 주가순자산비율 (Price to Book Ratio)
143 per: str = "" # 주가수익비율 (Price Earnings Ratio)
144 pgtr_ntby_qty: str = "" # 프로그램 매매 순매수 수량
145 prdy_ctrt: str = "" # 전일 대비 등락률 (%)
146 prdy_vrss: str = "" # 전일 대비 등락금액
147 prdy_vrss_sign: str = "" # 전일 대비 부호 (1:상승, 2:하락, 3:보합)
148 prdy_vrss_vol_rate: str = "" # 전일 대비 거래량 증감률 (%)
149 pvt_frst_dmrs_prc: str = "" # 예상체결가 첫번째 매도호가
150 pvt_frst_dmsp_prc: str = "" # 예상체결가 첫번째 매수호가
151 pvt_pont_val: str = "" # 예상체결가 기준 예상 체결가
152 pvt_scnd_dmrs_prc: str = "" # 예상체결가 두번째 매도호가
153 pvt_scnd_dmsp_prc: str = "" # 예상체결가 두번째 매수호가
154 rprs_mrkt_kor_name: str = "" # 대표시장명 (예: 코스피, 코스닥)
155 rstc_wdth_prc: str = "" # 가격제한폭 (상하한가 차이)
156 short_over_yn: str = "" # 공매도 과열 여부 (Y/N)
157 sltr_yn: str = "" # 정리매매 여부 (Y/N)
158 ssts_yn: str = "" # 정지 여부 (Y/N)
159 stac_month: str = "" # 결산월 (NXT 응답에 미포함)
160 stck_dryy_hgpr: str = "" # 당해 연도 최고가
161 stck_dryy_lwpr: str = "" # 당해 연도 최저가
162 stck_fcam: str = "" # 액면가
163 stck_hgpr: str # 금일 고가
164 stck_llam: str = "" # 종가 기준 시가총액 (원)
165 stck_lwpr: str # 금일 저가
166 stck_mxpr: str = "" # 상한가
167 stck_oprc: str # 시가
168 stck_prpr: str # 현재가
169 stck_sdpr: str # 기준가 (기준가격)
170 stck_shrn_iscd: str = "" # 단축 종목코드
171 stck_sspr: str = "" # 하한가
172 temp_stop_yn: str = "" # 일시 정지 여부 (Y/N)
173 vi_cls_code: str = "" # VI (변동성완화장치) 발동 여부
174 vol_tnrt: str = "" # 거래 회전율 (%)
175 w52_hgpr: str = "" # 52주 최고가
176 w52_hgpr_date: str = "" # 52주 최고가 기록일
177 w52_hgpr_vrss_prpr_ctrt: str = "" # 현재가 대비 52주 최고가 등락률 (%)
178 w52_lwpr: str = "" # 52주 최저가
179 w52_lwpr_date: str = "" # 52주 최저가 기록일
180 w52_lwpr_vrss_prpr_ctrt: str = "" # 현재가 대비 52주 최저가 등락률 (%)
181 wghn_avrg_stck_prc: str = "" # 가중평균주가
182 whol_loan_rmnd_rate: str = "" # 전체 대주잔고 비율 (%)
184 @property
185 def is_new_high(self) -> bool:
186 return self.new_hgpr_lwpr_cls_code in ("1", "신고가")
188 @property
189 def is_new_low(self) -> bool:
190 return self.new_hgpr_lwpr_cls_code in ("2", "신저가")
192 @property
193 def new_high_low_status(self) -> str:
194 if self.is_new_high:
195 return "신고가"
196 if self.is_new_low:
197 return "신저가"
198 return "-"
200 def to_dict(self):
201 return self.model_dump()
203 @classmethod
204 def from_dict(cls, data: dict, log_missing: bool = True) -> "ResStockFullInfoApiOutput":
205 return cls.model_validate(data)
208class ResTopMarketCapApiItem(BaseModel):
209 """
210 [시가총액 상위 종목 응답 아이템]
211 - 모든 수치는 API가 문자열(String)로 반환하므로 str 유지 (대형 정수/소수 정밀도 보존 목적)
212 - KIS 스펙(필드/의미/길이)을 주석으로 명시
213 """
215 # ── 필수/핵심 필드 ─────────────────────────────────────────────────────────────
216 mksc_shrn_iscd: str # 유가증권 단축 종목코드 (String, len<=9) e.g. '005930'
217 data_rank: str # 데이터 순위 (String, len<=10) e.g. '1'
218 hts_kor_isnm: str # HTS 한글 종목명 (String, len<=40) e.g. '삼성전자'
219 stck_avls: str # 시가총액 (String, len<=18) e.g. '467000000000000'
221 # ── 시세/등락 관련(옵셔널) ────────────────────────────────────────────────────
222 stck_prpr: Optional[str] = None # 주식 현재가 (String, len<=10)
223 prdy_vrss: Optional[str] = None # 전일 대비 (가격 차이) (String, len<=10)
224 prdy_vrss_sign: Optional[str] = None # 전일 대비 부호 (String, len<=1) e.g. '1', '2', '3' 등
225 prdy_ctrt: Optional[str] = None # 전일 대비율 (String, len<=82) e.g. '2.31'
227 # ── 거래/상장 관련(옵셔널) ────────────────────────────────────────────────────
228 acml_vol: Optional[str] = None # 누적 거래량 (String, len<=18)
229 lstn_stcn: Optional[str] = None # 상장 주수 (String, len<=18)
231 # ── 시장 비중(옵셔널) ────────────────────────────────────────────────────────
232 mrkt_whol_avls_rlim: Optional[str] = None # 시장 전체 시총 대비 비중 (String, len<=52)
234 # ── 과거 코드 호환용 Alias (옵셔널) ──────────────────────────────────────────
235 # 기존 일부 로직/테스트가 참조하던 필드들:
236 iscd: Optional[str] = None # (호환) 단축코드 별칭. 없으면 mksc_shrn_iscd로 채움
237 acc_trdvol: Optional[str] = None # (호환) 누적 거래량 별칭. 없으면 acml_vol로 채움
239 @model_validator(mode='after')
240 def sync_aliases(self) -> "ResTopMarketCapApiItem":
241 # 과거 호환: iscd가 없으면 mksc_shrn_iscd로 보완
242 if not self.iscd:
243 self.iscd = self.mksc_shrn_iscd
244 # 과거 호환: acc_trdvol <-> acml_vol 동기화
245 if self.acc_trdvol and not self.acml_vol:
246 self.acml_vol = self.acc_trdvol
247 elif self.acml_vol and not self.acc_trdvol:
248 self.acc_trdvol = self.acml_vol
249 return self
251 def to_dict(self):
252 """Dict 직렬화(테스트/로그용)."""
253 return self.model_dump()
255 @classmethod
256 def from_dict(cls, data: dict):
257 return cls.model_validate(data)
259 # 선택: 원시 API payload를 받아 alias 키까지 정규화하고 생성하고 싶다면 사용
260 @classmethod
261 def from_api(cls, payload: dict) -> "ResTopMarketCapApiItem":
262 """
263 - 공식 스펙 키 우선(mksc_shrn_iscd, data_rank, hts_kor_isnm, stck_avls, stck_prpr, prdy_vrss, prdy_vrss_sign,
264 prdy_ctrt, acml_vol, lstn_stcn, mrkt_whol_avls_rlim)
265 - 구키(alias)도 허용(iscd -> mksc_shrn_iscd, acc_trdvol -> acml_vol)
266 """
267 norm = dict(payload) if payload else {}
269 # 단축코드 alias
270 if "mksc_shrn_iscd" not in norm and "iscd" in norm: 270 ↛ 274line 270 didn't jump to line 274 because the condition on line 270 was always true
271 norm["mksc_shrn_iscd"] = norm.get("iscd")
273 # 거래량 alias
274 if "acml_vol" not in norm and "acc_trdvol" in norm: 274 ↛ 278line 274 didn't jump to line 278 because the condition on line 274 was always true
275 norm["acml_vol"] = norm.get("acc_trdvol")
277 # Pydantic 생성
278 return cls.model_validate(norm)
281class ResDailyChartApiItem(BaseModel):
282 stck_bsop_date: str
283 stck_oprc: str
284 stck_hgpr: str
285 stck_lwpr: str
286 stck_clpr: str
287 acml_vol: str = ""
289 def to_dict(self):
290 return self.model_dump()
292 @classmethod
293 def from_dict(cls, data: dict):
294 return cls.model_validate(data)
297class ResAccountBalanceApiOutput(BaseModel):
298 pdno: str
299 prdt_name: str
300 evlu_amt: str
302 def to_dict(self):
303 return self.model_dump()
305 @classmethod
306 def from_dict(cls, data: dict):
307 return cls.model_validate(data)
310class ResStockOrderApiOutput(BaseModel):
311 ordno: str
312 prdt_no: str
314 def to_dict(self):
315 return self.model_dump()
317 @classmethod
318 def from_dict(cls, data: dict):
319 return cls.model_validate(data)
322# 종목 요약 정보 응답 구조 (상승률 기반 필터링용 등)
323class ResBasicStockInfo(BaseModel):
324 code: str
325 name: str
326 # open_price: int
327 current_price: int
328 change_rate: float
329 prdy_ctrt: float
331 def to_dict(self):
332 return self.model_dump()
334 @classmethod
335 def from_dict(cls, data: dict):
336 return cls.model_validate(data)
339class ResFluctuation(BaseModel):
340 stck_shrn_iscd: str #주식 단축 종목코드
341 data_rank: str = "" #데이터 순위
342 hts_kor_isnm: str = "" #HTS 한글 종목명
343 stck_prpr: str #주식 현재가
344 prdy_vrss: str = "" #전일 대비
345 prdy_vrss_sign: str = "" #전일 대비 부호
346 prdy_ctrt: str = "" #전일 대비율
347 acml_vol: str = "" #누적 거래량
348 stck_hgpr: str = "" #주식 최고가
349 hgpr_hour: str = "" #최고가 시간
350 acml_hgpr_date: str = "" #누적 최고가 일자
351 stck_lwpr: str = "" #주식 최저가
352 lwpr_hour: str = "" #최저가 시간
353 acml_lwpr_date: str = "" #누적 최저가 일자
354 lwpr_vrss_prpr_rate: str = "" #최저가 대비 현재가 비율
355 dsgt_date_clpr_vrss_prpr_rate: str = "" #지정 일자 종가 대비 현재가 비
356 cnnt_ascn_dynu: str = "" #연속 상승 일수
357 hgpr_vrss_prpr_rate: str = "" #최고가 대비 현재가 비율
358 cnnt_down_dynu: str = "" #연속 하락 일수
359 oprc_vrss_prpr_sign: str = "" #시가2 대비 현재가 부호
360 oprc_vrss_prpr: str = "" #시가2 대비 현재가
361 oprc_vrss_prpr_rate: str = "" #시가2 대비 현재가 비율
362 prd_rsfl: str = "" #기간 등락
363 prd_rsfl_rate: str = "" #기간 등락 비율
365 def to_dict(self):
366 return self.model_dump()
368 @classmethod
369 def from_dict(cls, data: dict) -> "ResFluctuation":
370 # Pydantic handles missing fields if Optional, or we can use validator
371 return cls.model_validate(data)
374class ResBollingerBand(BaseModel):
375 code: str
376 date: str
377 close: Optional[float]
378 middle: Optional[float]
379 upper: Optional[float]
380 lower: Optional[float]
382 def to_dict(self):
383 return self.model_dump()
385 @classmethod
386 def from_dict(cls, data: dict):
387 return cls.model_validate(data)
390class ResRSI(BaseModel):
391 code: str
392 date: str
393 close: float
394 rsi: float
396 def to_dict(self):
397 return self.model_dump()
399 @classmethod
400 def from_dict(cls, data: dict):
401 return cls.model_validate(data)
404class ResMovingAverage(BaseModel):
405 code: str
406 date: str
407 close: float
408 ma: Optional[float]
410 def to_dict(self):
411 return self.model_dump()
413 @classmethod
414 def from_dict(cls, data: dict):
415 return cls.model_validate(data)
418class ResRelativeStrength(BaseModel):
419 """N일 수익률 (상대강도 원시값)."""
420 code: str
421 date: str
422 return_pct: float # N일 수익률 (%)
424 def to_dict(self):
425 return self.model_dump()
428# --- 공통 응답 구조 (유지 또는 dataclass로 래핑 가능) ---
430class ResCommonResponse(BaseModel, Generic[T]):
431 rt_cd: str
432 msg1: str
433 data: Optional[T] = None
435 def to_dict(self):
436 data_serialized = None
437 if hasattr(self.data, 'to_dict') and callable(getattr(self.data, 'to_dict')):
438 # data 필드 자체가 to_dict를 가진 객체인 경우
439 data_serialized = self.data.to_dict()
440 elif isinstance(self.data, BaseModel): 440 ↛ 441line 440 didn't jump to line 441 because the condition on line 440 was never true
441 data_serialized = self.data.model_dump()
442 elif isinstance(self.data, (list, tuple)):
443 # data 필드가 리스트/튜플인 경우, 내부 항목들도 재귀적으로 직렬화
444 data_serialized = []
445 for item in self.data:
446 if hasattr(item, 'to_dict') and callable(getattr(item, 'to_dict')):
447 data_serialized.append(item.to_dict())
448 elif isinstance(item, BaseModel): 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true
449 data_serialized.append(item.model_dump())
450 elif isinstance(item, (list, tuple, dict)): # 중첩된 리스트/딕셔너리도 처리
451 # 여기서 재귀적으로 self._serialize 같은 함수를 사용하면 좋지만,
452 # types.py에서는 cache_store._serialize를 직접 호출할 수 없으므로,
453 # to_dict를 가진 객체만 처리하거나, 더 일반적인 재귀 직렬화 로직을 이 안에 구현해야 함.
454 # 일단은 to_dict를 가진 객체만 처리하도록 간소화합니다.
455 data_serialized.append(item) # to_dict 없는 객체는 그대로
456 else:
457 data_serialized.append(item)
458 elif isinstance(self.data, dict):
459 # data 필드가 딕셔너리인 경우, 내부 값들도 재귀적으로 직렬화
460 data_serialized = {}
461 for k, v in self.data.items():
462 if hasattr(v, 'to_dict') and callable(getattr(v, 'to_dict')):
463 data_serialized[k] = v.to_dict()
464 elif isinstance(v, BaseModel): 464 ↛ 465line 464 didn't jump to line 465 because the condition on line 464 was never true
465 data_serialized[k] = v.model_dump()
466 elif isinstance(v, (list, tuple, dict)): # 중첩된 리스트/딕셔너리도 처리
467 data_serialized[k] = v # to_dict 없는 객체는 그대로
468 else:
469 data_serialized[k] = v
470 else:
471 # 그 외 기본 JSON 직렬화 가능 타입은 그대로 반환
472 data_serialized = self.data
474 return {
475 "rt_cd": self.rt_cd,
476 "msg1": self.msg1,
477 "data": data_serialized
478 }
480 @classmethod
481 def from_dict(cls, data: dict):
482 return cls.model_validate(data)