Coverage for services / stock_sync_service.py: 96%

1# utils/stock_info_updater.py 

2 

3import pandas as pd 

4import json 

5import os 

6import sqlite3 

7from pykrx import stock 

8from datetime import datetime 

9import FinanceDataReader as fdr 

10 

11ROOT_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) 

12DATA_DIR = os.path.join(ROOT_DIR, "data") 

13DB_FILE_PATH = os.path.join(DATA_DIR, "stock_code_list.db") 

14METADATA_PATH = os.path.join(DATA_DIR, "metadata.json") 

15 

16# 하위 호환용 (테스트 등에서 참조) 

17CSV_FILE_PATH = os.path.join(DATA_DIR, "stock_code_list.csv") 

18 

19TABLE_NAME = "stocks" 

20 

21 

22def _save_metadata(): 

23 metadata = { 

24 "last_updated": datetime.today().strftime("%Y-%m-%d") 

25 } 

26 with open(METADATA_PATH, "w", encoding="utf-8-sig") as f: 

27 json.dump(metadata, f) 

28 

29 

30def _load_metadata(): 

31 if not os.path.exists(METADATA_PATH): 

32 return None 

33 with open(METADATA_PATH, "r", encoding="utf-8-sig") as f: 

34 return json.load(f) 

35 

36 

37def _needs_update(max_age_days=7): 

38 metadata = _load_metadata() 

39 if not metadata: 

40 return True 

41 last_updated = datetime.strptime(metadata["last_updated"], "%Y-%m-%d") 

42 return (datetime.today() - last_updated).days > max_age_days 

43 

44 

45def save_stock_code_list(force_update=False): 

46 """ 

47 종목 코드 리스트 저장 (SQLite + 메타데이터). 

48 force_update=True일 경우 날짜와 무관하게 업데이트. 

49 """ 

50 save_stock_code_list_fdr(force_update=force_update) 

51 

52 

53def save_stock_code_list_fdr(force_update=False): 

54 if not force_update and not _needs_update(): 

55 print("✅ 최근 7일 이내에 이미 업데이트됨. 업데이트 생략.") 

56 return 

57 

58 try: 

59 print("🔄 FinanceDataReader를 통해 KRX 종목 목록을 다운로드합니다...") 

60 # 전체 종목 리스트 가져오기 

61 df_all = fdr.StockListing('KRX') 

62 

63 # 1. 필요한 컬럼만 추출 및 이름 변경 (pykrx 형식에 맞춤) 

64 # FinanceDataReader: Code -> 종목코드, Name -> 종목명 

65 # MarketId를 통해 시장구분 생성 

66 df_all = df_all[['Code', 'Name', 'MarketId']].copy() 

67 

68 # MarketId를 KOSPI/KOSDAQ으로 변환 (KONEX 포함 여부는 선택) 

69 market_map = { 

70 'STK': 'KOSPI', 

71 'KSQ': 'KOSDAQ', 

72 'KNX': 'KONEX' 

73 } 

74 df_all['시장구분'] = df_all['MarketId'].map(market_map) 

75 

76 # 컬럼명 최종 변경 

77 df = df_all.rename(columns={'Code': '종목코드', 'Name': '종목명'}) 

78 

79 # 필요한 시장만 필터링 (KONEX를 제외하려면 아래 주석 해제) 

80 df = df[df['시장구분'].isin(['KOSPI', 'KOSDAQ'])] 

81 

82 # --- 이후 DB 저장 로직은 기존과 동일 --- 

83 os.makedirs(DATA_DIR, exist_ok=True) 

84 df[['종목코드', '종목명', '시장구분']].to_csv(CSV_FILE_PATH, index=False, encoding="utf-8-sig") 

85 

86 conn = sqlite3.connect(DB_FILE_PATH) 

87 try: 

88 df[['종목코드', '종목명', '시장구분']].to_sql(TABLE_NAME, conn, if_exists="replace", index=False) 

89 conn.execute(f"CREATE INDEX IF NOT EXISTS idx_code ON {TABLE_NAME}(종목코드)") 

90 conn.execute(f"CREATE INDEX IF NOT EXISTS idx_name ON {TABLE_NAME}(종목명)") 

91 conn.commit() 

92 finally: 

93 conn.close() 

94 

95 _save_metadata() 

96 print(f"🟢 {len(df)}개 종목 저장 완료 (FDR 사용): {DB_FILE_PATH}") 

97 

98 except Exception as e: 

99 print(f"❌ 데이터 업데이트 실패: {e}") 

100 raise 

101 

102def load_stock_code_list(): 

103 conn = sqlite3.connect(DB_FILE_PATH) 

104 try: 

105 return pd.read_sql(f"SELECT * FROM {TABLE_NAME}", conn, dtype={"종목코드": str}) 

106 finally: 

107 conn.close()