"""
Aux Signal Service (port 8282, 172.30.0.25)

외부 보조 데이터 수집:
- 네이버 종목 integration API → 컨센서스 + 기관/외국인 일별 수급
- 한국은행 ECOS API → USD/KRW 환율, 국고채 10년 금리
"""
import os
import asyncio
import json
from datetime import date, datetime, timedelta
from typing import Optional, List

import asyncpg
import structlog
import httpx
from fastapi import FastAPI, Query, BackgroundTasks
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from pytz import timezone

# ─────────────────────────────────────────────────────────────
# 설정
# ─────────────────────────────────────────────────────────────
PG = {
    "host": os.getenv("POSTGRES_HOST", "postgres"),
    "port": int(os.getenv("POSTGRES_PORT", 5432)),
    "database": os.getenv("POSTGRES_DB", "trading_ai"),
    "user": os.getenv("POSTGRES_USER", "kyu"),
    "password": os.getenv("POSTGRES_PASSWORD", ""),
}
KST = timezone("Asia/Seoul")
ECOS_KEY = os.getenv("ECOS_API_KEY", "")

logger = structlog.get_logger()
app = FastAPI(title="Aux Signal")
pg_pool: Optional[asyncpg.Pool] = None
scheduler = AsyncIOScheduler(timezone=KST)

# ─────────────────────────────────────────────────────────────
# DDL
# ─────────────────────────────────────────────────────────────
DDL = """
CREATE TABLE IF NOT EXISTS analyst_consensus (
    stock_code     VARCHAR(10) PRIMARY KEY,
    target_price   BIGINT DEFAULT 0,
    recomm_mean    DOUBLE PRECISION DEFAULT 0,
    create_date    DATE,
    updated_at     TIMESTAMP DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS inst_daily_flow (
    stock_code         VARCHAR(10),
    trade_date         DATE,
    foreign_net        BIGINT DEFAULT 0,
    inst_net           BIGINT DEFAULT 0,
    individual_net     BIGINT DEFAULT 0,
    foreign_hold_ratio DOUBLE PRECISION DEFAULT 0,
    close_price        BIGINT DEFAULT 0,
    change_amount      BIGINT DEFAULT 0,
    PRIMARY KEY (stock_code, trade_date)
);
CREATE INDEX IF NOT EXISTS idx_flow_stock ON inst_daily_flow(stock_code, trade_date DESC);

CREATE TABLE IF NOT EXISTS macro_daily (
    indicator   VARCHAR(20),
    trade_date  DATE,
    value       DOUBLE PRECISION,
    created_at  TIMESTAMP DEFAULT NOW(),
    PRIMARY KEY (indicator, trade_date)
);
"""

# ─────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────
def _parse_int(v) -> int:
    if not v or v in ("-", ""):
        return 0
    s = str(v).replace(",", "").replace("+", "").replace(" ", "").strip()
    try: return int(s)
    except Exception: return 0


def _parse_float(v) -> float:
    if not v or v in ("-", ""):
        return 0.0
    s = str(v).replace(",", "").replace("%", "").replace("+", "").replace(" ", "").strip()
    try: return float(s)
    except Exception: return 0.0


# ─────────────────────────────────────────────────────────────
# 네이버 종목 integration API
# 응답: consensusInfo + dealTrendInfos (60일치 일별 매매동향)
# ─────────────────────────────────────────────────────────────
async def fetch_naver(client: httpx.AsyncClient, code: str) -> Optional[dict]:
    url = f"https://m.stock.naver.com/api/stock/{code}/integration"
    try:
        r = await client.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=15)
        if r.status_code != 200:
            return None
        return r.json()
    except Exception as e:
        logger.debug("naver.req_err", code=code, err=str(e))
        return None


async def save_consensus(conn, code: str, ci: dict):
    if not ci:
        return False
    tp = _parse_int(ci.get("priceTargetMean"))
    rm = _parse_float(ci.get("recommMean"))
    cd = ci.get("createDate")
    cd_date = None
    if cd:
        try:
            cd_date = datetime.strptime(cd, "%Y-%m-%d").date()
        except Exception:
            cd_date = None
    if tp == 0 and rm == 0:
        return False
    await conn.execute("""
        INSERT INTO analyst_consensus (stock_code, target_price, recomm_mean, create_date)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (stock_code) DO UPDATE
          SET target_price=$2, recomm_mean=$3, create_date=$4, updated_at=NOW()
    """, code, tp, rm, cd_date)
    return True


