Merge branch 'fix/news-sources-accuracy'

[소스 다변화] 라이브 검증(200+한글기사)된 10개 피드 추가:
연합인포맥스(채권·FX)·매경증권·한경 글로벌마켓·연합뉴스 산업·데일리안·
ZDNet코리아·테크M·전자신문 반도체·오토헤럴드·철강금속신문.
마켓/FX·IT·반도체·자동차·철강 버티컬로 커버리지 확대.
검증: 37소스 라이브 크롤 360건, fetch 에러 0.

[종목 귀속 정확성] extract_stocks의 한글명 부분일치(text.count) 오탐 차단.
일상어와 겹치는 모호 종목명 19개(대상·동양·동서·미래 등)는 본문에 주식
맥락 토큰(주가·실적·영업이익·코스피·㈜·반도체 등)이 있을 때만 인정.
보령·풍산·세방 등 고유명은 제외(누락 방지).
검증: '지원 대상으로'→무매칭, '대상 영업이익 증가'→대상 인정, 삼성전자 영향없음.

두 서비스 재빌드·재기동·라이브검증 완료.

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

bareunaapi/main.py

@@ -125,11 +125,29 @@ def extract_morphemes(text):
     except:
         return [w for w in text.split() if len(w) >= 2 and w not in STOPWORDS]
 
+# 일상어와 철자가 겹쳐 오탐이 잦은 짧은 종목명 — 본문에 주식 맥락 토큰이 같이 있을 때만 인정.
+# (예: "지원 대상으로 선정"→대상㈜ 오탐, "동양적 가치관"→동양 오탐 차단)
+# 보령·대덕·풍산·세방처럼 일상어로 거의 안 쓰이는 고유 종목명은 넣지 않음(오히려 누락 유발).
+AMBIGUOUS_NAMES = {
+    "대상", "동양", "동서", "미래", "대한", "고려", "한일", "서울", "부산",
+    "동국", "유성", "남성", "영원", "태양", "대성", "백산", "신성", "한창", "우성",
+}
+_STOCK_CTX = (
+    "주가", "주식", "증시", "상장", "상폐", "공시", "영업이익", "매출", "순이익",
+    "실적", "수주", "계약", "배당", "자사주", "유상증자", "무상증자", "목표주가",
+    "코스피", "코스닥", "급등", "급락", "상한가", "하한가", "㈜", "그룹", "지주",
+    "전자", "화학", "제약", "바이오", "반도체", "자동차", "건설", "증권", "철강",
+)
+
 def extract_stocks(text):
+    has_ctx = any(m in text for m in _STOCK_CTX)
     found = {}
     for name, code in state.stock_map.items():
         if len(name) < 2:
             continue                       # 1글자 종목명은 오탐 과다 → 제외
+        # 모호 종목명은 주식 맥락 토큰이 본문에 있을 때만 인정 (일상어 오탐 차단)
+        if name in AMBIGUOUS_NAMES and not has_ctx:
+            continue
         if name.isascii():
             # 영문/숫자 약칭(KT·SK·DB 등)은 단어경계 강제 (SKT·KTX 오탐 차단)
             pat = rf"(?<![A-Za-z0-9]){re.escape(name)}(?![A-Za-z0-9])"

dashboard-api/main.py

@@ -2084,6 +2084,38 @@ async def proxy_sector_concentration():
     return await _proxy_get(f"{SCORE_ENGINE_URL}/sector/concentration")
 
 
