auto-stock/backend/app/api/endpoints/tv_overlay.py

"""TradingView 스타일 프론트용 읽기 전용 GET 엔드포인트 (계약: tv_redesign/CONTRACT.md).

- GET /positions/overlay        — 포지션 오버레이(손절·목표·트레일링·시간손절)
- GET /risk/equity-history      — 일별 실현 순손익 곡선(전략 그룹별)
- GET /strategy-config/display  — §0 우선순위 실효 전략 파라미터

전부 SELECT 전용 · KIS 호출 없음 · DB 쓰기 없음. 라우터 단위 require_api_key (api/__init__.py).
"""
import logging
import math
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Depends, Query
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.database import get_db
from app.core.timeutil import now_kst
from app.models.models import ActiveTrade, Portfolio, TradeHistory
from app.services import tv_view_core as core

router = APIRouter()
logger = logging.getLogger(__name__)

_DT_STRATEGIES = ("daytrading", "day_trade")
_EXECUTED = ("EXECUTED", "COMPLETED", "FILLED")
_KST = timezone(timedelta(hours=9))


def _num(v: Any) -> Optional[float]:
    try:
        f = float(v)
    except (TypeError, ValueError):
        return None
    return f if math.isfinite(f) else None


def _cached_atr(stock_code: str, entry_at=None) -> Optional[float]:
    """swing/exit_service 폴백 경로(_stale_fallback_stop_pct_v2)와 같은 검증을 거친 ATR만 반환.

    나이 ≤ SWING_FALLBACK_PRICE_MAX_AGE_SEC 이고 진입 이후 값이어야 한다. 아니면 None.
    """
    try:
        from app.services.swing.exit_service import _last_atr
    except Exception as e:  # import 실패는 ATR 미상으로 처리하되 숨기지 않는다
        logger.warning("[tv_overlay] swing ATR 캐시 import 실패: %s", e)
        return None
    c = _last_atr.get(stock_code)
    if not (isinstance(c, tuple) and len(c) == 3):
        return None
    a, mono, wall = c
    if (time.monotonic() - mono) > float(settings.SWING_FALLBACK_PRICE_MAX_AGE_SEC):
        return None
    w, e_at = core.to_naive_kst(wall), core.to_naive_kst(entry_at)
    if e_at is not None and (w is None or w < e_at):
        return None
    return _num(a)


