chore: 누적 미커밋 작업분 일괄 커밋

이번 세션 외 그간 쌓인 변경 일괄 저장:
- bareunaapi: finance_dict 금융용어 / stock_loader 종목 로더 보강
- kis-api: 키움 토큰·수집 로직
- us-market / dart-collector: 수집 보강
- docker-compose: GEMINI_API_KEY 등 환경변수 추가
- score-engine/news-collector requirements, CLAUDE.md
- 신규: PROJECT.md, news-collector/sentiment_rules.py

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

kis-api/main.py

@@ -32,6 +32,8 @@ PG_PASS        = os.getenv("POSTGRES_PASSWORD", "")
 KIWOOM_APP_KEY    = os.getenv("KIWOOM_APP_KEY", "")
 KIWOOM_SECRET_KEY = os.getenv("KIWOOM_SECRET_KEY", "")
 KIWOOM_BASE_URL   = os.getenv("KIWOOM_BASE_URL", "https://api.kiwoom.com")
+TG_TOKEN          = os.getenv("TELEGRAM_BOT_TOKEN", "")
+TG_CHAT_ID        = os.getenv("TELEGRAM_CHAT_ID", "")
 
 pg_pool: Optional[asyncpg.Pool] = None
 redis_cl: Optional[aioredis.Redis] = None
@@ -68,6 +70,25 @@ class KiwoomToken:
 
 kiwoom_token = KiwoomToken()
 
