fix: 트레이딩 로직/AI 정확도 8종 수정 (Altman 과발생·CV누수·MACD·손절역전)

트레이딩 전문가 관점 로직 감사 후 라이브 추천·자동매매 영향 8종 수정.

[명백한 버그]
- MACD 시그널선: macd_line[-9:] 9개만 EMA→평활 거의 안 됨. 전체 시리즈로 수정 (ta-engine)
- 매수 손절가 역전: 하락추세(price<MA60)서 ma60*0.95가 진입가 위로 올라가 RR 역전·즉시손절
  → [-10%,-4%] 밴드로 클램프. 매도(숏)도 대칭 수정 (ta-engine)
- 12-1 모멘텀: closes[-1]은 보유데이터 길이(200~259)따라 룩백 가변→closes[min(251,len-1)] 12개월 고정

[모델 품질]
- Ridge 무스케일 학습: 피처 -100~수천 혼재로 L2 왜곡. StandardScaler 파이프라인 +
  계수 원본공간 역변환 저장(predict-price 무변경, 재구성 오차 3.5e-15)
- CV 패널 누수: 인덱스 분할이 같은 score_date를 train/test로 가르고 +30일 라벨윈도가
  test 피처와 겹침. 날짜분할 + 임바고(7d→12일/30d→39일)로 차단→정직한 IC/r2

[캘리브레이션·리스크]
- Altman Z: 2항 변형에 원본 4항 임계값(2.6/1.1) 적용→전시장 31%가 '부도위험' 오발생.
  실측분포(중앙값1.98/p10 0.51) 기준 부도0.7/안전3.0 재보정
- 일일 -3% halt: realized_pnl만 봐서 평가손실·무매도일 누락. 미실현손익 합산으로 수정

빌드·재기동·라이브 스모크 검증 완료. 자동매매 capital 실잔고 연동은 향후 과제(paper모드 무관).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

score-engine/main.py

@@ -969,9 +969,14 @@ def calc_piotroski_score(curr: dict, prev: dict) -> tuple[int, float, str]:
 # ── 알트만 Z-Score (단순화 — 운전자본·이익잉여금 데이터 부재) ─────
 def calc_altman_z(fin: dict, market_cap: int) -> tuple[float, str, str]:
     """
-    Z'' 비제조업 모델 일부 변형 (가용 변수만 사용)
+    Z'' 비제조업 모델의 2항 변형 (운전자본·이익잉여금 데이터 부재로 X1·X2 항 생략).
     Z_simple = 6.72*(EBIT/총자산) + 1.05*(시총/총부채)
-    > 2.6 안전 / 1.1~2.6 회색 / <1.1 부도위험
+
+    ※ 임계값 재보정: 원본 Z''의 안전선 2.6은 4항(X1·X2 포함) 기준값이라
+      2항만 쓰는 이 변형에 그대로 적용하면 Z가 체계적으로 낮게 나와 부도 신호가 과발생함
+      (실측: 전종목 중앙값 Z≈2.0, p10≈0.5, p75≈5.1인데 <1.1 적용 시 ~31%가 '부도위험').
+      → 2항 변형의 실측 분포에 맞춰 부도선 0.7(하위 ~15%), 안전선 3.0(상위 ~40%)로 보정.
+    > 3.0 안전 / 0.7~3.0 회색 / <0.7 부도위험
     returns: (z_score, signal '매수'|'매도'|'관망', reason)
     """
     op_pf = fin.get("operating_profit", 0) or 0
@@ -982,9 +987,9 @@ def calc_altman_z(fin: dict, market_cap: int) -> tuple[float, str, str]:
     a = op_pf / ta
     b = (market_cap / tl) if tl > 0 else 1.0
     z = 6.72 * a + 1.05 * b
-    if z >= 2.6:
+    if z >= 3.0:
         return round(z, 2), "매수", f"Altman Z {z:.1f} (안전)"
-    if z >= 1.1:
+    if z >= 0.7:
         return round(z, 2), "관망", ""
     return round(z, 2), "매도", f"Altman Z {z:.1f} (부도위험)"
 
@@ -1037,7 +1042,9 @@ async def calc_momentum(conn, stock_code: str, as_of: date | None = None) -> tup
     if len(closes) < 200:
         return 0.0, "관망", ""
     p_recent = closes[20][1]   # 1개월(거래일 ~21) 전
