"""
기술적 분석 엔진 (Technical Analysis Engine)
- 네이버 금융 차트 API (1차) / yfinance (2차 백업) OHLCV 수집
- MA5/20/60/120, RSI(14), MACD(12,26,9), 볼린저밴드(20,2), 스토캐스틱(14,3)
- 기술적 점수 (-100~100) 산출
- 매수/매도 목표가 T1/T2/T3 + 손절가 자동 계산
- vLLM AI 문장형 판단 생성
- 보유 포지션 손익 + 맞춤 전략 분석
"""
import asyncio, json, os, re, math
from datetime import datetime
from typing import Optional, List, Dict
import asyncpg, httpx, redis.asyncio as aioredis, structlog
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from fastapi import FastAPI, Query, Body
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

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", "")
OLLAMA_URL     = os.getenv("OLLAMA_URL", "http://ollama:11434")
HEADERS        = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}

pg_pool: Optional[asyncpg.Pool] = None
redis_cl: Optional[aioredis.Redis] = None
scheduler = AsyncIOScheduler(timezone="Asia/Seoul")

class Stats:
    analyzed = 0; errors = 0; last_run = ""
stats = Stats()

# ── 기술적 지표 계산 ──────────────────────────────────────

def _ema_series(values: List[float], period: int) -> List[float]:
    if not values or len(values) < period:
        return [values[-1]] * len(values) if values else []
    k = 2.0 / (period + 1)
    seed = sum(values[:period]) / period
    out = [seed]
    for v in values[period:]:
        out.append(v * k + out[-1] * (1 - k))
    return [out[0]] * (period - 1) + out

def _ma(closes: List[float], n: int) -> float:
    if not closes: return 0.0
    data = closes[-n:] if len(closes) >= n else closes
    return sum(data) / len(data)

def _rsi(closes: List[float], period: int = 14) -> float:
    if len(closes) < period + 1:
        return 50.0
    deltas = [closes[i] - closes[i-1] for i in range(1, len(closes))]
    gains = [max(d, 0.0) for d in deltas]
    losses = [max(-d, 0.0) for d in deltas]
    ag = sum(gains[:period]) / period
    al = sum(losses[:period]) / period
    for i in range(period, len(gains)):
        ag = (ag * (period - 1) + gains[i]) / period
        al = (al * (period - 1) + losses[i]) / period
    if al == 0:
        return 100.0
    return round(100 - 100 / (1 + ag / al), 2)

def _macd(closes: List[float]) -> tuple:
    if len(closes) < 26:
        return 0.0, 0.0, 0.0
    e12 = _ema_series(closes, 12)
    e26 = _ema_series(closes, 26)
    macd_line = [a - b for a, b in zip(e12, e26)]
    # 시그널선 = 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)

def _bollinger(closes: List[float], period: int = 20) -> tuple:
    if len(closes) < period:
        c = closes[-1] if closes else 0
        return float(c), float(c), float(c), 0.5
    recent = closes[-period:]
    ma = sum(recent) / period
    std = math.sqrt(sum((x - ma) ** 2 for x in recent) / period)
    upper = ma + 2 * std
    lower = ma - 2 * std
    cur = closes[-1]
    pct_b = (cur - lower) / (upper - lower) if upper != lower else 0.5
    return round(upper), round(ma), round(lower), round(pct_b, 3)

