diff --git a/.claude/settings.local.json b/.claude/settings.local.json index 29bb0e0..b0abcb0 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -37,7 +37,9 @@ "Bash(mkdir:*)", "Write(*)", "Edit(*)", - "MultiEdit(*)" + "MultiEdit(*)", + "Bash(docker cp backend/app/api/endpoints/portfolio.py autostock-backend:/app/app/api/endpoints/portfolio.py)", + "Bash(make batch-test)" ] }, "enableAllProjectMcpServers": true, diff --git a/CLAUDE.md b/CLAUDE.md index 820066a..690f8ed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -93,13 +93,13 @@ - **횡보장 (SIDEWAYS)**: ADX < 20 OR 기울기 ≈ 0. **(이 국면에서는 모든 진입 신호를 무시합니다)** ### 6.2 진입 조건 -- **상승장**: RSI(14) < 25, 가격이 볼린저 밴드(BB) 하단 부근 (-3% ~ +1.5%), 거래량 > 20일 평균 거래량(20MA). +- **상승장**: RSI(14) < 40, 가격이 볼린저 밴드(BB) 하단 부근 (-3% ~ +1.5%), 거래량 > 20일 평균 거래량(20MA). - **하락장**: RSI(14) < 30, 가격 < 볼린저 밴드 하단, 거래량 > 1.5배 * 20MA, 반전 캔들(Reversal candle) 필수. - **선제 진입 (Pre-emptive)**: 일봉 RSI < 40 AND 60분봉 RSI < 30 AND 5일 이동평균선(MA5) 상향 돌파. ### 6.3 청산 조건 -- **부분 매도**: RSI > 50 OR 가격 > 볼린저 밴드 중앙선 OR 수익 > 2*ATR. -- **전량 매도**: RSI > 65 OR 가격 > 볼린저 밴드 상단 (99%) OR 트레일링 스탑 도달. +- **부분 매도**: RSI >= 50 OR 가격 > 볼린저 밴드 중앙선 OR 수익 > 2*ATR. +- **전량 매도**: RSI >= 65 OR 가격 > 볼린저 밴드 상단 (99%) OR 트레일링 스탑 도달. --- diff --git a/backend/app/api/endpoints/portfolio.py b/backend/app/api/endpoints/portfolio.py index ec0b433..073611a 100644 --- a/backend/app/api/endpoints/portfolio.py +++ b/backend/app/api/endpoints/portfolio.py @@ -410,7 +410,19 @@ async def sync_portfolio(db: AsyncSession = Depends(get_db), force: bool = True) # 최신 상태가 'SELL'인 종목 식별 ghost_candidates = {code for code, status in stock_status_map.items() if status == "SELL"} - + + # 부분 매도 보정: KIS 잔고에서 여전히 수량이 있는 종목은 ghost에서 제외 + if ghost_candidates and "output1" in balance_data: + still_holding = { + item.get("pdno") + for item in balance_data["output1"] + if int(item.get("hldg_qty", 0)) > 0 + } + partial_sell_excluded = ghost_candidates & still_holding + if partial_sell_excluded: + logger.info(f"↩️ 부분매도 종목은 ghost 제외 (잔고 있음): {partial_sell_excluded}") + ghost_candidates -= still_holding + if ghost_candidates: logger.info(f"🚫 Ghost Stock Candidates (Last action today is SELL): {ghost_candidates}") diff --git a/backend/app/core/config.py b/backend/app/core/config.py index fb70872..0c49047 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -61,6 +61,10 @@ class Settings(BaseSettings): KIS_ACCOUNT_NO: str = "" KIS_BASE_URL: str = "" + # KRX OpenAPI (data-dbg.krx.co.kr) + KRX_API_KEY: str = "" + KRX_API_URL: str = "https://openapi.krx.co.kr" # 포털 URL (참고용) + # WebSocket 접속키 (수동 지정 시 우선, 비어 있으면 자동 발급) KIS_WS_APPROVAL_KEY: str = "" @@ -193,6 +197,10 @@ class Settings(BaseSettings): BEAR_MARKET_MAX_TRADES: int = 3 BEAR_MARKET_POSITION_RATIO: float = 0.5 + # 비상 정지 (전쟁/사이드카 등 급락 이벤트 대응) + TRADING_HALTED: bool = False # .env에서 TRADING_HALTED=true 로 즉시 전체 매수 중단 + MARKET_CRASH_HALT_PCT: float = -3.0 # 지수 당일 -3% 이상 급락 시 자동 매수 차단 + # 데이트레이딩 설정 (CLAUDE.md 섹션 8.2) DAY_TRADE_ENABLED: bool = True # 데이트레이딩 활성화 DAY_TRADE_MAX_POSITIONS: int = 3 # 동시 최대 종목 수 diff --git a/backend/app/services/korea_investment_api.py b/backend/app/services/korea_investment_api.py index 7a93f39..b6aea07 100644 --- a/backend/app/services/korea_investment_api.py +++ b/backend/app/services/korea_investment_api.py @@ -511,6 +511,76 @@ class KoreaInvestmentAPI: logger.error(f"Error fetching paginated chart for {stock_code}: {e}") raise + async def get_daily_ohlcv_range( + self, stock_code: str, start_date: str, end_date: str + ) -> list: + """날짜 범위 일봉 OHLCV 전체 조회 (FHKST03010100) + + KIS API는 1회 최대 100행 반환 → 날짜를 소급하며 start_date까지 전부 수집. + + Args: + stock_code: 종목코드 (예: "005930") + start_date: 조회 시작일 YYYYMMDD + end_date: 조회 종료일 YYYYMMDD + + Returns: + [{"stck_bsop_date": "20240101", "stck_oprc": ..., "stck_hgpr": ..., + "stck_lwpr": ..., "stck_clpr": ..., "acml_vol": ..., "prdy_ctrt": ...}, ...] + 최신 → 오래된 순 정렬 (KIS API 기본값) + """ + all_data: list = [] + current_end = end_date + + while True: + params = { + "FID_COND_MRKT_DIV_CODE": "J", + "FID_INPUT_ISCD": stock_code, + "FID_INPUT_DATE_1": start_date, + "FID_INPUT_DATE_2": current_end, + "FID_PERIOD_DIV_CODE": "D", + "FID_ORG_ADJ_PRC": "0", # 수정주가 + } + try: + result, _ = await self._make_request( + "GET", + "/uapi/domestic-stock/v1/quotations/inquire-daily-itemchartprice", + "FHKST03010100", + params=params, + ) + except Exception as exc: + logger.error( + "get_daily_ohlcv_range %s %s~%s 실패: %s", + stock_code, start_date, current_end, exc, + ) + break + + if not result or result.get("rt_cd") != "0": + logger.warning( + "get_daily_ohlcv_range %s: %s", + stock_code, result.get("msg1") if result else "no response", + ) + break + + output = result.get("output2") or [] + if not output: + break + + all_data.extend(output) + + # 가장 오래된 행의 날짜를 확인하고 그보다 하루 전으로 소급 + oldest_date = output[-1].get("stck_bsop_date", "") + if not oldest_date or oldest_date <= start_date: + break + + oldest_dt = datetime.strptime(oldest_date, "%Y%m%d") + current_end = (oldest_dt - timedelta(days=1)).strftime("%Y%m%d") + if current_end < start_date: + break + + await asyncio.sleep(self.min_request_interval) + + return all_data + async def get_overseas_price(self, symbol: str, exchange_code: str = "NASD") -> Dict[str, Any]: """해외주식 현재가 조회 (NASD, NYSE, AMEX)""" try: @@ -975,7 +1045,7 @@ class KoreaInvestmentAPI: "FID_BLNG_CLS_CODE": "0", "FID_TRGT_CLS_CODE": "111111111", "FID_TRGT_EXLS_CLS_CODE": "000000", - "FID_INPUT_PRICE_1": "", + "FID_INPUT_PRICE_1": "5000", # 5,000원 이상 종목만 조회 "FID_INPUT_PRICE_2": "", "FID_VOL_CNT": "", "FID_INPUT_DATE_1": "" @@ -1032,3 +1102,10 @@ class KoreaInvestmentAPI: # 싱글톤 인스턴스 korea_investment_api = KoreaInvestmentAPI() + +# 실계좌 시세조회 전용 싱글톤 (KIS_REAL_APP_KEY 설정 시 초기화) +# 모의투자 모드에서도 시세/차트/종목 조회는 실계좌 API를 사용한다. +try: + real_market_api = KoreaInvestmentAPI.for_real_market_data() +except Exception: + real_market_api = None diff --git a/backend/app/services/market_analyzer.py b/backend/app/services/market_analyzer.py index f107727..ef8532f 100644 --- a/backend/app/services/market_analyzer.py +++ b/backend/app/services/market_analyzer.py @@ -8,7 +8,7 @@ from pydantic import BaseModel from app.core.config import settings from app.core.market_schedule import is_market_open -from app.services.korea_investment_api import korea_investment_api +from app.services.korea_investment_api import korea_investment_api, real_market_api as _real_market_api from app.services.mean_reversion_strategy import ( MarketRegime, detect_market_regime, calculate_atr, calculate_adx ) @@ -77,7 +77,8 @@ async def check_market_regime() -> dict: if _market_regime_cache["result"] is not None and (now_ts - _market_regime_cache["ts"]) < _MARKET_REGIME_CACHE_TTL: return _market_regime_cache["result"] - kospi_chart = await korea_investment_api.get_daily_chart("069500", "D", limit=240) + _api = _real_market_api or korea_investment_api + kospi_chart = await _api.get_daily_chart("069500", "D", limit=240) if "output2" not in kospi_chart or not kospi_chart["output2"]: return {"regime": "neutral", "message": "시장 지수 확인 불가", "position_multiplier": 0.5} @@ -88,6 +89,22 @@ async def check_market_regime() -> dict: if len(closes) < 60: return {"regime": "neutral", "message": "데이터 부족", "position_multiplier": 0.5} + def _cache_and_return(result: dict) -> dict: + _market_regime_cache["result"] = result + _market_regime_cache["ts"] = time.monotonic() + return result + + # ── 수동 비상 정지 ──────────────────────────────────────────── + if settings.TRADING_HALTED: + return _cache_and_return({ + "regime": "halted", + "regime_200d": "bear", + "message": "⛔ TRADING_HALTED=true — 수동 비상 정지 중 (매수 전면 차단)", + "position_multiplier": 0.0, + "min_score": 9999, + "max_trades": 0, + }) + current_price = closes[0] ma20 = sum(closes[:20]) / 20 ma60 = sum(closes[:60]) / 60 @@ -100,6 +117,23 @@ async def check_market_regime() -> dict: is_bull_200 = current_price > ma20 ma200 = ma60 + # ── 당일 급락 감지 (사이드카/전쟁 등) ─────────────────────── + # closes[0]이 오늘 데이터일 때만 적용 (일봉 미확정 시 전일 종가가 closes[0]이 되어 오탐 발생) + latest_date = chart_data[0].get("stck_bsop_date", "") if chart_data else "" + today_str = _get_kst_now().strftime("%Y%m%d") + if latest_date == today_str and len(closes) >= 2 and closes[1] > 0: + today_change_pct = (closes[0] - closes[1]) / closes[1] * 100 + if today_change_pct <= settings.MARKET_CRASH_HALT_PCT: + return _cache_and_return({ + "regime": "crash", + "regime_200d": "bear", + "message": f"🚨 당일 급락 감지 ({today_change_pct:.1f}% ≤ {settings.MARKET_CRASH_HALT_PCT}%) — 매수 전면 차단", + "position_multiplier": 0.0, + "min_score": 9999, + "max_trades": 0, + "today_change_pct": round(today_change_pct, 2), + }) + consecutive_down = 0 for i in range(len(closes) - 1): if closes[i] < closes[i + 1]: @@ -109,11 +143,6 @@ async def check_market_regime() -> dict: drop_from_ma20 = ((current_price - ma20) / ma20) * 100 - def _cache_and_return(result: dict) -> dict: - _market_regime_cache["result"] = result - _market_regime_cache["ts"] = time.monotonic() - return result - if current_price < ma60 or consecutive_down >= 5: return _cache_and_return({ "regime": "strong_bearish", @@ -153,274 +182,318 @@ async def check_market_regime() -> dict: return {"regime": "neutral", "message": "시장 지수 확인 실패", "position_multiplier": 0.5} -async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: +def _analyze_chart_items( + chart_data: List[dict], + stock_code: str, + stock_name: str, + criteria: "RecommendationCriteria", +) -> Optional[StockRecommendation]: + """KIS 형식 chart_data (oldest first)로 기술적 분석 수행 — API 호출 없음. + + analyze_stock / analyze_stock_for_batch_db 양쪽에서 공유하는 순수 분석 로직. + """ + latest = chart_data[-1] if chart_data else None + if not latest: + return None + + if not stock_name: + stock_name = latest.get("hts_kor_isnm", stock_code) + + current_price = float(latest.get("stck_clpr", 0)) + price_date = latest.get("stck_bsop_date", "") + volume = int(latest.get("acml_vol", 0) or 0) + if current_price == 0: + logger.debug("필터[가격0] %s", stock_code) + return None + + # 가격 필터: 5,000원 미만 종목 제외 (CLAUDE.md 섹션 5.4) + if current_price < settings.MIN_PRICE_KRW: + logger.info("필터[저가] %s: %.0f원 < %d원", stock_code, current_price, settings.MIN_PRICE_KRW) + return None + + # 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족) + if volume < 1_000_000: + logger.info("필터[거래량] %s: %d주 < 100만주", stock_code, volume) + return None + + if len(chart_data) >= 2: + prev_price = float(chart_data[-2].get("stck_clpr", current_price)) + change_rate = ((current_price - prev_price) / prev_price) * 100 if prev_price > 0 else 0 + else: + change_rate = 0 + + closes = [float(item.get("stck_clpr", 0)) for item in chart_data] + volumes = [int(item.get("acml_vol", 0)) for item in chart_data] + + ma5 = sum(closes[-5:]) / 5 if len(closes) >= 5 else None + ma20 = sum(closes[-20:]) / 20 if len(closes) >= 20 else None + ma60 = sum(closes[-60:]) / 60 if len(closes) >= 60 else None + + avg_vol_20 = sum(volumes[-20:]) / 20 if len(volumes) >= 20 else 0 + volume_ratio = (volume / avg_vol_20) if avg_vol_20 > 0 else 0 + + if not criteria.for_batch and avg_vol_20 < 50000 and current_price < 50000: + return None + + disparity = (current_price / ma20) * 100 if ma20 else 100 + + trend_ok = None + if TREND_FILTER_ENABLED: + trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG) + + rsi = None + if len(closes) >= 15: + prices_for_rsi = closes[-15:] + gains = [] + losses = [] + for i in range(1, len(prices_for_rsi)): + diff = prices_for_rsi[i] - prices_for_rsi[i-1] + gains.append(max(diff, 0)) + losses.append(max(-diff, 0)) + avg_gain = sum(gains[-14:]) / 14 + avg_loss = sum(losses[-14:]) / 14 + if avg_loss == 0: + rsi = 100.0 + else: + rs = avg_gain / avg_loss + rsi = 100 - (100 / (1 + rs)) + + stochastic_k = None + stochastic_d = None + if len(chart_data) >= 14: + recent = chart_data[-14:] + highs = [float(item.get("stck_hgpr", 0)) for item in recent] + lows = [float(item.get("stck_lwpr", 0)) for item in recent] + closes_for_stoch = [float(item.get("stck_clpr", 0)) for item in recent] + highest_high = max(highs) + lowest_low = min(lows) + if highest_high != lowest_low: + stochastic_k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100 + stochastic_d_vals = [] + for j in range(len(recent)-3, len(recent)): + slice_high = max([float(item.get("stck_hgpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) + slice_low = min([float(item.get("stck_lwpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) + if slice_high != slice_low: + stochastic_d_vals.append(((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100) + if stochastic_d_vals: + stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals) + + bollinger_upper = bollinger_lower = bollinger_middle = None + if len(closes) >= 20: + window = closes[-20:] + middle = sum(window) / len(window) + variance = sum((p - middle) ** 2 for p in window) / len(window) + stddev = variance ** 0.5 + bollinger_middle = middle + bollinger_upper = middle + 2 * stddev + bollinger_lower = middle - 2 * stddev + + signals = [] + score = 0 + + if disparity > criteria.max_disparity: + score -= 5 + signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지") + + if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED: + overheat = analyze_overheat(closes, ma20) + if overheat["blocked"]: + score -= 10 + signals.append(overheat["reason"]) + + if criteria.use_volatility_breakout: + vb_result = analyze_volatility_breakout(chart_data) + if vb_result and vb_result["signal"] == "buy": + signals.append(vb_result["description"]) + score += 4 + + custom_result = analyze_custom_signals(chart_data) + if custom_result: + if custom_result["signal"] == "buy": + signals.append(custom_result["description"]) + score += 3 + elif custom_result["signal"] == "sell": + signals.append(custom_result["description"]) + score -= 2 + elif custom_result["signal"] == "exit": + signals.append(custom_result["description"]) + score -= 5 + elif custom_result["signal"] == "take_profit": + signals.append(custom_result["description"]) + score += 2 + + if criteria.use_macd: + result = analyze_macd(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 2 + + if criteria.use_rsi: + result = analyze_rsi(chart_data) + if result["signal"] == "buy": + if volume_ratio >= 1.2: + signals.append(result["description"] + " + 거래량증가") + score += 3 + else: + signals.append(result["description"] + " (거래량부족)") + score += 1 + + if criteria.use_bollinger: + result = analyze_bollinger(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 1 + + if criteria.use_moving_average: + result = analyze_moving_average(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 2 + + if criteria.use_stochastic: + result = analyze_stochastic(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 1 + + if criteria.use_adx: + result = analyze_adx(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 1 + + if criteria.use_obv: + result = analyze_obv(chart_data) + if result["signal"] == "buy": + signals.append(result["description"]) + score += 1 + + if volume_ratio >= 1.5 and change_rate > 0: + signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)") + score += 3 + elif volume_ratio < 0.8: + score -= 1 + + if TREND_FILTER_ENABLED and trend_ok is not None: + if trend_ok: + signals.append(f"추세상승(>{TREND_MA_SHORT}MA & {TREND_MA_SHORT}MA>{TREND_MA_LONG}MA, tol={int(TREND_TOLERANCE_PCT*10000)/100:.2f}%)") + score += 1 + else: + signals.append(f"추세약함({TREND_MA_SHORT}MA≤{TREND_MA_LONG}MA 또는 종가≤{TREND_MA_SHORT}MA)") + + if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio: + return None + + ml_score_val = None + ml_confidence_val = None + ml_signal_val = None + + if settings.ML_ENSEMBLE_ENABLED: + try: + from app.services.ml_ensemble import get_ensemble + ensemble = get_ensemble() + if ensemble.is_trained: + ml_result = ensemble.predict(chart_data) + if "error" not in ml_result: + ml_score_val = ml_result["ensemble_score"] + ml_confidence_val = ml_result["confidence"] + ml_signal_val = ml_result["signal"] + + rule_score_normalized = (score / 15.0) * 100.0 + combined_score = int( + rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT + + ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT + ) + score = max(0, int(combined_score / 100.0 * 15.0)) + + signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") + except Exception: + pass + + atr_14_val = None + adx_14_val = None try: - chart_response = await korea_investment_api.get_daily_chart(stock_code, "D") + if len(chart_data) >= 15: + _h = [float(d.get("stck_hgpr", 0)) for d in chart_data] + _l = [float(d.get("stck_lwpr", 0)) for d in chart_data] + _c = [float(d.get("stck_clpr", 0)) for d in chart_data] + atr_14_val = calculate_atr(_h, _l, _c) + if len(chart_data) >= 29: + adx_14_val = calculate_adx(_h, _l, _c) + except Exception as _e: + logger.debug(f"ATR/ADX 계산 실패 {stock_code}: {_e}") + + if criteria.for_batch or len(signals) > 0 or score > 0: + final_score = max(1, score) if len(signals) > 0 else score + return StockRecommendation( + stock_code=stock_code, + stock_name=stock_name, + current_price=current_price, + change_rate=change_rate, + signals=signals if signals else ["조건 분석 중"], + score=final_score, + price_date=price_date, + volume=volume, + rsi=rsi, + ma5=ma5, + ma20=ma20, + ma60=ma60, + stochastic_k=stochastic_k, + stochastic_d=stochastic_d, + bollinger_upper=bollinger_upper, + bollinger_lower=bollinger_lower, + bollinger_middle=bollinger_middle, + ml_score=ml_score_val, + ml_confidence=ml_confidence_val, + ml_signal=ml_signal_val, + atr_14=atr_14_val, + adx_14=adx_14_val, + ) + + return None + + +async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: + """KIS API로 차트를 조회한 뒤 _analyze_chart_items에 위임.""" + try: + _api = _real_market_api or korea_investment_api + chart_response = await _api.get_daily_chart(stock_code, "D") if "output2" not in chart_response or not chart_response["output2"]: return None - chart_data = list(reversed(chart_response["output2"])) - latest = chart_data[-1] if chart_data else None - if not latest: - return None - - if not stock_name: - stock_name = latest.get("hts_kor_isnm", stock_code) - - current_price = float(latest.get("stck_clpr", 0)) - price_date = latest.get("stck_bsop_date", "") - volume = int(latest.get("acml_vol", 0) or 0) - if current_price == 0: - return None - - # 가격 필터: 5,000원 미만 종목 제외 (CLAUDE.md 섹션 5.4) - if current_price < settings.MIN_PRICE_KRW: - return None - - # 전일 거래량 필터: 100만주 미만 종목 제외 (유동성 부족) - if volume < 1_000_000: - return None - - if len(chart_data) >= 2: - prev_price = float(chart_data[-2].get("stck_clpr", current_price)) - change_rate = ((current_price - prev_price) / prev_price) * 100 if prev_price > 0 else 0 - else: - change_rate = 0 - - closes = [float(item.get("stck_clpr", 0)) for item in chart_data] - volumes = [int(item.get("acml_vol", 0)) for item in chart_data] - - ma5 = sum(closes[-5:]) / 5 if len(closes) >= 5 else None - ma20 = sum(closes[-20:]) / 20 if len(closes) >= 20 else None - ma60 = sum(closes[-60:]) / 60 if len(closes) >= 60 else None - - avg_vol_20 = sum(volumes[-20:]) / 20 if len(volumes) >= 20 else 0 - volume_ratio = (volume / avg_vol_20) if avg_vol_20 > 0 else 0 - - if not criteria.for_batch and avg_vol_20 < 50000 and current_price < 50000: - return None - - disparity = (current_price / ma20) * 100 if ma20 else 100 - - trend_ok = None - if TREND_FILTER_ENABLED: - trend_ok = _passes_uptrend(closes, TREND_MA_SHORT, TREND_MA_LONG) - - rsi = None - if len(closes) >= 15: - prices_for_rsi = closes[-15:] - gains = [] - losses = [] - for i in range(1, len(prices_for_rsi)): - diff = prices_for_rsi[i] - prices_for_rsi[i-1] - gains.append(max(diff, 0)) - losses.append(max(-diff, 0)) - avg_gain = sum(gains[-14:]) / 14 - avg_loss = sum(losses[-14:]) / 14 - if avg_loss == 0: - rsi = 100.0 - else: - rs = avg_gain / avg_loss - rsi = 100 - (100 / (1 + rs)) - - stochastic_k = None - stochastic_d = None - if len(chart_data) >= 14: - recent = chart_data[-14:] - highs = [float(item.get("stck_hgpr", 0)) for item in recent] - lows = [float(item.get("stck_lwpr", 0)) for item in recent] - closes_for_stoch = [float(item.get("stck_clpr", 0)) for item in recent] - highest_high = max(highs) - lowest_low = min(lows) - if highest_high != lowest_low: - stochastic_k = ((closes_for_stoch[-1] - lowest_low) / (highest_high - lowest_low)) * 100 - stochastic_d_vals = [] - for j in range(len(recent)-3, len(recent)): - slice_high = max([float(item.get("stck_hgpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) - slice_low = min([float(item.get("stck_lwpr", 0)) for item in recent[j-3 if j-3>=0 else 0:j+1]]) - if slice_high != slice_low: - stochastic_d_vals.append(((float(recent[j].get("stck_clpr", 0)) - slice_low) / (slice_high - slice_low)) * 100) - if stochastic_d_vals: - stochastic_d = sum(stochastic_d_vals) / len(stochastic_d_vals) - - bollinger_upper = bollinger_lower = bollinger_middle = None - if len(closes) >= 20: - window = closes[-20:] - middle = sum(window) / len(window) - variance = sum((p - middle) ** 2 for p in window) / len(window) - stddev = variance ** 0.5 - bollinger_middle = middle - bollinger_upper = middle + 2 * stddev - bollinger_lower = middle - 2 * stddev - - signals = [] - score = 0 - - if disparity > criteria.max_disparity: - score -= 5 - signals.append(f"이격도 과열({disparity:.1f}% > {criteria.max_disparity}%) - 매수금지") - - if criteria.use_overheat_filter and OVERHEAT_FILTER_ENABLED: - overheat = analyze_overheat(closes, ma20) - if overheat["blocked"]: - score -= 10 - signals.append(overheat["reason"]) - - if criteria.use_volatility_breakout: - vb_result = analyze_volatility_breakout(chart_data) - if vb_result and vb_result["signal"] == "buy": - signals.append(vb_result["description"]) - score += 4 - - custom_result = analyze_custom_signals(chart_data) - if custom_result: - if custom_result["signal"] == "buy": - signals.append(custom_result["description"]) - score += 3 - elif custom_result["signal"] == "sell": - signals.append(custom_result["description"]) - score -= 2 - elif custom_result["signal"] == "exit": - signals.append(custom_result["description"]) - score -= 5 - elif custom_result["signal"] == "take_profit": - signals.append(custom_result["description"]) - score += 2 - - if criteria.use_macd: - result = analyze_macd(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 2 - - if criteria.use_rsi: - result = analyze_rsi(chart_data) - if result["signal"] == "buy": - if volume_ratio >= 1.2: - signals.append(result["description"] + " + 거래량증가") - score += 3 - else: - signals.append(result["description"] + " (거래량부족)") - score += 1 - - if criteria.use_bollinger: - result = analyze_bollinger(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 1 - - if criteria.use_moving_average: - result = analyze_moving_average(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 2 - - if criteria.use_stochastic: - result = analyze_stochastic(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 1 - - if criteria.use_adx: - result = analyze_adx(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 1 - - if criteria.use_obv: - result = analyze_obv(chart_data) - if result["signal"] == "buy": - signals.append(result["description"]) - score += 1 - - if volume_ratio >= 1.5 and change_rate > 0: - signals.append(f"거래량 실린 상승 ({volume_ratio:.1f}배)") - score += 3 - elif volume_ratio < 0.8: - score -= 1 - - if TREND_FILTER_ENABLED and trend_ok is not None: - if trend_ok: - signals.append(f"추세상승(>{TREND_MA_SHORT}MA & {TREND_MA_SHORT}MA>{TREND_MA_LONG}MA, tol={int(TREND_TOLERANCE_PCT*10000)/100:.2f}%)") - score += 1 - else: - signals.append(f"추세약함({TREND_MA_SHORT}MA≤{TREND_MA_LONG}MA 또는 종가≤{TREND_MA_SHORT}MA)") - - if criteria.min_volume_ratio > 0 and volume_ratio < criteria.min_volume_ratio: - return None - - ml_score_val = None - ml_confidence_val = None - ml_signal_val = None - - if settings.ML_ENSEMBLE_ENABLED: - try: - from app.services.ml_ensemble import get_ensemble - ensemble = get_ensemble() - if ensemble.is_trained: - ml_result = ensemble.predict(chart_data) - if "error" not in ml_result: - ml_score_val = ml_result["ensemble_score"] - ml_confidence_val = ml_result["confidence"] - ml_signal_val = ml_result["signal"] - - rule_score_normalized = (score / 15.0) * 100.0 - combined_score = int( - rule_score_normalized * settings.ML_ENSEMBLE_RULE_WEIGHT + - ml_score_val * settings.ML_ENSEMBLE_ML_WEIGHT - ) - score = max(0, int(combined_score / 100.0 * 15.0)) - - signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") - except Exception: - pass - - atr_14_val = None - adx_14_val = None - try: - if len(chart_data) >= 15: - _h = [float(d.get("stck_hgpr", 0)) for d in chart_data] - _l = [float(d.get("stck_lwpr", 0)) for d in chart_data] - _c = [float(d.get("stck_clpr", 0)) for d in chart_data] - atr_14_val = calculate_atr(_h, _l, _c) - if len(chart_data) >= 29: - adx_14_val = calculate_adx(_h, _l, _c) - except Exception as _e: - logger.debug(f"ATR/ADX 계산 실패 {stock_code}: {_e}") - - if criteria.for_batch or len(signals) > 0 or score > 0: - final_score = max(1, score) if len(signals) > 0 else score - - return StockRecommendation( - stock_code=stock_code, - stock_name=stock_name, - current_price=current_price, - change_rate=change_rate, - signals=signals if signals else ["조건 분석 중"], - score=final_score, - price_date=price_date, - volume=volume, - rsi=rsi, - ma5=ma5, - ma20=ma20, - ma60=ma60, - stochastic_k=stochastic_k, - stochastic_d=stochastic_d, - bollinger_upper=bollinger_upper, - bollinger_lower=bollinger_lower, - bollinger_middle=bollinger_middle, - ml_score=ml_score_val, - ml_confidence=ml_confidence_val, - ml_signal=ml_signal_val, - atr_14=atr_14_val, - adx_14=adx_14_val, - ) - - return None + return _analyze_chart_items(chart_data, stock_code, stock_name, criteria) except Exception as e: logger.error(f"Error analyzing {stock_code}: {str(e)}") return None + +async def analyze_stock_for_batch_db( + chart_data_newest_first: List[dict], + stock_code: str, + stock_name: str, +) -> Optional[StockRecommendation]: + """DB에서 가져온 chart_data (newest first)로 배치 분석 — KIS API 호출 없음.""" + criteria = RecommendationCriteria( + use_macd=True, + use_rsi=True, + use_bollinger=True, + use_moving_average=True, + use_stochastic=True, + use_adx=True, + use_obv=True, + use_volatility_breakout=True, + min_volume_ratio=0, + max_disparity=200.0, + for_batch=True, + ) + try: + chart_data = list(reversed(chart_data_newest_first)) # oldest first + return _analyze_chart_items(chart_data, stock_code, stock_name, criteria) + except Exception as e: + logger.error(f"Error analyzing {stock_code}: {str(e)}") + return None + + async def analyze_stock_for_batch(stock_code: str, stock_name: str) -> Optional[StockRecommendation]: criteria = RecommendationCriteria( use_macd=True, @@ -453,27 +526,30 @@ async def analyze_stock_with_indicators(stock_code: str, stock_name: str) -> Opt return await analyze_stock(stock_code, stock_name, criteria) async def get_all_tradable_stocks() -> List[tuple]: + SKIP_KEYWORDS = [ + "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", + "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE", + "선물", "스팩", "리츠", "SPAC", + ] try: + _list_api = _real_market_api or korea_investment_api await asyncio.sleep(3.0) - kospi_result = await korea_investment_api.get_stock_list("J") + kospi_result = await _list_api.get_stock_list("J") await asyncio.sleep(3.0) - kosdaq_result = await korea_investment_api.get_stock_list("Q") - + kosdaq_result = await _list_api.get_stock_list("Q") + stocks = [] - if "output" in kospi_result: - for item in kospi_result["output"]: - stock_code = item.get("mksc_shrn_iscd", "") - stock_name = item.get("hts_kor_isnm", "") - if stock_code and stock_name: + for result in (kospi_result, kosdaq_result): + if "output" in result: + for item in result["output"]: + stock_code = item.get("mksc_shrn_iscd", "") + stock_name = item.get("hts_kor_isnm", "") + if not stock_code or not stock_name: + continue + if any(kw in stock_name for kw in SKIP_KEYWORDS): + continue stocks.append((stock_code, stock_name)) - - if "output" in kosdaq_result: - for item in kosdaq_result["output"]: - stock_code = item.get("mksc_shrn_iscd", "") - stock_name = item.get("hts_kor_isnm", "") - if stock_code and stock_name: - stocks.append((stock_code, stock_name)) - + return stocks except Exception as e: logger.error(f"Error fetching stock list from KIS API: {e}", exc_info=True) diff --git a/backend/batch_daily_analysis.py b/backend/batch_daily_analysis.py index 6c343c8..efafce4 100644 --- a/backend/batch_daily_analysis.py +++ b/backend/batch_daily_analysis.py @@ -31,9 +31,11 @@ sys.path.insert(0, str(Path(__file__).parent)) from app.core.config import settings from app.core.database import AsyncSessionLocal # 🔧 공통 DB 모듈 사용 -from app.services.korea_investment_api import KoreaInvestmentAPI, korea_investment_api +import aiohttp +from sqlalchemy import select as sa_select +from app.models.models import StockDailyPrice from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_filter, run_db_scanner # 🔧 단일화 -from app.api.endpoints.recommendations import analyze_stock_for_batch, StockRecommendation +from app.services.market_analyzer import analyze_stock_for_batch_db, StockRecommendation import logging # 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출 @@ -55,33 +57,57 @@ MIN_STOCK_COUNT = 50 # 1. 시장 레짐 판별 (KOSPI 기준) # ------------------------------------------------------------------- +async def _fetch_db_chart(stock_code: str, limit: int = 240) -> list: + """stock_daily_prices DB에서 KIS 형식 차트 데이터 조회 (newest first). + + KIS API 호출 없이 collect_krx_data.py가 사전 수집한 데이터를 사용. + """ + async with AsyncSessionLocal() as db: + result = await db.execute( + sa_select(StockDailyPrice) + .where(StockDailyPrice.stock_code == stock_code) + .order_by(StockDailyPrice.trade_date.desc()) + .limit(limit) + ) + rows = result.scalars().all() + return [ + { + "stck_bsop_date": row.trade_date.strftime("%Y%m%d"), + "stck_clpr": str(row.close_price or 0), + "stck_oprc": str(row.open_price or 0), + "stck_hgpr": str(row.high_price or 0), + "stck_lwpr": str(row.low_price or 0), + "acml_vol": str(row.volume or 0), + "hts_kor_isnm": row.stock_name or "", + } + for row in rows + ] + + async def detect_and_log_market_regime() -> str: - """배치 시작 시 KOSPI 200일 데이터로 시장 레짐 판별. + """배치 시작 시 KOSPI200 레짐 판별 — DB에서 KODEX200 데이터 조회. - BULL / BEAR / SIDEWAYS 중 하나를 리턴하며 배치 전체에 로깅된다. + BULL / BEAR 중 하나를 리턴하며 배치 전체에 로깅된다. 실패 시 UNKNOWN을 리턴하여 배치를 차단하지 않는다. - - pykrx 대신 KIS API 사용 — KODEX 200(069500)으로 KOSPI 레짐 판별. - KODEX 200은 KOSPI 200을 추종하므로 지수 대용으로 적합하다. """ try: from app.services.mean_reversion_strategy import detect_market_regime - # KODEX 200(069500) 일봉 240개 = 약 1년치 데이터 - data = await korea_investment_api.get_daily_chart("069500", period="D", limit=240) - output = data.get("output2") or [] - - if not isinstance(output, list) or len(output) < 205: - logger.warning("⚠️ KODEX200 데이터 부족(%d행) — 레짐 판별 스킵", len(output) if output else 0) + chart_data = await _fetch_db_chart("069500", limit=240) + if not chart_data: + logger.warning("⚠️ KOSPI200(KODEX200) DB 데이터 없음 — 레짐 판별 스킵") return "UNKNOWN" - # KIS API는 최신순 반환 → 오래된 순(시계열 순)으로 역정렬 - closes = [float(item["stck_clpr"]) for item in reversed(output) if item.get("stck_clpr")] - regime = detect_market_regime(closes) + # DB 데이터는 newest first → oldest first 변환 후 detect_market_regime 호출 + closes = list(reversed([float(item["stck_clpr"]) for item in chart_data])) - regime_emoji = {"상승장": "📈 BULL", "하락장": "📉 BEAR", - "bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper()) - logger.info("🌏 시장 레짐: %s (KODEX200 %d일 기준)", regime_emoji, len(closes)) + if len(closes) < 205: + logger.warning("⚠️ KOSPI200 데이터 부족(%d행) — 레짐 판별 스킵", len(closes)) + return "UNKNOWN" + + regime = detect_market_regime(closes) + regime_emoji = {"bull": "📈 BULL", "bear": "📉 BEAR"}.get(regime.value, "➡️ " + regime.value.upper()) + logger.info("🌏 시장 레짐: %s (KODEX200 %d일 기준, DB)", regime_emoji, len(closes)) return regime.value.upper() except Exception as exc: @@ -93,87 +119,146 @@ async def detect_and_log_market_regime() -> str: # 2. 종목 목록 조회 # ------------------------------------------------------------------- -async def get_all_tradable_stocks(kis_api: KoreaInvestmentAPI, limit: int = 500) -> list[tuple[str, str]]: - """거래량 상위 종목 조회 (KIS API 사용) +async def get_all_tradable_stocks(limit: int = 500) -> list[tuple[str, str]]: + """거래 가능 종목 조회 (네이버 금융 sise 사용) - pykrx 의존성 제거 — KIS 거래량 순위 API(FHPST01710000) 사용. - 시가총액 기준 → 거래량 기준으로 변경 (유동성 높은 종목 우선). + 네이버 금융 전체 시세 페이지에서 KOSPI+KOSDAQ 종목 수집. + (data.krx.co.kr IP 차단 우회 — 네이버 서버는 별도) ETF / 레버리지 / 스팩 등 특수종목 자동 제외. - KOSPI + KOSDAQ 합산 거래량 상위 limit개 반환. Raises: RuntimeError: 조회 결과가 MIN_STOCK_COUNT(50) 미만인 경우 배치 중단 """ - logger.info("Fetching top %d stocks by volume from KIS API...", limit) + from bs4 import BeautifulSoup SKIP_KEYWORDS = [ "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", - "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", + "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "RISE", "선물", "스팩", "리츠", "SPAC", ] + HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; autostock-batch/1.0)"} - # 모의투자 모드에서 실계좌 시세 API를 사용할 수 있으면 준비 - real_market_api: Optional[KoreaInvestmentAPI] = None - if settings.KIS_MODE == "virtual" and settings.KIS_REAL_APP_KEY: - try: - real_market_api = KoreaInvestmentAPI.for_real_market_data() - logger.info("📡 KOSDAQ 조회: 실계좌 시세 API 사용 (모의투자 서버 제한 우회)") - except Exception as exc: - logger.warning("⚠️ 실계좌 시세 API 초기화 실패, 모의투자 API 사용: %s", exc) + logger.info("📡 네이버 금융 종목 조회 (KOSPI+KOSDAQ)") - stocks_with_vol = [] - for market_code, market_name in [("J", "KOSPI"), ("Q", "KOSDAQ")]: - # KOSDAQ은 실계좌 시세 API 우선 사용 (모의투자 서버는 KOSDAQ 미지원) - api_to_use = real_market_api if (market_code == "Q" and real_market_api) else kis_api - try: - result = await api_to_use.get_volume_rank(market_code) - items = result.get("output") or [] + async def _fetch_market_naver( + session: aiohttp.ClientSession, sosok: int, market_name: str + ) -> list[tuple[str, str, str]]: + """sosok=0: KOSPI, sosok=1: KOSDAQ""" + base_url = "https://finance.naver.com/sise/sise_market_sum.naver" + results: list[tuple[str, str, str]] = [] - if not isinstance(items, list): - logger.warning(" %s: output이 리스트 아님 — 건너뜀", market_name) - continue + # 첫 페이지로 총 페이지 수 파악 + async with session.get( + base_url, params={"sosok": sosok, "page": 1}, headers=HEADERS, + timeout=aiohttp.ClientTimeout(total=30) + ) as resp: + if resp.status != 200: + logger.warning(" 네이버 %s HTTP %s", market_name, resp.status) + return [] + html = await resp.text(encoding="euc-kr", errors="replace") - count = 0 - for item in items: - ticker = item.get("mksc_shrn_iscd") or item.get("stck_shrn_iscd") or "" - name = item.get("hts_kor_isnm") or item.get("stck_kor_isnm") or "" - volume = int(item.get("acml_vol") or 0) + soup = BeautifulSoup(html, "html.parser") + # 총 페이지 수 추출 + pager = soup.select("td.pgRR a") + if pager: + import re + m = re.search(r"page=(\d+)", pager[-1].get("href", "")) + total_pages = int(m.group(1)) if m else 1 + else: + total_pages = 1 + total_pages = min(total_pages, 30) # 최대 30페이지 (약 1,500종목) - if not ticker or not name: + # 첫 페이지 파싱 + def _parse_page(s: BeautifulSoup) -> list[tuple[str, str, str]]: + rows = s.select("table.type_2 tr") + items = [] + for row in rows: + a_tag = row.select_one("a.tltle") + if not a_tag: continue - if any(k in name for k in SKIP_KEYWORDS): + name = a_tag.get_text(strip=True) + href = a_tag.get("href", "") + import re as _re + m2 = _re.search(r"code=(\d+)", href) + if not m2: continue + code = m2.group(1) + if any(kw in name for kw in SKIP_KEYWORDS): + continue + items.append((code, name, market_name)) + return items - stocks_with_vol.append((ticker, name, volume, market_name)) - count += 1 + results.extend(_parse_page(soup)) - logger.info(" %s: %d개 종목 조회", market_name, count) + # 나머지 페이지 병렬 수집 + sem = asyncio.Semaphore(5) - except Exception as exc: - logger.error("❌ %s 종목 조회 실패: %s", market_name, exc) + async def _fetch_page(page: int): + async with sem: + try: + async with session.get( + base_url, params={"sosok": sosok, "page": page}, + headers=HEADERS, timeout=aiohttp.ClientTimeout(total=30) + ) as r: + if r.status != 200: + return [] + h = await r.text(encoding="euc-kr", errors="replace") + return _parse_page(BeautifulSoup(h, "html.parser")) + except Exception: + return [] - stocks_with_vol.sort(key=lambda x: x[2], reverse=True) - top_with_mkt = stocks_with_vol[:limit] - top_stocks = [(code, name) for code, name, _, _mkt in top_with_mkt] + tasks = [_fetch_page(p) for p in range(2, total_pages + 1)] + pages_data = await asyncio.gather(*tasks) + for page_items in pages_data: + results.extend(page_items) + + # 중복 제거 + seen: set[str] = set() + unique = [] + for item in results: + if item[0] not in seen: + seen.add(item[0]) + unique.append(item) + + logger.info(" %s: %d개 선정 (총 %d페이지)", market_name, len(unique), total_pages) + return unique + + kospi_stocks: list[tuple[str, str, str]] = [] + kosdaq_stocks: list[tuple[str, str, str]] = [] + async with aiohttp.ClientSession() as session: + kospi_stocks = await _fetch_market_naver(session, 0, "KOSPI") + kosdaq_stocks = await _fetch_market_naver(session, 1, "KOSDAQ") + + # 네이버 스크래핑 실패 시 KIS API로 폴백 + naver_total = len(kospi_stocks) + len(kosdaq_stocks) + if naver_total < MIN_STOCK_COUNT: + logger.warning( + "⚠️ 네이버 금융 조회 부족(%d개) — KIS API로 폴백", naver_total + ) + try: + from app.services.market_analyzer import get_all_tradable_stocks as _kis_stocks + kis_result = await _kis_stocks() + if len(kis_result) >= MIN_STOCK_COUNT: + logger.info("✅ KIS API 폴백 성공: %d개", len(kis_result)) + return kis_result[:limit] + except Exception as _exc: + logger.error("❌ KIS API 폴백도 실패: %s", _exc) + raise RuntimeError( + f"종목 조회 실패: 네이버({naver_total}개) + KIS API 모두 불가. " + f"네트워크 및 API 설정을 확인하세요." + ) + + # KOSPI 60% / KOSDAQ 40% 비율로 합산 + kospi_n = int(limit * 0.6) + kosdaq_n = limit - kospi_n + all_stocks = kospi_stocks[:kospi_n] + kosdaq_stocks[:kosdaq_n] + top_with_mkt = all_stocks[:limit] + top_stocks = [(code, name) for code, name, _ in top_with_mkt] kospi_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSPI") kosdaq_n = sum(1 for *_, mkt in top_with_mkt if mkt == "KOSDAQ") logger.info("✅ 선정 %d개 (KOSPI %d / KOSDAQ %d)", len(top_stocks), kospi_n, kosdaq_n) - if len(top_stocks) < MIN_STOCK_COUNT: - if settings.KIS_MODE == "virtual": - # 모의투자 서버는 KOSDAQ 거래량 순위 미지원 — 조회된 종목으로 진행 - logger.warning( - "⚠️ 모의투자 서버 종목 수 제한: %d개 (기준 %d개). " - "KOSDAQ 거래량 순위 미지원으로 인한 정상 동작 — 배치 계속 진행합니다.", - len(top_stocks), MIN_STOCK_COUNT - ) - else: - raise RuntimeError( - f"종목 조회 결과 너무 적음: {len(top_stocks)}개 (최소 {MIN_STOCK_COUNT}개 필요). " - "KIS API 오류 또는 장 휴장 여부를 확인하세요." - ) - return top_stocks @@ -189,12 +274,20 @@ async def analyze_one_stock( ) -> tuple[str, Optional[StockRecommendation]]: """종목 분석만 수행하고 결과 반환. DB 저장은 Streaming 저장기에서 처리. + DB(stock_daily_prices)에서 차트 데이터를 조회 — KIS API 호출 없음. + Returns: (status, result) status: "ok" | "no_chart" | "price_filter" | "error" """ try: logger.info("[%d/%d] 분석 중: %s (%s)", idx, total, stock_code, stock_name) - result = await analyze_stock_for_batch(stock_code, stock_name) + + chart_data = await _fetch_db_chart(stock_code) + if not chart_data: + logger.info("샘 차트없음(DB) %s (%s)", stock_code, stock_name) + return "no_chart", None + + result = await analyze_stock_for_batch_db(chart_data, stock_code, stock_name) if not result: logger.info("샘 차트없음 %s (%s)", stock_code, stock_name) @@ -300,11 +393,35 @@ async def batch_analyze_all_stocks(limit: int = None): logger.info("⚙️ 동시 처리: %d종목 | 벌크저장 단위: %d개 | API 세마포어: %dreq", BATCH_CONCURRENCY, BULK_SAVE_CHUNK, settings.API_MAX_CONCURRENT_REQUESTS) + # Step 0: DB 데이터 사전 확인 (collect_krx_data.py 미실행 감지) + async with AsyncSessionLocal() as _db: + _db_check = await _db.execute( + sa_select(StockDailyPrice.trade_date) + .order_by(StockDailyPrice.trade_date.desc()) + .limit(1) + ) + _last_row = _db_check.scalar() + if _last_row is None: + logger.error( + "❌ stock_daily_prices 테이블이 비어있습니다. " + "먼저 collect_krx_data.py를 실행하여 데이터를 수집하세요." + ) + raise RuntimeError("DB에 차트 데이터 없음 — collect_krx_data.py 선행 실행 필요") + from datetime import timedelta as _td + _days_old = (date.today() - _last_row).days + if _days_old > 3: + logger.warning( + "⚠️ DB 마지막 데이터: %s (%d일 전) — collect_krx_data.py --update 실행 권장", + _last_row, _days_old + ) + else: + logger.info("✅ DB 최신 데이터: %s", _last_row) + # Step 1: 시장 레짐 판별 regime = await detect_and_log_market_regime() - # Step 2: 종목 목록 조회 (모듈 레벨 싱글톤 재사용, 불필요한 새 인스턴스 생성 금지) - all_stocks = await get_all_tradable_stocks(korea_investment_api, limit=500) + # Step 2: 종목 목록 조회 (KRX OpenAPI) + all_stocks = await get_all_tradable_stocks(limit=500) filtered_stocks = all_stocks[:limit] if limit else all_stocks total = len(filtered_stocks) logger.info("📊 분석 대상: %d개 (시장 레짐: %s)", total, regime) @@ -314,43 +431,41 @@ async def batch_analyze_all_stocks(limit: int = None): queue: asyncio.Queue = asyncio.Queue(maxsize=BULK_SAVE_CHUNK * 3) semaphore = asyncio.Semaphore(BATCH_CONCURRENCY) + # saver_task를 먼저 시작 + saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date)) + async def analyze_one(i: int, code: str, name: str): try: async with semaphore: status, result = await analyze_one_stock(i, total, code, name) - await queue.put((status, code, name, result)) + # saver가 이미 종료(예외 등)됐으면 put 블로킹 방지 + if not saver_task.done(): + await queue.put((status, code, name, result)) + else: + logger.warning("⚠️ saver 종료됨 — %s 저장 건너뜀", code) except Exception as exc: logger.error("❌ analyze_one unexpected error %s: %s", code, exc) - await queue.put(("error", code, name, None)) + if not saver_task.done(): + await queue.put(("error", code, name, None)) - # 🔧 [fix] saver_task 예외 감지: asyncio.wait로 saver_task 비정상 종료 시 조기 감지 - saver_task = asyncio.create_task(_streaming_saver(queue, analysis_date)) - - # 분석 태스크 전체 실행 - analysis_gather = asyncio.gather( + # 분석 태스크 전체 실행 (완료까지 대기) + await asyncio.gather( *[analyze_one(i, c, n) for i, (c, n) in enumerate(filtered_stocks, 1)], return_exceptions=True, ) - # saver_task가 예외로 죽으면 analysis_gather도 조기 취소 - done, pending = await asyncio.wait( - [asyncio.ensure_future(analysis_gather), saver_task], - return_when=asyncio.FIRST_EXCEPTION, - ) - - # saver_task 예외 발생 시 분석 태스크 취소 처리 - for task in pending: - task.cancel() - try: - await task - except (asyncio.CancelledError, Exception): - pass - - # sentinel 전송 → 저장기 종료 (정상 흐름에서도 항상 실행) + # 모든 분석 완료 후 sentinel 전송 → saver 정상 종료 + # 기존 코드의 deadlock 원인: sentinel을 asyncio.wait() 이후에 보냈기 때문에 + # FIRST_EXCEPTION 모드에서 예외 없으면 ALL_COMPLETED 동작 → saver가 sentinel을 + # 기다리며 무한 대기 (deadlock) if not saver_task.done(): await queue.put(None) - stats = await saver_task if not saver_task.cancelled() else {} + stats = {} + try: + stats = await saver_task + except Exception as exc: + logger.error("❌ saver_task 최종 오류: %s", exc) ok_count = stats.get("ok", 0) no_chart = stats.get("no_chart", 0) @@ -375,7 +490,6 @@ async def batch_analyze_all_stocks(limit: int = None): # 결과 요약 duration = (datetime.now(KST) - start_time).total_seconds() - api_stats = korea_investment_api.get_monitor_stats() logger.info("=" * 60) logger.info("✅ Batch analysis completed!") @@ -386,9 +500,6 @@ async def batch_analyze_all_stocks(limit: int = None): logger.info(" ├─ 오류: %d건", err_count) logger.info("🔍 DB 스캐너 신호: %d개", len(scanner_tickers)) logger.info("⏱️ %.1f초 (%.1f분)", duration, duration / 60) - logger.info("📡 API 통계 — 호출:%d 오류:%d(%.1f%%) 평균:%.2f초", - api_stats['total_calls'], api_stats['total_errors'], - api_stats['error_rate'] * 100, api_stats['avg_duration']) logger.info("=" * 60) diff --git a/backend/collect_krx_data.py b/backend/collect_krx_data.py index 2fae511..965bf19 100644 --- a/backend/collect_krx_data.py +++ b/backend/collect_krx_data.py @@ -1,26 +1,31 @@ """ KRX 히스토리컬 데이터 수집 스크립트 -pykrx를 사용하여 KOSPI/KOSDAQ 종목의 일봉 OHLCV 데이터를 DB에 저장합니다. +KRX OpenAPI (data-dbg.krx.co.kr)를 사용하여 KOSPI/KOSDAQ 종목의 +일봉 OHLCV 데이터를 DB에 저장합니다. + +- KIS API 의존성 완전 제거 +- 날짜 1건 호출 → 해당일 전 종목 OHLCV 일괄 수신 (효율적) 사용법: - python collect_krx_data.py # 전체 수집 (10년, 주요 200종목) - python collect_krx_data.py --limit 5 --start-date 20240101 # 테스트 (5종목, 2024년~) - python collect_krx_data.py --update # 증분 업데이트 (마지막 저장일 이후) - python collect_krx_data.py --market kospi # KOSPI만 + python collect_krx_data.py # 전체 수집 (2020-01-01~) + python collect_krx_data.py --limit 5 # 최근 5거래일 (테스트) + python collect_krx_data.py --start-date 20240101 # 시작일 지정 + python collect_krx_data.py --update # 증분 업데이트 + python collect_krx_data.py --market kospi # KOSPI만 """ import argparse import asyncio import logging +import os import sys import time from datetime import date, datetime, timedelta from pathlib import Path -import numpy as np +import aiohttp import pandas as pd -from pykrx import stock as pykrx_stock # 프로젝트 루트를 Python 경로에 추가 sys.path.insert(0, str(Path(__file__).parent)) @@ -31,131 +36,186 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from app.core.database import AsyncSessionLocal, engine, Base from app.models.models import StockDailyPrice -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", -) -logger = logging.getLogger(__name__) +# ─── KRX OpenAPI 설정 ───────────────────────────────────────────── +# 실제 데이터 엔드포인트 (openapi.krx.co.kr 는 포털, 데이터는 data-dbg) +KRX_DATA_BASE = "https://data-dbg.krx.co.kr/svc/apis" -# ETF/특수 종목 제외 키워드 +try: + from app.core.config import settings as _cfg + KRX_API_KEY: str = _cfg.KRX_API_KEY +except Exception: + from dotenv import load_dotenv + load_dotenv(Path(__file__).parent.parent / ".env") + KRX_API_KEY = os.environ.get("KRX_API_KEY", "") + +# ─── ETF/특수 종목 제외 키워드 ──────────────────────────────────── SKIP_KEYWORDS = [ "ETF", "레버리지", "인버스", "KODEX", "TIGER", "ARIRANG", "KBSTAR", "HANARO", "SOL", "ACE", "KOSEF", "KINDEX", "선물", "스팩", "리츠", "SPAC", ] +# 날짜별 동시 요청 수 (KRX API 쓰로틀링 고려) +COLLECT_CONCURRENCY = 3 +REQUEST_INTERVAL = 0.5 # 요청 간 최소 대기(초) -def get_stock_list(market: str = "all", limit: int = 0) -> list: - """pykrx로 종목 목록 조회 - - Args: - market: "kospi", "kosdaq", "all" - limit: 0이면 전체, N이면 상위 N개 - - Returns: - [(stock_code, stock_name), ...] - """ - today = datetime.now().strftime("%Y%m%d") - stocks = [] - - markets = [] - if market in ("kospi", "all"): - markets.append("KOSPI") - if market in ("kosdaq", "all"): - markets.append("KOSDAQ") - - for mkt in markets: - try: - tickers = pykrx_stock.get_market_ticker_list(today, market=mkt) - for ticker in tickers: - name = pykrx_stock.get_market_ticker_name(ticker) - if any(k in name for k in SKIP_KEYWORDS): - continue - stocks.append((ticker, name)) - logger.info(f"{mkt}: {len(tickers)}개 조회, 필터링 후 {len(stocks)}개") - except Exception as e: - logger.error(f"{mkt} 종목 목록 조회 실패: {e}") - - # 중복 제거 - seen = set() - unique = [] - for code, name in stocks: - if code not in seen: - seen.add(code) - unique.append((code, name)) - - if limit > 0: - unique = unique[:limit] - - logger.info(f"총 {len(unique)}개 종목 선정") - return unique +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) -def fetch_ohlcv(stock_code: str, start_date: str, end_date: str) -> pd.DataFrame: - """pykrx로 일봉 OHLCV 데이터 조회 +# ─── 유틸 ───────────────────────────────────────────────────────── - Returns: - DataFrame with columns: date, open, high, low, close, volume, change_rate, market_cap - """ +def generate_trading_dates(start: str, end: str) -> list[str]: + """start ~ end 사이 평일(월~금) 날짜 문자열 목록 생성 (YYYYMMDD).""" + d = datetime.strptime(start, "%Y%m%d").date() + e = datetime.strptime(end, "%Y%m%d").date() + dates: list[str] = [] + while d <= e: + if d.weekday() < 5: # 0=Mon … 4=Fri + dates.append(d.strftime("%Y%m%d")) + d += timedelta(days=1) + return dates + + +def _int(val) -> int: + """문자열/숫자를 int로 안전 변환 (쉼표 포함 처리).""" try: - df = pykrx_stock.get_market_ohlcv_by_date( - start_date, end_date, stock_code, adjusted=True - ) - if df.empty: - return pd.DataFrame() + return int(str(val).replace(",", "").strip() or "0") + except (ValueError, TypeError): + return 0 - # 시가총액 조회 - try: - cap_df = pykrx_stock.get_market_cap_by_date(start_date, end_date, stock_code) - if not cap_df.empty: - df["시가총액"] = cap_df["시가총액"] - except Exception: - df["시가총액"] = None - df = df.reset_index() - df.columns = [c.replace("날짜", "date") for c in df.columns] +def _float(val) -> float: + """문자열/숫자를 float로 안전 변환.""" + try: + return float(str(val).replace(",", "").strip() or "0") + except (ValueError, TypeError): + return 0.0 - result = pd.DataFrame({ - "date": df.iloc[:, 0], # 날짜 - "open": df["시가"].astype(int), - "high": df["고가"].astype(int), - "low": df["저가"].astype(int), - "close": df["종가"].astype(int), - "volume": df["거래량"].astype(int), - "change_rate": df["등락률"] / 100.0 if "등락률" in df.columns else 0, - "market_cap": df.get("시가총액"), - }) - # 거래량 0인 날 제거 (거래정지일) - result = result[result["volume"] > 0].reset_index(drop=True) - return result +# ─── KRX API 호출 ───────────────────────────────────────────────── - except Exception as e: - logger.error(f"{stock_code} OHLCV 조회 실패: {e}") +async def _fetch_krx( + session: aiohttp.ClientSession, + endpoint: str, + bas_dd: str, +) -> list: + """KRX OpenAPI 단건 호출 → OutBlock_1 리스트 반환. + + endpoint 예: "sto/stk_bydd_trd" + """ + url = f"{KRX_DATA_BASE}/{endpoint}" + try: + async with session.get( + url, + params={"basDd": bas_dd}, + headers={"AUTH_KEY": KRX_API_KEY}, + timeout=aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status != 200: + logger.warning("KRX %s %s → HTTP %s", endpoint, bas_dd, resp.status) + return [] + data = await resp.json(content_type=None) + return data.get("OutBlock_1") or [] + except Exception as exc: + logger.error("KRX API 오류 [%s %s]: %s", endpoint, bas_dd, exc) + return [] + + +async def fetch_daily_prices( + session: aiohttp.ClientSession, + market: str, + bas_dd: str, +) -> pd.DataFrame: + """특정 날짜의 전 종목 OHLCV 조회. + + KRX 응답 필드: + ISU_SRT_CD : 단축 종목코드 (e.g. "005930") + ISU_ABBRV : 종목 약칭 + TDD_OPNPRC : 시가 + TDD_HGPRC : 고가 + TDD_LWPRC : 저가 + TDD_CLSPRC : 종가 + ACC_TRDVOL : 누적 거래량 + FLUC_RT : 등락률 (%) + """ + endpoint = "sto/stk_bydd_trd" if market == "kospi" else "sto/ksq_bydd_trd" + items = await _fetch_krx(session, endpoint, bas_dd) + if not items: return pd.DataFrame() + trade_date = datetime.strptime(bas_dd, "%Y%m%d").date() + rows: list[dict] = [] + + for item in items: + name = (item.get("ISU_ABBRV") or item.get("ISU_NM") or "").strip() + if any(kw in name for kw in SKIP_KEYWORDS): + continue + + code = item.get("ISU_SRT_CD", "").strip() + if not code: + continue + + volume = _int(item.get("ACC_TRDVOL")) + if volume == 0: + continue # 거래정지일 제외 + + close = _int(item.get("TDD_CLSPRC")) + if close == 0: + continue + + rows.append({ + "stock_code": code, + "stock_name": name, + "date": trade_date, + "open": _int(item.get("TDD_OPNPRC")), + "high": _int(item.get("TDD_HGPRC")), + "low": _int(item.get("TDD_LWPRC")), + "close": close, + "volume": volume, + "change_rate": _float(item.get("FLUC_RT")) / 100.0, + }) + + return pd.DataFrame(rows) if rows else pd.DataFrame() + + +async def get_stock_codes( + session: aiohttp.ClientSession, + market: str, + bas_dd: str, +) -> set[str]: + """특정 날짜 기준 상장 종목 코드 집합 조회 (선택적 필터용).""" + endpoint = "sto/stk_isu_base_info" if market == "kospi" else "sto/ksq_isu_base_info" + items = await _fetch_krx(session, endpoint, bas_dd) + codes: set[str] = set() + for item in items: + code = item.get("ISU_SRT_CD", "").strip() + if code: + codes.add(code) + return codes + + +# ─── DB 헬퍼 ────────────────────────────────────────────────────── async def create_tables(): - """DB 테이블 생성""" async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("DB 테이블 확인/생성 완료") -async def get_last_trade_date(session, stock_code: str) -> date | None: - """특정 종목의 DB 내 마지막 거래일 조회""" +async def get_last_trade_date(session) -> date | None: + """DB 내 마지막 거래일 조회 (전 종목 공통).""" result = await session.execute( - text( - "SELECT MAX(trade_date) FROM stock_daily_prices WHERE stock_code = :code" - ), - {"code": stock_code}, + text("SELECT MAX(trade_date) FROM stock_daily_prices") ) - row = result.scalar() - return row + return result.scalar() -async def upsert_daily_prices(session, stock_code: str, stock_name: str, df: pd.DataFrame): - """DB에 일봉 데이터 upsert (중복 시 업데이트)""" +async def upsert_daily_prices(session, df: pd.DataFrame) -> int: + """DB에 일봉 데이터 UPSERT (배치 500건).""" if df.empty: return 0 @@ -164,38 +224,35 @@ async def upsert_daily_prices(session, stock_code: str, stock_name: str, df: pd. trade_date = row["date"] if isinstance(trade_date, pd.Timestamp): trade_date = trade_date.date() - - record = { - "stock_code": stock_code, - "stock_name": stock_name, - "trade_date": trade_date, - "open_price": int(row["open"]), - "high_price": int(row["high"]), - "low_price": int(row["low"]), + records.append({ + "stock_code": row["stock_code"], + "stock_name": row["stock_name"], + "trade_date": trade_date, + "open_price": int(row["open"]), + "high_price": int(row["high"]), + "low_price": int(row["low"]), "close_price": int(row["close"]), - "volume": int(row["volume"]), + "volume": int(row["volume"]), "change_rate": float(row["change_rate"]) if pd.notna(row["change_rate"]) else None, - "market_cap": int(row["market_cap"]) if pd.notna(row.get("market_cap")) else None, - } - records.append(record) + "market_cap": None, + }) - # PostgreSQL ON CONFLICT upsert (batch) batch_size = 500 inserted = 0 for i in range(0, len(records), batch_size): - batch = records[i:i + batch_size] + batch = records[i : i + batch_size] stmt = pg_insert(StockDailyPrice).values(batch) stmt = stmt.on_conflict_do_update( constraint="uq_stock_daily", set_={ - "stock_name": stmt.excluded.stock_name, - "open_price": stmt.excluded.open_price, - "high_price": stmt.excluded.high_price, - "low_price": stmt.excluded.low_price, + "stock_name": stmt.excluded.stock_name, + "open_price": stmt.excluded.open_price, + "high_price": stmt.excluded.high_price, + "low_price": stmt.excluded.low_price, "close_price": stmt.excluded.close_price, - "volume": stmt.excluded.volume, + "volume": stmt.excluded.volume, "change_rate": stmt.excluded.change_rate, - "market_cap": stmt.excluded.market_cap, + "market_cap": stmt.excluded.market_cap, }, ) await session.execute(stmt) @@ -205,108 +262,146 @@ async def upsert_daily_prices(session, stock_code: str, stock_name: str, df: pd. return inserted +# ─── 날짜별 수집 ────────────────────────────────────────────────── + +async def collect_date( + i: int, + total: int, + bas_dd: str, + market: str, + semaphore: asyncio.Semaphore, + http_session: aiohttp.ClientSession, +) -> tuple[bool, int]: + """단일 날짜 수집 (세마포어 적용).""" + async with semaphore: + try: + markets = [] + if market in ("kospi", "all"): + markets.append("kospi") + if market in ("kosdaq", "all"): + markets.append("kosdaq") + + frames: list[pd.DataFrame] = [] + for mkt in markets: + df = await fetch_daily_prices(http_session, mkt, bas_dd) + if not df.empty: + frames.append(df) + await asyncio.sleep(REQUEST_INTERVAL) + + if not frames: + logger.info("[%d/%d] %s: 데이터 없음 (휴장일?)", i, total, bas_dd) + return True, 0 # 휴장일은 성공으로 처리 + + combined = pd.concat(frames, ignore_index=True) + # 같은 날짜 내 코드 중복 제거 (KOSPI/KOSDAQ 겹침 방지) + combined = combined.drop_duplicates(subset=["stock_code", "date"]) + + async with AsyncSessionLocal() as db: + count = await upsert_daily_prices(db, combined) + + logger.info("[%d/%d] %s: %d종목 저장", i, total, bas_dd, count) + return True, count + + except Exception as exc: + logger.error("[%d/%d] %s 실패: %s", i, total, bas_dd, exc) + return False, 0 + + +# ─── 메인 수집 로직 ─────────────────────────────────────────────── + async def collect_all( market: str = "all", limit: int = 0, - start_date: str = "20160101", + start_date: str = "20200101", update_mode: bool = False, ): - """메인 수집 로직""" + """메인 수집 로직 (KRX OpenAPI 기반, 날짜별 전종목 일괄 수집).""" + if not KRX_API_KEY: + logger.error( + "KRX_API_KEY가 설정되지 않았습니다.\n" + " 1) https://openapi.krx.co.kr 에서 API 키 발급\n" + " 2) .env 파일에 KRX_API_KEY=<발급받은키> 추가\n" + " 3) 다시 실행하세요." + ) + sys.exit(1) + await create_tables() end_date = datetime.now().strftime("%Y%m%d") + actual_start = start_date - # 종목 목록 - stock_list = get_stock_list(market=market, limit=limit) - if not stock_list: - logger.error("종목 목록이 비어있습니다") + if update_mode: + async with AsyncSessionLocal() as db: + last_date = await get_last_trade_date(db) + if last_date: + next_day = last_date + timedelta(days=1) + actual_start = next_day.strftime("%Y%m%d") + if next_day >= date.today(): + logger.info("이미 최신 데이터입니다 (마지막: %s)", last_date) + return + logger.info("증분 업데이트: %s ~", actual_start) + + dates = generate_trading_dates(actual_start, end_date) + if not dates: + logger.warning("수집할 날짜가 없습니다.") return - total = len(stock_list) - total_inserted = 0 - success_count = 0 - fail_count = 0 + # --limit N → 최근 N거래일만 (테스트용) + if limit > 0: + dates = dates[-limit:] - async with AsyncSessionLocal() as session: - for i, (stock_code, stock_name) in enumerate(stock_list): - try: - actual_start = start_date + total = len(dates) + logger.info("수집 날짜: %d일 (%s ~ %s)", total, dates[0], dates[-1]) - if update_mode: - last_date = await get_last_trade_date(session, stock_code) - if last_date: - # 마지막 저장일 다음 날부터 - next_day = last_date + timedelta(days=1) - actual_start = next_day.strftime("%Y%m%d") - if next_day >= date.today(): - logger.info(f"[{i+1}/{total}] {stock_code} ({stock_name}): 이미 최신") - success_count += 1 - continue + semaphore = asyncio.Semaphore(COLLECT_CONCURRENCY) + conn = aiohttp.TCPConnector(limit=10) - logger.info( - f"[{i+1}/{total}] {stock_code} ({stock_name}): " - f"{actual_start} ~ {end_date} 수집 중..." - ) + async with aiohttp.ClientSession(connector=conn) as http_session: + results = await asyncio.gather( + *[ + collect_date(i, total, d, market, semaphore, http_session) + for i, d in enumerate(dates, 1) + ], + return_exceptions=True, + ) - df = fetch_ohlcv(stock_code, actual_start, end_date) - if df.empty: - logger.warning(f" {stock_code}: 데이터 없음") - fail_count += 1 - continue - - count = await upsert_daily_prices(session, stock_code, stock_name, df) - total_inserted += count - success_count += 1 - logger.info(f" {stock_code}: {count}개 행 저장 (총 {len(df)}일)") - - except Exception as e: - logger.error(f" {stock_code} ({stock_name}) 실패: {e}") - fail_count += 1 - # 에러 시에도 계속 진행 - try: - await session.rollback() - except Exception: - pass - - # pykrx rate limit 방지 - time.sleep(0.5) + success = sum(1 for r in results if isinstance(r, tuple) and r[0]) + fail = total - success + total_rows = sum(r[1] for r in results if isinstance(r, tuple)) logger.info("=" * 60) - logger.info(f"수집 완료!") - logger.info(f" 성공: {success_count}/{total}") - logger.info(f" 실패: {fail_count}/{total}") - logger.info(f" 총 저장 행: {total_inserted:,}") + logger.info("수집 완료!") + logger.info(" 성공: %d/%d일", success, total) + logger.info(" 실패: %d/%d일", fail, total) + logger.info(" 총 저장 행: %s", f"{total_rows:,}") logger.info("=" * 60) def main(): - parser = argparse.ArgumentParser(description="KRX 히스토리컬 데이터 수집") + parser = argparse.ArgumentParser(description="KRX 히스토리컬 데이터 수집 (KRX OpenAPI)") parser.add_argument( - "--market", - choices=["kospi", "kosdaq", "all"], - default="all", + "--market", choices=["kospi", "kosdaq", "all"], default="all", help="수집 시장 (기본: all)", ) parser.add_argument( - "--limit", type=int, default=0, help="종목 수 제한 (0=전체, 테스트용)" + "--limit", type=int, default=0, + help="최근 N거래일만 수집 (0=전체, 테스트용)", ) parser.add_argument( - "--start-date", - default="20160101", - help="수집 시작일 YYYYMMDD (기본: 20160101)", + "--start-date", default="20200101", + help="수집 시작일 YYYYMMDD (기본: 20200101)", ) parser.add_argument( - "--update", - action="store_true", + "--update", action="store_true", help="증분 업데이트 모드 (마지막 저장일 이후만 수집)", ) args = parser.parse_args() - logger.info(f"KRX 데이터 수집 시작") - logger.info(f" 시장: {args.market}") - logger.info(f" 종목 제한: {'전체' if args.limit == 0 else args.limit}") - logger.info(f" 시작일: {args.start_date}") - logger.info(f" 증분 모드: {args.update}") + logger.info("KRX 데이터 수집 시작 (KRX OpenAPI)") + logger.info(" 시장: %s", args.market) + logger.info(" 날짜 제한: %s", "전체" if args.limit == 0 else f"최근 {args.limit}거래일") + logger.info(" 시작일: %s", args.start_date) + logger.info(" 증분 모드: %s", args.update) start_time = time.time() asyncio.run(collect_all( @@ -316,7 +411,7 @@ def main(): update_mode=args.update, )) elapsed = time.time() - start_time - logger.info(f"총 소요시간: {elapsed:.1f}초 ({elapsed/60:.1f}분)") + logger.info("총 소요시간: %.1f초 (%.1f분)", elapsed, elapsed / 60) if __name__ == "__main__": diff --git a/backend/requirements.txt b/backend/requirements.txt index 0b1ec2a..f852f8d 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -11,6 +11,7 @@ redis==5.0.1 aio-pika==9.3.1 websockets==12.0 httpx>=0.27.1 +aiohttp>=3.9.0 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 python-multipart>=0.0.9 @@ -26,9 +27,8 @@ pytest==7.4.3 pytest-asyncio==0.21.1 feedparser==6.0.11 joblib>=1.3.0 -pytz +beautifulsoup4>=4.12.0 # ML Ensemble torch>=2.1.0 lightgbm>=4.1.0 scikit-learn>=1.3.0 -pykrx>=1.0.44 diff --git a/backend/test_krx.py b/backend/test_krx.py new file mode 100644 index 0000000..ec06791 --- /dev/null +++ b/backend/test_krx.py @@ -0,0 +1,54 @@ +import os +import sys +import requests +from pathlib import Path +from dotenv import load_dotenv + +load_dotenv(Path(__file__).parent.parent / ".env") +AUTH_KEY = os.environ.get("KRX_API_KEY", "") +if not AUTH_KEY: + print("❌ KRX_API_KEY가 .env에 설정되지 않았습니다.") + sys.exit(1) + +print("=== KRX OTP 테스트 ===") + +# Step 1: OTP 발급 +otp_params = { + "bld": "dbms/MDC/STAT/standard/MDCSTAT02301", + "name": "fileDown", + "filetype": "json", + "url": "dbms/MDC/STAT/standard/MDCSTAT02301", + "trdDd": "20260304", + "AUTH_KEY": AUTH_KEY +} + +print("OTP 요청 중...") +otp_res = requests.post( + "https://openapi.krx.co.kr/contents/COM/GenerateOTP.cmd", + data=otp_params +) +print(f"OTP 상태코드: {otp_res.status_code}") +print(f"OTP 응답 전체: [{otp_res.text}]") + +if otp_res.status_code != 200 or not otp_res.text.strip(): + print("❌ OTP 발급 실패 — 여기서 중단") + exit(1) + +# Step 2: 데이터 요청 +print("\n데이터 요청 중...") +data_res = requests.post( + "https://openapi.krx.co.kr/contents/COM/GetData.cmd", + data={"code": otp_res.text} +) +print(f"데이터 상태코드: {data_res.status_code}") + +try: + data = data_res.json() + items = data.get("OutBlock_1") or data.get("output") or [] + print(f"✅ 종목 수: {len(items)}개") + if items: + print(f"첫 번째 항목 키: {list(items[0].keys())}") + print(f"첫 번째 종목: {items[0]}") +except Exception as e: + print(f"❌ JSON 파싱 실패: {e}") + print(f"응답 내용: {data_res.text[:500]}") diff --git a/backend/test_naver.py b/backend/test_naver.py new file mode 100644 index 0000000..69c3e42 --- /dev/null +++ b/backend/test_naver.py @@ -0,0 +1,30 @@ +import requests +from bs4 import BeautifulSoup + +HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; autostock-batch/1.0)"} + +url = "https://finance.naver.com/sise/sise_market_sum.naver" +resp = requests.get(url, params={"sosok": 0, "page": 1}, headers=HEADERS) +print(f"상태코드: {resp.status_code}") + +soup = BeautifulSoup(resp.content, "html.parser", from_encoding="euc-kr") + +# 어떤 table 태그들이 있는지 확인 +tables = soup.find_all("table") +print(f"\ntable 개수: {len(tables)}") +for i, t in enumerate(tables[:5]): + print(f" table[{i}] class={t.get('class')}") + +# type_2 테이블 직접 찾기 +t2 = soup.find("table", {"class": "type_2"}) +print(f"\ntype_2 테이블 존재: {t2 is not None}") + +if t2: + rows = t2.find_all("tr") + print(f"tr 개수: {len(rows)}") + for row in rows[:5]: + print(f" row: {str(row)[:200]}") +else: + # HTML 앞부분 출력 + print("\nHTML 앞 2000자:") + print(resp.text[:2000])