# Stock Screener Board 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:** KRX 전종목 일봉·수급 데이터를 매일 캐싱하고, 위생 게이트 1개 + 점수 노드 7개를 가중합해 강세주 후보 Top N을 산출하는 노드 기반 분석 보드를 구축한다. 평일 16:30 KST에 agent-office가 자동 실행해 텔레그램으로 결과를 전송한다.
**Architecture:** stock-lab에 `screener` 패키지 추가 — pykrx 캐시(snapshot.py) → ScreenContext(engine.py) → HygieneGate(GateNode) → ScoreNode 7개 → 가중합 → ATR 포지션 사이저. 프론트는 `/stock/screener` 별도 페이지에서 노드 폼·결과 표·히스토리·텔레그램 미리보기 제공. 자동 잡 트리거와 텔레그램 발신은 agent-office가 담당.
**Tech Stack:** Python(FastAPI, pandas, pykrx, pytest, SQLite) + React 18(Vite, react-router-dom) + agent-office(Python 잡)
---
## 개요 — 6 Phase
| Phase | 범위 | 작업 repo | Task 수 |
|-------|------|----------|---------|
| 0 | 백엔드 기반 — 의존성, 스키마, KRX 캐시, ScreenContext | `web-backend` | 4 |
| 1 | 노드 8개 (TDD) | `web-backend` | 9 |
| 2 | 엔진 · 포지션 사이저 · 텔레그램 빌더 · 레지스트리 | `web-backend` | 4 |
| 3 | FastAPI 라우터 | `web-backend` | 5 |
| 4 | 프론트엔드 페이지 | `web-ui` | 9 |
| 5 | agent-office 자동 잡 통합 | `web-backend` | 1 |
| 6 | 초기 백필 · 수동 검증 · 배포 | both | 3 |
**Total: 35 tasks.** 각 Phase 끝에 commit 정리. **commit은 해당 repo의 cwd에서만 수행** (`web-backend`와 `web-ui`는 별도 Git 저장소).
## 로컬 개발 환경 설정 (1회 — 이후 모든 pytest는 이 venv에서)
> ⚠️ 본 워크스페이스 규약: **Docker는 NAS에서만 구동.** 로컬에서는 venv로 직접 실행한다.
```powershell
cd C:\Users\jaeoh\Desktop\workspace\web-backend\stock-lab
python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt
pip install pytest httpx pandas
# (requirements.txt 변경은 Task 0.1에서. 추가 의존성 설치:)
# pip install finance-datareader beautifulsoup4 lxml
# ⚠️ 데이터 소스 변경 노트 (2026-05-12 실측 후 결정):
# plan의 spec은 "pykrx 하이브리드"였으나, 실측 결과 pykrx의 시장 전체 함수
# (get_market_ticker_list / get_market_cap / get_market_ohlcv_by_ticker)가 모두 KRX
# 인증 요구로 인해 비인증 호출 시 깨짐. 따라서 실제 구현은:
# - 종목 마스터 + 당일 일봉 + 5년치 일봉: FinanceDataReader (fdr)
# - 외국인/기관 수급: 네이버 금융 종목별 frgn 페이지 스크래핑 (시총 상위 500종목)
# Task 0.3 snapshot.py 코드는 implementer dispatch 시 새 방향으로 안내됨.
```
| 작업 | 어디서 실행 |
|------|------------|
| 백엔드 단위 테스트 (`pytest`) | **로컬 venv** (stock-lab cwd) |
| 컨테이너 빌드 (`docker compose build`) | **NAS SSH** — `ssh user@gahusb.synology.me` 후 운영 디렉토리에서 |
| Backfill (Task 6.1) | **NAS SSH** — 장시간 + 운영 DB 대상 |
| Agent-office 발동 (Task 5.x · 6.3) | **NAS** — git push로 webhook 자동 배포 후 운영 |
| 프론트 개발 (`npm run dev`) | **로컬** |
| 프론트 배포 (`npm run release:nas`) | **로컬** (Z 드라이브 NAS 마운트) |
이 plan의 모든 `pytest` 명령은 **stock-lab 디렉토리에서 venv 활성화 상태 가정**.
---
# Phase 0 — 백엔드 기반
## Task 0.1: pykrx 의존성 + 스키마 함수 골격
**Files:**
- Modify: `web-backend/stock-lab/requirements.txt`
- Create: `web-backend/stock-lab/app/screener/__init__.py`
- Modify: `web-backend/stock-lab/app/db.py` (스크리너 스키마 함수 호출 1줄 추가)
- [ ] **Step 1: requirements.txt에 데이터 라이브러리 추가**
`web-backend/stock-lab/requirements.txt`에 다음 의존성 추가:
```
finance-datareader>=0.9.96
beautifulsoup4>=4.12
lxml>=5.0
```
(`httpx`는 보통 이미 있으나 없으면 함께 추가.) 기존 pykrx 라인은 추가하지 않습니다 (실측 결과 시장 전체 함수가 KRX 인증 요구로 깨짐).
- [ ] **Step 2: screener 패키지 생성**
Create `web-backend/stock-lab/app/screener/__init__.py`:
```python
"""Stock screener — KRX 강세주 분석 노드 기반 보드.
See docs/superpowers/specs/2026-05-12-stock-screener-board-design.md
"""
from .engine import Screener, ScreenContext, ScreenerResult
from .registry import NODE_REGISTRY, GATE_REGISTRY
__all__ = [
"Screener", "ScreenContext", "ScreenerResult",
"NODE_REGISTRY", "GATE_REGISTRY",
]
```
(엔진·레지스트리 모듈은 Phase 1·2에서 만들지만, 이 import는 그때 활성화됨. 지금은 NameError로 import 시 실패하는 게 정상 — 나중에 채워짐.)
⚠️ Step 2의 import를 그대로 두면 다른 모듈 import 시 폭발합니다. **임시로 주석 처리하고 진행. Phase 2 끝에 다시 활성화.**
수정:
```python
"""Stock screener — KRX 강세주 분석 노드 기반 보드.
See docs/superpowers/specs/2026-05-12-stock-screener-board-design.md
"""
# Phase 2 완료 후 활성화:
# from .engine import Screener, ScreenContext, ScreenerResult
# from .registry import NODE_REGISTRY, GATE_REGISTRY
__all__ = []
```
- [ ] **Step 3: 로컬 venv에 데이터 라이브러리 설치**
```powershell
cd C:\Users\jaeoh\Desktop\workspace\web-backend\stock-lab
.\.venv\Scripts\Activate.ps1
pip install finance-datareader beautifulsoup4 lxml
```
Expected: 설치 성공.
> NAS 운영 컨테이너 재빌드는 본 plan 마지막의 **최종 배포** 단계에서 `git push` → webhook으로 자동 수행. 지금은 로컬 venv 동작만 검증.
- [ ] **Step 4: FDR + 네이버 동작 smoke test (one-off, 로컬 venv)**
```powershell
python -c "import FinanceDataReader as fdr; df = fdr.StockListing('KRX'); print('rows:', df.shape[0]); print(df.head(3)[['Code','Name','Market','Marcap','Close']])"
python -c "import httpx; from bs4 import BeautifulSoup; r = httpx.get('https://finance.naver.com/item/frgn.naver?code=005930', headers={'User-Agent':'Mozilla/5.0'}); print('status:', r.status_code); soup = BeautifulSoup(r.text,'lxml'); print('rows:', len(soup.select('table.type2 tr')))"
```
Expected: FDR rows ≥ 2,800. naver status 200, table rows > 5.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/requirements.txt stock-lab/app/screener/__init__.py
git commit -m "chore(stock-lab): FDR/네이버 데이터 의존성 + screener 패키지 골격"
```
---
## Task 0.2: 스키마 마이그레이션 (SQL DDL 7테이블)
**Files:**
- Create: `web-backend/stock-lab/app/screener/schema.py`
- Modify: `web-backend/stock-lab/app/db.py` (스키마 함수 import + 호출)
- Create: `web-backend/stock-lab/app/test_screener_schema.py`
- [ ] **Step 1: 테스트 먼저 작성**
Create `web-backend/stock-lab/app/test_screener_schema.py`:
```python
import sqlite3
from app.screener.schema import ensure_screener_schema
def test_creates_all_tables(tmp_path):
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
ensure_screener_schema(conn)
tables = {r[0] for r in conn.execute(
"SELECT name FROM sqlite_master WHERE type='table'"
).fetchall()}
expected = {
"krx_master", "krx_daily_prices", "krx_flow",
"screener_settings", "screener_runs", "screener_results",
}
assert expected.issubset(tables)
def test_settings_seeded_with_singleton_row(tmp_path):
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
ensure_screener_schema(conn)
rows = conn.execute("SELECT id FROM screener_settings").fetchall()
assert rows == [(1,)]
def test_idempotent(tmp_path):
db_path = tmp_path / "test.db"
conn = sqlite3.connect(db_path)
ensure_screener_schema(conn)
ensure_screener_schema(conn) # 두 번 호출해도 에러 없어야 함
rows = conn.execute("SELECT count(*) FROM screener_settings").fetchall()
assert rows == [(1,)]
```
- [ ] **Step 2: Run test, expect failure**
```bash
pytest app/test_screener_schema.py -v
```
Expected: FAIL with `ModuleNotFoundError: No module named 'app.screener.schema'`
- [ ] **Step 3: schema.py 작성**
Create `web-backend/stock-lab/app/screener/schema.py`:
```python
"""Screener schema bootstrap. Called once at module import via db.py."""
import json
import sqlite3
from datetime import datetime, timezone
DEFAULT_WEIGHTS = {
"foreign_buy": 1.0,
"volume_surge": 1.0,
"momentum": 1.0,
"high52w": 1.2,
"rs_rating": 1.2,
"ma_alignment": 1.0,
"vcp_lite": 0.8,
}
DEFAULT_NODE_PARAMS = {
"foreign_buy": {"window_days": 5},
"volume_surge": {"baseline_days": 20, "eval_days": 3},
"momentum": {"window_days": 20},
"high52w": {"window_days": 252},
"rs_rating": {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}},
"ma_alignment": {"ma_periods": [50, 150, 200]},
"vcp_lite": {"short_window": 40, "long_window": 252},
}
DEFAULT_GATE_PARAMS = {
"min_market_cap_won": 50_000_000_000,
"min_avg_value_won": 500_000_000,
"min_listed_days": 60,
"skip_managed": True,
"skip_preferred": True,
"skip_spac": True,
"skip_halted_days": 3,
}
DDL = """
CREATE TABLE IF NOT EXISTS krx_master (
ticker TEXT PRIMARY KEY,
name TEXT NOT NULL,
market TEXT NOT NULL,
market_cap INTEGER,
is_managed INTEGER NOT NULL DEFAULT 0,
is_preferred INTEGER NOT NULL DEFAULT 0,
is_spac INTEGER NOT NULL DEFAULT 0,
listed_date TEXT,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS krx_daily_prices (
ticker TEXT NOT NULL,
date TEXT NOT NULL,
open INTEGER, high INTEGER, low INTEGER, close INTEGER,
volume INTEGER,
value INTEGER,
PRIMARY KEY (ticker, date)
);
CREATE INDEX IF NOT EXISTS idx_prices_date ON krx_daily_prices(date);
CREATE TABLE IF NOT EXISTS krx_flow (
ticker TEXT NOT NULL,
date TEXT NOT NULL,
foreign_net INTEGER,
institution_net INTEGER,
PRIMARY KEY (ticker, date)
);
CREATE INDEX IF NOT EXISTS idx_flow_date ON krx_flow(date);
CREATE TABLE IF NOT EXISTS screener_settings (
id INTEGER PRIMARY KEY CHECK (id = 1),
weights_json TEXT NOT NULL,
node_params_json TEXT NOT NULL,
gate_params_json TEXT NOT NULL,
top_n INTEGER NOT NULL DEFAULT 20,
rr_ratio REAL NOT NULL DEFAULT 2.0,
atr_window INTEGER NOT NULL DEFAULT 14,
atr_stop_mult REAL NOT NULL DEFAULT 2.0,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS screener_runs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
asof TEXT NOT NULL,
mode TEXT NOT NULL,
status TEXT NOT NULL,
error TEXT,
started_at TEXT NOT NULL,
finished_at TEXT,
weights_json TEXT NOT NULL,
node_params_json TEXT NOT NULL,
gate_params_json TEXT NOT NULL,
top_n INTEGER NOT NULL,
survivors_count INTEGER,
telegram_sent INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_runs_asof ON screener_runs(asof DESC);
CREATE TABLE IF NOT EXISTS screener_results (
run_id INTEGER NOT NULL,
rank INTEGER NOT NULL,
ticker TEXT NOT NULL,
name TEXT NOT NULL,
total_score REAL NOT NULL,
scores_json TEXT NOT NULL,
close INTEGER,
market_cap INTEGER,
entry_price INTEGER,
stop_price INTEGER,
target_price INTEGER,
atr14 REAL,
PRIMARY KEY (run_id, ticker),
FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE
);
CREATE INDEX IF NOT EXISTS idx_results_run_rank ON screener_results(run_id, rank);
"""
def ensure_screener_schema(conn: sqlite3.Connection) -> None:
"""Create tables and seed default settings (idempotent)."""
conn.executescript(DDL)
existing = conn.execute("SELECT id FROM screener_settings WHERE id=1").fetchone()
if existing is None:
now = datetime.now(timezone.utc).isoformat()
conn.execute(
"""
INSERT INTO screener_settings (
id, weights_json, node_params_json, gate_params_json,
top_n, rr_ratio, atr_window, atr_stop_mult, updated_at
) VALUES (1, ?, ?, ?, 20, 2.0, 14, 2.0, ?)
""",
(
json.dumps(DEFAULT_WEIGHTS),
json.dumps(DEFAULT_NODE_PARAMS),
json.dumps(DEFAULT_GATE_PARAMS),
now,
),
)
conn.commit()
```
- [ ] **Step 4: Run test, expect pass**
```bash
pytest app/test_screener_schema.py -v
```
Expected: 3 passed.
- [ ] **Step 5: db.py에서 호출**
Read `web-backend/stock-lab/app/db.py` 파일을 확인하고 기존 스키마 부트스트랩 함수(예: `init_db()` 또는 connection factory)를 찾으세요. 그 함수 마지막에 다음을 추가:
```python
from app.screener.schema import ensure_screener_schema
# 기존 코드 다음에:
ensure_screener_schema(conn)
```
(정확한 위치는 db.py 구조에 따라 달라짐. 핵심: 앱 시작 시 1회 호출되도록.)
- [ ] **Step 6: 스키마 적용은 단위 테스트로 검증 (이미 Step 4에서 완료)**
운영 DB에 대한 실제 적용은 최종 배포 후 NAS에서 자동 수행됨 (`app/db.py` import 시 `ensure_screener_schema(conn)` 호출). 로컬 검증은 step 4의 pytest 통과로 충분.
- [ ] **Step 7: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/schema.py stock-lab/app/test_screener_schema.py stock-lab/app/db.py
git commit -m "feat(stock-lab): screener 스키마 7테이블 + 디폴트 설정 시드"
```
---
## Task 0.3: snapshot.py — KRX 캐시 갱신
**Files:**
- Create: `web-backend/stock-lab/app/screener/snapshot.py`
- Create: `web-backend/stock-lab/app/test_screener_snapshot.py`
- [ ] **Step 1: 테스트 먼저**
Create `web-backend/stock-lab/app/test_screener_snapshot.py`:
```python
import datetime as dt
import sqlite3
from unittest.mock import patch
import pandas as pd
import pytest
from app.screener.schema import ensure_screener_schema
from app.screener.snapshot import refresh_daily, backfill
@pytest.fixture
def conn(tmp_path):
db_path = tmp_path / "test.db"
c = sqlite3.connect(db_path)
ensure_screener_schema(c)
yield c
c.close()
def _stub_pykrx(monkeypatch):
"""pykrx 의 데이터 호출을 메모리 픽스쳐로 대체."""
def fake_ticker_list(date, market):
return ["005930", "035420"] if market == "KOSPI" else ["091990"]
def fake_ohlcv(date, ticker, *_args, **_kwargs):
return pd.DataFrame({
"시가": [70000], "고가": [72000], "저가": [69500],
"종가": [71000], "거래량": [12_000_000], "거래대금": [840_000_000_000],
}, index=[pd.Timestamp(date)])
def fake_market_cap(date):
return pd.DataFrame({
"시가총액": [420_000_000_000_000, 30_000_000_000_000, 10_000_000_000_000],
"상장주식수": [5_900_000_000, 164_000_000, 12_000_000],
}, index=["005930", "035420", "091990"])
def fake_trading_value(date, *_a, **_k):
return pd.DataFrame({
"외국인": [12_000_000_000, -3_000_000_000, 500_000_000],
"기관": [4_000_000_000, 8_000_000_000, -100_000_000],
}, index=["005930", "035420", "091990"])
monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_ticker_list", fake_ticker_list)
monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_ohlcv", fake_ohlcv)
monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_cap", fake_market_cap)
monkeypatch.setattr(
"app.screener.snapshot.pykrx_stock.get_market_trading_value_by_ticker",
fake_trading_value,
)
def test_refresh_daily_writes_master_prices_flow(conn, monkeypatch):
_stub_pykrx(monkeypatch)
asof = dt.date(2026, 5, 12)
summary = refresh_daily(conn, asof)
assert summary["master_count"] == 3
assert summary["prices_count"] == 3
assert summary["flow_count"] == 3
assert conn.execute(
"SELECT close FROM krx_daily_prices WHERE ticker='005930' AND date='2026-05-12'"
).fetchone()[0] == 71000
def test_refresh_daily_is_idempotent(conn, monkeypatch):
_stub_pykrx(monkeypatch)
asof = dt.date(2026, 5, 12)
refresh_daily(conn, asof)
refresh_daily(conn, asof) # 두 번 호출
cnt = conn.execute(
"SELECT count(*) FROM krx_daily_prices WHERE date='2026-05-12'"
).fetchone()[0]
assert cnt == 3 # 중복 row 없어야 함
```
- [ ] **Step 2: Run test, expect failure**
```bash
pytest app/test_screener_snapshot.py -v
```
Expected: FAIL — module not found.
- [ ] **Step 3: snapshot.py 구현**
Create `web-backend/stock-lab/app/screener/snapshot.py`:
```python
"""KRX daily snapshot loader (pykrx -> SQLite cache)."""
from __future__ import annotations
import datetime as dt
import logging
import sqlite3
from dataclasses import dataclass
import pandas as pd
from pykrx import stock as pykrx_stock
log = logging.getLogger(__name__)
@dataclass
class RefreshSummary:
asof: dt.date
master_count: int
prices_count: int
flow_count: int
failures: list[str]
def asdict(self) -> dict:
return {
"asof": self.asof.isoformat(),
"master_count": self.master_count,
"prices_count": self.prices_count,
"flow_count": self.flow_count,
"failures": self.failures,
}
def _date_str(d: dt.date) -> str:
return d.strftime("%Y%m%d")
def _iso(d: dt.date) -> str:
return d.isoformat()
def _gather_master(asof: dt.date) -> list[tuple]:
rows: list[tuple] = []
now_iso = dt.datetime.utcnow().isoformat()
for market in ("KOSPI", "KOSDAQ"):
tickers = pykrx_stock.get_market_ticker_list(_date_str(asof), market=market)
cap_df = pykrx_stock.get_market_cap(_date_str(asof))
for ticker in tickers:
try:
name = pykrx_stock.get_market_ticker_name(ticker)
except Exception:
name = ticker
market_cap = int(cap_df.loc[ticker, "시가총액"]) if ticker in cap_df.index else None
# 관리종목·우선주·스팩 판정은 종목명 휴리스틱(MVP)
is_preferred = 1 if name.endswith("우") else 0
is_spac = 1 if "스팩" in name else 0
is_managed = 0 # pykrx에 직접 플래그 없음 — Phase 후속에서 KRX 공시 파이프로
rows.append((
ticker, name, market, market_cap,
is_managed, is_preferred, is_spac,
None, # listed_date — MVP에서는 unknown
now_iso,
))
return rows
def _gather_prices(asof: dt.date, tickers: list[str]) -> list[tuple]:
rows: list[tuple] = []
iso = _iso(asof)
for t in tickers:
try:
df = pykrx_stock.get_market_ohlcv(_date_str(asof), _date_str(asof), t)
if df.empty:
continue
row = df.iloc[0]
rows.append((
t, iso,
int(row["시가"]), int(row["고가"]), int(row["저가"]), int(row["종가"]),
int(row["거래량"]), int(row["거래대금"]),
))
except Exception as e:
log.warning("price fetch failed for %s: %s", t, e)
return rows
def _gather_flow(asof: dt.date) -> list[tuple]:
iso = _iso(asof)
try:
df = pykrx_stock.get_market_trading_value_by_ticker(_date_str(asof), _date_str(asof))
rows = [
(idx, iso, int(r["외국인"]), int(r["기관"]))
for idx, r in df.iterrows()
]
return rows
except Exception as e:
log.warning("flow fetch failed: %s", e)
return []
def refresh_daily(conn: sqlite3.Connection, asof: dt.date) -> dict:
"""Pull pykrx data for asof and upsert into the cache tables."""
master_rows = _gather_master(asof)
conn.executemany(
"""
INSERT INTO krx_master (
ticker, name, market, market_cap,
is_managed, is_preferred, is_spac,
listed_date, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(ticker) DO UPDATE SET
name=excluded.name, market=excluded.market,
market_cap=excluded.market_cap,
is_managed=excluded.is_managed,
is_preferred=excluded.is_preferred,
is_spac=excluded.is_spac,
updated_at=excluded.updated_at
""",
master_rows,
)
tickers = [r[0] for r in master_rows]
price_rows = _gather_prices(asof, tickers)
conn.executemany(
"""
INSERT OR REPLACE INTO krx_daily_prices (
ticker, date, open, high, low, close, volume, value
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
price_rows,
)
flow_rows = _gather_flow(asof)
conn.executemany(
"""
INSERT OR REPLACE INTO krx_flow (
ticker, date, foreign_net, institution_net
) VALUES (?, ?, ?, ?)
""",
flow_rows,
)
conn.commit()
return RefreshSummary(
asof=asof,
master_count=len(master_rows),
prices_count=len(price_rows),
flow_count=len(flow_rows),
failures=[],
).asdict()
def backfill(conn: sqlite3.Connection, start: dt.date, end: dt.date) -> list[dict]:
"""Run refresh_daily for every weekday in [start, end] (inclusive).
공휴일도 포함 호출하되, pykrx가 빈 응답을 주므로 자연스럽게 skip 됨.
"""
results = []
d = start
while d <= end:
if d.weekday() < 5: # 0=Mon ... 4=Fri
try:
results.append(refresh_daily(conn, d))
except Exception as e:
log.error("backfill failed for %s: %s", d, e)
results.append({"asof": d.isoformat(), "error": str(e)})
d += dt.timedelta(days=1)
return results
```
- [ ] **Step 4: Run test, expect pass**
```bash
pytest app/test_screener_snapshot.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/snapshot.py stock-lab/app/test_screener_snapshot.py
git commit -m "feat(stock-lab): pykrx 일봉·수급 캐시 갱신 (refresh_daily + backfill)"
```
---
## Task 0.4: ScreenContext.load() — DB → DataFrame
**Files:**
- Create: `web-backend/stock-lab/app/screener/engine.py` (ScreenContext만 우선)
- Create: `web-backend/stock-lab/app/screener/_test_fixtures.py`
- Create: `web-backend/stock-lab/app/test_screener_context.py`
- [ ] **Step 1: 테스트 픽스쳐 헬퍼**
Create `web-backend/stock-lab/app/screener/_test_fixtures.py`:
```python
"""Synthetic fixtures for screener tests — no DB / no pykrx."""
import datetime as dt
import pandas as pd
def make_master(tickers: list[str], market_caps: dict | None = None,
preferred: set | None = None, managed: set | None = None) -> pd.DataFrame:
market_caps = market_caps or {t: 100_000_000_000 for t in tickers}
preferred = preferred or set()
managed = managed or set()
return pd.DataFrame([
{
"ticker": t,
"name": f"테스트{t}",
"market": "KOSPI",
"market_cap": market_caps.get(t),
"is_managed": int(t in managed),
"is_preferred": int(t in preferred),
"is_spac": 0,
"listed_date": None,
}
for t in tickers
]).set_index("ticker")
def make_prices(tickers: list[str], days: int = 260, start_close: int = 50000,
trend_pct: float = 0.0, asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame:
"""trend_pct: 일별 종가 등락률(%). 양수면 상승 추세."""
rows = []
for t in tickers:
close = start_close
for i in range(days):
day_idx = days - 1 - i # asof가 마지막
date = asof - dt.timedelta(days=day_idx)
high = int(close * 1.012)
low = int(close * 0.988)
rows.append({
"ticker": t, "date": date.isoformat(),
"open": close, "high": high, "low": low, "close": close,
"volume": 1_000_000, "value": close * 1_000_000,
})
close = int(close * (1 + trend_pct / 100))
return pd.DataFrame(rows)
def make_flow(tickers: list[str], days: int = 260,
foreign_per_day: dict | None = None,
asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame:
foreign_per_day = foreign_per_day or {t: 0 for t in tickers}
rows = []
for t in tickers:
for i in range(days):
day_idx = days - 1 - i
date = asof - dt.timedelta(days=day_idx)
rows.append({
"ticker": t, "date": date.isoformat(),
"foreign_net": foreign_per_day.get(t, 0),
"institution_net": 0,
})
return pd.DataFrame(rows)
def make_kospi(days: int = 260, start: int = 2500, trend_pct: float = 0.0,
asof: dt.date = dt.date(2026, 5, 12)) -> pd.Series:
values = []
dates = []
v = start
for i in range(days):
day_idx = days - 1 - i
d = asof - dt.timedelta(days=day_idx)
dates.append(d.isoformat())
values.append(v)
v = v * (1 + trend_pct / 100)
return pd.Series(values, index=dates, name="kospi")
```
Create `web-backend/stock-lab/app/test_screener_context.py`:
```python
import datetime as dt
import sqlite3
import pytest
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.schema import ensure_screener_schema
from app.screener._test_fixtures import make_master, make_prices, make_flow
@pytest.fixture
def conn(tmp_path):
db_path = tmp_path / "ctx.db"
c = sqlite3.connect(db_path)
ensure_screener_schema(c)
yield c
c.close()
def _seed(conn, master_df, prices_df, flow_df):
now = dt.datetime.utcnow().isoformat()
for t, row in master_df.iterrows():
conn.execute("""INSERT INTO krx_master (ticker,name,market,market_cap,
is_managed,is_preferred,is_spac,listed_date,updated_at)
VALUES (?,?,?,?,?,?,?,?,?)""",
(t, row["name"], row["market"], row["market_cap"],
row["is_managed"], row["is_preferred"], row["is_spac"], None, now))
prices_df.to_sql("krx_daily_prices", conn, if_exists="append", index=False)
flow_df.to_sql("krx_flow", conn, if_exists="append", index=False)
conn.commit()
def test_load_returns_dataframes(conn):
asof = dt.date(2026, 5, 12)
_seed(conn,
make_master(["005930", "035420"]),
make_prices(["005930", "035420"], days=30, asof=asof),
make_flow(["005930", "035420"], days=30, asof=asof))
ctx = ScreenContext.load(conn, asof, lookback_days=30)
assert ctx.asof == asof
assert set(ctx.master.index) == {"005930", "035420"}
assert ctx.prices.shape[0] == 60 # 2 종목 × 30일
assert ctx.flow.shape[0] == 60
def test_restrict_filters_tickers(conn):
asof = dt.date(2026, 5, 12)
_seed(conn,
make_master(["005930", "035420", "091990"]),
make_prices(["005930", "035420", "091990"], days=30, asof=asof),
make_flow(["005930", "035420", "091990"], days=30, asof=asof))
ctx = ScreenContext.load(conn, asof, lookback_days=30)
scoped = ctx.restrict(pd.Index(["005930"]))
assert list(scoped.master.index) == ["005930"]
assert (scoped.prices["ticker"] == "005930").all()
assert (scoped.flow["ticker"] == "005930").all()
```
- [ ] **Step 2: Run test, expect failure**
```bash
pytest app/test_screener_context.py -v
```
Expected: FAIL — `ScreenContext` not defined.
- [ ] **Step 3: engine.py에 ScreenContext 구현 (엔진 본체는 Phase 2에서)**
Create `web-backend/stock-lab/app/screener/engine.py`:
```python
"""Screener engine — ScreenContext & Screener.
Phase 0: ScreenContext만 구현.
Phase 2에서 Screener / combine() 추가.
"""
from __future__ import annotations
import datetime as dt
import sqlite3
from dataclasses import dataclass, replace
import pandas as pd
@dataclass(frozen=True)
class ScreenContext:
"""1회 실행 동안 공유되는 읽기 전용 데이터 컨테이너."""
master: pd.DataFrame # index=ticker
prices: pd.DataFrame # cols: ticker,date,open,high,low,close,volume,value
flow: pd.DataFrame # cols: ticker,date,foreign_net,institution_net
kospi: pd.Series # index=date(str), name="kospi"
asof: dt.date
@classmethod
def load(cls, conn: sqlite3.Connection, asof: dt.date,
lookback_days: int = 252 * 2) -> "ScreenContext":
cutoff = (asof - dt.timedelta(days=int(lookback_days * 1.5))).isoformat()
asof_iso = asof.isoformat()
master = pd.read_sql_query(
"SELECT * FROM krx_master",
conn, index_col="ticker",
)
prices = pd.read_sql_query(
"SELECT ticker,date,open,high,low,close,volume,value "
"FROM krx_daily_prices WHERE date BETWEEN ? AND ? ORDER BY date",
conn, params=(cutoff, asof_iso),
)
flow = pd.read_sql_query(
"SELECT ticker,date,foreign_net,institution_net "
"FROM krx_flow WHERE date BETWEEN ? AND ? ORDER BY date",
conn, params=(cutoff, asof_iso),
)
# KOSPI 지수는 MVP에서 005930(삼성전자) 대용 또는 별도 인덱스 캐시.
# 현 단계는 005930로 fallback. Phase 후속에서 ^KS11 시계열 별도 캐시.
kospi = pd.Series(dtype=float, name="kospi")
if "005930" in master.index and not prices.empty:
sub = prices[prices["ticker"] == "005930"].set_index("date")["close"]
kospi = sub.copy()
kospi.name = "kospi"
return cls(master=master, prices=prices, flow=flow, kospi=kospi, asof=asof)
def restrict(self, tickers) -> "ScreenContext":
tickers = pd.Index(tickers)
return replace(
self,
master=self.master.loc[self.master.index.intersection(tickers)],
prices=self.prices[self.prices["ticker"].isin(tickers)],
flow=self.flow[self.flow["ticker"].isin(tickers)],
)
def latest_close(self) -> pd.Series:
if self.prices.empty:
return pd.Series(dtype=float)
latest = self.prices.sort_values("date").groupby("ticker")["close"].last()
return latest
def latest_high(self) -> pd.Series:
if self.prices.empty:
return pd.Series(dtype=float)
return self.prices.sort_values("date").groupby("ticker")["high"].last()
```
- [ ] **Step 4: Run test, expect pass**
```bash
pytest app/test_screener_context.py -v
```
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/engine.py stock-lab/app/screener/_test_fixtures.py stock-lab/app/test_screener_context.py
git commit -m "feat(stock-lab): ScreenContext.load/restrict + 합성 픽스쳐"
```
---
# Phase 1 — 노드 8개 (TDD)
각 노드 패턴은 동일: `nodes/base.py` 추상 → 노드 구현 → 단위 테스트 → commit.
## Task 1.1: ScoreNode / GateNode 추상 + percentile_rank 유틸
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/__init__.py` (빈 파일)
- Create: `web-backend/stock-lab/app/screener/nodes/base.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_base.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_base.py`:
```python
import pandas as pd
import pytest
from app.screener.nodes.base import percentile_rank
def test_percentile_rank_basic():
s = pd.Series([10, 20, 30, 40, 50])
out = percentile_rank(s)
assert out.iloc[0] == pytest.approx(10.0) # 1/5 * 100 - half adjustment
assert out.iloc[-1] == pytest.approx(90.0)
assert (out >= 0).all() and (out <= 100).all()
def test_percentile_rank_all_equal_returns_50():
s = pd.Series([42, 42, 42, 42])
out = percentile_rank(s)
assert (out == 50.0).all()
def test_percentile_rank_handles_nan():
s = pd.Series([1.0, float("nan"), 3.0, 5.0])
out = percentile_rank(s)
assert pd.isna(out.iloc[1])
assert (out.dropna() >= 0).all()
```
- [ ] **Step 2: Run test, expect failure**
```bash
pytest app/test_screener_nodes_base.py -v
```
Expected: FAIL — module not found.
- [ ] **Step 3: base.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/__init__.py` (빈 파일).
Create `web-backend/stock-lab/app/screener/nodes/base.py`:
```python
"""Node base classes + helpers."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, ClassVar
import pandas as pd
class ScoreNode(ABC):
name: ClassVar[str]
label: ClassVar[str]
default_params: ClassVar[dict]
param_schema: ClassVar[dict]
@abstractmethod
def compute(self, ctx: "Any", params: dict) -> pd.Series:
"""returns Series indexed by ticker, 0..100 float."""
class GateNode(ABC):
name: ClassVar[str]
label: ClassVar[str]
default_params: ClassVar[dict]
param_schema: ClassVar[dict]
@abstractmethod
def filter(self, ctx: "Any", params: dict) -> pd.Index:
"""returns surviving tickers."""
def percentile_rank(series: pd.Series) -> pd.Series:
"""Percentile rank in [0, 100]. All-equal → 50. NaN preserved."""
if series.empty:
return series.astype(float)
if series.dropna().nunique() == 1:
return pd.Series(50.0, index=series.index)
ranked = series.rank(pct=True, na_option="keep") * 100.0
return ranked
```
- [ ] **Step 4: Run test, expect pass**
```bash
pytest app/test_screener_nodes_base.py -v
```
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/__init__.py stock-lab/app/screener/nodes/base.py stock-lab/app/test_screener_nodes_base.py
git commit -m "feat(stock-lab): ScoreNode/GateNode 추상 + percentile_rank 유틸"
```
---
## Task 1.2: HygieneGate
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/hygiene.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_hygiene.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_hygiene.py`:
```python
import datetime as dt
from app.screener.nodes.hygiene import HygieneGate
from app.screener.engine import ScreenContext
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=__import__("pandas").Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_filter_excludes_small_cap():
g = HygieneGate()
ctx = _ctx(
make_master(["A", "B"], market_caps={"A": 1_000_000_000, "B": 100_000_000_000}),
make_prices(["A", "B"], days=30),
make_flow(["A", "B"], days=30),
)
out = g.filter(ctx, {**g.default_params, "min_listed_days": 0})
assert list(out) == ["B"]
def test_filter_excludes_preferred():
g = HygieneGate()
ctx = _ctx(
make_master(["A", "B"], preferred={"B"}),
make_prices(["A", "B"], days=30),
make_flow(["A", "B"], days=30),
)
out = g.filter(ctx, {**g.default_params, "min_listed_days": 0})
assert list(out) == ["A"]
def test_filter_excludes_low_value():
g = HygieneGate()
prices = make_prices(["A", "B"], days=30)
prices.loc[prices["ticker"] == "A", "value"] = 100_000 # 매우 작음
ctx = _ctx(make_master(["A", "B"]), prices, make_flow(["A", "B"], days=30))
out = g.filter(ctx, {**g.default_params, "min_listed_days": 0})
assert list(out) == ["B"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_hygiene.py -v`
Expected: FAIL — module not found.
- [ ] **Step 3: hygiene.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/hygiene.py`:
```python
"""HygieneGate — pre-filter for screener."""
from __future__ import annotations
import pandas as pd
from .base import GateNode
class HygieneGate(GateNode):
name = "hygiene"
label = "위생 게이트"
default_params = {
"min_market_cap_won": 50_000_000_000,
"min_avg_value_won": 500_000_000,
"min_listed_days": 60,
"skip_managed": True,
"skip_preferred": True,
"skip_spac": True,
"skip_halted_days": 3,
}
param_schema = {
"type": "object",
"properties": {
"min_market_cap_won": {"type": "integer", "minimum": 0},
"min_avg_value_won": {"type": "integer", "minimum": 0},
"min_listed_days": {"type": "integer", "minimum": 0},
"skip_managed": {"type": "boolean"},
"skip_preferred": {"type": "boolean"},
"skip_spac": {"type": "boolean"},
"skip_halted_days": {"type": "integer", "minimum": 0},
},
}
def filter(self, ctx, params: dict) -> pd.Index:
master = ctx.master.copy()
prices = ctx.prices
# 시총
master = master[master["market_cap"].fillna(0) >= params["min_market_cap_won"]]
# 우선주·관리·스팩
if params.get("skip_preferred", True):
master = master[master["is_preferred"] == 0]
if params.get("skip_managed", True):
master = master[master["is_managed"] == 0]
if params.get("skip_spac", True):
master = master[master["is_spac"] == 0]
candidates = master.index
# 20일 평균 거래대금
if not prices.empty:
recent20 = (
prices[prices["ticker"].isin(candidates)]
.sort_values("date")
.groupby("ticker")
.tail(20)
)
avg_value = recent20.groupby("ticker")["value"].mean()
ok = avg_value[avg_value >= params["min_avg_value_won"]].index
candidates = candidates.intersection(ok)
# 최근 N일 거래정지 (volume==0 N일 이상)
halted_days = params.get("skip_halted_days", 3)
if halted_days > 0 and not prices.empty:
recent = (
prices[prices["ticker"].isin(candidates)]
.sort_values("date")
.groupby("ticker")
.tail(halted_days)
)
zero_count = recent.assign(z=lambda d: (d["volume"] == 0).astype(int)) \
.groupby("ticker")["z"].sum()
healthy = zero_count[zero_count < halted_days].index
candidates = candidates.intersection(healthy)
# 상장 N일 — MVP에선 listed_date null 허용, null이면 통과
return pd.Index(candidates)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_hygiene.py -v`
Expected: 3 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/hygiene.py stock-lab/app/test_screener_nodes_hygiene.py
git commit -m "feat(stock-lab): HygieneGate — 위생 필터 (시총/거래대금/우선주/관리종목)"
```
---
## Task 1.3: ForeignBuy
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/foreign_buy.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_foreign_buy.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_foreign_buy.py`:
```python
import datetime as dt
import pandas as pd
import pytest
from app.screener.engine import ScreenContext
from app.screener.nodes.foreign_buy import ForeignBuy
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_higher_foreign_buy_gets_higher_score():
asof = dt.date(2026, 5, 12)
master = make_master(["A", "B"])
prices = make_prices(["A", "B"], days=30, asof=asof)
flow = make_flow(["A", "B"], days=30, asof=asof,
foreign_per_day={"A": 100_000_000, "B": 0})
out = ForeignBuy().compute(_ctx(master, prices, flow), {"window_days": 5})
assert out["A"] > out["B"]
assert 0 <= out.min() <= out.max() <= 100
def test_all_zero_returns_50():
asof = dt.date(2026, 5, 12)
master = make_master(["A", "B"])
prices = make_prices(["A", "B"], days=30, asof=asof)
flow = make_flow(["A", "B"], days=30, asof=asof, foreign_per_day={"A": 0, "B": 0})
out = ForeignBuy().compute(_ctx(master, prices, flow), {"window_days": 5})
assert (out == 50.0).all()
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_foreign_buy.py -v`
Expected: FAIL.
- [ ] **Step 3: foreign_buy.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/foreign_buy.py`:
```python
"""외국인 N일 누적 순매수 강도 (시총 대비)."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class ForeignBuy(ScoreNode):
name = "foreign_buy"
label = "외국인 누적 순매수"
default_params = {"window_days": 5}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 1, "maximum": 60, "default": 5}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 5))
flow = ctx.flow
if flow.empty:
return pd.Series(dtype=float)
last_dates = (
flow.sort_values("date").groupby("ticker").tail(window)
)
net_sum = last_dates.groupby("ticker")["foreign_net"].sum()
market_cap = ctx.master["market_cap"].fillna(0).reindex(net_sum.index)
raw = (net_sum / market_cap.replace(0, pd.NA)).astype(float)
return percentile_rank(raw).fillna(50.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_foreign_buy.py -v`
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/foreign_buy.py stock-lab/app/test_screener_nodes_foreign_buy.py
git commit -m "feat(stock-lab): ForeignBuy 노드 — 외국인 N일 누적 순매수 강도"
```
---
## Task 1.4: VolumeSurge
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/volume_surge.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_volume_surge.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_volume_surge.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.volume_surge import VolumeSurge
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_recent_volume_surge_gets_higher_score():
asof = dt.date(2026, 5, 12)
master = make_master(["A", "B"])
prices = make_prices(["A", "B"], days=30, asof=asof)
# A는 최근 3일 거래량 10배로
mask = (prices["ticker"] == "A") & (prices["date"] >= (asof - dt.timedelta(days=3)).isoformat())
prices.loc[mask, "volume"] *= 10
flow = make_flow(["A", "B"], days=30, asof=asof)
out = VolumeSurge().compute(
_ctx(master, prices, flow),
{"baseline_days": 20, "eval_days": 3},
)
assert out["A"] > out["B"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_volume_surge.py -v`
- [ ] **Step 3: volume_surge.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/volume_surge.py`:
```python
"""거래량 급증 — log1p(recent/baseline)."""
import numpy as np
import pandas as pd
from .base import ScoreNode, percentile_rank
class VolumeSurge(ScoreNode):
name = "volume_surge"
label = "거래량 급증"
default_params = {"baseline_days": 20, "eval_days": 3}
param_schema = {
"type": "object",
"properties": {
"baseline_days": {"type": "integer", "minimum": 5, "maximum": 60, "default": 20},
"eval_days": {"type": "integer", "minimum": 1, "maximum": 10, "default": 3},
},
}
def compute(self, ctx, params: dict) -> pd.Series:
baseline = int(params.get("baseline_days", 20))
eval_d = int(params.get("eval_days", 3))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last_recent = ordered.groupby("ticker").tail(eval_d).groupby("ticker")["volume"].mean()
last_baseline = (
ordered.groupby("ticker")
.tail(baseline + eval_d)
.groupby("ticker")
.head(baseline)
.groupby("ticker")["volume"]
.mean()
)
ratio = last_recent / last_baseline.replace(0, pd.NA)
raw = np.log1p(ratio.astype(float))
return percentile_rank(raw).fillna(50.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_volume_surge.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/volume_surge.py stock-lab/app/test_screener_nodes_volume_surge.py
git commit -m "feat(stock-lab): VolumeSurge 노드 — log(최근/평균) 거래량 급증"
```
---
## Task 1.5: Momentum20
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/momentum.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_momentum.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_momentum.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.momentum import Momentum20
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_higher_momentum_gets_higher_score():
asof = dt.date(2026, 5, 12)
master = make_master(["UP", "DN"])
up = make_prices(["UP"], days=30, asof=asof, trend_pct=0.5)
dn = make_prices(["DN"], days=30, asof=asof, trend_pct=-0.3)
prices = pd.concat([up, dn], ignore_index=True)
flow = make_flow(["UP", "DN"], days=30, asof=asof)
out = Momentum20().compute(_ctx(master, prices, flow), {"window_days": 20})
assert out["UP"] > out["DN"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_momentum.py -v`
- [ ] **Step 3: momentum.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/momentum.py`:
```python
"""20일 모멘텀."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class Momentum20(ScoreNode):
name = "momentum"
label = "20일 모멘텀"
default_params = {"window_days": 20}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 5, "maximum": 120, "default": 20}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 20))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last = ordered.groupby("ticker").tail(window + 1)
def _ret(g):
if len(g) < window + 1:
return float("nan")
return g["close"].iloc[-1] / g["close"].iloc[0] - 1
raw = last.groupby("ticker").apply(_ret)
return percentile_rank(raw).fillna(50.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_momentum.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/momentum.py stock-lab/app/test_screener_nodes_momentum.py
git commit -m "feat(stock-lab): Momentum20 노드 — N일 수익률 백분위"
```
---
## Task 1.6: High52WProximity
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/high52w.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_high52w.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_high52w.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.high52w import High52WProximity
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_proximity_at_high_returns_100():
asof = dt.date(2026, 5, 12)
master = make_master(["A"])
prices = make_prices(["A"], days=260, asof=asof, trend_pct=0.05)
flow = make_flow(["A"], days=260, asof=asof)
out = High52WProximity().compute(_ctx(master, prices, flow), {"window_days": 252})
assert out["A"] >= 95
def test_proximity_below_70pct_returns_0():
asof = dt.date(2026, 5, 12)
master = make_master(["A"])
prices = make_prices(["A"], days=260, asof=asof, start_close=100000, trend_pct=-0.5)
flow = make_flow(["A"], days=260, asof=asof)
out = High52WProximity().compute(_ctx(master, prices, flow), {"window_days": 252})
assert out["A"] == 0
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_high52w.py -v`
- [ ] **Step 3: high52w.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/high52w.py`:
```python
"""52주 신고가 근접도 (룰 기반: 70% 미만 0점, 100% 도달 100점, 선형)."""
import pandas as pd
from .base import ScoreNode
class High52WProximity(ScoreNode):
name = "high52w"
label = "52주 신고가 근접도"
default_params = {"window_days": 252}
param_schema = {
"type": "object",
"properties": {
"window_days": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
window = int(params.get("window_days", 252))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
last = ordered.groupby("ticker").tail(window)
agg = last.groupby("ticker").agg(close=("close", "last"), high=("high", "max"))
proximity = (agg["close"] / agg["high"]).clip(upper=1.0)
score = ((proximity - 0.7) / 0.3).clip(lower=0.0, upper=1.0) * 100.0
return score.fillna(0.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_high52w.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/high52w.py stock-lab/app/test_screener_nodes_high52w.py
git commit -m "feat(stock-lab): High52WProximity 노드 — 신고가 대비 근접도 룰 점수"
```
---
## Task 1.7: RsRating
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/rs_rating.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_rs_rating.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_rs_rating.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.rs_rating import RsRating
from app.screener._test_fixtures import make_master, make_prices, make_flow, make_kospi
def _ctx(master, prices, flow, kospi):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=kospi, asof=dt.date(2026, 5, 12))
def test_outperformer_gets_higher_score():
asof = dt.date(2026, 5, 12)
master = make_master(["UP", "DN"])
up = make_prices(["UP"], days=260, asof=asof, trend_pct=0.3)
dn = make_prices(["DN"], days=260, asof=asof, trend_pct=-0.1)
prices = pd.concat([up, dn], ignore_index=True)
flow = make_flow(["UP", "DN"], days=260, asof=asof)
kospi = make_kospi(days=260, asof=asof, trend_pct=0.0)
out = RsRating().compute(_ctx(master, prices, flow, kospi),
RsRating.default_params)
assert out["UP"] > out["DN"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_rs_rating.py -v`
- [ ] **Step 3: rs_rating.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/rs_rating.py`:
```python
"""RS Rating — IBD 가중 (3m=2,6m=1,9m=1,12m=1)."""
import pandas as pd
from .base import ScoreNode, percentile_rank
_PERIOD_TO_DAYS = {"3m": 63, "6m": 126, "9m": 189, "12m": 252}
class RsRating(ScoreNode):
name = "rs_rating"
label = "RS Rating (시장 대비 상대강도)"
default_params = {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}}
param_schema = {
"type": "object",
"properties": {
"weights": {"type": "object"}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
weights: dict = params.get("weights", self.default_params["weights"])
prices = ctx.prices
kospi = ctx.kospi
if prices.empty or kospi.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
def _excess_for_ticker(g: pd.DataFrame) -> float:
closes = g.set_index("date")["close"]
total = 0.0
wsum = 0.0
for period, w in weights.items():
k = _PERIOD_TO_DAYS.get(period, 0)
if len(closes) <= k or len(kospi) <= k:
continue
r_stock = closes.iloc[-1] / closes.iloc[-(k + 1)] - 1
r_market = kospi.iloc[-1] / kospi.iloc[-(k + 1)] - 1
total += w * (r_stock - r_market)
wsum += w
return total / wsum if wsum else float("nan")
raw = ordered.groupby("ticker").apply(_excess_for_ticker)
return percentile_rank(raw).fillna(50.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_rs_rating.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/rs_rating.py stock-lab/app/test_screener_nodes_rs_rating.py
git commit -m "feat(stock-lab): RsRating 노드 — IBD 가중 시장초과수익 백분위"
```
---
## Task 1.8: MaAlignment
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/ma_alignment.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_ma_alignment.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_ma_alignment.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.ma_alignment import MaAlignment
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_strong_uptrend_returns_100():
asof = dt.date(2026, 5, 12)
master = make_master(["UP"])
prices = make_prices(["UP"], days=260, asof=asof, start_close=50000, trend_pct=0.2)
flow = make_flow(["UP"], days=260, asof=asof)
out = MaAlignment().compute(_ctx(master, prices, flow), MaAlignment.default_params)
assert out["UP"] == 100.0
def test_downtrend_returns_low():
asof = dt.date(2026, 5, 12)
master = make_master(["DN"])
prices = make_prices(["DN"], days=260, asof=asof, start_close=100000, trend_pct=-0.1)
flow = make_flow(["DN"], days=260, asof=asof)
out = MaAlignment().compute(_ctx(master, prices, flow), MaAlignment.default_params)
assert out["DN"] <= 20.0
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_ma_alignment.py -v`
- [ ] **Step 3: ma_alignment.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/ma_alignment.py`:
```python
"""이평선 정배열 점수 — 5개 조건 충족 개수 / 5 × 100."""
import pandas as pd
from .base import ScoreNode
class MaAlignment(ScoreNode):
name = "ma_alignment"
label = "이평선 정배열"
default_params = {"ma_periods": [50, 150, 200]}
param_schema = {
"type": "object",
"properties": {
"ma_periods": {"type": "array", "items": {"type": "integer"}}
},
}
def compute(self, ctx, params: dict) -> pd.Series:
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date")
def _score(g: pd.DataFrame) -> float:
closes = g["close"].astype(float)
if len(closes) < 252:
return float("nan")
close = closes.iloc[-1]
ma50 = closes.rolling(50).mean().iloc[-1]
ma150 = closes.rolling(150).mean().iloc[-1]
ma200 = closes.rolling(200).mean().iloc[-1]
low52 = closes.iloc[-252:].min()
conds = [
close > ma50,
ma50 > ma150,
ma150 > ma200,
close > ma200,
close >= low52 * 1.25,
]
return sum(conds) / 5 * 100.0
raw = ordered.groupby("ticker").apply(_score)
return raw.fillna(0.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_ma_alignment.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/ma_alignment.py stock-lab/app/test_screener_nodes_ma_alignment.py
git commit -m "feat(stock-lab): MaAlignment 노드 — 이평선 정배열 5조건 룰 점수"
```
---
## Task 1.9: VcpLite
**Files:**
- Create: `web-backend/stock-lab/app/screener/nodes/vcp_lite.py`
- Create: `web-backend/stock-lab/app/test_screener_nodes_vcp_lite.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_nodes_vcp_lite.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.nodes.vcp_lite import VcpLite
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_contracting_stock_scores_higher_than_expanding():
asof = dt.date(2026, 5, 12)
master = make_master(["CON", "EXP"])
prices = make_prices(["CON", "EXP"], days=260, asof=asof)
# CON: 최근 40일 변동성 축소 (high/low 좁힘)
mask_recent_con = (prices["ticker"] == "CON") & (
prices["date"] >= (asof - dt.timedelta(days=40)).isoformat()
)
prices.loc[mask_recent_con, "high"] = (prices.loc[mask_recent_con, "close"] * 1.003).astype(int)
prices.loc[mask_recent_con, "low"] = (prices.loc[mask_recent_con, "close"] * 0.997).astype(int)
# EXP: 최근 40일 변동성 확대
mask_recent_exp = (prices["ticker"] == "EXP") & (
prices["date"] >= (asof - dt.timedelta(days=40)).isoformat()
)
prices.loc[mask_recent_exp, "high"] = (prices.loc[mask_recent_exp, "close"] * 1.05).astype(int)
prices.loc[mask_recent_exp, "low"] = (prices.loc[mask_recent_exp, "close"] * 0.95).astype(int)
flow = make_flow(["CON", "EXP"], days=260, asof=asof)
out = VcpLite().compute(_ctx(master, prices, flow), VcpLite.default_params)
assert out["CON"] > out["EXP"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_nodes_vcp_lite.py -v`
- [ ] **Step 3: vcp_lite.py 구현**
Create `web-backend/stock-lab/app/screener/nodes/vcp_lite.py`:
```python
"""VCP-lite — 단기/장기 일중 변동성 비율 기반 수축률."""
import pandas as pd
from .base import ScoreNode, percentile_rank
class VcpLite(ScoreNode):
name = "vcp_lite"
label = "VCP-lite (변동성 수축)"
default_params = {"short_window": 40, "long_window": 252}
param_schema = {
"type": "object",
"properties": {
"short_window": {"type": "integer", "minimum": 10, "maximum": 120, "default": 40},
"long_window": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252},
},
}
def compute(self, ctx, params: dict) -> pd.Series:
short_w = int(params.get("short_window", 40))
long_w = int(params.get("long_window", 252))
prices = ctx.prices
if prices.empty:
return pd.Series(dtype=float)
ordered = prices.sort_values("date").copy()
ordered["range_pct"] = (ordered["high"] - ordered["low"]) / ordered["close"]
def _ratio(g: pd.DataFrame) -> float:
if len(g) < long_w:
return float("nan")
short_vol = g["range_pct"].tail(short_w).mean()
long_vol = g["range_pct"].tail(long_w).mean()
if long_vol == 0 or pd.isna(long_vol):
return float("nan")
return 1 - (short_vol / long_vol)
raw = ordered.groupby("ticker").apply(_ratio)
return percentile_rank(raw).fillna(50.0)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_nodes_vcp_lite.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/nodes/vcp_lite.py stock-lab/app/test_screener_nodes_vcp_lite.py
git commit -m "feat(stock-lab): VcpLite 노드 — 변동성 수축률 백분위"
```
---
# Phase 2 — 엔진 · 포지션 사이저 · 텔레그램 · 레지스트리
## Task 2.1: position_sizer.py — ATR Wilder + entry/stop/target
**Files:**
- Create: `web-backend/stock-lab/app/screener/position_sizer.py`
- Create: `web-backend/stock-lab/app/test_screener_position_sizer.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_position_sizer.py`:
```python
import datetime as dt
import pandas as pd
from app.screener.engine import ScreenContext
from app.screener.position_sizer import compute_atr_wilder, plan_positions
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=pd.Series(dtype=float, name="kospi"),
asof=dt.date(2026, 5, 12))
def test_atr_wilder_positive_and_smooth():
df = make_prices(["A"], days=30)
atr = compute_atr_wilder(df[df["ticker"] == "A"], window=14)
assert atr > 0
def test_plan_positions_returns_entry_stop_target():
asof = dt.date(2026, 5, 12)
master = make_master(["A"])
prices = make_prices(["A"], days=30, asof=asof, start_close=50000)
flow = make_flow(["A"], days=30, asof=asof)
ctx = _ctx(master, prices, flow)
sizing = plan_positions(ctx, ["A"], {"atr_window": 14, "atr_stop_mult": 2.0, "rr_ratio": 2.0})
row = sizing["A"]
assert row["entry_price"] > 0
assert row["stop_price"] < row["entry_price"]
assert row["target_price"] > row["entry_price"]
assert row["atr14"] > 0
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_position_sizer.py -v`
- [ ] **Step 3: position_sizer.py 구현**
Create `web-backend/stock-lab/app/screener/position_sizer.py`:
```python
"""ATR Wilder smoothing + entry/stop/target 계산."""
import pandas as pd
def compute_atr_wilder(df_one_ticker: pd.DataFrame, window: int = 14) -> float:
"""단일 종목 DataFrame(date·open·high·low·close)에 대해 Wilder ATR 마지막 값."""
g = df_one_ticker.sort_values("date").copy()
high = g["high"].astype(float)
low = g["low"].astype(float)
close = g["close"].astype(float)
prev_close = close.shift(1)
tr = pd.concat([
(high - low),
(high - prev_close).abs(),
(low - prev_close).abs(),
], axis=1).max(axis=1)
atr = tr.ewm(alpha=1 / window, adjust=False).mean()
return float(atr.iloc[-1])
def round_won(x: float) -> int:
return int(round(x))
def plan_positions(ctx, tickers: list[str], params: dict) -> dict:
"""각 ticker 에 대해 entry/stop/target/atr14 반환."""
atr_window = int(params.get("atr_window", 14))
stop_mult = float(params.get("atr_stop_mult", 2.0))
rr = float(params.get("rr_ratio", 2.0))
prices = ctx.prices.sort_values("date")
out: dict = {}
for t in tickers:
sub = prices[prices["ticker"] == t]
if sub.empty:
continue
close = float(sub["close"].iloc[-1])
atr14 = compute_atr_wilder(sub, window=atr_window)
entry = round_won(close * 1.005)
stop = round_won(close - stop_mult * atr14)
target = round_won(entry + rr * (entry - stop))
r_pct = (entry - stop) / entry * 100 if entry else 0.0
out[t] = {
"entry_price": entry,
"stop_price": stop,
"target_price": target,
"atr14": atr14,
"r_pct": r_pct,
}
return out
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_position_sizer.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/position_sizer.py stock-lab/app/test_screener_position_sizer.py
git commit -m "feat(stock-lab): position_sizer — ATR Wilder + entry/stop/target"
```
---
## Task 2.2: registry.py + engine.Screener + combine
**Files:**
- Create: `web-backend/stock-lab/app/screener/registry.py`
- Modify: `web-backend/stock-lab/app/screener/engine.py` (Screener·combine 추가)
- Create: `web-backend/stock-lab/app/test_screener_engine.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_engine.py`:
```python
import datetime as dt
import pandas as pd
import pytest
from app.screener.engine import ScreenContext, Screener, combine
from app.screener.nodes.hygiene import HygieneGate
from app.screener.nodes.foreign_buy import ForeignBuy
from app.screener.nodes.momentum import Momentum20
from app.screener._test_fixtures import make_master, make_prices, make_flow, make_kospi
def _ctx(master, prices, flow):
return ScreenContext(master=master, prices=prices, flow=flow,
kospi=make_kospi(days=260),
asof=dt.date(2026, 5, 12))
def test_combine_weighted_average():
scores = {
"foreign_buy": pd.Series({"A": 80, "B": 20}),
"momentum": pd.Series({"A": 60, "B": 40}),
}
weights = {"foreign_buy": 2.0, "momentum": 1.0}
out = combine(scores, weights)
# A: (80*2 + 60*1)/3 = 73.33
assert abs(out["A"] - 73.333) < 0.1
assert abs(out["B"] - 26.666) < 0.1
def test_combine_all_zero_weight_raises():
scores = {"foreign_buy": pd.Series({"A": 80})}
with pytest.raises(ValueError, match="no active"):
combine(scores, {"foreign_buy": 0})
def test_screener_run_end_to_end():
asof = dt.date(2026, 5, 12)
master = make_master(["GOOD", "SMALL"],
market_caps={"GOOD": 200_000_000_000, "SMALL": 1_000_000_000})
prices = make_prices(["GOOD", "SMALL"], days=260, asof=asof, trend_pct=0.1)
flow = make_flow(["GOOD", "SMALL"], days=260, asof=asof,
foreign_per_day={"GOOD": 100_000_000, "SMALL": 0})
ctx = _ctx(master, prices, flow)
screener = Screener(
gate=HygieneGate(),
score_nodes=[ForeignBuy(), Momentum20()],
weights={"foreign_buy": 1.0, "momentum": 1.0},
node_params={"foreign_buy": {"window_days": 5}, "momentum": {"window_days": 20}},
gate_params={**HygieneGate.default_params, "min_listed_days": 0},
top_n=10,
)
result = screener.run(ctx)
assert result.survivors_count == 1 # SMALL은 게이트 탈락
assert result.ranked.index[0] == "GOOD"
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_engine.py -v`
Expected: FAIL — Screener / combine 없음.
- [ ] **Step 3: registry.py 작성**
Create `web-backend/stock-lab/app/screener/registry.py`:
```python
"""Registry of node classes (single source of truth for /nodes endpoint)."""
from .nodes.hygiene import HygieneGate
from .nodes.foreign_buy import ForeignBuy
from .nodes.volume_surge import VolumeSurge
from .nodes.momentum import Momentum20
from .nodes.high52w import High52WProximity
from .nodes.rs_rating import RsRating
from .nodes.ma_alignment import MaAlignment
from .nodes.vcp_lite import VcpLite
NODE_REGISTRY: dict[str, type] = {
"foreign_buy": ForeignBuy,
"volume_surge": VolumeSurge,
"momentum": Momentum20,
"high52w": High52WProximity,
"rs_rating": RsRating,
"ma_alignment": MaAlignment,
"vcp_lite": VcpLite,
}
GATE_REGISTRY: dict[str, type] = {
"hygiene": HygieneGate,
}
```
- [ ] **Step 4: engine.py 확장**
Append to `web-backend/stock-lab/app/screener/engine.py`:
```python
# ---- combine + Screener (Phase 2) ----
from dataclasses import field as _field
import pandas as _pd
from . import position_sizer as _ps
def combine(scores: dict, weights: dict) -> pd.Series:
"""Weighted average across score nodes. ValueError if all weights = 0."""
active = {k: w for k, w in weights.items() if w > 0 and k in scores}
if not active:
raise ValueError("no active score nodes (all weights = 0)")
df = pd.DataFrame({k: scores[k] for k in active})
w = pd.Series(active)
weighted = (df.fillna(0).multiply(w, axis=1)).sum(axis=1) / w.sum()
return weighted
@dataclass
class ScreenerResult:
asof: dt.date
survivors_count: int
scores: dict # node name → pd.Series
weights: dict
ranked: pd.Series # ticker → total_score (sorted desc, head=top_n)
rows: list # list of dicts (for serialization)
warnings: list
class Screener:
def __init__(self, gate, score_nodes, weights: dict, node_params: dict,
gate_params: dict, top_n: int = 20, sizer_params: dict | None = None):
self.gate = gate
self.score_nodes = score_nodes
self.weights = weights
self.node_params = node_params
self.gate_params = gate_params
self.top_n = top_n
self.sizer_params = sizer_params or {"atr_window": 14, "atr_stop_mult": 2.0, "rr_ratio": 2.0}
def run(self, ctx: ScreenContext) -> ScreenerResult:
warnings: list[str] = []
survivors = self.gate.filter(ctx, self.gate_params)
if len(survivors) == 0:
raise ValueError("no survivors after hygiene gate")
if len(survivors) < 100:
warnings.append(f"survivors_count={len(survivors)} < 100 — 백분위 정규화 신뢰도 낮음")
scoped = ctx.restrict(survivors)
scores: dict = {}
for n in self.score_nodes:
w = self.weights.get(n.name, 0)
if w <= 0:
continue
try:
scores[n.name] = n.compute(scoped, self.node_params.get(n.name, {}))
except Exception as e:
warnings.append(f"node '{n.name}' failed: {e}")
scores[n.name] = pd.Series(0.0, index=scoped.master.index)
total = combine(scores, self.weights)
ranked = total.sort_values(ascending=False).head(self.top_n)
sizing = _ps.plan_positions(scoped, list(ranked.index), self.sizer_params)
latest_close = scoped.latest_close()
rows = []
for rank_idx, ticker in enumerate(ranked.index, start=1):
s = sizing.get(ticker, {})
row = {
"rank": rank_idx,
"ticker": ticker,
"name": str(scoped.master.loc[ticker, "name"]),
"total_score": float(ranked.loc[ticker]),
"scores": {k: float(v.get(ticker, 0.0)) for k, v in scores.items()},
"close": int(latest_close.get(ticker, 0)),
"market_cap": int(scoped.master.loc[ticker, "market_cap"] or 0),
"entry_price": s.get("entry_price"),
"stop_price": s.get("stop_price"),
"target_price": s.get("target_price"),
"atr14": s.get("atr14"),
"r_pct": s.get("r_pct"),
}
rows.append(row)
return ScreenerResult(
asof=ctx.asof, survivors_count=len(survivors),
scores=scores, weights=self.weights,
ranked=ranked, rows=rows, warnings=warnings,
)
```
- [ ] **Step 5: __init__.py 활성화**
Modify `web-backend/stock-lab/app/screener/__init__.py`:
```python
"""Stock screener — KRX 강세주 분석 노드 기반 보드."""
from .engine import Screener, ScreenContext, ScreenerResult
from .registry import NODE_REGISTRY, GATE_REGISTRY
__all__ = [
"Screener", "ScreenContext", "ScreenerResult",
"NODE_REGISTRY", "GATE_REGISTRY",
]
```
- [ ] **Step 6: Run test, expect pass**
Run: `pytest app/test_screener_engine.py -v`
Expected: 3 passed.
- [ ] **Step 7: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/registry.py stock-lab/app/screener/engine.py stock-lab/app/screener/__init__.py stock-lab/app/test_screener_engine.py
git commit -m "feat(stock-lab): Screener 엔진 + combine + ScreenerResult + 노드 레지스트리"
```
---
## Task 2.3: telegram.py — 메시지 빌더
**Files:**
- Create: `web-backend/stock-lab/app/screener/telegram.py`
- Create: `web-backend/stock-lab/app/test_screener_telegram.py`
- [ ] **Step 1: 테스트**
Create `web-backend/stock-lab/app/test_screener_telegram.py`:
```python
import datetime as dt
from app.screener.telegram import build_telegram_payload
def test_build_payload_includes_top10_and_link():
rows = [
{
"rank": i, "ticker": f"00{i:04}", "name": f"종목{i}",
"total_score": 90 - i,
"scores": {"foreign_buy": 80 + i, "volume_surge": 60, "momentum": 70,
"high52w": 75, "rs_rating": 85, "ma_alignment": 80, "vcp_lite": 30},
"close": 50000, "entry_price": 50250, "stop_price": 48500,
"target_price": 53750, "r_pct": 3.5,
}
for i in range(1, 21)
]
p = build_telegram_payload(
asof=dt.date(2026, 5, 12),
mode="auto",
survivors_count=612,
top_n=20,
rows=rows,
run_id=42,
)
assert p["parse_mode"] == "MarkdownV2"
text = p["text"]
assert "2026-05-12" in text
assert "종목1" in text
assert "종목10" in text
assert "종목11" not in text # 본문 1-10만
assert "run_id=42" in text
def test_score_threshold_filters_icons():
rows = [{
"rank": 1, "ticker": "A", "name": "A주",
"total_score": 80,
"scores": {"foreign_buy": 90, "volume_surge": 50, "momentum": 70,
"high52w": 30, "rs_rating": 80, "ma_alignment": 80, "vcp_lite": 60},
"close": 50000, "entry_price": 50250, "stop_price": 48500,
"target_price": 53750, "r_pct": 3.5,
}]
p = build_telegram_payload(dt.date(2026, 5, 12), "auto", 100, 1, rows, run_id=1)
# foreign_buy(90), momentum(70), rs_rating(80), ma_alignment(80) 만 표시 (≥70)
assert "👤외" in p["text"]
assert "🚀모" in p["text"]
assert "💪RS" in p["text"]
assert "📈MA" in p["text"]
assert "⚡거" not in p["text"]
assert "🆙고" not in p["text"]
assert "🌀VCP" not in p["text"]
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_telegram.py -v`
- [ ] **Step 3: telegram.py 구현**
Create `web-backend/stock-lab/app/screener/telegram.py`:
```python
"""Telegram payload builder. Caller (agent-office) handles actual delivery."""
from __future__ import annotations
import datetime as dt
NODE_ICONS = {
"foreign_buy": "👤외",
"volume_surge": "⚡거",
"momentum": "🚀모",
"high52w": "🆙고",
"rs_rating": "💪RS",
"ma_alignment": "📈MA",
"vcp_lite": "🌀VCP",
}
PAGE_BASE = "https://gahusb.synology.me/stock/screener"
def _escape_md(s: str) -> str:
"""Minimal MarkdownV2 escape — extend if formatting breaks."""
for ch in r"\_*[]()~`>#+-=|{}.!":
s = s.replace(ch, "\\" + ch)
return s
def _format_won(n: int | None) -> str:
if n is None:
return "-"
return f"{int(n):,}"
def build_telegram_payload(asof: dt.date, mode: str, survivors_count: int,
top_n: int, rows: list[dict], run_id: int | None) -> dict:
title = "*KRX 강세주 스크리너*"
header = (
f"🎯 {title} — {asof.isoformat()} \\({mode}\\)\n"
f"통과 {survivors_count}종 / Top {top_n} / 본문 1\\-10"
)
lines = []
for r in rows[:10]:
icons = " ".join(
NODE_ICONS[name] for name, sc in r["scores"].items()
if sc >= 70 and name in NODE_ICONS
)
score_str = f"{r['total_score']:.1f}"
lines.append(
f"{r['rank']}\\. *{_escape_md(r['name'])}* `{r['ticker']}` "
f"⭐ {_escape_md(score_str)}\n"
f" {icons}\n"
f" 진입 {_format_won(r.get('entry_price'))} "
f"손절 {_format_won(r.get('stop_price'))} "
f"익절 {_format_won(r.get('target_price'))} "
f"\\(R {_escape_md(f'{r.get('r_pct', 0):.1f}')}%\\)"
)
link = f"🔗 전체 결과·11~20위:\n{PAGE_BASE}?run\\_id={run_id}" if run_id else ""
text = header + "\n\n" + "\n\n".join(lines) + ("\n\n" + link if link else "")
return {
"chat_target": "default",
"parse_mode": "MarkdownV2",
"text": text,
}
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_telegram.py -v`
Expected: 2 passed.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/telegram.py stock-lab/app/test_screener_telegram.py
git commit -m "feat(stock-lab): telegram.py 메시지 빌더 (Top10 + 아이콘 + 페이지 링크)"
```
---
# Phase 3 — FastAPI 라우터
## Task 3.1: schemas.py (Pydantic)
**Files:**
- Create: `web-backend/stock-lab/app/screener/schemas.py`
- [ ] **Step 1: schemas.py 작성**
Create `web-backend/stock-lab/app/screener/schemas.py`:
```python
from __future__ import annotations
from typing import Literal, Optional
from pydantic import BaseModel, Field
class NodeMeta(BaseModel):
name: str
label: str
default_params: dict
param_schema: dict
class NodesResponse(BaseModel):
score_nodes: list[NodeMeta]
gate_nodes: list[NodeMeta]
class SettingsBody(BaseModel):
weights: dict[str, float]
node_params: dict[str, dict] = Field(default_factory=dict)
gate_params: dict
top_n: int = 20
rr_ratio: float = 2.0
atr_window: int = 14
atr_stop_mult: float = 2.0
class SettingsResponse(SettingsBody):
updated_at: str
class RunRequest(BaseModel):
mode: Literal["preview", "manual_save", "auto"] = "preview"
asof: Optional[str] = None
weights: Optional[dict[str, float]] = None
node_params: Optional[dict[str, dict]] = None
gate_params: Optional[dict] = None
top_n: Optional[int] = None
class ResultRow(BaseModel):
rank: int
ticker: str
name: str
total_score: float
scores: dict[str, float]
close: int
market_cap: int
entry_price: Optional[int] = None
stop_price: Optional[int] = None
target_price: Optional[int] = None
atr14: Optional[float] = None
r_pct: Optional[float] = None
class TelegramPayload(BaseModel):
chat_target: str
parse_mode: str
text: str
class RunResponse(BaseModel):
asof: str
mode: str
status: Literal["success", "failed", "skipped_holiday"]
run_id: Optional[int] = None
survivors_count: Optional[int] = None
weights: dict[str, float]
top_n: int
results: list[ResultRow] = Field(default_factory=list)
telegram_payload: Optional[TelegramPayload] = None
warnings: list[str] = Field(default_factory=list)
error: Optional[str] = None
class RunSummary(BaseModel):
id: int
asof: str
mode: str
status: str
started_at: str
finished_at: Optional[str] = None
top_n: int
survivors_count: Optional[int] = None
telegram_sent: bool
```
- [ ] **Step 2: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/schemas.py
git commit -m "feat(stock-lab): screener Pydantic 스키마"
```
---
## Task 3.2: router.py — /nodes + /settings
**Files:**
- Create: `web-backend/stock-lab/app/screener/router.py`
- Create: `web-backend/stock-lab/app/test_screener_router.py`
- [ ] **Step 1: 테스트 (nodes + settings round-trip)**
Create `web-backend/stock-lab/app/test_screener_router.py`:
```python
import os
import sqlite3
import pytest
from fastapi.testclient import TestClient
from app.main import app # 가정: app.main 에 FastAPI 인스턴스 노출
from app.screener.schema import ensure_screener_schema
@pytest.fixture(autouse=True)
def isolated_db(tmp_path, monkeypatch):
db_path = tmp_path / "screener_router.db"
c = sqlite3.connect(db_path)
ensure_screener_schema(c)
c.close()
monkeypatch.setenv("STOCK_DB_PATH", str(db_path))
@pytest.fixture
def client():
return TestClient(app)
def test_get_nodes_lists_7_score_and_1_gate(client):
r = client.get("/api/stock/screener/nodes")
assert r.status_code == 200
body = r.json()
assert len(body["score_nodes"]) == 7
assert len(body["gate_nodes"]) == 1
assert {n["name"] for n in body["score_nodes"]} == {
"foreign_buy", "volume_surge", "momentum",
"high52w", "rs_rating", "ma_alignment", "vcp_lite",
}
def test_settings_get_returns_defaults(client):
r = client.get("/api/stock/screener/settings")
assert r.status_code == 200
body = r.json()
assert body["weights"]["foreign_buy"] == 1.0
assert body["top_n"] == 20
def test_settings_put_then_get_round_trip(client):
new_settings = {
"weights": {"foreign_buy": 2.5, "momentum": 1.0, "volume_surge": 1.0,
"high52w": 1.2, "rs_rating": 1.2, "ma_alignment": 1.0, "vcp_lite": 0.8},
"node_params": {"foreign_buy": {"window_days": 7}},
"gate_params": {"min_market_cap_won": 100_000_000_000,
"min_avg_value_won": 500_000_000,
"min_listed_days": 60,
"skip_managed": True, "skip_preferred": True, "skip_spac": True,
"skip_halted_days": 3},
"top_n": 30,
"rr_ratio": 2.5,
"atr_window": 14,
"atr_stop_mult": 2.0,
}
r = client.put("/api/stock/screener/settings", json=new_settings)
assert r.status_code == 200
r2 = client.get("/api/stock/screener/settings")
body = r2.json()
assert body["weights"]["foreign_buy"] == 2.5
assert body["top_n"] == 30
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_router.py -v`
Expected: FAIL (router 없음).
- [ ] **Step 3: router.py — /nodes + /settings 구현**
Create `web-backend/stock-lab/app/screener/router.py`:
```python
"""FastAPI router for /api/stock/screener/*"""
from __future__ import annotations
import datetime as dt
import json
import os
import sqlite3
from typing import Optional
from fastapi import APIRouter, HTTPException
from . import schemas
from .registry import NODE_REGISTRY, GATE_REGISTRY
router = APIRouter(prefix="/api/stock/screener")
def _db_path() -> str:
return os.environ.get("STOCK_DB_PATH", "/data/stock.db")
def _conn() -> sqlite3.Connection:
return sqlite3.connect(_db_path())
# ---------- /nodes ----------
@router.get("/nodes", response_model=schemas.NodesResponse)
def get_nodes():
score_nodes = [
schemas.NodeMeta(
name=cls.name, label=cls.label,
default_params=cls.default_params, param_schema=cls.param_schema,
)
for cls in NODE_REGISTRY.values()
]
gate_nodes = [
schemas.NodeMeta(
name=cls.name, label=cls.label,
default_params=cls.default_params, param_schema=cls.param_schema,
)
for cls in GATE_REGISTRY.values()
]
return schemas.NodesResponse(score_nodes=score_nodes, gate_nodes=gate_nodes)
# ---------- /settings ----------
@router.get("/settings", response_model=schemas.SettingsResponse)
def get_settings():
with _conn() as c:
row = c.execute(
"SELECT weights_json, node_params_json, gate_params_json, "
"top_n, rr_ratio, atr_window, atr_stop_mult, updated_at "
"FROM screener_settings WHERE id=1"
).fetchone()
if row is None:
raise HTTPException(503, "settings not initialized")
return schemas.SettingsResponse(
weights=json.loads(row[0]),
node_params=json.loads(row[1]),
gate_params=json.loads(row[2]),
top_n=row[3], rr_ratio=row[4], atr_window=row[5], atr_stop_mult=row[6],
updated_at=row[7],
)
@router.put("/settings", response_model=schemas.SettingsResponse)
def put_settings(body: schemas.SettingsBody):
now = dt.datetime.utcnow().isoformat()
with _conn() as c:
c.execute(
"""UPDATE screener_settings SET
weights_json=?, node_params_json=?, gate_params_json=?,
top_n=?, rr_ratio=?, atr_window=?, atr_stop_mult=?, updated_at=?
WHERE id=1""",
(
json.dumps(body.weights), json.dumps(body.node_params),
json.dumps(body.gate_params),
body.top_n, body.rr_ratio, body.atr_window, body.atr_stop_mult, now,
),
)
c.commit()
return schemas.SettingsResponse(**body.model_dump(), updated_at=now)
```
- [ ] **Step 4: app/main.py 에 router 등록**
Read `web-backend/stock-lab/app/main.py` 파일 확인. FastAPI 앱이 `app = FastAPI(...)`로 만들어지는 위치에 다음 라인 추가:
```python
from app.screener.router import router as screener_router
app.include_router(screener_router)
```
- [ ] **Step 5: Run test, expect pass**
Run: `pytest app/test_screener_router.py -v`
Expected: 3 passed.
- [ ] **Step 6: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/router.py stock-lab/app/main.py stock-lab/app/test_screener_router.py
git commit -m "feat(stock-lab): /nodes + /settings 라우터 + main.py include"
```
---
## Task 3.3: /run 엔드포인트
**Files:**
- Modify: `web-backend/stock-lab/app/screener/router.py` (append /run)
- Modify: `web-backend/stock-lab/app/test_screener_router.py` (append tests)
- [ ] **Step 1: 테스트 추가**
Append to `web-backend/stock-lab/app/test_screener_router.py`:
```python
from app.screener._test_fixtures import make_master, make_prices, make_flow
def _seed_min(conn, asof_iso="2026-05-12"):
import datetime as dt
now = dt.datetime.utcnow().isoformat()
# 시총 큰 종목 2개 + 작은 종목 1개
rows = [
("BIG1", "큰주식1", "KOSPI", 200_000_000_000, 0, 0, 0, None, now),
("BIG2", "큰주식2", "KOSPI", 100_000_000_000, 0, 0, 0, None, now),
("SMALL", "작은주식", "KOSPI", 1_000_000_000, 0, 0, 0, None, now),
]
for r in rows:
conn.execute("""INSERT INTO krx_master (ticker,name,market,market_cap,
is_managed,is_preferred,is_spac,listed_date,updated_at)
VALUES (?,?,?,?,?,?,?,?,?)""", r)
asof = dt.date(2026, 5, 12)
p = make_prices(["BIG1", "BIG2", "SMALL"], days=260, asof=asof)
f = make_flow(["BIG1", "BIG2", "SMALL"], days=260, asof=asof,
foreign_per_day={"BIG1": 100_000_000, "BIG2": 50_000_000, "SMALL": 0})
p.to_sql("krx_daily_prices", conn, if_exists="append", index=False)
f.to_sql("krx_flow", conn, if_exists="append", index=False)
conn.commit()
def test_run_preview_no_save(client, tmp_path):
db_path = os.environ["STOCK_DB_PATH"]
c = sqlite3.connect(db_path)
_seed_min(c)
c.close()
r = client.post("/api/stock/screener/run", json={"mode": "preview", "asof": "2026-05-12"})
assert r.status_code == 200
body = r.json()
assert body["status"] == "success"
assert body["run_id"] is None
assert body["telegram_payload"] is not None
c = sqlite3.connect(db_path)
cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0]
assert cnt == 0
def test_run_manual_save_writes_row(client):
db_path = os.environ["STOCK_DB_PATH"]
c = sqlite3.connect(db_path)
_seed_min(c)
c.close()
r = client.post("/api/stock/screener/run",
json={"mode": "manual_save", "asof": "2026-05-12"})
assert r.status_code == 200
assert r.json()["run_id"] is not None
c = sqlite3.connect(db_path)
cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0]
assert cnt == 1
def test_run_holiday_returns_skipped():
"""공휴일 처리 — TODO Phase 6에서 holidays.json 통합 시 활성화."""
pytest.skip("holidays.json integration is part of Task 6.x")
```
- [ ] **Step 2: Run test, expect failure**
Run: `pytest app/test_screener_router.py -k run -v`
Expected: FAIL — /run 없음.
- [ ] **Step 3: /run 구현**
Append to `web-backend/stock-lab/app/screener/router.py`:
```python
# ---------- /run ----------
from . import telegram as _tg
from .engine import Screener, ScreenContext
from .registry import NODE_REGISTRY, GATE_REGISTRY
def _resolve_asof(asof_str: Optional[str], conn: sqlite3.Connection) -> dt.date:
if asof_str:
return dt.date.fromisoformat(asof_str)
row = conn.execute("SELECT max(date) FROM krx_daily_prices").fetchone()
if not row or row[0] is None:
raise HTTPException(503, "no snapshot available — run /snapshot/refresh first")
return dt.date.fromisoformat(row[0])
def _load_settings(conn) -> dict:
row = conn.execute(
"SELECT weights_json,node_params_json,gate_params_json,top_n,"
"rr_ratio,atr_window,atr_stop_mult FROM screener_settings WHERE id=1"
).fetchone()
return {
"weights": json.loads(row[0]),
"node_params": json.loads(row[1]),
"gate_params": json.loads(row[2]),
"top_n": row[3],
"rr_ratio": row[4],
"atr_window": row[5],
"atr_stop_mult": row[6],
}
def _persist_run(conn, asof, mode, weights, node_params, gate_params, top_n,
result, started_at, finished_at) -> int:
cur = conn.execute(
"""INSERT INTO screener_runs (asof,mode,status,started_at,finished_at,
weights_json,node_params_json,gate_params_json,top_n,survivors_count,telegram_sent)
VALUES (?,?,?,?,?,?,?,?,?,?,0)""",
(asof.isoformat(), mode, "success", started_at, finished_at,
json.dumps(weights), json.dumps(node_params), json.dumps(gate_params),
top_n, result.survivors_count),
)
run_id = cur.lastrowid
for row in result.rows:
conn.execute(
"""INSERT INTO screener_results (run_id,rank,ticker,name,total_score,
scores_json,close,market_cap,entry_price,stop_price,target_price,atr14)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
(run_id, row["rank"], row["ticker"], row["name"], row["total_score"],
json.dumps(row["scores"]), row["close"], row["market_cap"],
row["entry_price"], row["stop_price"], row["target_price"], row["atr14"]),
)
conn.commit()
return run_id
@router.post("/run", response_model=schemas.RunResponse)
def post_run(body: schemas.RunRequest):
started_at = dt.datetime.utcnow().isoformat()
with _conn() as c:
try:
asof = _resolve_asof(body.asof, c)
except HTTPException:
raise
defaults = _load_settings(c)
if body.mode == "auto":
weights = defaults["weights"]
node_params = defaults["node_params"]
gate_params = defaults["gate_params"]
top_n = defaults["top_n"]
else:
weights = body.weights if body.weights is not None else defaults["weights"]
node_params = body.node_params if body.node_params is not None else defaults["node_params"]
gate_params = body.gate_params if body.gate_params is not None else defaults["gate_params"]
top_n = body.top_n if body.top_n is not None else defaults["top_n"]
sizer_params = {
"atr_window": defaults["atr_window"],
"atr_stop_mult": defaults["atr_stop_mult"],
"rr_ratio": defaults["rr_ratio"],
}
ctx = ScreenContext.load(c, asof)
score_nodes = [cls() for name, cls in NODE_REGISTRY.items() if weights.get(name, 0) > 0]
gate = GATE_REGISTRY["hygiene"]()
try:
screener = Screener(
gate=gate, score_nodes=score_nodes, weights=weights,
node_params=node_params, gate_params=gate_params,
top_n=top_n, sizer_params=sizer_params,
)
result = screener.run(ctx)
except ValueError as e:
raise HTTPException(422, str(e))
finished_at = dt.datetime.utcnow().isoformat()
run_id = None
if body.mode in ("manual_save", "auto"):
run_id = _persist_run(c, asof, body.mode, weights, node_params, gate_params,
top_n, result, started_at, finished_at)
payload = _tg.build_telegram_payload(
asof=asof, mode=body.mode, survivors_count=result.survivors_count,
top_n=top_n, rows=result.rows, run_id=run_id,
)
return schemas.RunResponse(
asof=asof.isoformat(), mode=body.mode, status="success",
run_id=run_id, survivors_count=result.survivors_count,
weights=weights, top_n=top_n,
results=result.rows,
telegram_payload=schemas.TelegramPayload(**payload),
warnings=result.warnings,
)
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_router.py -k run -v`
Expected: 2 passed (third = skipped).
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py
git commit -m "feat(stock-lab): /run 엔드포인트 — preview/manual_save/auto 모드 매트릭스"
```
---
## Task 3.4: /snapshot/refresh + /runs
**Files:**
- Modify: `web-backend/stock-lab/app/screener/router.py`
- [ ] **Step 1: 라우터 추가**
Append to `web-backend/stock-lab/app/screener/router.py`:
```python
# ---------- /snapshot/refresh ----------
from . import snapshot as _snap
@router.post("/snapshot/refresh")
def post_snapshot_refresh(asof: Optional[str] = None):
asof_date = dt.date.fromisoformat(asof) if asof else dt.date.today()
if asof_date.weekday() >= 5:
return {"asof": asof_date.isoformat(), "status": "skipped_weekend"}
with _conn() as c:
summary = _snap.refresh_daily(c, asof_date)
return summary
# ---------- /runs ----------
@router.get("/runs", response_model=list[schemas.RunSummary])
def list_runs(limit: int = 30):
with _conn() as c:
rows = c.execute(
"SELECT id,asof,mode,status,started_at,finished_at,top_n,"
"survivors_count,telegram_sent FROM screener_runs "
"ORDER BY asof DESC, id DESC LIMIT ?", (limit,),
).fetchall()
return [
schemas.RunSummary(
id=r[0], asof=r[1], mode=r[2], status=r[3],
started_at=r[4], finished_at=r[5], top_n=r[6],
survivors_count=r[7], telegram_sent=bool(r[8]),
)
for r in rows
]
@router.get("/runs/{run_id}")
def get_run(run_id: int):
with _conn() as c:
meta = c.execute(
"SELECT id,asof,mode,status,started_at,finished_at,top_n,"
"survivors_count,telegram_sent,weights_json,node_params_json,gate_params_json "
"FROM screener_runs WHERE id=?",
(run_id,),
).fetchone()
if not meta:
raise HTTPException(404, "run not found")
rows = c.execute(
"SELECT rank,ticker,name,total_score,scores_json,close,market_cap,"
"entry_price,stop_price,target_price,atr14 "
"FROM screener_results WHERE run_id=? ORDER BY rank",
(run_id,),
).fetchall()
return {
"meta": {
"id": meta[0], "asof": meta[1], "mode": meta[2], "status": meta[3],
"started_at": meta[4], "finished_at": meta[5], "top_n": meta[6],
"survivors_count": meta[7], "telegram_sent": bool(meta[8]),
"weights": json.loads(meta[9]),
"node_params": json.loads(meta[10]),
"gate_params": json.loads(meta[11]),
},
"results": [
{
"rank": r[0], "ticker": r[1], "name": r[2],
"total_score": r[3], "scores": json.loads(r[4]),
"close": r[5], "market_cap": r[6],
"entry_price": r[7], "stop_price": r[8], "target_price": r[9],
"atr14": r[10],
}
for r in rows
],
}
```
- [ ] **Step 2: 테스트 추가**
Append to `web-backend/stock-lab/app/test_screener_router.py`:
```python
def test_runs_list_and_detail(client):
db_path = os.environ["STOCK_DB_PATH"]
c = sqlite3.connect(db_path)
_seed_min(c)
c.close()
saved = client.post(
"/api/stock/screener/run",
json={"mode": "manual_save", "asof": "2026-05-12"},
).json()
run_id = saved["run_id"]
list_r = client.get("/api/stock/screener/runs?limit=5")
assert list_r.status_code == 200
assert any(r["id"] == run_id for r in list_r.json())
detail = client.get(f"/api/stock/screener/runs/{run_id}")
assert detail.status_code == 200
assert detail.json()["meta"]["id"] == run_id
assert isinstance(detail.json()["results"], list)
```
- [ ] **Step 3: Run test, expect pass**
Run: `pytest app/test_screener_router.py -v`
Expected: 모든 테스트 pass.
- [ ] **Step 4: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py
git commit -m "feat(stock-lab): /snapshot/refresh + /runs 리스트·상세 라우터"
```
---
## Task 3.5: 공휴일 처리 + asof 자동 추정 + skipped_holiday
**Files:**
- Modify: `web-backend/stock-lab/app/screener/router.py`
- Modify: `web-backend/stock-lab/app/test_screener_router.py` (holiday 테스트 활성화)
- [ ] **Step 1: holidays.json 로딩 헬퍼 추가**
Append to `web-backend/stock-lab/app/screener/router.py` (파일 상단의 imports 다음에):
```python
import json as _json
import pathlib as _pathlib
_HOLIDAYS_CACHE: set[str] | None = None
def _holidays() -> set[str]:
global _HOLIDAYS_CACHE
if _HOLIDAYS_CACHE is None:
path = _pathlib.Path(__file__).resolve().parent.parent / "holidays.json"
try:
with path.open(encoding="utf-8") as f:
data = _json.load(f)
_HOLIDAYS_CACHE = set(data) if isinstance(data, list) else set(data.keys())
except FileNotFoundError:
_HOLIDAYS_CACHE = set()
return _HOLIDAYS_CACHE
def _is_holiday(d: dt.date) -> bool:
return d.weekday() >= 5 or d.isoformat() in _holidays()
```
- [ ] **Step 2: post_run에 holiday 분기 추가**
`post_run` 함수의 `started_at = dt.datetime.utcnow().isoformat()` 다음, `with _conn() as c:` 진입 직후:
```python
try:
asof = _resolve_asof(body.asof, c)
except HTTPException:
raise
# Skipped holiday handling for mode='auto'
if body.mode == "auto" and _is_holiday(asof):
return schemas.RunResponse(
asof=asof.isoformat(), mode="auto", status="skipped_holiday",
run_id=None, survivors_count=None,
weights={}, top_n=0,
results=[], telegram_payload=None,
warnings=[f"{asof.isoformat()} is a holiday — skipped"],
)
```
- [ ] **Step 3: 테스트 활성화**
Replace the skipped test in `app/test_screener_router.py`:
```python
def test_run_holiday_returns_skipped(client):
# 토요일
r = client.post("/api/stock/screener/run",
json={"mode": "auto", "asof": "2026-05-09"}) # 2026-05-09 is Saturday
assert r.status_code == 200
assert r.json()["status"] == "skipped_holiday"
```
- [ ] **Step 4: Run test, expect pass**
Run: `pytest app/test_screener_router.py -v`
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py
git commit -m "feat(stock-lab): /run mode=auto 공휴일·주말 skipped_holiday 처리"
```
---
# Phase 4 — 프론트엔드 (web-ui)
**Phase 전체에서 commit은 `web-ui` 디렉토리에서 수행.**
## Task 4.1: api.js 헬퍼 7개 추가
**Files:**
- Modify: `web-ui/src/api.js`
- [ ] **Step 1: 헬퍼 추가**
Read `web-ui/src/api.js` 끝부분에 다음 7개 헬퍼 추가:
```javascript
// ---- Stock Screener ----
export const getScreenerNodes = () => apiGet ('/api/stock/screener/nodes');
export const getScreenerSettings = () => apiGet ('/api/stock/screener/settings');
export const saveScreenerSettings = (body) => apiPut ('/api/stock/screener/settings', body);
export const runScreener = (body) => apiPost('/api/stock/screener/run', body);
export const refreshScreenerSnap = () => apiPost('/api/stock/screener/snapshot/refresh');
export const listScreenerRuns = (limit = 30) => apiGet (`/api/stock/screener/runs?limit=${limit}`);
export const getScreenerRun = (id) => apiGet (`/api/stock/screener/runs/${id}`);
```
- [ ] **Step 2: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/api.js
git commit -m "feat(stock): screener API 헬퍼 7개"
```
---
## Task 4.2: 라우트 + 네비게이션 등록
**Files:**
- Modify: `web-ui/src/routes.jsx`
- Modify: `web-ui/src/Router.jsx`
- [ ] **Step 1: routes.jsx 수정**
Read `web-ui/src/routes.jsx` 검토 후 라우트 배열에 추가 (기존 `/stock` 항목 근처):
```jsx
{ path: '/stock/screener', label: '스크리너', element: lazy(() => import('./pages/stock/screener/Screener')) },
```
(파일의 정확한 패턴은 기존 항목을 따라가세요. lazy import 사용 여부 등.)
- [ ] **Step 2: Router.jsx 수정**
Read `web-ui/src/Router.jsx` 검토. 라우트 추가 패턴이 별도로 있으면 그곳에 동일하게 `/stock/screener` 등록.
- [ ] **Step 3: 빌드 확인**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run dev
```
브라우저에서 `/stock/screener` 접근. (페이지는 아직 비어 있어도 무방, 라우트만 매칭 확인)
- [ ] **Step 4: Commit (페이지 빈 컴포넌트 임시 작성 후)**
임시 placeholder 페이지를 위해 4.3을 먼저 작성하고 함께 commit. **이 단계는 4.3 commit에 통합.**
---
## Task 4.3: Screener.jsx 페이지 골격 + 빈 컴포넌트
**Files:**
- Create: `web-ui/src/pages/stock/screener/Screener.jsx`
- Create: `web-ui/src/pages/stock/screener/Screener.css`
- [ ] **Step 1: 페이지 골격**
Create `web-ui/src/pages/stock/screener/Screener.jsx`:
```jsx
import React from 'react';
import { Link } from 'react-router-dom';
import './Screener.css';
import { useScreenerMeta } from './hooks/useScreenerMeta';
import { useScreenerSettings } from './hooks/useScreenerSettings';
import { useScreenerRun } from './hooks/useScreenerRun';
import { useScreenerHistory } from './hooks/useScreenerHistory';
import GatePanel from './components/GatePanel';
import NodePanel from './components/NodePanel';
import GlobalControls from './components/GlobalControls';
import ResultTable from './components/ResultTable';
import TelegramPreview from './components/TelegramPreview';
import RunHistoryList from './components/RunHistoryList';
export default function Screener() {
const { meta, loading: metaLoading } = useScreenerMeta();
const { settings, dirty, setLocal, save } = useScreenerSettings();
const { result, running, runPreview, runSave } = useScreenerRun();
const { runs, runs_loading, selectRun, selectedRun } = useScreenerHistory();
const activeResult = selectedRun || result;
if (metaLoading || !meta || !settings) {
return 로딩 중…
;
}
return (
);
}
```
- [ ] **Step 2: Screener.css 골격**
Create `web-ui/src/pages/stock/screener/Screener.css`:
```css
.screener-page {
padding: 24px;
color: var(--text, #e5e7eb);
background: var(--bg, #0b0f17);
min-height: 100vh;
}
.screener-header {
display: flex;
align-items: flex-end;
justify-content: space-between;
margin-bottom: 24px;
}
.screener-header h1 {
font-size: 28px;
margin: 0 0 4px 0;
}
.screener-header .meta {
color: #9ca3af;
font-size: 13px;
margin: 0;
}
.screener-header nav a {
margin-left: 12px;
color: #9ca3af;
text-decoration: none;
}
.screener-grid {
display: grid;
grid-template-columns: 320px 1fr 280px;
gap: 24px;
}
@media (max-width: 1023px) {
.screener-grid { grid-template-columns: 1fr; }
}
.screener-loading { padding: 80px; text-align: center; color: #9ca3af; }
```
- [ ] **Step 3: 빈 hooks 4개 stub 작성 (다음 task에서 채움)**
Create `web-ui/src/pages/stock/screener/hooks/useScreenerMeta.js`:
```javascript
import { useEffect, useState } from 'react';
import { getScreenerNodes } from '../../../../api';
export function useScreenerMeta() {
const [meta, setMeta] = useState(null);
const [loading, setLoading] = useState(true);
useEffect(() => {
getScreenerNodes().then((m) => { setMeta(m); setLoading(false); });
}, []);
return { meta, loading };
}
```
Create `useScreenerSettings.js`:
```javascript
import { useEffect, useState } from 'react';
import { getScreenerSettings, saveScreenerSettings } from '../../../../api';
export function useScreenerSettings() {
const [remote, setRemote] = useState(null);
const [local, setLocal] = useState(null);
useEffect(() => {
getScreenerSettings().then((s) => { setRemote(s); setLocal(s); });
}, []);
const dirty = remote && local && JSON.stringify(remote) !== JSON.stringify(local);
async function save() {
if (!local) return;
const saved = await saveScreenerSettings({
weights: local.weights, node_params: local.node_params, gate_params: local.gate_params,
top_n: local.top_n, rr_ratio: local.rr_ratio,
atr_window: local.atr_window, atr_stop_mult: local.atr_stop_mult,
});
setRemote(saved);
setLocal(saved);
}
return { settings: local, dirty, setLocal, save };
}
```
Create `useScreenerRun.js`:
```javascript
import { useState } from 'react';
import { runScreener } from '../../../../api';
export function useScreenerRun() {
const [result, setResult] = useState(null);
const [running, setRunning] = useState(false);
async function call(mode, settings) {
setRunning(true);
try {
const body = {
mode,
weights: settings.weights,
node_params: settings.node_params,
gate_params: settings.gate_params,
top_n: settings.top_n,
};
const r = await runScreener(body);
setResult(r);
return r;
} finally {
setRunning(false);
}
}
return {
result, running,
runPreview: (s) => call('preview', s),
runSave: (s) => call('manual_save', s),
};
}
```
Create `useScreenerHistory.js`:
```javascript
import { useEffect, useState } from 'react';
import { listScreenerRuns, getScreenerRun } from '../../../../api';
export function useScreenerHistory() {
const [runs, setRuns] = useState([]);
const [loading, setLoading] = useState(true);
const [selectedRun, setSelectedRun] = useState(null);
useEffect(() => {
listScreenerRuns(30).then((r) => { setRuns(r); setLoading(false); });
}, []);
async function selectRun(id) {
if (!id) { setSelectedRun(null); return; }
const detail = await getScreenerRun(id);
setSelectedRun({
asof: detail.meta.asof,
mode: detail.meta.mode,
status: detail.meta.status,
run_id: detail.meta.id,
survivors_count: detail.meta.survivors_count,
weights: detail.meta.weights,
top_n: detail.meta.top_n,
results: detail.results,
telegram_payload: null, // history detail은 메시지 미리보기 비포함 (MVP)
warnings: [],
meta: detail.meta,
});
}
return { runs, runs_loading: loading, selectedRun, selectRun };
}
```
- [ ] **Step 4: 빈 컴포넌트 5개 stub**
Create `web-ui/src/pages/stock/screener/components/GatePanel.jsx`:
```jsx
export default function GatePanel({ meta, value, onChange }) {
return {meta?.label ?? '게이트'}
TODO: 게이트 파라미터 폼
;
}
```
Create `NodePanel.jsx`:
```jsx
export default function NodePanel({ meta, weights, params, onWeights, onParams }) {
return 점수 노드 ({meta.length})
TODO: 노드별 카드
;
}
```
Create `GlobalControls.jsx`:
```jsx
export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
return (
);
}
```
Create `ResultTable.jsx`:
```jsx
export default function ResultTable({ result }) {
if (!result) return 아직 결과 없음. "지금 실행"을 눌러보세요.
;
return (
Top {result.top_n} · 통과 {result.survivors_count}
| # | 종목 | 총점 | 진입 | 손절 | 익절 | R% |
|---|
{(result.results || []).map((r) => (
| {r.rank} | {r.name} ({r.ticker}) |
{r.total_score?.toFixed?.(1)} |
{r.entry_price?.toLocaleString?.()} |
{r.stop_price?.toLocaleString?.()} |
{r.target_price?.toLocaleString?.()} |
{r.r_pct?.toFixed?.(1)} |
))}
);
}
```
Create `TelegramPreview.jsx`:
```jsx
export default function TelegramPreview({ payload }) {
if (!payload) return null;
return (
);
}
```
Create `RunHistoryList.jsx`:
```jsx
export default function RunHistoryList({ runs, loading, onSelect, selectedId }) {
if (loading) return ;
return (
최근 실행
{(runs || []).map((r) => (
- onSelect(r.id)}>
{r.asof} · {r.mode}
))}
);
}
```
추가 css to Screener.css:
```css
.screener-card {
background: #0f1623;
border: 1px solid #1f2937;
border-radius: 8px;
padding: 16px;
margin-bottom: 16px;
}
.screener-card h3 { margin: 0 0 12px 0; font-size: 15px; }
```
- [ ] **Step 5: 빌드 확인**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run dev
```
브라우저: `/stock/screener` 접속, 페이지 로드 확인. (백엔드 미동작이어도 hook 호출 실패 메시지 정도가 보임)
- [ ] **Step 6: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/pages/stock/screener src/routes.jsx src/Router.jsx
git commit -m "feat(stock): /stock/screener 페이지 골격 + hooks 4개 + 컴포넌트 stub 6개"
```
---
## Task 4.4: NodeCard — param_schema 기반 자동 폼
**Files:**
- Create: `web-ui/src/pages/stock/screener/components/NodeCard.jsx`
- Modify: `web-ui/src/pages/stock/screener/components/NodePanel.jsx`
- [ ] **Step 1: NodeCard 작성**
Create `web-ui/src/pages/stock/screener/components/NodeCard.jsx`:
```jsx
import React from 'react';
export default function NodeCard({ meta, weight, params, onWeightChange, onParamsChange }) {
const enabled = (weight ?? 0) > 0;
return (
);
}
function ParamRow({ paramKey, prop, value, disabled, onChange }) {
const type = prop.type;
if (type === 'integer' || type === 'number') {
return (
{paramKey}
onChange(type === 'integer' ? parseInt(e.target.value, 10) : parseFloat(e.target.value))}
style={{ width: 80 }}
/>
);
}
if (type === 'boolean') {
return (
);
}
// object/array는 MVP에서 read-only JSON 표시 (RsRating의 weights 등)
return (
{paramKey}: {JSON.stringify(value)}
);
}
```
- [ ] **Step 2: NodePanel 갱신**
Replace `web-ui/src/pages/stock/screener/components/NodePanel.jsx`:
```jsx
import NodeCard from './NodeCard';
export default function NodePanel({ meta, weights, params, onWeights, onParams }) {
return (
점수 노드 ({meta.length})
{meta.map((m) => (
onWeights({ ...weights, [m.name]: w })}
onParamsChange={(p) => onParams({ ...params, [m.name]: p })}
/>
))}
);
}
```
- [ ] **Step 3: CSS 추가**
Append to `Screener.css`:
```css
.node-card {
background: #0a0f1a;
border: 1px solid #1f2937;
border-radius: 6px;
padding: 10px;
font-size: 13px;
}
.node-card-header { font-weight: 500; margin-bottom: 6px; }
.weight-row, .param-row { display: flex; align-items: center; gap: 6px; margin-top: 6px; }
```
- [ ] **Step 4: 빌드 확인**
`npm run dev`로 노드 7개 카드가 슬라이더·체크박스·숫자 입력으로 렌더링되는지 확인.
- [ ] **Step 5: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/pages/stock/screener/components/NodeCard.jsx src/pages/stock/screener/components/NodePanel.jsx src/pages/stock/screener/Screener.css
git commit -m "feat(stock): NodeCard 자동 폼 (param_schema 기반) + NodePanel 통합"
```
---
## Task 4.5: GatePanel + GlobalControls 본구현
**Files:**
- Modify: `web-ui/src/pages/stock/screener/components/GatePanel.jsx`
- Modify: `web-ui/src/pages/stock/screener/components/GlobalControls.jsx`
- [ ] **Step 1: GatePanel 갱신**
Replace `GatePanel.jsx`:
```jsx
export default function GatePanel({ meta, value, onChange }) {
if (!meta) return null;
const props = meta.param_schema?.properties || {};
return (
{meta.label}
통과 조건 — 통과한 종목만 점수 노드에 전달
{Object.entries(props).map(([key, prop]) => (
onChange({ ...value, [key]: v })} />
))}
);
}
function GateField({ paramKey, prop, value, onChange }) {
if (prop.type === 'integer') {
return (
onChange(parseInt(e.target.value, 10))}
style={{ flex: 1 }} />
);
}
if (prop.type === 'boolean') {
return (
);
}
return null;
}
```
- [ ] **Step 2: GlobalControls 갱신**
Replace `GlobalControls.jsx`:
```jsx
export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
return (
);
}
```
- [ ] **Step 3: 빌드 확인**
`npm run dev`로 게이트 폼·글로벌 옵션 렌더링 확인.
- [ ] **Step 4: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/pages/stock/screener/components/GatePanel.jsx src/pages/stock/screener/components/GlobalControls.jsx
git commit -m "feat(stock): GatePanel 자동 폼 + GlobalControls (TopN/ATR/RR + 3버튼)"
```
---
## Task 4.6: ResultTable 본구현 (ScoreChips)
**Files:**
- Create: `web-ui/src/pages/stock/screener/components/ScoreChips.jsx`
- Modify: `web-ui/src/pages/stock/screener/components/ResultTable.jsx`
- [ ] **Step 1: ScoreChips**
Create `ScoreChips.jsx`:
```jsx
const NODE_ICONS = {
foreign_buy: { icon: '👤', label: '외국인' },
volume_surge: { icon: '⚡', label: '거래량' },
momentum: { icon: '🚀', label: '모멘텀' },
high52w: { icon: '🆙', label: '52w고' },
rs_rating: { icon: '💪', label: 'RS' },
ma_alignment: { icon: '📈', label: '정배열' },
vcp_lite: { icon: '🌀', label: 'VCP' },
};
export default function ScoreChips({ scores }) {
return (
{Object.entries(scores || {}).map(([name, s]) => {
const meta = NODE_ICONS[name];
if (!meta) return null;
const active = s >= 70;
return (
{meta.icon}{Math.round(s)}
);
})}
);
}
```
- [ ] **Step 2: ResultTable 갱신**
Replace `ResultTable.jsx`:
```jsx
import ScoreChips from './ScoreChips';
export default function ResultTable({ result }) {
if (!result) {
return (
아직 결과 없음. "지금 실행"을 눌러보세요.
);
}
if (result.warnings?.length) {
// 경고 배너만 노출하고 표는 계속 렌더
}
return (
Top {result.top_n} · 통과 {result.survivors_count} · {result.asof}
{result.warnings?.length > 0 && (
⚠ {result.warnings.join(' · ')}
)}
| # | 종목 | 총점 | 노드 |
진입 | 손절 | 익절 | R% |
{(result.results || []).map((r) => (
| {r.rank} |
{r.name}
{r.ticker} |
{r.total_score?.toFixed(1)} |
|
{r.entry_price?.toLocaleString?.()} |
{r.stop_price?.toLocaleString?.()} |
{r.target_price?.toLocaleString?.()} |
{r.r_pct?.toFixed?.(1)} |
))}
);
}
```
- [ ] **Step 3: CSS 추가**
Append to `Screener.css`:
```css
.screener-table {
width: 100%;
font-size: 13px;
border-collapse: collapse;
}
.screener-table th { text-align: left; padding: 8px; background: #0a0f1a; color: #9ca3af; font-weight: 500; border-bottom: 1px solid #1f2937; }
.screener-table td { padding: 8px; border-bottom: 1px solid #1a2230; vertical-align: middle; }
.screener-table tr:hover { background: #0a0f1a; }
```
- [ ] **Step 4: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/pages/stock/screener/components/ScoreChips.jsx src/pages/stock/screener/components/ResultTable.jsx src/pages/stock/screener/Screener.css
git commit -m "feat(stock): ResultTable 본구현 + ScoreChips (노드 칩 + 70점 강조)"
```
---
## Task 4.7: CLAUDE.md API 표 갱신 + Stock 페이지에 스크리너 링크
**Files:**
- Modify: `web-ui/CLAUDE.md`
- Modify: `web-ui/src/pages/stock/Stock.jsx` (또는 StockTrade.jsx — 스크리너 링크 위치)
- [ ] **Step 1: CLAUDE.md API 표에 7행 추가**
Read `web-ui/CLAUDE.md`의 "API 엔드포인트 목록" 표를 찾고, 마지막에 다음 7행 추가:
```markdown
| 스크리너 | GET | `/api/stock/screener/nodes` |
| 스크리너 | GET/PUT | `/api/stock/screener/settings` |
| 스크리너 | POST | `/api/stock/screener/run` — body: `{ mode, asof?, weights?, ... }` |
| 스크리너 | POST | `/api/stock/screener/snapshot/refresh` |
| 스크리너 | GET | `/api/stock/screener/runs?limit=N` |
| 스크리너 | GET | `/api/stock/screener/runs/:id` |
```
또한 "페이지 구조" 표에 `/stock/screener` 행 추가:
```markdown
| `/stock/screener` | `Screener` | 노드 기반 강세주 스크리너 (점수 노드 7 + 위생 게이트 + ATR 포지션 사이저) |
```
- [ ] **Step 2: Stock 페이지에 스크리너 링크**
Read `web-ui/src/pages/stock/Stock.jsx` 상단 헤더 또는 nav 영역에 다음 링크 추가 (기존 `` 패턴 옆에):
```jsx
스크리너
```
- [ ] **Step 3: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add CLAUDE.md src/pages/stock/Stock.jsx
git commit -m "docs(stock): CLAUDE.md 스크리너 API 표 추가 + Stock 페이지 링크"
```
---
## Task 4.8: 모바일 레이아웃 점검 + 작은 화면 적층
**Files:**
- Modify: `web-ui/src/pages/stock/screener/Screener.css`
- [ ] **Step 1: 모바일 CSS 추가**
Append to `Screener.css`:
```css
@media (max-width: 1023px) {
.screener-page { padding: 16px; }
.screener-header { flex-direction: column; align-items: flex-start; gap: 8px; }
.screener-grid { gap: 16px; }
.screener-left { order: 1; }
.screener-center { order: 2; }
.screener-right { order: 3; }
.screener-table { font-size: 12px; }
.screener-table th, .screener-table td { padding: 6px 4px; }
}
@media (max-width: 640px) {
.screener-page { padding: 12px; }
.screener-card { padding: 12px; }
}
```
- [ ] **Step 2: 브라우저 모바일 폭(<768px) 확인**
`npm run dev`로 Chrome DevTools에서 iPhone 사이즈 토글 → 카드 세로 적층, 표 가로 스크롤 확인.
- [ ] **Step 3: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
git add src/pages/stock/screener/Screener.css
git commit -m "style(stock): 스크리너 모바일 적층 + 표 가로 스크롤"
```
---
## Task 4.9: ESLint + 빌드 검증
**Files:** (변경 없음, 검증만)
- [ ] **Step 1: lint**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run lint
```
오류가 있으면 fix 후 재실행.
- [ ] **Step 2: build**
```bash
npm run build
```
`dist/` 생성 확인.
- [ ] **Step 3: fix가 있었으면 commit**
```bash
git add -A
git commit -m "chore(stock): 스크리너 lint/build 정리"
```
(변경 없으면 step 건너뜀)
---
# Phase 5 — agent-office 자동 잡 통합
## Task 5.1: agent-office에 stock_screener 잡 추가
**Files:** (`web-backend/agent-office/` 구조에 따라 위치 결정 — 기존 realestate-lab/뉴스 알림 잡 패턴을 그대로 따라가세요)
agent-office가 어떤 형태로 잡을 정의하는지(YAML config, Python module, cron entry 등)를 먼저 확인해야 합니다. 가장 가능성 있는 패턴 3가지:
**Patten A — 잡 module 추가**: `web-backend/agent-office/app/jobs/stock_screener.py`
**Pattern B — config 항목 추가**: `web-backend/agent-office/jobs.yaml`에 항목 1개
**Pattern C — agent 명령**: `agent-office` 내부 명령 핸들러에 `stock_agent screener` 추가
이 task는 **반드시 기존 코드를 먼저 읽어 패턴을 확인한 뒤** 그 패턴에 맞춰 잡을 추가하세요.
- [ ] **Step 1: 기존 자동 잡 패턴 식별 (로컬 파일 시스템 탐색)**
로컬 파일을 직접 읽어 패턴 파악 (Grep/Glob 도구 사용 — Docker 불필요):
- `web-backend/agent-office/app/` 디렉토리 트리 확인
- "telegram" 키워드 grep — 어떤 모듈에서 메시지를 보내는지
- "schedule|cron|APScheduler|tick" grep — 잡을 어떻게 등록하는지
- realestate-lab 매칭 알림이 가장 유사한 기존 잡이므로 그 코드를 통째로 읽기
가장 비슷한 기존 잡 1개의 코드를 통째로 읽고, 그 구조를 그대로 따라갑니다.
- [ ] **Step 2: stock_screener 잡 작성**
기존 패턴이 함수 기반(예: `def realestate_match_job(): ...`)이라면 다음을 같은 위치에 추가:
```python
"""Stock screener auto job — 평일 16:30 KST."""
import logging
import os
from datetime import datetime, timezone, timedelta
import httpx
log = logging.getLogger(__name__)
STOCK_LAB_URL = os.environ.get("STOCK_LAB_URL", "http://stock-lab:18500")
async def run_stock_screener_job():
"""Refresh KRX snapshot, run screener auto, dispatch telegram."""
today = datetime.now(timezone(timedelta(hours=9))).date()
asof = today.isoformat()
try:
async with httpx.AsyncClient(timeout=180) as client:
# 1. Snapshot refresh (실패해도 다음 단계 계속)
try:
await client.post(f"{STOCK_LAB_URL}/api/stock/screener/snapshot/refresh",
json={"asof": asof})
except Exception as e:
log.warning("snapshot refresh failed: %s", e)
# 2. Run auto
r = await client.post(
f"{STOCK_LAB_URL}/api/stock/screener/run",
json={"mode": "auto", "asof": asof},
)
r.raise_for_status()
body = r.json()
status = body.get("status")
if status == "skipped_holiday":
log.info("stock_screener skipped (holiday/weekend) %s", asof)
return
if status != "success":
log.error("stock_screener failed: %s", body.get("error"))
_notify_self("stock_screener 잡 실패: " + (body.get("error") or "unknown"))
return
payload = body.get("telegram_payload")
if payload:
# 기존 agent-office 텔레그램 전송 함수 호출 — 이름은 기존 코드 따름
from .telegram import send_message # 예시
await send_message(payload["text"], parse_mode=payload["parse_mode"])
log.info("stock_screener sent telegram for %s", asof)
except Exception as e:
log.exception("stock_screener job crashed")
_notify_self(f"stock_screener 잡 예외: {e}")
def _notify_self(msg: str):
"""운영자에게 알림 — 기존 패턴 차용."""
# TODO: 기존 운영 알림 채널 함수로 연결
log.warning(msg)
```
기존 스케줄러에 등록 (APScheduler·cron entry·기존 main loop에 추가):
- 매일 한국 시간 16:30 트리거
- 평일만 (스케줄러가 weekday 필터 지원하면 거기서, 아니면 잡 내부에서 분기)
- [ ] **Step 3: 환경변수 점검**
`web-backend/agent-office/.env`(또는 docker-compose.yml의 environment 섹션)에서 stock-lab으로의 내부 URL 확인. 보통 도커 네트워크에서 서비스명으로 접근:
```
STOCK_LAB_URL=http://stock-lab:18500
```
- [ ] **Step 4: 단위 테스트 (로컬 venv, 함수 직접 호출)**
가능하면 `run_stock_screener_job`에 대한 단위 테스트를 추가 (httpx mock으로 stock-lab 응답 stub). 기존 agent-office가 venv로 로컬 실행 가능하면 그것으로 실행. 그렇지 않으면 함수 시그니처와 로직 코드 리뷰만 하고 통과.
- [ ] **Step 5: 배포 후 NAS에서 실전 발동 확인 (Task 6.3에서)**
agent-office 컨테이너 재시작·로그 확인은 NAS SSH에서 수행 (Task 6.3 참조). 로컬에서는 코드만 작성하고 commit.
- [ ] **Step 6: Commit**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git add agent-office/
git commit -m "feat(agent-office): stock screener 평일 16:30 KST 자동 잡 + 텔레그램 전송"
```
---
# Phase 6 — 초기 백필 · 수동 검증 · 배포
## Task 6.1: 초기 KRX 데이터 백필 (2년치)
**Files:** (스크립트 1회용 실행, 코드 신규 없음)
- [ ] **Step 1: NAS SSH로 접속 후 백필 (운영 DB 대상)**
```bash
ssh user@gahusb.synology.me
cd /volume1/docker/webpage
sudo docker compose exec stock-lab python -c "
import datetime as dt, sqlite3
from app.screener.snapshot import backfill
start = dt.date(2024, 5, 12)
end = dt.date(2026, 5, 11)
conn = sqlite3.connect('/data/stock.db')
results = backfill(conn, start, end)
ok = sum(1 for r in results if 'error' not in r)
print(f'backfill done: {ok}/{len(results)} days')
"
```
⚠ 시간이 오래 걸립니다 (~30분~수 시간, 2,700종목 × ~500 거래일). NAS 안에서 백그라운드(`nohup`/`screen`) 실행 권장. 끝나기를 기다리고 결과 로그 확인.
- [ ] **Step 2: DB 행 수 확인 (NAS)**
```bash
sudo docker compose exec stock-lab sqlite3 /data/stock.db "
SELECT count(*) AS prices FROM krx_daily_prices;
SELECT count(*) AS flows FROM krx_flow;
SELECT count(*) AS master FROM krx_master;
"
```
prices·flow 둘 다 100만+ 행 정도 나와야 정상.
- [ ] **Step 3: (선택) DB 백업 (NAS)**
```bash
sudo docker compose exec stock-lab sh -c "cp /data/stock.db /data/stock-pre-screener-backup.db"
```
---
## Task 6.2: 수동 검증 — Top 20이 합리적인가
- [ ] **Step 1: 프론트에서 1회 미리보기 실행**
`/stock/screener`로 이동 → "지금 실행 (미리보기)" 클릭. 결과 Top 20 표 확인.
- [ ] **Step 2: 합리성 체크리스트 (사용자 눈으로)**
다음 중 다수가 "예"여야 정상:
- [ ] 상위 종목들이 최근 1~3개월간 외국인 누적 매수 + 거래량 증가 + 우상향 추세인가?
- [ ] 작전주·관리종목·우선주가 포함되어 있지 않은가?
- [ ] 시총이 모두 500억 이상인가?
- [ ] 각 종목의 ATR 기반 손절가가 진입가 대비 -2~-7% 범위인가? (R% 컬럼)
- [ ] 노드 칩(👤외 ⚡거 등)이 종목마다 0~5개 정도 점등되는가? (모두 점등 또는 모두 비점등은 의심)
이상치가 보이면 노드 알고리즘 또는 데이터 캐시 문제 → 개별 디버그.
- [ ] **Step 3: 모바일 화면 확인**
iPhone 사이즈로 좌측 패널 접힘·결과 표 가로 스크롤 OK 확인.
---
## Task 6.3: 자동 잡 실전 발동 & 텔레그램 도착 확인
- [ ] **Step 1: NAS SSH에서 자동 잡 다음 평일 16:30 모니터링**
```bash
ssh user@gahusb.synology.me
cd /volume1/docker/webpage
sudo docker compose logs -f agent-office | grep stock_screener
```
- [ ] **Step 2: 텔레그램 메시지 도착 확인**
- 헤더에 분석 기준일·통과 종목 수 표시되는가
- Top 10이 한 메시지에 다 들어왔는가 (텔레그램 4096자 한도 위반 없음)
- 노드 아이콘이 70점 이상에서만 점등됐는가
- 진입가/손절가/익절가/R% 모두 표시됐는가
- 페이지 링크 클릭 시 해당 `run_id` 결과가 프론트에 로드되는가
- [ ] **Step 3: 운영 상황에서 누적 검증 (1주일)**
매일 텔레그램 도착 + `/stock/screener` 히스토리에 자동 잡 row가 매일 1개씩 추가되는지 확인.
---
## 최종 commit & 배포
- [ ] **Step 1: 프론트 배포**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run release:nas
```
Z 드라이브 마운트 필수 (CLAUDE.md 참조).
- [ ] **Step 2: 백엔드 배포 (이미 docker compose up이면 자동, 아니면 git push로 webhook 트리거)**
```bash
cd C:\Users\jaeoh\Desktop\workspace\web-backend
git push
```
Gitea webhook → NAS deployer가 stock-lab + agent-office 컨테이너 자동 재시작.
- [ ] **Step 3: 운영 사이트에서 1회 확인**
`https://gahusb.synology.me/stock/screener` 접근 → 데이터 정상 로드 확인.
---
# 끝