# Signal V2 Phase 4 — Signal Generator Implementation Plan

> **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.

**Goal:** signal_v2 에 매수/매도 신호 생성 레이어 추가. Phase 2/3a/3b 의 모든 state 산출 → Phase 0 spec §6.1-§6.3 룰 → `state.signals[ticker]` (Phase 0 spec §5.2 schema) + `SignalDedup` 24h 차단.

**Architecture:** 순수 함수 `generate_signals(state, dedup, settings)` 가 매 분봉 cycle 후 호출. 매수 (Hard gate 4 조건 + soft confidence > 0.7) + 매도 (손절>이상>익절 우선순위). 6 env 외부화 (운영 튜닝).

**Tech Stack:** Python 순수 함수 / pytest / SignalDedup (Phase 2) / 외부 의존성 없음

**Spec:** `web-ui/docs/superpowers/specs/2026-05-17-signal-v2-phase4-signal-generator.md`

---

## 파일 구조

| 파일 | 책임 |
|------|------|
| `signal_v2/config.py` | (수정) Settings 에 6 env field 추가 |
| `signal_v2/state.py` | (수정) PollState `signals` field 추가 |
| `signal_v2/signal_generator.py` | (신규) `generate_signals(state, dedup, settings)` + 8 helper |
| `signal_v2/pull_worker.py` | (수정) `poll_loop` signature + 매 cycle 후 `generate_signals` 호출 |
| `signal_v2/main.py` | (수정) lifespan 의 poll_task 호출에 `dedup` + `settings` 전달 |
| `signal_v2/tests/test_signal_generator.py` | (신규) 9 단위 케이스 |
| `signal_v2/tests/test_pull_worker.py` | (수정) integration 1 케이스 추가 |

7 파일 변경, **10 신규 테스트** (45 → 55).

---

## Task 순서

```
Task 1: foundation (config 6 env + state signals field)
Task 2: signal_generator.py + 9 단위 tests (TDD)
Task 3: pull_worker + main.py 통합 + 1 integration test
Task 4: 사용자 수동 (.env optional + smoke + push)
```

---

### Task 1: foundation (config + state)

**Files:**
- Modify: `web-ai/signal_v2/config.py`
- Modify: `web-ai/signal_v2/state.py`

- [ ] **Step 1: Update config.py with 6 new fields**

Read `web-ai/signal_v2/config.py`. Add 6 fields to Settings (after `chronos_model` field, before properties):

```python
    stop_loss_pct: float = field(
        default_factory=lambda: float(os.getenv("STOP_LOSS_PCT", "-0.07"))
    )
    take_profit_pct: float = field(
        default_factory=lambda: float(os.getenv("TAKE_PROFIT_PCT", "0.15"))
    )
    chronos_spread_threshold: float = field(
        default_factory=lambda: float(os.getenv("CHRONOS_SPREAD_THRESHOLD", "0.6"))
    )
    asking_bid_ratio_threshold: float = field(
        default_factory=lambda: float(os.getenv("ASKING_BID_RATIO_THRESHOLD", "0.6"))
    )
    confidence_threshold: float = field(
        default_factory=lambda: float(os.getenv("CONFIDENCE_THRESHOLD", "0.7"))
    )
    min_momentum_for_buy: str = field(
        default_factory=lambda: os.getenv("MIN_MOMENTUM_FOR_BUY", "strong_up")
    )
```

- [ ] **Step 2: Update state.py with signals field**

Read `web-ai/signal_v2/state.py`. Add `signals` field to PollState (after `minute_momentum`):

```python
    signals: dict[str, dict] = field(default_factory=dict)
```

- [ ] **Step 3: Smoke import test**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -c "from signal_v2.config import get_settings; from signal_v2.state import state; s = get_settings(); print(f'stop_loss={s.stop_loss_pct}, conf_threshold={s.confidence_threshold}, min_momentum={s.min_momentum_for_buy}'); print(state)"
```
Expected: `stop_loss=-0.07, conf_threshold=0.7, min_momentum=strong_up` + state print with `signals={}`.

- [ ] **Step 4: Run existing tests — no regression**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -m pytest signal_v2/tests -q 2>&1 | tail -3
```
Expected: 45 passed.

- [ ] **Step 5: Commit**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
git add signal_v2/config.py signal_v2/state.py
git commit -m "$(cat <<'EOF'
feat(signal_v2-phase4): foundation — 6 env thresholds + state.signals

