Coverage for view / web / routes / stock.py: 95%

1""" 

2종목 조회 관련 API 엔드포인트 (index.html). 

3현재가, 차트(OHLCV), 기술 지표, 시장 상태, 환경 전환. 

4""" 

5import asyncio 

6import time 

7from fastapi import APIRouter, HTTPException, Query 

8from common.types import Exchange 

9from view.web.api_common import _get_ctx, _serialize_response, EnvironmentRequest 

10import view.web.api_common as api_common 

11 

12router = APIRouter() 

13 

14# /api/status 결과를 5초간 캐시하여 페이지 전환 시 broker API 반복 호출 방지 

15_status_cache = None 

16_status_cache_ts = 0.0 

17_STATUS_CACHE_TTL = 5.0 

18 

19 

20@router.get("/status") 

21async def get_status(): 

22 """시장 상태 및 환경 정보.""" 

23 global _status_cache, _status_cache_ts 

24 

25 ctx = _get_ctx() 

26 if ctx is None: 

27 return {"market_open": False, "env_type": "미설정", "current_time": "", "initialized": False} 

28 

29 now = time.monotonic() 

30 if _status_cache is not None and (now - _status_cache_ts) < _STATUS_CACHE_TTL: 

31 # 캐시된 결과 반환 (현재 시각만 갱신) 

32 _status_cache["current_time"] = ctx.get_current_time_str() 

33 return _status_cache 

34 

35 result = { 

36 "market_open": await ctx.is_market_open_now(), 

37 "env_type": ctx.get_env_type(), 

38 "current_time": ctx.get_current_time_str(), 

39 "initialized": ctx.initialized 

40 } 

41 _status_cache = result 

42 _status_cache_ts = now 

43 return result 

44 

45 

46@router.get("/stocks/list") 

47async def get_stocks_list(): 

48 """전 종목 리스트 반환 (클라이언트 자동완성용, localStorage 캐싱 대상).""" 

49 ctx = _get_ctx() 

50 stock_list = [ 

51 {"c": code, "n": name} 

52 for name, code in ctx.stock_code_repository.name_to_code.items() 

53 ] 

54 return {"stocks": stock_list, "count": len(stock_list)} 

55 

56 

57@router.get("/stock/search") 

58async def search_stock_by_name(q: str = ""): 

59 """종목명 부분 일치 검색 (자동완성용).""" 

60 ctx = _get_ctx() 

61 if not q or len(q.strip()) == 0: 

62 return {"results": []} 

63 results = ctx.stock_code_repository.search_by_name(q.strip()) 

64 return {"results": results} 

65 

66 

67@router.get("/stock/{code}") 

68async def get_stock_price(code: str, exchange: str = Query("KRX")): 

69 """현재가 조회. 종목명이 들어오면 종목코드로 변환 후 조회. exchange=KRX|NXT|UN 선택 가능.""" 

70 ctx = _get_ctx() 

71 # 숫자가 아닌 입력(종목명)이면 코드로 변환 

72 if not code.isdigit(): 

73 resolved = ctx.stock_code_repository.get_code_by_name(code) 

74 if not resolved: 74 ↛ 76line 74 didn't jump to line 76 because the condition on line 74 was always true

75 return {"rt_cd": "1", "msg1": f"종목명 '{code}'에 해당하는 종목코드를 찾을 수 없습니다.", "data": None} 

76 code = resolved 

77 try: 

78 exchange_enum = Exchange(exchange.upper()) 

79 except ValueError: 

80 exchange_enum = Exchange.KRX 

81 t_start = ctx.pm.start_timer() 

82 resp = await ctx.stock_query_service.handle_get_current_stock_price(code, caller="stock.py - get_stock_price", exchange=exchange_enum) 

83 result = _serialize_response(resp) 

84 

85 ctx.pm.log_timer(f"get_stock_price({code})", t_start) 

86 

