diff --git a/docs/ai_org.md b/docs/ai_org.md
new file mode 100644
index 0000000..8ed70d1
--- /dev/null
+++ b/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)
diff --git a/score-engine/main.py b/score-engine/main.py
index 8268ec3..00a08ae 100644
--- a/score-engine/main.py
+++ b/score-engine/main.py
@@ -2061,8 +2061,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 필터
@@ -2818,7 +2820,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 +2855,8 @@ async def calculate_daily_scores(as_of: date | None = None):
         except Exception as e:
             logger.warning("new_buy.err", error=str(e))
 
-    # 텔레그램 알림
-    if strong_buy or strong_sell:
+    # 텔레그램 알림 (notify=True 정기 1회만 — 매 호출 발신 방지)
+    if notify and (strong_buy or strong_sell):
         lines = [f"<b>📊 Trading AI 일간 리포트 ({today})</b>\n"]
         if strong_buy:
             lines.append("🟢 <b>강력매수 추천 (버핏 가치필터 통과)</b>")
@@ -3027,7 +3029,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(lambda: calculate_daily_scores(notify=True), "cron",
                       day_of_week="mon-fri", hour=16, minute=30,
                       id="daily_score", replace_existing=True)
     # 텔레그램 정기 알림 하루 2회로 축소 (사용자 요청 2026-06-02):
@@ -3038,6 +3040,14 @@ 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)
     # 성과 추적: 매일 18시 가격 업데이트
     scheduler.add_job(update_performance_prices, "cron",
                       day_of_week="mon-fri", hour=18, minute=0,
@@ -3144,13 +3154,117 @@ 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}")
+    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 수익률·알파) 집계.
+    매수계열 알파>0 & 매도계열 알파<0 이면 방식 유효."""
+    async with pg_pool.acquire() as conn:
+        grades = await conn.fetch("""
+            SELECT recommendation rec, COUNT(*) n,
+                   AVG(return_7d) ret7, AVG(alpha_7d) a7,
+                   AVG(return_30d) ret30, AVG(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)
+    order = {"강력매수": 0, "매수관심": 1, "관망": 2, "매도관심": 3, "강력매도": 4}
+    rows = sorted([dict(g) for g in grades], key=lambda x: order.get(x["rec"], 9))
+    def wavg(sel, key):
+        tot = sum(r["n"] for r in rows if r["rec"] in sel)
+        s = sum((r[key] or 0) * r["n"] for r in rows if r["rec"] in sel)
+        return round(s / tot, 2) if tot else None
+    buy_a, sell_a = wavg(("강력매수", "매수관심"), "a7"), wavg(("강력매도", "매도관심"), "a7")
+    ok = (buy_a is not None and sell_a is not None and buy_a > 0 and sell_a < 0)
+    return {
+        "days": days,
+        "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,
+        "verdict": "양호 (매수 알파>0, 매도 알파<0)" if ok else "교정필요 (매수/매도 변별력 부족)",
+    }
+
+@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일·7일 알파)</b>",
+             f"판정: <b>{a['verdict']}</b>",
+             f"매수계열 알파 {a['buy_alpha7']} / 매도계열 알파 {a['sell_alpha7']}\n"]
+    for g in a["grades"]:
+        lines.append(f"{g['rec']}: n{g['n']} 수익{g['ret7']}% 알파{g['alpha7']}% 상승{g['up7_pct']}%")
+    await send_telegram("\n".join(lines))
+    logger.info("accuracy.report.sent", verdict=a["verdict"])
+
 @app.get("/health")
 async def health():
     return {"status": "ok"}
 
 @app.post("/score/calculate")
-async def manual_calc():
-    n = await calculate_daily_scores()
+async def manual_calc(notify: bool = Query(default=False)):
+    n = await calculate_daily_scores(notify=notify)
     return {"status": "done", "scored": n}
 
 
diff --git a/ta-engine/main.py b/ta-engine/main.py
index 6b01a92..e91ac90 100644
--- a/ta-engine/main.py
+++ b/ta-engine/main.py
@@ -713,16 +713,31 @@ async def init_db():
 
 # ── 전체 분석 작업 ────────────────────────────────────────
 
-async def job_analyze():
-    logger.info("ta.job.start")
+async def job_analyze(limit: int = 500):
+    """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 +755,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 +763,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 +783,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(lambda: job_analyze(limit=500), "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(lambda: job_analyze(limit=0), "cron", day_of_week="mon-fri",
                       hour=16, minute=15, id="ta_close", replace_existing=True)
     scheduler.start()
     logger.info("ta-engine.started")
@@ -856,8 +871,8 @@ async def buy_candidates(limit: int = Query(default=20)):
     return result
 
 @app.post("/analyze/all")
-async def analyze_all():
-    asyncio.create_task(job_analyze())
+async def analyze_all(limit: int = 0):
+    asyncio.create_task(job_analyze(limit=limit))
     return {"status": "started"}
 
 @app.post("/analyze/{code}")