config.py: STOP_LOSS_PCT / TAKE_PROFIT_PCT / CHRONOS_SPREAD_THRESHOLD /
ASKING_BID_RATIO_THRESHOLD / CONFIDENCE_THRESHOLD / MIN_MOMENTUM_FOR_BUY
env vars with sensible defaults (Phase 0 spec §6.1-§6.2 values).

state.py: PollState.signals dict[ticker, signal_body] for Phase 5 input.

45 existing tests still pass.

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

---

### Task 2: signal_generator.py + 9 단위 tests

**Files:**
- Create: `web-ai/signal_v2/signal_generator.py`
- Create: `web-ai/signal_v2/tests/test_signal_generator.py`

- [ ] **Step 1: Write 9 failing tests**

Create `web-ai/signal_v2/tests/test_signal_generator.py`:

```python
"""Tests for signal_generator."""
from collections import deque
from pathlib import Path
from unittest.mock import MagicMock

import pytest

from signal_v2.signal_generator import generate_signals
from signal_v2.state import PollState


def _settings(**overrides):
    """Build a Settings-like object for tests (avoid env)."""
    defaults = dict(
        stop_loss_pct=-0.07,
        take_profit_pct=0.15,
        chronos_spread_threshold=0.6,
        asking_bid_ratio_threshold=0.6,
        confidence_threshold=0.7,
        min_momentum_for_buy="strong_up",
    )
    defaults.update(overrides)
    m = MagicMock()
    for k, v in defaults.items():
        setattr(m, k, v)
    return m


def _make_state_with_buy_candidate(
    ticker="005930", name="삼성전자", rank=1,
    chronos_median=0.02, chronos_q10=-0.01, chronos_q90=0.04, chronos_conf=0.85,
    momentum="strong_up", bid_ratio=0.7, current_price=78500,
):
    state = PollState()
    state.screener_preview = {"items": [{"ticker": ticker, "name": name}]}
    state.chronos_predictions[ticker] = {
        "median": chronos_median, "q10": chronos_q10, "q90": chronos_q90,
        "conf": chronos_conf, "as_of": "2026-05-17T16:00:00+09:00",
    }
    state.minute_momentum[ticker] = momentum
    state.asking_price[ticker] = {
        "bid_total": int(bid_ratio * 1000),
        "ask_total": int((1 - bid_ratio) * 1000),
        "bid_ratio": bid_ratio,
        "current_price": current_price,
        "as_of": "2026-05-17T16:00:01+09:00",
    }
    return state


def _make_state_with_holding(
    ticker="005930", name="삼성전자",
    pnl_pct=0.0, avg_price=75000, current_price=75000,
):
    state = PollState()
    state.portfolio = {"holdings": [{
        "ticker": ticker, "name": name,
        "avg_price": avg_price, "current_price": current_price,
        "pnl_pct": pnl_pct, "profit_rate": pnl_pct * 100,
        "quantity": 100, "broker": "키움",
    }]}
    state.screener_preview = {"items": []}
    return state


@pytest.fixture
def dedup_mock():
    d = MagicMock()
    d.is_recent.return_value = False
    return d


def test_buy_signal_when_all_conditions_pass_and_confidence_high(dedup_mock):
    state = _make_state_with_buy_candidate()
    generate_signals(state, dedup_mock, _settings())
    assert "005930" in state.signals
    sig = state.signals["005930"]
    assert sig["action"] == "buy"
    assert sig["confidence_webai"] > 0.7
    dedup_mock.record.assert_called()


def test_silent_when_chronos_median_negative(dedup_mock):
    state = _make_state_with_buy_candidate(chronos_median=-0.01)
    generate_signals(state, dedup_mock, _settings())
    assert "005930" not in state.signals


def test_silent_when_distribution_spread_too_wide(dedup_mock):
    # spread = (0.5 - (-0.5)) / max(|0.001|, 0.001) = 1000 → > 0.6
    state = _make_state_with_buy_candidate(
        chronos_median=0.001, chronos_q10=-0.5, chronos_q90=0.5,
    )
    generate_signals(state, dedup_mock, _settings())
    assert "005930" not in state.signals


def test_silent_when_momentum_not_strong_up(dedup_mock):
    state = _make_state_with_buy_candidate(momentum="weak_up")
    generate_signals(state, dedup_mock, _settings())
    assert "005930" not in state.signals


def test_silent_when_bid_ratio_below_threshold(dedup_mock):
    state = _make_state_with_buy_candidate(bid_ratio=0.5)
    generate_signals(state, dedup_mock, _settings())
    assert "005930" not in state.signals


def test_silent_when_confidence_below_threshold(dedup_mock):
    # chronos_conf low + rank=20 → confidence < 0.7
    state = _make_state_with_buy_candidate(chronos_conf=0.3)
    # add 19 fake items to push rank to 20
    state.screener_preview["items"] = (
        [{"ticker": f"FAKE{i:03d}"} for i in range(19)]
        + [{"ticker": "005930", "name": "삼성전자"}]
    )
    generate_signals(state, dedup_mock, _settings())
    # confidence_webai = 0.3*0.5 + 1.0*0.3 + 0.05*0.2 = 0.15 + 0.3 + 0.01 = 0.46 < 0.7
    assert "005930" not in state.signals


def test_sell_signal_when_stop_loss_triggered(dedup_mock):
    state = _make_state_with_holding(pnl_pct=-0.08, current_price=69000, avg_price=75000)
    generate_signals(state, dedup_mock, _settings())
    assert "005930" in state.signals
    sig = state.signals["005930"]
    assert sig["action"] == "sell"
    assert sig["confidence_webai"] == 1.0  # 손절선 즉시
    assert sig["pnl_pct"] == -0.08


def test_sell_signal_when_take_profit_triggered(dedup_mock):
    state = _make_state_with_holding(pnl_pct=0.16, current_price=87000, avg_price=75000)
    generate_signals(state, dedup_mock, _settings())
    assert "005930" in state.signals
    sig = state.signals["005930"]
    assert sig["action"] == "sell"
    assert sig["confidence_webai"] == 0.6  # 익절선 검토 알림


def test_silent_when_dedup_recently_sent(dedup_mock):
    state = _make_state_with_buy_candidate()
    dedup_mock.is_recent.return_value = True  # dedup 차단
    generate_signals(state, dedup_mock, _settings())
    assert "005930" not in state.signals
    dedup_mock.record.assert_not_called()
```