async def save_flow(conn, code: str, dt_infos: list) -> int:
    saved = 0
    for r in dt_infos:
        try:
            bz = r.get("bizdate")
            if not bz: continue
            trade_dt = datetime.strptime(bz, "%Y%m%d").date()
            await conn.execute("""
                INSERT INTO inst_daily_flow
                  (stock_code, trade_date, foreign_net, inst_net,
                   individual_net, foreign_hold_ratio,
                   close_price, change_amount)
                VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
                ON CONFLICT (stock_code, trade_date) DO UPDATE
                  SET foreign_net=$3, inst_net=$4, individual_net=$5,
                      foreign_hold_ratio=$6, close_price=$7, change_amount=$8
            """, code, trade_dt,
                _parse_int(r.get("foreignerPureBuyQuant")),
                _parse_int(r.get("organPureBuyQuant")),
                _parse_int(r.get("individualPureBuyQuant")),
                _parse_float(r.get("foreignerHoldRatio")),
                _parse_int(r.get("closePrice")),
                _parse_int(r.get("compareToPreviousClosePrice")))
            saved += 1
        except Exception as e:
            logger.debug("flow.save_err", code=code, err=str(e))
    return saved


async def collect_naver_data(count: int = 500):
    """시총 상위 N개 종목 컨센서스 + 일별 수급 수집"""
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT DISTINCT ON (d.stock_code) d.stock_code,
                   COALESCE(p.market_cap, 0) AS mc
            FROM dart_corps d
            LEFT JOIN stock_prices p ON p.stock_code=d.stock_code
            WHERE d.is_active=true
            ORDER BY d.stock_code, p.collected_at DESC NULLS LAST
        """)
        top = sorted(rows, key=lambda r: -r["mc"])[:count]
        logger.info("naver.collect_start", count=len(top))

        cons_saved, flow_saved = 0, 0
        async with httpx.AsyncClient() as client:
            for i, row in enumerate(top):
                code = row["stock_code"]
                j = await fetch_naver(client, code)
                if j:
                    async with pg_pool.acquire() as c2:
                        if await save_consensus(c2, code, j.get("consensusInfo") or {}):
                            cons_saved += 1
                        s = await save_flow(c2, code, j.get("dealTrendInfos") or [])
                        flow_saved += s
                # 네이버 rate-limit 회피
                if i < len(top) - 1:
                    await asyncio.sleep(0.3)
    logger.info("naver.collect_done", consensus=cons_saved, flow_rows=flow_saved)
    return {"consensus": cons_saved, "flow_rows": flow_saved}


# ─────────────────────────────────────────────────────────────
# ECOS 매크로
# 731Y001 → 환율 (0000001=USD/KRW)
# 817Y002 → 시장금리 (010195000=국고채 10년)
# ─────────────────────────────────────────────────────────────
ECOS_BASE = "https://ecos.bok.or.kr/api/StatisticSearch"

ECOS_SERIES = [
    # (indicator_name, stat_code, item_code1, freq)
    ("usdkrw",     "731Y001", "0000001",   "D"),
    ("kor_10y",    "817Y002", "010195000", "D"),
    ("kor_3y",     "817Y002", "010190000", "D"),
    ("kospi",      "802Y001", "0001000",   "D"),
]


async def fetch_ecos_series(client: httpx.AsyncClient, stat: str, item: str,
                              freq: str = "D", days: int = 90) -> list:
    if not ECOS_KEY:
        return []
    end = date.today().strftime("%Y%m%d")
    start = (date.today() - timedelta(days=days)).strftime("%Y%m%d")
    url = f"{ECOS_BASE}/{ECOS_KEY}/json/kr/1/200/{stat}/{freq}/{start}/{end}/{item}"
    try:
        r = await client.get(url, timeout=15)
        if r.status_code != 200:
            return []
        j = r.json()
        return (j.get("StatisticSearch") or {}).get("row", []) or []
    except Exception as e:
        logger.warning("ecos.req_err", stat=stat, item=item, err=str(e))
        return []


async def collect_ecos_macro(days: int = 90):
    if not ECOS_KEY:
        logger.error("ecos.no_key")
        return {"saved": 0, "err": "ECOS_API_KEY missing"}
    saved = 0
    async with httpx.AsyncClient() as client:
        for name, stat, item, freq in ECOS_SERIES:
            rows = await fetch_ecos_series(client, stat, item, freq, days)
            async with pg_pool.acquire() as conn:
                for r in rows:
                    try:
                        t = r.get("TIME")
                        v = r.get("DATA_VALUE")
                        if not t or not v:
                            continue
                        dt = datetime.strptime(t, "%Y%m%d").date()
                        await conn.execute("""
                            INSERT INTO macro_daily (indicator, trade_date, value)
                            VALUES ($1, $2, $3)
                            ON CONFLICT (indicator, trade_date) DO UPDATE
                              SET value=$3
                        """, name, dt, float(v))
                        saved += 1
                    except Exception as e:
                        logger.debug("ecos.save_err", err=str(e))
            await asyncio.sleep(0.2)
    logger.info("ecos.collect_done", saved=saved)
    return {"saved": saved}


# ─────────────────────────────────────────────────────────────
# 시작/종료
# ─────────────────────────────────────────────────────────────
@app.on_event("startup")
async def on_start():
    global pg_pool
    pg_pool = await asyncpg.create_pool(**PG, min_size=2, max_size=10)
    async with pg_pool.acquire() as conn:
        await conn.execute(DDL)

    # 네이버 컨센서스/수급: 매일 평일 16:30 (장 마감 후)
    scheduler.add_job(collect_naver_data, CronTrigger(
        day_of_week="mon-fri", hour=16, minute=30),
        id="naver_collect", replace_existing=True)
    # ECOS 매크로: 매일 새벽 5시
    scheduler.add_job(collect_ecos_macro, CronTrigger(hour=5),
        id="ecos_macro", replace_existing=True)
    scheduler.start()
    logger.info("aux-signal.started")


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


# ─────────────────────────────────────────────────────────────
# REST API
# ─────────────────────────────────────────────────────────────
@app.get("/health")
async def health():
    return {"ok": True, "service": "aux-signal",
            "ts": datetime.now(KST).isoformat()}


@app.post("/collect/naver")
async def manual_naver(count: int = Query(default=500, ge=10, le=2000),
                        bg: BackgroundTasks = None):
    if bg:
        bg.add_task(collect_naver_data, count)
        return {"status": "queued", "count": count}
    return await collect_naver_data(count)


@app.post("/collect/macro")
async def manual_macro(days: int = Query(default=90, ge=7, le=365)):
    return await collect_ecos_macro(days)


@app.get("/consensus/{code}")
async def get_consensus(code: str):
    async with pg_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT * FROM analyst_consensus WHERE stock_code=$1", code)
        return dict(row) if row else {"err": "no data"}


@app.get("/flow/{code}")
async def get_flow(code: str, days: int = Query(default=30, ge=1, le=90)):
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT trade_date, foreign_net, inst_net, individual_net,
                   foreign_hold_ratio, close_price, change_amount
            FROM inst_daily_flow
            WHERE stock_code=$1 AND trade_date >= CURRENT_DATE - $2::int
            ORDER BY trade_date DESC
        """, code, days)
        return [dict(r) for r in rows]