+# ── AI 투자조직 패널 (데이터건강·정확도·핫검증·결정안) ──────────────
+@app.get("/api/data-health")
+async def proxy_data_health():
+    """데이터 무결성 (score-engine /data-health)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/data-health")
+
+@app.get("/api/accuracy")
+async def proxy_accuracy(days: int = Query(default=90, ge=7, le=365)):
+    """추천 등급별 정확도 (score-engine /accuracy)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/accuracy?days={days}")
+
+@app.get("/api/hot-validate")
+async def proxy_hot_validate():
+    """핫종목 검증 (score-engine /hot/validate)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/hot/validate", timeout=20.0)
+
+@app.get("/api/decisions")
+async def proxy_decisions():
+    """오늘의 결정안 (score-engine /decisions)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/decisions")
+
+@app.get("/api/decisions-accuracy")
+async def proxy_decisions_accuracy(days: int = Query(default=60, ge=7, le=365)):
+    """결정안 실측 성과 (score-engine /decisions/accuracy)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/decisions/accuracy?days={days}")
+
+@app.get("/api/trade-history")
+async def proxy_trade_history(days: int = Query(default=30, ge=1, le=365)):
+    """모의매매 이력 + 손익 (score-engine /trade/history)"""
+    return await _proxy_get(f"{SCORE_ENGINE_URL}/trade/history?days={days}")
+
+
 async def _enrich_kr_names(rows):
     """rows의 kr_code/stock_code에 dart_corps.corp_name을 kr_name으로 첨부"""
     if not isinstance(rows, list) or not rows:

docs/ai_org.md

@@ -0,0 +1,68 @@
+# AI 주식 투자 회사 — 마스터 기획 (데이터 우선 재설계)
+
+> 사용자 의도: AI 에이전트들이 매일 회사처럼 스스로 운영. 사용자는 결과+승인.
+> **핵심 원칙(재정립)**: 지능·자동실행보다 **데이터 신뢰성과 정확도 검증이 먼저**.
+>   "쓰레기 입력 → 쓰레기 추천". 토대가 검증되기 전엔 위층을 신뢰하지 않는다.
+> 선택: 두뇌=하이브리드(EXAONE 1차 + Gemini 고급검증) / 승인=자동실행+사후보고 / 풀 조직.
+
+## 왜 순서를 바꿨나 (2026-06-03 발견된 토대 결함)
+- 시세: 키움 토큰 race로 stale → 복구함.
+- 일봉(stock_ohlcv): 06-02 366종목/오늘 0개. ka10005(30봉 한계)로 백필 안 됨 → **네이버 siseJson 경로 필요**.
+- TA: `/analyze/all`이 dart_corps LIMIT 500 중 163개만, `is_active=true` 필터 없음.
+- 정확도: 매일 "당일 실측 대비" 검증이 운영 중심이 아님.
+→ 이 상태로 에이전트 조직을 올리면 추천이 오염된다. 데이터·검증을 먼저 세운다.
+
+---
+
+## 로드맵 (각 Phase는 "검증 통과"가 다음 진행 조건)
+
+### Phase 0 — 데이터 신뢰 기반 (지금 시작)
+목표: 모든 핵심 데이터가 **매일 최신·전종목·무결**하고, 깨지면 **즉시 감지**.
+- [ ] OHLCV 백필 정상화: 네이버 siseJson 경로로 06-02~당일 + KOSPI 지수 채움. 키움은 보조.
+- [ ] TA 커버리지: 전 활성종목 대상 + `is_active=true` 필터.
+- [ ] **데이터 무결성 모니터**(신규): 시세/일봉/TA/점수/뉴스의 신선도·종목수·결측을 매시간 점검,
+      기준 미달이면 텔레그램 경고 1건. (= "데이터 제대로 쌓이는지" 상시 감시)
+- 검증: 모니터가 "전부 GREEN" 며칠 유지. LS 등 표본 종목 시세=일봉=TA=점수 일치.
+
+### Phase 1 — 정확도 검증 하베스트 (백본)
+목표: 모든 예측을 **매일 당일 실측과 대조**해 맞는지 수치로 안다.
+- [ ] 일일 검증 잡: 어제 추천/예측 vs 오늘 실제 등락 → 등급별 적중률·평균수익·알파·IC(스피어만) 기록.
+      (기존 verify_predictions_job / /backtest / /learn-weights 를 이 백본으로 통합·정규화)
+- [ ] **핫종목 검증팀**을 여기 편입: 키움 핫종목(ka10023 거래량급증 등) × 시스템 추천 교차 →
+      노이즈(작전·테마·일시급등) 필터 → 며칠 뒤 실측으로 "노이즈 판별이 맞았나" 정확도 추적.
+- [ ] 정확도 리포트(텔레그램/대시보드): 방식이 맞는지 매일 사용자에게 숫자로 보고.
+- 검증: 적중률·IC가 무작위(0) 대비 유의미하게 양(+)인지. 아니면 방식 수정.
+
+### Phase 2 — AI 조직 (신뢰 데이터 위에서)
+- [ ] 🔍 리서치 2명: `run_deep_analysis(model="hybrid")` 재사용 (EXAONE↔Gemini 교차검증).
+- [ ] ⚠️ 리스크 매니저: 사이징·섹터집중·레짐·분식·단기약세·일일손실 게이트 정식화.
+- [ ] 🧭 CIO(신규, Gemini): 점수+논거+리스크+핫검증 종합 → "오늘의 결정안"(매수/매도/보류+확신도+사이즈+논거).
+- 비용 가드: EXAONE 전종목 1차 → 상위 N(기본10)만 Gemini. `gemini_call_log` 일일한도 재사용.
+- 검증: 결정안을 1~2주 dry-run, Phase 1 하베스트로 실측 적중 확인.
+
+### Phase 3 — 자동실행 + 사후보고
+- 기본 **모의투자 유지**(`KIS_IS_PAPER=true`). 실계좌 전환은 별도 명시 승인.
+- 사전승인 필수("중요") 기본: ①1회 매수 ≥50만원 ②신규종목 ③일일손실 -2% 근접.
+- 그 외 가드 내 자동실행 + 텔레그램 사후보고. 기존 7가드·halt 유지.
+- 검증: 모의계좌 체결·손익이 결정안과 일치, 가드가 실제로 막는지.
+
+---
+
+## 조직도 (Phase 2에서 완성, 기존 재사용 위주)
+| 직책 | 구현 | 상태 |
+|------|------|------|
+| 📥 데이터팀 | 서비스 7종 | ✅ (Phase 0에서 신뢰성 보강) |
+| 🔍 리서치 ×2 | run_deep_analysis(hybrid) | ✅ 재사용 |
+| 💰 퀀트 | score-engine 10공식 | ✅ |
+| 🔥 핫종목 검증팀 | 키움 핫 × 추천 노이즈필터 | 🟡 Phase 1 |
+| ⚠️ 리스크 매니저 | 가드 묶음 | 🟡 Phase 2 |
+| 🧭 CIO | 종합 결정안 (Gemini) | ❌ Phase 2 |
+| 📋 리포터 / ✅ 사용자 | 텔레그램 보고·승인 | 🟡 Phase 3 |
+
+## 신규 산출물(누적)
+- `data_health` 점검 모듈 + 신선도 모니터 (Phase 0)
+- OHLCV siseJson 백필 보강 (Phase 0)
+- `prediction_accuracy` 검증 하베스트 + 리포트 (Phase 1)
+- `/hot/validate` 핫종목 노이즈필터 (Phase 1)
+- `daily_decisions` 테이블 + CIO 로직 (Phase 2)
+- 자동실행/사후보고 배선 (Phase 3)

kis-api/main.py

@@ -44,10 +44,19 @@ scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
 class KiwoomToken:
     token: str = ""
     expires_at: datetime = datetime.min
+    _lock = asyncio.Lock()
 
     async def get(self, client: httpx.AsyncClient) -> str:
         if self.token and datetime.now() < self.expires_at:
             return self.token
+        # 동시 갱신 방지: 락 안에서 재확인(double-check) — 키움은 앱키당 토큰 1개만
+        # 유효(재발급 시 이전 토큰 무효화)하므로 thundering-herd 시 무효 토큰을 붙들게 됨
+        async with self._lock:
+            if self.token and datetime.now() < self.expires_at:
+                return self.token
+            return await self._refresh(client)
+
+    async def _refresh(self, client: httpx.AsyncClient) -> str:
         resp = await client.post(
             f"{KIWOOM_BASE_URL}/oauth2/token",
             json={"grant_type": "client_credentials",

news-collector/main.py

@@ -261,6 +261,17 @@ RSS_SOURCES = [
     ("더벨",           "https://www.thebell.co.kr/free/content/xmlService.asp"),
     ("세계일보",       "https://www.segye.com/RSS/economyRss.xml"),
     ("SBS Biz",        "https://news.sbs.co.kr/news/SectionRssFeed.do?sectionId=EC"),
+    # ── 증권/마켓·산업 버티컬 확장 (2026-06 검증된 라이브 피드) ──
+    ("연합인포맥스",   "https://news.einfomax.co.kr/rss/allArticle.xml"),   # 채권·FX·마켓 전문
+    ("매일경제증권",   "https://www.mk.co.kr/rss/50200011/"),               # 증권
+    ("한국경제글로벌", "https://www.hankyung.com/feed/international"),       # 글로벌마켓
+    ("연합뉴스산업",   "https://www.yna.co.kr/rss/industry.xml"),           # 산업
+    ("데일리안",       "https://www.dailian.co.kr/rss/economy"),
+    ("ZDNet코리아",    "https://feeds.feedburner.com/zdkorea"),             # IT/테크
+    ("테크M",          "https://www.techm.kr/rss/allArticle.xml"),          # IT/테크
+    ("전자신문반도체", "https://rss.etnews.com/Section902.xml"),            # 반도체 버티컬
+    ("오토헤럴드",     "https://www.autoherald.co.kr/rss/allArticle.xml"),  # 자동차 버티컬
+    ("철강금속신문",   "https://www.snmnews.com/rss/allArticle.xml"),       # 철강/소재 버티컬
 ]
 
 def parse_rss_date(date_str: str) -> str:

score-engine/main.py

@@ -743,8 +743,9 @@ def calc_foreign_score(foreign_data: list) -> tuple[float, str]:
     return max(-100.0, min(100.0, score)), reason
 
 
-def calc_short_score(short_data: list) -> tuple[float, str]:
+def calc_short_score(short_data: list, rsi: float | None = None) -> tuple[float, str]:
-    """공매도 점수 (-100~+100), 공매도 많을수록 패널티"""
+    """공매도 점수 (-100~+100). 기본은 공매도 많을수록 패널티지만,
+    공매도 과다 + 과매도(RSI≤35)면 숏커버링 반등(스퀴즈) 기대로 가산 — 양방향."""
     if not short_data:
         return 0.0, ""
     score = 0.0
@@ -767,6 +768,12 @@ def calc_short_score(short_data: list) -> tuple[float, str]:
             elif bal_chg_pct > 5: score -= 10
             elif bal_chg_pct < -20: score += 15
             elif bal_chg_pct < -5:  score += 7
+    # 숏스퀴즈 기대: 공매도 과다(거래비중≥5%) + 과매도(RSI≤35) → 숏커버링 반등 가능
+    # → 단방향 패널티를 상쇄/가산 (공매도 많을수록·과매도 깊을수록 가산 ↑, 최대 +35)
+    if rsi is not None and weight >= 5.0 and rsi <= 35.0:
+        squeeze = min(35.0, (weight - 5.0) * 2.0 + (35.0 - rsi) * 0.8)
+        score += squeeze
+        reason = (reason or "") + f" 숏스퀴즈기대(RSI{rsi:.0f})"
     return max(-100.0, min(100.0, score)), reason
 
 
@@ -962,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
@@ -975,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} (부도위험)"
 
@@ -1030,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
@@ -1999,6 +2013,7 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict:
             SELECT stock_code,
                    SUM(inst_net) AS inst5d,
                    SUM(foreign_net) AS for5d,
+                   SUM(individual_net) AS ind5d,
                    AVG(close_price)::float AS avg_price
             FROM inst_daily_flow
             WHERE trade_date >= CURRENT_DATE - 7
@@ -2009,6 +2024,7 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict:
             SELECT stock_code,
                    SUM(inst_net) AS inst5d,
                    SUM(foreign_net) AS for5d,
+                   SUM(individual_net) AS ind5d,
                    AVG(close_price)::float AS avg_price
             FROM inst_daily_flow
             WHERE trade_date BETWEEN $1::date - 7 AND $1::date
@@ -2018,25 +2034,35 @@ async def _load_inst_flow_map(conn, as_of: date | None = None) -> dict:
 
 
 def calc_inst_flow_signal(flow: dict) -> tuple[float, str]:
-    """기관 5일 순매수가 평균거래량 대비 의미 있으면 + 가산. 외국인과 같은 방향이면 추가 가중."""
+    """기관+개인 수급 시그널 (±15). 기관 순매수 가산(외국인 동행 시 가중),
+    개인은 컨트래리언: 개인 과열매수=감점, 개인이탈+큰손매집=가점."""
     if not flow:
         return 0.0, ""
     inst5 = int(flow.get("inst5d") or 0)
     for5 = int(flow.get("for5d") or 0)
-    if inst5 == 0 and for5 == 0:
+    ind5 = int(flow.get("ind5d") or 0)
+    if inst5 == 0 and for5 == 0 and ind5 == 0:
         return 0.0, ""
-    # 기관 시그널: tanh 스케일 (포화)
     import math
+    # 기관 시그널: tanh 스케일 (포화) + 외국인 동행 가중
     inst_score = math.tanh(inst5 / 5_000_000) * 6.0
-    # 외국인 같은 방향이면 +4, 반대면 -2
     if inst5 * for5 > 0:
         inst_score += 4.0 if abs(for5) > 1_000_000 else 2.0
     elif inst5 * for5 < 0:
         inst_score -= 2.0
-    sig = max(-10.0, min(10.0, inst_score))
+    # 개인 컨트래리언: 순매수→감점(과열), 순매도→가점 (소폭)
-    direction = "매수" if inst5 > 0 else "매도"
+    ind_score = -math.tanh(ind5 / 5_000_000) * 3.0
-    reason = f"기관 5d {direction}{abs(inst5)//10000:,}만주"
+    smart = inst5 + for5
-    return sig, reason
+    if ind5 > 0 and smart < 0:    ind_score -= 2.0   # 개인이 받고 큰손 던짐 = 분산
+    elif ind5 < 0 and smart > 0:  ind_score += 2.0   # 큰손 매집·개인 이탈 = 매집
+    ind_score = max(-5.0, min(5.0, ind_score))
+    sig = max(-15.0, min(15.0, inst_score + ind_score))
+    parts = []
+    if inst5 != 0:
+        parts.append(f"기관5d {'매수' if inst5>0 else '매도'}{abs(inst5)//10000:,}만주")
+    if abs(ind_score) >= 1.5:
+        parts.append(f"개인{'과열매수' if ind5>0 else '이탈'}")
+    return sig, " ".join(parts)
 
 
 def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float, str]:
@@ -2061,8 +2087,10 @@ def calc_valuation_percentile(per_history: list, cur_per: float) -> tuple[float,
     return 0.0, ""
 
 
-async def calculate_daily_scores(as_of: date | None = None):
+async def calculate_daily_scores(as_of: date | None = None, notify: bool = False):
     """일간 점수 계산. as_of=None이면 today (운영 모드), as_of=date이면 그 시점 기준 (백필 모드).