- [ ] **Step 2: Run tests to verify FAIL**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -m pytest signal_v2/tests/test_signal_generator.py -v 2>&1 | tail -10
```
Expected: ImportError (signal_v2.signal_generator missing).

- [ ] **Step 3: Implement signal_generator.py**

Create `web-ai/signal_v2/signal_generator.py`:

```python
"""Phase 4 — 매수/매도 신호 생성.

순수 함수 generate_signals(state, dedup, settings). state 를 mutate.
"""
from __future__ import annotations
import logging
from datetime import datetime
from zoneinfo import ZoneInfo

logger = logging.getLogger(__name__)
KST = ZoneInfo("Asia/Seoul")

# 분봉 모멘텀 → linear score
MOMENTUM_SCORES = {
    "strong_up": 1.0,
    "weak_up": 0.7,
    "neutral": 0.5,
    "weak_down": 0.3,
    "strong_down": 0.0,
}


def generate_signals(state, dedup, settings) -> None:
    """Phase 4 entry — state mutating. 매수/매도 룰 적용."""
    _evaluate_buy_signals(state, dedup, settings)
    _evaluate_sell_signals(state, dedup, settings)


# ----- 매수 -----

def _evaluate_buy_signals(state, dedup, settings) -> None:
    candidates = _buy_candidates(state)
    for ticker, name, rank in candidates:
        if not _check_buy_hard_gate(state, ticker, settings):
            continue
        confidence = _compute_buy_confidence(state, ticker, rank)
        if confidence <= settings.confidence_threshold:
            continue
        if dedup.is_recent(ticker, "buy", within_hours=24):
            continue
        state.signals[ticker] = _build_buy_signal(state, ticker, name, rank, confidence)
        dedup.record(ticker, "buy", confidence=confidence)


def _buy_candidates(state) -> list[tuple[str, str, int | None]]:
    """screener Top-N (rank 1..N) + portfolio (rank=None)."""
    candidates: list[tuple[str, str, int | None]] = []
    seen: set[str] = set()
    # Screener Top-N
    if state.screener_preview is not None:
        for i, item in enumerate(state.screener_preview.get("items", [])):
            ticker = item.get("ticker")
            if not ticker or ticker in seen:
                continue
            seen.add(ticker)
            name = item.get("name", ticker)
            candidates.append((ticker, name, i + 1))
    # Portfolio holdings
    if state.portfolio is not None:
        for h in state.portfolio.get("holdings", []):
            ticker = h.get("ticker")
            if not ticker or ticker in seen:
                continue
            seen.add(ticker)
            candidates.append((ticker, h.get("name", ticker), None))
    return candidates


