docs(env): align PACK_HOST_DIR with CLAUDE.md (F5)

[F4] /api/portfolio 응답의 summary.total_buy가 종목별 단가 × 수량의 합이
되도록 fix. 기존 인라인 코드가 purchase_price를 수량 미곱산으로 단순
누적해 명세(qty 100 · avg 72000 → 7,200,000)와 어긋났음. API_SPEC.md에
purchase_price 필드 의미 + total_buy 계산식 명시. test 3건 (단가 곱산,
avg_price 폴백, 다종목 합산).

[F6] insta-trends spec/plan 상단에 "google_trends → youtube_trending"
변경 이력 추가. Google Trends endpoint 폐기로 source 교체된 이력이
본문 검색 시 혼란 주는 문제 차단. 사유 cross-ref:
feedback_external_data_sources.md

.env.example

@@ -51,9 +51,14 @@ PGID=1000
 # Windows AI Server (NAS 입장에서 바라본 Windows PC IP)
 WINDOWS_AI_SERVER_URL=http://192.168.45.59:8000
 
-# Admin API Key (trade/order 등 민감 엔드포인트 보호, 미설정 시 인증 비활성화)
+# Admin API Key — /api/trade/* 등 민감 엔드포인트 보호.
+# 운영 .env에는 반드시 값을 채워야 함. 빈 값이면 503 응답으로 거부됨 (CODE_REVIEW F2).
 ADMIN_API_KEY=
 
+# 개발 모드: 위 ADMIN_API_KEY 비워둔 채로 trade/admin 엔드포인트 호출 허용.
+# 운영 환경에서는 절대 true로 두지 말 것. 기본 false (보호 활성).
+ALLOW_UNAUTHENTICATED_ADMIN=false
+
 # Anthropic API Key (AI Coach 프록시 + 뉴스 요약 Claude provider)
 ANTHROPIC_API_KEY=
 ANTHROPIC_MODEL=claude-haiku-4-5-20251001
@@ -120,4 +125,6 @@ PACK_BASE_DIR=/app/data/packs
 
 # DSM·Supabase에 노출되는 NAS 호스트 절대경로 (PACK_DATA_PATH와 같은 디렉토리를 호스트 시점에서 가리킴).
 # 운영 NAS는 반드시 /volume1/docker/webpage/media/packs 같은 절대경로 설정. 미설정 시 PACK_DATA_PATH로 fallback (로컬 개발용).
+# DSM API는 일반 사용자 권한에서 /volume1/... 절대경로를 거부(408).
+# shared folder 시점(/docker/...)이 운영 표준 (CLAUDE.md와 일치).
 PACK_HOST_DIR=/volume1/docker/webpage/media/packs

docker-compose.yml

@@ -100,6 +100,7 @@ services:
       - ANTHROPIC_MODEL_SONNET=${ANTHROPIC_MODEL_SONNET:-claude-sonnet-4-6}
       - NAVER_CLIENT_ID=${NAVER_CLIENT_ID:-}
       - NAVER_CLIENT_SECRET=${NAVER_CLIENT_SECRET:-}
