diff --git a/backend/app/api/endpoints/ml_ensemble.py b/backend/app/api/endpoints/ml_ensemble.py index c34df15..9ae1d71 100644 --- a/backend/app/api/endpoints/ml_ensemble.py +++ b/backend/app/api/endpoints/ml_ensemble.py @@ -13,6 +13,7 @@ import logging import asyncio from app.services.ml_ensemble import get_ensemble +from app.services.feature_engineering import MIN_BARS_FOR_PREDICT from app.services.korea_investment_api import korea_investment_api from app.core.config import settings @@ -48,12 +49,22 @@ async def predict_stock(stock_code: str): try: chart_resp = await korea_investment_api.get_daily_chart(stock_code, "D") - chart_data = chart_resp.get("output", []) + # get_daily_chart 는 페이지네이션 결과를 output2 에 담는다. + # (과거 이 핸들러는 "output" 을 읽어 항상 빈 리스트 → 400 이었다. 2026-09-28) + chart_data = chart_resp.get("output2") or chart_resp.get("output") or [] - if not chart_data or len(chart_data) < 61: - raise HTTPException(status_code=400, detail=f"차트 데이터 부족: {len(chart_data) if chart_data else 0}일 (최소 61일 필요)") + if len(chart_data) < MIN_BARS_FOR_PREDICT: + raise HTTPException( + status_code=400, + detail=( + f"차트 데이터 부족: {len(chart_data)}봉 " + f"(최소 {MIN_BARS_FOR_PREDICT}봉 필요 — " + f"LSTM 시퀀스 + 지표 워밍업)" + ), + ) - result = ensemble.predict(chart_data) + # async 핸들러에서 동기 predict() 를 호출하면 이벤트 루프를 막는다. + result = await ensemble.predict_async(chart_data) result["stock_code"] = stock_code return result diff --git a/backend/app/services/feature_engineering.py b/backend/app/services/feature_engineering.py index 9403811..779a37a 100644 --- a/backend/app/services/feature_engineering.py +++ b/backend/app/services/feature_engineering.py @@ -41,6 +41,33 @@ def chart_data_to_dataframe(chart_data: list) -> pd.DataFrame: return df +def chart_data_to_dataframe_v2(chart_data: list) -> pd.DataFrame: + """KIS 형식 차트 데이터를 **입력 순서와 무관하게** 시간순 DataFrame으로 변환. + + chart_data_to_dataframe() 은 "입력이 최신순"이라는 가정 아래 무조건 뒤집는다. + 그런데 market_analyzer 는 oldest-first 로 정규화한 배열을 넘기므로 + (analyze_stock: `list(reversed(kis_output2))`, naver: oldest-first) + 그 함수를 그대로 쓰면 **시계열이 역방향으로 뒤집힌 상태**로 피처가 계산된다. + 학습 데이터(db_data_to_dataframe → trade_date ASC)는 정방향이므로 + 추론이 학습과 정반대 방향의 시계열을 보게 되는 결함이었다. (2026-09-28) + + stck_bsop_date 를 기준으로 오름차순 정렬하여 방향 의존성을 제거한다. + 날짜가 없으면 기존 동작(최신순 가정 → reverse)으로 폴백한다. + """ + if not chart_data: + return chart_data_to_dataframe(chart_data) + + dates = [str(item.get("stck_bsop_date", "") or "") for item in chart_data] + if not all(len(d) == 8 and d.isdigit() for d in dates): + # 날짜 누락/형식 이상 → 기존 규약(최신순 입력) 유지 + return chart_data_to_dataframe(chart_data) + + order = sorted(range(len(chart_data)), key=lambda i: dates[i]) + ordered = [chart_data[i] for i in order] + # chart_data_to_dataframe 은 뒤집으므로, 정렬 결과를 한 번 더 뒤집어 넘긴다 + return chart_data_to_dataframe(list(reversed(ordered))) + + def db_data_to_dataframe(rows: list) -> pd.DataFrame: """DB StockDailyPrice 레코드를 DataFrame(시간순)으로 변환. @@ -271,6 +298,26 @@ FEATURE_COLUMNS = [ # LSTM 시퀀스에 사용할 피처 (5개) LSTM_SEQUENCE_FEATURES = ["close", "volume", "rsi_14", "ma20_ratio", "volatility_20d"] +# LSTM 시퀀스 길이 +LSTM_SEQUENCE_LENGTH = 60 + +# ─── 워밍업 상수 (extract_features() 구현과 1:1 대응, 실측값) ─── +# 각 피처가 처음으로 non-NaN 이 되는 인덱스: +# rsi_14=14, macd=25, macd_signal/hist=33, ma60_ratio=59, +# ma20_ratio/bb_position/volume_ratio/disparity_20=19, volatility_20d=20 +# → 탭뷸러 마지막 행이 전부 유효하려면 인덱스 59 이상 (봉 60개) +TABULAR_WARMUP_BARS = 60 +# → LSTM 시퀀스(마지막 60행)가 전부 유효하려면 시퀀스 시작 인덱스 >= 20 +# (volatility_20d 의 첫 유효 인덱스) +SEQUENCE_FEATURE_WARMUP_BARS = 20 + +# predict() 에 필요한 최소 봉수 = 시퀀스 60 + 시퀀스 피처 워밍업 20 = 80 +# (61봉이면 탭뷸러는 통과하지만 시퀀스 앞쪽 20행이 NaN → 모델이 NaN 을 전파해 +# `int(prob_up*100)` 에서 ValueError. 2026-09-23~28 ML 전면 미작동 원인) +MIN_BARS_FOR_PREDICT = LSTM_SEQUENCE_LENGTH + SEQUENCE_FEATURE_WARMUP_BARS +# 권장 봉수: RSI(Wilder)·EMA 등 재귀 지표의 번인을 감안한 여유치 +RECOMMENDED_BARS_FOR_PREDICT = 120 + def prepare_training_data( df: pd.DataFrame, @@ -314,17 +361,6 @@ def prepare_training_data( if valid_mask[i] and np.any(np.isnan(tabular[i])): valid_mask[i] = False - valid_indices = np.where(valid_mask)[0] - - if len(valid_indices) == 0: - return np.array([]), np.array([]), np.array([]), np.array([]) - - # 탭뷸러 피처 - X_tabular = tabular[valid_indices] - - # 레이블 - y = labels[valid_indices].astype(int) - # LSTM 시퀀스 피처 준비 # 시퀀스에 사용할 raw 피처 결합 seq_cols = [] @@ -337,6 +373,31 @@ def prepare_training_data( seq_cols.append(np.zeros(n, dtype=float)) seq_raw = np.column_stack(seq_cols) # (n, 5) + # [2026-09-28] 시퀀스 윈도우 NaN 체크 — 기존에는 `valid_mask[:60]=False` 만 있어 + # 인덱스 60~79 샘플의 시퀀스(seq_raw[idx-60:idx])에 워밍업 NaN 이 섞여 들어갔다. + # X_seq 에 NaN 이 하나라도 있으면 `X_seq.mean(axis=(0,1))` = NaN 이 되어 + # seq_mean/seq_std 가 NaN → LSTM 이 학습·추론 전체에서 NaN 을 뱉는다. + seq_row_finite = np.all(np.isfinite(seq_raw), axis=1) + # 누적합으로 각 윈도우의 유효 행 수를 O(n) 에 계산 + finite_cumsum = np.concatenate(([0], np.cumsum(seq_row_finite.astype(int)))) + for i in range(n): + if not valid_mask[i]: + continue + start = i - seq_window + if start < 0 or (finite_cumsum[i] - finite_cumsum[start]) < seq_window: + valid_mask[i] = False + + valid_indices = np.where(valid_mask)[0] + + if len(valid_indices) == 0: + return np.array([]), np.array([]), np.array([]), np.array([]) + + # 탭뷸러 피처 + X_tabular = tabular[valid_indices] + + # 레이블 + y = labels[valid_indices].astype(int) + X_seq = np.zeros((len(valid_indices), seq_window, len(LSTM_SEQUENCE_FEATURES)), dtype=float) for j, idx in enumerate(valid_indices): X_seq[j] = seq_raw[idx - seq_window:idx] @@ -350,17 +411,54 @@ def extract_single_features(chart_data: list) -> Tuple[Optional[np.ndarray], Opt Returns: tabular_features: (20,) 탭뷸러 피처 또는 None sequence_features: (60, 5) LSTM 시퀀스 또는 None - """ - if not chart_data or len(chart_data) < 61: - return None, None - df = chart_data_to_dataframe(chart_data) + 실패 사유가 필요하면 extract_single_features_v2() 를 사용하세요. + """ + tabular, sequence, _reason = extract_single_features_v2(chart_data) + return tabular, sequence + + +def extract_single_features_v2( + chart_data: list, +) -> Tuple[Optional[np.ndarray], Optional[np.ndarray], Optional[str]]: + """extract_single_features() + 실패 사유 반환 (2026-09-28 신설). + + 기존 버전의 결함 두 가지를 여기서 바로잡는다. + + 1) **최소 봉수 61 → 80**. + 61봉이면 탭뷸러 마지막 행은 유효하지만(워밍업 60) LSTM 시퀴스 + `seq_raw[-60:]` 의 앞쪽 행에 NaN 이 남는다 + (volatility_20d 첫 유효 인덱스 20, ma20_ratio 19, rsi_14 14). + 그 NaN 은 예외를 던지지 않고 LSTM → 가중평균 확률로 **전파**되어 + `int(prob_up * 100)` 에서 `ValueError: cannot convert float NaN to integer` + 를 유발했다. 봉수를 워밍업에 맞춰 올리는 것이 근본 수정이다. + + 2) **입력 순서 의존 제거**. chart_data_to_dataframe() 은 입력이 최신순이라 + 가정해 무조건 뒤집는데, 호출부(market_analyzer)는 oldest-first 를 넘긴다. + → 시계열이 역방향으로 뒤집힌 상태로 피처가 계산되고 있었다. + chart_data_to_dataframe_v2() 로 날짜 기준 정렬한다. + + Returns: + (tabular(20,), sequence(60,5), None) 또는 (None, None, 실패사유) + """ + if not chart_data: + return None, None, "차트 데이터 없음" + if len(chart_data) < MIN_BARS_FOR_PREDICT: + return None, None, ( + f"봉수 부족 {len(chart_data)}/{MIN_BARS_FOR_PREDICT}" + f"(시퀀스{LSTM_SEQUENCE_LENGTH}+워밍업{SEQUENCE_FEATURE_WARMUP_BARS})" + ) + + df = chart_data_to_dataframe_v2(chart_data) + if len(df) < MIN_BARS_FOR_PREDICT: + return None, None, f"변환 후 봉수 부족 {len(df)}/{MIN_BARS_FOR_PREDICT}" features_df = extract_features(df) # 마지막 행의 탭뷸러 피처 - last_row = features_df[FEATURE_COLUMNS].iloc[-1].values - if np.any(np.isnan(last_row)): - return None, None + last_row = features_df[FEATURE_COLUMNS].iloc[-1].values.astype(float) + if not np.all(np.isfinite(last_row)): + bad = [c for c, v in zip(FEATURE_COLUMNS, last_row) if not np.isfinite(v)] + return None, None, f"탭뷸러 피처 무효: {','.join(bad)}" # LSTM 시퀀스 (최근 60일) seq_cols = [] @@ -373,9 +471,16 @@ def extract_single_features(chart_data: list) -> Tuple[Optional[np.ndarray], Opt seq_cols.append(np.zeros(len(df), dtype=float)) seq_raw = np.column_stack(seq_cols) - if len(seq_raw) < 60: - return None, None + if len(seq_raw) < LSTM_SEQUENCE_LENGTH: + return None, None, f"시퀀스 길이 부족 {len(seq_raw)}/{LSTM_SEQUENCE_LENGTH}" - sequence = seq_raw[-60:] # (60, 5) + sequence = seq_raw[-LSTM_SEQUENCE_LENGTH:].astype(float) # (60, 5) - return last_row, sequence + # 최소 봉수를 만족해도 데이터에 결측/0가격이 섞이면 NaN 이 남을 수 있다. + # 여기서 반드시 잡아야 모델이 NaN 을 전파하지 않는다. + if not np.all(np.isfinite(sequence)): + bad_idx = np.where(~np.isfinite(sequence).all(axis=0))[0] + bad = [LSTM_SEQUENCE_FEATURES[i] for i in bad_idx] + return None, None, f"시퀀스 피처 무효: {','.join(bad)}" + + return last_row, sequence, None diff --git a/backend/app/services/market_analyzer.py b/backend/app/services/market_analyzer.py index 5494c1a..963312c 100644 --- a/backend/app/services/market_analyzer.py +++ b/backend/app/services/market_analyzer.py @@ -8,9 +8,10 @@ """ import asyncio import logging +import re import time import pytz -from datetime import datetime +from datetime import datetime, timedelta from typing import List, Optional from pydantic import BaseModel @@ -429,13 +430,144 @@ def _calculate_technical_indicators(chart_data: List[dict], closes: List[float]) return indicators +# ─── 네이버 일봉 봉수 (2026-09-28) ───────────────────────────────────────── +# get_naver_daily_chart(days=N) 은 [오늘-(N+30)일, 오늘] 캘린더 창을 요청한다. +# days=80 → 110일 창 → 실측 75~77 거래일 (추석 연휴 포함 시 75) +# days=200 → 230일 창 → 약 155 거래일 +# ML 최소 요구치는 80봉(시퀀스 60 + 시퀀스 피처 워밍업 20)이므로 기존 기본값으로는 +# 항상 미달이었다. ML 입력만 확장하고 룰 입력은 기존 창으로 자른다. +_NAVER_ML_CHART_DAYS = 200 +_NAVER_LEGACY_DAYS = 80 # 기존 동작(=get_naver_daily_chart 기본값) 재현용 + + +def _trim_to_legacy_window( + chart_data: Optional[List[dict]], legacy_days: int, +) -> Optional[List[dict]]: + """확장 조회 결과를 기존 창(days=legacy_days)과 **동일한 배열**로 자른다. + + get_naver_daily_chart 의 startTime 은 `오늘 - (days+30)` 이므로 같은 경계로 자르면 + 기존 호출과 같은 봉 집합이 된다 → RSI/BB/ATR/ADX 등 룰 지표 값이 변하지 않는다. + (ADX 는 Wilder 재귀라 배열 길이에 민감하므로 이 절단이 필수다.) + """ + if not chart_data: + return chart_data + cutoff = (datetime.now() - timedelta(days=legacy_days + 30)).strftime("%Y%m%d") + trimmed = [ + item for item in chart_data + if str(item.get("stck_bsop_date", "") or "") >= cutoff + ] + return trimmed if trimmed else chart_data + + +# ─── ML 앙상블 스킵 집계 (2026-09-28) ─────────────────────────────────────── +# 9/23~9/28 닷새간 ml_score 가 200/200 NULL 이었는데 아무 로그도 남지 않았다. +# 원인은 `except Exception: pass` 였다. 사유별로 집계해 반드시 눈에 띄게 만든다. +_ML_SKIP_STATS: dict = {"total": 0, "reasons": {}, "logged": 0} +_ML_SUCCESS_COUNT: dict = {"n": 0} +_ML_SKIP_LOG_EVERY = 50 # 최초 1회 + 이후 N회마다 1줄 +_DIGITS_RE = re.compile(r"\d+") + + +def get_ml_skip_stats() -> dict: + """ML 스킵 집계 스냅샷 (배치 사이클 종료 시 요약 로그용).""" + return { + "ml_ok": _ML_SUCCESS_COUNT["n"], + "ml_skipped": _ML_SKIP_STATS["total"], + "reasons": dict(_ML_SKIP_STATS["reasons"]), + } + + +def reset_ml_skip_stats() -> None: + """배치 사이클 시작 시 호출 — 집계 초기화.""" + _ML_SKIP_STATS["total"] = 0 + _ML_SKIP_STATS["reasons"] = {} + _ML_SKIP_STATS["logged"] = 0 + _ML_SUCCESS_COUNT["n"] = 0 + + +def log_ml_skip_summary(context: str = "") -> None: + """사이클 종료 시 사유별 집계를 1줄로 남긴다 (warning 수준).""" + snap = get_ml_skip_stats() + if snap["ml_skipped"] == 0: + logger.info("[ML] %s 예측 성공 %d종목, 스킵 0건", context, snap["ml_ok"]) + return + detail = ", ".join(f"{k}={v}" for k, v in sorted( + snap["reasons"].items(), key=lambda kv: -kv[1])) + logger.warning( + "[ML] %s 스킵 %d종목 (성공 %d종목) — 사유별: %s", + context, snap["ml_skipped"], snap["ml_ok"], detail, + ) + + +def _record_ml_skip(stock_code: str, reason: str) -> None: + """스킵 사유 집계 + 최초/주기적 warning 로그.""" + # 숫자(봉수 등)를 지워 사유 단위로 묶는다 — "봉수 부족 75/80" 과 "77/80" 이 한 버킷 + key = _DIGITS_RE.sub("N", reason)[:80] if reason else "unknown" + _ML_SKIP_STATS["total"] += 1 + _ML_SKIP_STATS["reasons"][key] = _ML_SKIP_STATS["reasons"].get(key, 0) + 1 + n = _ML_SKIP_STATS["total"] + if n == 1 or n % _ML_SKIP_LOG_EVERY == 0: + _ML_SKIP_STATS["logged"] += 1 + logger.warning("[ML] 예측 스킵 %s: %s (누적 %d건)", stock_code, reason, n) + + +def _run_ml_ensemble(chart_data: List[dict], stock_code: str) -> Optional[dict]: + """ML 앙상블 예측 — 실패는 집계·로그로 드러내고 None 을 반환한다. + + ML 실패가 룰 기반 분석 저장을 막지 않도록 여기서 흡수하되, + **조용히 넘기지 않는다**(기존 `except Exception: pass` 가 9/23 무증상의 원인). + """ + try: + from app.services.ml_ensemble import get_ensemble + ensemble = get_ensemble() + if not ensemble.is_trained: + _record_ml_skip(stock_code, "모델 미학습") + return None + ml_result = ensemble.predict(chart_data) # 예외를 던지지 않는 계약 + if "error" in ml_result: + _record_ml_skip(stock_code, f"{ml_result['error']} (봉수 {len(chart_data)})") + return None + if ml_result.get("degraded_models"): + logger.warning( + "[ML] %s 일부 모델 제외 상태로 예측: %s", + stock_code, ml_result["degraded_models"], + ) + _ML_SUCCESS_COUNT["n"] += 1 + return ml_result + except Exception as e: + # predict() 는 예외를 내지 않아야 하므로 여기에 오면 계약 위반 = 버그다. + logger.warning( + "[ML] %s 예측 중 예상치 못한 예외 (%s): %s", + stock_code, type(e).__name__, e, exc_info=True, + ) + _record_ml_skip(stock_code, f"예외 {type(e).__name__}") + return None + + def _build_analysis_result( chart_data: List[dict], chart_info: dict, indicators: dict, criteria: "RecommendationCriteria", ) -> Optional[StockRecommendation]: - """Build analysis result with signals and score. + """Build analysis result with signals and score. (기존 시그니처 유지)""" + return _build_analysis_result_v2(chart_data, chart_info, indicators, criteria) + + +def _build_analysis_result_v2( + chart_data: List[dict], + chart_info: dict, + indicators: dict, + criteria: "RecommendationCriteria", + ml_chart_data: Optional[List[dict]] = None, +) -> Optional[StockRecommendation]: + """_build_analysis_result + ML 전용 장기 차트 주입 (2026-09-28). + + Args: + ml_chart_data: ML 앙상블에만 사용할 더 긴 차트(≥80봉). None 이면 chart_data 사용. + **룰 지표(RSI/BB/ATR/ADX)는 chart_data 만으로 계산되므로 값이 변하지 않는다.** + ADX(Wilder)는 배열 전체를 훑는 재귀 지표라 봉수가 바뀌면 값이 달라진다 — + 그래서 룰 경로의 입력은 절대 늘리지 않고 ML 입력만 분리했다. Returns StockRecommendation or None based on score threshold and criteria. """ @@ -590,26 +722,21 @@ def _build_analysis_result( 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"] + # ML 결과는 chart_data 보다 긴 시계열이 있으면 그쪽을 쓴다 (봉수 부족 방지). + ml_result = _run_ml_ensemble(ml_chart_data or chart_data, stock_code) + if ml_result is not None: + 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)) + 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 + signals.append(f"ML앙상블: {ml_signal_val}({ml_score_val}점, 신뢰도{ml_confidence_val:.0%})") if criteria.for_batch or score >= 2: final_score = max(1, score) if len(signals) > 0 else score @@ -656,6 +783,20 @@ def _analyze_chart_items( 2. 기술적 지표 계산 3. 분석 결과 생성 (신호 및 점수) """ + return _analyze_chart_items_v2(chart_data, stock_code, stock_name, criteria) + + +def _analyze_chart_items_v2( + chart_data: List[dict], + stock_code: str, + stock_name: str, + criteria: "RecommendationCriteria", + ml_chart_data: Optional[List[dict]] = None, +) -> Optional[StockRecommendation]: + """_analyze_chart_items + ML 전용 장기 차트 주입 (2026-09-28). + + ml_chart_data 는 ML 앙상블에만 전달되며 룰 지표 계산에는 쓰이지 않는다. + """ # 1. Extract and validate chart data chart_info = _extract_chart_data(chart_data, stock_code, stock_name) if not chart_info: @@ -665,7 +806,9 @@ def _analyze_chart_items( indicators = _calculate_technical_indicators(chart_data, chart_info["closes"]) # 3. Build analysis result with signals and scoring - return _build_analysis_result(chart_data, chart_info, indicators, criteria) + return _build_analysis_result_v2( + chart_data, chart_info, indicators, criteria, ml_chart_data=ml_chart_data, + ) async def analyze_stock(stock_code: str, stock_name: str, criteria: RecommendationCriteria) -> Optional[StockRecommendation]: @@ -676,24 +819,35 @@ async def analyze_stock(stock_code: str, stock_name: str, criteria: Recommendati """ try: chart_data = None + ml_chart_data = None # 배치 분석: 네이버 일봉 우선 사용 if criteria.for_batch: try: from app.services.naver_finance import get_naver_daily_chart - chart_data = await get_naver_daily_chart(stock_code) - except Exception: - pass + # ML 은 최소 80봉이 필요하다(시퀀스60+워밍업20). 기존 기본값 days=80 은 + # 캘린더 110일 창 → 실제 75~77봉밖에 오지 않아 ML 이 항상 NaN 으로 터졌다. + ml_chart_data = await get_naver_daily_chart( + stock_code, days=_NAVER_ML_CHART_DAYS, + ) + # 룰 지표는 기존과 동일한 창으로 계산한다(ADX Wilder 는 봉수 의존). + chart_data = _trim_to_legacy_window(ml_chart_data, _NAVER_LEGACY_DAYS) + except Exception as e: + logger.warning("네이버 일봉 조회 실패 %s: %s", stock_code, e) # 네이버 실패 또는 실시간 분석 → KIS API 폴백 if chart_data is None: _api = _real_market_api or korea_investment_api + # limit 기본값 240 — ML 최소 요구치(80봉)를 충분히 넘는다 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"])) + ml_chart_data = None # KIS 경로는 동일 배열을 그대로 ML 에 사용 - return _analyze_chart_items(chart_data, stock_code, stock_name, criteria) + return _analyze_chart_items_v2( + chart_data, stock_code, stock_name, criteria, ml_chart_data=ml_chart_data, + ) except Exception as e: logger.error(f"Error analyzing {stock_code}: {str(e)}") return None diff --git a/backend/app/services/ml_ensemble.py b/backend/app/services/ml_ensemble.py index 3bba48c..2500092 100644 --- a/backend/app/services/ml_ensemble.py +++ b/backend/app/services/ml_ensemble.py @@ -21,7 +21,10 @@ import joblib from app.services.feature_engineering import ( FEATURE_COLUMNS, LSTM_SEQUENCE_FEATURES, + LSTM_SEQUENCE_LENGTH, + MIN_BARS_FOR_PREDICT, extract_single_features, + extract_single_features_v2, chart_data_to_dataframe, extract_features, prepare_training_data, @@ -33,6 +36,35 @@ logger = logging.getLogger(__name__) MODEL_DIR = Path(__file__).parent.parent.parent / "models" / "ensemble" +_CLASS_NAMES = ("down", "hold", "up") + + +def _per_class_metrics(y_true: np.ndarray, y_pred: np.ndarray) -> Dict[str, Dict[str, float]]: + """3클래스 precision/recall/support (sklearn 의존 없이 계산). + + 모델 교체 판단의 근거로 metrics.json 에 기록한다. + 정확도만으로는 "전부 HOLD" 모델이 좋아 보이는 문제를 잡을 수 없다. + """ + y_true = np.asarray(y_true, dtype=int) + y_pred = np.asarray(y_pred, dtype=int) + out: Dict[str, Dict[str, float]] = {} + for cls, name in enumerate(_CLASS_NAMES): + tp = int(np.sum((y_pred == cls) & (y_true == cls))) + fp = int(np.sum((y_pred == cls) & (y_true != cls))) + fn = int(np.sum((y_pred != cls) & (y_true == cls))) + precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 + recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0 + f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) > 0 else 0.0 + out[name] = { + "precision": round(float(precision), 4), + "recall": round(float(recall), 4), + "f1": round(float(f1), 4), + "support": int(np.sum(y_true == cls)), + "predicted": int(np.sum(y_pred == cls)), + } + return out + + # ─── PyTorch LSTM 모델 정의 ─── def _get_torch(): @@ -251,14 +283,32 @@ class MLEnsemble: Returns: 학습 메트릭 """ + from datetime import datetime, timezone, timedelta + self._ensure_dir() - metrics = {} + metrics: Dict[str, Any] = {} if len(X_tabular) < 50: return {"error": f"학습 데이터 부족: {len(X_tabular)}개 (최소 50개 필요)"} logger.info(f"앙상블 학습 시작: {len(X_tabular)}개 샘플, 클래스 분포: {np.bincount(y)}") + # ─── 학습 메타데이터 (모델 품질·교체 판단의 근거) ─── + _KST = timezone(timedelta(hours=9)) + _class_counts = np.bincount(np.asarray(y, dtype=int), minlength=3) + metrics["trained_at"] = datetime.now(_KST).isoformat() + metrics["n_samples"] = int(len(X_tabular)) + metrics["n_features"] = int(X_tabular.shape[1]) if X_tabular.ndim > 1 else 0 + metrics["sequence_shape"] = list(np.asarray(X_seq).shape[1:]) if len(X_seq) else [] + metrics["class_distribution"] = { + "down": int(_class_counts[0]), "hold": int(_class_counts[1]), "up": int(_class_counts[2]), + } + metrics["class_ratio"] = { + k: round(float(v) / max(1, len(y)), 4) + for k, v in zip(("down", "hold", "up"), _class_counts[:3]) + } + metrics["weights"] = {"lgbm": self.weights[0], "lstm": self.weights[1], "rf": self.weights[2]} + # 탭뷸러 피처 스케일링 from sklearn.preprocessing import StandardScaler self.scaler = StandardScaler() @@ -272,6 +322,12 @@ class MLEnsemble: X_train, X_val = X_scaled[:split_idx], X_scaled[split_idx:] y_train, y_val = y[:split_idx], y[split_idx:] X_seq_train, X_seq_val = X_seq[:split_idx], X_seq[split_idx:] + metrics["n_train"] = int(len(X_train)) + metrics["n_val"] = int(len(X_val)) + _val_counts = np.bincount(np.asarray(y_val, dtype=int), minlength=3) + metrics["val_class_distribution"] = { + "down": int(_val_counts[0]), "hold": int(_val_counts[1]), "up": int(_val_counts[2]), + } # 1. LightGBM logger.info(">>> LightGBM 학습 중...") @@ -299,6 +355,7 @@ class MLEnsemble: lgbm_val_pred = self.lgbm_model.predict(X_val) lgbm_acc = np.mean(lgbm_val_pred == y_val) metrics["lgbm_accuracy"] = float(lgbm_acc) + metrics["lgbm_per_class"] = _per_class_metrics(y_val, lgbm_val_pred) logger.info(f"LightGBM 검증 정확도: {lgbm_acc:.4f}") importance = self.lgbm_model.feature_importances_ @@ -331,7 +388,12 @@ class MLEnsemble: lstm_val_pred = np.argmax(lstm_val_probs, axis=1) lstm_acc = np.mean(lstm_val_pred == y_val) metrics["lstm_accuracy"] = float(lstm_acc) + metrics["lstm_per_class"] = _per_class_metrics(y_val, lstm_val_pred) metrics["lstm_training"] = lstm_metrics + # seq_std 가 0 에 수렴하면(예: 전부 동일값) 정규화가 폭주한다 → 기록해 둔다 + if self.lstm_model.seq_std is not None: + metrics["lstm_seq_std_min"] = float(np.min(self.lstm_model.seq_std)) + metrics["lstm_seq_mean_abs_max"] = float(np.max(np.abs(self.lstm_model.seq_mean))) logger.info(f"LSTM 검증 정확도: {lstm_acc:.4f}") except Exception as e: logger.error(f"LSTM 학습 실패: {e}") @@ -353,54 +415,86 @@ class MLEnsemble: rf_val_pred = self.rf_model.predict(X_val) rf_acc = np.mean(rf_val_pred == y_val) metrics["rf_accuracy"] = float(rf_acc) + metrics["rf_per_class"] = _per_class_metrics(y_val, rf_val_pred) logger.info(f"Random Forest 검증 정확도: {rf_acc:.4f}") except Exception as e: logger.error(f"Random Forest 학습 실패: {e}") metrics["rf_error"] = str(e) # 앙상블 검증 - ensemble_probs = self._ensemble_predict_proba(X_val, X_seq_val) + _degraded: list = [] + ensemble_probs = self._ensemble_predict_proba(X_val, X_seq_val, degraded=_degraded) + if _degraded: + metrics["degraded_models"] = _degraded if ensemble_probs is not None: ensemble_pred = np.argmax(ensemble_probs, axis=1) ensemble_acc = np.mean(ensemble_pred == y_val) metrics["ensemble_accuracy"] = float(ensemble_acc) + metrics["ensemble_per_class"] = _per_class_metrics(y_val, ensemble_pred) + _pred_counts = np.bincount(np.asarray(ensemble_pred, dtype=int), minlength=3) + metrics["ensemble_pred_distribution"] = { + "down": int(_pred_counts[0]), "hold": int(_pred_counts[1]), "up": int(_pred_counts[2]), + } logger.info(f"앙상블 검증 정확도: {ensemble_acc:.4f}") + else: + metrics["ensemble_error"] = "모든 모델 예측 실패" - self._save_models() + # ⚠️ 순서 중요: _save_models() 가 self.training_metrics 를 metrics.json 으로 + # 쓴다. 예전에는 저장이 대입보다 **먼저** 실행돼 항상 이전 값({})이 기록됐고, + # 그래서 운영 metrics.json 이 빈 dict 였다. (2026-09-28 수정) self.is_trained = True self.training_metrics = metrics + self._save_models() logger.info(f"앙상블 학습 완료: {metrics}") return metrics - def _ensemble_predict_proba(self, X_tabular: np.ndarray, X_seq: np.ndarray) -> Optional[np.ndarray]: - """3개 모델의 가중 평균 확률""" + def _ensemble_predict_proba(self, X_tabular: np.ndarray, X_seq: np.ndarray, + degraded: Optional[list] = None) -> Optional[np.ndarray]: + """3개 모델의 가중 평균 확률. + + Args: + degraded: (선택) 제외된 모델 사유를 담을 리스트. 호출부가 넘기면 + "lgbm:non-finite" 같은 문자열이 추가된다. + + 비유한(NaN/inf) 확률을 뱉는 모델은 앙상블에서 **제외**한다. + 예전에는 NaN 이 그대로 가중평균에 섞여 최종 확률 전체를 NaN 으로 만들고 + `int(prob_up*100)` 에서 예외로 터졌다. 제외 사실은 로그와 degraded 로 + 드러내며, 조용히 넘기지 않는다. + """ probs_list = [] weights_list = [] + def _record(name: str, reason: str): + logger.warning("[ML] %s 모델 제외: %s", name, reason) + if degraded is not None: + degraded.append(f"{name}:{reason}") + + def _add(name: str, probs, weight: float): + arr = np.asarray(probs, dtype=float) + if not np.all(np.isfinite(arr)): + _record(name, "non-finite") + return + probs_list.append(arr) + weights_list.append(weight) + if self.lgbm_model is not None: try: - lgbm_probs = self.lgbm_model.predict_proba(X_tabular) - probs_list.append(lgbm_probs) - weights_list.append(self.weights[0]) - except Exception: - pass + _add("lgbm", self.lgbm_model.predict_proba(X_tabular), self.weights[0]) + except Exception as e: + _record("lgbm", f"{type(e).__name__}: {e}") if self.lstm_model is not None and self.lstm_model.model is not None: try: - lstm_probs = self.lstm_model.predict_proba(X_seq) - probs_list.append(lstm_probs) - weights_list.append(self.weights[1]) - except Exception: - pass + _add("lstm", self.lstm_model.predict_proba(X_seq), self.weights[1]) + except Exception as e: + _record("lstm", f"{type(e).__name__}: {e}") if self.rf_model is not None: try: - rf_probs = self.rf_model.predict_proba(X_tabular) - probs_list.append(rf_probs) - weights_list.append(self.weights[2]) - except Exception: - pass + _add("rf", self.rf_model.predict_proba(X_tabular), self.weights[2]) + except Exception as e: + _record("rf", f"{type(e).__name__}: {e}") if not probs_list: return None @@ -421,22 +515,63 @@ class MLEnsemble: Returns: {ensemble_score, confidence, signal, model_scores, probabilities} + 또는 {"error": 사유, ...} — **어떤 입력에도 예외를 던지지 않는다.** + + 주의: 호출부는 반드시 `"error" in result` 를 확인해야 한다. + error 응답의 ensemble_score=50 은 "중립 플레이스홀더"이며 모델 판단이 아니다. """ + try: + return self._predict_impl(chart_data) + except Exception as e: + # predict() 는 절대 예외를 밖으로 내지 않는다. 호출부의 + # `except Exception: pass` 에 삼켜져 5일간 무증상이던 것이 9/23 사고였다. + logger.warning("[ML] predict 예외 (%s): %s", type(e).__name__, e, exc_info=True) + return { + "error": f"예측 예외 {type(e).__name__}: {e}", + "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD", + } + + def _predict_impl(self, chart_data: list) -> Dict[str, Any]: + """predict() 의 실제 구현 (예외 가드는 predict() 가 담당).""" if not self.is_trained: return {"error": "모델이 학습되지 않았습니다", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"} + if self.scaler is None: + return {"error": "스케일러 없음", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"} - tabular, sequence = extract_single_features(chart_data) + tabular, sequence, reason = extract_single_features_v2(chart_data) if tabular is None or sequence is None: - return {"error": "피처 추출 실패 (데이터 부족)", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"} + return { + "error": f"피처 추출 실패: {reason or '데이터 부족'}", + "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD", + } - tabular_scaled = self.scaler.transform(tabular.reshape(1, -1)) - sequence_batch = sequence.reshape(1, 60, len(LSTM_SEQUENCE_FEATURES)) + tabular_scaled = np.asarray(self.scaler.transform(tabular.reshape(1, -1)), dtype=float) + if not np.all(np.isfinite(tabular_scaled)): + # 스케일러 분산 0 등으로 스케일링 결과가 깨진 경우 — 점수를 만들지 않는다 + return { + "error": "스케일링 결과 비유한(NaN/inf)", + "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD", + } + sequence_batch = sequence.reshape(1, LSTM_SEQUENCE_LENGTH, len(LSTM_SEQUENCE_FEATURES)) - probs = self._ensemble_predict_proba(tabular_scaled, sequence_batch) + degraded: list = [] + probs = self._ensemble_predict_proba(tabular_scaled, sequence_batch, degraded=degraded) if probs is None: - return {"error": "모든 모델 예측 실패", "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD"} + return { + "error": "모든 모델 예측 실패" + (f" ({','.join(degraded)})" if degraded else ""), + "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD", + } - prob_down, prob_hold, prob_up = probs[0] + probs = np.asarray(probs, dtype=float) + if probs.shape[0] < 1 or probs.shape[1] < 3 or not np.all(np.isfinite(probs[0])): + # NaN 을 0 이나 그럴듯한 값으로 치환하지 않는다 — 0점은 "ML이 매수 반대"로 + # 오해되므로, 정직하게 error 를 돌려 호출부가 ML 없이 진행하게 한다. + return { + "error": "앙상블 확률 비유한(NaN/inf)", + "ensemble_score": 50, "confidence": 0.0, "signal": "HOLD", + } + + prob_down, prob_hold, prob_up = probs[0][:3] ensemble_score = int(prob_up * 100) max_prob = max(prob_down, prob_hold, prob_up) confidence = float(max_prob) @@ -473,6 +608,8 @@ class MLEnsemble: "confidence": round(confidence, 4), "signal": signal, "model_scores": model_scores, + # 비유한 확률/예외로 앙상블에서 빠진 모델 — 비어 있어야 정상 + "degraded_models": degraded, "probabilities": { "up": round(float(prob_up), 4), "hold": round(float(prob_hold), 4), diff --git a/backend/batch_daily_analysis.py b/backend/batch_daily_analysis.py index c2fec7d..8360292 100644 --- a/backend/batch_daily_analysis.py +++ b/backend/batch_daily_analysis.py @@ -35,6 +35,9 @@ from sqlalchemy import select as sa_select, func as sa_func from app.models.models import StockDailyPrice from app.services.analysis_db_service import bulk_upsert_analysis, passes_price_filter, run_db_scanner # 🔧 단일화 from app.services.market_analyzer import analyze_stock_for_batch_db, StockRecommendation +from app.services.market_analyzer import ( + reset_ml_skip_stats, log_ml_skip_summary, +) import logging # 🔧 [fix] logging.basicConfig는 __main__ 블록에서만 호출 @@ -516,6 +519,10 @@ 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) + # ML 스킵 집계 초기화 — 사이클 종료 시 사유별 1줄 요약을 남긴다. + # 9/23~9/28 닷새간 ml_score 가 200/200 NULL 이었는데 로그가 0줄이었다. (2026-09-28) + reset_ml_skip_stats() + # Step 0: DB 데이터 사전 확인 (collect_krx_data.py 미실행 감지) async with AsyncSessionLocal() as _db: _db_check = await _db.execute( @@ -629,6 +636,7 @@ async def batch_analyze_all_stocks(limit: int = None): logger.info(" ├─ 가격필터: %d건", price_flt) logger.info(" ├─ 오류: %d건", err_count) logger.info("🔍 DB 스캐너 신호: %d개", len(scanner_tickers)) + log_ml_skip_summary("일일배치") logger.info("⏱️ %.1f초 (%.1f분)", duration, duration / 60) logger.info("=" * 60) diff --git a/backend/tests/test_ml_feature_direction.py b/backend/tests/test_ml_feature_direction.py new file mode 100644 index 0000000..3ded953 --- /dev/null +++ b/backend/tests/test_ml_feature_direction.py @@ -0,0 +1,114 @@ +"""ML 피처 파이프라인 회귀 테스트 (2026-09-28). + +두 결함을 behavioral 하게 고정한다. + +1. **시계열 방향**: market_analyzer 는 chart_data 를 oldest-first 로 정규화해 넘기는데 + chart_data_to_dataframe() 은 "입력은 최신순" 가정으로 무조건 뒤집었다. + → ML 추론이 학습(trade_date ASC)과 정반대 방향의 시계열을 봤다. +2. **시퀀스 워밍업 NaN**: 탭뷸러만 NaN 검사를 받고 LSTM 시퀀스는 검사되지 않아 + 61~79봉 구간이 `int(nan)` → ValueError 크래시 밴드였다. +""" + +import numpy as np +import pytest + +from app.services.feature_engineering import ( + LSTM_SEQUENCE_LENGTH, + MIN_BARS_FOR_PREDICT, + SEQUENCE_FEATURE_WARMUP_BARS, + chart_data_to_dataframe_v2, + extract_single_features_v2, +) + + +def _make_chart(n: int, oldest_first: bool = True) -> list: + """상승 추세 합성 일봉. stck_bsop_date 는 영업일 근사(연속 날짜).""" + rows = [] + for i in range(n): + base = 50_000 + i * 100 + rows.append({ + "stck_bsop_date": f"2026{(1 + i // 28):02d}{(1 + i % 28):02d}", + "stck_oprc": str(base), + "stck_hgpr": str(base + 300), + "stck_lwpr": str(base - 300), + "stck_clpr": str(base + 100), + "acml_vol": str(1_000_000 + i * 1_000), + }) + return rows if oldest_first else list(reversed(rows)) + + +# ─── 1. 방향 무관성 ──────────────────────────────────────────────────────── + +def test_dataframe_v2_is_input_order_agnostic(): + """oldest-first / newest-first 어느 쪽을 넣어도 동일한 시간순 DataFrame.""" + df_old = chart_data_to_dataframe_v2(_make_chart(120, oldest_first=True)) + df_new = chart_data_to_dataframe_v2(_make_chart(120, oldest_first=False)) + + assert list(df_old["date"]) == list(df_new["date"]) + # 시간 오름차순이어야 한다 + dates = [str(d) for d in df_old["date"]] + assert dates == sorted(dates) + + +def test_features_identical_regardless_of_input_order(): + """방향이 뒤집힌 입력에서도 피처가 같아야 한다 (구버전은 달랐다).""" + tab_a, seq_a, err_a = extract_single_features_v2(_make_chart(150, True)) + tab_b, seq_b, err_b = extract_single_features_v2(_make_chart(150, False)) + + assert err_a is None and err_b is None, (err_a, err_b) + np.testing.assert_allclose(tab_a, tab_b, rtol=0, atol=0) + np.testing.assert_allclose(seq_a, seq_b, rtol=0, atol=0) + + +def test_uptrend_is_not_read_as_downtrend(): + """상승 추세 입력이 실제로 상승으로 읽혀야 한다. + + 구버전은 oldest-first 입력을 뒤집어 상승장을 하락장으로 계산했다. + change_rate_1d 부호로 방향을 고정한다. + """ + from app.services.feature_engineering import FEATURE_COLUMNS + + tab, _, err = extract_single_features_v2(_make_chart(150, oldest_first=True)) + assert err is None, err + idx = FEATURE_COLUMNS.index("change_rate_1d") + assert tab[idx] > 0, f"상승 추세인데 change_rate_1d={tab[idx]} (방향 역전 의심)" + + +# ─── 2. 시퀀스 NaN 밴드 ─────────────────────────────────────────────────── + +def test_min_bars_covers_sequence_warmup(): + """최소 봉수는 시퀀스 길이 + 시퀀스 피처 워밍업을 모두 덮어야 한다.""" + assert MIN_BARS_FOR_PREDICT >= LSTM_SEQUENCE_LENGTH + SEQUENCE_FEATURE_WARMUP_BARS + + +@pytest.mark.parametrize("n", [0, 1, 30, 60, 61, 70, 79]) +def test_insufficient_bars_return_reason_not_nan(n): + """봉수 부족은 NaN 을 흘리지 않고 reason 을 반환한다 (구 61~79 크래시 밴드).""" + tab, seq, err = extract_single_features_v2(_make_chart(n)) + assert tab is None and seq is None + assert err, f"n={n} 에서 사유 없이 실패" + + +@pytest.mark.parametrize("n", [80, 81, 120, 200]) +def test_sufficient_bars_produce_finite_features(n): + """최소 봉수 이상에서는 탭뷸러·시퀀스 모두 유한해야 한다.""" + tab, seq, err = extract_single_features_v2(_make_chart(n)) + assert err is None, f"n={n}: {err}" + assert np.all(np.isfinite(tab)), "탭뷸러에 NaN/Inf" + assert np.all(np.isfinite(seq)), "LSTM 시퀀스에 NaN/Inf" + assert seq.shape[0] == LSTM_SEQUENCE_LENGTH + + +# ─── 3. predict() 무예외 계약 ───────────────────────────────────────────── + +@pytest.mark.parametrize("bad", [None, [], [{}], [1, 2, 3], "not-a-list"]) +def test_predict_never_raises_on_bad_input(bad): + """predict() 는 어떤 입력에도 예외를 던지지 않고 dict 를 반환한다. + + 구버전은 NaN 이 섞이면 `int(nan)` → ValueError 로 호출부를 터뜨렸고, + market_analyzer 의 `except Exception: pass` 가 그것을 조용히 삼켰다. + """ + from app.services.ml_ensemble import MLEnsemble + + result = MLEnsemble().predict(bad) + assert isinstance(result, dict)