"""
키움증권 REST API 기반 주가·수급·공매도 수집 서비스
- ka10001: 현재가 + 재무지표 (PER/PBR/ROE/EPS/BPS/외국인비중/시가총액)
- ka10005: 일봉 OHLCV + 외국인·기관 순매수
- ka10008: 외국인 종목별 매매동향 (일자별 보유비중 변화)
- ka10014: 공매도 추이 (잔고수량·거래비중)
- Redis db=3 price:{code} 형식 유지 (score-engine·ta-engine 호환)
"""
import asyncio, json, os, re
from datetime import datetime, timedelta
from typing import Optional
import asyncpg, httpx, redis.asyncio as aioredis, structlog
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware

structlog.configure(processors=[
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.add_log_level,
    structlog.processors.JSONRenderer(),
])
logger = structlog.get_logger()

REDIS_HOST     = os.getenv("REDIS_HOST", "redis")
REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
PG_HOST        = os.getenv("POSTGRES_HOST", "postgres")
PG_PORT        = int(os.getenv("POSTGRES_PORT", "5432"))
PG_DB          = os.getenv("POSTGRES_DB", "trading_ai")
PG_USER        = os.getenv("POSTGRES_USER", "kyu")
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
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",
                  "appkey": KIWOOM_APP_KEY,
                  "secretkey": KIWOOM_SECRET_KEY},
            headers={"Content-Type": "application/json;charset=UTF-8"},
            timeout=10)
        data = resp.json()
        if data.get("return_code", -1) != 0:
            raise RuntimeError(f"토큰 발급 실패: {data.get('return_msg')}")
        self.token = data["token"]
        # expires_dt: "20260508151325" 형식
        exp_str = data.get("expires_dt", "")
        try:
            self.expires_at = datetime.strptime(exp_str, "%Y%m%d%H%M%S") - timedelta(minutes=5)
        except Exception:
            self.expires_at = datetime.now() + timedelta(hours=23)
        logger.info("kiwoom.token.refreshed", expires=exp_str)
        return self.token

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,
                      body: dict, cont_yn: str = "N", next_key: str = "",
                      return_headers: bool = False):
    token = await kiwoom_token.get(client)
    headers = {
        "Content-Type": "application/json;charset=UTF-8",
        "Authorization": f"Bearer {token}",
        "api-id": api_id,
        "cont-yn": cont_yn,
    }
    if next_key:
        headers["next-key"] = next_key
    # 글로벌 페이싱 + 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()

def to_int(v: str) -> int:
    try:
        return int(str(v).replace(",", "").lstrip("+").replace(" ", "") or 0)
    except:
        return 0

def to_float(v: str) -> float:
    try:
        return float(str(v).replace(",", "").lstrip("+").replace(" ", "") or 0)
    except:
        return 0.0

# ── DB 초기화 ─────────────────────────────────────────────

