auto-stock/backend/app/services/db_worker.py

"""Background DB Writer Worker

ARCHITECTURE (CLAUDE.md 3.2):
  - 실시간/콜백 스레드에서는 절대 DB 세션을 직접 열지 않는다.
  - 모든 DB 쓰기 작업은 이 모듈의 db_task_queue 로 위임한다.
  - 워커가 큐를 소비하며 단일 세션에서 순차 처리 → 락 경합 제거.

Supported task types:
  - insert_trade_history       : dict  → TradeHistory INSERT
  - update_trade_history       : dict  → TradeHistory UPDATE by id (체결 확인 후 슬리피지/수량 반영)
  - update_trade_history_by_order_id : dict → TradeHistory UPDATE by order_id (KIS 주문번호 기준)
  - upsert_stock_analysis      : dict  → StockAnalysis single UPSERT
  - bulk_upsert_stock_analysis : list  → StockAnalysis batch UPSERT
  - upsert_stock_ban           : dict  → StockBan UPSERT (loss_tracker)
  - insert_news                 : dict  → News INSERT (뉴스 감성 분석 결과 저장)
  - log_mcp_deny               : dict  → MCP 거부 이벤트 로그 (DB 저장 없이 로그만)
  - upsert_portfolio           : dict  → Portfolio INSERT/UPDATE (strategy_type 포함)
  - upsert_active_trade        : dict  → ActiveTrade UPSERT (strategy_name 포함)
  - close_active_trade         : dict  → ActiveTrade 상태를 'closed'로 변경
  - upsert_stock_ban_v2        : dict  → StockBanV2 UPSERT (user_id, stock_code 스코프 — 멀티테넌트)

멀티테넌트(사용자 스코프) 확장 (docs/multi_tenant_design.md §3.4):
  - insert_trade_history / update_trade_history(_by_order_id) / upsert_portfolio /
    upsert_active_trade / close_active_trade / insert_trade_journal 는 payload
    data["user_id"] 가 있으면 조회 필터·생성 컬럼에 반영하고, 없으면(None) 기존과
    100% 동일하게 동작한다(하위 호환). 새 task 타입을 추가하지 않는다.
  - stock_bans 는 PK가 stock_code 단일이라 사용자 스코프로 확장 불가 → 신규 테이블
    stock_bans_v2(user_id, stock_code UNIQUE 쌍) + 신규 task upsert_stock_ban_v2 사용.
"""
import asyncio
import json
import logging
import uuid
from datetime import date, datetime
from pathlib import Path

from sqlalchemy import or_, select

from app.core.database import AsyncSessionLocal
from app.models.models import TradeHistory
from app.core.timeutil import now_kst

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────────────────────
# 전략명 상수 / 필터 헬퍼 (CLAUDE.md §8.5 항목 3·6)
#
# DT_STRATEGY_NAMES 는 mdd_manager.DT_STRATEGY_NAMES 와 **동일 목록**이어야 한다.
# 여기(db_worker)를 정본으로 두는 이유:
#   - db_worker 는 어떤 서비스 모듈도 모듈 수준에서 import 하지 않아 순환 참조가 없다.
#   - ActiveTrade 를 실제로 쓰는 계층이므로 필터 표현을 한 곳에 모을 수 있다.
# mdd_manager 쪽은 담당 에이전트가 달라 이번에 수정하지 않는다(후속 과제).
# ─────────────────────────────────────────────────────────────────────────────
DT_STRATEGY_NAMES: tuple[str, ...] = ("day_trade", "daytrading")


def not_dt_strategy_clause(column):
    """DT 계열 전략을 제외하는 SQL 3값 논리 안전 필터.

    `column != "day_trade"` 는 column IS NULL 인 legacy 행까지 함께 탈락시킨다
    (SQL 3값 논리: NULL != 'x' → UNKNOWN → WHERE 에서 거짓 취급).
    → NULL 행을 명시적으로 포함시킨 뒤 DT 계열만 NOT IN 으로 제외한다.

    Args:
        column: ActiveTrade.strategy_name 등 비교 대상 컬럼

    Returns:
        SQLAlchemy boolean expression
    """
    return or_(column.is_(None), column.notin_(DT_STRATEGY_NAMES))