def _check_buy_hard_gate(state, ticker: str, settings) -> bool:
    pred = state.chronos_predictions.get(ticker)
    if pred is None or pred["median"] <= 0:
        return False
    spread = (pred["q90"] - pred["q10"]) / max(abs(pred["median"]), 0.001)
    if spread >= settings.chronos_spread_threshold:
        return False
    momentum = state.minute_momentum.get(ticker)
    if momentum != settings.min_momentum_for_buy:
        return False
    ap = state.asking_price.get(ticker)
    if ap is None or ap["bid_ratio"] < settings.asking_bid_ratio_threshold:
        return False
    return True


def _compute_buy_confidence(state, ticker: str, rank: int | None) -> float:
    pred = state.chronos_predictions[ticker]
    chronos_conf = pred["conf"]
    minute_score = MOMENTUM_SCORES.get(state.minute_momentum.get(ticker, "neutral"), 0.5)
    screener_norm = 1 - (rank - 1) / 20 if rank is not None else 0.0
    return chronos_conf * 0.5 + minute_score * 0.3 + screener_norm * 0.2


def _build_buy_signal(state, ticker: str, name: str, rank: int | None, confidence: float) -> dict:
    ap = state.asking_price[ticker]
    pred = state.chronos_predictions[ticker]
    return {
        "ticker": ticker,
        "name": name,
        "action": "buy",
        "confidence_webai": confidence,
        "current_price": ap["current_price"],
        "avg_price": None,
        "pnl_pct": None,
        "context": _build_context(state, ticker, rank),
        "as_of": datetime.now(KST).isoformat(),
    }


# ----- 매도 -----

def _evaluate_sell_signals(state, dedup, settings) -> None:
    if state.portfolio is None:
        return
    for holding in state.portfolio.get("holdings", []):
        ticker = holding.get("ticker")
        if not ticker:
            continue
        sell = _try_stop_loss(state, holding, settings)
        if sell is None:
            sell = _try_anomaly(state, holding, settings)
        if sell is None:
            sell = _try_take_profit(state, holding, settings)
        if sell is None:
            continue
        if dedup.is_recent(ticker, "sell", within_hours=24):
            continue
        state.signals[ticker] = sell
        dedup.record(ticker, "sell", confidence=sell["confidence_webai"])


def _try_stop_loss(state, holding: dict, settings) -> dict | None:
    pnl = holding.get("pnl_pct")
    if pnl is None or pnl >= settings.stop_loss_pct:
        return None
    return _build_sell_signal(state, holding, confidence=1.0, reason="stop_loss")


def _try_take_profit(state, holding: dict, settings) -> dict | None:
    pnl = holding.get("pnl_pct")
    if pnl is None or pnl <= settings.take_profit_pct:
        return None
    return _build_sell_signal(state, holding, confidence=0.6, reason="take_profit")


def _try_anomaly(state, holding: dict, settings) -> dict | None:
    ticker = holding["ticker"]
    pred = state.chronos_predictions.get(ticker)
    if pred is None or pred["median"] >= -0.01:
        return None
    momentum = state.minute_momentum.get(ticker)
    if momentum != "strong_down":
        return None
    ap = state.asking_price.get(ticker)
    if ap is None:
        return None
    if ap["bid_ratio"] > (1 - settings.asking_bid_ratio_threshold):
        return None  # 매도세 60% 미만
    minute_score = 1.0 - MOMENTUM_SCORES.get(momentum, 0.5)  # 반전
    confidence = pred["conf"] * 0.5 + minute_score * 0.3 + 1.0 * 0.2
    if confidence <= settings.confidence_threshold:
        return None
    return _build_sell_signal(state, holding, confidence=confidence, reason="anomaly")