async def init_db():
    async with pg_pool.acquire() as c:
        await c.execute("""
            CREATE TABLE IF NOT EXISTS stock_prices (
                id SERIAL PRIMARY KEY,
                stock_code VARCHAR(10) NOT NULL,
                stock_name VARCHAR(100) DEFAULT '',
                price INTEGER DEFAULT 0,
                change_pct FLOAT DEFAULT 0,
                change_amount INTEGER DEFAULT 0,
                volume BIGINT DEFAULT 0,
                high INTEGER DEFAULT 0,
                low INTEGER DEFAULT 0,
                open_price INTEGER DEFAULT 0,
                market_cap BIGINT DEFAULT 0,
                per FLOAT DEFAULT 0,
                pbr FLOAT DEFAULT 0,
                eps FLOAT DEFAULT 0,
                bps FLOAT DEFAULT 0,
                roe FLOAT DEFAULT 0,
                ev FLOAT DEFAULT 0,
                high_52w INTEGER DEFAULT 0,
                low_52w INTEGER DEFAULT 0,
                foreign_ratio FLOAT DEFAULT 0,
                credit_ratio FLOAT DEFAULT 0,
                float_shares BIGINT DEFAULT 0,
                collected_at TIMESTAMP DEFAULT NOW()
            )""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_sp_code ON stock_prices(stock_code)")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_sp_time ON stock_prices(collected_at DESC)")

        await c.execute("""
            CREATE TABLE IF NOT EXISTS stock_ohlcv (
                id SERIAL PRIMARY KEY,
                stock_code VARCHAR(10) NOT NULL,
                dt DATE NOT NULL,
                open_price INTEGER DEFAULT 0,
                high_price INTEGER DEFAULT 0,
                low_price INTEGER DEFAULT 0,
                close_price INTEGER DEFAULT 0,
                volume BIGINT DEFAULT 0,
                trade_amount BIGINT DEFAULT 0,
                foreign_ratio FLOAT DEFAULT 0,
                foreign_net BIGINT DEFAULT 0,
                institution_net BIGINT DEFAULT 0,
                individual_net BIGINT DEFAULT 0,
                created_at TIMESTAMP DEFAULT NOW(),
                UNIQUE(stock_code, dt)
            )""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ohlcv_code_dt ON stock_ohlcv(stock_code, dt DESC)")

        await c.execute("""
            CREATE TABLE IF NOT EXISTS stock_foreign_flow (
                id SERIAL PRIMARY KEY,
                stock_code VARCHAR(10) NOT NULL,
                dt DATE NOT NULL,
                close_price INTEGER DEFAULT 0,
                change_qty BIGINT DEFAULT 0,
                hold_qty BIGINT DEFAULT 0,
                hold_ratio FLOAT DEFAULT 0,
                limit_ratio FLOAT DEFAULT 0,
                created_at TIMESTAMP DEFAULT NOW(),
                UNIQUE(stock_code, dt)
            )""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ff_code_dt ON stock_foreign_flow(stock_code, dt DESC)")

        await c.execute("""
            CREATE TABLE IF NOT EXISTS stock_short_sale (
                id SERIAL PRIMARY KEY,
                stock_code VARCHAR(10) NOT NULL,
                dt DATE NOT NULL,
                close_price INTEGER DEFAULT 0,
                short_qty BIGINT DEFAULT 0,
                short_balance_qty BIGINT DEFAULT 0,
                trade_weight FLOAT DEFAULT 0,
                short_avg_price INTEGER DEFAULT 0,
                created_at TIMESTAMP DEFAULT NOW(),
                UNIQUE(stock_code, dt)
            )""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ss_code_dt ON stock_short_sale(stock_code, dt DESC)")

        await c.execute("""
            CREATE TABLE IF NOT EXISTS trade_signals (
                id SERIAL PRIMARY KEY,
                stock_code VARCHAR(10) NOT NULL,
                stock_name VARCHAR(100) DEFAULT '',
                signal_type VARCHAR(10) NOT NULL,
                current_price INTEGER DEFAULT 0,
                target_price INTEGER DEFAULT 0,
                stop_loss INTEGER DEFAULT 0,
                expected_return_pct FLOAT DEFAULT 0,
                risk_reward_ratio FLOAT DEFAULT 0,
                confidence FLOAT DEFAULT 0,
                reason TEXT DEFAULT '',
                news_score FLOAT DEFAULT 0,
                dart_score FLOAT DEFAULT 0,
                price_momentum FLOAT DEFAULT 0,
                foreign_net_5d BIGINT DEFAULT 0,
                short_weight FLOAT DEFAULT 0,
                created_at TIMESTAMP DEFAULT NOW()
            )""")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ts_code ON trade_signals(stock_code)")
        await c.execute("CREATE INDEX IF NOT EXISTS idx_ts_time ON trade_signals(created_at DESC)")
    logger.info("kiwoom.db.initialized")

# ── 수집 함수 ─────────────────────────────────────────────

async def fetch_basic_info(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> Optional[dict]:
    """ka10001: 현재가 + 재무지표 (PER/PBR/ROE/EPS/BPS/외국인비중/시가총액)"""
    async with sem:
        try:
            d = await kiwoom_post(client, "/api/dostk/stkinfo", "ka10001", {"stk_cd": code})
            if d.get("return_code", -1) != 0:
                return None
            return {
                "code": code,
                "name": d.get("stk_nm", ""),
                "price": abs(to_int(d.get("cur_prc", "0"))),
                "change_amount": to_int(d.get("pred_pre", "0")),
                "change_pct": to_float(d.get("flu_rt", "0")),
                "volume": to_int(d.get("trde_qty", "0")),
                "open_price": abs(to_int(d.get("open_pric", "0"))),
                "high": abs(to_int(d.get("high_pric", "0"))),
                "low": abs(to_int(d.get("low_pric", "0"))),
                "market_cap": to_int(d.get("mac", "0")) * 100_000_000,  # 억 → 원
                "per": to_float(d.get("per", "0")),
                "pbr": to_float(d.get("pbr", "0")),
                "eps": to_float(d.get("eps", "0")),
                "bps": to_float(d.get("bps", "0")),
                "roe": to_float(d.get("roe", "0")),
                "ev": to_float(d.get("ev", "0")),
                "high_52w": abs(to_int(d.get("250hgst", "0"))),
                "low_52w": abs(to_int(d.get("250lwst", "0"))),
                "foreign_ratio": to_float(d.get("for_exh_rt", "0")),
                "credit_ratio": to_float(d.get("crd_rt", "0")),
                "float_shares": to_int(d.get("flo_stk", "0")),
                "sale_amt": to_int(d.get("sale_amt", "0")),
                "operating_profit": to_int(d.get("bus_pro", "0")),
                "net_income": to_int(d.get("cup_nga", "0")),
                "timestamp": datetime.now().isoformat(),
            }
        except Exception as e:
            logger.debug("ka10001.err", code=code, error=str(e))
            return None

async def fetch_ohlcv(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore, days: int = 365) -> list:
    """ka10005: 일봉 OHLCV + 외국인·기관 순매수"""
    async with sem:
        try:
            today = datetime.now().strftime("%Y%m%d")
            result = []
            cont_yn, next_key = "N", ""
            for _ in range(20):                       # 안전 상한(20p ≈ 365봉 충분)
                d, hdr = await kiwoom_post(
                    client, "/api/dostk/mrkcond", "ka10005",
                    {"stk_cd": code, "dt": today, "req_cnt": days},
                    cont_yn=cont_yn, next_key=next_key, return_headers=True)
                for r in d.get("stk_ddwkmm", []):
                    cp = abs(to_int(r.get("close_pric", "0")))
                    if cp <= 0:
                        continue
                    result.append({
                        "dt": r.get("date", ""),
                        "open": abs(to_int(r.get("open_pric", "0"))),
                        "high": abs(to_int(r.get("high_pric", "0"))),
                        "low": abs(to_int(r.get("low_pric", "0"))),
                        "close": cp,
                        "volume": to_int(r.get("trde_qty", "0")),
                        "trade_amount": to_int(r.get("trde_prica", "0")),
                        "foreign_ratio": to_float(r.get("for_poss", "0")),
                        "foreign_net": to_int(r.get("for_netprps", "0")),
                        "institution_net": to_int(r.get("orgn_netprps", "0")),
                        "individual_net": to_int(r.get("ind_netprps", "0")),
                    })
                # 키움 연속조회: 응답 cont-yn=Y + next-key 있으면 다음 페이지
                if (len(result) >= days or hdr.get("cont-yn") != "Y"
                        or not hdr.get("next-key")):
                    break
                cont_yn, next_key = "Y", hdr.get("next-key", "")
            return result
        except Exception as e:
            logger.debug("ka10005.err", code=code, error=str(e))
            return []

async def fetch_foreign_flow(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> list:
    """ka10008: 외국인 종목별 매매동향"""
    async with sem:
        try:
            d = await kiwoom_post(client, "/api/dostk/frgnistt", "ka10008", {"stk_cd": code})
            rows = d.get("stk_frgnr", [])
            result = []
            for r in rows:
                result.append({
                    "dt": r.get("dt", ""),
                    "close_price": abs(to_int(r.get("close_pric", "0"))),
                    "change_qty": to_int(r.get("chg_qty", "0")),
                    "hold_qty": to_int(r.get("poss_stkcnt", "0")),
                    "hold_ratio": to_float(r.get("wght", "0")),
                    "limit_ratio": to_float(r.get("limit_exh_rt", "0")),
                })
            return result
        except Exception as e:
            logger.debug("ka10008.err", code=code, error=str(e))
            return []

async def fetch_minute_chart(client: httpx.AsyncClient, code: str, tic_scope: str = "1") -> list:
    """ka10080: 주식분봉차트조회. tic_scope: 1, 3, 5, 10, 15, 30, 45, 60 분"""
    try:
        d = await kiwoom_post(client, "/api/dostk/chart", "ka10080",
            {"stk_cd": code, "tic_scope": tic_scope, "upd_stkpc_tp": "1"})
        if d.get("return_code", -1) != 0:
            return []
        rows = d.get("stk_min_pole_chart_qry", []) or d.get("stk_min_pole", []) or []
        result = []
        for r in rows:
            result.append({
                "dt":     r.get("cntr_tm",  r.get("dt", "")),
                "open":   abs(to_int(r.get("open_pric", "0"))),
                "high":   abs(to_int(r.get("high_pric", "0"))),
                "low":    abs(to_int(r.get("low_pric",  "0"))),
                "close":  abs(to_int(r.get("cur_prc",   r.get("close_pric", "0")))),
                "volume": to_int(r.get("trde_qty",  "0")),
            })
        return result
    except Exception as e:
        logger.debug("ka10080.err", code=code, error=str(e))
        return []

async def fetch_orderbook(client: httpx.AsyncClient, code: str) -> dict:
    """ka10004: 호가잔량 (10단계 매수/매도). path=/api/dostk/mrkcond"""
    try:
        d = await kiwoom_post(client, "/api/dostk/mrkcond", "ka10004", {"stk_cd": code})
        if d.get("return_code", -1) != 0:
            return {}
        bid, ask = [], []
        # 매도 호가 1~10 (sel_1th_pre_bid = 가격, sel_1th_pre_req = 잔량)
        for i in range(1, 11):
            p = abs(to_int(d.get(f"sel_{i}th_pre_bid", "0")))
            q = to_int(d.get(f"sel_{i}th_pre_req", "0"))
            if p > 0:
                ask.append({"price": p, "qty": q})
        # 매수 호가 1~10
        for i in range(1, 11):
            p = abs(to_int(d.get(f"buy_{i}th_pre_bid", "0")))
            q = to_int(d.get(f"buy_{i}th_pre_req", "0"))
            if p > 0:
                bid.append({"price": p, "qty": q})
        return {
            "code": code,
            "ask": ask,            # 매도 (가격 낮은 것부터)
            "bid": bid,            # 매수 (가격 높은 것부터)
            "ask_total": to_int(d.get("tot_sel_req", "0")),
            "bid_total": to_int(d.get("tot_buy_req", "0")),
            "base_time": d.get("bid_req_base_tm", ""),
            "timestamp": datetime.now().isoformat(),
        }
    except Exception as e:
        logger.debug("ka10004.err", code=code, error=str(e))
        return {}

async def fetch_volume_surge() -> list:
    """ka10023: 거래량급증 종목. path=/api/dostk/rkinfo"""
    try:
        async with httpx.AsyncClient() as c:
            d = await kiwoom_post(c, "/api/dostk/rkinfo", "ka10023",
                {"mrkt_tp": "000",     # 000=전체, 001=코스피, 101=코스닥
                 "sort_tp": "1",        # 1=급증량, 2=급증률
                 "tm_tp":   "2",        # 1=분, 2=전일
                 "trde_qty_tp": "5",   # 1=5천주, 2=1만주, 5=5만주
                 "tm": "",
                 "stk_cnd": "0",        # 0=전체
                 "pric_tp": "0",        # 0=전체
                 "stex_tp": "3"})        # 3=통합
            if d.get("return_code", -1) != 0:
                return []
            rows = d.get("trde_qty_sdnin", []) or []
            result = []
            for r in rows[:50]:
                raw_code = r.get("stk_cd", "")
                # NXT 등 접미사 제거 ('_AL', '_NX' 등)
                clean_code = raw_code.split("_")[0] if raw_code else ""
                result.append({
                    "code":       clean_code,
                    "name":       r.get("stk_nm", ""),
                    "price":      abs(to_int(r.get("cur_prc", "0"))),
                    "change_pct": to_float(r.get("flu_rt", "0")),
                    "volume":     to_int(r.get("now_trde_qty", r.get("trde_qty", "0"))),
                    "prev_volume":to_int(r.get("pred_trde_qty", "0")),
                    "surge_rate": to_float(r.get("sdnin_rt", r.get("sdnin_qty", "0"))),
                })
            return result
    except Exception as e:
        logger.debug("ka10023.err", error=str(e))
        return []

async def fetch_short_sale(client: httpx.AsyncClient, code: str, sem: asyncio.Semaphore) -> list:
    """ka10014: 공매도 추이"""
    async with sem:
        try:
            today = datetime.now().strftime("%Y%m%d")
            month_ago = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d")
            d = await kiwoom_post(client, "/api/dostk/shsa", "ka10014",
                                  {"stk_cd": code, "strt_dt": month_ago, "end_dt": today})
            rows = d.get("shrts_trnsn", [])
            result = []
            for r in rows:
                result.append({
                    "dt": r.get("dt", ""),
                    "close_price": abs(to_int(r.get("close_pric", "0"))),
                    "short_qty": to_int(r.get("shrts_qty", "0")),
                    "short_balance_qty": to_int(r.get("ovr_shrts_qty", "0")),
                    "trade_weight": to_float(r.get("trde_wght", "0")),
                    "short_avg_price": to_int(r.get("shrts_avg_pric", "0")),
                })
            return result
        except Exception as e:
            logger.debug("ka10014.err", code=code, error=str(e))
            return []

# ── 종목 코드 로드 ────────────────────────────────────────

async def get_stock_codes(limit: int = 0) -> list:
    """is_active 종목 전체 (limit=0이면 제한 없음)"""
    if pg_pool:
        try:
            if limit > 0:
                rows = await pg_pool.fetch(
                    "SELECT stock_code FROM dart_corps WHERE is_active=TRUE ORDER BY stock_code LIMIT $1", limit)
            else:
                rows = await pg_pool.fetch(
                    "SELECT stock_code FROM dart_corps WHERE is_active=TRUE ORDER BY stock_code")
            codes = [r["stock_code"] for r in rows if r["stock_code"]]
            if len(codes) >= 50:
                return codes
        except Exception:
            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):
    if redis_cl:
        redis_data = {
            "code": info["code"], "name": info["name"],
            "price": info["price"], "change_pct": info["change_pct"],
            "change_amount": info["change_amount"],
            "volume": info["volume"], "high": info["high"], "low": info["low"],
            "open_price": info["open_price"],
            "market_cap": info["market_cap"],
            "per": info["per"], "pbr": info["pbr"],
            "eps": info["eps"], "bps": info["bps"],
            "roe": info["roe"], "ev": info["ev"],
            "high_52w": info["high_52w"], "low_52w": info["low_52w"],
            "foreign_ratio": info["foreign_ratio"],
            "credit_ratio": info["credit_ratio"],
            "timestamp": info["timestamp"],
        }
        await redis_cl.set(f"price:{info['code']}", json.dumps(redis_data, ensure_ascii=False), ex=600)

    if pg_pool:
        try:
            async with pg_pool.acquire() as c:
                await c.execute("""
                    INSERT INTO stock_prices (
                        stock_code, stock_name, price, change_pct, change_amount,
                        volume, high, low, open_price, market_cap,
                        per, pbr, eps, bps, roe, ev,
                        high_52w, low_52w, foreign_ratio, credit_ratio, float_shares, collected_at)
                    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22)
                """, info["code"], info["name"], info["price"], info["change_pct"], info["change_amount"],
                    info["volume"], info["high"], info["low"], info["open_price"], info["market_cap"],
                    info["per"], info["pbr"], info["eps"], info["bps"], info["roe"], info["ev"],
                    info["high_52w"], info["low_52w"], info["foreign_ratio"], info["credit_ratio"],
                    info["float_shares"], datetime.now())
        except Exception as e:
            logger.debug("save_price.err", code=info["code"], error=str(e))

async def save_ohlcv(code: str, rows: list):
    if not rows or not pg_pool:
        return
    # Redis에 최근 60일 일봉 저장 (ta-engine 보조)
    if redis_cl:
        await redis_cl.set(f"ohlcv:{code}", json.dumps(rows[:60], ensure_ascii=False), ex=86400)
    async with pg_pool.acquire() as c:
        for r in rows:
            try:
                dt = datetime.strptime(r["dt"], "%Y%m%d").date()
                await c.execute("""
                    INSERT INTO stock_ohlcv (
                        stock_code, dt, open_price, high_price, low_price, close_price,
                        volume, trade_amount, foreign_ratio, foreign_net, institution_net, individual_net)
                    VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)
                    ON CONFLICT (stock_code, dt) DO UPDATE SET
                        close_price=$6, volume=$7, foreign_ratio=$9,
                        foreign_net=$10, institution_net=$11, individual_net=$12
                """, code, dt, r["open"], r["high"], r["low"], r["close"],
                    r["volume"], r["trade_amount"], r["foreign_ratio"],
                    r["foreign_net"], r["institution_net"], r["individual_net"])
            except Exception:
                pass

async def save_foreign_flow(code: str, rows: list):
    if not rows or not pg_pool:
        return
    if redis_cl and rows:
        await redis_cl.set(f"foreign:{code}", json.dumps(rows[:20], ensure_ascii=False), ex=86400)
    async with pg_pool.acquire() as c:
        for r in rows:
            try:
                dt = datetime.strptime(r["dt"], "%Y%m%d").date()
                await c.execute("""
                    INSERT INTO stock_foreign_flow (
                        stock_code, dt, close_price, change_qty, hold_qty, hold_ratio, limit_ratio)
                    VALUES ($1,$2,$3,$4,$5,$6,$7)
                    ON CONFLICT (stock_code, dt) DO UPDATE SET
                        change_qty=$4, hold_qty=$5, hold_ratio=$6, limit_ratio=$7
                """, code, dt, r["close_price"], r["change_qty"],
                    r["hold_qty"], r["hold_ratio"], r["limit_ratio"])
            except Exception:
                pass

async def save_short_sale(code: str, rows: list):
    if not rows or not pg_pool:
        return
    if redis_cl and rows:
        await redis_cl.set(f"short:{code}", json.dumps(rows[:20], ensure_ascii=False), ex=86400)
    async with pg_pool.acquire() as c:
        for r in rows:
            try:
                dt = datetime.strptime(r["dt"], "%Y%m%d").date()
                await c.execute("""
                    INSERT INTO stock_short_sale (
                        stock_code, dt, close_price, short_qty, short_balance_qty,
                        trade_weight, short_avg_price)
                    VALUES ($1,$2,$3,$4,$5,$6,$7)
                    ON CONFLICT (stock_code, dt) DO UPDATE SET
                        short_qty=$4, short_balance_qty=$5, trade_weight=$6, short_avg_price=$7
                """, code, dt, r["close_price"], r["short_qty"],
                    r["short_balance_qty"], r["trade_weight"], r["short_avg_price"])
            except Exception:
                pass

# ── 수집 작업 ─────────────────────────────────────────────

class Stats:
    collected = 0; errors = 0; last_run = ""; last_full_run = ""

stats = Stats()

_job_price_running = False

async def job_price():
    """평일 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
    _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).
    일별은 days=10(경량, ON CONFLICT 갱신). 백필은 /collect/full?days=400."""
    codes = await get_stock_codes(0)
    if not codes:
        return
    sem = asyncio.Semaphore(5)  # 야간 수집 느리게
    ok = 0
    async with httpx.AsyncClient() as client:
        for code in codes:
            try:
                ohlcv, foreign, short = await asyncio.gather(
                    fetch_ohlcv(client, code, sem, days),
                    fetch_foreign_flow(client, code, sem),
                    fetch_short_sale(client, code, sem),
                    return_exceptions=True
                )
                if isinstance(ohlcv, list):
                    await save_ohlcv(code, ohlcv)
                if isinstance(foreign, list):
                    await save_foreign_flow(code, foreign)
                if isinstance(short, list):
                    await save_short_sale(code, short)
                ok += 1
            except Exception as e:
                logger.debug("job_full.err", code=code, error=str(e))
    stats.last_full_run = datetime.now().isoformat()
    logger.info("job_full.done", ok=ok, total=len(codes))

# ── 시그널 생성 ───────────────────────────────────────────

def calc_signal(info: dict, news_sc: float, dart_sc: float,
                foreign_net_5d: int, short_weight: float) -> Optional[dict]:
    price = info["price"]
    if price <= 0:
        return None
    h52 = info.get("high_52w", 0)
    l52 = info.get("low_52w", 0)
    cpct = info.get("change_pct", 0)
    foreign_ratio = info.get("foreign_ratio", 0)

    # 외국인 수급 보너스 (5일 누적 순매수 양수면 +10, 음수면 -10)
    foreign_bonus = min(10, max(-10, foreign_net_5d / 500_000)) if foreign_net_5d else 0
    # 공매도 패널티 (비중 5% 이상이면 -5)
    short_penalty = -5 if short_weight >= 5 else 0

    total = (news_sc * 0.35 + dart_sc * 0.25
             + (cpct * 10) * 0.25 + foreign_bonus * 0.10 + short_penalty * 0.05)
    total = max(-100, min(100, total))
    pos = (price - l52) / (h52 - l52) if h52 > l52 > 0 else 0.5

    if total >= 30:
        sig = "매수"
        tp = int(price + (h52 - price) * (0.3 + min(total, 100) / 200)) if h52 > price else int(price * 1.10)
        sl = int(price * (0.92 + pos * 0.03))
        conf = min(95, 50 + total * 0.3 + (1 - pos) * 20)
    elif total <= -30:
        sig = "매도"
        tp = int(price - (price - l52) * (0.3 + min(abs(total), 100) / 200)) if l52 > 0 and l52 < price else int(price * 0.90)
        sl = int(price * 1.05)
        conf = min(95, 50 + abs(total) * 0.3 + pos * 20)
    else:
        return None

    er = ((tp - price) / price) * 100 if sig == "매수" else ((price - tp) / price) * 100
    risk = abs(sl - price)
    reward = abs(tp - price)
    rr = reward / risk if risk > 0 else 0
    return {
        "signal_type": sig, "current_price": price, "target_price": tp, "stop_loss": sl,
        "expected_return_pct": round(er, 2), "risk_reward_ratio": round(rr, 2),
        "confidence": round(conf, 1), "price_momentum": round(cpct, 2),
        "news_score": round(news_sc, 1), "dart_score": round(dart_sc, 1),
        "foreign_net_5d": foreign_net_5d, "short_weight": short_weight,
    }

async def gen_signals():
    if not pg_pool:
        return
    async with pg_pool.acquire() as c:
        scores = await c.fetch("""
            SELECT stock_code, stock_name, news_score, dart_score, total_score
            FROM stock_scores
            WHERE score_date = (SELECT MAX(score_date) FROM stock_scores)
              AND (total_score >= 30 OR total_score <= -30)
        """)
        n = 0
        for row in scores:
            code = row["stock_code"]
            if not code or len(code) != 6:
                continue
            if not redis_cl:
                continue
            cached = await redis_cl.get(f"price:{code}")
            if not cached:
                continue
            info = json.loads(cached)

            # 외국인 5일 순매수 합산
            foreign_net_5d = 0
            f_cached = await redis_cl.get(f"foreign:{code}")
            if f_cached:
                f_rows = json.loads(f_cached)
                foreign_net_5d = sum(r.get("change_qty", 0) for r in f_rows[:5])

            # 공매도 최근 비중
            short_weight = 0.0
            s_cached = await redis_cl.get(f"short:{code}")
            if s_cached:
                s_rows = json.loads(s_cached)
                if s_rows:
                    short_weight = s_rows[0].get("trade_weight", 0.0)

            sig = calc_signal(info, row["news_score"], row["dart_score"],
                              foreign_net_5d, short_weight)
            if not sig:
                continue
            reasons = await c.fetch("""
                SELECT reason FROM news_analysis
                WHERE primary_stock=$1 AND intensity>=2
                ORDER BY analyzed_at DESC LIMIT 2
            """, code)
            reason = " | ".join([r["reason"][:100] for r in reasons])
            await c.execute("""
                INSERT INTO trade_signals (
                    stock_code, stock_name, signal_type, current_price,
                    target_price, stop_loss, expected_return_pct, risk_reward_ratio,
                    confidence, reason, news_score, dart_score, price_momentum,
                    foreign_net_5d, short_weight)
                VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15)
                ON CONFLICT (stock_code, (created_at::date)) DO UPDATE SET
                    signal_type=EXCLUDED.signal_type, current_price=EXCLUDED.current_price,
                    target_price=EXCLUDED.target_price, stop_loss=EXCLUDED.stop_loss,
                    expected_return_pct=EXCLUDED.expected_return_pct,
                    risk_reward_ratio=EXCLUDED.risk_reward_ratio,
                    confidence=EXCLUDED.confidence, reason=EXCLUDED.reason,
                    news_score=EXCLUDED.news_score, dart_score=EXCLUDED.dart_score,
                    price_momentum=EXCLUDED.price_momentum,
                    foreign_net_5d=EXCLUDED.foreign_net_5d, short_weight=EXCLUDED.short_weight
            """, code, row["stock_name"], sig["signal_type"], sig["current_price"],
                sig["target_price"], sig["stop_loss"], sig["expected_return_pct"],
                sig["risk_reward_ratio"], sig["confidence"], reason,
                sig["news_score"], sig["dart_score"], sig["price_momentum"],
                sig["foreign_net_5d"], sig["short_weight"])
            n += 1
    logger.info("signals.done", count=n)

# ── FastAPI ───────────────────────────────────────────────

app = FastAPI(title="키움증권 주가·수급·공매도")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

@app.on_event("startup")
async def startup():
    global pg_pool, redis_cl
    pg_pool = await asyncpg.create_pool(
        host=PG_HOST, port=PG_PORT, database=PG_DB,
        user=PG_USER, password=PG_PASS, min_size=2, max_size=5)
    redis_cl = aioredis.Redis(
        host=REDIS_HOST, port=6379, password=REDIS_PASSWORD, db=3, decode_responses=True)
    await init_db()
    # 현재가: 평일 9~16시 2분마다 (전체)
    scheduler.add_job(job_price, "cron", day_of_week="mon-fri",
                      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)
    scheduler.start()
    logger.info("kiwoom-api.started")

@app.on_event("shutdown")
async def shutdown():
    scheduler.shutdown()
    if pg_pool:
        await pg_pool.close()
    if redis_cl:
        await redis_cl.aclose()

@app.get("/health")
async def health():
    lu = await redis_cl.get("prices:last_update") if redis_cl else ""
    return JSONResponse(content={
        "status": "ok", "collected": stats.collected, "errors": stats.errors,
        "last_run": stats.last_run, "last_full_run": stats.last_full_run,
        "last_update": lu or ""
    })

@app.get("/price/{code}")
async def price(code: str):
    if redis_cl:
        c = await redis_cl.get(f"price:{code}")
        if c:
            return JSONResponse(content=json.loads(c))
    # 실시간 단건 조회
    sem = asyncio.Semaphore(1)
    async with httpx.AsyncClient() as client:
        info = await fetch_basic_info(client, code, sem)
        if info:
            await save_price(info)
            return JSONResponse(content=info)
    return JSONResponse(content={"error": "not found"}, status_code=404)

@app.get("/prices")
async def prices(limit: int = Query(default=50)):
    if not redis_cl:
        return JSONResponse(content={"data": {}})
    keys = await redis_cl.keys("price:*")
    result = {}
    for k in keys[:limit]:
        c = await redis_cl.get(k)
        if c:
            d = json.loads(c)
            result[d["code"]] = d
    return JSONResponse(content={"count": len(result), "data": result})

@app.get("/foreign/{code}")
async def foreign(code: str):
    if redis_cl:
        c = await redis_cl.get(f"foreign:{code}")
        if c:
            return JSONResponse(content={"code": code, "data": json.loads(c)})
    sem = asyncio.Semaphore(1)
    async with httpx.AsyncClient() as client:
        rows = await fetch_foreign_flow(client, code, sem)
        await save_foreign_flow(code, rows)
        return JSONResponse(content={"code": code, "data": rows})

@app.get("/short/{code}")
async def short_sale(code: str):
    if redis_cl:
        c = await redis_cl.get(f"short:{code}")
        if c:
            return JSONResponse(content={"code": code, "data": json.loads(c)})
    sem = asyncio.Semaphore(1)
    async with httpx.AsyncClient() as client:
        rows = await fetch_short_sale(client, code, sem)
        await save_short_sale(code, rows)
        return JSONResponse(content={"code": code, "data": rows})

@app.get("/ohlcv/{code}")
async def ohlcv(code: str, days: int = Query(default=60)):
    if redis_cl:
        c = await redis_cl.get(f"ohlcv:{code}")
        if c:
            data = json.loads(c)
            return JSONResponse(content={"code": code, "data": data[:days]})
    sem = asyncio.Semaphore(1)
    async with httpx.AsyncClient() as client:
        rows = await fetch_ohlcv(client, code, sem, days)
        await save_ohlcv(code, rows)
        return JSONResponse(content={"code": code, "data": rows})

@app.get("/signals")
async def signals(days: int = Query(default=7)):
    async with pg_pool.acquire() as c:
        rows = await c.fetch("""
            SELECT * FROM trade_signals
            WHERE created_at >= NOW() - INTERVAL '%s days'
            ORDER BY confidence DESC LIMIT 50
        """ % days)
    return [dict(r) for r in rows]

@app.get("/signals/{code}")
async def stock_signals(code: str):
    async with pg_pool.acquire() as c:
        rows = await c.fetch("""
            SELECT * FROM trade_signals WHERE stock_code=$1
            ORDER BY created_at DESC LIMIT 10
        """, code)
    return [dict(r) for r in rows]

@app.get("/summary/{code}")
async def summary(code: str):
    """종목 종합 요약 (현재가 + 외국인 + 공매도 + 최근 신호)"""
    result = {"code": code}
    if redis_cl:
        for key in [f"price:{code}", f"foreign:{code}", f"short:{code}", f"ohlcv:{code}"]:
            c = await redis_cl.get(key)
            if c:
                field = key.split(":")[0]
                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())
    return {"status": "started", "job": "price"}

@app.post("/collect/full")
async def collect_full(days: int = Query(default=10, ge=1, le=600)):
    asyncio.create_task(job_full(days))
    return {"status": "started", "job": "full", "days": days}

# ── 추가: 분봉 / 호가 / 거래량급증 ─────────────────────────

@app.get("/minute/{code}")
async def minute_chart(code: str, scope: str = Query(default="5", description="분봉 단위: 1,3,5,10,15,30,60")):
    """ka10080: 분봉차트 (실시간 호출)"""
    async with httpx.AsyncClient() as c:
        data = await fetch_minute_chart(c, code, scope)
    return JSONResponse(content={"code": code, "scope": scope, "data": data})

@app.get("/orderbook/{code}")
async def orderbook(code: str):
    """ka10004: 10단계 호가 잔량"""
    async with httpx.AsyncClient() as c:
        data = await fetch_orderbook(c, code)
    return JSONResponse(content=data or {"code": code, "ask": [], "bid": []})

@app.get("/volume-surge")
async def volume_surge():
    """ka10023: 거래량 급증 종목 TOP50 (Redis 5분 캐시)"""
    cached = None
    if redis_cl:
        try:
            v = await redis_cl.get("vol_surge:list")
            if v: cached = json.loads(v)
        except: pass
    if cached:
        return JSONResponse(content={"data": cached, "cached": True})
    data = await fetch_volume_surge()
    if redis_cl and data:
        try: await redis_cl.setex("vol_surge:list", 300, json.dumps(data))
        except: pass
    return JSONResponse(content={"data": data, "cached": False})