diff --git a/score-engine/main.py b/score-engine/main.py
index 39ae1f1..bdb26ea 100644
--- a/score-engine/main.py
+++ b/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 폴드 생성.
- 각 fold i ∈ [1..n_folds]: train = [0..i*block], test = [i*block..(i+1)*block].
- test 데이터는 학습에 절대 안 들어감 → leakage 없음.
+ 시간순 정렬된 rows를 날짜(score_date) 경계로 (n_folds+1) 블록으로 나눈 expanding-window 폴드.
+ 각 fold i: test=다음 날짜블록, train=그 이전 날짜들. 단,
+ ① 같은 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_end = min(n, (i + 1) * block)
- tr = rows[:tr_end]; te = rows[tr_end:te_end]
- if len(tr) < 5 or len(te) < 3: continue
+ te_start_date = dates[i * dblock]
+ te_end_date = dates[min(len(dates), (i + 1) * dblock) - 1]
+ emb_cut = te_start_date - timedelta(days=embargo_days)
+ 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)
- y_p = m.predict(X_te)
- fold_metrics_lin.append(_eval_metrics(y_te, y_p))
- last_lin = m # 마지막(최대 train) 모델 저장용
+ m = make_pipeline(StandardScaler(), Ridge(alpha=1.0)).fit(X_tr, y_tr)
+ fold_metrics_lin.append(_eval_metrics(y_te, m.predict(X_te)))
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)
- lin_coef = {fn: round(float(c), 6) for fn, c in zip(LEARN_FEATURE_NAMES, final_lin.coef_)}
+ final_pipe = make_pipeline(StandardScaler(), Ridge(alpha=1.0)).fit(X_all, y_all)
+ _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"]:
- loss_pct = (pnl["realized_pnl"] or 0) / capital * 100
- if loss_pct <= TRADE_SETTINGS["daily_loss_limit_pct"]:
- await conn.execute("""
- UPDATE trading_daily_pnl SET halted=true WHERE dt=CURRENT_DATE
- """)
- await send_telegram(
- f"🛑 일일 손실 한도 도달\n"
- f"손실 {loss_pct:.2f}% ≤ -{abs(TRADE_SETTINGS['daily_loss_limit_pct'])}%\n"
- f"오늘 자동매매 중단됨"
- )
- return
+ realized = (pnl["realized_pnl"] if pnl else 0) or 0
+ # 미실현 손익까지 합산. 기존엔 realized만 봐서 ① 평가손실이 아무리 커도
+ # 매도 전엔 halt가 안 걸리고 ② 매도 없는 날은 pnl row 자체가 없어 체크를 건너뛰는
+ # 구멍이 있었음 → 보유 평가손익을 더해 포트폴리오 기준으로 판단.
+ unreal = await conn.fetchval("""
+ SELECT COALESCE(SUM((st.price - p.buy_price) * p.qty), 0)
+ FROM user_portfolio p
+ JOIN stock_technical st ON st.stock_code = p.stock_code
+ WHERE p.active = true AND p.buy_price > 0 AND st.price > 0
+ """) or 0
+ loss_pct = (realized + unreal) / capital * 100
+ 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"🛑 일일 손실 한도 도달\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")]
diff --git a/ta-engine/main.py b/ta-engine/main.py
index a45a4c2..c1edc6c 100644
--- a/ta-engine/main.py
+++ b/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진입가가 되어
+ # 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 {