def _build_sell_signal(state, holding: dict, confidence: float, reason: str) -> dict:
    ticker = holding["ticker"]
    return {
        "ticker": ticker,
        "name": holding.get("name", ticker),
        "action": "sell",
        "confidence_webai": confidence,
        "current_price": holding.get("current_price"),
        "avg_price": holding.get("avg_price"),
        "pnl_pct": holding.get("pnl_pct"),
        "context": _build_context(state, ticker, rank=None, sell_reason=reason),
        "as_of": datetime.now(KST).isoformat(),
    }


# ----- Context -----

def _build_context(state, ticker: str, rank: int | None, sell_reason: str | None = None) -> dict:
    pred = state.chronos_predictions.get(ticker) or {}
    ap = state.asking_price.get(ticker) or {}
    news_item = _find_news_sentiment(state, ticker)
    screener_scores = _find_screener_scores(state, ticker)
    context: dict = {
        "chronos_pred_1d": pred.get("median"),
        "chronos_pred_conf": pred.get("conf"),
        "chronos_q10": pred.get("q10"),
        "chronos_q90": pred.get("q90"),
        "screener_rank": rank,
        "screener_scores": screener_scores,
        "minute_momentum": state.minute_momentum.get(ticker),
        "asking_bid_ratio": ap.get("bid_ratio"),
        "news_sentiment": news_item.get("score") if news_item else None,
        "news_reason": news_item.get("reason") if news_item else None,
    }
    if sell_reason is not None:
        context["sell_reason"] = sell_reason
    return context


def _find_news_sentiment(state, ticker: str) -> dict | None:
    if state.news_sentiment is None:
        return None
    for item in state.news_sentiment.get("items", []):
        if item.get("ticker") == ticker:
            return item
    return None


def _find_screener_scores(state, ticker: str) -> dict | None:
    if state.screener_preview is None:
        return None
    for item in state.screener_preview.get("items", []):
        if item.get("ticker") == ticker:
            return item.get("scores")
    return None
```

- [ ] **Step 4: Run tests to verify PASS**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -m pytest signal_v2/tests/test_signal_generator.py -v 2>&1 | tail -15
```
Expected: 9 passed.

Full suite:
```bash
python -m pytest signal_v2/tests -q 2>&1 | tail -3
```
Expected: 54 passed.

If any test fails, examine the assertion + impl. Common gotchas:
- Confidence calculation order — chronos*0.5 + minute*0.3 + screener*0.2
- Stop loss `<` (strict) vs `<=` — spec says "도달 시" so use `<` strict
- screener_norm when rank=None → 0.0 (not 1.0)

- [ ] **Step 5: Commit**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
git add signal_v2/signal_generator.py signal_v2/tests/test_signal_generator.py
git commit -m "$(cat <<'EOF'
feat(signal_v2-phase4): signal_generator + 9 unit tests

generate_signals(state, dedup, settings) → state mutating:
- Buy: screener Top-N + portfolio. Hard gate (chronos median > 0 +
  spread < 0.6 + momentum strong_up + bid_ratio >= 0.6) + soft
  confidence (chronos*0.5 + minute*0.3 + screener*0.2) > 0.7.
- Sell: portfolio only. Priority stop_loss > anomaly > take_profit.
  Stop loss confidence 1.0 (immediate), take_profit 0.6 (review).
- SignalDedup 24h via dedup.is_recent/record per (ticker, action).
- State signal dict matches Phase 0 spec §5.2 schema.

54 tests pass.

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

---

### Task 3: pull_worker + main.py integration + 1 test

**Files:**
- Modify: `web-ai/signal_v2/pull_worker.py`
- Modify: `web-ai/signal_v2/main.py`
- Modify: `web-ai/signal_v2/tests/test_pull_worker.py`

- [ ] **Step 1: Write failing integration test**

Append to `web-ai/signal_v2/tests/test_pull_worker.py`:

```python
def test_poll_loop_calls_generate_signals_after_cycle(monkeypatch):
    """매 cycle 후 generate_signals 호출 + state.signals 갱신."""
    from unittest.mock import MagicMock
    from signal_v2.state import PollState

    state = PollState()
    state.portfolio = {"holdings": [{
        "ticker": "005930", "name": "삼성전자",
        "avg_price": 75000, "current_price": 69000,
        "pnl_pct": -0.08, "profit_rate": -8.0,
        "quantity": 100, "broker": "키움",
    }]}
    state.screener_preview = {"items": []}

    dedup = MagicMock()
    dedup.is_recent.return_value = False

    settings = MagicMock()
    settings.stop_loss_pct = -0.07
    settings.take_profit_pct = 0.15
    settings.chronos_spread_threshold = 0.6
    settings.asking_bid_ratio_threshold = 0.6
    settings.confidence_threshold = 0.7
    settings.min_momentum_for_buy = "strong_up"

    from signal_v2.signal_generator import generate_signals
    # Call generate_signals directly to verify state mutation through the public API.
    generate_signals(state, dedup, settings)

    # Stop loss should trigger
    assert "005930" in state.signals
    assert state.signals["005930"]["action"] == "sell"
    assert state.signals["005930"]["confidence_webai"] == 1.0
    dedup.record.assert_called_with("005930", "sell", confidence=1.0)
```