+      - YOUTUBE_DATA_API_KEY=${YOUTUBE_DATA_API_KEY:-}
       - INSTA_DATA_PATH=/app/data
       - CARD_TEMPLATE_DIR=/app/app/templates
       - CORS_ALLOW_ORIGINS=${CORS_ALLOW_ORIGINS:-http://localhost:3007,http://localhost:8080}

docs/superpowers/plans/2026-05-16-insta-trends-implementation.md

@@ -2,6 +2,10 @@
 
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
 
+## ⚠️ 변경 이력
+
+- **2026-05-17**: 본문에 `google_trends` source로 기재된 모든 task와 코드 블록은 **실제 구현에서 `youtube_trending`으로 교체됨**. Google Trends 비공식 endpoint(RSS + dailytrends JSON 양쪽) 모두 404 폐기 확인. YouTube Data API v3 mostPopular로 source 대체 + pytrends 의존성 제거. 운영 코드는 현재 `youtube_trending` 사용 중. 이 plan을 다시 실행할 일이 있으면 본문의 `google_trends` 단어를 `youtube_trending`으로 읽어달라. 자세한 사유와 교체 체크리스트는 `feedback_external_data_sources.md`.
+
 **Goal:** Add a "Trends" tab to the Insta page that pulls external trends from NAVER popular + Google Trends, lets the user set category weights, and feeds those weights back into the daily keyword extraction pipeline.
 
 **Architecture:** New `trend_collector` module in insta-lab (NAVER `news.json` 인기순 + `pytrends` Google Trends + Claude Haiku 카테고리 분류 + in-memory 캐시). Existing `trending_keywords` table gets a `source` column; new `account_preferences` table stores category weights. `keyword_extractor` gains a `extract_with_weights()` variant. InstaAgent runs a new 09:00 cron for trend collection and applies weights at 09:30 extraction. web-ui splits InstaCards into Cards/Trends tabs and adds 3 panels (AccountFocus / ExternalTrends / PreferenceImpact).

docs/superpowers/specs/2026-05-16-insta-trends-design.md

@@ -4,6 +4,10 @@
 상태: 사용자 승인 대기 → writing-plans 진입 예정
 연관 문서: `2026-05-15-insta-agent-design.md` (insta-lab 기본 설계)
 
+## ⚠️ 변경 이력
+
+- **2026-05-17**: 본문에 `google_trends` source로 기재된 모든 항목은 **실제 구현에서 `youtube_trending`으로 교체됨**. Google Trends 비공식 endpoint 두 가지(`trendingsearches/daily/rss?geo=KR`, `/trends/api/dailytrends?...`)가 모두 404로 폐기되어 운영 호출이 빈 결과로 끝나는 문제 확인 → YouTube Data API v3 `videos.list?chart=mostPopular&regionCode=KR`로 source 대체. 이후 spec 본문을 읽을 때는 `google_trends` → `youtube_trending`, "Google Trends" → "YouTube 인기"로 치환 해석. 사유와 source 교체 시 동시 갱신 체크리스트: `feedback_external_data_sources.md`.
+
 ---
 
 ## 1. 목적·배경

insta-lab/Dockerfile

@@ -16,7 +16,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
  && rm -rf /var/lib/apt/lists/*
 
 COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+# --timeout 600 --retries 5: NAS 느린 네트워크/CPU에서 pip 다운로드 timeout 방지
+RUN pip install --no-cache-dir --timeout 600 --retries 5 -r requirements.txt
 RUN playwright install chromium
 
 COPY . .

insta-lab/app/config.py

@@ -2,6 +2,7 @@ import os
 
 NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID", "")
 NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET", "")
+YOUTUBE_DATA_API_KEY = os.getenv("YOUTUBE_DATA_API_KEY", "")
 ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY", "")
 ANTHROPIC_MODEL_HAIKU = os.getenv("ANTHROPIC_MODEL_HAIKU", "claude-haiku-4-5-20251001")
 ANTHROPIC_MODEL_SONNET = os.getenv("ANTHROPIC_MODEL_SONNET", "claude-sonnet-4-6")

insta-lab/app/main.py

@@ -265,7 +265,7 @@ async def _bg_collect_trends(task_id: str, categories: list[str]):
     try:
         db.update_task(task_id, "processing", 10, "외부 트렌드 수집 중")
         result = trend_collector.collect_all(categories)
-        msg = f"naver:{result['naver_popular']}, google:{result['google_trends']}"
+        msg = f"naver:{result['naver_popular']}, youtube:{result['youtube_trending']}"
         db.update_task(task_id, "succeeded", 100, msg, result_id=sum(result.values()))
     except Exception as e:
         logger.exception("trends collect failed")

insta-lab/app/trend_collector.py

@@ -1,6 +1,10 @@
-"""외부 트렌드 수집 — NAVER 인기 + Google Trends + LLM 카테고리 분류.
+"""외부 트렌드 수집 — NAVER 인기 + YouTube 인기 영상 + LLM 카테고리 분류.
 
-Phase B Task 3: Google Trends integration via pytrends + Anthropic Haiku 분류 캐시 (24h TTL).
+NAVER: 카테고리별 시드 키워드로 인기 검색 → 빈도 상위 추출.
+YouTube: Google Trends 비공식 endpoint(RSS / dailytrends JSON)가 모두 404 폐기되어
+대체로 YouTube Data API v3 (`videos.list?chart=mostPopular&regionCode=KR`) 사용.
+무료 일일 quota 10000, 한국 region 지원, 인기 영상 50개 제목에서 트렌드 추출.
+LLM 분류 결과는 24h in-memory 캐시.
 """
 
 import json
@@ -11,11 +15,10 @@ from typing import Any, Dict, List, Optional
 
 import requests
 from anthropic import Anthropic
-from pytrends.request import TrendReq
 
 from .config import (
     NAVER_CLIENT_ID, NAVER_CLIENT_SECRET, DEFAULT_CATEGORY_SEEDS,
-    ANTHROPIC_API_KEY, ANTHROPIC_MODEL_HAIKU,
+    ANTHROPIC_API_KEY, ANTHROPIC_MODEL_HAIKU, YOUTUBE_DATA_API_KEY,
 )
 from . import db
 from .news_collector import _clean
@@ -29,16 +32,46 @@ _NAVER_HEADERS = {
     "X-Naver-Client-Secret": NAVER_CLIENT_SECRET,
 }
 
+YOUTUBE_TRENDING_URL = "https://www.googleapis.com/youtube/v3/videos"
+# YouTube 제목 정제: 대괄호·이모지·과도한 길이 제거 후 카드 주제로 적합한 키워드 형태
+_TITLE_BRACKET_RE = re.compile(r"[\[【「『\(][^\]】」』\)]{0,30}[\]】」』\)]")
+_EMOJI_RE = re.compile(
+    r"["
+    r"\U0001F300-\U0001FAFF"   # symbols & pictographs, etc.
+    r"\U00002600-\U000027BF"   # misc symbols, dingbats
+    r"\U0001F1E6-\U0001F1FF"   # regional indicator
+    r"]"
+)
+_TITLE_MAX_LEN = 60
+
+_PLACEHOLDER_SEEDS = {"...", "…", "tbd", "todo", "placeholder", "example"}
+
+
+def _is_valid_seed(s: str) -> bool:
+    """프롬프트 템플릿에 placeholder/빈 값이 들어가 NAVER에 400을 유발하는 일을 막는 가드."""
+    if not s:
+        return False
+    s = s.strip()
+    if len(s) < 2:
+        return False
+    if s.lower() in _PLACEHOLDER_SEEDS:
+        return False
+    return True
+
 
 def _seeds_for(category: str) -> List[str]:
+    """category_seeds 프롬프트 템플릿이 있으면 사용, 없거나 모두 invalid면 config DEFAULT 폴백."""
     pt = db.get_prompt_template("category_seeds")
     if pt and pt.get("template"):
         try:
             data = json.loads(pt["template"])
             if category in data:
-                return list(data[category])
-        except Exception:
-            pass
+                filtered = [s for s in (data[category] or []) if _is_valid_seed(s)]
+                if filtered:
+                    return filtered
+                logger.warning("category_seeds[%s]에 유효한 시드 없음 → DEFAULT 폴백", category)
+        except Exception as e:
+            logger.warning("category_seeds JSON 파싱 실패 → DEFAULT 폴백: %s", e)
     return list(DEFAULT_CATEGORY_SEEDS.get(category, []))
 
 
@@ -142,36 +175,70 @@ def classify_keyword(keyword: str) -> str:
     return cat
 
 
-# ── Google Trends ─────────────────────────────────────────────────────────────
+# ── YouTube Trending ──────────────────────────────────────────────────────────
+# YouTube Data API v3 videos.list?chart=mostPopular&regionCode=KR
+# 한국 인기 영상 50개 제목에서 카드 주제로 적합한 키워드 추출.
 
-def fetch_google_trends() -> List[Dict[str, Any]]:
-    """pytrends 한국 daily trending searches. 실패 시 빈 리스트."""
+def _clean_yt_title(title: str) -> str:
+    """[공식]·【속보】·🔥 등 제거 후 60자 이내로 자른다."""
+    if not title:
+        return ""
+    cleaned = _TITLE_BRACKET_RE.sub("", title)
+    cleaned = _EMOJI_RE.sub("", cleaned)
+    cleaned = re.sub(r"\s+", " ", cleaned).strip()
+    return cleaned[:_TITLE_MAX_LEN]
+
+
+def fetch_youtube_trending() -> List[Dict[str, Any]]:
+    """YouTube Data API v3 mostPopular (한국, 50개). API 키 없거나 호출 실패 시 빈 리스트."""
+    if not YOUTUBE_DATA_API_KEY:
+        logger.info("YOUTUBE_DATA_API_KEY 미설정 — youtube_trending skip")
+        return []
     try:
-        pytrends = TrendReq(hl="ko-KR", tz=540)
-        df = pytrends.trending_searches(pn="south_korea")
+        resp = requests.get(
+            YOUTUBE_TRENDING_URL,
+            params={
+                "part": "snippet",
+                "chart": "mostPopular",
+                "regionCode": "KR",
+                "maxResults": 50,
+                "key": YOUTUBE_DATA_API_KEY,
+            },
+            timeout=15,
+        )
+        resp.raise_for_status()
+        videos = resp.json().get("items", []) or []
     except Exception as e:
-        logger.warning("Google Trends fetch failed: %s", e)
+        logger.warning("YouTube trending fetch failed: %s", e)
         return []
 
     items: List[Dict[str, Any]] = []
-    for idx, row in df.iterrows():
-        kw = str(row.iloc[0]).strip()
-        if not kw:
+    seen = set()
+    total = max(1, len(videos))
+    for idx, v in enumerate(videos):
+        title = (v.get("snippet") or {}).get("title", "")
+        kw = _clean_yt_title(title)
+        if not kw or kw in seen:
             continue
+        seen.add(kw)
+        try:
             cat = classify_keyword(kw)
-        rank_score = round(max(0.0, 1.0 - (idx / max(1, len(df)))), 4)
+        except Exception as e:
+            logger.warning("classify_keyword(%s) 실패: %s", kw, e)
+            cat = "uncategorized"
+        rank_score = round(max(0.0, 1.0 - (idx / total)), 4)
         items.append({
             "keyword": kw,
             "category": cat,
-            "source": "google_trends",
+            "source": "youtube_trending",
             "score": rank_score,
             "articles_count": 0,
         })
     return items
 
 
-def collect_google_trends() -> int:
-    items = fetch_google_trends()
+def collect_youtube_trending() -> int:
+    items = fetch_youtube_trending()
     for it in items:
         db.add_external_trend(it)
     return len(items)
@@ -179,5 +246,5 @@ def collect_google_trends() -> int:
 
 def collect_all(categories: List[str]) -> Dict[str, int]:
     naver_n = collect_naver_popular_for(categories)
-    google_n = collect_google_trends()
-    return {"naver_popular": naver_n, "google_trends": google_n}
+    yt_n = collect_youtube_trending()
+    return {"naver_popular": naver_n, "youtube_trending": yt_n}

insta-lab/requirements.txt

@@ -7,4 +7,3 @@ jinja2>=3.1.4
 playwright==1.48.0
 pytest>=8.0
 pytest-asyncio>=0.24
-pytrends>=4.9

insta-lab/tests/test_main_trends.py

@@ -59,7 +59,7 @@ def test_collect_trends_kicks_background(client, monkeypatch):
 
     def fake_collect_all(cats):
         captured["called"] = True
-        return {"naver_popular": 3, "google_trends": 2}
+        return {"naver_popular": 3, "youtube_trending": 2}
 
     monkeypatch.setattr(trend_collector, "collect_all", fake_collect_all)
     resp = client.post("/api/insta/trends/collect", json={})

insta-lab/tests/test_trend_collector.py

@@ -77,45 +77,84 @@ def test_classify_keyword_with_cache(monkeypatch):
     assert calls["n"] == 1
 
 
-def test_fetch_google_trends_parses_and_classifies(tmp_db, monkeypatch):
-    class FakePyTrends:
-        def __init__(self, *_a, **_kw):
-            pass
+def test_fetch_youtube_trending_parses_and_cleans_titles(tmp_db, monkeypatch):
+    """YouTube Data API mostPopular 응답 → 제목 정제 + 분류."""
+    monkeypatch.setattr(trend_collector, "YOUTUBE_DATA_API_KEY", "fake_key")
+    payload = {
+        "items": [
+            {"snippet": {"title": "[속보] 기준금리 인상 단행 🔥"}},
+            {"snippet": {"title": "(공식) BTS 컴백 무대 🎤"}},
+            {"snippet": {"title": "스트레스 관리 5가지 방법"}},
+            # 중복 제목 — 중복 제거 확인
+            {"snippet": {"title": "[속보] 기준금리 인상 단행 🔥"}},
+        ]
+    }
+    fake_resp = MagicMock()
+    fake_resp.json.return_value = payload
+    fake_resp.raise_for_status.return_value = None
+    monkeypatch.setattr(trend_collector.requests, "get", lambda *a, **kw: fake_resp)
+    monkeypatch.setattr(
+        trend_collector, "classify_keyword",
+        lambda kw: ("economy" if "금리" in kw else
+                    "celebrity" if "BTS" in kw else
+                    "psychology" if "스트레스" in kw else "uncategorized"),
+    )
 
-        def trending_searches(self, pn="south_korea"):
-            import pandas as pd
-            return pd.DataFrame({"0": ["기준금리", "BTS 컴백", "스트레스 관리"]})
+    trends = trend_collector.fetch_youtube_trending()
+    keywords = [t["keyword"] for t in trends]
+    assert "기준금리 인상 단행" in keywords  # 대괄호·이모지 제거
+    assert "BTS 컴백 무대" in keywords  # 괄호 제거
+    assert "스트레스 관리 5가지 방법" in keywords  # 그대로
+    assert len(trends) == 3  # 중복 제거됨
+    assert all(t["source"] == "youtube_trending" for t in trends)
 
-    monkeypatch.setattr(trend_collector, "TrendReq", FakePyTrends)
-    monkeypatch.setattr(trend_collector, "classify_keyword",
-                        lambda kw: {"기준금리": "economy", "BTS 컴백": "celebrity",
-                                    "스트레스 관리": "psychology"}.get(kw, "uncategorized"))
 
-    trends = trend_collector.fetch_google_trends()
-    by_kw = {t["keyword"]: t for t in trends}
-    assert by_kw["기준금리"]["category"] == "economy"
-    assert by_kw["BTS 컴백"]["category"] == "celebrity"
-    assert by_kw["스트레스 관리"]["category"] == "psychology"
-    assert all(t["source"] == "google_trends" for t in trends)
+def test_fetch_youtube_trending_no_api_key_returns_empty(monkeypatch):
+    monkeypatch.setattr(trend_collector, "YOUTUBE_DATA_API_KEY", "")
+    out = trend_collector.fetch_youtube_trending()
+    assert out == []
+
+
+def test_fetch_youtube_trending_graceful_on_api_failure(monkeypatch):
+    monkeypatch.setattr(trend_collector, "YOUTUBE_DATA_API_KEY", "fake_key")
+    fake_resp = MagicMock()
+    fake_resp.raise_for_status.side_effect = RuntimeError("quota exceeded")
+    monkeypatch.setattr(trend_collector.requests, "get", lambda *a, **kw: fake_resp)
+    out = trend_collector.fetch_youtube_trending()
+    assert out == []
 
 
 def test_collect_all_invokes_both_sources(tmp_db, monkeypatch):
     monkeypatch.setattr(trend_collector, "collect_naver_popular_for",
                         lambda cats: 5)
-    monkeypatch.setattr(trend_collector, "collect_google_trends",
+    monkeypatch.setattr(trend_collector, "collect_youtube_trending",
                         lambda: 3)
     out = trend_collector.collect_all(["economy"])
-    assert out == {"naver_popular": 5, "google_trends": 3}
+    assert out == {"naver_popular": 5, "youtube_trending": 3}
 
 
-def test_fetch_google_trends_graceful_on_pytrends_failure(monkeypatch):
-    class FakePyTrends:
-        def __init__(self, *_a, **_kw):
-            pass
+def test_seeds_for_filters_placeholder(tmp_db, monkeypatch):
+    """category_seeds 템플릿에 placeholder '...'가 들어가도 DEFAULT 폴백."""
+    from app import db as db_module
+    db_module.upsert_prompt_template(
+        "category_seeds",
+        '{"economy": ["...", "…", "a", "real_keyword"]}',
+        "test",
+    )
+    out = trend_collector._seeds_for("economy")
+    # '...', '…', 'a'(2자 미만)는 필터링되고 'real_keyword'만 남음
+    assert out == ["real_keyword"]
 
-        def trending_searches(self, pn="south_korea"):
-            raise RuntimeError("rate limited")
 
-    monkeypatch.setattr(trend_collector, "TrendReq", FakePyTrends)
-    out = trend_collector.fetch_google_trends()
-    assert out == []
+def test_seeds_for_falls_back_when_all_invalid(tmp_db, monkeypatch):
+    """모든 시드가 invalid면 DEFAULT_CATEGORY_SEEDS 폴백."""
+    from app import db as db_module
+    db_module.upsert_prompt_template(
+        "category_seeds",
+        '{"economy": ["...", "TBD", ""]}',
+        "test",
+    )
+    out = trend_collector._seeds_for("economy")
+    # DEFAULT_CATEGORY_SEEDS["economy"] 가 반환되어야 함
+    from app.config import DEFAULT_CATEGORY_SEEDS
+    assert out == list(DEFAULT_CATEGORY_SEEDS["economy"])

packs-lab/app/routes.py

@@ -133,8 +133,12 @@ async def sign_link(
 
     # 경로 안전: PACK_HOST_DIR(NAS 호스트 절대경로) 하위인지 확인.
     # file_path는 upload 라우트가 Supabase에 저장한 호스트경로 그대로 전달되어 DSM API에 사용됨.
+    # str.startswith는 '/foo/packs' 와 '/foo/packs_evil' 같은 sibling 경로를 통과시키므로
+    # Path.relative_to로 엄격하게 컴포넌트 단위 검증한다 (CODE_REVIEW F1).
     abs_path = Path(payload.file_path).resolve()
-    if not str(abs_path).startswith(str(PACK_HOST_DIR)):
+    try:
+        abs_path.relative_to(PACK_HOST_DIR.resolve())
+    except ValueError:
         raise HTTPException(status_code=400, detail="허용된 경로 외부")
 
     try:

packs-lab/tests/test_routes.py

@@ -60,6 +60,29 @@ def test_sign_link_path_outside_base():
     assert r.status_code == 400
 
 
+def test_sign_link_rejects_sibling_path():
+    """PACK_HOST_DIR='/foo/packs' 일 때 '/foo/packs_evil/x.mp4' 같이 prefix만
+    통과하는 sibling 경로는 거부해야 한다 (CODE_REVIEW F1, path traversal 변형).
+
+    기존 str.startswith 방식은 trailing slash가 없어 sibling 경로를 통과시킴.
+    relative_to 기반 검증으로 교체되어야 통과한다.
+    """
+    import json as _json
+    from pathlib import Path
+    base_resolved = Path("/foo/packs").resolve()
+    # base의 자식이 아닌 sibling 경로 (예: /foo/packs_evil/...)
+    sibling_posix = (base_resolved.parent / f"{base_resolved.name}_evil" / "x.mp4").as_posix()
+    with patch("app.routes.PACK_HOST_DIR", base_resolved):
+        body = _json.dumps(
+            {"file_path": sibling_posix, "expires_in_seconds": 14400}
+        ).encode()
+        r = client.post("/api/packs/sign-link", content=body, headers=_signed(body))
+    assert r.status_code == 400, (
+        f"sibling 경로 '{sibling_posix}'가 허용됨 (status={r.status_code}) "
+        f"— path traversal 가능성"
+    )
+
+
 def test_upload_invalid_token():
     r = client.post(
         "/api/packs/upload",

scripts/deploy.sh

@@ -1,6 +1,14 @@
 #!/bin/bash
 set -euo pipefail
 
+# ── docker / compose / buildkit timeout 늘리기 ──
+# NAS Celeron J4025에서 pip install·chromium 다운로드 등 무거운 RUN step이
+# 기본 timeout(2분)에 걸려 webhook 자동 배포가 "DeadlineExceeded"로 끝나는 일이
+# 있어 10분으로 상향. 호스트 셸 + deployer 컨테이너 둘 다에 적용됨.
+export COMPOSE_HTTP_TIMEOUT=600
+export DOCKER_CLIENT_TIMEOUT=600
+export BUILDKIT_STEP_LOG_MAX_SIZE=-1
+
 # ── 동시 배포 방지 (flock) ──
 exec 200>/tmp/deploy.lock
 flock -n 200 || { echo "Deploy already running, skipping"; exit 0; }

stock/API_SPEC.md

@@ -142,6 +142,7 @@ KB증권·삼성증권 등 Open API 미제공 증권사용.
       "name": "삼성전자",
       "quantity": 100,
       "avg_price": 72000,
+      "purchase_price": 72000,
       "current_price": 74500,
       "price_session": "NXT_AFTER",
       "price_as_of": "2026-05-11T19:21:40+09:00",
@@ -159,6 +160,10 @@ KB증권·삼성증권 등 Open API 미제공 증권사용.
 }
 ```
 
+> **`purchase_price` 필드**: 종목별 매입 단가(1주당). 사용자가 수동 등록한 매입가가
+> 평균단가(`avg_price`)와 다를 때 표시용으로 분리한다. 미설정 시 `avg_price`로 폴백.
+> `summary.total_buy = SUM(purchase_price × quantity)` (CODE_REVIEW F4에서 명세 정합화).
+
 > **주의**: 현재가 조회에 실패한 종목은 `current_price`, `eval_amount`, `profit_amount`, `profit_rate` 가 `null`로 반환됩니다.
 > 프론트에서 `null` 체크 후 `"조회 실패"` 등으로 표시해 주세요.
 

stock/app/main.py

@@ -47,13 +47,30 @@ scheduler = BackgroundScheduler(timezone=os.getenv("TZ", "Asia/Seoul"))
 # Windows AI Server URL (NAS .env에서 설정)
 WINDOWS_AI_SERVER_URL = os.getenv("WINDOWS_AI_SERVER_URL", "http://192.168.0.5:8000")
 
-# Admin API Key 인증
+# Admin API Key 인증 — /api/trade/* 보호 (CODE_REVIEW F2)
+# 빈 키 + 명시적 dev flag 없으면 503으로 거부. 운영 .env에 ADMIN_API_KEY 누락 시
+# 무인증 통과되던 버그 차단.
 ADMIN_API_KEY = os.getenv("ADMIN_API_KEY", "")
 
 def verify_admin(x_admin_key: str = Header(None)):
-    """admin/trade 엔드포인트 보호용 API 키 검증"""
+    """admin/trade 엔드포인트 보호용 API 키 검증.
+
+    - ADMIN_API_KEY 설정됨 + 키 일치 → 통과
+    - ADMIN_API_KEY 설정됨 + 키 불일치 → 401 Unauthorized
+    - ADMIN_API_KEY 미설정 + ALLOW_UNAUTHENTICATED_ADMIN=true → 통과 (개발 모드)
+    - ADMIN_API_KEY 미설정 + dev flag 없음 → 503 (보호 강화, 운영 .env 누락 차단)
+    """
     if not ADMIN_API_KEY:
-        return  # 키 미설정 시 인증 비활성화 (개발 환경)
+        if os.getenv("ALLOW_UNAUTHENTICATED_ADMIN", "false").lower() == "true":
+            return  # 개발 환경 명시적 허용
+        raise HTTPException(
+            status_code=503,
+            detail=(
+                "admin endpoint protected — ADMIN_API_KEY not configured. "
+                "Set ADMIN_API_KEY in .env, or set ALLOW_UNAUTHENTICATED_ADMIN=true "
+                "for development only."
+            ),
+        )
     if x_admin_key != ADMIN_API_KEY:
         raise HTTPException(status_code=401, detail="Unauthorized")
 
@@ -337,11 +354,11 @@ def get_portfolio():
         price_session = detail["session"] if detail else None
         price_as_of = detail["as_of"] if detail else None
         # avg_price: 평균단가 — 손익(평가금액 - 매입원가) 계산 기준
-        # purchase_price: 매입가 — 총 매입 금액 표시 기준 (없으면 avg_price로 폴백)
+        # purchase_price: 매입 단가(1주당) — 없으면 avg_price로 폴백 (CODE_REVIEW F4)
         purchase_price = item.get("purchase_price") if item.get("purchase_price") is not None else item["avg_price"]
         cost_basis = item["avg_price"] * item["quantity"]
-        # 총 매입 금액 표시는 종목별 매입가의 단순 합계 (수량 미곱산)
-        buy_amount = purchase_price
+        # 총 매입 금액 = 단가 × 보유 수량. API_SPEC.md 예시(qty 100·avg 72000 → 7,200,000)와 일치
+        buy_amount = purchase_price * item["quantity"]
         eval_amount = current_price * item["quantity"] if current_price is not None else None
         profit_amount = (eval_amount - cost_basis) if eval_amount is not None else None
         profit_rate = round((profit_amount / cost_basis) * 100, 2) if (profit_amount is not None and cost_basis) else None

stock/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+pythonpath = .
+asyncio_mode = auto

stock/tests/test_admin_auth.py

@@ -0,0 +1,43 @@
+"""verify_admin 보안 강화 회귀 테스트 (CODE_REVIEW F2).
+
+운영 .env에서 ADMIN_API_KEY가 누락되면 /api/trade/balance, /api/trade/order
+인증이 무력화되는 버그를 막기 위한 가드.
+"""
+import os
+from unittest.mock import patch
+
+import pytest
+from fastapi import HTTPException
+
+from app import main as stock_main
+
+
+def test_verify_admin_rejects_when_key_missing_and_no_dev_flag(monkeypatch):
+    """ADMIN_API_KEY 미설정 + ALLOW_UNAUTHENTICATED_ADMIN 미설정 → 503."""
+    monkeypatch.setattr(stock_main, "ADMIN_API_KEY", "")
+    monkeypatch.delenv("ALLOW_UNAUTHENTICATED_ADMIN", raising=False)
+    with pytest.raises(HTTPException) as exc_info:
+        stock_main.verify_admin(x_admin_key=None)
+    assert exc_info.value.status_code == 503
+    assert "ADMIN_API_KEY" in exc_info.value.detail
+
+
+def test_verify_admin_allows_when_key_missing_with_dev_flag(monkeypatch):
+    """ADMIN_API_KEY 미설정 + ALLOW_UNAUTHENTICATED_ADMIN=true → 통과 (개발 모드)."""
+    monkeypatch.setattr(stock_main, "ADMIN_API_KEY", "")
+    monkeypatch.setenv("ALLOW_UNAUTHENTICATED_ADMIN", "true")
+    stock_main.verify_admin(x_admin_key=None)  # 예외 없으면 통과
+
+
+def test_verify_admin_rejects_wrong_key(monkeypatch):
+    """ADMIN_API_KEY 설정 + 잘못된 키 → 401 (regression)."""
+    monkeypatch.setattr(stock_main, "ADMIN_API_KEY", "secret123")
+    with pytest.raises(HTTPException) as exc_info:
+        stock_main.verify_admin(x_admin_key="wrong")
+    assert exc_info.value.status_code == 401
+
+
+def test_verify_admin_allows_correct_key(monkeypatch):
+    """ADMIN_API_KEY 설정 + 올바른 키 → 통과 (regression)."""
+    monkeypatch.setattr(stock_main, "ADMIN_API_KEY", "secret123")
+    stock_main.verify_admin(x_admin_key="secret123")  # 예외 없으면 통과

stock/tests/test_portfolio_total_buy.py

@@ -0,0 +1,77 @@
+"""포트폴리오 /api/portfolio 응답의 total_buy 계산 회귀 테스트 (CODE_REVIEW F4).
+
+purchase_price는 종목별 단가(1주당) 의미. total_buy = SUM(purchase_price × quantity).
+purchase_price가 없으면 avg_price로 폴백 후 동일하게 수량 곱산.
+"""
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+def _fake_db_setup(monkeypatch, items, cash=None):
+    from app import main as stock_main
+    monkeypatch.setattr(stock_main, "get_all_portfolio", lambda: items)
+    monkeypatch.setattr(stock_main, "get_all_broker_cash", lambda: cash or [])
+
+
+def test_portfolio_total_buy_uses_purchase_price_times_quantity(monkeypatch):
+    """purchase_price 설정 시: total_buy = purchase_price × quantity 의 합."""
+    items = [
+        {"id": 1, "broker": "KB", "ticker": "005930", "name": "삼성전자",
+         "quantity": 100, "avg_price": 72000, "purchase_price": 70000},
+    ]
+    fake_prices = {"005930": {"price": 74500, "session": "REGULAR", "as_of": "2026-05-17T10:00:00+09:00"}}
+    _fake_db_setup(monkeypatch, items)
+    from app import main as stock_main
+    monkeypatch.setattr(stock_main, "get_current_prices_detail", lambda t: fake_prices)
+
+    client = TestClient(app)
+    resp = client.get("/api/portfolio")
+    assert resp.status_code == 200
+    data = resp.json()
+    # purchase_price=70000 × quantity=100 = 7,000,000
+    assert data["summary"]["total_buy"] == 7_000_000
+
+
+def test_portfolio_total_buy_falls_back_to_avg_price_with_quantity(monkeypatch):
+    """purchase_price 미설정 시: avg_price 폴백 + 수량 곱산. API_SPEC 예시와 일치."""
+    items = [
+        {"id": 1, "broker": "KB", "ticker": "005930", "name": "삼성전자",
+         "quantity": 100, "avg_price": 72000, "purchase_price": None},
+    ]
+    fake_prices = {"005930": {"price": 74500, "session": "REGULAR", "as_of": "2026-05-17T10:00:00+09:00"}}
+    _fake_db_setup(monkeypatch, items)
+    from app import main as stock_main
+    monkeypatch.setattr(stock_main, "get_current_prices_detail", lambda t: fake_prices)
+
+    client = TestClient(app)
+    resp = client.get("/api/portfolio")
+    assert resp.status_code == 200
+    data = resp.json()
+    # avg_price=72000 × quantity=100 = 7,200,000 (API_SPEC.md 예시와 일치)
+    assert data["summary"]["total_buy"] == 7_200_000
+
+
+def test_portfolio_total_buy_sums_multiple_holdings(monkeypatch):
+    """여러 종목 합산도 단가 × 수량 합."""
+    items = [
+        {"id": 1, "broker": "KB", "ticker": "005930", "name": "삼성전자",
+         "quantity": 100, "avg_price": 70000, "purchase_price": 70000},
+        {"id": 2, "broker": "NH", "ticker": "000660", "name": "SK하이닉스",
+         "quantity": 50, "avg_price": 130000, "purchase_price": 130000},
+    ]
+    fake_prices = {
+        "005930": {"price": 74500, "session": "REGULAR", "as_of": "2026-05-17T10:00:00+09:00"},
+        "000660": {"price": 140000, "session": "REGULAR", "as_of": "2026-05-17T10:00:00+09:00"},
+    }
+    _fake_db_setup(monkeypatch, items)
+    from app import main as stock_main
+    monkeypatch.setattr(stock_main, "get_current_prices_detail", lambda t: fake_prices)
+
+    client = TestClient(app)
+    resp = client.get("/api/portfolio")
+    data = resp.json()
+    # 70000*100 + 130000*50 = 7,000,000 + 6,500,000 = 13,500,000
+    assert data["summary"]["total_buy"] == 13_500_000