def _stochastic(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> tuple:
    if len(closes) < period:
        return 50.0, 50.0
    h = max(highs[-period:])
    l = min(lows[-period:])
    k = ((closes[-1] - l) / (h - l) * 100) if h != l else 50.0
    ks = []
    for i in range(3):
        idx = -(3 - i)
        hh = max(highs[idx - period + 1:idx + 1] if idx != -1 else highs[-period:])
        ll = min(lows[idx - period + 1:idx + 1] if idx != -1 else lows[-period:])
        ks.append(((closes[idx] - ll) / (hh - ll) * 100) if hh != ll else 50.0)
    d = sum(ks) / len(ks)
    return round(k, 2), round(d, 2)

def _vol_ratio(volumes: List[float], period: int = 20) -> float:
    if len(volumes) < period + 1:
        return 1.0
    avg = sum(volumes[-period - 1:-1]) / period
    return round(volumes[-1] / avg, 2) if avg > 0 else 1.0

def _atr(highs: List[float], lows: List[float], closes: List[float], period: int = 14) -> float:
    """Average True Range — 변동성 측정"""
    if len(closes) < period + 1:
        return 0.0
    trs = []
    for i in range(1, len(closes)):
        tr = max(
            highs[i] - lows[i],
            abs(highs[i] - closes[i-1]),
            abs(lows[i]  - closes[i-1]),
        )
        trs.append(tr)
    recent = trs[-period:]
    return round(sum(recent) / period, 2) if recent else 0.0


def _obv(closes: List[float], volumes: List[float]) -> tuple:
    """On-Balance Volume — 거래량 누적, 가격 상승일 +volume, 하락일 -volume"""
    if len(closes) < 2: return 0, 0
    obv = 0.0
    obvs = [0.0]
    for i in range(1, len(closes)):
        if closes[i] > closes[i-1]: obv += volumes[i]
        elif closes[i] < closes[i-1]: obv -= volumes[i]
        obvs.append(obv)
    # 최근 OBV 추세 (20일 평균 대비)
    recent = obvs[-20:] if len(obvs) >= 20 else obvs
    avg = sum(recent) / len(recent) if recent else 0
    obv_trend = "상승" if obv > avg * 1.05 else ("하락" if obv < avg * 0.95 else "중립")
    return int(obv), obv_trend


def _vwap(highs: List[float], lows: List[float], closes: List[float], volumes: List[float],
          period: int = 20) -> float:
    """Volume Weighted Average Price — 거래량 가중 평균가, 기관 매매 기준선"""
    if len(closes) < period: return float(closes[-1]) if closes else 0
    cum_vp = sum(((highs[i] + lows[i] + closes[i]) / 3) * volumes[i]
                 for i in range(-period, 0))
    cum_v = sum(volumes[-period:])
    return round(cum_vp / cum_v, 2) if cum_v > 0 else 0


def _ichimoku(highs: List[float], lows: List[float], closes: List[float]) -> dict:
    """일목균형표 (Ichimoku Kinko Hyo) — 5개 라인"""
    if len(closes) < 52: return {}
    # 전환선(Tenkan-sen): (9일 고가 + 9일 저가) / 2
    tenkan = (max(highs[-9:]) + min(lows[-9:])) / 2
    # 기준선(Kijun-sen): (26일 고가 + 26일 저가) / 2
    kijun  = (max(highs[-26:]) + min(lows[-26:])) / 2
    # 선행스팬1(Senkou Span A): (전환+기준)/2, 26일 후
    span_a = (tenkan + kijun) / 2
    # 선행스팬2(Senkou Span B): (52일 고가 + 52일 저가)/2, 26일 후
    span_b = (max(highs[-52:]) + min(lows[-52:])) / 2
    # 후행스팬(Chikou Span): 종가, 26일 전
    chikou = closes[-26] if len(closes) > 26 else closes[-1]
    cur = closes[-1]
    cloud_top = max(span_a, span_b)
    cloud_bot = min(span_a, span_b)
    pos = "구름위" if cur > cloud_top else ("구름아래" if cur < cloud_bot else "구름속")
    return {
        "tenkan": int(tenkan), "kijun": int(kijun),
        "span_a": int(span_a), "span_b": int(span_b), "chikou": int(chikou),
        "cloud_pos": pos,
    }


def calc_indicators(ohlcv: List[dict]) -> dict:
    if len(ohlcv) < 5:
        return {}
    closes  = [float(d["close"])  for d in ohlcv]
    highs   = [float(d["high"])   for d in ohlcv]
    lows    = [float(d["low"])    for d in ohlcv]
    volumes = [float(d["volume"]) for d in ohlcv]

    bb_upper, bb_mid, bb_lower, pct_b = _bollinger(closes, 20)
    stoch_k, stoch_d                  = _stochastic(highs, lows, closes, 14)
    macd, macd_signal, macd_hist      = _macd(closes)
    atr14                             = _atr(highs, lows, closes, 14)
    obv_val, obv_trend                = _obv(closes, volumes)
    vwap_val                          = _vwap(highs, lows, closes, volumes, 20)
    ichi                              = _ichimoku(highs, lows, closes)

    return {
        "price":       int(closes[-1]),
        "ma5":         round(_ma(closes, 5)),
        "ma20":        round(_ma(closes, 20)),
        "ma60":        round(_ma(closes, 60)),
        "ma120":       round(_ma(closes, 120)),
        "rsi":         _rsi(closes, 14),
        "macd":        macd,
        "macd_signal": macd_signal,
        "macd_hist":   macd_hist,
        "bb_upper":    bb_upper,
        "bb_mid":      bb_mid,
        "bb_lower":    bb_lower,
        "pct_b":       pct_b,
        "stoch_k":     stoch_k,
        "stoch_d":     stoch_d,
        "vol_ratio":   _vol_ratio(volumes, 20),
        "atr14":       atr14,
        "high_52w":    int(max(highs[-min(len(highs), 252):])),
        "low_52w":     int(min(lows[-min(len(lows), 252):])),
        "obv":         obv_val,
        "obv_trend":   obv_trend,
        "vwap20":      vwap_val,
        "ichimoku":    ichi,
    }

def calc_tech_score(ind: dict) -> tuple:
    """기술적 점수 (-100~100)와 근거 신호 목록 반환"""
    if not ind:
        return 0.0, []

    price = ind["price"]
    score = 0.0
    signals: List[str] = []

    # ── 이동평균 (±40) ──────────────────────────
    if ind["ma5"] > ind["ma20"]:
        score += 10; signals.append("MA5>MA20 단기상승")
    else:
        score -= 10; signals.append("MA5<MA20 단기하락")

    if ind["ma20"] > ind["ma60"]:
        score += 8; signals.append("MA20>MA60 중기상승")
    else:
        score -= 8

    if ind["ma60"] > ind["ma120"]:
        score += 7
    else:
        score -= 7

    if price > ind["ma20"]:
        score += 8; signals.append("현재가 MA20 위")
    elif price < ind["ma60"]:
        score -= 8; signals.append("현재가 MA60 아래")

    # 정배열/역배열
    if ind["ma5"] > ind["ma20"] > ind["ma60"] > ind["ma120"]:
        score += 7; signals.append("정배열")
    elif ind["ma5"] < ind["ma20"] < ind["ma60"] < ind["ma120"]:
        score -= 7; signals.append("역배열")

    # ── RSI (±25) ───────────────────────────────
    rsi = ind["rsi"]
    if rsi <= 30:
        score += 25; signals.append(f"RSI 과매도({rsi:.0f})")
    elif rsi <= 40:
        score += 15; signals.append(f"RSI 저점({rsi:.0f})")
    elif rsi <= 60:
        score += 5
    elif rsi <= 70:
        score -= 5
    else:
        score -= 20; signals.append(f"RSI 과매수({rsi:.0f})")

    # ── MACD (±20) ──────────────────────────────
    if ind["macd_hist"] > 0 and ind["macd"] > ind["macd_signal"]:
        score += 20; signals.append("MACD 골든크로스")
    elif ind["macd_hist"] > 0:
        score += 8
    elif ind["macd_hist"] < 0 and ind["macd"] < ind["macd_signal"]:
        score -= 20; signals.append("MACD 데드크로스")
    else:
        score -= 5

    # ── 볼린저밴드 (±15) ────────────────────────
    pb = ind["pct_b"]
    if pb < 0.1:
        score += 15; signals.append("볼밴 하단(과매도)")
    elif pb < 0.3:
        score += 8
    elif pb > 0.9:
        score -= 15; signals.append("볼밴 상단(과매수)")
    elif pb > 0.7:
        score -= 5

    # ── 스토캐스틱 (±10) ────────────────────────
    sk, sd = ind["stoch_k"], ind["stoch_d"]
    if sk < 20 and sk > sd:
        score += 10; signals.append("스토캐스틱 바닥반등")
    elif sk > 80 and sk < sd:
        score -= 10; signals.append("스토캐스틱 고점하락")

    # ── 거래량 보너스 (±5) ──────────────────────
    if ind["vol_ratio"] > 1.5:
        if score > 0:
            score += 5; signals.append("거래량 급증(매수세)")
        else:
            score -= 5; signals.append("거래량 급증(매도세)")

    return round(max(-100.0, min(100.0, score)), 1), signals

def calc_price_targets(price: int, ind: dict, sig: str) -> dict:
    """매수/매도 목표가(T1/T2/T3) + 손절가 계산 (10원 단위 반올림)"""
    if not ind or price <= 0:
        return {}

    def r10(p): return int(round(p / 10) * 10)

    h52    = ind.get("high_52w", price * 1.3)
    l52    = ind.get("low_52w",  price * 0.7)
    bb_up  = ind.get("bb_upper", price * 1.05)
    bb_dn  = ind.get("bb_lower", price * 0.95)
    ma20   = ind.get("ma20", price)
    ma60   = ind.get("ma60", price)

    if sig == "매수":
        # 진입: 현재가 기준 -2% (기술지표 확인 후 매수)
        entry = r10(price * 0.98)
        # T1: +7% (단기), T2: +14% (중기), T3: min(+22%, 52주고가 -3%)
        t1   = r10(price * 1.07)
        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%, -4%] 밴드로 제한해 항상 진입가 아래 유지.
        # (하락추세로 price<MA60이면 MA60*0.95가 현재가 위로 올라가 손절가>진입가가 되어
        #  RR이 역전·즉시 손절되는 버그 차단)
        raw_stop = max(price * 0.92, ma60 * 0.95)
        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 아래)
        atr = ind.get("atr14", 0)
        atr_trailing = r10(price - 2 * atr) if atr > 0 else stop
        return {
            "entry_price": entry,
            "t1": t1, "t1_pct": er1, "t1_sell_pct": 50,
            "t2": t2, "t2_pct": round((t2 - price) / price * 100, 1), "t2_sell_pct": 30,
            "t3": t3, "t3_pct": round((t3 - price) / price * 100, 1), "t3_sell_pct": 20,
            "stop_loss": stop, "stop_pct": -sl_r,
            "atr14": atr,
            "trailing_stop": atr_trailing,
            "risk_reward": round(er1 / sl_r, 2) if sl_r > 0 else 0,
            "exit_strategy": "T1 50% + T2 30% + T3 20% 분할매도, 손절 또는 trailing(ATR×2) 도달시 전량",
        }
    else:  # 매도 / 관망(음수)
        entry = r10(price * 1.02)
        t1   = r10(price * 0.93)
        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(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 {
            "entry_price": entry,
            "t1": t1, "t1_pct": -er1,
            "t2": t2, "t2_pct": -round((price - t2) / price * 100, 1),
            "t3": t3, "t3_pct": -round((price - t3) / price * 100, 1),
            "stop_loss": stop, "stop_pct": sl_r,
            "risk_reward": round(er1 / sl_r, 2) if sl_r > 0 else 0,
        }

# ── OHLCV 수집 (네이버 차트 → yfinance 백업) ─────────────

async def get_ohlcv_naver_chart(client: httpx.AsyncClient, code: str, count: int = 120) -> List[dict]:
    """네이버 차트 API (fchart)"""
    try:
        r = await client.get(
            f"https://fchart.stock.naver.com/sise.nhn?symbol={code}&timeframe=day&count={count}&requestType=0",
            headers=HEADERS, timeout=10)
        items = re.findall(r'data="([^"]+)"', r.text)
        data = []
        for item in items:
            p = item.split("|")
            if len(p) >= 6 and all(x.strip() for x in p[:5]):
                try:
                    data.append({
                        "date": p[0], "open": int(p[1]), "high": int(p[2]),
                        "low": int(p[3]), "close": int(p[4]),
                        "volume": int(p[5]) if p[5].strip() else 0,
                    })
                except ValueError:
                    pass
        return data
    except Exception:
        return []

async def get_ohlcv_naver_sise(client: httpx.AsyncClient, code: str, pages: int = 7) -> List[dict]:
    """네이버 일별시세 페이지 (차트 API 실패 시 2차)"""
    data = []
    try:
        for page in range(1, pages + 1):
            r = await client.get(
                f"https://finance.naver.com/item/sise_day.naver?code={code}&page={page}",
                headers=HEADERS, timeout=12)
            r.encoding = "euc-kr"
            # 날짜+종가+전일비+시가+고가+저가+거래량 패턴
            rows = re.findall(
                r'(\d{4}\.\d{2}\.\d{2})[^<]*</span>.*?'
                r'<span[^>]*>([\d,]+)</span>.*?'   # 종가
                r'(?:.*?){3}'
                r'<span[^>]*>([\d,]+)</span>.*?'   # 시가
                r'<span[^>]*>([\d,]+)</span>.*?'   # 고가
                r'<span[^>]*>([\d,]+)</span>.*?'   # 저가
                r'<span[^>]*>([\d,]+)</span>',     # 거래량
                r.text, re.DOTALL)
            if not rows:
                # 단순 종가만 추출 (더 넓은 패턴)
                simple = re.findall(
                    r'class="tah p10 gray03">(\d{4}\.\d{2}\.\d{2})<.*?'
                    r'class="tah p11">([\d,]+)<',
                    r.text, re.DOTALL)
                for date_str, close_str in simple:
                    close = int(close_str.replace(",", ""))
                    if close > 0:
                        data.append({"date": date_str.replace(".", ""),
                                     "open": close, "high": close,
                                     "low": close, "close": close, "volume": 0})
            else:
                for m in rows:
                    date_str = m[0].replace(".", "")
                    close = int(m[1].replace(",", ""))
                    open_ = int(m[2].replace(",", "")) if m[2] else close
                    high  = int(m[3].replace(",", "")) if m[3] else close
                    low   = int(m[4].replace(",", "")) if m[4] else close
                    vol   = int(m[5].replace(",", "")) if m[5] else 0
                    if close > 0:
                        data.append({"date": date_str, "open": open_,
                                     "high": high, "low": low,
                                     "close": close, "volume": vol})
            if len(data) >= 120:
                break
            await asyncio.sleep(0.15)
    except Exception as e:
        logger.warning("ohlcv.sise.err", code=code, error=str(e))
    return data[:120]

async def get_ohlcv_yfinance(code: str) -> List[dict]:
    """yfinance 최종 백업"""
    try:
        import yfinance as yf
        loop = asyncio.get_event_loop()
        def _fetch():
            t = yf.Ticker(f"{code}.KS")
            return t.history(period="1y")
        h = await loop.run_in_executor(None, _fetch)
        if h.empty:
            return []
        return [{"date": idx.strftime("%Y%m%d"),
                 "open": int(row["Open"] or 0), "high": int(row["High"] or 0),
                 "low": int(row["Low"] or 0), "close": int(row["Close"] or 0),
                 "volume": int(row["Volume"] or 0)}
                for idx, row in h.iterrows() if row["Close"] and row["High"]]
    except Exception as e:
        logger.warning("ohlcv.yfinance.err", code=code, error=str(e))
        return []

async def get_ohlcv(client: httpx.AsyncClient, code: str, count: int = 120) -> List[dict]:
    data = await get_ohlcv_naver_chart(client, code, count)
    if len(data) < 20:
        data = await get_ohlcv_naver_sise(client, code)
    if len(data) < 20:
        logger.info("ohlcv.fallback.yfinance", code=code)
        await asyncio.sleep(0.5)  # rate limit 방지
        data = await get_ohlcv_yfinance(code)
    return data

# ── vLLM AI 판단문 생성 ───────────────────────────────────

async def generate_ai_opinion(client: httpx.AsyncClient, code: str, name: str,
                               ind: dict, tech_score: float, signals: List[str],
                               targets: dict, news_score: float = 0) -> str:
    """vLLM으로 문장형 투자 판단 생성"""
    sig = "매수" if tech_score >= 30 else ("매도" if tech_score <= -30 else "관망")
    price = ind.get("price", 0)
    rsi   = ind.get("rsi", 50)
    h52   = ind.get("high_52w", price)
    l52   = ind.get("low_52w",  price)
    pos52 = int((price - l52) / (h52 - l52) * 100) if h52 != l52 else 50

    prompt = f"""다음 주식 데이터를 바탕으로 투자자에게 명확한 매매 판단을 3~5문장으로 설명하세요.
한국어로, 구체적인 가격과 수치를 포함해서 작성하세요.

종목: {name}({code})
현재가: {price:,}원
기술점수: {tech_score}점 / 신호: {sig}
이동평균: MA5={ind.get('ma5',0):,} MA20={ind.get('ma20',0):,} MA60={ind.get('ma60',0):,}
RSI: {rsi} / MACD히스토그램: {'양수(골든)' if ind.get('macd_hist',0)>0 else '음수(데드)'}
볼린저%B: {ind.get('pct_b',0.5)*100:.0f}%
52주위치: 하단에서 {pos52}%
뉴스감성점수: {news_score:.0f}
기술신호: {', '.join(signals[:4])}
{f"1차목표가: {targets.get('t1',0):,}원 / 손절가: {targets.get('stop_loss',0):,}원" if targets else ""}

투자 판단 (3~5문장):"""

    try:
        r = await client.post(f"{OLLAMA_URL}/v1/chat/completions", json={
            "model": "exaone3.5:7.8b",
            "messages": [
                {"role": "system", "content": "당신은 한국 주식 전문 애널리스트입니다. 기술적 분석 데이터를 바탕으로 명확하고 실용적인 투자 의견을 제시합니다."},
                {"role": "user",   "content": prompt}
            ],
            "max_tokens": 300, "temperature": 0.2
        }, timeout=60)
        return r.json()["choices"][0]["message"]["content"].strip()
    except Exception as e:
        logger.warning("ai_opinion.err", code=code, error=str(e))
        return ""

# ── 포지션 손익 분석 ──────────────────────────────────────

class PositionRequest(BaseModel):
    code: str
    name: str = ""
    buy_price: int
    qty: int

def analyze_position(price: int, buy_price: int, qty: int,
                     ind: dict, tech_score: float) -> dict:
    """보유 포지션 기반 맞춤 전략 계산"""
    pnl      = (price - buy_price) * qty
    pnl_pct  = (price - buy_price) / buy_price * 100
    total_buy = buy_price * qty
    h52 = ind.get("high_52w", price * 1.3)
    l52 = ind.get("low_52w",  price * 0.7)
    ma20 = ind.get("ma20", price)
    ma60 = ind.get("ma60", price)
    bbu  = ind.get("bb_upper", price * 1.05)

    def r10(p): return int(round(p / 10) * 10)

    # 손절선: 매입가 -8% 또는 MA60 -3% 중 높은 것
    stop = max(r10(buy_price * 0.92), r10(ma60 * 0.97))

    # 목표가
    t1 = r10(max(buy_price * 1.08, bbu * 0.97))   # 본전+8% 또는 볼밴 상단
    t2 = r10(max((price + h52) / 2, buy_price * 1.15))
    t3 = r10(max(h52 * 0.97, buy_price * 1.25))

    # 추가매수 구간 (물타기) - 현재가 -5%, -10%
    avg_down1_price = r10(price * 0.95)
    avg_down1_qty   = max(1, qty // 3)
    avg_down1_avg   = (total_buy + avg_down1_price * avg_down1_qty) / (qty + avg_down1_qty)

    avg_down2_price = r10(price * 0.90)
    avg_down2_qty   = max(1, qty // 2)
    avg_down2_avg   = (total_buy + avg_down2_price * avg_down2_qty) / (qty + avg_down2_qty)

    return {
        "pnl":          pnl,
        "pnl_pct":      round(pnl_pct, 2),
        "total_buy":    total_buy,
        "current_value": price * qty,
        "stop_loss":    stop,
        "stop_pnl":     (stop - buy_price) * qty,
        "t1": t1, "t1_pnl": (t1 - buy_price) * qty,
        "t2": t2, "t2_pnl": (t2 - buy_price) * qty,
        "t3": t3, "t3_pnl": (t3 - buy_price) * qty,
        "avg_down": [
            {"price": avg_down1_price, "add_qty": avg_down1_qty,
             "new_avg": round(avg_down1_avg), "add_cost": avg_down1_price * avg_down1_qty},
            {"price": avg_down2_price, "add_qty": avg_down2_qty,
             "new_avg": round(avg_down2_avg), "add_cost": avg_down2_price * avg_down2_qty},
        ],
        "action": (
            "손절 고려" if price <= stop else
            "추가매수 검토" if pnl_pct < -5 and tech_score >= 0 else
            "홀드" if -5 <= pnl_pct < 5 else
            "1차 익절 고려" if pnl_pct >= 10 else "홀드"
        ),
    }

# ── 단일 종목 분석 ────────────────────────────────────────

EXCLUDE_KEYWORDS = (
    "기업인수목적", "선박투자회사", "부동산투자회사", "특별자산", "인프라투자",
    "사모투자", "맥쿼리", "리츠", "REITs",
)

async def analyze_stock(client: httpx.AsyncClient, code: str, name: str = "",
                        with_ai: bool = False, news_score: float = 0) -> Optional[dict]:
    # 이름이 없으면 DB에서 조회
    if not name and pg_pool:
        try:
            async with pg_pool.acquire() as conn:
                name = await conn.fetchval(
                    "SELECT corp_name FROM dart_corps WHERE stock_code=$1", code) or ""
        except: pass

    # SPAC·REITs·선박펀드 등 제외
    if any(kw in name for kw in EXCLUDE_KEYWORDS):
        return None

    ohlcv = await get_ohlcv(client, code, 120)
    if len(ohlcv) < 20:
        return None

    ind = calc_indicators(ohlcv)
    if not ind:
        return None

    # 주가 500원 미만 penny stock 제외
    if ind.get("price", 0) < 500:
        return None

    tech_score, signals = calc_tech_score(ind)
    sig = "매수" if tech_score >= 30 else ("매도" if tech_score <= -30 else "관망")
    # 관망도 목표가 계산 (기술점수 양수면 매수 기준, 음수면 매도 기준)
    tgt_sig = "매수" if tech_score >= 0 else "매도"
    targets = calc_price_targets(ind["price"], ind, tgt_sig)

    ai_opinion = ""
    if with_ai and sig != "관망":
        ai_opinion = await generate_ai_opinion(
            client, code, name, ind, tech_score, signals, targets, news_score)

    result = {
        "code": code, "name": name,
        "tech_score": tech_score, "signal": sig,
        "signals": signals, "indicators": ind, "targets": targets,
        "ai_opinion": ai_opinion,
        "analyzed_at": datetime.now().isoformat(),
    }

    if redis_cl:
        try:
            await redis_cl.set(f"ta:{code}", json.dumps(result, ensure_ascii=False), ex=1800)
        except: pass

    if pg_pool:
        try:
            async with pg_pool.acquire() as conn:
                await conn.execute("""
                    INSERT INTO stock_technical (
                        stock_code, stock_name, price,
                        ma5, ma20, ma60, ma120,
                        rsi, macd, macd_signal, macd_hist,
                        bb_upper, bb_mid, bb_lower, pct_b,
                        stoch_k, stoch_d, vol_ratio,
                        tech_score, signal, signals, targets, analyzed_at
                    ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23)
                    ON CONFLICT (stock_code) DO UPDATE SET
                        stock_name=$2, price=$3,
                        ma5=$4, ma20=$5, ma60=$6, ma120=$7,
                        rsi=$8, macd=$9, macd_signal=$10, macd_hist=$11,
                        bb_upper=$12, bb_mid=$13, bb_lower=$14, pct_b=$15,
                        stoch_k=$16, stoch_d=$17, vol_ratio=$18,
                        tech_score=$19, signal=$20, signals=$21, targets=$22, analyzed_at=$23
                """,
                    code, name, ind["price"],
                    ind["ma5"], ind["ma20"], ind["ma60"], ind["ma120"],
                    ind["rsi"], ind["macd"], ind["macd_signal"], ind["macd_hist"],
                    ind["bb_upper"], ind["bb_mid"], ind["bb_lower"], ind["pct_b"],
                    ind["stoch_k"], ind["stoch_d"], ind["vol_ratio"],
                    tech_score, sig,
                    json.dumps(signals, ensure_ascii=False),
                    json.dumps(targets, ensure_ascii=False),
                    datetime.now())
        except Exception as e:
            logger.warning("ta.db.err", code=code, error=str(e))

    return result

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

async def init_db():
    async with pg_pool.acquire() as conn:
        await conn.execute("""
            CREATE TABLE IF NOT EXISTS stock_technical (
                id          SERIAL PRIMARY KEY,
                stock_code  VARCHAR(10) UNIQUE NOT NULL,
                stock_name  VARCHAR(100) DEFAULT '',
                price       INTEGER DEFAULT 0,
                ma5         FLOAT DEFAULT 0,
                ma20        FLOAT DEFAULT 0,
                ma60        FLOAT DEFAULT 0,
                ma120       FLOAT DEFAULT 0,
                rsi         FLOAT DEFAULT 50,
                macd        FLOAT DEFAULT 0,
                macd_signal FLOAT DEFAULT 0,
                macd_hist   FLOAT DEFAULT 0,
                bb_upper    FLOAT DEFAULT 0,
                bb_mid      FLOAT DEFAULT 0,
                bb_lower    FLOAT DEFAULT 0,
                pct_b       FLOAT DEFAULT 0.5,
                stoch_k     FLOAT DEFAULT 50,
                stoch_d     FLOAT DEFAULT 50,
                vol_ratio   FLOAT DEFAULT 1,
                tech_score  FLOAT DEFAULT 0,
                signal      VARCHAR(10) DEFAULT '관망',
                signals     JSONB DEFAULT '[]'::jsonb,
                targets     JSONB DEFAULT '{}'::jsonb,
                analyzed_at TIMESTAMP DEFAULT NOW()
            )
        """)
        await conn.execute("CREATE INDEX IF NOT EXISTS idx_ta_score  ON stock_technical(tech_score DESC)")
        await conn.execute("CREATE INDEX IF NOT EXISTS idx_ta_signal ON stock_technical(signal)")
    logger.info("ta.db.initialized")

# ── 전체 분석 작업 ────────────────────────────────────────

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:
                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 Exception as e:
                logger.warning("ta.codes.err", error=str(e))

        if not codes:
            for sosok in [0, 1]:
                for page in range(1, 30):
                    try:
                        r = await client.get(
                            f"https://finance.naver.com/sise/sise_market_sum.naver?sosok={sosok}&page={page}",
                            headers=HEADERS, timeout=15)
                        r.encoding = "euc-kr"
                        found = re.findall(r'main\.naver\?code=(\d{6})[^>]*>([^<]+)</a>', r.text)
                        if not found: break
                        codes.extend([(c.strip(), n.strip()) for c, n in found])
                        await asyncio.sleep(0.2)
                    except: break
                    if len(codes) >= 500: break

        ok = 0
        for code, name in codes:
            if not code or len(code) != 6: continue
            try:
                result = await analyze_stock(client, code, name)
                if result: ok += 1
            except Exception as e:
                stats.errors += 1
                logger.warning("ta.analyze.err", code=code, error=str(e))
            await asyncio.sleep(0.2)

        stats.analyzed += ok
        stats.last_run = datetime.now().isoformat()
        logger.info("ta.job.done", analyzed=ok, requested=len(codes))

# ── 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=5, decode_responses=True)
    await init_db()
    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(_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")

@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():
    return {"status": "ok", "analyzed": stats.analyzed,
            "errors": stats.errors, "last_run": stats.last_run}

@app.get("/technical/{code}")
async def technical(code: str):
    if redis_cl:
        cached = await redis_cl.get(f"ta:{code}")
        if cached:
            return JSONResponse(content=json.loads(cached))
    if pg_pool:
        async with pg_pool.acquire() as conn:
            row = await conn.fetchrow("SELECT * FROM stock_technical WHERE stock_code=$1", code)
            if row:
                d = dict(row)
                d["analyzed_at"] = str(d["analyzed_at"])
                for k in ("signals", "targets"):
                    if isinstance(d[k], str):
                        d[k] = json.loads(d[k])
                return JSONResponse(content=d)
    # 실시간 분석
    async with httpx.AsyncClient() as client:
        result = await analyze_stock(client, code)
        if result:
            return JSONResponse(content=result)
    return JSONResponse(content={"error": "not found"}, status_code=404)

@app.get("/ranking")
async def ranking(limit: int = Query(default=30), signal: str = Query(default="")):
    async with pg_pool.acquire() as conn:
        if signal:
            rows = await conn.fetch(
                "SELECT * FROM stock_technical WHERE signal=$1 ORDER BY tech_score DESC LIMIT $2",
                signal, limit)
        else:
            rows = await conn.fetch(
                "SELECT * FROM stock_technical ORDER BY tech_score DESC LIMIT $1", limit)
    result = []
    for row in rows:
        d = dict(row)
        d["analyzed_at"] = str(d["analyzed_at"])
        for k in ("signals", "targets"):
            if isinstance(d[k], str):
                d[k] = json.loads(d[k])
        result.append(d)
    return result

@app.get("/buy-candidates")
async def buy_candidates(limit: int = Query(default=20)):
    """기술적 매수 후보 (점수 30 이상) + 펀더멘탈 점수 합산"""
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT t.*,
                   s.news_score, s.dart_score, s.recommendation AS fundamental_rec,
                   s.total_score AS fundamental_total
            FROM stock_technical t
            LEFT JOIN stock_scores s
                   ON t.stock_code = s.stock_code
                  AND s.score_date = (SELECT MAX(score_date) FROM stock_scores)
            WHERE t.signal = '매수' AND t.tech_score >= 30
            ORDER BY (t.tech_score + COALESCE(s.total_score, 0)) DESC
            LIMIT $1
        """, limit)
    result = []
    for row in rows:
        d = dict(row)
        d["analyzed_at"] = str(d["analyzed_at"])
        for k in ("signals", "targets"):
            if isinstance(d.get(k), str):
                d[k] = json.loads(d[k])
        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(limit: int = 0):
    asyncio.create_task(job_analyze(limit=limit))
    return {"status": "started"}

@app.post("/analyze/{code}")
async def analyze_single(code: str, ai: bool = False):
    async with httpx.AsyncClient() as client:
        result = await analyze_stock(client, code, with_ai=ai)
        if result:
            return JSONResponse(content=result)
    return JSONResponse(content={"error": "analysis failed"}, status_code=500)

# ── 보유 포지션 맞춤 분석 ────────────────────────────────

@app.post("/position")
async def position_analysis(req: PositionRequest, ai: bool = False):
    """보유 종목 매입가/수량 기반 맞춤 손익 + 전략 분석"""
    code = req.code

    # 캐시 확인
    result = None
    if redis_cl:
        try:
            cached = await redis_cl.get(f"ta:{code}")
            if cached:
                result = json.loads(cached)
        except: pass

    if not result:
        async with httpx.AsyncClient() as client:
            result = await analyze_stock(client, code, req.name, with_ai=False)

    if not result:
        return JSONResponse(content={"error": "종목 분석 실패"}, status_code=500)

    ind        = result.get("indicators", {})
    tech_score = result.get("tech_score", 0)
    price      = ind.get("price", 0)

    pos = analyze_position(price, req.buy_price, req.qty, ind, tech_score)

    # AI 판단문 (요청 시)
    ai_opinion = result.get("ai_opinion", "")
    if ai and not ai_opinion:
        async with httpx.AsyncClient() as client:
            ai_opinion = await generate_ai_opinion(
                client, code, req.name or result.get("name", code),
                ind, tech_score, result.get("signals", []),
                result.get("targets", {}))

    return {
        "code": code,
        "name": req.name or result.get("name", code),
        "buy_price": req.buy_price,
        "qty": req.qty,
        "current_price": price,
        "tech_score": tech_score,
        "signal": result.get("signal"),
        "signals": result.get("signals", []),
        "indicators": ind,
        "position": pos,
        "targets": result.get("targets", {}),
        "ai_opinion": ai_opinion,
        "analyzed_at": result.get("analyzed_at"),
    }

# ── 종목 전체 리포트 (AI 판단문 포함) ────────────────────

@app.get("/report/{code}")
async def full_report(code: str):
    """기술적 분석 + AI 판단문 + 뉴스감성 통합 리포트"""
    news_score = 0.0
    if pg_pool:
        try:
            async with pg_pool.acquire() as conn:
                row = await conn.fetchrow(
                    "SELECT news_score FROM stock_scores WHERE stock_code=$1 "
                    "ORDER BY score_date DESC LIMIT 1", code)
                if row:
                    news_score = float(row["news_score"] or 0)
        except: pass

    async with httpx.AsyncClient() as client:
        result = await analyze_stock(client, code, with_ai=True, news_score=news_score)

    if not result:
        return JSONResponse(content={"error": "분석 실패"}, status_code=500)

    # DB에서 추가 정보
    extra = {}
    if pg_pool:
        try:
            async with pg_pool.acquire() as conn:
                score_row = await conn.fetchrow(
                    "SELECT * FROM stock_scores WHERE stock_code=$1 "
                    "ORDER BY score_date DESC LIMIT 1", code)
                news_rows = await conn.fetch(
                    "SELECT title, sentiment, intensity, reason "
                    "FROM news_analysis WHERE primary_stock=$1 "
                    "ORDER BY analyzed_at DESC LIMIT 5", code)
                if score_row:
                    extra["score"] = dict(score_row)
                    extra["score"]["score_date"] = str(extra["score"]["score_date"])
                extra["recent_news"] = [dict(r) for r in news_rows]
        except: pass

    return {**result, **extra, "news_score": news_score}