- [ ] **Step 2: Run test to verify PASS (signal_generator from Task 2 already exists)**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -m pytest signal_v2/tests/test_pull_worker.py::test_poll_loop_calls_generate_signals_after_cycle -v 2>&1 | tail -10
```
Expected: PASS (test exercises generate_signals directly — public API integration).

- [ ] **Step 3: Update pull_worker.py — poll_loop signature + cycle integration**

Read `web-ai/signal_v2/pull_worker.py`. Modify the `poll_loop` signature to accept dedup + settings:

```python
async def poll_loop(
    client, state, shutdown,
    kis_client=None, chronos=None,
    dedup=None, settings=None,
) -> None:
    """...existing docstring..."""
    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")
            try:
                update_minute_momentum_for_all(state)
            except Exception:
                logger.exception("minute momentum update failed")
            if _is_post_close_trigger(now) and chronos is not None and kis_client is not None:
                try:
                    await _run_post_close_cycle(kis_client, chronos, state)
                except Exception:
                    logger.exception("post-close cycle failed")
            # Phase 4: generate signals
            if dedup is not None and settings is not None:
                try:
                    from signal_v2.signal_generator import generate_signals
                    generate_signals(state, dedup, settings)
                except Exception:
                    logger.exception("generate_signals failed")
        interval = _next_interval(now)
        try:
            await asyncio.wait_for(shutdown.wait(), timeout=interval)
            break
        except asyncio.TimeoutError:
            continue
    logger.info("poll_loop ended")
```

- [ ] **Step 4: Update main.py — pass dedup + settings to poll_loop**

Read `web-ai/signal_v2/main.py`. Find the `asyncio.create_task(poll_loop(...))` call inside `lifespan` and add `dedup` + `settings` params:

```python
    _ctx.poll_task = asyncio.create_task(
        poll_loop(
            _ctx.client, state_mod.state, _ctx.shutdown,
            kis_client=_ctx.kis_client,
            chronos=_ctx.chronos,
            dedup=_ctx.dedup,
            settings=settings,
        )
    )
```

- [ ] **Step 5: Run full test suite**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
python -m pytest signal_v2/tests -q 2>&1 | tail -3
```
Expected: 55 passed (54 + 1 new integration).

- [ ] **Step 6: Commit**

```bash
cd /c/Users/jaeoh/Desktop/workspace/web-ai
git add signal_v2/pull_worker.py signal_v2/main.py signal_v2/tests/test_pull_worker.py
git commit -m "$(cat <<'EOF'
feat(signal_v2-phase4): pull_worker + main.py integrate signal generator

poll_loop signature now accepts dedup + settings. After each cycle
(stock pull + minute momentum + post-close), call generate_signals
to populate state.signals. main.py lifespan passes _ctx.dedup and
settings to poll_loop.

1 integration test added. 55 tests pass.

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

---

### Task 4: 사용자 수동 — .env optional + smoke + push

**This task requires user action.**

- [ ] **Step 1: .env optional**

6 env 의 default 가 Phase 0 spec 값과 동일 — `.env` 변경 불필요. 운영 검증 후 조정 시:
```
STOP_LOSS_PCT=-0.07
TAKE_PROFIT_PCT=0.15
CHRONOS_SPREAD_THRESHOLD=0.6
ASKING_BID_RATIO_THRESHOLD=0.6
CONFIDENCE_THRESHOLD=0.7
MIN_MOMENTUM_FOR_BUY=strong_up
```

- [ ] **Step 2: signal_v2 재시작**

기존 signal_v2 가 가동 중이면 Ctrl+C 후:
```powershell
cd C:\Users\jaeoh\Desktop\workspace\web-ai\signal_v2
.\start.bat
```
기대: 정상 시작 (signal_generator 자동 호출 — 매 cycle 마다).

- [ ] **Step 3: state.signals 검증 (수동)**

운영 시간대라면 cycle 진행 + state.signals 채워질 수 있음. 수동 검증:
```powershell
cd C:\Users\jaeoh\Desktop\workspace\web-ai
python -c "
import asyncio
from signal_v2.config import get_settings
from signal_v2.kis_client import KISClient
from signal_v2.chronos_predictor import ChronosPredictor
from signal_v2.state import PollState
from signal_v2.rate_limit import SignalDedup
from signal_v2.pull_worker import _run_post_close_cycle, update_minute_momentum_for_all
from signal_v2.signal_generator import generate_signals