+    notify=True일 때만 텔레그램 일간리포트/신규강력매수 발신 (16:30 정기 1회만 True).
+    그 외 호출(통합 워크플로우 30분마다·매시간 스코어·18:30 등)은 silent로 점수만 갱신.
     백필 모드는 look-ahead bias 차단:
       - 사후 학습 캐시(reliability/source_credibility) 미적용
       - weight_config는 config_date <= as_of 필터
@@ -2319,12 +2347,13 @@ async def calculate_daily_scores(as_of: date | None = None):
                 except: pass
 
             # 기술적 점수 (stock_technical 테이블 - Redis TTL/DB 불일치 방지)
-            technical_score = 0.0
+            technical_score = 0.0; rsi_val = None
             try:
                 ta_row = await conn.fetchrow(
-                    "SELECT tech_score FROM stock_technical WHERE stock_code=$1", stock)
+                    "SELECT tech_score, rsi FROM stock_technical WHERE stock_code=$1", stock)
                 if ta_row:
                     technical_score = float(ta_row["tech_score"] or 0)
+                    rsi_val = float(ta_row["rsi"]) if ta_row["rsi"] is not None else None
             except: pass
 
             # 외국인 수급 점수 (Redis foreign:{code})
@@ -2343,7 +2372,7 @@ async def calculate_daily_scores(as_of: date | None = None):
                 s_raw = await redis_cl.get(f"short:{stock}")
                 if s_raw:
                     s_data = json.loads(s_raw)
-                    short_score, short_reason = calc_short_score(s_data)
+                    short_score, short_reason = calc_short_score(s_data, rsi=rsi_val)
                     short_weight_val = s_data[0].get("trade_weight", 0) if s_data else 0
             except: pass
 
@@ -2818,7 +2847,7 @@ async def calculate_daily_scores(as_of: date | None = None):
     logger.info("scoring.done", scored=scored, recommended=len(rec_rows))
 
     # 신규 강력매수 즉시 알림 (어제는 강력매수 아니었는데 오늘 새로 등장한 종목)