-    p_year   = closes[-1][1]   # 약 12개월 전
+    # 12개월 전(거래일 252) 고정. closes[-1]을 쓰면 보유 데이터 길이(200~259)에 따라
+    # 룩백이 종목마다 달라져 모멘텀 비교가 불가능해짐 → 252봉으로 캡(부족 시 최장 사용).
+    p_year   = closes[min(251, len(closes) - 1)][1]
     if p_year <= 0:
         return 0.0, "관망", ""
     mom = (p_recent - p_year) / p_year * 100
@@ -4510,22 +4517,43 @@ def _eval_metrics(y_true, y_pred) -> dict:
     return out
 
 
-def _walk_forward_folds(rows, n_folds: int):
+def _walk_forward_folds(rows, n_folds: int, embargo_days: int = 0):
     """
-    시간순 정렬된 rows를 (n_folds+1) 블록으로 나누어 expanding window 폴드 생성.
+    시간순 정렬된 rows를 날짜(score_date) 경계로 (n_folds+1) 블록으로 나눈 expanding-window 폴드.
-    각 fold i ∈ [1..n_folds]: train = [0..i*block], test = [i*block..(i+1)*block].
+    각 fold i: test=다음 날짜블록, train=그 이전 날짜들. 단,
-    test 데이터는 학습에 절대 안 들어감 → leakage 없음.
+      ① 같은 score_date가 train/test에 쪼개지지 않도록 '날짜' 단위로 분할
+         (인덱스 분할은 한 날짜의 종목들이 train/test로 갈라져 시장 단면 누수 발생)
+      ② train의 마지막 embargo_days 구간은 purge — train 라벨의 미래 수익 윈도(+N일)가
+         test 피처 시점과 겹치는 패널 데이터 누수(López de Prado purge/embargo) 차단.
     """
     n = len(rows)
     if n < (n_folds + 1) * 3:
         return []
-    block = n // (n_folds + 1)
+    dates = sorted({r["score_date"] for r in rows})
+    if len(dates) < n_folds + 1:
+        # 고유 날짜가 적으면 날짜 분할 불가 → 인덱스 분할로 폴백(임바고는 그대로 적용)
+        block = n // (n_folds + 1)
+        folds = []
+        for i in range(1, n_folds + 1):
+            te = rows[i * block:min(n, (i + 1) * block)]
+            if not te:
+                continue
+            emb_cut = te[0]["score_date"] - timedelta(days=embargo_days)
+            tr = [r for r in rows[:i * block] if r["score_date"] <= emb_cut]
+            if len(tr) < 5 or len(te) < 3:
+                continue
+            folds.append((tr, te))
+        return folds
+    dblock = len(dates) // (n_folds + 1)
     folds = []
     for i in range(1, n_folds + 1):
-        tr_end = i * block
+        te_start_date = dates[i * dblock]
-        te_end = min(n, (i + 1) * block)
+        te_end_date = dates[min(len(dates), (i + 1) * dblock) - 1]
-        tr = rows[:tr_end]; te = rows[tr_end:te_end]
+        emb_cut = te_start_date - timedelta(days=embargo_days)
-        if len(tr) < 5 or len(te) < 3: continue
+        tr = [r for r in rows if r["score_date"] <= emb_cut]
+        te = [r for r in rows if te_start_date <= r["score_date"] <= te_end_date]
+        if len(tr) < 5 or len(te) < 3:
+            continue
         folds.append((tr, te))
     return folds
 