async def main():
    s = get_settings()
    kc = KISClient(app_key=s.kis_app_key, app_secret=s.kis_app_secret, account=s.kis_account, is_virtual=s.kis_is_virtual, v1_token_path=s.v1_token_path)
    cp = ChronosPredictor(model_name=s.chronos_model)
    dedup = SignalDedup(s.db_path)
    state = PollState()
    state.portfolio = {'holdings': [{'ticker': '005930', 'name': '삼성전자', 'avg_price': 75000, 'current_price': 78500, 'pnl_pct': 0.047, 'profit_rate': 4.67, 'quantity': 100, 'broker': '키움'}]}
    state.screener_preview = {'items': []}
    try:
        await _run_post_close_cycle(kc, cp, state)
        update_minute_momentum_for_all(state)
        generate_signals(state, dedup, s)
        print('Signals:', state.signals)
    finally:
        await kc.close()
asyncio.run(main())
"
```
Expected: `Signals: {}` (정상 — pnl_pct 0.047 은 손절/익절 트리거 안 함, 매수 조건 다 만족 어려움) 또는 일부 신호 dict.

- [ ] **Step 4: V1 무영향**

V1 정상 가동 확인.

- [ ] **Step 5: push**

```powershell
cd C:\Users\jaeoh\Desktop\workspace\web-ai
git push
```

- [ ] **Step 6: 결과 보고**

- Step 2 (signal_v2 시작): PASS / FAIL
- Step 3 (state.signals 검증): PASS — empty dict or 신호 결과 공유 / FAIL
- Step 4 (V1 무영향): PASS / FAIL
- Step 5 (push): PASS / FAIL

전체 PASS 시 **Phase 4 완료** → Phase 5 (agent-office /signal + Qwen3 + 이중 텔레그램) brainstorming.

---

## Self-Review

**1. Spec coverage:**

| Spec § | 요구사항 | Plan task |
|--------|----------|----------|
| §2 ① signal_generator | Task 2 ✅ |
| §2 ② config 6 env | Task 1 ✅ |
| §2 ③ state.signals | Task 1 ✅ |
| §2 ④ pull_worker integration | Task 3 ✅ |
| §2 ⑤ main.py lifespan | Task 3 ✅ |
| §2 ⑥ 10 tests | Task 2 (9) + Task 3 (1) = 10 ✅ |
| §4 매수 룰 + confidence | Task 2 (_check_buy_hard_gate + _compute_buy_confidence) ✅ |
| §5 매도 룰 + dedup | Task 2 (_try_stop_loss/anomaly/take_profit + dedup.is_recent/record) ✅ |
| §6 state 통합 + pull_worker | Task 1 + Task 3 ✅ |
| §7 signal_generator 구조 | Task 2 Step 3 (8 helpers) ✅ |
| §8 10 테스트 케이스 | Task 2-3 ✅ |
| §9 DoD 8 항목 | Task 1-4 합산 ✅ |

No gaps.

**2. Placeholder scan**: No "TBD" / "implement later". 각 step 의 코드 + 명령 모두 명시.

**3. Type consistency:**
- `generate_signals(state, dedup, settings) -> None` consistent Task 2 + Task 3 ✅
- `MOMENTUM_SCORES` 매핑 consistent (1.0/0.7/0.5/0.3/0.0) ✅
- Settings field names consistent Task 1 + Task 2 (stop_loss_pct, etc.) ✅
- PollState.signals dict[str, dict] consistent ✅
- helper signatures (_check_buy_hard_gate, _compute_buy_confidence, _try_stop_loss, _try_anomaly, _try_take_profit, _build_buy_signal, _build_sell_signal, _build_context) consistent ✅

Plan passes self-review.