-    if not backfill_mode and strong_buy:
+    if notify and not backfill_mode and strong_buy:
         try:
             async with pg_pool.acquire() as nc:
                 prev_codes = {r["stock_code"] for r in await nc.fetch("""
@@ -2853,8 +2882,8 @@ async def calculate_daily_scores(as_of: date | None = None):
         except Exception as e:
             logger.warning("new_buy.err", error=str(e))
 
-    # 텔레그램 알림
+    # 텔레그램 알림 (notify=True 정기 1회만 — 매 호출 발신 방지)
-    if strong_buy or strong_sell:
+    if notify and (strong_buy or strong_sell):
         lines = [f"<b>📊 Trading AI 일간 리포트 ({today})</b>\n"]
         if strong_buy:
             lines.append("🟢 <b>강력매수 추천 (버핏 가치필터 통과)</b>")
@@ -2876,9 +2905,17 @@ async def calculate_daily_scores(as_of: date | None = None):
 
 # ── 정기 브리핑 ───────────────────────────────────────────
 
+_last_briefing_sent: datetime | None = None   # 중복/폭주 차단 쓰로틀
+
 async def send_briefing():
-    """정기 시황 브리핑 — 종목당 1카드(매매가/포지션/재무/근거 통합)"""
+    """정기 시황 브리핑 — 종목당 1카드(매매가/포지션/재무/근거 통합).
+    30분 내 재호출은 무시 — n8n 중복/재시도 폭주로 같은 브리핑이 여러 건 발송되는 것 방지."""
+    global _last_briefing_sent
     now = datetime.now()
+    if _last_briefing_sent and (now - _last_briefing_sent).total_seconds() < 1800:
+        logger.info("briefing.throttled", since=(now - _last_briefing_sent).total_seconds())
+        return
+    _last_briefing_sent = now
     ta_redis = aioredis.Redis(
         host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=5, decode_responses=True)
 
@@ -3018,6 +3055,17 @@ async def send_briefing():
 app = FastAPI(title="종목 점수 엔진")
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
 
+async def _daily_score_notify():
+    """16:30 정기 일간리포트 — 코루틴 함수로 등록해야 APScheduler가 await함
+    (lambda로 감싸면 coroutine never awaited 버그)."""
+    await calculate_daily_scores(notify=True)
+
+async def _learn_weights_job():
+    await learn_weights(days=90, segment="all")
+
+async def _learn_pricing_job():
+    await learn_pricing(days=90, segment="all", target="return_7d", n_folds=5)
+
 @app.on_event("startup")
 async def startup():
     global pg_pool, redis_cl
@@ -3027,7 +3075,7 @@ async def startup():
     redis_cl = aioredis.Redis(
         host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=3, decode_responses=True)
     await init_db()
-    scheduler.add_job(calculate_daily_scores, "cron",
+    scheduler.add_job(_daily_score_notify, "cron",
                       day_of_week="mon-fri", hour=16, minute=30,
                       id="daily_score", replace_existing=True)
     # 텔레그램 정기 알림 하루 2회로 축소 (사용자 요청 2026-06-02):
@@ -3038,6 +3086,26 @@ async def startup():
     scheduler.add_job(cleanup_old_data, "cron",
                       hour=4, minute=0,
                       id="cleanup", replace_existing=True)
+    # 데이터 무결성 모니터: 평일 10~17시 매시간 (RED면 텔레그램 경고, 3h 쓰로틀)
+    scheduler.add_job(data_health_monitor_job, "cron",
+                      day_of_week="mon-fri", hour="10-17", minute=5,
+                      id="data_health", replace_existing=True)
+    # 정확도 검증 리포트: 매주 일요일 10시 (방식이 실측 대비 맞는지)
+    scheduler.add_job(accuracy_report_job, "cron",
+                      day_of_week="sun", hour=10, minute=0,
+                      id="accuracy_report", replace_existing=True)
+    # 핫종목 검증팀: 평일 09:35 검증통과 핫종목 1회 보고 (dry-run)
+    scheduler.add_job(hot_validate_report_job, "cron",
+                      day_of_week="mon-fri", hour=9, minute=35,
+                      id="hot_validate", replace_existing=True)
+    # CIO 오늘의 결정안: 평일 09:20 생성+보고 (dry-run, 자동실행 OFF)
+    scheduler.add_job(cio_decisions_job, "cron",
+                      day_of_week="mon-fri", hour=9, minute=20,
+                      id="cio_decisions", replace_existing=True)
+    # 결정안 성과 채점: 매일 18:10 (성과가격 갱신 18:00 이후)
+    scheduler.add_job(verify_decisions_job, "cron",
+                      hour=18, minute=10,
+                      id="verify_decisions", replace_existing=True)
     # 성과 추적: 매일 18시 가격 업데이트
     scheduler.add_job(update_performance_prices, "cron",
                       day_of_week="mon-fri", hour=18, minute=0,
@@ -3050,11 +3118,10 @@ async def startup():
     # 04:00 — 공식 가중치 학습 (90일 백테스트)
     # 05:00 — 예상가 모델 학습 (선형회귀 + RF + XGBoost)
     # 두 함수 모두 표본 부족 시 graceful (return early) — 데이터 누적되면 자동 활성화
-    scheduler.add_job(lambda: learn_weights(days=90, segment="all"), "cron",
+    scheduler.add_job(_learn_weights_job, "cron",
                       day_of_week="sun", hour=4, minute=0,
                       id="learn_weights", replace_existing=True)
-    scheduler.add_job(lambda: learn_pricing(days=90, segment="all",
+    scheduler.add_job(_learn_pricing_job, "cron",
-                      target="return_7d", n_folds=5), "cron",
                       day_of_week="sun", hour=5, minute=0,
                       id="learn_pricing", replace_existing=True)
     # AI 심층분석 — 비용 폭주로 자동 호출 임시 비활성화 (2026-05-29)
@@ -3144,13 +3211,425 @@ async def cleanup_old_data():
                 prices=deleted_prices, recs=deleted_recs,
                 news=deleted_news, signals=deleted_signals)
 
+# ── 데이터 무결성 모니터 ("데이터 제대로 쌓이는지" 상시 감시) ──────────
+_last_health_alert: dict = {}   # 동일 결함 반복경고 방지 (3시간 쓰로틀)
+
+async def check_data_health() -> dict:
+    """핵심 데이터 신선도·커버리지 점검 → 항목별 GREEN/YELLOW/RED."""
+    checks = []
+    def add(name, status, detail): checks.append({"name": name, "status": status, "detail": detail})
+    async with pg_pool.acquire() as c:
+        n = await c.fetchval("SELECT COUNT(DISTINCT stock_code) FROM stock_prices WHERE collected_at::date=CURRENT_DATE") or 0
+        add("시세(stock_prices)", "GREEN" if n>=2000 else "YELLOW" if n>=500 else "RED", f"오늘 {n}종목")
+
+        last_dt = await c.fetchval("SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code<>'KOSPI'")
+        ocnt = await c.fetchval("SELECT COUNT(DISTINCT stock_code) FROM stock_ohlcv WHERE dt=$1", last_dt) if last_dt else 0
+        oage = (date.today()-last_dt).days if last_dt else 999
+        add("일봉(stock_ohlcv)", "GREEN" if (ocnt>=2000 and oage<=4) else "YELLOW" if ocnt>=1000 else "RED", f"{last_dt} {ocnt}종목")
+
+        n = await c.fetchval("SELECT COUNT(*) FROM stock_technical WHERE analyzed_at::date=CURRENT_DATE") or 0
+        add("기술분석(TA)", "GREEN" if n>=1500 else "YELLOW" if n>=300 else "RED", f"오늘 {n}건")
+
+        n = await c.fetchval("SELECT COUNT(*) FROM stock_scores WHERE score_date=CURRENT_DATE") or 0
+        add("점수(stock_scores)", "GREEN" if n>=1000 else "YELLOW" if n>=100 else "RED", f"오늘 {n}종목")
+
+        n = await c.fetchval("SELECT COUNT(*) FROM news_analysis WHERE analyzed_at > now()-interval '24 hours'") or 0
+        add("뉴스(24h)", "GREEN" if n>=100 else "YELLOW" if n>=10 else "RED", f"{n}건")
+
+        kd = await c.fetchval("SELECT MAX(dt) FROM stock_ohlcv WHERE stock_code='KOSPI'")
+        kage = (date.today()-kd).days if kd else 999
+        add("KOSPI지수 일봉", "GREEN" if kage<=4 else "RED", f"{kd}")
+
+        # OHLCV 이상치: 한국 일일 가격제한 ±30% 초과는 정의상 불량(스케일버그·권리락 미조정).
+        # 35% 마진으로 정상 상한가(±30%)는 제외. 데이터품질 경고라 YELLOW, 급증 시만 RED.
+        bad = await c.fetchval("""
+            WITH px AS (
+              SELECT close_price, LAG(close_price) OVER (PARTITION BY stock_code ORDER BY dt) AS prev_close
+              FROM stock_ohlcv WHERE dt > CURRENT_DATE - 7 AND stock_code<>'KOSPI'
+            )
+            SELECT COUNT(*) FROM px
+            WHERE close_price > 0 AND prev_close > 0 AND abs(close_price::float/prev_close - 1) > 0.35
+        """) or 0
+        add("OHLCV 이상치(±30%위반)", "GREEN" if bad==0 else "YELLOW" if bad<=50 else "RED", f"최근7일 {bad}행")
+    worst = "RED" if any(x["status"]=="RED" for x in checks) else ("YELLOW" if any(x["status"]=="YELLOW" for x in checks) else "GREEN")
+    return {"overall": worst, "checks": checks, "checked_at": datetime.now().isoformat()}
+
+async def data_health_monitor_job():
+    """평일 장중~마감후 점검. RED 있으면 텔레그램 경고 1건 (3시간 쓰로틀)."""
+    try:
+        h = await check_data_health()
+    except Exception as e:
+        logger.error("data_health.err", error=str(e)); return
+    reds = [x for x in h["checks"] if x["status"]=="RED"]
+    if not reds: return
+    key = ",".join(sorted(x["name"] for x in reds))
+    now = datetime.now()
+    last = _last_health_alert.get(key)
+    if last and (now-last).total_seconds() < 10800: return
+    _last_health_alert[key] = now
+    lines = ["🚨 <b>데이터 무결성 경고</b>"] + [f"❌ {x['name']}: {x['detail']}" for x in reds]
+    lines.append("\n수집/분석 파이프라인 점검 필요")
+    await send_telegram("\n".join(lines))
+    logger.info("data_health.alert", reds=len(reds))
+
+@app.get("/data-health")
+async def data_health_endpoint():
+    return await check_data_health()
+
+# ── 정확도 검증 하베스트 ("방식이 맞는지" 실측 대비) ──────────────────
+async def compute_accuracy(days: int = 90) -> dict:
+    """추천 등급별 사후 정확도. recommendation_performance(실측 7d/30d 수익률·알파) 집계.
+    동전주 폭등·불량 가격데이터 이상치가 평균(mean)을 왜곡하므로 중앙값(median)으로 집계.
+    매수계열 알파>0 & 매도계열 알파<0 이면 방식 유효."""
+    async with pg_pool.acquire() as conn:
+        grades = await conn.fetch("""
+            SELECT recommendation rec, COUNT(*) n,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY return_7d) ret7,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_7d) a7,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY return_30d) ret30,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_30d) a30,
+                   AVG(CASE WHEN return_7d>0 THEN 1.0 ELSE 0 END) up7
+            FROM recommendation_performance
+            WHERE return_7d IS NOT NULL AND rec_date >= CURRENT_DATE - ($1::int)
+            GROUP BY recommendation
+        """, days)
+        pooled = await conn.fetchrow("""
+            SELECT percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_7d)
+                     FILTER (WHERE recommendation IN ('강력매수','매수관심')) buy_a7,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_7d)
+                     FILTER (WHERE recommendation IN ('강력매도','매도관심')) sell_a7,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_30d)
+                     FILTER (WHERE recommendation IN ('강력매수','매수관심')) buy_a30,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_30d)
+                     FILTER (WHERE recommendation IN ('강력매도','매도관심')) sell_a30,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_30d)
+                     FILTER (WHERE recommendation='강력매수') sb_a30,
+                   percentile_cont(0.5) WITHIN GROUP (ORDER BY alpha_30d)
+                     FILTER (WHERE recommendation='강력매도') ss_a30
+            FROM recommendation_performance
+            WHERE return_7d IS NOT NULL AND rec_date >= CURRENT_DATE - ($1::int)
+        """, days)
+    order = {"강력매수": 0, "매수관심": 1, "관망": 2, "매도관심": 3, "강력매도": 4}
+    rows = sorted([dict(g) for g in grades], key=lambda x: order.get(x["rec"], 9))
+    def rnd(v): return round(v, 2) if v is not None else None
+    buy_a, sell_a = rnd(pooled["buy_a7"]), rnd(pooled["sell_a7"])
+    buy_a30, sell_a30 = rnd(pooled["buy_a30"]), rnd(pooled["sell_a30"])
+    sb30, ss30 = rnd(pooled["sb_a30"]), rnd(pooled["ss_a30"])
+    spread30 = round(sb30 - ss30, 2) if (sb30 is not None and ss30 is not None) else None
+    # 판정은 30일 기준(가치투자 시계열·이상치 robust median). 7일은 단기 노이즈라 참고용.
+    if buy_a30 is None or sell_a30 is None or spread30 is None:
+        verdict = "30일 표본 부족 — 7일 참고"
+    elif buy_a30 > 0 and sell_a30 < 0 and spread30 >= 5:
+        verdict = "양호 (30일 매수>0·매도<0·스프레드≥5%p)"
+    elif spread30 >= 5 and sell_a30 < 0:
+        verdict = "부분유효 (강력매수 변별 양호, 매수계열 알파 음전)"
+    else:
+        verdict = "교정필요 (30일 변별력 부족)"
+    return {
+        "days": days,
+        "agg": "median",
+        "basis": "30d",
+        "grades": [{"rec": r["rec"], "n": r["n"],
+                    "ret7": round(r["ret7"] or 0, 2), "alpha7": round(r["a7"] or 0, 2),
+                    "ret30": round(r["ret30"], 2) if r["ret30"] is not None else None,
+                    "alpha30": round(r["a30"], 2) if r["a30"] is not None else None,
+                    "up7_pct": round(100 * (r["up7"] or 0))} for r in rows],
+        "buy_alpha7": buy_a, "sell_alpha7": sell_a,
+        "buy_alpha30": buy_a30, "sell_alpha30": sell_a30, "spread30": spread30,
+        "verdict": verdict,
+    }
+
+@app.get("/accuracy")
+async def accuracy_endpoint(days: int = Query(default=90, ge=7, le=365)):
+    return await compute_accuracy(days)
+
+async def accuracy_report_job():
+    """주간 정확도 리포트 — 방식이 실측 대비 맞는지 텔레그램 보고."""
+    try:
+        a = await compute_accuracy(90)
+    except Exception as e:
+        logger.error("accuracy.err", error=str(e)); return
+    lines = ["📈 <b>추천 정확도 리포트 (최근90일·30일 알파·중앙값)</b>",
+             f"판정: <b>{a['verdict']}</b>",
+             f"매수계열 알파 {a['buy_alpha30']} / 매도계열 알파 {a['sell_alpha30']} / 강력매수−강력매도 스프레드 {a['spread30']}%p\n"]
+    for g in a["grades"]:
+        lines.append(f"{g['rec']}: n{g['n']} 30일수익{g['ret30']}% 알파{g['alpha30']}% (7일알파{g['alpha7']}%)")
+    await send_telegram("\n".join(lines))
+    logger.info("accuracy.report.sent", verdict=a["verdict"])
+
+# ── 🔥 핫종목 검증팀 (키움 핫 × 가치/품질 노이즈필터, dry-run) ──────────
+async def compute_hot_validate(top: int = 30) -> dict:
+    """키움 거래량급증(ka10023) × 가치/품질 필터 → 핫한데 가치없는 노이즈
+    (ETF·파생·작전·적자·분식·음수점수) 제거 → '검증통과'만 추림."""
+    hot = []
+    try:
+        async with httpx.AsyncClient() as cli:
+            r = await cli.get("http://kis-api:8585/volume-surge", timeout=15)
+            hot = ((r.json() or {}).get("data") or [])[:top]
+    except Exception as e:
+        logger.warning("hot_validate.fetch_err", error=str(e))
+        return {"error": "키움 핫종목 조회 실패", "detail": str(e)}
+    codes = [h["code"] for h in hot if h.get("code")]
+    if not codes:
+        return {"hot_count": 0, "passed": [], "noise": [], "watch": []}
+    async with pg_pool.acquire() as conn:
+        rows = await conn.fetch("""
+            SELECT h.code, c.corp_name, c.is_active,
+                   sc.total_score, sc.recommendation, sc.beneish_score,
+                   pr.market_cap, fin.operating_profit
+            FROM unnest($1::text[]) AS h(code)
+            LEFT JOIN dart_corps c ON c.stock_code = h.code
+            LEFT JOIN LATERAL (SELECT total_score, recommendation, beneish_score
+                               FROM stock_scores WHERE stock_code=h.code
+                               ORDER BY score_date DESC LIMIT 1) sc ON true
+            LEFT JOIN LATERAL (SELECT market_cap FROM stock_prices WHERE stock_code=h.code
+                               ORDER BY collected_at DESC LIMIT 1) pr ON true
+            LEFT JOIN LATERAL (SELECT operating_profit FROM dart_financials
+                               WHERE stock_code=h.code AND reprt_code='11011'
+                               ORDER BY bsns_year DESC LIMIT 1) fin ON true
+        """, codes)
+    by = {r["code"]: r for r in rows}
+    passed, noise, watch = [], [], []
+    for h in hot:
+        code = h.get("code"); r = by.get(code)
+        it = {"code": code, "name": (r["corp_name"] if r and r["corp_name"] else h.get("name", "")),
+              "change_pct": h.get("change_pct"), "surge_rate": h.get("surge_rate"),
+              "score": round(r["total_score"], 1) if r and r["total_score"] is not None else None,
+              "reco": r["recommendation"] if r else None}
+        if not r or not r["is_active"]:
+            it["verdict"], it["reason"] = "노이즈", "ETF/파생/비상장(가치판단 불가)"; noise.append(it); continue
+        mc, op, sc, bn = r["market_cap"] or 0, r["operating_profit"], r["total_score"], r["beneish_score"]
+        if mc and mc < 10_000_000_000:
+            it["verdict"], it["reason"] = "노이즈", f"초소형 시총{mc//100000000}억(작전위험)"; noise.append(it); continue
+        if op is not None and op <= 0:
+            it["verdict"], it["reason"] = "노이즈", "영업적자"; noise.append(it); continue
+        if bn is not None and bn >= 50:
+            it["verdict"], it["reason"] = "노이즈", f"분식의심(Beneish {bn:.0f})"; noise.append(it); continue
+        if sc is None:
+            it["verdict"], it["reason"] = "관찰", "점수 미산출"; watch.append(it); continue
+        if sc < 0:
+            it["verdict"], it["reason"] = "노이즈", f"투자부적합(점수{sc:.0f})"; noise.append(it); continue
+        if sc >= 40 and r["recommendation"] in ("강력매수", "매수관심"):
+            it["verdict"], it["reason"] = "검증통과", f"가치+모멘텀(점수{sc:.0f}·{r['recommendation']})"; passed.append(it); continue
+        it["verdict"], it["reason"] = "관찰", f"중립(점수{sc:.0f})"; watch.append(it)
+    return {"hot_count": len(hot), "passed_count": len(passed),
+            "noise_count": len(noise), "watch_count": len(watch),
+            "noise_filtered_pct": round(100 * len(noise) / len(hot)) if hot else 0,
+            "passed": passed, "watch": watch, "noise": noise}
+
+@app.get("/hot/validate")
+async def hot_validate_endpoint(top: int = Query(default=30, ge=5, le=50)):
+    return await compute_hot_validate(top)
+
+async def hot_validate_report_job():
+    """평일 09:35 — 검증통과 핫종목만 텔레그램 보고 (dry-run, 1일 1회)."""
+    try:
+        v = await compute_hot_validate(30)
+    except Exception as e:
+        logger.error("hot_validate.err", error=str(e)); return
+    if v.get("error") or not v.get("passed"):
+        return
+    lines = [f"🔥 <b>검증통과 핫종목</b> (키움 핫 {v['hot_count']}개 중 노이즈 {v['noise_filtered_pct']}% 제거)"]
+    for p in v["passed"][:10]:
+        lines.append(f"✅ {p['name']}({p['code']}) {p['change_pct']:+.1f}% — {p['reason']}")
+    await send_telegram("\n".join(lines))
+    logger.info("hot_validate.report.sent", passed=v["passed_count"])
+
+# ── 🧭 CIO 종합 에이전트 (오늘의 결정안, dry-run) ──────────────────
+def _conviction(score: float, votes: int) -> int:
+    if score >= 70 and votes >= 3: return 5
+    if score >= 55 and votes >= 2: return 4
+    if score >= 45: return 3
+    if score >= 35: return 2
+    return 1
+
+async def run_cio_decisions(top: int = 8) -> dict:
+    """점수+보팅+리스크+핫검증 종합 → 오늘의 매수/매도 결정안 (dry-run, 승인 전).
+    매수=추천 상위 후보, 매도=보유종목(user_portfolio) 등급악화·손절. entry_price 기록(사후추적)."""
+    today = date.today()
+    async with pg_pool.acquire() as conn:
+        await conn.execute("""
+            CREATE TABLE IF NOT EXISTS daily_decisions (
+                id SERIAL PRIMARY KEY,
+                decision_date DATE NOT NULL,
+                stock_code VARCHAR(10) NOT NULL,
+                stock_name VARCHAR(100) DEFAULT '',
+                action VARCHAR(20) NOT NULL,
+                conviction INTEGER DEFAULT 0,
+                size_pct FLOAT DEFAULT 0,
+                total_score FLOAT,
+                thesis TEXT DEFAULT '',
+                risk_notes TEXT DEFAULT '',
+                status VARCHAR(20) DEFAULT 'proposed',
+                created_at TIMESTAMP DEFAULT NOW(),
+                UNIQUE(decision_date, stock_code)
+            )
+        """)
+        for col in ("entry_price BIGINT DEFAULT 0", "return_7d FLOAT",
+                    "kospi_return_7d FLOAT", "alpha_7d FLOAT", "outcome VARCHAR(10)"):
+            await conn.execute(f"ALTER TABLE daily_decisions ADD COLUMN IF NOT EXISTS {col}")
+
+        regime_label, _ = await calc_market_regime(conn)
+        hotv = await compute_hot_validate(30)
+        hot_pass = {p["code"] for p in (hotv.get("passed") or [])} if not hotv.get("error") else set()
+
+        async def _cur_price(code):
+            return await conn.fetchval(
+                "SELECT price FROM stock_prices WHERE stock_code=$1 ORDER BY collected_at DESC LIMIT 1", code)
+
+        saved = []
+        # ── 매수 결정안 (추천 상위 후보) ──
+        cands = await conn.fetch("""
+            SELECT s.stock_code, COALESCE(d.corp_name, s.stock_code) name,
+                   s.total_score, s.recommendation, s.buy_votes,
+                   s.position_size_pct, s.top_reasons, s.beneish_score, s.earnings_quality
+            FROM stock_scores s JOIN dart_corps d ON d.stock_code = s.stock_code
+            WHERE s.score_date = $1 AND d.is_active = true
+              AND s.recommendation IN ('강력매수', '매수관심') AND s.buy_votes >= 1
+            ORDER BY s.total_score DESC LIMIT $2
+        """, today, top)
+        for c in cands:
+            conv = _conviction(c["total_score"] or 0, c["buy_votes"] or 0)
+            if c["stock_code"] in hot_pass: conv = min(5, conv + 1)
+            size = round(c["position_size_pct"] or (conv * 2.0), 1)
+            if regime_label == "약세": size = round(size * 0.5, 1)
+            risk = []
+            if regime_label == "약세": risk.append("시장 약세(사이즈↓)")
+            if (c["beneish_score"] or 0) >= 50: risk.append("분식의심")
+            if c["earnings_quality"] is not None and c["earnings_quality"] < 0: risk.append("이익품질 낮음")
+            if c["stock_code"] in hot_pass: risk.append("핫종목 검증통과")
+            ep = await _cur_price(c["stock_code"]) or 0
+            await conn.execute("""
+                INSERT INTO daily_decisions
+                  (decision_date,stock_code,stock_name,action,conviction,size_pct,
+                   total_score,thesis,risk_notes,entry_price,status)
+                VALUES ($1,$2,$3,'매수',$4,$5,$6,$7,$8,$9,'proposed')
+                ON CONFLICT (decision_date,stock_code) DO UPDATE SET
+                  action='매수',conviction=$4,size_pct=$5,total_score=$6,
+                  thesis=$7,risk_notes=$8,entry_price=$9
+            """, today, c["stock_code"], c["name"], conv, size,
+                c["total_score"], (c["top_reasons"] or "")[:300], " · ".join(risk), int(ep))
+            saved.append({"code": c["stock_code"], "name": c["name"], "action": "매수",
+                          "conviction": conv, "size_pct": size,
+                          "score": round(c["total_score"], 1) if c["total_score"] is not None else None,
+                          "risk": " · ".join(risk)})
+        # ── 매도 결정안 (보유종목 등급악화/손절) ──
+        holds = await conn.fetch("""
+            SELECT p.stock_code, p.stock_name, p.buy_price,
+                   s.total_score, s.recommendation, s.sell_votes, s.top_reasons
+            FROM user_portfolio p
+            LEFT JOIN stock_scores s ON s.stock_code=p.stock_code AND s.score_date=$1
+            WHERE p.active = true
+        """, today)
+        for h in holds:
+            cp = await _cur_price(h["stock_code"]) or 0
+            loss = ((cp - h["buy_price"]) / h["buy_price"] * 100) if (h["buy_price"] and cp) else 0
+            reco = h["recommendation"]; reasons = []; sell = False
+            if reco in ("강력매도", "매도관심"): sell = True; reasons.append(f"등급 {reco}")
+            if loss <= -8: sell = True; reasons.append(f"손절({loss:.0f}%)")
+            if (h["sell_votes"] or 0) >= 3: sell = True; reasons.append(f"매도보팅{h['sell_votes']}")
+            if not sell: continue
+            conv = 5 if (reco == "강력매도" or loss <= -12) else 3
+            await conn.execute("""
+                INSERT INTO daily_decisions
+                  (decision_date,stock_code,stock_name,action,conviction,size_pct,
+                   total_score,thesis,risk_notes,entry_price,status)
+                VALUES ($1,$2,$3,'매도',$4,0,$5,$6,$7,$8,'proposed')
+                ON CONFLICT (decision_date,stock_code) DO UPDATE SET
+                  action='매도',conviction=$4,total_score=$5,thesis=$6,risk_notes=$7,entry_price=$8
+            """, today, h["stock_code"], h["stock_name"], conv,
+                h["total_score"], (h["top_reasons"] or "")[:200], " · ".join(reasons), int(cp))
+            saved.append({"code": h["stock_code"], "name": h["stock_name"], "action": "매도",
+                          "conviction": conv, "size_pct": 0,
+                          "score": round(h["total_score"], 1) if h["total_score"] is not None else None,
+                          "risk": " · ".join(reasons)})
+    return {"decision_date": str(today), "regime": regime_label, "count": len(saved),
+            "buy": sum(1 for x in saved if x["action"] == "매수"),
+            "sell": sum(1 for x in saved if x["action"] == "매도"),
+            "decisions": saved}
+
+@app.post("/decisions/generate")
+async def decisions_generate(top: int = Query(default=8, ge=1, le=20)):
+    return await run_cio_decisions(top)
+
+@app.get("/decisions")
+async def decisions_get():
+    async with pg_pool.acquire() as conn:
+        rows = await conn.fetch("""
+            SELECT stock_code, stock_name, action, conviction, size_pct, total_score,
+                   thesis, risk_notes, status
+            FROM daily_decisions WHERE decision_date = CURRENT_DATE
+            ORDER BY conviction DESC, total_score DESC
+        """)
+    return [dict(r) for r in rows]
+
+async def verify_decisions_job():
+    """결정안 7일 성과 채점 (return/alpha/정답여부) — '회사 결정이 실제 맞았나'. 매일 18:10."""
+    scored = 0
+    async with pg_pool.acquire() as conn:
+        rows = await conn.fetch("""
+            SELECT id, stock_code, action, entry_price, decision_date
+            FROM daily_decisions
+            WHERE return_7d IS NULL AND entry_price > 0
+              AND decision_date <= CURRENT_DATE - 7 AND decision_date >= CURRENT_DATE - 60
+        """)
+        for r in rows:
+            target = r["decision_date"] + timedelta(days=7)
+            if target > date.today(): continue
+            price = await _close_near(conn, r["stock_code"], target)
+            if price is None: continue
+            ret = (price - r["entry_price"]) / r["entry_price"] * 100
+            kret = await _kospi_return_between(conn, r["decision_date"], target)
+            alpha = (ret - kret) if kret is not None else None
+            outcome = None
+            if alpha is not None:
+                outcome = ("정답" if alpha > 0 else "오답") if r["action"] == "매수" \
+                          else ("정답" if alpha < 0 else "오답")
+            await conn.execute("""UPDATE daily_decisions
+                SET return_7d=$1, kospi_return_7d=$2, alpha_7d=$3, outcome=$4 WHERE id=$5""",
+                ret, kret, alpha, outcome, r["id"])
+            scored += 1
+    logger.info("verify_decisions.done", scored=scored)
+
+@app.get("/decisions/accuracy")
+async def decisions_accuracy(days: int = Query(default=60, ge=7, le=365)):
+    """CIO 결정안의 실측 성과 — 자동실행 게이트."""
+    async with pg_pool.acquire() as conn:
+        rows = await conn.fetch("""
+            SELECT action, COUNT(*) n, AVG(return_7d) ret, AVG(alpha_7d) alpha,
+                   AVG(CASE WHEN outcome='정답' THEN 1.0 ELSE 0 END) hit
+            FROM daily_decisions
+            WHERE return_7d IS NOT NULL AND decision_date >= CURRENT_DATE - ($1::int)
+            GROUP BY action
+        """, days)
+    return {"days": days, "by_action": [
+        {"action": r["action"], "n": r["n"],
+         "avg_return7": round(r["ret"] or 0, 2), "avg_alpha7": round(r["alpha"] or 0, 2),
+         "hit_rate": round(100 * (r["hit"] or 0))} for r in rows]}
+
+async def cio_decisions_job():
+    """평일 09:20 — CIO 오늘의 결정안 생성 + 텔레그램 보고 (dry-run, 자동실행 OFF)."""
+    try:
+        d = await run_cio_decisions(8)
+    except Exception as e:
+        logger.error("cio.err", error=str(e)); return
+    if not d.get("decisions"): return
+    lines = [f"🧭 <b>오늘의 결정안</b> ({d['decision_date']}, 시장:{d['regime']}) — dry-run"]
+    for x in d["decisions"]:
+        lines.append(f"{'★'*x['conviction']} <b>{x['name']}</b>({x['code']}) {x['action']} "
+                     f"비중{x['size_pct']}% (점수{x['score']})"
+                     + (f"\n  ⚠️{x['risk']}" if x['risk'] else ""))
+    lines.append("\n(자동실행 OFF — 검증 단계. 정확도 양전환 시 실행 연결)")
+    await send_telegram("\n".join(lines))
+    logger.info("cio.report.sent", count=d["count"])
+
 @app.get("/health")
 async def health():
     return {"status": "ok"}
 
 @app.post("/score/calculate")
-async def manual_calc():
+async def manual_calc(notify: bool = Query(default=False)):
-    n = await calculate_daily_scores()
+    n = await calculate_daily_scores(notify=notify)
     return {"status": "done", "scored": n}
 
 
@@ -4038,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 []
+    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):
-        tr_end = i * block
+            te = rows[i * block:min(n, (i + 1) * block)]
-        te_end = min(n, (i + 1) * block)
+            if not te:
-        tr = rows[:tr_end]; te = rows[tr_end:te_end]
+                continue
-        if len(tr) < 5 or len(te) < 3: 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):
+        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
 
@@ -4103,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 = []
@@ -4190,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
@@ -5475,6 +5986,7 @@ async def ensure_trading_tables():
 # 자동매매 안전 설정 — 보수적 기본값
 TRADE_SETTINGS = {
     "enabled":            True,     # 자동매매 ON/OFF 토글
+    "auto_execute":       True,     # 제안 즉시 자동 체결(모의). False면 텔레그램 버튼 승인 대기
     "max_position_pct":   10.0,     # 종목당 자본 ≤ 10%
     "daily_loss_limit_pct": -3.0,   # 일일 손실 -3% 도달 시 매매 중단
     "max_orders_per_day": 10,
@@ -5798,10 +6310,8 @@ async def trade_sell_scan():
     return {"checked": len(codes), "proposed": len(results), "results": results}
 
 
-@app.post("/trade/confirm/{order_id}")
+async def _fill_order(conn, order_id: int, auto: bool = False) -> dict:
-async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 override)
+    """주문 체결(모의=DB 기록, 실제 브로커 전송 없음). auto=True면 자동매매 체결."""
-    """매수/매도 confirm 통합 처리."""
-    async with pg_pool.acquire() as conn:
     order = await conn.fetchrow(
         "SELECT * FROM trading_orders WHERE id=$1 AND status='pending'", order_id)
     if not order:
@@ -5810,17 +6320,17 @@ async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 overri
         await conn.execute(
             "UPDATE trading_orders SET status='expired' WHERE id=$1", order_id)
         return {"error": "주문 만료"}
+    tag = "자동체결" if auto else "체결"
     if order["side"] == "buy":
         await conn.execute("""
             INSERT INTO user_portfolio (stock_code, stock_name, buy_price, qty, memo)
             VALUES ($1, $2, $3, $4, $5)
         """, order["stock_code"], order["stock_name"], order["price"],
             order["qty"], f"자동매매 #{order_id} (모의)")
-            msg = (f"✅ <b>매수 체결 #{order_id}</b>\n"
+        msg = (f"✅ <b>매수 {tag} #{order_id}</b> (모의)\n"
                f"{order['stock_name']}({order['stock_code']}) "
                f"{order['qty']}주 × {order['price']:,}원")
     else:  # sell
-            # 보유 종목 비활성화 + 실현손익 계산
         held = await conn.fetchrow("""
             SELECT id, buy_price, qty FROM user_portfolio
             WHERE stock_code=$1 AND active=true LIMIT 1
