Files
web-page/docs/superpowers/plans/2026-05-12-stock-screener-board.md
gahusb 3b66a47316 docs(plan): 데이터 소스 pykrx → FDR + 네이버 스크래핑 (Task 0.1/0.3)
실측 결과 pykrx의 시장 전체 함수 (get_market_ticker_list,
get_market_cap, get_market_ohlcv_by_ticker)가 모두 KRX 인증
요구로 깨짐. Task 0.1 의존성을 finance-datareader + bs4 + lxml
로 교체하고 Task 0.3 snapshot.py는 FDR + 네이버 frgn 스크래핑
방식으로 재작성 (implementer dispatch 시 인라인 안내).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-12 04:03:31 +09:00

4354 lines
138 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# 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 <div className="screener-loading">로딩 </div>;
}
return (
<div className="screener-page">
<header className="screener-header">
<div>
<h1>스크리너</h1>
<p className="meta">
최근 자동 : {runs?.find(r => r.mode === 'auto')?.asof ?? '-'}
· 분석 기준일: {activeResult?.asof ?? settings.asof ?? '-'}
</p>
</div>
<nav>
<Link to="/stock">시장</Link>
<Link to="/stock/trade">트레이드</Link>
</nav>
</header>
<div className="screener-grid">
<aside className="screener-left">
<GatePanel meta={meta.gate_nodes[0]} value={settings.gate_params} onChange={(p) => setLocal({...settings, gate_params: p})} />
<NodePanel meta={meta.score_nodes} weights={settings.weights} params={settings.node_params}
onWeights={(w) => setLocal({...settings, weights: w})}
onParams={(p) => setLocal({...settings, node_params: p})} />
<GlobalControls settings={settings} setSettings={setLocal}
onRun={() => runPreview(settings)}
onSave={() => runSave(settings)}
onPersist={save}
dirty={dirty}
running={running} />
</aside>
<main className="screener-center">
<ResultTable result={activeResult} />
<TelegramPreview payload={activeResult?.telegram_payload} />
</main>
<aside className="screener-right">
<RunHistoryList runs={runs} loading={runs_loading} onSelect={selectRun}
selectedId={selectedRun?.meta?.id} />
</aside>
</div>
</div>
);
}
```
- [ ] **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 <section className="screener-card"><h3>{meta?.label ?? '게이트'}</h3><p style={{fontSize: 12, color:'#9ca3af'}}>TODO: 게이트 파라미터 </p></section>;
}
```
Create `NodePanel.jsx`:
```jsx
export default function NodePanel({ meta, weights, params, onWeights, onParams }) {
return <section className="screener-card"><h3>점수 노드 ({meta.length})</h3><p style={{fontSize: 12, color:'#9ca3af'}}>TODO: 노드별 카드</p></section>;
}
```
Create `GlobalControls.jsx`:
```jsx
export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
return (
<section className="screener-card">
<button onClick={onRun} disabled={running}>{running ? '실행 중…' : '지금 실행 (미리보기)'}</button>
<button onClick={onSave} disabled={running} style={{ marginTop: 8 }}>스냅샷 저장</button>
<button onClick={onPersist} disabled={!dirty} style={{ marginTop: 8 }}>설정 저장</button>
</section>
);
}
```
Create `ResultTable.jsx`:
```jsx
export default function ResultTable({ result }) {
if (!result) return <section className="screener-card"><p style={{color:'#9ca3af'}}>아직 결과 없음. "지금 실행" 눌러보세요.</p></section>;
return (
<section className="screener-card">
<h3>Top {result.top_n} · 통과 {result.survivors_count}</h3>
<table style={{ width: '100%', fontSize: 13 }}>
<thead>
<tr><th>#</th><th>종목</th><th>총점</th><th>진입</th><th>손절</th><th>익절</th><th>R%</th></tr>
</thead>
<tbody>
{(result.results || []).map((r) => (
<tr key={r.ticker}>
<td>{r.rank}</td><td>{r.name} ({r.ticker})</td>
<td>{r.total_score?.toFixed?.(1)}</td>
<td>{r.entry_price?.toLocaleString?.()}</td>
<td>{r.stop_price?.toLocaleString?.()}</td>
<td>{r.target_price?.toLocaleString?.()}</td>
<td>{r.r_pct?.toFixed?.(1)}</td>
</tr>
))}
</tbody>
</table>
</section>
);
}
```
Create `TelegramPreview.jsx`:
```jsx
export default function TelegramPreview({ payload }) {
if (!payload) return null;
return (
<section className="screener-card">
<h3>텔레그램 미리보기</h3>
<pre style={{whiteSpace:'pre-wrap', fontFamily:'monospace', fontSize:12}}>{payload.text}</pre>
</section>
);
}
```
Create `RunHistoryList.jsx`:
```jsx
export default function RunHistoryList({ runs, loading, onSelect, selectedId }) {
if (loading) return <section className="screener-card"><p>로딩</p></section>;
return (
<section className="screener-card">
<h3>최근 실행</h3>
<ul style={{listStyle:'none', padding:0, margin:0, fontSize:13}}>
{(runs || []).map((r) => (
<li key={r.id} style={{padding:'6px 0', borderBottom:'1px solid #1f2937', cursor:'pointer',
color: selectedId === r.id ? '#fbbf24' : '#e5e7eb'}}
onClick={() => onSelect(r.id)}>
{r.asof} · {r.mode}
</li>
))}
</ul>
</section>
);
}
```
추가 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 (
<div className="node-card" style={{ opacity: enabled ? 1 : 0.6 }}>
<div className="node-card-header">
<label style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
<input
type="checkbox"
checked={enabled}
onChange={(e) => onWeightChange(e.target.checked ? (weight || 1) : 0)}
/>
<span>{meta.label}</span>
</label>
</div>
<div className="node-card-body">
<div className="weight-row">
<span style={{ width: 50, fontSize: 12, color: '#9ca3af' }}>가중치</span>
<input
type="range" min="0" max="3" step="0.1"
value={weight ?? 0}
disabled={!enabled}
onChange={(e) => onWeightChange(parseFloat(e.target.value))}
style={{ flex: 1 }}
/>
<span style={{ width: 32, textAlign: 'right', fontSize: 12 }}>{(weight ?? 0).toFixed(1)}</span>
</div>
{Object.entries(meta.param_schema?.properties || {}).map(([key, prop]) => (
<ParamRow
key={key}
paramKey={key}
prop={prop}
value={params?.[key] ?? meta.default_params?.[key]}
disabled={!enabled}
onChange={(v) => onParamsChange({ ...params, [key]: v })}
/>
))}
</div>
</div>
);
}
function ParamRow({ paramKey, prop, value, disabled, onChange }) {
const type = prop.type;
if (type === 'integer' || type === 'number') {
return (
<div className="param-row">
<span style={{ width: 100, fontSize: 12 }}>{paramKey}</span>
<input
type="number"
min={prop.minimum} max={prop.maximum}
step={type === 'integer' ? 1 : 0.1}
value={value ?? ''}
disabled={disabled}
onChange={(e) => onChange(type === 'integer' ? parseInt(e.target.value, 10) : parseFloat(e.target.value))}
style={{ width: 80 }}
/>
</div>
);
}
if (type === 'boolean') {
return (
<div className="param-row">
<label>
<input type="checkbox" checked={!!value} disabled={disabled}
onChange={(e) => onChange(e.target.checked)} />
<span style={{ marginLeft: 6, fontSize: 12 }}>{paramKey}</span>
</label>
</div>
);
}
// object/array는 MVP에서 read-only JSON 표시 (RsRating의 weights 등)
return (
<div className="param-row" style={{ fontSize: 11, color: '#9ca3af' }}>
{paramKey}: <code>{JSON.stringify(value)}</code>
</div>
);
}
```
- [ ] **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 (
<section className="screener-card">
<h3>점수 노드 ({meta.length})</h3>
<div style={{ display: 'flex', flexDirection: 'column', gap: 12 }}>
{meta.map((m) => (
<NodeCard
key={m.name}
meta={m}
weight={weights[m.name]}
params={params[m.name]}
onWeightChange={(w) => onWeights({ ...weights, [m.name]: w })}
onParamsChange={(p) => onParams({ ...params, [m.name]: p })}
/>
))}
</div>
</section>
);
}
```
- [ ] **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 (
<section className="screener-card">
<h3>{meta.label}</h3>
<p style={{ fontSize: 11, color: '#9ca3af', marginTop: 0 }}>
통과 조건 통과한 종목만 점수 노드에 전달
</p>
{Object.entries(props).map(([key, prop]) => (
<GateField key={key} paramKey={key} prop={prop}
value={value?.[key] ?? meta.default_params?.[key]}
onChange={(v) => onChange({ ...value, [key]: v })} />
))}
</section>
);
}
function GateField({ paramKey, prop, value, onChange }) {
if (prop.type === 'integer') {
return (
<div className="param-row">
<label style={{ width: 160, fontSize: 12 }}>{paramKey}</label>
<input type="number" value={value ?? ''}
min={prop.minimum} onChange={(e) => onChange(parseInt(e.target.value, 10))}
style={{ flex: 1 }} />
</div>
);
}
if (prop.type === 'boolean') {
return (
<div className="param-row">
<label>
<input type="checkbox" checked={!!value} onChange={(e) => onChange(e.target.checked)} />
<span style={{ marginLeft: 6, fontSize: 12 }}>{paramKey}</span>
</label>
</div>
);
}
return null;
}
```
- [ ] **Step 2: GlobalControls 갱신**
Replace `GlobalControls.jsx`:
```jsx
export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
return (
<section className="screener-card">
<h3>실행 옵션</h3>
<div className="param-row">
<label style={{ width: 80, fontSize: 12 }}>Top N</label>
<input type="number" value={settings.top_n}
onChange={(e) => setSettings({ ...settings, top_n: parseInt(e.target.value, 10) })}
min={5} max={100} style={{ width: 80 }} />
</div>
<div className="param-row">
<label style={{ width: 80, fontSize: 12 }}>ATR window</label>
<input type="number" value={settings.atr_window}
onChange={(e) => setSettings({ ...settings, atr_window: parseInt(e.target.value, 10) })}
min={5} max={50} style={{ width: 80 }} />
</div>
<div className="param-row">
<label style={{ width: 80, fontSize: 12 }}>손절 ×ATR</label>
<input type="number" value={settings.atr_stop_mult} step={0.1}
onChange={(e) => setSettings({ ...settings, atr_stop_mult: parseFloat(e.target.value) })}
min={0.5} max={5} style={{ width: 80 }} />
</div>
<div className="param-row">
<label style={{ width: 80, fontSize: 12 }}>R:R 비율</label>
<input type="number" value={settings.rr_ratio} step={0.1}
onChange={(e) => setSettings({ ...settings, rr_ratio: parseFloat(e.target.value) })}
min={1} max={10} style={{ width: 80 }} />
</div>
<button onClick={onRun} disabled={running}
style={{ marginTop: 16, width: '100%', padding: 10, background: '#fbbf24', color: '#0b0f17', border: 'none', borderRadius: 6, fontWeight: 600 }}>
{running ? '실행 중…' : '지금 실행 (미리보기)'}
</button>
<button onClick={onSave} disabled={running}
style={{ marginTop: 8, width: '100%', padding: 8 }}>
스냅샷 저장
</button>
<button onClick={onPersist} disabled={!dirty}
style={{ marginTop: 8, width: '100%', padding: 8, opacity: dirty ? 1 : 0.5 }}>
설정 저장 (디폴트 갱신)
</button>
</section>
);
}
```
- [ ] **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 (
<div style={{ display: 'flex', gap: 4, flexWrap: 'wrap' }}>
{Object.entries(scores || {}).map(([name, s]) => {
const meta = NODE_ICONS[name];
if (!meta) return null;
const active = s >= 70;
return (
<span key={name}
title={`${meta.label}: ${s.toFixed?.(0) ?? s}`}
style={{
padding: '2px 6px', borderRadius: 4, fontSize: 11,
background: active ? '#fbbf24' : '#1f2937',
color: active ? '#0b0f17' : '#9ca3af',
}}>
{meta.icon}{Math.round(s)}
</span>
);
})}
</div>
);
}
```
- [ ] **Step 2: ResultTable 갱신**
Replace `ResultTable.jsx`:
```jsx
import ScoreChips from './ScoreChips';
export default function ResultTable({ result }) {
if (!result) {
return (
<section className="screener-card">
<p style={{ color: '#9ca3af' }}>아직 결과 없음. "지금 실행" 눌러보세요.</p>
</section>
);
}
if (result.warnings?.length) {
// 경고 배너만 노출하고 표는 계속 렌더
}
return (
<section className="screener-card">
<div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
<h3 style={{ margin: 0 }}>
Top {result.top_n} · 통과 {result.survivors_count} · {result.asof}
</h3>
{result.warnings?.length > 0 && (
<div style={{
background: '#7c2d12', color: '#fde68a', padding: '4px 10px',
borderRadius: 4, fontSize: 12,
}}>
{result.warnings.join(' · ')}
</div>
)}
</div>
<div style={{ overflowX: 'auto', marginTop: 12 }}>
<table className="screener-table">
<thead>
<tr>
<th>#</th><th>종목</th><th>총점</th><th>노드</th>
<th>진입</th><th>손절</th><th>익절</th><th>R%</th>
</tr>
</thead>
<tbody>
{(result.results || []).map((r) => (
<tr key={r.ticker}>
<td>{r.rank}</td>
<td>{r.name}<br /><span style={{ fontSize: 11, color: '#9ca3af' }}>{r.ticker}</span></td>
<td style={{ fontWeight: 600 }}>{r.total_score?.toFixed(1)}</td>
<td><ScoreChips scores={r.scores} /></td>
<td>{r.entry_price?.toLocaleString?.()}</td>
<td>{r.stop_price?.toLocaleString?.()}</td>
<td>{r.target_price?.toLocaleString?.()}</td>
<td>{r.r_pct?.toFixed?.(1)}</td>
</tr>
))}
</tbody>
</table>
</div>
</section>
);
}
```
- [ ] **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 영역에 다음 링크 추가 (기존 `<Link to="/stock/trade">` 패턴 옆에):
```jsx
<Link to="/stock/screener">스크리너</Link>
```
- [ ] **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` 접근 → 데이터 정상 로드 확인.
---
# 끝