+
+class _RateLimiter:
+    """Kiwoom 글로벌 레이트리밋 — 초당 max_rate건으로 직렬 페이싱 (429 방지)."""
+    def __init__(self, max_rate: float):
+        self.min_gap = 1.0 / max_rate
+        self._last = 0.0
+        self._lock = asyncio.Lock()
+
+    async def acquire(self):
+        async with self._lock:
+            loop = asyncio.get_running_loop()
+            wait = self.min_gap - (loop.time() - self._last)
+            if wait > 0:
+                await asyncio.sleep(wait)
+            self._last = loop.time()
+
+# 실측: 페이싱 시 10/s도 정상. 보수적으로 6/s (≈2,300종목 6분 주기).
+kiwoom_rate = _RateLimiter(6.0)
+
 # ── 키움 API 호출 헬퍼 ────────────────────────────────────
 
 async def kiwoom_post(client: httpx.AsyncClient, endpoint: str, api_id: str,
@@ -82,8 +103,15 @@ async def kiwoom_post(client: httpx.AsyncClient, endpoint: str, api_id: str,
     }
     if next_key:
         headers["next-key"] = next_key
-    r = await client.post(f"{KIWOOM_BASE_URL}{endpoint}",
-                          headers=headers, json=body, timeout=15)
+    # 글로벌 페이싱 + 429(요청한도 초과) 재시도
+    for attempt in range(3):
+        await kiwoom_rate.acquire()
+        r = await client.post(f"{KIWOOM_BASE_URL}{endpoint}",
+                              headers=headers, json=body, timeout=15)
+        if r.status_code == 429 and attempt < 2:
+            await asyncio.sleep(0.6 * (attempt + 1))
+            continue
+        break
     if return_headers:                       # 연속조회용 cont-yn/next-key 응답헤더 필요
         return r.json(), r.headers
     return r.json()
@@ -440,6 +468,28 @@ async def get_stock_codes(limit: int = 0) -> list:
             pass
     return []
 
+
+async def get_priority_codes(min_mcap: int = 30_000_000_000) -> list:
+    """5분 수집 우선군 — 시총 min_mcap(기본 300억) 이상 활성종목 (잡주 제외).
+    조회 실패·표본 부족 시 전체 종목으로 폴백."""
+    if pg_pool:
+        try:
+            rows = await pg_pool.fetch("""
+                SELECT d.stock_code FROM dart_corps d
+                JOIN LATERAL (
+                  SELECT market_cap FROM stock_prices
+                  WHERE stock_code=d.stock_code AND market_cap>0
+                  ORDER BY collected_at DESC LIMIT 1) p ON true
+                WHERE d.is_active=TRUE AND p.market_cap >= $1
+                ORDER BY d.stock_code
+            """, min_mcap)
+            codes = [r["stock_code"] for r in rows if r["stock_code"]]
+            if len(codes) >= 50:
+                return codes
+        except Exception:
+            pass
+    return await get_stock_codes(0)
+
 # ── 저장 함수 ─────────────────────────────────────────────
 
 async def save_price(info: dict):
@@ -551,29 +601,88 @@ class Stats:
 
 stats = Stats()
 
+_job_price_running = False
+
 async def job_price():
-    """평일 9~16시 5분마다: 현재가·재무지표 수집 (ka10001)"""
-    codes = await get_stock_codes(0)
-    if not codes:
-        logger.warning("job_price.no_codes")
+    """평일 9~15시: 시총 300억+ 우선군 현재가 수집 (ka10001).
+    잡주(시총<300억)는 후순위 — job_full(17시)에서 일봉으로 커버.
+    글로벌 6/s 페이싱으로 한 사이클 수 분 소요 → 중복 실행 가드."""
+    global _job_price_running
+    if _job_price_running:
+        logger.info("job_price.skip_running")
         return
-    sem = asyncio.Semaphore(10)  # 동시 10개 제한
-    ok = 0
-    async with httpx.AsyncClient() as client:
-        tasks = [fetch_basic_info(client, code, sem) for code in codes]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        for info in results:
-            if isinstance(info, dict) and info.get("price", 0) > 0:
-                await save_price(info)
-                ok += 1
-            else:
-                stats.errors += 1
-    await gen_signals()
-    stats.collected += ok
-    stats.last_run = datetime.now().isoformat()
-    if redis_cl:
-        await redis_cl.set("prices:last_update", datetime.now().isoformat())
-    logger.info("job_price.done", ok=ok, total=len(codes))
+    _job_price_running = True
+    try:
+        codes = await get_priority_codes()
+        if not codes:
+            logger.warning("job_price.no_codes")
+            return
+        sem = asyncio.Semaphore(8)  # 동시 in-flight 한도 (실제 속도는 kiwoom_rate가 제어)
+        ok = 0
+        async with httpx.AsyncClient() as client:
+            tasks = [fetch_basic_info(client, code, sem) for code in codes]
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            for info in results:
+                if isinstance(info, dict) and info.get("price", 0) > 0:
+                    await save_price(info)
+                    ok += 1
+                else:
+                    stats.errors += 1
+        await gen_signals()
+        stats.collected += ok
+        stats.last_run = datetime.now().isoformat()
+        if redis_cl:
+            await redis_cl.set("prices:last_update", datetime.now().isoformat())
+        try:
+            await check_price_alerts()
+        except Exception as e:
+            logger.warning("alert.check_err", error=str(e))
+        logger.info("job_price.done", ok=ok, total=len(codes))
+    finally:
+        _job_price_running = False
+
+_job_priority_running = False
+
+async def job_price_priority():
+    """30초마다: 보유 + 강력매수 + 매수관심 종목만 빠른 수집 (자동매매·알림 latency 단축).
+    키움 rate limit 안전 — 우선순위 종목은 보통 30~50건이라 6/s × 10초 이내 완료."""
+    global _job_priority_running
+    if _job_priority_running:
+        return
+    _job_priority_running = True
+    try:
+        async with pg_pool.acquire() as conn:
+            rows = await conn.fetch("""
+                SELECT DISTINCT stock_code FROM (
+                    SELECT stock_code FROM user_portfolio WHERE active=true
+                    UNION
+                    SELECT stock_code FROM user_alerts WHERE active=true
+                    UNION
+                    SELECT stock_code FROM stock_scores
+                    WHERE score_date=(SELECT MAX(score_date) FROM stock_scores)
+                      AND recommendation IN ('강력매수','매수관심')
+                ) AS t LIMIT 60
+            """)
+        codes = [r["stock_code"] for r in rows]
+        if not codes:
+            return
+        sem = asyncio.Semaphore(8)
+        ok = 0
+        async with httpx.AsyncClient() as client:
+            tasks = [fetch_basic_info(client, c, sem) for c in codes]
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            for info in results:
+                if isinstance(info, dict) and info.get("price", 0) > 0:
+                    await save_price(info)
+                    ok += 1
+        try:
+            await check_price_alerts()      # 알림 즉시 평가
+        except Exception as e:
+            logger.warning("priority.alert_err", error=str(e))
+        logger.info("job_price_priority.done", ok=ok, total=len(codes))
+    finally:
+        _job_priority_running = False
+
 
 async def job_full(days: int = 10):
     """평일 17:00: 일봉·외국인·공매도 전체 수집 (ka10005·ka10008·ka10014).
@@ -736,9 +845,12 @@ async def startup():
     redis_cl = aioredis.Redis(
         host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=3, decode_responses=True)
     await init_db()
-    # 현재가: 평일 9~16시 5분마다
+    # 현재가: 평일 9~16시 2분마다 (전체)
     scheduler.add_job(job_price, "cron", day_of_week="mon-fri",
-                      hour="9-15", minute="*/5", id="price", replace_existing=True)
+                      hour="9-15", minute="*/2", id="price", replace_existing=True)
+    # Fast track: 보유 + 강력매수 + 매수관심만 30초마다 (자동매매 latency)
+    scheduler.add_job(job_price_priority, "cron", day_of_week="mon-fri",
+                      hour="9-15", second="*/30", id="price_priority", replace_existing=True)
     # 장 마감 후 전체 수집: 평일 17:00
     scheduler.add_job(job_full, "cron", day_of_week="mon-fri",
                       hour="17", minute="0", id="full", replace_existing=True)
@@ -858,6 +970,229 @@ async def summary(code: str):
                 result[field] = json.loads(c) if field != "price" else json.loads(c)
     return JSONResponse(content=result)
 
+# ── 가격 알림 (user_alerts 트리거 + 5가지 물타기 진단) ────────
+
+async def send_telegram(msg: str):
+    if not TG_TOKEN or not TG_CHAT_ID:
+        return
+    try:
+        async with httpx.AsyncClient() as c:
+            await c.post(
+                f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage",
+                json={"chat_id": TG_CHAT_ID, "text": msg, "parse_mode": "HTML"},
+                timeout=10)
+    except Exception as e:
+        logger.warning("telegram.err", error=str(e))
+
+
+async def diagnose_avg_down(conn, code: str, current_price: float) -> tuple[int, list[str]]:
+    """물타기 진단 5가지 조건. (충족 개수, 표시줄 리스트) 반환."""
+    score = 0
+    lines = []
+
+    # 1. RSI ≤ 30 (과매도)
+    rsi = await conn.fetchval("""
+        SELECT rsi FROM stock_technical WHERE stock_code=$1
+    """, code)
+    rsi_disp = f"{rsi:.0f}" if rsi is not None else "N/A"
+    if rsi is not None and rsi <= 30:
+        score += 1
+        lines.append(f"✅ RSI {rsi_disp} (과매도)")
+    elif rsi is not None and rsi <= 40:
+        lines.append(f"⚠️ RSI {rsi_disp} (약한 과매도)")
+    else:
+        lines.append(f"❌ RSI {rsi_disp} (과매도 아님)")
+
+    # 2. 거래량 비율 ≤ 1.5 (패닉 매도 끝)
+    vol_ratio = await conn.fetchval("""
+        SELECT vol_ratio FROM stock_technical WHERE stock_code=$1
+    """, code)
+    vol_disp = f"{vol_ratio:.2f}" if vol_ratio is not None else "N/A"
+    if vol_ratio is not None and vol_ratio <= 1.5:
+        score += 1
+        lines.append(f"✅ 거래량비율 {vol_disp} (안정)")
+    else:
+        lines.append(f"❌ 거래량비율 {vol_disp} (패닉 진행)")
+
+    # 3. 공매도 잔고 감소 추세 (최근 5일)
+    short_rows = await conn.fetch("""
+        SELECT short_balance_qty FROM stock_short_sale
+        WHERE stock_code=$1 ORDER BY dt DESC LIMIT 5
+    """, code)
+    if len(short_rows) >= 2:
+        latest = short_rows[0]["short_balance_qty"] or 0
+        older  = short_rows[-1]["short_balance_qty"] or 0
+        if older > 0 and latest < older:
+            chg = (latest - older) / older * 100
+            score += 1
+            lines.append(f"✅ 공매도잔고 {chg:+.1f}% (감소)")
+        else:
+            lines.append(f"❌ 공매도잔고 증가/유지")
+    else:
+        lines.append("⚠️ 공매도 데이터 부족")
+
+    # 4. 최근 24시간 악재(intensity≥3) 없음
+    bad_news = await conn.fetchval("""
+        SELECT COUNT(*) FROM news_analysis
+        WHERE primary_stock=$1
+          AND sentiment='악재' AND intensity >= 3
+          AND analyzed_at >= NOW() - INTERVAL '24 hours'
+    """, code)
+    if bad_news == 0:
+        score += 1
+        lines.append("✅ 최근 악재 없음")
+    else:
+        lines.append(f"❌ 최근 악재 {bad_news}건")
+
+    # 5. 시장 레짐 = 강세/중립 (약세장 회피)
+    regime = await conn.fetchval("""
+        SELECT regime FROM market_regime ORDER BY dt DESC LIMIT 1
+    """)
+    if regime in ("강세", "중립"):
+        score += 1
+        lines.append(f"✅ 시장레짐 {regime}")
+    else:
+        lines.append(f"❌ 시장레짐 {regime} (약세장 회피)")
+
+    return score, lines
+
+
+async def check_price_alerts():
+    """수집 후 호출: active=true 알림 중 임계값 도달한 것 처리."""
+    if not pg_pool:
+        return
+    async with pg_pool.acquire() as conn:
+        alerts = await conn.fetch("""
+            SELECT a.id, a.stock_code, a.alert_type, a.threshold,
+                   d.corp_name
+            FROM user_alerts a
+            LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
+            WHERE a.active = true
+        """)
+        for a in alerts:
+            code = a["stock_code"]
+            price = 0.0
+            # 1차: Redis price:{code} (평일 수집 직후)
+            cached = await redis_cl.get(f"price:{code}") if redis_cl else None
+            if cached:
+                try:
+                    price = float(json.loads(cached).get("price", 0))
+                except Exception:
+                    price = 0.0
+            # 2차 폴백: stock_ohlcv 최신 종가 (장 닫힘·주말·실패 대비)
+            if price <= 0:
+                p = await conn.fetchval("""
+                    SELECT close_price FROM stock_ohlcv
+                    WHERE stock_code=$1 ORDER BY dt DESC LIMIT 1
+                """, code)
+                price = float(p) if p else 0.0
+            if price <= 0:
+                continue
+            hit = False
+            if a["alert_type"] == "price_below" and price <= a["threshold"]:
+                hit = True
+            elif a["alert_type"] == "price_above" and price >= a["threshold"]:
+                hit = True
+            if not hit:
+                continue
+
+            score, lines = await diagnose_avg_down(conn, code, price)
+            verdict = "🟢 물타기 OK" if score >= 4 else ("🟡 신중" if score >= 3 else "🔴 NO")
+            name = a["corp_name"] or code
+
+            # LLM hybrid 분석 동봉 (score-engine /deep-analysis 호출)
+            llm_block = ""
+            try:
+                async with httpx.AsyncClient(timeout=120) as c:
+                    r = await c.get(
+                        f"http://score-engine:8686/deep-analysis/{code}",
+                        params={"refresh": "true", "model": "hybrid", "notify": "false"})
+                    if r.status_code == 200:
+                        d = r.json()
+                        if not d.get("error"):
+                            agr = d.get("agreement")
+                            agr_tag = "✅두AI일치" if agr is True else ("⚠️AI의견갈림" if agr is False else "")
+                            llm_block = (
+                                f"\n\n🤖 <b>AI 하이브리드 분석 {agr_tag}</b>\n"
+                                f"판단: {d.get('recommendation','-')} (확신도 {d.get('conviction',0)}/5)\n"
+                                f"논거: {d.get('thesis','')[:250]}\n"
+                                f"🎯 목표 {int(d.get('target_price') or 0):,}원 / "
+                                f"손절 {int(d.get('stop_loss') or 0):,}원"
+                            )
+                            if agr is False and d.get("exaone"):
+                                e = d["exaone"]
+                                llm_block += (
+                                    f"\n🔄 EXAONE 의견: {e.get('recommendation')} "
+                                    f"{e.get('conviction',0)}/5 — {e.get('thesis','')[:120]}"
+                                )
+            except Exception as e:
+                logger.warning("alert.llm_err", code=code, error=str(e))
+
+            msg = (
+                f"🔔 <b>가격 알림: {name} ({code})</b>\n"
+                f"현재가: {int(price):,}원\n"
+                f"임계값: {int(a['threshold']):,}원 ({a['alert_type']})\n\n"
+                f"📊 <b>물타기 진단 ({score}/5)</b>\n"
+                + "\n".join(lines)
+                + f"\n\n<b>종합: {verdict}</b>"
+                + llm_block
+            )
+            await send_telegram(msg)
+            await conn.execute("""
+                UPDATE user_alerts
+                SET active=false, last_triggered=NOW()
+                WHERE id=$1
+            """, a["id"])
+            logger.info("alert.fired", code=code, price=price,
+                        threshold=a["threshold"], verdict=verdict)
+
+
+@app.post("/alerts/register")
+async def register_alert(code: str = Query(...),
+                         alert_type: str = Query(..., regex="^(price_below|price_above)$"),
+                         threshold: float = Query(..., gt=0),
+                         user_id: int = Query(default=1)):
+    async with pg_pool.acquire() as c:
+        row = await c.fetchrow("""
+            INSERT INTO user_alerts (user_id, stock_code, alert_type, threshold, active)
+            VALUES ($1, $2, $3, $4, true)
+            RETURNING id
+        """, user_id, code, alert_type, threshold)
+    return {"id": row["id"], "code": code, "alert_type": alert_type, "threshold": threshold}
+
+
+@app.get("/alerts")
+async def list_alerts(active_only: bool = Query(default=True)):
+    async with pg_pool.acquire() as c:
+        if active_only:
+            rows = await c.fetch("""
+                SELECT a.*, d.corp_name FROM user_alerts a
+                LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
+                WHERE a.active=true ORDER BY a.created_at DESC
+            """)
+        else:
+            rows = await c.fetch("""
+                SELECT a.*, d.corp_name FROM user_alerts a
+                LEFT JOIN dart_corps d ON d.stock_code=a.stock_code
+                ORDER BY a.created_at DESC LIMIT 50
+            """)
+    return [dict(r) for r in rows]
+
+
+@app.post("/alerts/check")
+async def trigger_alert_check():
+    """수동 알림 체크 — 주말·장마감·테스트 시 사용 (job_price 내 자동 호출 외 추가 경로)."""
+    await check_price_alerts()
+    return {"status": "checked"}
+
+
+@app.delete("/alerts/{alert_id}")
+async def delete_alert(alert_id: int):
+    async with pg_pool.acquire() as c:
+        await c.execute("UPDATE user_alerts SET active=false WHERE id=$1", alert_id)
+    return {"status": "deactivated", "id": alert_id}
+
+
 @app.post("/collect/price")
 async def collect_price():
     asyncio.create_task(job_price())