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

"""TradingView 스타일 프론트용 읽기 전용 뷰 계산 (순수 함수 — settings/DB/네트워크 참조 없음).

계약: scratchpad/tv_redesign/CONTRACT.md. 여기서는 집계·변환만 하고 임계값을 정의하지 않는다.
"""
from __future__ import annotations

import math
from datetime import date, datetime, timedelta, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple

_KST = timezone(timedelta(hours=9))
_DT_NAMES = ("daytrading", "day_trade", "dt")


def group_of_strategy(strategy_type: Optional[str]) -> str:
    """TradeHistory.strategy_type → 'dt' | 'swing' | 'other' (NULL 은 컬럼 기본값 swing 과 동일 취급하지 않고 other)."""
    s = (strategy_type or "").strip().lower()
    if s in _DT_NAMES:
        return "dt"
    if s == "swing":
        return "swing"
    return "other"


def build_equity_points(
    rows: Iterable[Tuple[date, Optional[str], Optional[float]]],
    pct_base: Optional[float],
) -> Tuple[List[Dict[str, Any]], int]:
    """(체결일, strategy_type, net) → 일별 포인트. net 이 None 인 행은 합계에서 제외하고 개수만 반환.

    cum_net 은 조회 구간 시작을 0 으로 한 누적 실현 순손익, peak 는 그 누적의 최고치(시작 0 포함).
    drawdown_pct = (cum_net - peak) / pct_base * 100 (≤0). pct_base(계좌 기준 잔고)를 모르면 None.
    """
    per_day: Dict[date, Dict[str, float]] = {}
    null_rows = 0
    for d, st, net in rows:
        if net is None:
            null_rows += 1
            continue
        try:
            v = float(net)
        except (TypeError, ValueError):
            null_rows += 1
            continue
        if not math.isfinite(v):
            null_rows += 1
            continue
        b = per_day.setdefault(d, {"swing": 0.0, "dt": 0.0, "other": 0.0})
        b[group_of_strategy(st)] += v
    points: List[Dict[str, Any]] = []
    cum = 0.0
    peak = 0.0
    for d in sorted(per_day):
        b = per_day[d]
        day_total = b["swing"] + b["dt"] + b["other"]
        cum += day_total
        peak = max(peak, cum)
        dd = round((cum - peak) / pct_base * 100, 4) if pct_base and pct_base > 0 else None
        points.append({
            "date": d.isoformat(),
            "realized_net": round(day_total, 2),
            "realized_net_swing": round(b["swing"], 2),
            "realized_net_dt": round(b["dt"], 2),
            "realized_net_other": round(b["other"], 2),
            "cum_net": round(cum, 2),
            "peak": round(peak, 2),
            "drawdown_pct": dd,
        })
    return points, null_rows


def plausible_stop_price(stored: Any, entry: float) -> Optional[float]:
    """ActiveTrade.stop_loss 가 '가격'으로 보이는지 판정 (퍼센트·가격 혼재 — §8.5-47).

    0 < stored < entry 이고 entry 의 절반 이상일 때만 가격으로 인정한다. -7/7 같은 퍼센트
    표기이거나 범위 밖이면 None (거짓 값 금지).
    """
    try:
        s = float(stored)
    except (TypeError, ValueError):
        return None
    if not (math.isfinite(s) and math.isfinite(entry)) or entry <= 0:
        return None
    return s if (0.5 * entry <= s < entry) else None


def plausible_target_price(stored: Any, entry: float) -> Optional[float]:
    try:
        t = float(stored)
    except (TypeError, ValueError):
        return None
    if not (math.isfinite(t) and math.isfinite(entry)) or entry <= 0:
        return None
    return t if entry < t <= entry * 3 else None


def to_naive_kst(dt: Any) -> Optional[datetime]:
    if isinstance(dt, str):
        try:
            dt = datetime.fromisoformat(dt)
        except ValueError:
            return None
    if not isinstance(dt, datetime):
        return None
    if dt.tzinfo is not None:
        dt = dt.astimezone(_KST).replace(tzinfo=None)
    return dt


def minute_rows_to_candles(rows: List[dict], today: date, step_min: int) -> List[Dict[str, Any]]:
    """KIS 분봉 output2 → [{time(epoch s), open, high, low, close, volume}] (당일만, 오름차순).

    time 은 KST 벽시계를 UTC epoch 로 변환한 실제 epoch 초. step_min>1 이면 1분봉을 버킷 집계.
    중복 키(날짜+시각)는 마지막 값, 0/결측 OHLC 봉은 제외한다.
    """
    ymd = today.strftime("%Y%m%d")
    uniq: Dict[str, dict] = {}
    for r in rows or []:
        if str(r.get("stck_bsop_date", "")) != ymd:
            continue
        hhmm = str(r.get("stck_cntg_hour", "")).zfill(6)
        uniq[hhmm] = r
    out: List[Dict[str, Any]] = []
    cur: Optional[Dict[str, Any]] = None
    for hhmm in sorted(uniq):
        r = uniq[hhmm]
        try:
            o = float(r.get("stck_oprc") or 0)
            h = float(r.get("stck_hgpr") or 0)
            lo = float(r.get("stck_lwpr") or 0)
            c = float(r.get("stck_prpr") or 0)
            v = int(float(r.get("cntg_vol") or 0))
            hh, mm = int(hhmm[0:2]), int(hhmm[2:4])
        except (TypeError, ValueError):
            continue
        if min(o, h, lo, c) <= 0:
            continue
        mins = hh * 60 + mm
        bucket = mins - (mins % step_min) if step_min > 1 else mins
        ts = int(datetime(today.year, today.month, today.day, bucket // 60, bucket % 60, tzinfo=_KST).timestamp())
        if cur is not None and cur["time"] == ts:
            cur["high"] = max(cur["high"], h)
            cur["low"] = min(cur["low"], lo)
            cur["close"] = c
            cur["volume"] += v
        else:
            cur = {"time": ts, "open": o, "high": h, "low": lo, "close": c, "volume": v}
            out.append(cur)
    return out