fix(signal_v2-phase3a): V1 KIS env pattern + test isolation fix

config.py: kis_env_type (virtual/real) + KIS_REAL_*/KIS_VIRTUAL_* env
variables (V1 호환). kis_app_key/kis_app_secret/kis_account properties
auto-select based on env type.

main.py: KIS not-configured warning uses descriptive message including
env type + expected var name.

test_main.py: monkeypatch load_dotenv to no-op + setenv empty string
(instead of delenv) — defeats .env re-read on importlib.reload.
Pre-existing test_startup_warns_if_webai_api_key_missing also fixed.

33/33 tests pass (was 31/33).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.gitignore

@@ -47,9 +47,11 @@ daily_trade_history.json
 watchlist.json
 bot_ipc.json
 
-# Test
+# Test (top-level only; signal_v2/tests tracked separately)
 tests/
 tests/*
+!signal_v2/tests/
+!signal_v2/tests/**
 
 # System
 Thumbs.db
@@ -59,3 +61,7 @@ Desktop.ini
 KIS_SETUP.md
 # Claude Code subagent state
 .claude/
+
+# Signal V2 runtime data
+signal_v2/data/*.db
+signal_v2/data/*.db-*

requirements.txt

@@ -0,0 +1,9 @@
+# Signal V2 dependencies (added 2026-05-16, Phase 2)
+httpx>=0.27
+fastapi>=0.110
+uvicorn>=0.27
+python-dotenv>=1.0
+pytest>=8.0
+pytest-asyncio>=0.23
+respx>=0.21
+websockets>=12

signal_v2/config.py

@@ -0,0 +1,56 @@
+"""Signal V2 환경변수 로딩."""
+import os
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv(Path(__file__).parent.parent / ".env")
+
+
+@dataclass(frozen=True)
+class Settings:
+    stock_api_url: str = field(
+        default_factory=lambda: os.getenv("STOCK_API_URL", "").rstrip("/")
+    )
+    webai_api_key: str = field(
+        default_factory=lambda: os.getenv("WEBAI_API_KEY", "").strip()
+    )
+    port: int = field(default_factory=lambda: int(os.getenv("SIGNAL_V2_PORT", "8001")))
+    db_path: Path = field(
+        default_factory=lambda: Path(__file__).parent / "data" / "signal_v2.db"
+    )
+    # KIS — V1 호환 패턴 (KIS_ENV_TYPE virtual/real)
+    kis_env_type: str = field(default_factory=lambda: os.getenv("KIS_ENV_TYPE", "virtual").lower())
+    kis_real_app_key: str = field(default_factory=lambda: os.getenv("KIS_REAL_APP_KEY", "").strip())
+    kis_real_app_secret: str = field(default_factory=lambda: os.getenv("KIS_REAL_APP_SECRET", "").strip())
+    kis_real_account: str = field(default_factory=lambda: os.getenv("KIS_REAL_ACCOUNT", "").strip())
+    kis_virtual_app_key: str = field(default_factory=lambda: os.getenv("KIS_VIRTUAL_APP_KEY", "").strip())
+    kis_virtual_app_secret: str = field(default_factory=lambda: os.getenv("KIS_VIRTUAL_APP_SECRET", "").strip())
+    kis_virtual_account: str = field(default_factory=lambda: os.getenv("KIS_VIRTUAL_ACCOUNT", "").strip())
+    v1_token_path: Path = field(
+        default_factory=lambda: Path(
+            os.getenv("V1_TOKEN_PATH",
+                      str(Path(__file__).parent.parent / "signal_v1" / "data" / "kis_token.json"))
+        )
+    )
+
+    @property
+    def kis_is_virtual(self) -> bool:
+        return self.kis_env_type != "real"
+
+    @property
+    def kis_app_key(self) -> str:
+        return self.kis_real_app_key if self.kis_env_type == "real" else self.kis_virtual_app_key
+
+    @property
+    def kis_app_secret(self) -> str:
+        return self.kis_real_app_secret if self.kis_env_type == "real" else self.kis_virtual_app_secret
+
+    @property
+    def kis_account(self) -> str:
+        return self.kis_real_account if self.kis_env_type == "real" else self.kis_virtual_account
+
+
+def get_settings() -> Settings:
+    return Settings()

signal_v2/holidays.json

@@ -0,0 +1,18 @@
+[
+  "2026-01-01",
+  "2026-01-28",
+  "2026-01-29",
+  "2026-01-30",
+  "2026-03-01",
+  "2026-05-05",
+  "2026-05-25",
+  "2026-06-06",
+  "2026-08-15",
+  "2026-09-24",
+  "2026-09-25",
+  "2026-09-26",
+  "2026-10-03",
+  "2026-10-09",
+  "2026-12-25",
+  "2026-12-31"
+]

signal_v2/kis_client.py

@@ -0,0 +1,155 @@
+"""KIS REST API client — 분봉 + 호가. V1 토큰 read-only 공유."""
+from __future__ import annotations
+import asyncio
+import json
+import logging
+import time
+from datetime import datetime
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+import httpx
+
+logger = logging.getLogger(__name__)
+KST = ZoneInfo("Asia/Seoul")
+
+_MAX_ATTEMPTS = 3
+_THROTTLE_INTERVAL = 0.5  # 초당 2회 제한
+
+
+class KISClient:
+    """KIS REST (분봉 + 호가). V1 토큰 파일 read-only."""
+
+    def __init__(
+        self,
+        app_key: str, app_secret: str, account: str, is_virtual: bool,
+        v1_token_path: Path,
+        timeout: float = 10.0,
+    ):
+        self._app_key = app_key
+        self._app_secret = app_secret
+        self._account = account
+        self._is_virtual = is_virtual
+        self._v1_token_path = Path(v1_token_path)
+        self._base_url = (
+            "https://openapivts.koreainvestment.com:29443" if is_virtual
+            else "https://openapi.koreainvestment.com:9443"
+        )
+        self._client = httpx.AsyncClient(timeout=timeout)
+        self._token_cache: tuple[str, float] | None = None  # (token, file_mtime)
+        self._last_throttle_at = 0.0
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    def _read_v1_token(self) -> str:
+        if not self._v1_token_path.exists():
+            raise RuntimeError(f"V1 token file missing: {self._v1_token_path}")
+        mtime = self._v1_token_path.stat().st_mtime
+        if self._token_cache and self._token_cache[1] == mtime:
+            return self._token_cache[0]
+        data = json.loads(self._v1_token_path.read_text(encoding="utf-8"))
+        token = data.get("access_token", "")
+        if not token:
+            raise RuntimeError("V1 token file has no access_token")
+        self._token_cache = (token, mtime)
+        return token
+
+    async def _throttle(self) -> None:
+        elapsed = time.monotonic() - self._last_throttle_at
+        if elapsed < _THROTTLE_INTERVAL:
+            await asyncio.sleep(_THROTTLE_INTERVAL - elapsed)
+        self._last_throttle_at = time.monotonic()
+
+    def _common_headers(self, tr_id: str) -> dict[str, str]:
+        token = self._read_v1_token()
+        return {
+            "authorization": f"Bearer {token}",
+            "appkey": self._app_key,
+            "appsecret": self._app_secret,
+            "tr_id": tr_id,
+            "custtype": "P",
+        }
+
+    async def _request_with_retry(
+        self, method: str, path: str, tr_id: str, **kwargs,
+    ) -> dict:
+        url = f"{self._base_url}{path}"
+        headers = self._common_headers(tr_id)
+        for attempt in range(_MAX_ATTEMPTS):
+            await self._throttle()
+            try:
+                response = await self._client.request(
+                    method, url, headers=headers, **kwargs
+                )
+                if response.status_code == 429:
+                    if attempt < _MAX_ATTEMPTS - 1:
+                        await asyncio.sleep(2**attempt)
+                        continue
+                    response.raise_for_status()
+                response.raise_for_status()
+                return response.json()
+            except httpx.TimeoutException:
+                if attempt < _MAX_ATTEMPTS - 1:
+                    await asyncio.sleep(2**attempt)
+                    continue
+                raise
+        raise RuntimeError("retry exhausted")
+
+    async def get_minute_ohlcv(self, ticker: str) -> list[dict]:
+        """현재 시점 직전 30개 1분봉 OHLCV (TR_ID FHKST03010200)."""
+        path = "/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
+        params = {
+            "FID_ETC_CLS_CODE": "",
+            "FID_COND_MRKT_DIV_CODE": "J",
+            "FID_INPUT_ISCD": ticker,
+            "FID_INPUT_HOUR_1": datetime.now(KST).strftime("%H%M%S"),
+            "FID_PW_DATA_INCU_YN": "N",
+        }
+        raw = await self._request_with_retry(
+            "GET", path, tr_id="FHKST03010200", params=params,
+        )
+        output2 = raw.get("output2", [])
+        bars = []
+        for row in output2:
+            try:
+                date = row["stck_bsop_date"]
+                hhmmss = row["stck_cntg_hour"]
+                dt = datetime.strptime(f"{date} {hhmmss}", "%Y%m%d %H%M%S").replace(tzinfo=KST)
+                bars.append({
+                    "datetime": dt.isoformat(),
+                    "open": int(row["stck_oprc"]),
+                    "high": int(row["stck_hgpr"]),
+                    "low": int(row["stck_lwpr"]),
+                    "close": int(row["stck_prpr"]),
+                    "volume": int(row["cntg_vol"]),
+                })
+            except (KeyError, ValueError) as e:
+                logger.warning("skip malformed bar for %s: %r", ticker, e)
+        # KIS returns descending; reverse to ascending (most recent last)
+        bars.reverse()
+        return bars
+
+    async def get_asking_price(self, ticker: str) -> dict:
+        """현재 호가 + 매수/매도 잔량 (TR_ID FHKST01010200)."""
+        path = "/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn"
+        params = {
+            "FID_COND_MRKT_DIV_CODE": "J",
+            "FID_INPUT_ISCD": ticker,
+        }
+        raw = await self._request_with_retry(
+            "GET", path, tr_id="FHKST01010200", params=params,
+        )
+        output1 = raw.get("output1", {})
+        bid_total = int(output1.get("total_bidp_rsqn", 0))
+        ask_total = int(output1.get("total_askp_rsqn", 0))
+        total = bid_total + ask_total
+        bid_ratio = bid_total / total if total > 0 else 0.0
+        current_price = int(output1.get("stck_prpr", 0))
+        return {
+            "bid_total": bid_total,
+            "ask_total": ask_total,
+            "bid_ratio": bid_ratio,
+            "current_price": current_price,
+            "as_of": datetime.now(KST).isoformat(),
+        }

signal_v2/kis_websocket.py

@@ -0,0 +1,186 @@
+"""KIS WebSocket — approval_key + 실시간 호가 구독."""
+from __future__ import annotations
+import asyncio
+import json
+import logging
+from datetime import datetime
+from typing import Callable
+from zoneinfo import ZoneInfo
+
+import httpx
+import websockets
+
+logger = logging.getLogger(__name__)
+KST = ZoneInfo("Asia/Seoul")
+
+# KIS 호가 메시지 필드 인덱스 (운영 환경 검증 필요)
+# H0STASP0 응답: ticker | time | current_price | ... | ask_total | bid_total
+# 본 spec/plan 의 가정: 마지막 2개 필드가 ask_total / bid_total
+_ASKING_TICKER_IDX = 0
+_ASKING_TIME_IDX = 1
+_ASKING_CURRENT_PRICE_IDX = 2
+_ASKING_TOTAL_ASK_IDX = -2
+_ASKING_TOTAL_BID_IDX = -1
+
+
+class KISWebSocket:
+    """KIS WebSocket client. approval_key 발급 + 호가 실시간."""
+
+    def __init__(self, app_key: str, app_secret: str, is_virtual: bool):
+        self._app_key = app_key
+        self._app_secret = app_secret
+        self._is_virtual = is_virtual
+        self._base_rest = (
+            "https://openapivts.koreainvestment.com:29443" if is_virtual
+            else "https://openapi.koreainvestment.com:9443"
+        )
+        self._ws_url = (
+            "ws://ops.koreainvestment.com:31000" if is_virtual
+            else "ws://ops.koreainvestment.com:21000"
+        )
+        self._approval_key: str | None = None
+        self._ws = None
+        self._subscriptions: set[str] = set()
+        self._on_asking_price: Callable[[str, dict], None] | None = None
+        self._recv_task: asyncio.Task | None = None
+        self._shutdown = asyncio.Event()
+
+    async def _fetch_approval_key(self) -> str:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.post(
+                f"{self._base_rest}/oauth2/Approval",
+                json={
+                    "grant_type": "client_credentials",
+                    "appkey": self._app_key,
+                    "secretkey": self._app_secret,
+                },
+            )
+            response.raise_for_status()
+            data = response.json()
+            self._approval_key = data["approval_key"]
+            return self._approval_key
+
+    async def _connect(self):
+        return await websockets.connect(self._ws_url)
+
+    async def _connect_with_backoff(self):
+        """연결 시도 with exponential backoff (1s → 2s → 4s → max 30s)."""
+        for attempt in range(10):
+            try:
+                ws = await self._connect()
+                return ws
+            except Exception as e:
+                wait = min(2**attempt, 30)
+                logger.warning(
+                    "KIS WebSocket connect failed (attempt %d): %r — retrying in %ds",
+                    attempt + 1, e, wait,
+                )
+                await asyncio.sleep(wait)
+        raise RuntimeError("KIS WebSocket connect exhausted retries")
+
+    async def start(
+        self, tickers: list[str],
+        on_asking_price: Callable[[str, dict], None],
+    ) -> None:
+        if self._approval_key is None:
+            await self._fetch_approval_key()
+        self._on_asking_price = on_asking_price
+        self._ws = await self._connect_with_backoff()
+        for ticker in tickers:
+            await self.subscribe(ticker)
+        self._recv_task = asyncio.create_task(self._receive_loop())
+
+    async def subscribe(self, ticker: str) -> None:
+        if self._ws is None or self._approval_key is None:
+            raise RuntimeError("KIS WebSocket not started")
+        msg = json.dumps({
+            "header": {
+                "approval_key": self._approval_key,
+                "custtype": "P",
+                "tr_type": "1",
+                "content-type": "utf-8",
+            },
+            "body": {
+                "input": {"tr_id": "H0STASP0", "tr_key": ticker},
+            },
+        })
+        await self._ws.send(msg)
+        self._subscriptions.add(ticker)
+
+    async def unsubscribe(self, ticker: str) -> None:
+        if self._ws is None or self._approval_key is None:
+            return
+        msg = json.dumps({
+            "header": {
+                "approval_key": self._approval_key,
+                "custtype": "P",
+                "tr_type": "2",
+                "content-type": "utf-8",
+            },
+            "body": {
+                "input": {"tr_id": "H0STASP0", "tr_key": ticker},
+            },
+        })
+        await self._ws.send(msg)
+        self._subscriptions.discard(ticker)
+
+    async def close(self) -> None:
+        self._shutdown.set()
+        if self._recv_task is not None:
+            self._recv_task.cancel()
+            try:
+                await self._recv_task
+            except asyncio.CancelledError:
+                pass
+        if self._ws is not None:
+            await self._ws.close()
+
+    async def _receive_loop(self) -> None:
+        while not self._shutdown.is_set():
+            try:
+                raw = await self._ws.recv()
+            except websockets.ConnectionClosed:
+                logger.warning("KIS WebSocket closed — reconnecting")
+                self._ws = await self._connect_with_backoff()
+                for ticker in list(self._subscriptions):
+                    await self.subscribe(ticker)
+                continue
+            if not isinstance(raw, str):
+                continue
+            parsed = self._parse_asking_price(raw)
+            if parsed is not None and self._on_asking_price is not None:
+                ticker, data = parsed
+                try:
+                    self._on_asking_price(ticker, data)
+                except Exception:
+                    logger.exception("on_asking_price callback failed")
+
+    def _parse_asking_price(self, raw: str) -> tuple[str, dict] | None:
+        """KIS H0STASP0 raw → (ticker, asking_price dict).
+
+        Raw format: '0|H0STASP0|<count>|<data>' where data = '^'-joined fields.
+        Field indices (운영 검증 필요): 마지막 2개 가정 (ask, bid).
+        """
+        try:
+            parts = raw.split("|")
+            if len(parts) < 4 or parts[1] != "H0STASP0":
+                return None
+            fields = parts[3].split("^")
+            ticker = fields[_ASKING_TICKER_IDX]
+            current_price_str = fields[_ASKING_CURRENT_PRICE_IDX]
+            current_price = int(current_price_str) if current_price_str.lstrip("-").isdigit() else 0
+            ask_str = fields[_ASKING_TOTAL_ASK_IDX]
+            bid_str = fields[_ASKING_TOTAL_BID_IDX]
+            ask_total = int(ask_str) if ask_str.lstrip("-").isdigit() else 0
+            bid_total = int(bid_str) if bid_str.lstrip("-").isdigit() else 0
+            total = bid_total + ask_total
+            return ticker, {
+                "bid_total": bid_total,
+                "ask_total": ask_total,
+                "bid_ratio": bid_total / total if total > 0 else 0.0,
+                "current_price": current_price,
+                "as_of": datetime.now(KST).isoformat(),
+            }
+        except (IndexError, ValueError) as e:
+            logger.warning("parse_asking_price failed: %r", e)
+            return None

signal_v2/main.py

@@ -0,0 +1,114 @@
+"""FastAPI app — Signal V2 Pull Worker."""
+from __future__ import annotations
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from signal_v2 import state as state_mod
+from signal_v2.config import get_settings
+from signal_v2.kis_client import KISClient
+from signal_v2.kis_websocket import KISWebSocket
+from signal_v2.pull_worker import poll_loop, make_asking_price_callback
+from signal_v2.rate_limit import SignalDedup
+from signal_v2.stock_client import StockClient
+
+logger = logging.getLogger(__name__)
+
+
+class AppContext:
+    client: StockClient | None = None
+    dedup: SignalDedup | None = None
+    shutdown: asyncio.Event | None = None
+    poll_task: asyncio.Task | None = None
+    kis_client: KISClient | None = None
+    kis_ws: KISWebSocket | None = None
+
+
+_ctx = AppContext()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    settings = get_settings()
+    if not settings.webai_api_key:
+        logger.warning(
+            "WEBAI_API_KEY not configured — stock API calls will fail with 401"
+        )
+    if not settings.kis_app_key:
+        logger.warning(
+            "KIS app_key not configured (KIS_ENV_TYPE=%s, KIS_%s_APP_KEY missing) — KIS REST/WebSocket disabled",
+            settings.kis_env_type, settings.kis_env_type.upper()
+        )
+
+    _ctx.client = StockClient(settings.stock_api_url, settings.webai_api_key)
+    _ctx.dedup = SignalDedup(settings.db_path)
+    _ctx.shutdown = asyncio.Event()
+
+    # KIS only if app_key configured
+    if settings.kis_app_key:
+        _ctx.kis_client = KISClient(
+            app_key=settings.kis_app_key,
+            app_secret=settings.kis_app_secret,
+            account=settings.kis_account,
+            is_virtual=settings.kis_is_virtual,
+            v1_token_path=settings.v1_token_path,
+        )
+        _ctx.kis_ws = KISWebSocket(
+            app_key=settings.kis_app_key,
+            app_secret=settings.kis_app_secret,
+            is_virtual=settings.kis_is_virtual,
+        )
+        # Subscribe portfolio holdings (if any)
+        try:
+            portfolio = await _ctx.client.get_portfolio()
+            tickers = [h["ticker"] for h in portfolio.get("holdings", []) if "ticker" in h]
+            cb = make_asking_price_callback(state_mod.state)
+            await _ctx.kis_ws.start(tickers, cb)
+        except Exception:
+            logger.exception("KIS WebSocket startup failed — continuing without realtime asking_price")
+
+    _ctx.poll_task = asyncio.create_task(
+        poll_loop(
+            _ctx.client, state_mod.state, _ctx.shutdown,
+            kis_client=_ctx.kis_client,
+        )
+    )
+
+    yield
+
+    # Shutdown
+    if _ctx.shutdown is not None:
+        _ctx.shutdown.set()
+    if _ctx.poll_task is not None:
+        try:
+            await asyncio.wait_for(_ctx.poll_task, timeout=5.0)
+        except asyncio.TimeoutError:
+            _ctx.poll_task.cancel()
+            try:
+                await _ctx.poll_task
+            except asyncio.CancelledError:
+                pass
+    if _ctx.kis_ws is not None:
+        await _ctx.kis_ws.close()
+    if _ctx.kis_client is not None:
+        await _ctx.kis_client.close()
+    if _ctx.client is not None:
+        await _ctx.client.close()
+
+
+app = FastAPI(
+    title="Signal V2 Pull Worker", version="0.1.0", lifespan=lifespan
+)
+
+
+@app.get("/health")
+async def health():
+    settings = get_settings()
+    return {
+        "status": "online",
+        "stock_api_url": settings.stock_api_url,
+        "last_poll": state_mod.state.last_updated,
+        "cache_size": _ctx.client.cache_size() if _ctx.client is not None else 0,
+    }

signal_v2/pull_worker.py

@@ -0,0 +1,127 @@
+"""Polling loop — async cron + state update."""
+from __future__ import annotations
+import asyncio
+import logging
+from collections import deque
+from datetime import datetime
+
+from signal_v2.kis_client import KISClient
+from signal_v2.scheduler import (
+    KST, _is_market_day, _is_polling_window, _next_interval,
+)
+from signal_v2.state import PollState
+from signal_v2.stock_client import StockClient
+
+logger = logging.getLogger(__name__)
+
+
+async def poll_loop(
+    client: StockClient, state: PollState, shutdown: asyncio.Event,
+    kis_client: KISClient | None = None,
+) -> None:
+    """FastAPI lifespan 에서 asyncio.create_task 로 시작."""
+    logger.info("poll_loop started")
+    while not shutdown.is_set():
+        now = datetime.now(KST)
+        if _is_market_day(now) and _is_polling_window(now):
+            try:
+                await _run_polling_cycle(client, state, kis_client=kis_client)
+            except Exception:
+                logger.exception("poll cycle failed")
+        interval = _next_interval(now)
+        try:
+            await asyncio.wait_for(shutdown.wait(), timeout=interval)
+            break
+        except asyncio.TimeoutError:
+            continue
+    logger.info("poll_loop ended")
+
+
+async def _run_polling_cycle(
+    client: StockClient, state: PollState,
+    kis_client: KISClient | None = None,
+) -> None:
+    """기존 3 endpoint (stock) + KIS 분봉 fetch."""
+    portfolio, sentiment, screener = await asyncio.gather(
+        client.get_portfolio(),
+        client.get_news_sentiment(),
+        client.run_screener_preview(),
+        return_exceptions=True,
+    )
+    now_iso = datetime.now(KST).isoformat()
+
+    for name, result in (
+        ("portfolio", portfolio),
+        ("news_sentiment", sentiment),
+        ("screener_preview", screener),
+    ):
+        if isinstance(result, dict):
+            setattr(state, name, result)
+            state.last_updated[name] = now_iso
+            state.fetch_errors[name] = 0
+        else:
+            state.fetch_errors[name] = state.fetch_errors.get(name, 0) + 1
+            logger.warning("fetch %s failed: %r", name, result)
+
+    # KIS 분봉 + 호가 (kis_client 주어졌을 때만)
+    if kis_client is not None:
+        try:
+            await _run_kis_minute_cycle(kis_client, state)
+        except Exception:
+            logger.exception("kis minute cycle failed")
+
+
+async def _run_kis_minute_cycle(kis_client: KISClient, state: PollState) -> None:
+    """KIS 분봉 + 호가 fetch + state 갱신.
+
+    - 분봉: portfolio + screener Top-N union 종목 모두
+    - 호가 (REST): screener-only 종목 (portfolio 는 WebSocket 으로 들어옴)
+    """
+    portfolio_tickers = _portfolio_tickers(state)
+    screener_tickers = _screener_tickers(state)
+    all_tickers = list(set(portfolio_tickers) | set(screener_tickers))
+
+    # 분봉 fetch (병렬)
+    minute_results = await asyncio.gather(*[
+        kis_client.get_minute_ohlcv(t) for t in all_tickers
+    ], return_exceptions=True)
+    now_iso = datetime.now(KST).isoformat()
+    for ticker, result in zip(all_tickers, minute_results):
+        if isinstance(result, list):
+            buf = state.minute_bars.setdefault(ticker, deque(maxlen=60))
+            buf.extend(result)
+            state.last_updated[f"minute_bars/{ticker}"] = now_iso
+        else:
+            state.fetch_errors[f"minute_bars/{ticker}"] = (
+                state.fetch_errors.get(f"minute_bars/{ticker}", 0) + 1
+            )
+
+    # 호가 fetch (REST) — screener-only
+    screener_only = list(set(screener_tickers) - set(portfolio_tickers))
+    asking_results = await asyncio.gather(*[
+        kis_client.get_asking_price(t) for t in screener_only
+    ], return_exceptions=True)
+    for ticker, result in zip(screener_only, asking_results):
+        if isinstance(result, dict):
+            state.asking_price[ticker] = result
+            state.last_updated[f"asking_price/{ticker}"] = now_iso
+
+
+def make_asking_price_callback(state: PollState):
+    """KIS WebSocket on_asking_price callback factory."""
+    def _cb(ticker: str, data: dict) -> None:
+        state.asking_price[ticker] = data
+        state.last_updated[f"asking_price/{ticker}"] = datetime.now(KST).isoformat()
+    return _cb
+
+
+def _portfolio_tickers(state: PollState) -> list[str]:
+    if state.portfolio is None:
+        return []
+    return [h["ticker"] for h in state.portfolio.get("holdings", []) if "ticker" in h]
+
+
+def _screener_tickers(state: PollState) -> list[str]:
+    if state.screener_preview is None:
+        return []
+    return [i["ticker"] for i in state.screener_preview.get("items", []) if "ticker" in i]

signal_v2/pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests

signal_v2/rate_limit.py

@@ -0,0 +1,73 @@
+"""SignalDedup — SQLite-backed 24h duplicate signal blocker."""
+from __future__ import annotations
+import sqlite3
+from contextlib import contextmanager
+from datetime import datetime, timedelta
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+KST = ZoneInfo("Asia/Seoul")
+
+
+def _now_iso() -> str:
+    """Test seam — overridable via monkeypatch."""
+    return datetime.now(KST).isoformat()
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS signal_dedup (
+    ticker      TEXT NOT NULL,
+    action      TEXT NOT NULL,
+    last_sent   TEXT NOT NULL,
+    confidence  REAL NOT NULL,
+    PRIMARY KEY (ticker, action)
+);
+CREATE INDEX IF NOT EXISTS idx_signal_dedup_last_sent
+    ON signal_dedup(last_sent);
+"""
+
+
+class SignalDedup:
+    """24h dedup interface. WAL + busy_timeout=120000."""
+
+    def __init__(self, db_path: Path):
+        self._db_path = Path(db_path)
+        self._db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._init_schema()
+
+    @contextmanager
+    def _conn(self):
+        conn = sqlite3.connect(self._db_path, timeout=120.0)
+        try:
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute("PRAGMA busy_timeout=120000")
+            yield conn
+        finally:
+            conn.close()
+
+    def _init_schema(self) -> None:
+        with self._conn() as conn:
+            conn.executescript(_SCHEMA)
+            conn.commit()
+
+    def is_recent(self, ticker: str, action: str, within_hours: int = 24) -> bool:
+        threshold_dt = datetime.fromisoformat(_now_iso()) - timedelta(hours=within_hours)
+        threshold_iso = threshold_dt.isoformat()
+        with self._conn() as conn:
+            row = conn.execute(
+                "SELECT last_sent FROM signal_dedup WHERE ticker = ? AND action = ?",
+                (ticker, action),
+            ).fetchone()
+        return row is not None and row[0] >= threshold_iso
+
+    def record(self, ticker: str, action: str, confidence: float) -> None:
+        with self._conn() as conn:
+            conn.execute(
+                """INSERT INTO signal_dedup (ticker, action, last_sent, confidence)
+                   VALUES (?, ?, ?, ?)
+                   ON CONFLICT (ticker, action) DO UPDATE
+                   SET last_sent = excluded.last_sent,
+                       confidence = excluded.confidence""",
+                (ticker, action, _now_iso(), confidence),
+            )
+            conn.commit()

signal_v2/scheduler.py

@@ -0,0 +1,91 @@
+"""Polling scheduler — 시간대별 분기 + 휴장일 처리."""
+from __future__ import annotations
+import json
+import logging
+from datetime import datetime, timedelta, time
+from pathlib import Path
+from zoneinfo import ZoneInfo
+
+logger = logging.getLogger(__name__)
+
+KST = ZoneInfo("Asia/Seoul")
+_HOLIDAYS_PATH = Path(__file__).parent / "holidays.json"
+_HOLIDAYS: set[str] = set(json.loads(_HOLIDAYS_PATH.read_text(encoding="utf-8")))
+
+# Market windows (정규장)
+_PRE_OPEN = time(7, 0)
+_OPEN = time(9, 0)
+_CLOSE = time(15, 30)
+_POST_END = time(20, 0)
+
+# NXT windows (시간외)
+_NXT_PRE_END = time(23, 30)
+_NXT_POST_OPEN = time(4, 30)
+# 23:30 - 04:30 (dead zone) skip
+
+
+def _is_market_day(now: datetime) -> bool:
+    """평일 + 휴장일 아닌 날."""
+    if now.weekday() >= 5:  # Sat/Sun
+        return False
+    return now.strftime("%Y-%m-%d") not in _HOLIDAYS
+
+
+def _is_polling_window(now: datetime) -> bool:
+    """폴링 윈도우: 07:00-23:30 + 04:30-07:00."""
+    t = now.time()
+    return (
+        (_PRE_OPEN <= t < _NXT_PRE_END)
+        or (_NXT_POST_OPEN <= t < _PRE_OPEN)
+    )
+
+
+def _next_interval(now: datetime) -> float:
+    """다음 폴링까지 sleep 초수."""
+    if not _is_market_day(now):
+        return _seconds_until_next_market_open(now)
+
+    t = now.time()
+    if _PRE_OPEN <= t < _OPEN:
+        return 300.0  # 장전 5분
+    elif _OPEN <= t < _CLOSE:
+        return 60.0   # 장중 1분
+    elif _CLOSE <= t < _POST_END:
+        return 300.0  # 장후 5분
+    elif _POST_END <= t < _NXT_PRE_END:
+        return 300.0  # NXT 야간 5분
+    elif _NXT_POST_OPEN <= t < _PRE_OPEN:
+        return 300.0  # NXT 새벽 5분
+    else:
+        # Dead zone (23:30-04:30) — wait until next 04:30
+        return _seconds_until_nxt_or_market_open(now)
+
+
+def _seconds_until_nxt_or_market_open(now: datetime) -> float:
+    """다음 04:30 (NXT 새벽 start) 까지 초수. 휴장일은 다음 영업일 07:00."""
+    candidate = now.replace(hour=4, minute=30, second=0, microsecond=0)
+    if candidate <= now:
+        candidate += timedelta(days=1)
+
+    for _ in range(14):
+        if _is_market_day(candidate):
+            return (candidate - now).total_seconds()
+        candidate += timedelta(days=1)
+
+    logger.warning("could not find next market day within 14 days")
+    return 86400.0
+
+
+def _seconds_until_next_market_open(now: datetime) -> float:
+    """다음 영업일의 07:00 KST 까지 초수 (휴장일/주말용)."""
+    candidate = now.replace(hour=7, minute=0, second=0, microsecond=0)
+    if candidate <= now:
+        candidate += timedelta(days=1)
+
+    for _ in range(14):  # safety bound (max 2 weeks of holidays)
+        if _is_market_day(candidate):
+            return (candidate - now).total_seconds()
+        candidate += timedelta(days=1)
+
+    logger.warning("could not find next market day within 14 days")
+    return 86400.0

signal_v2/start.bat

@@ -0,0 +1,3 @@
+@echo off
+cd /d "%~dp0\.."
+python -m uvicorn signal_v2.main:app --host 0.0.0.0 --port 8001

signal_v2/state.py

@@ -0,0 +1,18 @@
+"""PollState — process-wide singleton."""
+from collections import deque
+from dataclasses import dataclass, field
+
+
+@dataclass
+class PollState:
+    portfolio: dict | None = None
+    news_sentiment: dict | None = None
+    screener_preview: dict | None = None
+    # Phase 3a additions
+    minute_bars: dict[str, deque] = field(default_factory=dict)
+    asking_price: dict[str, dict] = field(default_factory=dict)
+    last_updated: dict[str, str] = field(default_factory=dict)
+    fetch_errors: dict[str, int] = field(default_factory=dict)
+
+
+state = PollState()

signal_v2/stock_client.py

@@ -0,0 +1,128 @@
+"""Stock API HTTP client — async httpx + retry + memory cache."""
+from __future__ import annotations
+import asyncio
+import logging
+import time
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Cache TTL by endpoint (seconds)
+_TTL = {
+    "portfolio": 60.0,
+    "news-sentiment": 300.0,
+    "screener-preview": 60.0,
+}
+
+# Retry policy
+_MAX_ATTEMPTS = 3
+_RETRY_STATUSES = {429, 500, 502, 503, 504}
+
+
+class StockClient:
+    """stock API wrapper. Async httpx + self-retry + memory cache."""
+
+    def __init__(self, base_url: str, api_key: str, timeout: float = 10.0):
+        self._base_url = base_url.rstrip("/")
+        self._api_key = api_key
+        self._client = httpx.AsyncClient(timeout=timeout)
+        # cache: key → (data, timestamp_monotonic)
+        self._cache: dict[str, tuple[Any, float]] = {}
+
+    async def close(self) -> None:
+        await self._client.aclose()
+
+    def cache_size(self) -> int:
+        """Number of cached endpoint responses (public surface for /health)."""
+        return len(self._cache)
+
+    async def get_portfolio(self) -> dict:
+        return await self._cached_request(
+            "portfolio", "GET", "/api/webai/portfolio"
+        )
+
+    async def get_news_sentiment(self, date: str | None = None) -> dict:
+        path = "/api/webai/news-sentiment"
+        if date is not None:
+            path += f"?date={date}"
+        cache_key = f"news-sentiment:{date or 'latest'}"
+        return await self._cached_request(
+            cache_key, "GET", path, _ttl_key="news-sentiment"
+        )
+
+    async def run_screener_preview(
+        self, weights: dict | None = None, top_n: int = 20
+    ) -> dict:
+        body = {"mode": "preview", "top_n": top_n}
+        if weights is not None:
+            body["weights"] = weights
+        return await self._cached_request(
+            "screener-preview",
+            "POST",
+            "/api/stock/screener/run",
+            json=body,
+            _ttl_key="screener-preview",
+        )
+
+    async def _cached_request(
+        self,
+        cache_key: str,
+        method: str,
+        path: str,
+        *,
+        _ttl_key: str | None = None,
+        **kwargs,
+    ) -> dict:
+        ttl_key = _ttl_key or cache_key
+        ttl = _TTL.get(ttl_key, 60.0)
+        # Fresh cache hit?
+        if cache_key in self._cache:
+            data, ts = self._cache[cache_key]
+            if time.monotonic() - ts < ttl:
+                return data
+
+        # Fetch (with retry)
+        try:
+            data = await self._request_with_retry(method, path, **kwargs)
+            self._cache[cache_key] = (data, time.monotonic())
+            return data
+        except httpx.HTTPError:
+            # Stale fallback: serve old cached value if exists
+            if cache_key in self._cache:
+                stale_data, stale_ts = self._cache[cache_key]
+                age = time.monotonic() - stale_ts
+                logger.warning(
+                    "serving stale cache for %s (age=%.1fs)", cache_key, age
+                )
+                return stale_data
+            raise
+
+    async def _request_with_retry(self, method: str, path: str, **kwargs) -> dict:
+        url = f"{self._base_url}{path}"
+        headers = self._auth_headers()
+        for attempt in range(_MAX_ATTEMPTS):
+            try:
+                response = await self._client.request(
+                    method, url, headers=headers, **kwargs
+                )
+                if response.status_code in _RETRY_STATUSES:
+                    if attempt < _MAX_ATTEMPTS - 1:
+                        await asyncio.sleep(2**attempt)
+                        continue
+                    response.raise_for_status()
+                response.raise_for_status()
+                return response.json()
+            except httpx.TimeoutException:
+                if attempt < _MAX_ATTEMPTS - 1:
+                    await asyncio.sleep(2**attempt)
+                    continue
+                raise
+            except httpx.HTTPStatusError:
+                raise
+        # Unreachable: every iteration either returns or raises
+        raise RuntimeError("_request_with_retry exhausted loop without raising")
+
+    def _auth_headers(self) -> dict[str, str]:
+        return {"X-WebAI-Key": self._api_key}

signal_v2/tests/conftest.py

@@ -0,0 +1,18 @@
+"""Pytest fixtures for signal_v2 tests."""
+from pathlib import Path
+
+import pytest
+import respx
+
+
+@pytest.fixture
+def tmp_dedup_db(tmp_path) -> Path:
+    """SQLite 단위 테스트용 임시 DB path."""
+    return tmp_path / "test_signal_v2.db"
+
+
+@pytest.fixture
+def mock_stock_api():
+    """respx 로 stock API mock. base_url 은 테스트마다 임의."""
+    with respx.mock(base_url="https://test.stock.local", assert_all_called=False) as mock:
+        yield mock

signal_v2/tests/test_kis_client.py

@@ -0,0 +1,128 @@
+"""Tests for KISClient (REST)."""
+import json
+from pathlib import Path
+
+import httpx
+import pytest
+import respx
+
+from signal_v2.kis_client import KISClient
+
+
+@pytest.fixture
+def fake_v1_token(tmp_path):
+    """V1 토큰 파일 fixture."""
+    token_file = tmp_path / "kis_token.json"
+    token_file.write_text(json.dumps({
+        "access_token": "test-kis-token-abc123",
+        "token_expired": "2099-12-31 23:59:59",
+    }))
+    return token_file
+
+
+@pytest.fixture
+def kis_client_factory(fake_v1_token):
+    def _make():
+        return KISClient(
+            app_key="test-app-key",
+            app_secret="test-app-secret",
+            account="50000000-01",
+            is_virtual=True,
+            v1_token_path=fake_v1_token,
+        )
+    return _make
+
+
+@respx.mock
+async def test_get_minute_ohlcv_normal_returns_30_bars(kis_client_factory):
+    """정상 200 → 30개 분봉 list 반환."""
+    sample_output2 = [
+        {
+            "stck_bsop_date": "20260518",
+            "stck_cntg_hour": f"09{m:02d}00",
+            "stck_oprc": "78000", "stck_hgpr": "78500",
+            "stck_lwpr": "77800", "stck_prpr": "78300",
+            "cntg_vol": "12345",
+        }
+        for m in range(30)  # 9:00-9:29 = 30 bars
+    ]
+    respx.get(
+        "https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
+    ).mock(
+        return_value=httpx.Response(200, json={"output2": sample_output2})
+    )
+
+    client = kis_client_factory()
+    try:
+        bars = await client.get_minute_ohlcv("005930")
+        assert len(bars) == 30
+        assert bars[0]["close"] == 78300
+        assert "datetime" in bars[0]
+    finally:
+        await client.close()
+
+
+@respx.mock
+async def test_get_minute_ohlcv_429_retry_then_success(kis_client_factory, monkeypatch):
+    """429 → exponential backoff → 200."""
+    sleep_calls = []
+    async def fake_sleep(s): sleep_calls.append(s)
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    respx.get(
+        "https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
+    ).mock(side_effect=[
+        httpx.Response(429, text="rate limit"),
+        httpx.Response(200, json={"output2": []}),
+    ])
+    client = kis_client_factory()
+    try:
+        result = await client.get_minute_ohlcv("005930")
+        assert result == []
+        assert 1 in sleep_calls
+    finally:
+        await client.close()
+
+
+@respx.mock
+async def test_get_minute_ohlcv_uses_v1_token(kis_client_factory, fake_v1_token):
+    """KIS 호출 헤더에 V1 토큰 파일의 access_token 사용."""
+    route = respx.get(
+        "https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-time-itemchartprice"
+    ).mock(return_value=httpx.Response(200, json={"output2": []}))
+
+    client = kis_client_factory()
+    try:
+        await client.get_minute_ohlcv("005930")
+        assert route.called
+        req = route.calls.last.request
+        # check authorization header contains the V1 token
+        auth = req.headers.get("authorization", "")
+        assert "test-kis-token-abc123" in auth
+    finally:
+        await client.close()
+
+
+@respx.mock
+async def test_get_asking_price_computes_bid_ratio(kis_client_factory):
+    """호가 응답 → bid_total/(bid+ask) bid_ratio 계산."""
+    respx.get(
+        "https://openapivts.koreainvestment.com:29443/uapi/domestic-stock/v1/quotations/inquire-asking-price-exp-ccn"
+    ).mock(return_value=httpx.Response(200, json={
+        "output1": {
+            "total_bidp_rsqn": "600",
+            "total_askp_rsqn": "400",
+            "stck_prpr": "78500",
+        }
+    }))
+
+    client = kis_client_factory()
+    try:
+        data = await client.get_asking_price("005930")
+        assert data["bid_total"] == 600
+        assert data["ask_total"] == 400
+        assert abs(data["bid_ratio"] - 0.6) < 1e-9
+        assert data["current_price"] == 78500
+        assert "as_of" in data
+    finally:
+        await client.close()

signal_v2/tests/test_kis_websocket.py

@@ -0,0 +1,94 @@
+"""Tests for KISWebSocket."""
+import asyncio
+import json
+from unittest.mock import AsyncMock, MagicMock
+
+import httpx
+import pytest
+import respx
+
+from signal_v2.kis_websocket import KISWebSocket
+
+
+BASE_REST = "https://openapivts.koreainvestment.com:29443"
+
+
+@respx.mock
+async def test_fetch_approval_key_via_oauth_endpoint():
+    """POST /oauth2/Approval → approval_key 추출."""
+    respx.post(f"{BASE_REST}/oauth2/Approval").mock(
+        return_value=httpx.Response(200, json={"approval_key": "test-approval-key-xyz"})
+    )
+    ws = KISWebSocket(app_key="k", app_secret="s", is_virtual=True)
+    key = await ws._fetch_approval_key()
+    assert key == "test-approval-key-xyz"
+    assert ws._approval_key == "test-approval-key-xyz"
+
+
+async def test_subscribe_sends_h0stasp0_message():
+    """subscribe() → WebSocket 으로 H0STASP0 구독 메시지 전송."""
+    sent_messages = []
+    mock_ws = AsyncMock()
+    mock_ws.send = AsyncMock(side_effect=lambda m: sent_messages.append(m))
+
+    ws = KISWebSocket(app_key="k", app_secret="s", is_virtual=True)
+    ws._approval_key = "test-key"
+    ws._ws = mock_ws
+    await ws.subscribe("005930")
+    assert ws._subscriptions == {"005930"}
+    assert len(sent_messages) == 1
+    msg = json.loads(sent_messages[0])
+    assert msg["header"]["tr_type"] == "1"  # subscribe
+    assert msg["body"]["input"]["tr_id"] == "H0STASP0"
+    assert msg["body"]["input"]["tr_key"] == "005930"
+
+
+def test_parse_asking_price_extracts_bid_ask_totals():
+    """KIS raw '0|H0STASP0|001|...' → (ticker, dict).
+
+    KIS 호가 메시지 형식 — KIS 공식 spec 의 정확한 필드 인덱스 운영 검증 필요.
+    본 테스트는 implementer 의 _parse_asking_price 구현 인덱스에 맞춰서 sample 작성.
+    """
+    ws = KISWebSocket(app_key="k", app_secret="s", is_virtual=True)
+    # Build a sample raw message — implementer 가 _ASKING_TOTAL_BID/ASK 인덱스에
+    # 맞춰서 필드 배치하면 됨. 예: 마지막 2개 필드를 bid_total / ask_total 로.
+    fields = ["005930", "091500", "78500"]  # ticker, time, current_price
+    fields.extend(["0"] * 40)  # padding (KIS 의 실 필드 수 ~50개)
+    fields.append("400")  # ask_total
+    fields.append("600")  # bid_total
+    raw = f"0|H0STASP0|001|{'^'.join(fields)}"
+
+    result = ws._parse_asking_price(raw)
+    assert result is not None, "parse_asking_price returned None"
+    ticker, data = result
+    assert ticker == "005930"
+    assert "bid_total" in data
+    assert "ask_total" in data
+    assert "bid_ratio" in data
+    assert "current_price" in data
+    # bid_total=600, ask_total=400, bid_ratio=0.6
+    assert data["bid_total"] == 600
+    assert data["ask_total"] == 400
+    assert abs(data["bid_ratio"] - 0.6) < 1e-9
+
+
+async def test_reconnect_on_disconnect_with_backoff(monkeypatch):
+    """연결 끊김 → exponential backoff retry. _connect_with_backoff() 검증."""
+    sleep_calls = []
+    async def fake_sleep(s): sleep_calls.append(s)
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    ws = KISWebSocket(app_key="k", app_secret="s", is_virtual=True)
+    # Mock _connect to fail twice then succeed
+    call_count = [0]
+    async def fake_connect():
+        call_count[0] += 1
+        if call_count[0] < 3:
+            raise ConnectionError("fake disconnect")
+        return AsyncMock()
+    monkeypatch.setattr(ws, "_connect", fake_connect)
+
+    result = await ws._connect_with_backoff()
+    assert call_count[0] == 3  # 2 fails + 1 success
+    # exponential 1s, 2s
+    assert sleep_calls[:2] == [1, 2]

signal_v2/tests/test_main.py

@@ -0,0 +1,62 @@
+"""Tests for FastAPI main app."""
+import logging
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+def test_health_endpoint_returns_status_online(monkeypatch):
+    monkeypatch.setenv("STOCK_API_URL", "https://test.stock.local")
+    monkeypatch.setenv("WEBAI_API_KEY", "test-secret")
+    # Reload modules so they pick up the new env
+    import importlib
+    from signal_v2 import config as cfg
+    importlib.reload(cfg)
+    from signal_v2 import main as main_mod
+    importlib.reload(main_mod)
+    with TestClient(main_mod.app) as client:
+        r = client.get("/health")
+        assert r.status_code == 200
+        body = r.json()
+        assert body["status"] == "online"
+        assert body["stock_api_url"] == "https://test.stock.local"
+
+
+def test_startup_warns_if_webai_api_key_missing(monkeypatch, caplog):
+    # Use setenv with empty string + no-op load_dotenv to defeat .env re-read on reload
+    monkeypatch.setattr("signal_v2.config.load_dotenv", lambda *a, **k: None)
+    monkeypatch.setenv("WEBAI_API_KEY", "")
+    monkeypatch.setenv("STOCK_API_URL", "https://test.stock.local")
+    import importlib
+    from signal_v2 import config as cfg
+    importlib.reload(cfg)
+    # After reload, load_dotenv reference is fresh — re-patch
+    monkeypatch.setattr("signal_v2.config.load_dotenv", lambda *a, **k: None)
+    from signal_v2 import main as main_mod
+    importlib.reload(main_mod)
+    with caplog.at_level(logging.WARNING, logger="signal_v2.main"):
+        with TestClient(main_mod.app) as client:
+            client.get("/health")
+        assert any("WEBAI_API_KEY" in rec.message for rec in caplog.records)
+
+
+def test_startup_warns_if_kis_app_key_missing(monkeypatch, caplog):
+    """KIS app_key 미설정 시 startup WARNING (KIS 호출 disabled) — V1 패턴."""
+    monkeypatch.setattr("signal_v2.config.load_dotenv", lambda *a, **k: None)
+    monkeypatch.setenv("STOCK_API_URL", "https://test.stock.local")
+    monkeypatch.setenv("WEBAI_API_KEY", "test-secret")
+    # V1 pattern: kis_env_type=virtual, both virtual keys empty
+    monkeypatch.setenv("KIS_ENV_TYPE", "virtual")
+    monkeypatch.setenv("KIS_VIRTUAL_APP_KEY", "")
+    monkeypatch.setenv("KIS_REAL_APP_KEY", "")
+
+    import importlib
+    from signal_v2 import config as cfg
+    importlib.reload(cfg)
+    monkeypatch.setattr("signal_v2.config.load_dotenv", lambda *a, **k: None)
+    from signal_v2 import main as main_mod
+    importlib.reload(main_mod)
+    with caplog.at_level(logging.WARNING, logger="signal_v2.main"):
+        with TestClient(main_mod.app) as client:
+            client.get("/health")
+        assert any("KIS" in rec.message and "app_key" in rec.message.lower() for rec in caplog.records)

signal_v2/tests/test_pull_worker.py

@@ -0,0 +1,55 @@
+"""Tests for pull_worker (Phase 3a additions)."""
+from collections import deque
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from signal_v2.state import PollState
+
+
+async def test_minute_polling_cycle_updates_state_minute_bars():
+    """KIS REST mock 의 분봉 데이터가 state.minute_bars[ticker] deque 에 들어간다."""
+    from signal_v2.pull_worker import _run_kis_minute_cycle
+
+    state = PollState()
+    state.portfolio = {"holdings": [{"ticker": "005930"}, {"ticker": "000660"}]}
+    state.screener_preview = {
+        "items": [{"ticker": "005930"}, {"ticker": "035720"}]
+    }
+
+    kis_client_mock = MagicMock()
+    kis_client_mock.get_minute_ohlcv = AsyncMock(side_effect=[
+        [{"datetime": "2026-05-18T09:00:00+09:00", "open": 78000,
+          "high": 78500, "low": 77900, "close": 78300, "volume": 12345}],
+        [{"datetime": "2026-05-18T09:00:00+09:00", "open": 180000,
+          "high": 181000, "low": 179800, "close": 180500, "volume": 5000}],
+        [{"datetime": "2026-05-18T09:00:00+09:00", "open": 51000,
+          "high": 51200, "low": 50800, "close": 51100, "volume": 8000}],
+    ])
+    kis_client_mock.get_asking_price = AsyncMock(return_value={
+        "bid_total": 600, "ask_total": 400, "bid_ratio": 0.6,
+        "current_price": 51100, "as_of": "2026-05-18T09:00:30+09:00",
+    })
+
+    await _run_kis_minute_cycle(kis_client_mock, state)
+
+    # 3 unique tickers (005930, 000660, 035720)
+    assert "005930" in state.minute_bars
+    assert "000660" in state.minute_bars
+    assert "035720" in state.minute_bars
+    assert len(state.minute_bars["005930"]) >= 1
+    # asking_price 만 screener-only ticker (035720) 에 들어가야 함
+    # (portfolio = 005930, 000660 는 WebSocket 으로 들어옴)
+    assert "035720" in state.asking_price
+
+
+def test_websocket_message_updates_state_asking_price():
+    """WebSocket callback factory → state.asking_price 갱신."""
+    from signal_v2.pull_worker import make_asking_price_callback
+
+    state = PollState()
+    cb = make_asking_price_callback(state)
+    cb("005930", {"bid_total": 1000, "ask_total": 800, "bid_ratio": 0.555,
+                  "current_price": 78500, "as_of": "2026-05-18T10:00:00+09:00"})
+    assert state.asking_price["005930"]["bid_total"] == 1000
+    assert "asking_price/005930" in state.last_updated

signal_v2/tests/test_rate_limit.py

@@ -0,0 +1,34 @@
+"""Tests for SignalDedup."""
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from signal_v2.rate_limit import SignalDedup
+
+KST = ZoneInfo("Asia/Seoul")
+
+
+def test_is_recent_returns_false_for_new_ticker_action(tmp_dedup_db):
+    dedup = SignalDedup(tmp_dedup_db)
+    assert dedup.is_recent("005930", "buy") is False
+
+
+def test_is_recent_returns_true_within_24h(tmp_dedup_db):
+    dedup = SignalDedup(tmp_dedup_db)
+    dedup.record("005930", "buy", confidence=0.82)
+    assert dedup.is_recent("005930", "buy") is True
+
+
+def test_is_recent_returns_false_after_24h(tmp_dedup_db, monkeypatch):
+    dedup = SignalDedup(tmp_dedup_db)
+    # Record with a timestamp 25 hours ago
+    now = datetime.now(KST)
+    fake_now = now - timedelta(hours=25)
+    monkeypatch.setattr(
+        "signal_v2.rate_limit._now_iso", lambda: fake_now.isoformat()
+    )
+    dedup.record("005930", "buy", confidence=0.82)
+    # Reset to real now for is_recent check
+    monkeypatch.setattr(
+        "signal_v2.rate_limit._now_iso", lambda: now.isoformat()
+    )
+    assert dedup.is_recent("005930", "buy", within_hours=24) is False

signal_v2/tests/test_scheduler.py

@@ -0,0 +1,81 @@
+"""Tests for scheduler interval logic."""
+from datetime import datetime
+
+import pytest
+
+from signal_v2.scheduler import _next_interval, _is_market_day, KST
+
+
+def _kst(year, month, day, hour, minute=0):
+    return datetime(year, month, day, hour, minute, tzinfo=KST)
+
+
+def test_next_interval_pre_market_5min():
+    now = _kst(2026, 5, 18, 8, 30)  # Monday 08:30
+    assert _next_interval(now) == 300
+
+
+def test_next_interval_market_open_1min():
+    now = _kst(2026, 5, 18, 10, 0)  # Monday 10:00
+    assert _next_interval(now) == 60
+
+
+def test_next_interval_post_market_5min():
+    now = _kst(2026, 5, 18, 17, 0)  # Monday 17:00
+    assert _next_interval(now) == 300
+
+
+def test_next_interval_overnight_skip_to_next_morning():
+    now = _kst(2026, 5, 18, 2, 30)  # Monday 02:30 (dead zone, not NXT window)
+    interval = _next_interval(now)
+    # Dead zone 23:30-04:30 → next 04:30 is ~2h away
+    assert 2 * 3600 - 60 < interval < 2 * 3600 + 60
+
+
+def test_next_interval_holiday_skip():
+    # 2026-05-05 어린이날 (Tuesday holiday)
+    now = _kst(2026, 5, 5, 10, 0)
+    assert _is_market_day(now) is False
+    interval = _next_interval(now)
+    # Next: 2026-05-06 (Wed) 07:00, ~21h away
+    assert 20 * 3600 < interval < 22 * 3600
+
+
+def test_next_interval_at_market_open_boundary():
+    """09:00:00 정확 second → 60초 (market 구간 진입)."""
+    now = _kst(2026, 5, 18, 9, 0)  # Monday 09:00:00
+    assert _next_interval(now) == 60
+
+
+def test_next_interval_at_market_close_boundary():
+    """15:30:00 정확 second → 300초 (post-market 구간 진입)."""
+    now = _kst(2026, 5, 18, 15, 30)  # Monday 15:30:00
+    assert _next_interval(now) == 300
+
+
+def test_next_interval_at_polling_window_end_boundary():
+    """23:30:00 정확 second → dead zone skip (다음 04:30 까지)."""
+    now = _kst(2026, 5, 18, 23, 30)  # Monday 23:30:00 (NXT_PRE_END boundary)
+    interval = _next_interval(now)
+    # Dead zone 23:30-04:30 → next 04:30 is ~5h away
+    assert 5 * 3600 - 60 < interval < 5 * 3600 + 60
+
+
+def test_next_interval_nxt_evening_5min():
+    """22:00 평일 (NXT 야간) → 300 (5분)."""
+    now = _kst(2026, 5, 18, 22, 0)
+    assert _next_interval(now) == 300
+
+
+def test_next_interval_nxt_dawn_5min():
+    """05:30 평일 (NXT 새벽) → 300 (5분)."""
+    now = _kst(2026, 5, 18, 5, 30)
+    assert _next_interval(now) == 300
+
+
+def test_next_interval_dead_zone_skip():
+    """02:00 평일 (dead zone 23:30-04:30) → 다음 04:30 까지 (~9000s)."""
+    now = _kst(2026, 5, 18, 2, 0)
+    interval = _next_interval(now)
+    # 02:00 → 04:30 = 2.5h = 9000s
+    assert 9000 - 60 < interval < 9000 + 60

signal_v2/tests/test_stock_client.py

@@ -0,0 +1,168 @@
+"""Tests for stock_client.StockClient."""
+import asyncio
+import logging
+import pytest
+import httpx
+
+from signal_v2.stock_client import StockClient
+
+
+BASE_URL = "https://test.stock.local"
+API_KEY = "test-secret"
+
+
+async def test_get_portfolio_normal_returns_dict_with_pnl_pct(mock_stock_api):
+    """정상 200 응답 + cache 저장."""
+    mock_stock_api.get("/api/webai/portfolio").mock(
+        return_value=httpx.Response(
+            200,
+            json={
+                "holdings": [{"ticker": "005930", "pnl_pct": 0.047}],
+                "cash": [],
+                "summary": {},
+            },
+        )
+    )
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        result = await client.get_portfolio()
+        assert result["holdings"][0]["pnl_pct"] == 0.047
+        # Cache populated
+        assert len(client._cache) >= 1
+    finally:
+        await client.close()
+
+
+async def test_get_portfolio_uses_cache_within_ttl(mock_stock_api):
+    """60s TTL 내 두번째 호출 = mock 콜 1회."""
+    route = mock_stock_api.get("/api/webai/portfolio").mock(
+        return_value=httpx.Response(
+            200, json={"holdings": [], "cash": [], "summary": {}}
+        )
+    )
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        await client.get_portfolio()
+        await client.get_portfolio()  # second call within TTL
+        assert route.call_count == 1
+    finally:
+        await client.close()
+
+
+async def test_get_portfolio_refetches_after_ttl_expiry(mock_stock_api, monkeypatch):
+    """TTL 만료 후 재호출 = mock 콜 2회. time.monotonic 모킹."""
+    route = mock_stock_api.get("/api/webai/portfolio").mock(
+        return_value=httpx.Response(
+            200, json={"holdings": [], "cash": [], "summary": {}}
+        )
+    )
+    # Fake clock: starts at 0, jumps to 61 between calls
+    fake_time = [0.0]
+    monkeypatch.setattr(
+        "signal_v2.stock_client.time.monotonic", lambda: fake_time[0]
+    )
+
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        await client.get_portfolio()
+        fake_time[0] = 61.0  # 60s TTL 만료
+        await client.get_portfolio()
+        assert route.call_count == 2
+    finally:
+        await client.close()
+
+
+async def test_get_portfolio_retries_3_times_on_timeout(mock_stock_api, monkeypatch):
+    """timeout 2번 + 200 1번 → 최종 성공. exponential sleep 호출 검증."""
+    sleep_calls = []
+
+    async def fake_sleep(s):
+        sleep_calls.append(s)
+
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    mock_stock_api.get("/api/webai/portfolio").mock(
+        side_effect=[
+            httpx.TimeoutException("timeout 1"),
+            httpx.TimeoutException("timeout 2"),
+            httpx.Response(
+                200, json={"holdings": [], "cash": [], "summary": {}}
+            ),
+        ]
+    )
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        result = await client.get_portfolio()
+        assert result["holdings"] == []
+        assert sleep_calls == [1, 2]  # exponential 1s, 2s
+    finally:
+        await client.close()
+
+
+async def test_get_portfolio_429_triggers_backoff(mock_stock_api, monkeypatch):
+    """429 → 1s backoff → 200."""
+    sleep_calls = []
+
+    async def fake_sleep(s):
+        sleep_calls.append(s)
+
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    mock_stock_api.get("/api/webai/portfolio").mock(
+        side_effect=[
+            httpx.Response(429, text="rate limit"),
+            httpx.Response(
+                200, json={"holdings": [], "cash": [], "summary": {}}
+            ),
+        ]
+    )
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        result = await client.get_portfolio()
+        assert result["holdings"] == []
+        assert sleep_calls == [1]
+    finally:
+        await client.close()
+
+
+async def test_get_portfolio_falls_back_to_stale_on_all_failures(
+    mock_stock_api, monkeypatch, caplog
+):
+    """cache 에 이전 성공 응답 + 모든 retry 5xx → stale 반환 + logger.warning."""
+    # No-op sleep for fast test
+    async def fake_sleep(s):
+        return None
+    monkeypatch.setattr("asyncio.sleep", fake_sleep)
+
+    # Patch time.monotonic BEFORE first call so cached timestamp uses fake clock
+    fake_time = [0.0]
+    monkeypatch.setattr(
+        "signal_v2.stock_client.time.monotonic", lambda: fake_time[0]
+    )
+
+    # First call succeeds
+    route1 = mock_stock_api.get("/api/webai/portfolio").mock(
+        return_value=httpx.Response(
+            200,
+            json={"holdings": [{"ticker": "005930"}], "cash": [], "summary": {}},
+        )
+    )
+    client = StockClient(BASE_URL, API_KEY)
+    try:
+        first = await client.get_portfolio()
+        assert first["holdings"][0]["ticker"] == "005930"
+
+        # Advance fake clock past TTL (60s) so cache is stale
+        fake_time[0] = 61.0
+
+        # Now mock to return 500s persistently
+        route1.mock(return_value=httpx.Response(500, text="server error"))
+
+        with caplog.at_level(logging.WARNING, logger="signal_v2.stock_client"):
+            result = await client.get_portfolio()
+            assert result["holdings"][0]["ticker"] == "005930"  # stale data returned
+            assert any(
+                "stale" in rec.message.lower() for rec in caplog.records
+            )
+    finally:
+        await client.close()