def _swing_overlay(row: ActiveTrade, pf_row: Optional[Portfolio], now) -> Dict[str, Any]:
    entry = _num(row.entry_price) or 0.0
    qty = int(row.quantity or 0)
    # 현재가: Portfolio(KIS sync 가 갱신) 우선, 없으면 ActiveTrade
    cur = _num(pf_row.current_price) if pf_row is not None else None
    if cur is None:
        cur = _num(row.current_price)
    avg = _num(pf_row.average_price) if pf_row is not None else None
    pnl = pnl_pct = None
    if cur is not None and entry > 0:
        pnl = round((cur - entry) * qty, 0)
        pnl_pct = round((cur - entry) / entry * 100, 2)

    atr = _cached_atr(row.stock_code, row.created_at)
    stop = basis = stop_alt = stop_note = None
    stored = core.plausible_stop_price(row.stop_loss, entry)
    if entry > 0:
        try:
            from app.services.mean_reversion_strategy import resolve_swing_stop_params_v2
            pct, mult = resolve_swing_stop_params_v2(atr, cur if cur else entry)
        except Exception as e:
            logger.warning("[tv_overlay] 손절 파라미터 조회 실패(%s): %s", row.stock_code, e)
            pct = mult = None
        if pct is not None and atr:
            # §5.2: 가격 기준 비교 → 더 높은 손절가가 타이트
            stop = round(max(entry - mult * atr, entry * (1 + pct / 100)), 2)
            basis = "atr"
        elif stored is not None:
            stop, basis = round(stored, 2), "stored"
        elif pct is not None:
            # ATR 미상 — 퍼센트 손절만(실제 손절은 ATR 손절과 비교한 타이트한 쪽이라 이보다 높을 수 있다)
            stop, basis = round(entry * (1 + pct / 100), 2), "pct"
            try:
                # 고변동(ATR/종가>3%) 가정 손절률 — 신선한 ATR 이 없어 어느 쪽인지 알 수 없다
                hv_pct, _ = resolve_swing_stop_params_v2(entry * 0.04, entry)
                stop_alt = round(entry * (1 + hv_pct / 100), 2)
                stop_note = "ATR 미상: stop_loss=일반 변동성 손절가, stop_loss_alt=고변동 손절가 (실제는 둘 사이이거나 ATR 손절로 더 높을 수 있음)"
            except Exception as e:
                logger.warning("[tv_overlay] 고변동 손절 조회 실패(%s): %s", row.stock_code, e)

    rem = _num(row.remaining_pct)
    trailing_active = rem is not None and rem <= 0.5
    highest = _num(row.highest_since_entry)
    trailing_stop = None
    if trailing_active and atr and highest:
        try:
            from app.services.mean_reversion_strategy import resolve_swing_trailing_atr_mult_v2
            trailing_stop = round(highest - resolve_swing_trailing_atr_mult_v2() * atr, 2)
        except Exception as e:
            logger.warning("[tv_overlay] 트레일링 배수 조회 실패(%s): %s", row.stock_code, e)

    et = core.to_naive_kst(row.created_at)
    return {
        "stock_code": row.stock_code,
        "stock_name": row.stock_name,
        "strategy": (row.strategy_name or "swing"),
        "entry_price": entry,
        "avg_price": avg if avg is not None else entry,
        "quantity": qty,
        "current_price": cur,
        "pnl": pnl,
        "pnl_pct": pnl_pct,
        "stop_loss": stop,
        "stop_basis": basis,
        "stop_loss_alt": stop_alt,
        "stop_note": stop_note,
        "target_price": core.plausible_target_price(row.take_profit, entry),
        "trailing_active": bool(trailing_active),
        "trailing_stop": trailing_stop,
        "highest_price": highest,
        "entry_time": et.isoformat() if et else None,
        "elapsed_min": int((now - et).total_seconds() // 60) if et else None,
        "time_stop_at": None,  # 스윙 시간손절은 KRX 거래일 기준(20일)이라 시각으로 환산하지 않는다
    }


def _dt_overlay(code: str, pos: Dict[str, Any], now) -> Dict[str, Any]:
    entry = _num(pos.get("entry_price")) or 0.0
    qty = int(pos.get("quantity", 0) or 0)
    cur = _num(pos.get("current_price"))
    pnl = pnl_pct = None
    if cur is not None and cur > 0 and entry > 0:
        pnl = round((cur - entry) * qty, 0)
        pnl_pct = round((cur - entry) / entry * 100, 2)
    et = core.to_naive_kst(pos.get("entry_time"))
    stop = _num(pos.get("stop_loss"))
    trailing_active = bool(pos.get("partial_sold", False))
    return {
        "stock_code": code,
        "stock_name": pos.get("stock_name") or code,
        "strategy": "dt",
        "entry_price": entry,
        "avg_price": entry,
        "quantity": qty,
        "current_price": cur if (cur is not None and cur > 0) else None,
        "pnl": pnl,
        "pnl_pct": pnl_pct,
        "stop_loss": stop,
        "stop_basis": "stored" if stop is not None else None,
        "target_price": _num(pos.get("target_price")),  # 분할 후 inf 는 None
        "trailing_active": trailing_active,
        "trailing_stop": _num(pos.get("trailing_stop")),
        "highest_price": _num(pos.get("highest_price")),
        "entry_time": et.isoformat() if et else None,
        "elapsed_min": int((now - et).total_seconds() // 60) if et else None,
        "time_stop_at": (et + timedelta(minutes=int(settings.DAY_TRADE_TIME_STOP_MINUTES))).isoformat() if et else None,
    }


@router.get("/positions/overlay")
async def get_positions_overlay(
    code: Optional[str] = Query(None, description="종목코드(생략 시 전체)"),
    db: AsyncSession = Depends(get_db),
):
    now = datetime.now(_KST).replace(tzinfo=None)
    q = select(ActiveTrade).where(ActiveTrade.status == "active")
    from app.services.db_worker import not_dt_strategy_clause
    q = q.where(not_dt_strategy_clause(ActiveTrade.strategy_name))
    if code:
        q = q.where(ActiveTrade.stock_code == code)
    # 종목당 청산이 쓰는 1행(get_swing_active_trade 와 동일: id 최대)만
    raw = (await db.execute(q.order_by(ActiveTrade.id.desc()))).scalars().all()
    seen, rows = set(), []
    for r in raw:
        if r.stock_code not in seen:
            seen.add(r.stock_code)
            rows.append(r)
    pf: Dict[str, Portfolio] = {}
    if rows:
        codes = list(seen)
        pf = {p.stock_code: p for p in (await db.execute(
            select(Portfolio).where(Portfolio.stock_code.in_(codes)))).scalars().all()}
    # Portfolio 에 보유가 없는 active 행(고아)은 제외
    out: List[Dict[str, Any]] = [
        _swing_overlay(r, pf[r.stock_code], now)
        for r in rows if r.stock_code in pf and int(pf[r.stock_code].quantity or 0) > 0
    ]

    # DT: 인메모리 포지션(읽기 전용)
    try:
        from app.services.daytrading.position_manager import day_trading_state
        for c, p in list(day_trading_state.positions.items()):
            if code and c != code:
                continue
            out.append(_dt_overlay(c, p, now))
    except Exception as e:
        logger.warning("[tv_overlay] DT 인메모리 포지션 조회 실패: %s", e)
    return {"positions": out, "as_of": datetime.now(_KST).isoformat()}


@router.get("/risk/equity-history")
async def get_equity_history(
    days: int = Query(90, ge=1, le=730),
    db: AsyncSession = Depends(get_db),
):
    since = now_kst() - timedelta(days=days)
    ts = func.coalesce(TradeHistory.executed_at, TradeHistory.created_at)
    res = await db.execute(
        select(ts, TradeHistory.strategy_type, TradeHistory.net_profit_loss, TradeHistory.currency_code)
        .where(TradeHistory.order_type == "SELL")
        .where(TradeHistory.status.in_(_EXECUTED))
        .where(ts >= since)
        .order_by(ts)
    )
    rows, excluded = [], 0
    for t, st, net, cur in res.all():
        if cur is not None and str(cur).upper() != "KRW":
            excluded += 1  # 미국/USD 행은 원화 곡선에 섞지 않는다
            continue
        rows.append((t.date(), st, net))
    # 낙폭 % 기준: MDD 관리자가 보는 계좌 peak (없으면 최근 잔고). 모르면 drawdown_pct=null
    base = None
    try:
        from app.services.mdd_manager import mdd_manager
        base = _num(getattr(mdd_manager, "peak_balance", 0)) or _num(getattr(mdd_manager, "_last_balance", 0))
        if base is not None and base <= 0:
            base = None
    except Exception as e:
        logger.warning("[tv_overlay] mdd_manager 조회 실패 — drawdown_pct 미산출: %s", e)
    points, null_rows = core.build_equity_points(rows, base)
    return {
        "cost_basis": "net_after_costs",
        "points": points,
        "mdd_thresholds": mdd_thresholds(),
        "null_net_rows": null_rows,
        "excluded_non_krw_rows": excluded,
        "drawdown_pct_base": base,
    }


def mdd_thresholds() -> Dict[str, float]:
    """MDD 단계 임계(%, 음수) — mdd_manager.update() 가 실제로 전이하는 값.

    mdd_manager 는 -10/-20 리터럴로 전이하고 swing.scheduler 가 그 값을 상수로 보관한다
    (settings.MDD_*_THRESHOLD 는 실효 손잡이가 아님 — scheduler 의 mismatch 경보 참조).
    scheduler 를 못 불러오면 settings 로 폴백(warning).
    """
    try:
        from app.services.swing.scheduler import _MDD_MANAGER_PAUSE_PCT, _MDD_MANAGER_FORCE_PCT
        return {"reduce_pct": float(_MDD_MANAGER_PAUSE_PCT), "force_pct": float(_MDD_MANAGER_FORCE_PCT)}
    except Exception as e:
        logger.warning("[tv_overlay] scheduler MDD 상수 조회 실패 — settings 폴백: %s", e)
        return {"reduce_pct": float(settings.MDD_PAUSE_THRESHOLD), "force_pct": float(settings.MDD_FORCE_THRESHOLD)}


def _eff_num(key: str, fallback: float) -> tuple:
    """strategy_cfg(DB) → settings 순 실효값과 출처."""
    from app.services.strategy_config_service import strategy_cfg
    v = _num(strategy_cfg.get(key))
    if v is not None and v > 0:
        return v, "strategy_cfg"
    return float(fallback), "settings"


@router.get("/strategy-config/display")
async def get_strategy_display():
    from app.services.mean_reversion_strategy import (
        resolve_swing_stop_params_v2, resolve_swing_trailing_atr_mult_v2, resolve_swing_exit_rsi_v2,
    )
    adp = None
    try:
        from app.services.adaptive_strategy import get_current_adaptive_params
        adp = get_current_adaptive_params()
    except Exception as e:
        logger.warning("[strategy-display] 적응형 파라미터 조회 실패 — strategy_cfg/settings 사용: %s", e)
    adp_live = adp is not None and not getattr(adp, "fallback_used", True)

    bull, src_bull = _eff_num("MEAN_REVERSION_RSI_THRESHOLD", settings.MEAN_REVERSION_RSI_THRESHOLD)
    bear, src_bear = _eff_num("MEAN_REVERSION_BEAR_RSI_THRESHOLD", settings.MEAN_REVERSION_BEAR_RSI_THRESHOLD)
    period, _ = _eff_num("RSI_PERIOD", 14)
    if adp_live:
        bull = float(adp.rsi_entry)
        bear = float(adp.bear_rsi_entry)
    # 청산 RSI: check_exit_signal 과 같은 함수(settings 기본 + 적응형)
    part, full, exit_adp = resolve_swing_exit_rsi_v2()
    part, full = float(part), float(full)
    sl_pct, atr_mult = resolve_swing_stop_params_v2(None, 0.0)  # 일반 변동성 기준(적응형 → cfg → settings)
    trail = resolve_swing_trailing_atr_mult_v2()
    if adp_live:
        source = "adaptive"
    elif "strategy_cfg" in (src_bull, src_bear):
        source = "strategy_cfg"
    else:
        source = "settings"
    exit_source = "adaptive" if exit_adp else "settings"
    return {
        "rsi": {"period": int(period), "bull_entry": bull, "bear_entry": bear,
                "partial_exit": part, "full_exit": full},
        "atr": {"stop_mult": atr_mult, "trailing_mult": trail},
        "stop_loss_pct": sl_pct,
        "dt": {
            "daily_loss_pct": float(settings.DAY_TRADE_DAILY_LOSS_LIMIT) * 100,
            "time_stop_min": int(settings.DAY_TRADE_TIME_STOP_MINUTES),
            "time_stop_min_gain": float(settings.DAY_TRADE_TIME_STOP_MIN_GAIN) * 100,
            "eod_flat": f"{int(settings.DAY_TRADE_EOD_HOUR):02d}:{int(settings.DAY_TRADE_EOD_MIN):02d}",
            "entry_cutoff": f"{int(settings.DAY_TRADE_ENTRY_CUTOFF_HOUR):02d}:{int(settings.DAY_TRADE_ENTRY_CUTOFF_MIN):02d}",
        },
        "mdd": mdd_thresholds(),
        "adaptive_mode": getattr(adp, "trading_mode", None) if adp is not None else None,
        "source": source,
        "sources": {"entry": source, "exit": exit_source},
    }