# upsert_active_trade payload 중 ORM 컬럼이 아닌 '메타' 키 (setattr/생성 인자 제외 대상)
_ACTIVE_TRADE_META_KEYS = (
    "remaining_pct_multiply",
    "strategy_names",
    "strategy_name_not",
)


def _apply_strategy_name_filter(query, data: dict, column):
    """payload 의 전략명 필터 키를 SELECT 쿼리에 적용.

    지원 키 (우선순위 순 — 상위 키가 있으면 하위 키는 무시):
      1) strategy_name      : str  → column == value        (기존 동작, 무변경)
      2) strategy_names     : list → column IN (...)
      3) strategy_name_not  : str | list → column IS NULL OR column NOT IN (...)

    세 키가 모두 없거나 비어 있으면 query 를 **그대로** 반환하므로,
    새 키를 쓰지 않는 기존 호출부의 동작은 수정 전과 완전히 동일하다.
    """
    strategy_name = data.get("strategy_name")
    if strategy_name:
        return query.where(column == strategy_name)

    names = data.get("strategy_names")
    if names:
        _names = [n for n in names if n]
        if _names:
            return query.where(column.in_(_names))
        return query

    excluded = data.get("strategy_name_not")
    if excluded:
        if isinstance(excluded, str):
            excluded = [excluded]
        _excluded = [n for n in excluded if n]
        if _excluded:
            return query.where(or_(column.is_(None), column.notin_(_excluded)))
    return query


# ─────────────────────────────────────────────────────────────────────────────
# DLQ (Dead Letter Queue) — 3회 재시도 실패 시 파일로 백업
# ─────────────────────────────────────────────────────────────────────────────
DLQ_DIR = Path.home() / ".cache" / "auto-stock" / "dlq"


async def _save_to_dlq(task: dict, error: str) -> None:
    """3회 재시도 실패한 task를 DLQ 파일로 저장.

    저장 위치: ~/.cache/auto-stock/dlq/{timestamp}_{task_type}_{uuid}.json
    파일 I/O는 asyncio.to_thread로 감싸 이벤트 루프 블로킹을 방지한다.
    """
    try:
        task_type = task.get("type", "unknown")
        filename = (
            f"{datetime.now().strftime('%Y%m%d_%H%M%S')}"
            f"_{task_type}"
            f"_{uuid.uuid4().hex[:8]}.json"
        )
        dlq_entry = {
            "task": task,
            "error": str(error),
            "failed_at": datetime.now().isoformat(),
            "retry_count": task.get("_retry_count", 0),
        }

        def _write():
            DLQ_DIR.mkdir(parents=True, exist_ok=True)
            (DLQ_DIR / filename).write_text(
                json.dumps(dlq_entry, default=str, ensure_ascii=False),
                encoding="utf-8",
            )

        await asyncio.to_thread(_write)
        logger.critical(
            "[DLQ] Task saved to DLQ: %s | type=%s | error=%s",
            filename, task_type, error,
        )
    except Exception as e:
        logger.critical(
            "[DLQ] Failed to save to DLQ: %s | task_type=%s",
            e, task.get("type", "unknown"),
        )


# ─────────────────────────────────────────────────────────────────────────────
# 전역 큐 — 앱 어디서나 import 하여 put() 호출
# [이슈8 수정] asyncio.Queue()를 모듈 로드 시점에 생성하면 이벤트 루프가 없는
# 환경(pytest, 스크립트 임포트 등)에서 RuntimeError가 발생할 수 있음.
# _db_task_queue를 None으로 두고, 첫 접근 시 lazy 초기화하는 패턴으로 변경.
# 외부 코드는 기존처럼 db_task_queue 이름으로 접근 가능 (get_db_task_queue() 내부 참조).
# ─────────────────────────────────────────────────────────────────────────────
_db_task_queue: asyncio.Queue | None = None


