Coverage for view / web / routes / ranking.py: 100%

1""" 

2랭킹/시가총액 관련 API 엔드포인트 (ranking.html, marketcap.html). 

3""" 

4from fastapi import APIRouter, HTTPException 

5from common.types import ErrorCode 

6from view.web.api_common import _get_ctx, _serialize_response, _serialize_list_items 

7 

8router = APIRouter() 

9 

10 

11@router.get("/ranking/progress") 

12async def get_ranking_progress(): 

13 """투자자 랭킹 수집 진행률 조회.""" 

14 ctx = _get_ctx() 

15 if ctx.ranking_task: 

16 return ctx.ranking_task.get_investor_ranking_progress() 

17 return {"running": False, "processed": 0, "total": 0, "collected": 0, "elapsed": 0.0} 

18 

19 

20@router.get("/ranking/{category}") 

21async def get_ranking(category: str): 

22 """랭킹 조회 (rise/fall/volume/trading_value/foreign_buy/foreign_sell/inst_buy/inst_sell/prsn_buy/prsn_sell).""" 

23 ctx = _get_ctx() 

24 t_start = ctx.pm.start_timer() 

25 valid = ("rise", "fall", "volume", "trading_value", 

26 "foreign_buy", "foreign_sell", "inst_buy", "inst_sell", "prsn_buy", "prsn_sell", 

27 "program_buy", "program_sell") 

28 if category not in valid: 

29 raise HTTPException(status_code=400, detail=f"category는 {', '.join(valid)} 중 하나여야 합니다.") 

30 

31 resp = await ctx.stock_query_service.handle_get_top_stocks(category) 

32 

33 ctx.pm.log_timer(f"get_ranking({category})", t_start) 

34 if resp and resp.rt_cd == ErrorCode.SUCCESS.value: 

35 return { 

36 "rt_cd": resp.rt_cd, 

37 "msg1": resp.msg1, 

38 "data": _serialize_list_items(resp.data) 

39 } 

40 return _serialize_response(resp) 

41 

42 

43@router.get("/top-market-cap") 

44async def get_top_market_cap(limit: int = 20, market: str = "0001"): 

45 """시가총액 상위 종목. market: 0001=거래소(코스피), 1001=코스닥""" 

46 ctx = _get_ctx() 

47 t_start = ctx.pm.start_timer() 

48 if market not in ("0000", "0001", "1001", "2001"): 

49 market = "0001" 

50 resp = await ctx.broker.get_top_market_cap_stocks_code(market, limit) 

51 ctx.pm.log_timer(f"get_top_market_cap({market}, {limit})", t_start) 

52 if resp and resp.rt_cd == ErrorCode.SUCCESS.value: 

53 items = resp.data or [] 

54 data = [] 

55 for idx, item in enumerate(items, 1): 

56 get = (lambda k: getattr(item, k, None)) if not isinstance(item, dict) else item.get 

57 data.append({ 

58 "rank": str(idx), 

59 "name": get("hts_kor_isnm") or "", 

60 "code": get("mksc_shrn_iscd") or get("iscd") or "", 

61 "current_price": get("stck_prpr") or "0", 

62 "change_rate": get("prdy_ctrt") or "0", 

63 "market_cap": get("stck_avls") or "0", 

64 }) 

65 return {"rt_cd": resp.rt_cd, "msg1": resp.msg1, "data": data} 

66 return _serialize_response(resp)