@@ -5830,7 +6340,6 @@ async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 overri
             realized = (order["price"] - held["buy_price"]) * order["qty"]
             await conn.execute(
                 "UPDATE user_portfolio SET active=false WHERE id=$1", held["id"])
-            # 일일 손익 누적
         await conn.execute("""
             INSERT INTO trading_daily_pnl (dt, realized_pnl, trades_count)
             VALUES (CURRENT_DATE, $1, 1)
@@ -5840,7 +6349,7 @@ async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 overri
               updated_at = NOW()
         """, realized)
         sign = "🟢" if realized > 0 else "🔴"
-            msg = (f"{sign} <b>매도 체결 #{order_id}</b>\n"
+        msg = (f"{sign} <b>매도 {tag} #{order_id}</b> (모의)\n"
                f"{order['stock_name']}({order['stock_code']}) "
                f"{order['qty']}주 × {order['price']:,}원\n"
                f"실현손익: {realized:+,}원")
@@ -5852,6 +6361,51 @@ async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 overri
     await send_telegram(msg)
     return {"status": "filled", "order_id": order_id}
 
+@app.post("/trade/confirm/{order_id}")
+async def trade_confirm_v2(order_id: int):    # noqa: F811 (이전 정의 override)
+    """매수/매도 confirm 통합 처리 (수동 버튼)."""
+    async with pg_pool.acquire() as conn:
+        return await _fill_order(conn, order_id, auto=False)
+
+@app.get("/trade/history")
+async def trade_history(days: int = Query(default=30, ge=1, le=365)):
+    """모의매매 이력 + 손익 — 체결내역·일별손익·보유 평가손익."""
+    async with pg_pool.acquire() as conn:
+        fills = await conn.fetch("""
+            SELECT stock_code, stock_name, side, qty, price, filled_at
+            FROM trading_orders
+            WHERE status='filled' AND filled_at >= CURRENT_DATE - ($1::int)
+            ORDER BY filled_at DESC
+        """, days)
+        daily = await conn.fetch("""
+            SELECT dt, realized_pnl, unrealized_pnl, trades_count, halted
+            FROM trading_daily_pnl WHERE dt >= CURRENT_DATE - ($1::int) ORDER BY dt DESC
+        """, days)
+        holds = await conn.fetch("""
+            SELECT p.stock_code, p.stock_name, p.buy_price, p.qty,
+                   (SELECT price FROM stock_technical WHERE stock_code=p.stock_code) cur
+            FROM user_portfolio p WHERE active=true
+        """)
+    realized = sum((r["realized_pnl"] or 0) for r in daily)
+    holdings, unreal = [], 0
+    for h in holds:
+        cur = h["cur"] or h["buy_price"]
+        pl = int((cur - h["buy_price"]) * h["qty"]) if h["buy_price"] else 0
+        unreal += pl
+        holdings.append({"code": h["stock_code"], "name": h["stock_name"],
+                         "buy_price": h["buy_price"], "qty": h["qty"], "cur_price": cur,
+                         "pl": pl, "pl_pct": round((cur / h["buy_price"] - 1) * 100, 1) if h["buy_price"] else 0})
+    return {
+        "realized_pnl": realized, "unrealized_pnl": unreal, "total_pnl": realized + unreal,
+        "fill_count": len(fills),
+        "holdings": holdings,
+        "daily_pnl": [{"dt": str(r["dt"]), "realized": r["realized_pnl"],
+                       "unrealized": r["unrealized_pnl"], "trades": r["trades_count"],
+                       "halted": r["halted"]} for r in daily],
+        "fills": [{"code": f["stock_code"], "name": f["stock_name"], "side": f["side"],
+                   "qty": f["qty"], "price": f["price"], "at": str(f["filled_at"])} for f in fills],
+    }
+
 
 async def auto_trade_scan_job():
     """5분마다: 매수 신호 + 매도 신호 자동 스캔. 한도 도달 시 자동 halt."""
@@ -5861,16 +6415,31 @@ 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만 봐서 ① 평가손실이 아무리 커도
+            # 매도 전엔 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("""
-                        UPDATE trading_daily_pnl SET halted=true WHERE dt=CURRENT_DATE
+                    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"실현+평가 손실 {loss_pct:.2f}% ≤ -{abs(TRADE_SETTINGS['daily_loss_limit_pct'])}%\n"
+                    f"(실현 {realized:+,}원 / 평가 {int(unreal):+,}원)\n"
                     f"오늘 자동매매 중단됨"
                 )
                 return