def get_db_task_queue() -> asyncio.Queue:
    """이벤트 루프 기동 후 최초 호출 시 Queue를 생성(lazy init)하여 반환."""
    global _db_task_queue
    if _db_task_queue is None:
        _db_task_queue = asyncio.Queue()
    return _db_task_queue


# 하위 호환: 기존 코드가 `db_task_queue.put()` 형태로 직접 참조하는 경우를 위해
# 프록시 객체 대신 get_db_task_queue()를 통해 접근하도록 모듈 수준 alias 제공.
# 단, 모듈 임포트 시점에 이벤트 루프가 없어도 오류가 발생하지 않음.
class _QueueProxy:
    """asyncio.Queue를 lazy init으로 감싸는 투명 프록시."""

    def __getattr__(self, name):
        return getattr(get_db_task_queue(), name)

    async def put(self, item):
        await get_db_task_queue().put(item)

    async def get(self):
        return await get_db_task_queue().get()

    def task_done(self):
        get_db_task_queue().task_done()

    def qsize(self):
        return get_db_task_queue().qsize()

    def empty(self):
        return get_db_task_queue().empty()


db_task_queue: asyncio.Queue = _QueueProxy()  # type: ignore[assignment]


# ─────────────────────────────────────────────────────────────────────────────
# 편의 함수 — 호출 측에서 dict 구성 없이 바로 사용
# ─────────────────────────────────────────────────────────────────────────────
async def enqueue_trade_history(data: dict) -> None:
    """매매일지 INSERT 를 큐에 등록."""
    await db_task_queue.put({"type": "insert_trade_history", "data": data})