@@ -4575,33 +4603,44 @@ async def learn_pricing(days: int = Query(default=180, ge=14, le=730),
         import numpy as np
         from sklearn.linear_model import Ridge
         from sklearn.ensemble import RandomForestRegressor
+        from sklearn.preprocessing import StandardScaler
+        from sklearn.pipeline import make_pipeline
     except Exception as e:
         return {**out, "err": f"sklearn import 실패: {e}"}
 
-    folds = _walk_forward_folds(rows, n_folds)
+    # 라벨 수익 윈도 길이만큼 train↔test 사이 임바고 (7d→+11일, 30d→+38일 윈도 → +1 여유)
+    embargo_days = {"return_7d": 12, "return_30d": 39, "alpha_30d": 39}.get(target, 39)
+    folds = _walk_forward_folds(rows, n_folds, embargo_days=embargo_days)
     if not folds:
         return {**out, "err": "fold 구성 실패 (표본 부족)"}
 
-    # ── 1. Linear (Ridge) ─────────────────────────────────
+    # ── 1. Linear (Ridge + 표준화) ─────────────────────────
+    # 피처 스케일이 -100~수천(점수 vs Amihud illiq vs log_mcap)으로 천차만별이라
+    # Ridge L2 패널티가 스케일에 휘둘려 계수가 왜곡됨 → 폴드마다 train에만 fit한
+    # StandardScaler를 파이프라인으로 결합. (scaler는 train 통계만 사용 → 누수 없음)
     fold_metrics_lin = []
-    last_lin = None
     for tr, te in folds:
         X_tr = np.array([_row_features(r) for r in tr])
         y_tr = np.array([float(r[target]) for r in tr])
         X_te = np.array([_row_features(r) for r in te])
         y_te = np.array([float(r[target]) for r in te])
-        m = Ridge(alpha=1.0).fit(X_tr, y_tr)
+        m = make_pipeline(StandardScaler(), Ridge(alpha=1.0)).fit(X_tr, y_tr)
-        y_p = m.predict(X_te)
+        fold_metrics_lin.append(_eval_metrics(y_te, m.predict(X_te)))
-        fold_metrics_lin.append(_eval_metrics(y_te, y_p))
-        last_lin = m  # 마지막(최대 train) 모델 저장용
 
     linear_metrics = _aggregate_fold_metrics(fold_metrics_lin)
 
-    # 전체 데이터 재학습 (배포용 모델)
+    # 전체 데이터 재학습 (배포용 모델) — 표준화 공간에서 학습 후 계수를 원본 피처 공간으로
+    # 역변환해 저장. /predict-price가 원본 피처로 intercept+Σcoef·x를 그대로 계산하므로
+    # 스케일러를 따로 들고 다닐 필요 없이 동일 예측이 보장됨.
     X_all = np.array([_row_features(r) for r in rows])
     y_all = np.array([float(r[target]) for r in rows])
-    final_lin = Ridge(alpha=1.0).fit(X_all, y_all)
+    final_pipe = make_pipeline(StandardScaler(), Ridge(alpha=1.0)).fit(X_all, y_all)
-    lin_coef = {fn: round(float(c), 6) for fn, c in zip(LEARN_FEATURE_NAMES, final_lin.coef_)}
+    _scaler = final_pipe.named_steps["standardscaler"]
+    _ridge = final_pipe.named_steps["ridge"]
+    _scale = np.where(_scaler.scale_ == 0, 1.0, _scaler.scale_)
+    raw_coef = _ridge.coef_ / _scale
+    lin_intercept = float(_ridge.intercept_ - np.sum(_ridge.coef_ * _scaler.mean_ / _scale))
+    lin_coef = {fn: round(float(c), 6) for fn, c in zip(LEARN_FEATURE_NAMES, raw_coef)}
 
     # ── 2. Random Forest ──────────────────────────────────
     fold_metrics_rf = []
@@ -4662,7 +4701,7 @@ async def learn_pricing(days: int = Query(default=180, ge=14, le=730),
                 sample_size=$10, period_days=$11
         """, today_d, segment, target,
             json.dumps(LEARN_FEATURE_NAMES), json.dumps(lin_coef),
-            float(final_lin.intercept_),
+            lin_intercept,
             linear_metrics.get("r2_oos"), linear_metrics.get("ic_spearman"),
             linear_metrics.get("hit_ratio"), len(rows), days)
         # RF
@@ -6376,19 +6415,34 @@ async def auto_trade_scan_job():
         async with pg_pool.acquire() as conn:
             pnl = await conn.fetchrow(
                 "SELECT realized_pnl, halted FROM trading_daily_pnl WHERE dt=CURRENT_DATE")
+            if pnl and pnl["halted"]:
+                return  # 이미 중단된 날
             capital = TRADE_SETTINGS["default_capital"]
-            if pnl and not pnl["halted"]:
+            realized = (pnl["realized_pnl"] if pnl else 0) or 0
-                loss_pct = (pnl["realized_pnl"] or 0) / capital * 100
+            # 미실현 손익까지 합산. 기존엔 realized만 봐서 ① 평가손실이 아무리 커도
-                if loss_pct <= TRADE_SETTINGS["daily_loss_limit_pct"]:
+            # 매도 전엔 halt가 안 걸리고 ② 매도 없는 날은 pnl row 자체가 없어 체크를 건너뛰는
-                    await conn.execute("""
+            # 구멍이 있었음 → 보유 평가손익을 더해 포트폴리오 기준으로 판단.
-                        UPDATE trading_daily_pnl SET halted=true WHERE dt=CURRENT_DATE
+            unreal = await conn.fetchval("""
-                    """)
+                SELECT COALESCE(SUM((st.price - p.buy_price) * p.qty), 0)
-                    await send_telegram(
+                FROM user_portfolio p
-                        f"🛑 <b>일일 손실 한도 도달</b>\n"
+                JOIN stock_technical st ON st.stock_code = p.stock_code
-                        f"손실 {loss_pct:.2f}% ≤ -{abs(TRADE_SETTINGS['daily_loss_limit_pct'])}%\n"
+                WHERE p.active = true AND p.buy_price > 0 AND st.price > 0
-                        f"오늘 자동매매 중단됨"
+            """) or 0
-                    )
+            loss_pct = (realized + unreal) / capital * 100
-                    return
+            if loss_pct <= TRADE_SETTINGS["daily_loss_limit_pct"]:
+                await conn.execute("""
+                    INSERT INTO trading_daily_pnl (dt, unrealized_pnl, halted)
+                    VALUES (CURRENT_DATE, $1, true)
+                    ON CONFLICT (dt) DO UPDATE SET
+                      unrealized_pnl=$1, halted=true, updated_at=NOW()
+                """, int(unreal))
+                await send_telegram(
+                    f"🛑 <b>일일 손실 한도 도달</b>\n"
+                    f"실현+평가 손실 {loss_pct:.2f}% ≤ -{abs(TRADE_SETTINGS['daily_loss_limit_pct'])}%\n"
+                    f"(실현 {realized:+,}원 / 평가 {int(unreal):+,}원)\n"
+                    f"오늘 자동매매 중단됨"
+                )
+                return
             # 매도 스캔 먼저 (보유 종목 리스크 관리 우선)
             sell_codes = [r["stock_code"] for r in await conn.fetch(
                 "SELECT stock_code FROM user_portfolio WHERE active=true")]

ta-engine/main.py

@@ -80,7 +80,9 @@ def _macd(closes: List[float]) -> tuple:
     e12 = _ema_series(closes, 12)
     e26 = _ema_series(closes, 26)
     macd_line = [a - b for a, b in zip(e12, e26)]
-    signal_line = _ema_series(macd_line[-9:] if len(macd_line) >= 9 else macd_line, 9)
+    # 시그널선 = MACD선 전체에 대한 9-EMA. 과거 9개(-9:)만 쓰면 평활이 거의 안 돼
+    # 표준 MACD 시그널과 달라짐 → 전체 시리즈로 EMA 계산.
+    signal_line = _ema_series(macd_line, 9)
     macd = macd_line[-1]
     signal = signal_line[-1]
     return round(macd, 4), round(signal, 4), round(macd - signal, 4)
@@ -335,9 +337,11 @@ def calc_price_targets(price: int, ind: dict, sig: str) -> dict:
         t2   = r10(price * 1.14)
         t3   = r10(min(price * 1.22, h52 * 0.97))
         t3   = t3 if t3 > t2 else r10(price * 1.22)
-        # 손절: max(-8%, MA60 -5%) — 최대 -10% 이내 제한
+        # 손절: max(-8%, MA60 -5%)를 [-10%, -4%] 밴드로 제한해 항상 진입가 아래 유지.
+        # (하락추세로 price<MA60이면 MA60*0.95가 현재가 위로 올라가 손절가>진입가가 되어
+        #  RR이 역전·즉시 손절되는 버그 차단)
         raw_stop = max(price * 0.92, ma60 * 0.95)
-        stop = r10(max(raw_stop, price * 0.90))  # 최소 -10%
+        stop = r10(min(max(raw_stop, price * 0.90), price * 0.96))
         er1  = round((t1 - price) / price * 100, 1)
         sl_r = round(abs(stop - price) / price * 100, 1)
         # M3: ATR 기반 trailing stop (현재가 기준 2 ATR 아래)
@@ -360,8 +364,10 @@ def calc_price_targets(price: int, ind: dict, sig: str) -> dict:
         t2   = r10(price * 0.86)
         t3   = r10(max(price * 0.78, l52 * 1.03))
         t3   = t3 if t3 < t2 else r10(price * 0.78)
+        # 숏 손절: [+4%, +10%] 밴드로 제한해 항상 진입가(+2%) 위 유지
+        # (price>MA20이면 ma20*1.05가 현재가 아래로 내려가 손절가<진입가 역전되는 버그 차단)
         raw_stop = min(price * 1.08, ma20 * 1.05)
-        stop = r10(min(raw_stop, price * 1.10))
+        stop = r10(max(min(raw_stop, price * 1.10), price * 1.04))
         er1  = round((price - t1) / price * 100, 1)
         sl_r = round(abs(stop - price) / price * 100, 1)
         return {