@@ -5906,6 +6475,13 @@ async def auto_trade_scan_job():
                 sig = await evaluate_buy_signal(conn, c)
                 if sig["ok"]:
                     await propose_buy_order(conn, c)
+            # 자동실행(모의): auto_execute면 오늘 제안된 pending 주문 즉시 체결
+            if TRADE_SETTINGS.get("auto_execute"):
+                pend = await conn.fetch(
+                    "SELECT id FROM trading_orders WHERE status='pending' "
+                    "AND proposed_at::date=CURRENT_DATE")
+                for p in pend:
+                    await _fill_order(conn, p["id"], auto=True)
         logger.info("auto_trade_scan.done")
     except Exception as e:
         logger.error("auto_trade_scan.err", error=str(e))

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 {
@@ -713,16 +719,31 @@ async def init_db():
 
 # ── 전체 분석 작업 ────────────────────────────────────────
 
-async def job_analyze():
+async def job_analyze(limit: int = 500):
-    logger.info("ta.job.start")
+    """limit>0: 시총 상위 N개(장중 경량). limit=0: 전 활성종목(장마감 풀커버).
+    is_active=true 필터 필수 — 상장폐지 제외 + LS 등 누락 방지."""
+    logger.info("ta.job.start", limit=limit)
     async with httpx.AsyncClient() as client:
         codes: List[tuple] = []
 
         if pg_pool:
             try:
-                rows = await pg_pool.fetch("SELECT stock_code, corp_name FROM dart_corps LIMIT 500")
+                q = """
+                    SELECT c.stock_code, c.corp_name
+                    FROM dart_corps c
+                    LEFT JOIN (
+                        SELECT DISTINCT ON (stock_code) stock_code, market_cap
+                        FROM stock_prices ORDER BY stock_code, collected_at DESC
+                    ) p ON p.stock_code = c.stock_code
+                    WHERE c.is_active = true
+                    ORDER BY COALESCE(p.market_cap, 0) DESC
+                """
+                if limit and limit > 0:
+                    q += f" LIMIT {int(limit)}"
+                rows = await pg_pool.fetch(q)
                 codes = [(r["stock_code"], r["corp_name"] or "") for r in rows if r["stock_code"]]
-            except: pass
+            except Exception as e:
+                logger.warning("ta.codes.err", error=str(e))
 
         if not codes:
             for sosok in [0, 1]:
@@ -740,7 +761,7 @@ async def job_analyze():
                     if len(codes) >= 500: break
 
         ok = 0
-        for code, name in codes[:500]:
+        for code, name in codes:
             if not code or len(code) != 6: continue
             try:
                 result = await analyze_stock(client, code, name)
@@ -748,11 +769,11 @@ async def job_analyze():
             except Exception as e:
                 stats.errors += 1
                 logger.warning("ta.analyze.err", code=code, error=str(e))
-            await asyncio.sleep(0.4)
+            await asyncio.sleep(0.2)
 
         stats.analyzed += ok
         stats.last_run = datetime.now().isoformat()
-        logger.info("ta.job.done", analyzed=ok)
+        logger.info("ta.job.done", analyzed=ok, requested=len(codes))
 
 # ── FastAPI ────────────────────────────────────────────────
 
@@ -768,9 +789,9 @@ async def startup():
     redis_cl = aioredis.Redis(
         host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=5, decode_responses=True)
     await init_db()
-    scheduler.add_job(job_analyze, "cron", day_of_week="mon-fri",
+    scheduler.add_job(_ta_intraday, "cron", day_of_week="mon-fri",
                       hour="9-16", minute="*/30", id="ta_30m", replace_existing=True)
-    scheduler.add_job(job_analyze, "cron", day_of_week="mon-fri",
+    scheduler.add_job(_ta_close, "cron", day_of_week="mon-fri",
                       hour=16, minute=15, id="ta_close", replace_existing=True)
     scheduler.start()
     logger.info("ta-engine.started")
@@ -855,9 +876,15 @@ async def buy_candidates(limit: int = Query(default=20)):
         result.append(d)
     return result
 
+async def _ta_intraday():   # 코루틴 함수로 등록 (lambda 감싸면 APScheduler가 await 못함)
+    await job_analyze(limit=500)
+
+async def _ta_close():
+    await job_analyze(limit=0)
+
 @app.post("/analyze/all")
-async def analyze_all():
+async def analyze_all(limit: int = 0):
-    asyncio.create_task(job_analyze())
+    asyncio.create_task(job_analyze(limit=limit))
     return {"status": "started"}
 
 @app.post("/analyze/{code}")