async def enqueue_update_trade_history(trade_id: int, updates: dict) -> None:
    """기존 TradeHistory 레코드 UPDATE 를 큐에 등록.

    주로 체결 확인(verify_and_sync) 후 executed_price/quantity/status/slippage 반영 시 사용.

    Args:
        trade_id: 업데이트할 TradeHistory.id
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history",
        "data": {"id": trade_id, **updates},
    })


async def enqueue_upsert_analysis(row: dict) -> None:
    """단건 StockAnalysis UPSERT 를 큐에 등록."""
    await db_task_queue.put({"type": "upsert_stock_analysis", "data": row})


async def enqueue_bulk_upsert_analysis(rows: list[dict]) -> None:
    """다건 StockAnalysis batch UPSERT 를 큐에 등록."""
    if rows:
        await db_task_queue.put({"type": "bulk_upsert_stock_analysis", "data": rows})


async def enqueue_db_write(task: dict) -> None:
    """범용 DB 쓰기 태스크를 큐에 등록.

    task 는 반드시 'type' 키를 포함해야 하며, 워커가 지원하는 task type 중 하나여야 한다.
    예: {"type": "log_mcp_deny", "reason": "...", "trade_plan": {...}}
    """
    await db_task_queue.put(task)


async def enqueue_trade_journal_raw(data: dict) -> None:
    """TradeJournal INSERT 를 큐에 등록 (순수 dict 전달용)."""
    await db_task_queue.put({"type": "insert_trade_journal", "data": data})


async def enqueue_stock_ban_upsert(stock_code: str, consecutive_losses: int, ban_until) -> None:
    """StockBan UPSERT 를 큐에 등록 (loss_tracker에서 호출)."""
    await db_task_queue.put({
        "type": "upsert_stock_ban",
        "data": {
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


async def enqueue_stock_ban_upsert_v2(
    user_id: int, stock_code: str, consecutive_losses: int, ban_until
) -> None:
    """[v2] StockBanV2 UPSERT 를 큐에 등록 (사용자 스코프 loss_tracker에서 호출).

    stock_bans(PK=stock_code, 단일 계정 전제)와 달리 stock_bans_v2 는
    UNIQUE(user_id, stock_code) 이므로 동일 종목을 사용자별로 독립 추적한다.
    """
    await db_task_queue.put({
        "type": "upsert_stock_ban_v2",
        "data": {
            "user_id": user_id,
            "stock_code": stock_code,
            "consecutive_losses": consecutive_losses,
            "ban_until": ban_until,
        }
    })


async def enqueue_portfolio_upsert(data: dict) -> None:
    """Portfolio INSERT/UPDATE 를 큐에 등록.

    data 필수 키: stock_code, quantity, average_price
    data 선택 키: stock_name, current_price, profit_loss, profit_loss_rate,
                  market, strategy_type ('swing' | 'daytrading')
    """
    await db_task_queue.put({"type": "upsert_portfolio", "data": data})


async def enqueue_active_trade_upsert(data: dict) -> None:
    """ActiveTrade UPSERT 를 큐에 등록.

    data 필수 키: stock_code, quantity, entry_price
    data 선택 키: strategy_name ('swing' | 'day_trade'), stop_loss,
                  take_profit, entry_reason, highest_since_entry
    data 선택 키(조회 필터 전용, ORM 컬럼 아님 — _ACTIVE_TRADE_META_KEYS):
                  strategy_names(list → IN),
                  strategy_name_not(str|list → NOT IN + strategy_name IS NULL 포함)
                  우선순위: strategy_name > strategy_names > strategy_name_not
    """
    await db_task_queue.put({"type": "upsert_active_trade", "data": data})


async def enqueue_close_active_trade(stock_code: str, strategy_name: str | None = None) -> None:
    """ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.

    전략명 부정 조건(strategy_name_not) 등 확장 필터가 필요하면
    시그니처 변경 없이 enqueue_db_write() 로 payload 를 직접 넘긴다:
        await enqueue_db_write({
            "type": "close_active_trade",
            "data": {"stock_code": code, "strategy_name_not": DT_STRATEGY_NAMES},
        })
    """
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name},
    })


async def enqueue_close_active_trade_v2(
    user_id: int, stock_code: str, strategy_name: str | None = None
) -> None:
    """[v2] 사용자 스코프 ActiveTrade 를 'closed' 상태로 변경하는 태스크를 큐에 등록.

    close_active_trade 와 동일 task type("close_active_trade")을 재사용한다 —
    핸들러가 data.get("user_id")로 필터를 추가하므로 새 타입이 필요 없다.
    """
    await db_task_queue.put({
        "type": "close_active_trade",
        "data": {"stock_code": stock_code, "strategy_name": strategy_name, "user_id": user_id},
    })


async def enqueue_update_trade_history_by_order_id(order_id: str, updates: dict) -> None:
    """KIS 주문번호(order_id)로 TradeHistory UPDATE 를 큐에 등록.

    trade_id 대신 order_id 로 레코드를 조회하여 업데이트한다.
    enqueue_trade_history() 로 INSERT 한 레코드의 id 를 모르는 경우 사용.

    Args:
        order_id: KIS 주문번호 (TradeHistory.order_id)
        updates:  업데이트할 필드 dict (예: {"executed_quantity": 10, "status": "COMPLETED"})
    """
    await db_task_queue.put({
        "type": "update_trade_history_by_order_id",
        "data": {"order_id": order_id, **updates},
    })


async def enqueue_trade_journal(trade, content: str) -> None:
    """TradeJournal INSERT 를 큐에 등록 (main.py post_market_scheduler 용).

    CLAUDE.md 3.2: 스케줄러(실시간 계층)에서 직접 DB 커밋 금지 → 큐 경유.
    """
    from datetime import datetime
    from app.services.trade_analyzer import _to_float, _parse_strategy, _parse_exit_type

    pl_rate = _to_float(trade.profit_loss_rate)
    strategy = _parse_strategy(trade.entry_reason)
    exit_type = _parse_exit_type(trade.exit_reason)
    mood = "positive" if pl_rate > 0 else ("neutral" if pl_rate > -3.0 else "negative")
    tags = [strategy, exit_type]
    if abs(pl_rate) > 5.0:
        tags.append("significant")

    await db_task_queue.put({
        "type": "insert_trade_journal",
        "data": {
            "trade_id": trade.id,
            "stock_code": trade.stock_code,
            "stock_name": trade.stock_name,
            "journal_date": trade.executed_at or trade.created_at or now_kst(),
            "content": content,
            "strategy_name": strategy,
            "mood": mood,
            "tags": tags,
        },
    })


# ─────────────────────────────────────────────────────────────────────────────
# 워커 본체
# ─────────────────────────────────────────────────────────────────────────────
async def db_writer_worker() -> None:
    """lifespan 에서 asyncio.create_task() 로 실행되는 영구 백그라운드 워커.

    큐에서 task 를 꺼내 단일 AsyncSession 안에서 처리한 뒤
    commit / rollback 을 자동 관리한다.
    """
    logger.info("🛠️  Background DB Writer Worker started")
    queue = get_db_task_queue()  # 워커 진입 시 큐 인스턴스 고정
    while True:
        task = await queue.get()
        try:
            task_type = task.get("type")
            data = task.get("data")

            async with AsyncSessionLocal() as session:
                # ── 1. 매매일지 INSERT ─────────────────────────────────────
                if task_type == "insert_trade_history":
                    session.add(TradeHistory(**data))
                    await session.commit()
                    logger.info(
                        "✅ [Worker] TradeHistory saved: %s",
                        data.get("stock_code", "?"),
                    )

                # ── 2. TradeHistory UPDATE (체결 확인 후 슬리피지/수량 반영) ──
                elif task_type == "update_trade_history":
                    trade_id = data.get("id")
                    user_id = data.get("user_id")
                    if trade_id:
                        query = select(TradeHistory).where(TradeHistory.id == trade_id)
                        if user_id is not None:
                            query = query.where(TradeHistory.user_id == user_id)
                        row = await session.execute(query)
                        trade = row.scalar_one_or_none()
                        if trade:
                            for key, val in data.items():
                                if key != "id" and val is not None and hasattr(trade, key):
                                    setattr(trade, key, val)
                            await session.commit()
                            logger.info(
                                "✅ [Worker] TradeHistory updated: id=%s status=%s",
                                trade_id,
                                data.get("status", "?"),
                            )

                # ── 2b. TradeHistory UPDATE by order_id (KIS 주문번호 기준) ──
                elif task_type == "update_trade_history_by_order_id":
                    order_id = data.get("order_id")
                    user_id = data.get("user_id")
                    if order_id:
                        query = select(TradeHistory).where(TradeHistory.order_id == order_id)
                        if user_id is not None:
                            query = query.where(TradeHistory.user_id == user_id)
                        row = await session.execute(query)
                        trade = row.scalar_one_or_none()
                        if trade:
                            for key, val in data.items():
                                if key != "order_id" and val is not None and hasattr(trade, key):
                                    setattr(trade, key, val)
                            await session.commit()
                            logger.info(
                                "✅ [Worker] TradeHistory updated by order_id=%s status=%s",
                                order_id,
                                data.get("status", "?"),
                            )
                        else:
                            logger.warning(
                                "⚠️ [Worker] TradeHistory not found for order_id=%s",
                                order_id,
                            )

                # ── 3. 단건 StockAnalysis UPSERT ──────────────────────────
                elif task_type == "upsert_stock_analysis":
                    from app.services.analysis_db_service import _execute_upsert
                    await _execute_upsert(session, [data])
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockAnalysis upsert: %s",
                        data.get("stock_code", "?"),
                    )

                # ── 3. 다건 StockAnalysis batch UPSERT ────────────────────
                elif task_type == "bulk_upsert_stock_analysis":
                    from app.services.analysis_db_service import _execute_upsert
                    await _execute_upsert(session, data)          # data = list[dict]
                    await session.commit()
                    logger.info(
                        "✅ [Worker] Bulk StockAnalysis upsert: %d rows", len(data)
                    )

                # ── 4. StockBan UPSERT ────────────────────────────────────────
                elif task_type == "upsert_stock_ban":
                    from app.models.models import StockBan
                    from app.core.dt_utils import to_naive_utc

                    stock_code = data.get("stock_code")
                    consecutive_losses = data.get("consecutive_losses")
                    # stock_bans.ban_until은 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
                    ban_until = to_naive_utc(data.get("ban_until"))

                    result = await session.execute(select(StockBan).where(StockBan.stock_code == stock_code))
                    record = result.scalar_one_or_none()

                    if record is None:
                        record = StockBan(
                            stock_code=stock_code,
                            consecutive_losses=consecutive_losses,
                            ban_until=ban_until,
                        )
                        session.add(record)
                    else:
                        record.consecutive_losses = consecutive_losses
                        record.ban_until = ban_until

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockBan upsert: %s (losses=%d, ban_until=%s)",
                        stock_code,
                        consecutive_losses,
                        ban_until.isoformat() if ban_until else "None",
                    )

                # ── 4v2. StockBanV2 UPSERT (user_id, stock_code 스코프) ────────
                elif task_type == "upsert_stock_ban_v2":
                    from app.models.models import StockBanV2
                    from sqlalchemy.dialects.postgresql import insert as pg_insert
                    from app.core.dt_utils import to_naive_utc

                    user_id = data.get("user_id")
                    stock_code = data.get("stock_code")
                    consecutive_losses = data.get("consecutive_losses")
                    # stock_bans_v2.ban_until도 tz-naive 컬럼 — tz-aware 값이 들어와도 안전하게 저장.
                    ban_until = to_naive_utc(data.get("ban_until"))

                    # PostgreSQL INSERT ... ON CONFLICT DO UPDATE — 여러 워커/사용자
                    # 사이클이 동시에 같은 (user_id, stock_code)를 갱신해도 경쟁 안전.
                    stmt = pg_insert(StockBanV2).values(
                        user_id=user_id,
                        stock_code=stock_code,
                        consecutive_losses=consecutive_losses,
                        ban_until=ban_until,
                    )
                    stmt = stmt.on_conflict_do_update(
                        constraint="uq_stock_bans_v2_user_stock",
                        set_={
                            "consecutive_losses": stmt.excluded.consecutive_losses,
                            "ban_until": stmt.excluded.ban_until,
                            "updated_at": now_kst(),
                        },
                    )
                    await session.execute(stmt)
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] StockBanV2 upsert: user=%s %s (losses=%s, ban_until=%s)",
                        user_id, stock_code, consecutive_losses,
                        ban_until.isoformat() if ban_until else "None",
                    )

                # ── 4b. News INSERT (중복 URL 무시) ────────────────────────────
                elif task_type == "insert_news":
                    from app.models.models import News

                    news_url = data.get("url")
                    is_dup = False
                    # URL 기반 중복 방지 (동일 URL은 skip)
                    if news_url:
                        dup_result = await session.execute(
                            select(News.id).where(News.url == news_url).limit(1)
                        )
                        if dup_result.scalar_one_or_none() is not None:
                            is_dup = True
                            logger.debug(
                                "⏭️ [Worker] News skip (dup URL): %s",
                                (data.get("title") or "?")[:40],
                            )

                    if not is_dup:
                        session.add(News(**data))
                        await session.commit()
                        logger.debug(
                            "✅ [Worker] News saved: %s",
                            (data.get("title") or "?")[:40],
                        )

                # ── 5. MCP 거부 이벤트 로그 ───────────────────────────────────
                elif task_type == "log_mcp_deny":
                    # DB 저장 없이 로그만 기록 (필요 시 별도 테이블 확장 가능)
                    logger.warning(
                        "🛑 [Worker] MCP DENY logged: reason=%s, trade_plan=%s",
                        task.get("reason", "unknown"),
                        task.get("trade_plan"),
                    )

                # ── 6. Portfolio UPSERT ────────────────────────────────────────
                elif task_type == "upsert_portfolio":
                    from app.models.models import Portfolio

                    stock_code = data.get("stock_code")
                    user_id = data.get("user_id")
                    query = select(Portfolio).where(Portfolio.stock_code == stock_code)
                    if user_id is not None:
                        query = query.where(Portfolio.user_id == user_id)
                    result = await session.execute(query)
                    record = result.scalar_one_or_none()

                    if record is None:
                        record = Portfolio(**data)
                        session.add(record)
                    else:
                        for key, val in data.items():
                            if key != "stock_code" and val is not None:
                                setattr(record, key, val)

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] Portfolio upsert: %s (strategy=%s)",
                        stock_code,
                        data.get("strategy_type", "swing"),
                    )

                # ── 7. ActiveTrade UPSERT ──────────────────────────────────────
                elif task_type == "upsert_active_trade":
                    from app.models.models import ActiveTrade

                    stock_code = data.get("stock_code")
                    strategy_name = data.get("strategy_name")
                    user_id = data.get("user_id")

                    query = select(ActiveTrade).where(
                        ActiveTrade.stock_code == stock_code,
                        ActiveTrade.status == "active",
                    )
                    # [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
                    # strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
                    query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
                    if user_id is not None:
                        query = query.where(ActiveTrade.user_id == user_id)
                        # [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
                        # scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
                        # 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
                        # (멀티테넌트 플래그 ON 전제조건)
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()
                    else:
                        # legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
                        # 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
                        # MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()

                    if record is None:
                        # remaining_pct_multiply·전략명 필터 키는 신규 생성 시 무시
                        create_data = {
                            k: v for k, v in data.items()
                            if k not in _ACTIVE_TRADE_META_KEYS
                        }
                        # 필수 컬럼(quantity/entry_price) 없이 신규 생성 요청이 오면
                        # IntegrityError로 3회 재시도만 낭비 → 경고 후 스킵
                        _missing = [k for k in ("quantity", "entry_price") if create_data.get(k) is None]
                        if _missing:
                            logger.warning(
                                "⚠️ [Worker] ActiveTrade upsert 스킵: %s active 행 없음 + 필수 키 누락 %s (keys=%s)",
                                stock_code, _missing, sorted(create_data.keys()),
                            )
                            continue
                        record = ActiveTrade(**create_data)
                        session.add(record)
                    else:
                        # remaining_pct_multiply: 기존 remaining_pct에 비율을 곱함
                        # (MDD 축소 시 기존 분할매도 이력 반영)
                        # data.get() 사용 — pop()으로 task dict를 파괴적으로 변경하면
                        # 재시도 시 이 키가 사라져 no-op으로 성공 처리되는 버그 방지
                        multiply_ratio = data.get("remaining_pct_multiply")
                        if multiply_ratio is not None:
                            current_remaining = getattr(record, "remaining_pct", 1.0) or 1.0
                            record.remaining_pct = float(current_remaining) * float(multiply_ratio)
                        for key, val in data.items():
                            if key == "stock_code" or key in _ACTIVE_TRADE_META_KEYS:
                                continue
                            if val is not None:
                                setattr(record, key, val)

                    await session.commit()
                    logger.debug(
                        "✅ [Worker] ActiveTrade upsert: %s (strategy=%s)",
                        stock_code,
                        strategy_name,
                    )

                # ── 8. ActiveTrade close ───────────────────────────────────────
                elif task_type == "close_active_trade":
                    from app.models.models import ActiveTrade
                    from datetime import datetime as _dt

                    stock_code = data.get("stock_code")
                    strategy_name = data.get("strategy_name")
                    user_id = data.get("user_id")

                    query = select(ActiveTrade).where(
                        ActiveTrade.stock_code == stock_code,
                        ActiveTrade.status == "active",
                    )
                    # [§8.5 항목6] strategy_name(동등) / strategy_names(IN) /
                    # strategy_name_not(NOT IN + NULL 포함) — 새 키 미사용 시 무변경.
                    query = _apply_strategy_name_filter(query, data, ActiveTrade.strategy_name)
                    if user_id is not None:
                        query = query.where(ActiveTrade.user_id == user_id)
                        # [2026-09-27 §8.5 후속] legacy 경로와 동일하게 '가장 최근 행' 우선.
                        # scalar_one_or_none() 은 한 사용자가 같은 종목에 비-DT active 행을
                        # 2건 이상 보유하면 MultipleResultsFound → 3회 재시도 후 DLQ 로 빠진다.
                        # (멀티테넌트 플래그 ON 전제조건)
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()
                    else:
                        # legacy(user_id=None) 경로: 사용자 다중화로 동일 종목 active
                        # 행이 2건 이상 존재할 수 있어 scalar_one_or_none() 이
                        # MultipleResultsFound 를 던질 수 있음 → 가장 최근 행 우선.
                        query = query.order_by(ActiveTrade.id.desc()).limit(1)
                        result = await session.execute(query)
                        record = result.scalars().first()

                    if record:
                        record.status = "closed"
                        record.updated_at = now_kst()
                        await session.commit()
                        logger.debug(
                            "✅ [Worker] ActiveTrade closed: %s (strategy=%s)",
                            stock_code,
                            strategy_name,
                        )

                # ── 9. TradeJournal INSERT ────────────────────────────────────
                elif task_type == "insert_trade_journal":
                    from app.models.models import TradeJournal

                    session.add(TradeJournal(**data))
                    await session.commit()
                    logger.debug(
                        "✅ [Worker] TradeJournal saved: %s",
                        data.get("stock_code", "?"),
                    )

                else:
                    logger.warning("⚠️  [Worker] Unknown task type: %s", task_type)

        except asyncio.CancelledError:
            # 큐 drain: 종료 전 남은 태스크 처리하여 데이터 손실 방지
            remaining = 0
            while not queue.empty():
                try:
                    drain_task = queue.get_nowait()
                except asyncio.QueueEmpty:
                    # get_nowait()이 실패해 아이템을 꺼내지 못했으므로 task_done() 호출 금지
                    break
                try:
                    drain_type = drain_task.get("type", "unknown")
                    drain_data = drain_task.get("data")
                    async with AsyncSessionLocal() as drain_session:
                        if drain_type == "insert_trade_history":
                            drain_session.add(TradeHistory(**drain_data))
                            await drain_session.commit()
                        elif drain_type in (
                            "update_trade_history",
                            "update_trade_history_by_order_id",
                            "upsert_stock_analysis",
                            "bulk_upsert_stock_analysis",
                            "upsert_stock_ban",
                            "upsert_stock_ban_v2",
                            "upsert_portfolio",
                            "upsert_active_trade",
                            "close_active_trade",
                            "insert_trade_journal",
                        ):
                            # 단순 재큐잉 대신, 중요 태스크만 커밋 시도
                            # insert_trade_history 외에는 복잡한 로직이므로 DLQ 저장
                            await _save_to_dlq(drain_task, "shutdown_drain")
                        else:
                            await _save_to_dlq(drain_task, "shutdown_drain_unknown_type")
                    remaining += 1
                except Exception as drain_err:
                    logger.error(f"[DBWorker] Drain 실패: {drain_err}")
                    try:
                        await _save_to_dlq(drain_task, str(drain_err))
                    except Exception:
                        pass
                    remaining += 1
                finally:
                    queue.task_done()
            if remaining:
                logger.info(f"[DBWorker] 종료 시 {remaining}건 drain 완료")
            logger.info("⏹️  DB Writer Worker stopped")
            break
        except Exception as exc:
            retry_count = task.get("_retry_count", 0)
            max_retries = 3
            task_type = task.get("type", "unknown")

            if retry_count < max_retries:
                task["_retry_count"] = retry_count + 1
                delay = min(2 ** retry_count, 8)  # 1, 2, 4초 지수 백오프
                logger.warning(
                    "⚠️ [Worker] DB task retry %d/%d (type=%s): %s — %ds 후 재시도",
                    retry_count + 1, max_retries, task_type, exc, delay,
                )
                await asyncio.sleep(delay)
                await queue.put(task)  # 큐에 재등록
            else:
                logger.error(
                    "❌ [Worker] DB task DROPPED after %d retries (type=%s): %s",
                    max_retries, task_type, exc, exc_info=True,
                )
                # DLQ: 최종 실패 task를 파일로 백업하여 데이터 손실 방지
                await _save_to_dlq(task, str(exc))
        finally:
            queue.task_done()