Coverage for brokers / korea_investment / korea_invest_params_provider.py: 96%
400 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-04 15:08 +0000
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-04 15:08 +0000
1from __future__ import annotations
2from dataclasses import dataclass, asdict
3from typing import Dict, Optional, Literal
4from core.market_clock import MarketClock
6MarketCode = Literal["J", "Q", "NX", "UN"] # J: 코스피(KRX), Q: 코스닥, NX: NXT, UN: 통합
7tm = MarketClock()
9# ---- 개별 파라미터 dataclass들 ----
11@dataclass(frozen=True)
12class SearchInfoParams:
13 pdno: str
14 prdt_type_cd: str # 종목코드
16 @classmethod
17 def of(cls, stock_code: str, prdt_type_cd: prdt_type_cd): # type: ignore
18 return cls(pdno=stock_code, prdt_type_cd=prdt_type_cd)
20 def to_dict(self) -> Dict[str, str]:
21 return asdict(self)
24@dataclass(frozen=True)
25class InquirePriceParams:
26 fid_cond_mrkt_div_code: str # "J"
27 fid_input_iscd: str # 종목코드
29 @classmethod
30 def of(cls, stock_code: str, market: MarketCode = "J"):
31 return cls(fid_cond_mrkt_div_code=market, fid_input_iscd=stock_code)
33 def to_dict(self) -> Dict[str, str]:
34 return asdict(self)
37@dataclass(frozen=True)
38class AskingPriceParams:
39 fid_cond_mrkt_div_code: str
40 fid_input_iscd: str
42 @classmethod
43 def of(cls, stock_code: str, market: MarketCode = "J"):
44 return cls(fid_cond_mrkt_div_code=market, fid_input_iscd=stock_code)
46 def to_dict(self) -> Dict[str, str]:
47 return asdict(self)
50@dataclass(frozen=True)
51class TimeConcludeParams:
52 fid_cond_mrkt_div_code: str
53 fid_input_iscd: str
55 @classmethod
56 def of(cls, stock_code: str, market: MarketCode = "J"):
57 return cls(fid_cond_mrkt_div_code=market, fid_input_iscd=stock_code)
59 def to_dict(self) -> Dict[str, str]:
60 return asdict(self)
63@dataclass(frozen=True)
64class DailyItemChartPriceParams:
65 fid_cond_mrkt_div_code: str # 보통 "J"
66 fid_input_iscd: str # 종목코드
67 fid_input_date_1: str # 조회 시작일
68 fid_input_date_2: str # 조회 종료일
69 fid_period_div_code: str # 일/주/월/년
70 fid_org_adj_prc: Literal["0", "1"] # 수정주가 여부
72 @classmethod
73 def daily_itemchartprice(cls, stock_code: str, start_date: str, end_date: str, period: str, market: MarketCode = "J", adj: Literal["0", "1"] = "0"):
74 return cls(
75 fid_cond_mrkt_div_code=market,
76 fid_input_iscd=stock_code,
77 fid_input_date_1=tm.to_yyyymmdd(start_date),
78 fid_input_date_2=tm.to_yyyymmdd(end_date),
79 fid_period_div_code=period,
80 fid_org_adj_prc=adj,
81 )
84 def to_dict(self) -> Dict[str, str]:
85 return asdict(self)
87@dataclass(frozen=True)
88class TimeItemChartPriceParams:
89 fid_cond_mrkt_div_code: str # 보통 "J"
90 fid_input_iscd: str # 종목코드
91 fid_input_hour_1: str # 입력시간
92 fid_pw_data_incu_yn: str # 과거 데이터 포함 여부
93 fid_etc_cls_code: str # 기타 구분 코드
95 @classmethod
96 def time_itemchartprice(cls, stock_code: str, fid_input_hour_1: str, fid_pw_data_incu_yn: str, fid_etc_cls_code: str, market: MarketCode = "J", adj: Literal["0", "1"] = "0"):
97 return cls(
98 fid_cond_mrkt_div_code=market,
99 fid_input_iscd=stock_code,
100 fid_input_hour_1=fid_input_hour_1,
101 fid_pw_data_incu_yn=fid_pw_data_incu_yn,
102 fid_etc_cls_code=fid_etc_cls_code
103 )
106 def to_dict(self) -> Dict[str, str]:
107 return asdict(self)
109@dataclass(frozen=True)
110class TimeDailyItemChartPriceParams:
111 fid_cond_mrkt_div_code: str # 보통 "J"
112 fid_input_iscd: str # 종목코드
113 fid_input_hour_1: str # 입력시간 (ex 13시 130000)
114 fid_input_date_1: str # 입력날짜 (YYYYMMDD)
115 fid_pw_data_incu_yn: str # 과거 데이터 포함 여부
116 fid_fake_tick_incu_yn: str # 허봉 포함 여부
118 @classmethod
119 def time_daily_itemchartprice(cls, stock_code: str, fid_input_hour_1: str, fid_input_date_1: str, fid_pw_data_incu_yn: str, fid_fake_tick_incu_yn: str, market: MarketCode = "J", adj: Literal["0", "1"] = "0"):
120 return cls(
121 fid_cond_mrkt_div_code=market,
122 fid_input_iscd=stock_code,
123 fid_input_hour_1=fid_input_hour_1,
124 fid_input_date_1=fid_input_date_1,
125 fid_pw_data_incu_yn=fid_pw_data_incu_yn,
126 fid_fake_tick_incu_yn=fid_fake_tick_incu_yn
127 )
130 def to_dict(self) -> Dict[str, str]:
131 return asdict(self)
134@dataclass(frozen=True)
135class VolumeRankParams:
136 FID_COND_MRKT_DIV_CODE: str
137 FID_COND_SCR_DIV_CODE: str
138 FID_INPUT_ISCD: str
139 FID_DIV_CLS_CODE: str
140 FID_BLNG_CLS_CODE: str
141 FID_TRGT_CLS_CODE: str
142 FID_TRGT_EXLS_CLS_CODE: str
143 FID_INPUT_PRICE_1: str
144 FID_INPUT_PRICE_2: str
145 FID_VOL_CNT: str
146 FID_INPUT_DATE_1: str
148 @classmethod
149 def default(cls, market: MarketCode = "J"):
150 return cls(
151 FID_COND_MRKT_DIV_CODE=market,
152 FID_COND_SCR_DIV_CODE="20171",
153 FID_INPUT_ISCD="0000",
154 FID_DIV_CLS_CODE="0",
155 FID_BLNG_CLS_CODE="0",
156 FID_TRGT_CLS_CODE="0",
157 FID_TRGT_EXLS_CLS_CODE="0000000000",
158 FID_INPUT_PRICE_1="",
159 FID_INPUT_PRICE_2="",
160 FID_VOL_CNT="",
161 FID_INPUT_DATE_1="",
162 )
164 @classmethod
165 def trading_value(cls, market: MarketCode = "J"):
166 """거래대금 순위 (HTS [0171] 화면의 거래대금 탭)"""
167 return cls(
168 FID_COND_MRKT_DIV_CODE=market,
169 FID_COND_SCR_DIV_CODE="20173",
170 FID_INPUT_ISCD="0000",
171 FID_DIV_CLS_CODE="0",
172 FID_BLNG_CLS_CODE="0",
173 FID_TRGT_CLS_CODE="0",
174 FID_TRGT_EXLS_CLS_CODE="0000000000",
175 FID_INPUT_PRICE_1="",
176 FID_INPUT_PRICE_2="",
177 FID_VOL_CNT="",
178 FID_INPUT_DATE_1="",
179 )
181 def to_dict(self) -> Dict[str, str]:
182 return asdict(self)
185@dataclass(frozen=True)
186class MarketCapScreenParams:
187 fid_cond_mrkt_div_code: str # "J"
188 fid_cond_scr_div_code: str # "20174" (시총 상위 등)
189 fid_div_cls_code: str # "0"
190 fid_input_iscd: str # "0000" (전체) 또는 업종코드
191 fid_trgt_cls_code: str # "20"
192 fid_trgt_exls_cls_code: str # "20"
193 fid_input_price_1: str
194 fid_input_price_2: str
195 fid_vol_cnt: str
197 @classmethod
198 def top_market_cap(cls, market: MarketCode = "J", input_iscd: str = "0000"):
199 return cls(
200 fid_cond_mrkt_div_code=market,
201 fid_cond_scr_div_code="20174",
202 fid_div_cls_code="0",
203 fid_input_iscd=input_iscd,
204 fid_trgt_cls_code="",
205 fid_trgt_exls_cls_code="",
206 fid_input_price_1="",
207 fid_input_price_2="",
208 fid_vol_cnt="",
209 )
211 def to_dict(self) -> Dict[str, str]:
212 return asdict(self)
215@dataclass(frozen=True)
216class SearchStockParams:
217 word: str
219 @classmethod
220 def of(cls, keyword: str):
221 return cls(word=keyword)
223 def to_dict(self) -> Dict[str, str]:
224 return asdict(self)
227@dataclass(frozen=True)
228class ItemNewsParams:
229 fid_input_iscd: str
231 @classmethod
232 def of(cls, stock_code: str):
233 return cls(fid_input_iscd=stock_code)
235 def to_dict(self) -> Dict[str, str]:
236 return asdict(self)
239@dataclass(frozen=True)
240class FinancialRatioParams:
241 """기업 재무비율 조회 파라미터. (KIS 문서 확인 필요)"""
242 fid_cond_mrkt_div_code: str # 시장구분코드 ("J": 주식)
243 fid_div_cls_code: str # "0" (전체)
244 fid_input_iscd: str # 종목코드
246 @classmethod
247 def of(cls, stock_code: str, market: MarketCode = "J"):
248 return cls(fid_cond_mrkt_div_code=market, fid_div_cls_code="0", fid_input_iscd=stock_code)
250 def to_dict(self) -> Dict[str, str]:
251 return asdict(self)
254@dataclass(frozen=True)
255class ETFInfoParams:
256 fid_cond_mrkt_div_code: str
257 fid_input_iscd: str
259 @classmethod
260 def of(cls, etf_code: str, market: MarketCode = "J"):
261 return cls(fid_cond_mrkt_div_code=market, fid_input_iscd=etf_code)
263 def to_dict(self) -> Dict[str, str]:
264 return asdict(self)
267@dataclass(frozen=True)
268class CheckHolidayParams:
269 BASS_DT: str # 기준일자 (YYYYMMDD)
270 CTX_AREA_NK: str = ""
271 CTX_AREA_FK: str = ""
273 @classmethod
274 def of(cls, date: str):
275 return cls(BASS_DT=date)
277 def to_dict(self) -> Dict[str, str]:
278 return asdict(self)
280@dataclass(frozen=True)
281class FluctuationParams:
282 fid_rsfl_rate2: str # 등락 비율2 ( ~비율 ) 공백이면 전체
283 fid_cond_mrkt_div_code: str # 시장구분 (주식 J)
284 fid_cond_scr_div_code: str # 조건 화면 분류 코드 (20170)
285 fid_input_iscd: str # 0000(전체) 0001(코스피) 1001(코스닥) 2001(코스피200)
286 fid_rank_sort_cls_code: str # 0:상승율 1:하락율 2:시가대비상승율 3:시가대비하락율 4:변동율
287 fid_input_cnt_1: str # 0:전체 또는 누적일수
288 fid_prc_cls_code: str # (정렬코드 0: 저가대비/종가대비, 1: 고가대비/종가대비, 기타: 0=전체)
289 fid_input_price_1: str # 가격 ~
290 fid_input_price_2: str # ~ 가격
291 fid_vol_cnt: str # 거래량 ~
292 fid_trgt_cls_code: str # 대상 (0:전체)
293 fid_trgt_exls_cls_code: str # 대상제외 (0:전체)
294 fid_div_cls_code: str # 분류 (0:전체)
295 fid_rsfl_rate1: str # 등락 비율1 (비율 ~)
297 def to_dict(self) -> Dict[str, str]:
298 return asdict(self)
300 # ---- 기본 빌더: 공백/기본값 채우기 ----
301 @staticmethod
302 def _base(
303 market: MarketCode = "J",
304 input_iscd: str = "0000",
305 rank_sort: Literal["0", "1", "2", "3", "4"] = "0",
306 prc_cls: Optional[str] = None,
307 input_cnt_1: str = "0",
308 rsfl_rate1: str = "",
309 rsfl_rate2: str = "",
310 price_1: str = "",
311 price_2: str = "",
312 vol_cnt: str = "",
313 trgt_cls: str = "0",
314 trgt_exls: str = "0",
315 div_cls: str = "0",
316 ) -> "FluctuationParams":
317 # rank_sort(정렬)에 따른 prc_cls 기본값
318 # 0:상승율순 -> 0:저가대비, 1:종가대비 (기본 '1' 권장)
319 # 1:하락율순 -> 0:고가대비, 1:종가대비 (기본 '1' 권장)
320 # 2/3/4 -> 0:전체
321 if prc_cls is None: 321 ↛ 327line 321 didn't jump to line 327 because the condition on line 321 was always true
322 if rank_sort in ("0", "1"): 322 ↛ 325line 322 didn't jump to line 325 because the condition on line 322 was always true
323 prc_cls = "1" # 종가대비 기본
324 else:
325 prc_cls = "0" # 전체
327 return FluctuationParams(
328 fid_rsfl_rate2=rsfl_rate2,
329 fid_cond_mrkt_div_code=market,
330 fid_cond_scr_div_code="20170",
331 fid_input_iscd=input_iscd,
332 fid_rank_sort_cls_code=rank_sort,
333 fid_input_cnt_1=input_cnt_1,
334 fid_prc_cls_code=prc_cls,
335 fid_input_price_1=price_1,
336 fid_input_price_2=price_2,
337 fid_vol_cnt=vol_cnt,
338 fid_trgt_cls_code=trgt_cls,
339 fid_trgt_exls_cls_code=trgt_exls,
340 fid_div_cls_code=div_cls,
341 fid_rsfl_rate1=rsfl_rate1,
342 )
344 # ---- 편의 빌더들 ----
345 @classmethod
346 def rising(
347 cls,
348 market: MarketCode = "J",
349 *,
350 input_iscd: str = "0000",
351 input_cnt_1: str = "0",
352 prc_cls: Optional[str] = None, # "0"(저가대비) / "1"(종가대비)
353 rsfl_rate1: str = "",
354 rsfl_rate2: str = "",
355 price_1: str = "",
356 price_2: str = "",
357 vol_cnt: str = "",
358 trgt_cls: str = "0",
359 trgt_exls: str = "0",
360 div_cls: str = "0",
361 ) -> "FluctuationParams":
362 return cls._base(
363 market=market, input_iscd=input_iscd,
364 rank_sort="0", prc_cls=prc_cls, input_cnt_1=input_cnt_1,
365 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
366 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
367 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls
368 )
370 @classmethod
371 def falling(
372 cls,
373 market: MarketCode = "J",
374 *,
375 input_iscd: str = "0000",
376 input_cnt_1: str = "0",
377 prc_cls: Optional[str] = None, # "0"(고가대비) / "1"(종가대비)
378 rsfl_rate1: str = "",
379 rsfl_rate2: str = "",
380 price_1: str = "",
381 price_2: str = "",
382 vol_cnt: str = "",
383 trgt_cls: str = "0",
384 trgt_exls: str = "0",
385 div_cls: str = "0",
386 ) -> "FluctuationParams":
387 return cls._base(
388 market=market, input_iscd=input_iscd,
389 rank_sort="1", prc_cls=prc_cls, input_cnt_1=input_cnt_1,
390 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
391 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
392 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls
393 )
395 @classmethod
396 def since_open_rise(
397 cls,
398 market: MarketCode = "J",
399 *,
400 input_iscd: str = "0000",
401 input_cnt_1: str = "0",
402 prc_cls: Optional[str] = None, # "0"(저가대비) / "1"(종가대비)
403 rsfl_rate1: str = "",
404 rsfl_rate2: str = "",
405 price_1: str = "",
406 price_2: str = "",
407 vol_cnt: str = "",
408 trgt_cls: str = "0",
409 trgt_exls: str = "0",
410 div_cls: str = "0",
411 ) -> "FluctuationParams":
412 # 2:시가대비상승율
413 return cls._base(market=market, input_iscd=input_iscd, rank_sort="2", prc_cls=prc_cls, input_cnt_1=input_cnt_1,
414 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
415 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
416 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls)
418 @classmethod
419 def since_open_fall(
420 cls,
421 market: MarketCode = "J",
422 *,
423 input_iscd: str = "0000",
424 input_cnt_1: str = "0",
425 prc_cls: Optional[str] = None, # "0"(저가대비) / "1"(종가대비)
426 rsfl_rate1: str = "",
427 rsfl_rate2: str = "",
428 price_1: str = "",
429 price_2: str = "",
430 vol_cnt: str = "",
431 trgt_cls: str = "0",
432 trgt_exls: str = "0",
433 div_cls: str = "0",
434 ) -> "FluctuationParams":
435 # 3:시가대비하락율
436 return cls._base(market=market, input_iscd=input_iscd, rank_sort="3", prc_cls=prc_cls, input_cnt_1=input_cnt_1,
437 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
438 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
439 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls)
441 @classmethod
442 def volatility(
443 cls,
444 market: MarketCode = "J",
445 *,
446 input_iscd: str = "0000",
447 input_cnt_1: str = "0",
448 prc_cls: Optional[str] = None, # "0"(저가대비) / "1"(종가대비)
449 rsfl_rate1: str = "",
450 rsfl_rate2: str = "",
451 price_1: str = "",
452 price_2: str = "",
453 vol_cnt: str = "",
454 trgt_cls: str = "0",
455 trgt_exls: str = "0",
456 div_cls: str = "0",
457 ) -> "FluctuationParams":
458 # 4:변동율
459 return cls._base(market=market, input_iscd=input_iscd, rank_sort="4", prc_cls=prc_cls, input_cnt_1=input_cnt_1,
460 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
461 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
462 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls)
465@dataclass(frozen=True)
466class InvestorTradeByStockDailyParams:
467 """종목별 투자자 매매동향(일별) 파라미터"""
468 FID_COND_MRKT_DIV_CODE: str # 시장구분코드 (J:KRX, NX:NXT, UN:통합)
469 FID_INPUT_ISCD: str # 종목코드 (6자리)
470 FID_INPUT_DATE_1: str # 입력 날짜 (YYYYMMDD)
471 FID_ORG_ADJ_PRC: str # 수정주가 원주가 가격 (공란)
472 FID_ETC_CLS_CODE: str # 기타 구분 코드 ("1")
474 @classmethod
475 def of(cls, stock_code: str, date: str, market: str = "J"):
476 return cls(
477 FID_COND_MRKT_DIV_CODE=market,
478 FID_INPUT_ISCD=stock_code,
479 FID_INPUT_DATE_1=date,
480 FID_ORG_ADJ_PRC="",
481 FID_ETC_CLS_CODE="1",
482 )
484 def to_dict(self) -> Dict[str, str]:
485 return asdict(self)
488@dataclass(frozen=True)
489class ProgramTradeByStockDailyParams:
490 """종목별 프로그램매매추이(일별) 파라미터"""
491 FID_COND_MRKT_DIV_CODE: str # 시장구분코드 (J:KRX, NX:NXT, UN:통합)
492 FID_INPUT_ISCD: str # 종목코드 (6자리)
493 FID_INPUT_DATE_1: str # 입력 날짜 (YYYYMMDD)
495 @classmethod
496 def of(cls, stock_code: str, date: str, market: str = "J"):
497 return cls(
498 FID_COND_MRKT_DIV_CODE=market,
499 FID_INPUT_ISCD=stock_code,
500 FID_INPUT_DATE_1=date,
501 )
503 def to_dict(self) -> Dict[str, str]:
504 return asdict(self)
507@dataclass(frozen=True)
508class MultiPriceParams:
509 """복수종목 현재가 조회 파라미터 (최대 30종목)"""
510 fid_cond_mrkt_div_code_1: str
511 fid_input_iscd_1: str
512 fid_cond_mrkt_div_code_2: str = ""
513 fid_input_iscd_2: str = ""
514 fid_cond_mrkt_div_code_3: str = ""
515 fid_input_iscd_3: str = ""
516 fid_cond_mrkt_div_code_4: str = ""
517 fid_input_iscd_4: str = ""
518 fid_cond_mrkt_div_code_5: str = ""
519 fid_input_iscd_5: str = ""
520 fid_cond_mrkt_div_code_6: str = ""
521 fid_input_iscd_6: str = ""
522 fid_cond_mrkt_div_code_7: str = ""
523 fid_input_iscd_7: str = ""
524 fid_cond_mrkt_div_code_8: str = ""
525 fid_input_iscd_8: str = ""
526 fid_cond_mrkt_div_code_9: str = ""
527 fid_input_iscd_9: str = ""
528 fid_cond_mrkt_div_code_10: str = ""
529 fid_input_iscd_10: str = ""
530 fid_cond_mrkt_div_code_11: str = ""
531 fid_input_iscd_11: str = ""
532 fid_cond_mrkt_div_code_12: str = ""
533 fid_input_iscd_12: str = ""
534 fid_cond_mrkt_div_code_13: str = ""
535 fid_input_iscd_13: str = ""
536 fid_cond_mrkt_div_code_14: str = ""
537 fid_input_iscd_14: str = ""
538 fid_cond_mrkt_div_code_15: str = ""
539 fid_input_iscd_15: str = ""
540 fid_cond_mrkt_div_code_16: str = ""
541 fid_input_iscd_16: str = ""
542 fid_cond_mrkt_div_code_17: str = ""
543 fid_input_iscd_17: str = ""
544 fid_cond_mrkt_div_code_18: str = ""
545 fid_input_iscd_18: str = ""
546 fid_cond_mrkt_div_code_19: str = ""
547 fid_input_iscd_19: str = ""
548 fid_cond_mrkt_div_code_20: str = ""
549 fid_input_iscd_20: str = ""
550 fid_cond_mrkt_div_code_21: str = ""
551 fid_input_iscd_21: str = ""
552 fid_cond_mrkt_div_code_22: str = ""
553 fid_input_iscd_22: str = ""
554 fid_cond_mrkt_div_code_23: str = ""
555 fid_input_iscd_23: str = ""
556 fid_cond_mrkt_div_code_24: str = ""
557 fid_input_iscd_24: str = ""
558 fid_cond_mrkt_div_code_25: str = ""
559 fid_input_iscd_25: str = ""
560 fid_cond_mrkt_div_code_26: str = ""
561 fid_input_iscd_26: str = ""
562 fid_cond_mrkt_div_code_27: str = ""
563 fid_input_iscd_27: str = ""
564 fid_cond_mrkt_div_code_28: str = ""
565 fid_input_iscd_28: str = ""
566 fid_cond_mrkt_div_code_29: str = ""
567 fid_input_iscd_29: str = ""
568 fid_cond_mrkt_div_code_30: str = ""
569 fid_input_iscd_30: str = ""
571 @classmethod
572 def of(cls, stock_codes: list[str], market: MarketCode = "J") -> "MultiPriceParams":
573 """종목코드 리스트로 파라미터 생성 (최대 30개)"""
574 if not stock_codes: 574 ↛ 575line 574 didn't jump to line 575 because the condition on line 574 was never true
575 raise ValueError("stock_codes가 비어 있습니다.")
576 if len(stock_codes) > 30: 576 ↛ 577line 576 didn't jump to line 577 because the condition on line 576 was never true
577 raise ValueError(f"최대 30종목까지 조회 가능합니다. (요청: {len(stock_codes)}개)")
578 kwargs = {}
579 for i, code in enumerate(stock_codes, start=1):
580 kwargs[f"fid_cond_mrkt_div_code_{i}"] = market
581 kwargs[f"fid_input_iscd_{i}"] = code
582 return cls(**kwargs)
584 def to_dict(self) -> Dict[str, str]:
585 return {k: v for k, v in asdict(self).items() if v}
588@dataclass(frozen=True)
589class AccountBalanceParams:
590 CANO: str # 계좌번호 앞 8자리
591 ACNT_PRDT_CD: str # 계좌상품코드 (기본: "01")
592 AFHR_FLPR_YN: str = "N" # 시간외단일가 포함여부
593 FNCG_AMT_AUTO_RDPT_YN: str = "N" # 융자금액자동상환여부
594 FUND_STTL_ICLD_YN: str = "N" # 펀드결제분포함여부
595 INQR_DVSN: str = "01" # 조회구분코드 (기본: 01)
596 OFL_YN: str = "N" # 오프라인여부
597 PRCS_DVSN: str = "01" # 처리구분코드
598 UNPR_DVSN: str = "01" # 단가구분코드
599 CTX_AREA_FK100: str = "" # 연속조회검색조건100
600 CTX_AREA_NK100: str = "" # 연속조회키100
602 def to_dict(self) -> Dict[str, str]:
603 return asdict(self)
605 @classmethod
606 def create(
607 cls,
608 cano: str,
609 acnt_prdt_cd: str = "01",
610 afhr_flpr_yn: str = "N",
611 fncg_amt_auto_rdpt_yn: str = "N",
612 fund_sttl_icld_yn: str = "N",
613 inqr_dvsn: str = "01",
614 ofl_yn: str = "N",
615 prcs_dvsn: str = "01",
616 unpr_dvsn: str = "01",
617 ctx_area_fk100: str = "",
618 ctx_area_nk100: str = "",
619 ) -> "AccountBalanceParams":
620 return cls(
621 CANO=cano,
622 ACNT_PRDT_CD=acnt_prdt_cd,
623 AFHR_FLPR_YN=afhr_flpr_yn,
624 FNCG_AMT_AUTO_RDPT_YN=fncg_amt_auto_rdpt_yn,
625 FUND_STTL_ICLD_YN=fund_sttl_icld_yn,
626 INQR_DVSN=inqr_dvsn,
627 OFL_YN=ofl_yn,
628 PRCS_DVSN=prcs_dvsn,
629 UNPR_DVSN=unpr_dvsn,
630 CTX_AREA_FK100=ctx_area_fk100,
631 CTX_AREA_NK100=ctx_area_nk100,
632 )
635@dataclass(frozen=True)
636class OrderCashBody:
637 CANO: str
638 ACNT_PRDT_CD: str
639 PDNO: str
640 ORD_DVSN: str # 00 지정가 / 01 시장가 ...
641 ORD_QTY: str
642 ORD_UNPR: str # 시장가일 땐 "0" 등 규약대로
643 EXCG_ID_DVSN_CD: str = "" # 거래소 ID 구분 코드 (NXT: "NX", KRX: "" 또는 "KRX")
644 SLL_TYPE: str = "" # 매도 유형 (NXT 전용)
645 CNDT_PRIC: str = "" # 조건가격 (NXT 전용)
647 def to_dict(self):
648 return asdict(self)
651# ---- 얇은 파사드: 기존 코드에서 함수 호출만으로 dict를 얻을 수 있게 ----
653class Params:
654 """기존 코드 변경 최소화를 위한 dict 파사드"""
656 @staticmethod
657 def search_info(stock_code: str, prdt_type_cd: MarketCode) -> Dict[str, str]:
658 return SearchInfoParams.of(stock_code, prdt_type_cd).to_dict()
660 @staticmethod
661 def inquire_price(stock_code: str, market: MarketCode = "J") -> Dict[str, str]:
662 return InquirePriceParams.of(stock_code, market).to_dict()
664 @staticmethod
665 def inquire_conclusion(stock_code: str, market: MarketCode = "J") -> Dict[str, str]:
666 return InquirePriceParams.of(stock_code, market).to_dict()
668 @staticmethod
669 def asking_price(stock_code: str, market: MarketCode = "J") -> Dict[str, str]:
670 return AskingPriceParams.of(stock_code, market).to_dict()
672 @staticmethod
673 def time_conclude(stock_code: str, market: MarketCode = "J") -> Dict[str, str]:
674 return TimeConcludeParams.of(stock_code, market).to_dict()
676 @staticmethod
677 def daily_itemchartprice(stock_code: str, start_date: str, end_date: str, period: str, market: MarketCode = "J", adj: Literal["0", "1"] = "0") -> \
678 Dict[str, str]:
679 return DailyItemChartPriceParams.daily_itemchartprice(stock_code, start_date, end_date, period, market, adj).to_dict()
681 @staticmethod
682 def time_itemchartprice(stock_code: str, input_hour: str,
683 include_past: str, etc_cls_code: str, market: MarketCode = "UN") -> dict:
684 return TimeItemChartPriceParams.time_itemchartprice(stock_code, input_hour, include_past, etc_cls_code, market).to_dict()
686 @staticmethod
687 def time_daily_itemchartprice(stock_code: str, input_hour: str, input_date: str, include_past: str, fid_pw_data_incu_yn: str, market: MarketCode = "UN") -> \
688 Dict[str, str]:
689 return TimeDailyItemChartPriceParams.time_daily_itemchartprice(stock_code, input_hour, input_date, include_past, fid_pw_data_incu_yn, market).to_dict()
691 @staticmethod
692 def volume_rank(market: MarketCode = "J") -> Dict[str, str]:
693 return VolumeRankParams.default(market).to_dict()
695 @staticmethod
696 def trading_value_rank(market: MarketCode = "J") -> Dict[str, str]:
697 return VolumeRankParams.trading_value(market).to_dict()
701 @staticmethod
702 def top_market_cap(market: MarketCode = "J", input_iscd: str = "0000") -> Dict[str, str]:
703 return MarketCapScreenParams.top_market_cap(market, input_iscd).to_dict()
705 @staticmethod
706 def search_stock(keyword: str) -> Dict[str, str]:
707 return SearchStockParams.of(keyword).to_dict()
709 @staticmethod
710 def item_news(stock_code: str) -> Dict[str, str]:
711 return ItemNewsParams.of(stock_code).to_dict()
713 @staticmethod
714 def financial_ratio(stock_code: str, market: MarketCode = "J") -> Dict[str, str]:
715 return FinancialRatioParams.of(stock_code, market).to_dict()
717 @staticmethod
718 def check_holiday(date: str) -> Dict[str, str]:
719 return CheckHolidayParams.of(date).to_dict()
721 @staticmethod
722 def etf_info(etf_code: str, market: MarketCode = "J") -> Dict[str, str]:
723 return ETFInfoParams.of(etf_code, market).to_dict()
725 @staticmethod
726 def fluctuation_rise(
727 market: MarketCode = "J",
728 *,
729 input_iscd: str = "0000",
730 input_cnt_1: str = "0",
731 prc_cls: Optional[str] = None, # 0/1
732 rsfl_rate1: str = "",
733 rsfl_rate2: str = "",
734 price_1: str = "",
735 price_2: str = "",
736 vol_cnt: str = "",
737 trgt_cls: str = "0",
738 trgt_exls: str = "0",
739 div_cls: str = "0",
740 ) -> Dict[str, str]:
741 return FluctuationParams.rising(
742 market, input_iscd=input_iscd, input_cnt_1=input_cnt_1, prc_cls=prc_cls,
743 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
744 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
745 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls
746 ).to_dict()
748 @staticmethod
749 def fluctuation_fall(
750 market: MarketCode = "J",
751 *,
752 input_iscd: str = "0000",
753 input_cnt_1: str = "0",
754 prc_cls: Optional[str] = None, # 0/1
755 rsfl_rate1: str = "",
756 rsfl_rate2: str = "",
757 price_1: str = "",
758 price_2: str = "",
759 vol_cnt: str = "",
760 trgt_cls: str = "0",
761 trgt_exls: str = "0",
762 div_cls: str = "0",
763 ) -> Dict[str, str]:
764 return FluctuationParams.falling(
765 market, input_iscd=input_iscd, input_cnt_1=input_cnt_1, prc_cls=prc_cls,
766 rsfl_rate1=rsfl_rate1, rsfl_rate2=rsfl_rate2,
767 price_1=price_1, price_2=price_2, vol_cnt=vol_cnt,
768 trgt_cls=trgt_cls, trgt_exls=trgt_exls, div_cls=div_cls
769 ).to_dict()
771 @staticmethod
772 def investor_trade_by_stock_daily(stock_code: str, date: str, market: str = "J") -> Dict[str, str]:
773 return InvestorTradeByStockDailyParams.of(stock_code, date, market).to_dict()
775 @staticmethod
776 def program_trade_by_stock_daily(stock_code: str, date: str, market: str = "J") -> Dict[str, str]:
777 return ProgramTradeByStockDailyParams.of(stock_code, date, market).to_dict()
779 @staticmethod
780 def multi_price(stock_codes: list[str], market: MarketCode = "J") -> Dict[str, str]:
781 return MultiPriceParams.of(stock_codes, market).to_dict()
783 @staticmethod
784 def account_balance(
785 cano: str,
786 acnt_prdt_cd: str = "01",
787 afhr_flpr_yn: str = "N",
788 fncg_amt_auto_rdpt_yn: str = "N",
789 fund_sttl_icld_yn: str = "N",
790 inqr_dvsn: str = "01",
791 ofl_yn: str = "N",
792 prcs_dvsn: str = "01",
793 unpr_dvsn: str = "01",
794 ctx_area_fk100: str = "",
795 ctx_area_nk100: str = "",
796 ) -> Dict[str, str]:
797 return AccountBalanceParams.create(
798 cano, acnt_prdt_cd,
799 afhr_flpr_yn, fncg_amt_auto_rdpt_yn,
800 fund_sttl_icld_yn, inqr_dvsn, ofl_yn,
801 prcs_dvsn, unpr_dvsn,
802 ctx_area_fk100, ctx_area_nk100
803 ).to_dict()
805 @staticmethod
806 def order_cash_body(*, cano: str, acnt_prdt_cd: str, pdno: str,
807 ord_dvsn: str, ord_qty: str | int, ord_unpr: str | int,
808 excg_id_dvsn_cd: str = "", sll_type: str = "", cndt_pric: str = "") -> dict:
809 return OrderCashBody(
810 CANO=cano,
811 ACNT_PRDT_CD=acnt_prdt_cd,
812 PDNO=pdno,
813 ORD_DVSN=ord_dvsn,
814 ORD_QTY=str(ord_qty),
815 ORD_UNPR=str(ord_unpr),
816 EXCG_ID_DVSN_CD=excg_id_dvsn_cd,
817 SLL_TYPE=sll_type,
818 CNDT_PRIC=cndt_pric,
819 ).to_dict()