87 # 현재가 조회 후 OHLCV 2년치 백그라운드 프리로드 (캐시 miss 시에만 실제 API 호출) 

88 async def _preload_ohlcv(): 

89 try: 

90 await ctx.stock_query_service.get_ohlcv(code, caller="preload_on_price_query") 

91 except Exception: 

92 pass 

93 asyncio.create_task(_preload_ohlcv()) 

94 

95 return result 

96 

97 

98@router.get("/chart/{code}") 

99async def get_stock_chart(code: str, period: str = "D", indicators: bool = False, exchange: str = Query("KRX")): 

100 """종목의 OHLCV 차트 데이터 조회 (기본 일봉). indicators=true 시 MA+BB 지표 포함. exchange=KRX|NXT|UN 선택 가능.""" 

101 ctx = _get_ctx() 

102 try: 

103 exchange_enum = Exchange(exchange.upper()) 

104 except ValueError: 

105 exchange_enum = Exchange.KRX 

106 t_start = ctx.pm.start_timer() 

107 if indicators: 

108 resp = await ctx.stock_query_service.get_ohlcv_with_indicators(code, period, caller="stock.py - get_stock_chart") 

109 else: 

110 resp = await ctx.stock_query_service.get_ohlcv(code, period, caller="stock.py - get_stock_chart", exchange=exchange_enum) 

111 result = _serialize_response(resp) 

112 

113 ctx.pm.log_timer(f"get_stock_chart({code}, indicators={indicators})", t_start) 

114 return result 

115 

116 

117@router.get("/indicator/bollinger/{code}") 

118async def get_bollinger_bands(code: str, period: int = 20, std_dev: float = 2.0): 

119 """볼린저 밴드 조회 (기본: 20일, 2표준편차)""" 

120 ctx = _get_ctx() 

121 t_start = ctx.pm.start_timer() 

122 resp = await ctx.indicator_service.get_bollinger_bands(code, period, std_dev) 

123 result = _serialize_response(resp) 

124 

125 ctx.pm.log_timer(f"get_bollinger_bands({code})", t_start) 

126 return result 

127 

128 

129@router.get("/indicator/rsi/{code}") 

130async def get_rsi(code: str, period: int = 14): 

131 """RSI 조회 (기본: 14일)""" 

132 ctx = _get_ctx() 

133 t_start = ctx.pm.start_timer() 

134 resp = await ctx.indicator_service.get_rsi(code, period) 

135 result = _serialize_response(resp) 

136 

137 ctx.pm.log_timer(f"get_rsi({code})", t_start) 

138 return result 

139 

140 

141@router.get("/indicator/ma/{code}") 

142async def get_moving_average(code: str, period: int = 20, method: str = "sma"): 

143 """이동평균선 조회 (기본: 20일, sma)""" 

144 ctx = _get_ctx() 

145 t_start = ctx.pm.start_timer() 

146 resp = await ctx.indicator_service.get_moving_average(code, period, method) 

147 result = _serialize_response(resp) 

148 

149 ctx.pm.log_timer(f"get_moving_average({code})", t_start) 

150 return result 

151 

152 

153@router.post("/environment") 

154async def change_environment(req: EnvironmentRequest): 

155 """거래 환경 변경 (모의/실전).""" 

156 global _status_cache, _status_cache_ts 

157 ctx = api_common._ctx 

158 if ctx is None: 

159 raise HTTPException(status_code=503, detail="서비스가 초기화되지 않았습니다.") 

160 success = await ctx.initialize_services(is_paper_trading=req.is_paper) 

161 # 환경 전환 시 상태 캐시 무효화 

162 _status_cache = None 

163 _status_cache_ts = 0.0 

164 if not success: 

165 raise HTTPException(status_code=500, detail="환경 전환 실패 (토큰 발급 오류)") 

166 ctx.start_background_tasks() 

167 return {"success": True, "env_type": ctx.get_env_type()}