@app.get("/macro/latest")
async def macro_latest():
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT DISTINCT ON (indicator) indicator, trade_date, value
            FROM macro_daily
            ORDER BY indicator, trade_date DESC
        """)
        return {r["indicator"]: {"date": str(r["trade_date"]),
                                   "value": float(r["value"])}
                for r in rows}


@app.get("/macro/{indicator}")
async def macro_series(indicator: str, days: int = Query(default=30, ge=1, le=365)):
    async with pg_pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT trade_date, value FROM macro_daily
            WHERE indicator=$1 AND trade_date >= CURRENT_DATE - $2::int
            ORDER BY trade_date DESC
        """, indicator, days)
        return [{"date": str(r["trade_date"]), "value": float(r["value"])}
                for r in rows]


@app.get("/stats")
async def stats():
    async with pg_pool.acquire() as conn:
        c = await conn.fetchval("SELECT COUNT(*) FROM analyst_consensus")
        f = await conn.fetchrow(
            "SELECT COUNT(*) AS rows, COUNT(DISTINCT stock_code) AS codes,"
            " MAX(trade_date) AS latest FROM inst_daily_flow")
        m = await conn.fetchrow(
            "SELECT COUNT(*) AS rows, COUNT(DISTINCT indicator) AS indicators,"
            " MAX(trade_date) AS latest FROM macro_daily")
        return {"consensus_stocks": c,
                "flow": dict(f) if f else {},
                "macro": dict(m) if m else {}}