From f7323a5b7268073af3605cf9b2c0629f3de45d90 Mon Sep 17 00:00:00 2001 From: gahusb Date: Tue, 12 May 2026 03:48:20 +0900 Subject: [PATCH] =?UTF-8?q?docs(plan):=20Stock=20Screener=20Board=20MVP=20?= =?UTF-8?q?=EA=B5=AC=ED=98=84=20plan?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 6 Phase × 35 task. Phase 0(백엔드 기반)·Phase 1(노드 8개 TDD)· Phase 2(엔진/사이저/텔레그램)·Phase 3(라우터)·Phase 4(프론트)· Phase 5(agent-office 통합)·Phase 6(백필·검증·배포). 모든 task에 TDD step + 코드 + 명령 명시. 로컬 venv 기반 실행으로 메모리 규약(로컬 docker 금지) 준수. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../plans/2026-05-12-stock-screener-board.md | 4336 +++++++++++++++++ 1 file changed, 4336 insertions(+) create mode 100644 docs/superpowers/plans/2026-05-12-stock-screener-board.md diff --git a/docs/superpowers/plans/2026-05-12-stock-screener-board.md b/docs/superpowers/plans/2026-05-12-stock-screener-board.md new file mode 100644 index 0000000..54e52e1 --- /dev/null +++ b/docs/superpowers/plans/2026-05-12-stock-screener-board.md @@ -0,0 +1,4336 @@ +# 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에 pykrx 추가는 Task 0.1에서 진행. 그 이후 다음으로 pykrx도 설치:) +# pip install pykrx +``` + +| 작업 | 어디서 실행 | +|------|------------| +| 백엔드 단위 테스트 (`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에 pykrx 추가** + +`web-backend/stock-lab/requirements.txt`에 `pykrx>=1.0.45` 한 줄 추가. (기존 줄 위 또는 아래 어느 곳이든.) + +- [ ] **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에 pykrx 설치** + +```powershell +cd C:\Users\jaeoh\Desktop\workspace\web-backend\stock-lab +.\.venv\Scripts\Activate.ps1 +pip install pykrx +``` + +Expected: 설치 성공. 실패하면 numpy/pandas 호환성 점검(pykrx는 pandas≥1.5 요구). + +> NAS 운영 컨테이너 재빌드는 본 plan 마지막의 **최종 배포** 단계에서 `git push` → webhook으로 자동 수행. 지금은 로컬 venv 동작만 검증. + +- [ ] **Step 4: pykrx 동작 smoke test (one-off, 로컬 venv)** + +```powershell +python -c "from pykrx import stock; print(stock.get_market_ticker_list('20260512', market='KOSPI')[:5])" +``` + +Expected: 5개 KOSPI 종목 코드 출력. 실패 시 IP 차단/네트워크 확인. + +- [ ] **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): pykrx 의존성 + screener 패키지 골격" +``` + +--- + +## Task 0.2: 스키마 마이그레이션 (SQL DDL 7테이블) + +**Files:** +- Create: `web-backend/stock-lab/app/screener/schema.py` +- Modify: `web-backend/stock-lab/app/db.py` (스키마 함수 import + 호출) +- Create: `web-backend/stock-lab/app/test_screener_schema.py` + +- [ ] **Step 1: 테스트 먼저 작성** + +Create `web-backend/stock-lab/app/test_screener_schema.py`: +```python +import sqlite3 +from app.screener.schema import ensure_screener_schema + + +def test_creates_all_tables(tmp_path): + db_path = tmp_path / "test.db" + conn = sqlite3.connect(db_path) + ensure_screener_schema(conn) + + tables = {r[0] for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall()} + + expected = { + "krx_master", "krx_daily_prices", "krx_flow", + "screener_settings", "screener_runs", "screener_results", + } + assert expected.issubset(tables) + + +def test_settings_seeded_with_singleton_row(tmp_path): + db_path = tmp_path / "test.db" + conn = sqlite3.connect(db_path) + ensure_screener_schema(conn) + + rows = conn.execute("SELECT id FROM screener_settings").fetchall() + assert rows == [(1,)] + + +def test_idempotent(tmp_path): + db_path = tmp_path / "test.db" + conn = sqlite3.connect(db_path) + ensure_screener_schema(conn) + ensure_screener_schema(conn) # 두 번 호출해도 에러 없어야 함 + + rows = conn.execute("SELECT count(*) FROM screener_settings").fetchall() + assert rows == [(1,)] +``` + +- [ ] **Step 2: Run test, expect failure** + +```bash +pytest app/test_screener_schema.py -v +``` +Expected: FAIL with `ModuleNotFoundError: No module named 'app.screener.schema'` + +- [ ] **Step 3: schema.py 작성** + +Create `web-backend/stock-lab/app/screener/schema.py`: +```python +"""Screener schema bootstrap. Called once at module import via db.py.""" + +import json +import sqlite3 +from datetime import datetime, timezone + +DEFAULT_WEIGHTS = { + "foreign_buy": 1.0, + "volume_surge": 1.0, + "momentum": 1.0, + "high52w": 1.2, + "rs_rating": 1.2, + "ma_alignment": 1.0, + "vcp_lite": 0.8, +} +DEFAULT_NODE_PARAMS = { + "foreign_buy": {"window_days": 5}, + "volume_surge": {"baseline_days": 20, "eval_days": 3}, + "momentum": {"window_days": 20}, + "high52w": {"window_days": 252}, + "rs_rating": {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}}, + "ma_alignment": {"ma_periods": [50, 150, 200]}, + "vcp_lite": {"short_window": 40, "long_window": 252}, +} +DEFAULT_GATE_PARAMS = { + "min_market_cap_won": 50_000_000_000, + "min_avg_value_won": 500_000_000, + "min_listed_days": 60, + "skip_managed": True, + "skip_preferred": True, + "skip_spac": True, + "skip_halted_days": 3, +} + +DDL = """ +CREATE TABLE IF NOT EXISTS krx_master ( + ticker TEXT PRIMARY KEY, + name TEXT NOT NULL, + market TEXT NOT NULL, + market_cap INTEGER, + is_managed INTEGER NOT NULL DEFAULT 0, + is_preferred INTEGER NOT NULL DEFAULT 0, + is_spac INTEGER NOT NULL DEFAULT 0, + listed_date TEXT, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS krx_daily_prices ( + ticker TEXT NOT NULL, + date TEXT NOT NULL, + open INTEGER, high INTEGER, low INTEGER, close INTEGER, + volume INTEGER, + value INTEGER, + PRIMARY KEY (ticker, date) +); +CREATE INDEX IF NOT EXISTS idx_prices_date ON krx_daily_prices(date); + +CREATE TABLE IF NOT EXISTS krx_flow ( + ticker TEXT NOT NULL, + date TEXT NOT NULL, + foreign_net INTEGER, + institution_net INTEGER, + PRIMARY KEY (ticker, date) +); +CREATE INDEX IF NOT EXISTS idx_flow_date ON krx_flow(date); + +CREATE TABLE IF NOT EXISTS screener_settings ( + id INTEGER PRIMARY KEY CHECK (id = 1), + weights_json TEXT NOT NULL, + node_params_json TEXT NOT NULL, + gate_params_json TEXT NOT NULL, + top_n INTEGER NOT NULL DEFAULT 20, + rr_ratio REAL NOT NULL DEFAULT 2.0, + atr_window INTEGER NOT NULL DEFAULT 14, + atr_stop_mult REAL NOT NULL DEFAULT 2.0, + updated_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS screener_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + asof TEXT NOT NULL, + mode TEXT NOT NULL, + status TEXT NOT NULL, + error TEXT, + started_at TEXT NOT NULL, + finished_at TEXT, + weights_json TEXT NOT NULL, + node_params_json TEXT NOT NULL, + gate_params_json TEXT NOT NULL, + top_n INTEGER NOT NULL, + survivors_count INTEGER, + telegram_sent INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_runs_asof ON screener_runs(asof DESC); + +CREATE TABLE IF NOT EXISTS screener_results ( + run_id INTEGER NOT NULL, + rank INTEGER NOT NULL, + ticker TEXT NOT NULL, + name TEXT NOT NULL, + total_score REAL NOT NULL, + scores_json TEXT NOT NULL, + close INTEGER, + market_cap INTEGER, + entry_price INTEGER, + stop_price INTEGER, + target_price INTEGER, + atr14 REAL, + PRIMARY KEY (run_id, ticker), + FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE +); +CREATE INDEX IF NOT EXISTS idx_results_run_rank ON screener_results(run_id, rank); +""" + + +def ensure_screener_schema(conn: sqlite3.Connection) -> None: + """Create tables and seed default settings (idempotent).""" + conn.executescript(DDL) + existing = conn.execute("SELECT id FROM screener_settings WHERE id=1").fetchone() + if existing is None: + now = datetime.now(timezone.utc).isoformat() + conn.execute( + """ + INSERT INTO screener_settings ( + id, weights_json, node_params_json, gate_params_json, + top_n, rr_ratio, atr_window, atr_stop_mult, updated_at + ) VALUES (1, ?, ?, ?, 20, 2.0, 14, 2.0, ?) + """, + ( + json.dumps(DEFAULT_WEIGHTS), + json.dumps(DEFAULT_NODE_PARAMS), + json.dumps(DEFAULT_GATE_PARAMS), + now, + ), + ) + conn.commit() +``` + +- [ ] **Step 4: Run test, expect pass** + +```bash +pytest app/test_screener_schema.py -v +``` +Expected: 3 passed. + +- [ ] **Step 5: db.py에서 호출** + +Read `web-backend/stock-lab/app/db.py` 파일을 확인하고 기존 스키마 부트스트랩 함수(예: `init_db()` 또는 connection factory)를 찾으세요. 그 함수 마지막에 다음을 추가: + +```python +from app.screener.schema import ensure_screener_schema + +# 기존 코드 다음에: +ensure_screener_schema(conn) +``` + +(정확한 위치는 db.py 구조에 따라 달라짐. 핵심: 앱 시작 시 1회 호출되도록.) + +- [ ] **Step 6: 스키마 적용은 단위 테스트로 검증 (이미 Step 4에서 완료)** + +운영 DB에 대한 실제 적용은 최종 배포 후 NAS에서 자동 수행됨 (`app/db.py` import 시 `ensure_screener_schema(conn)` 호출). 로컬 검증은 step 4의 pytest 통과로 충분. + +- [ ] **Step 7: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/schema.py stock-lab/app/test_screener_schema.py stock-lab/app/db.py +git commit -m "feat(stock-lab): screener 스키마 7테이블 + 디폴트 설정 시드" +``` + +--- + +## Task 0.3: snapshot.py — KRX 캐시 갱신 + +**Files:** +- Create: `web-backend/stock-lab/app/screener/snapshot.py` +- Create: `web-backend/stock-lab/app/test_screener_snapshot.py` + +- [ ] **Step 1: 테스트 먼저** + +Create `web-backend/stock-lab/app/test_screener_snapshot.py`: +```python +import datetime as dt +import sqlite3 +from unittest.mock import patch +import pandas as pd +import pytest + +from app.screener.schema import ensure_screener_schema +from app.screener.snapshot import refresh_daily, backfill + + +@pytest.fixture +def conn(tmp_path): + db_path = tmp_path / "test.db" + c = sqlite3.connect(db_path) + ensure_screener_schema(c) + yield c + c.close() + + +def _stub_pykrx(monkeypatch): + """pykrx 의 데이터 호출을 메모리 픽스쳐로 대체.""" + def fake_ticker_list(date, market): + return ["005930", "035420"] if market == "KOSPI" else ["091990"] + + def fake_ohlcv(date, ticker, *_args, **_kwargs): + return pd.DataFrame({ + "시가": [70000], "고가": [72000], "저가": [69500], + "종가": [71000], "거래량": [12_000_000], "거래대금": [840_000_000_000], + }, index=[pd.Timestamp(date)]) + + def fake_market_cap(date): + return pd.DataFrame({ + "시가총액": [420_000_000_000_000, 30_000_000_000_000, 10_000_000_000_000], + "상장주식수": [5_900_000_000, 164_000_000, 12_000_000], + }, index=["005930", "035420", "091990"]) + + def fake_trading_value(date, *_a, **_k): + return pd.DataFrame({ + "외국인": [12_000_000_000, -3_000_000_000, 500_000_000], + "기관": [4_000_000_000, 8_000_000_000, -100_000_000], + }, index=["005930", "035420", "091990"]) + + monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_ticker_list", fake_ticker_list) + monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_ohlcv", fake_ohlcv) + monkeypatch.setattr("app.screener.snapshot.pykrx_stock.get_market_cap", fake_market_cap) + monkeypatch.setattr( + "app.screener.snapshot.pykrx_stock.get_market_trading_value_by_ticker", + fake_trading_value, + ) + + +def test_refresh_daily_writes_master_prices_flow(conn, monkeypatch): + _stub_pykrx(monkeypatch) + + asof = dt.date(2026, 5, 12) + summary = refresh_daily(conn, asof) + + assert summary["master_count"] == 3 + assert summary["prices_count"] == 3 + assert summary["flow_count"] == 3 + assert conn.execute( + "SELECT close FROM krx_daily_prices WHERE ticker='005930' AND date='2026-05-12'" + ).fetchone()[0] == 71000 + + +def test_refresh_daily_is_idempotent(conn, monkeypatch): + _stub_pykrx(monkeypatch) + asof = dt.date(2026, 5, 12) + refresh_daily(conn, asof) + refresh_daily(conn, asof) # 두 번 호출 + cnt = conn.execute( + "SELECT count(*) FROM krx_daily_prices WHERE date='2026-05-12'" + ).fetchone()[0] + assert cnt == 3 # 중복 row 없어야 함 +``` + +- [ ] **Step 2: Run test, expect failure** + +```bash +pytest app/test_screener_snapshot.py -v +``` +Expected: FAIL — module not found. + +- [ ] **Step 3: snapshot.py 구현** + +Create `web-backend/stock-lab/app/screener/snapshot.py`: +```python +"""KRX daily snapshot loader (pykrx -> SQLite cache).""" + +from __future__ import annotations + +import datetime as dt +import logging +import sqlite3 +from dataclasses import dataclass + +import pandas as pd +from pykrx import stock as pykrx_stock + +log = logging.getLogger(__name__) + + +@dataclass +class RefreshSummary: + asof: dt.date + master_count: int + prices_count: int + flow_count: int + failures: list[str] + + def asdict(self) -> dict: + return { + "asof": self.asof.isoformat(), + "master_count": self.master_count, + "prices_count": self.prices_count, + "flow_count": self.flow_count, + "failures": self.failures, + } + + +def _date_str(d: dt.date) -> str: + return d.strftime("%Y%m%d") + + +def _iso(d: dt.date) -> str: + return d.isoformat() + + +def _gather_master(asof: dt.date) -> list[tuple]: + rows: list[tuple] = [] + now_iso = dt.datetime.utcnow().isoformat() + for market in ("KOSPI", "KOSDAQ"): + tickers = pykrx_stock.get_market_ticker_list(_date_str(asof), market=market) + cap_df = pykrx_stock.get_market_cap(_date_str(asof)) + for ticker in tickers: + try: + name = pykrx_stock.get_market_ticker_name(ticker) + except Exception: + name = ticker + market_cap = int(cap_df.loc[ticker, "시가총액"]) if ticker in cap_df.index else None + # 관리종목·우선주·스팩 판정은 종목명 휴리스틱(MVP) + is_preferred = 1 if name.endswith("우") else 0 + is_spac = 1 if "스팩" in name else 0 + is_managed = 0 # pykrx에 직접 플래그 없음 — Phase 후속에서 KRX 공시 파이프로 + rows.append(( + ticker, name, market, market_cap, + is_managed, is_preferred, is_spac, + None, # listed_date — MVP에서는 unknown + now_iso, + )) + return rows + + +def _gather_prices(asof: dt.date, tickers: list[str]) -> list[tuple]: + rows: list[tuple] = [] + iso = _iso(asof) + for t in tickers: + try: + df = pykrx_stock.get_market_ohlcv(_date_str(asof), _date_str(asof), t) + if df.empty: + continue + row = df.iloc[0] + rows.append(( + t, iso, + int(row["시가"]), int(row["고가"]), int(row["저가"]), int(row["종가"]), + int(row["거래량"]), int(row["거래대금"]), + )) + except Exception as e: + log.warning("price fetch failed for %s: %s", t, e) + return rows + + +def _gather_flow(asof: dt.date) -> list[tuple]: + iso = _iso(asof) + try: + df = pykrx_stock.get_market_trading_value_by_ticker(_date_str(asof), _date_str(asof)) + rows = [ + (idx, iso, int(r["외국인"]), int(r["기관"])) + for idx, r in df.iterrows() + ] + return rows + except Exception as e: + log.warning("flow fetch failed: %s", e) + return [] + + +def refresh_daily(conn: sqlite3.Connection, asof: dt.date) -> dict: + """Pull pykrx data for asof and upsert into the cache tables.""" + master_rows = _gather_master(asof) + conn.executemany( + """ + INSERT INTO krx_master ( + ticker, name, market, market_cap, + is_managed, is_preferred, is_spac, + listed_date, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(ticker) DO UPDATE SET + name=excluded.name, market=excluded.market, + market_cap=excluded.market_cap, + is_managed=excluded.is_managed, + is_preferred=excluded.is_preferred, + is_spac=excluded.is_spac, + updated_at=excluded.updated_at + """, + master_rows, + ) + tickers = [r[0] for r in master_rows] + + price_rows = _gather_prices(asof, tickers) + conn.executemany( + """ + INSERT OR REPLACE INTO krx_daily_prices ( + ticker, date, open, high, low, close, volume, value + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + price_rows, + ) + + flow_rows = _gather_flow(asof) + conn.executemany( + """ + INSERT OR REPLACE INTO krx_flow ( + ticker, date, foreign_net, institution_net + ) VALUES (?, ?, ?, ?) + """, + flow_rows, + ) + conn.commit() + + return RefreshSummary( + asof=asof, + master_count=len(master_rows), + prices_count=len(price_rows), + flow_count=len(flow_rows), + failures=[], + ).asdict() + + +def backfill(conn: sqlite3.Connection, start: dt.date, end: dt.date) -> list[dict]: + """Run refresh_daily for every weekday in [start, end] (inclusive). + + 공휴일도 포함 호출하되, pykrx가 빈 응답을 주므로 자연스럽게 skip 됨. + """ + results = [] + d = start + while d <= end: + if d.weekday() < 5: # 0=Mon ... 4=Fri + try: + results.append(refresh_daily(conn, d)) + except Exception as e: + log.error("backfill failed for %s: %s", d, e) + results.append({"asof": d.isoformat(), "error": str(e)}) + d += dt.timedelta(days=1) + return results +``` + +- [ ] **Step 4: Run test, expect pass** + +```bash +pytest app/test_screener_snapshot.py -v +``` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/snapshot.py stock-lab/app/test_screener_snapshot.py +git commit -m "feat(stock-lab): pykrx 일봉·수급 캐시 갱신 (refresh_daily + backfill)" +``` + +--- + +## Task 0.4: ScreenContext.load() — DB → DataFrame + +**Files:** +- Create: `web-backend/stock-lab/app/screener/engine.py` (ScreenContext만 우선) +- Create: `web-backend/stock-lab/app/screener/_test_fixtures.py` +- Create: `web-backend/stock-lab/app/test_screener_context.py` + +- [ ] **Step 1: 테스트 픽스쳐 헬퍼** + +Create `web-backend/stock-lab/app/screener/_test_fixtures.py`: +```python +"""Synthetic fixtures for screener tests — no DB / no pykrx.""" + +import datetime as dt +import pandas as pd + + +def make_master(tickers: list[str], market_caps: dict | None = None, + preferred: set | None = None, managed: set | None = None) -> pd.DataFrame: + market_caps = market_caps or {t: 100_000_000_000 for t in tickers} + preferred = preferred or set() + managed = managed or set() + return pd.DataFrame([ + { + "ticker": t, + "name": f"테스트{t}", + "market": "KOSPI", + "market_cap": market_caps.get(t), + "is_managed": int(t in managed), + "is_preferred": int(t in preferred), + "is_spac": 0, + "listed_date": None, + } + for t in tickers + ]).set_index("ticker") + + +def make_prices(tickers: list[str], days: int = 260, start_close: int = 50000, + trend_pct: float = 0.0, asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame: + """trend_pct: 일별 종가 등락률(%). 양수면 상승 추세.""" + rows = [] + for t in tickers: + close = start_close + for i in range(days): + day_idx = days - 1 - i # asof가 마지막 + date = asof - dt.timedelta(days=day_idx) + high = int(close * 1.012) + low = int(close * 0.988) + rows.append({ + "ticker": t, "date": date.isoformat(), + "open": close, "high": high, "low": low, "close": close, + "volume": 1_000_000, "value": close * 1_000_000, + }) + close = int(close * (1 + trend_pct / 100)) + return pd.DataFrame(rows) + + +def make_flow(tickers: list[str], days: int = 260, + foreign_per_day: dict | None = None, + asof: dt.date = dt.date(2026, 5, 12)) -> pd.DataFrame: + foreign_per_day = foreign_per_day or {t: 0 for t in tickers} + rows = [] + for t in tickers: + for i in range(days): + day_idx = days - 1 - i + date = asof - dt.timedelta(days=day_idx) + rows.append({ + "ticker": t, "date": date.isoformat(), + "foreign_net": foreign_per_day.get(t, 0), + "institution_net": 0, + }) + return pd.DataFrame(rows) + + +def make_kospi(days: int = 260, start: int = 2500, trend_pct: float = 0.0, + asof: dt.date = dt.date(2026, 5, 12)) -> pd.Series: + values = [] + dates = [] + v = start + for i in range(days): + day_idx = days - 1 - i + d = asof - dt.timedelta(days=day_idx) + dates.append(d.isoformat()) + values.append(v) + v = v * (1 + trend_pct / 100) + return pd.Series(values, index=dates, name="kospi") +``` + +Create `web-backend/stock-lab/app/test_screener_context.py`: +```python +import datetime as dt +import sqlite3 +import pytest +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.schema import ensure_screener_schema +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +@pytest.fixture +def conn(tmp_path): + db_path = tmp_path / "ctx.db" + c = sqlite3.connect(db_path) + ensure_screener_schema(c) + yield c + c.close() + + +def _seed(conn, master_df, prices_df, flow_df): + now = dt.datetime.utcnow().isoformat() + for t, row in master_df.iterrows(): + conn.execute("""INSERT INTO krx_master (ticker,name,market,market_cap, + is_managed,is_preferred,is_spac,listed_date,updated_at) + VALUES (?,?,?,?,?,?,?,?,?)""", + (t, row["name"], row["market"], row["market_cap"], + row["is_managed"], row["is_preferred"], row["is_spac"], None, now)) + prices_df.to_sql("krx_daily_prices", conn, if_exists="append", index=False) + flow_df.to_sql("krx_flow", conn, if_exists="append", index=False) + conn.commit() + + +def test_load_returns_dataframes(conn): + asof = dt.date(2026, 5, 12) + _seed(conn, + make_master(["005930", "035420"]), + make_prices(["005930", "035420"], days=30, asof=asof), + make_flow(["005930", "035420"], days=30, asof=asof)) + + ctx = ScreenContext.load(conn, asof, lookback_days=30) + + assert ctx.asof == asof + assert set(ctx.master.index) == {"005930", "035420"} + assert ctx.prices.shape[0] == 60 # 2 종목 × 30일 + assert ctx.flow.shape[0] == 60 + + +def test_restrict_filters_tickers(conn): + asof = dt.date(2026, 5, 12) + _seed(conn, + make_master(["005930", "035420", "091990"]), + make_prices(["005930", "035420", "091990"], days=30, asof=asof), + make_flow(["005930", "035420", "091990"], days=30, asof=asof)) + + ctx = ScreenContext.load(conn, asof, lookback_days=30) + scoped = ctx.restrict(pd.Index(["005930"])) + + assert list(scoped.master.index) == ["005930"] + assert (scoped.prices["ticker"] == "005930").all() + assert (scoped.flow["ticker"] == "005930").all() +``` + +- [ ] **Step 2: Run test, expect failure** + +```bash +pytest app/test_screener_context.py -v +``` +Expected: FAIL — `ScreenContext` not defined. + +- [ ] **Step 3: engine.py에 ScreenContext 구현 (엔진 본체는 Phase 2에서)** + +Create `web-backend/stock-lab/app/screener/engine.py`: +```python +"""Screener engine — ScreenContext & Screener. + +Phase 0: ScreenContext만 구현. +Phase 2에서 Screener / combine() 추가. +""" + +from __future__ import annotations + +import datetime as dt +import sqlite3 +from dataclasses import dataclass, replace + +import pandas as pd + + +@dataclass(frozen=True) +class ScreenContext: + """1회 실행 동안 공유되는 읽기 전용 데이터 컨테이너.""" + master: pd.DataFrame # index=ticker + prices: pd.DataFrame # cols: ticker,date,open,high,low,close,volume,value + flow: pd.DataFrame # cols: ticker,date,foreign_net,institution_net + kospi: pd.Series # index=date(str), name="kospi" + asof: dt.date + + @classmethod + def load(cls, conn: sqlite3.Connection, asof: dt.date, + lookback_days: int = 252 * 2) -> "ScreenContext": + cutoff = (asof - dt.timedelta(days=int(lookback_days * 1.5))).isoformat() + asof_iso = asof.isoformat() + + master = pd.read_sql_query( + "SELECT * FROM krx_master", + conn, index_col="ticker", + ) + prices = pd.read_sql_query( + "SELECT ticker,date,open,high,low,close,volume,value " + "FROM krx_daily_prices WHERE date BETWEEN ? AND ? ORDER BY date", + conn, params=(cutoff, asof_iso), + ) + flow = pd.read_sql_query( + "SELECT ticker,date,foreign_net,institution_net " + "FROM krx_flow WHERE date BETWEEN ? AND ? ORDER BY date", + conn, params=(cutoff, asof_iso), + ) + + # KOSPI 지수는 MVP에서 005930(삼성전자) 대용 또는 별도 인덱스 캐시. + # 현 단계는 005930로 fallback. Phase 후속에서 ^KS11 시계열 별도 캐시. + kospi = pd.Series(dtype=float, name="kospi") + if "005930" in master.index and not prices.empty: + sub = prices[prices["ticker"] == "005930"].set_index("date")["close"] + kospi = sub.copy() + kospi.name = "kospi" + + return cls(master=master, prices=prices, flow=flow, kospi=kospi, asof=asof) + + def restrict(self, tickers) -> "ScreenContext": + tickers = pd.Index(tickers) + return replace( + self, + master=self.master.loc[self.master.index.intersection(tickers)], + prices=self.prices[self.prices["ticker"].isin(tickers)], + flow=self.flow[self.flow["ticker"].isin(tickers)], + ) + + def latest_close(self) -> pd.Series: + if self.prices.empty: + return pd.Series(dtype=float) + latest = self.prices.sort_values("date").groupby("ticker")["close"].last() + return latest + + def latest_high(self) -> pd.Series: + if self.prices.empty: + return pd.Series(dtype=float) + return self.prices.sort_values("date").groupby("ticker")["high"].last() +``` + +- [ ] **Step 4: Run test, expect pass** + +```bash +pytest app/test_screener_context.py -v +``` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/engine.py stock-lab/app/screener/_test_fixtures.py stock-lab/app/test_screener_context.py +git commit -m "feat(stock-lab): ScreenContext.load/restrict + 합성 픽스쳐" +``` + +--- + +# Phase 1 — 노드 8개 (TDD) + +각 노드 패턴은 동일: `nodes/base.py` 추상 → 노드 구현 → 단위 테스트 → commit. + +## Task 1.1: ScoreNode / GateNode 추상 + percentile_rank 유틸 + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/__init__.py` (빈 파일) +- Create: `web-backend/stock-lab/app/screener/nodes/base.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_base.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_base.py`: +```python +import pandas as pd +import pytest + +from app.screener.nodes.base import percentile_rank + + +def test_percentile_rank_basic(): + s = pd.Series([10, 20, 30, 40, 50]) + out = percentile_rank(s) + assert out.iloc[0] == pytest.approx(10.0) # 1/5 * 100 - half adjustment + assert out.iloc[-1] == pytest.approx(90.0) + assert (out >= 0).all() and (out <= 100).all() + + +def test_percentile_rank_all_equal_returns_50(): + s = pd.Series([42, 42, 42, 42]) + out = percentile_rank(s) + assert (out == 50.0).all() + + +def test_percentile_rank_handles_nan(): + s = pd.Series([1.0, float("nan"), 3.0, 5.0]) + out = percentile_rank(s) + assert pd.isna(out.iloc[1]) + assert (out.dropna() >= 0).all() +``` + +- [ ] **Step 2: Run test, expect failure** + +```bash +pytest app/test_screener_nodes_base.py -v +``` +Expected: FAIL — module not found. + +- [ ] **Step 3: base.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/__init__.py` (빈 파일). + +Create `web-backend/stock-lab/app/screener/nodes/base.py`: +```python +"""Node base classes + helpers.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any, ClassVar + +import pandas as pd + + +class ScoreNode(ABC): + name: ClassVar[str] + label: ClassVar[str] + default_params: ClassVar[dict] + param_schema: ClassVar[dict] + + @abstractmethod + def compute(self, ctx: "Any", params: dict) -> pd.Series: + """returns Series indexed by ticker, 0..100 float.""" + + +class GateNode(ABC): + name: ClassVar[str] + label: ClassVar[str] + default_params: ClassVar[dict] + param_schema: ClassVar[dict] + + @abstractmethod + def filter(self, ctx: "Any", params: dict) -> pd.Index: + """returns surviving tickers.""" + + +def percentile_rank(series: pd.Series) -> pd.Series: + """Percentile rank in [0, 100]. All-equal → 50. NaN preserved.""" + if series.empty: + return series.astype(float) + if series.dropna().nunique() == 1: + return pd.Series(50.0, index=series.index) + ranked = series.rank(pct=True, na_option="keep") * 100.0 + return ranked +``` + +- [ ] **Step 4: Run test, expect pass** + +```bash +pytest app/test_screener_nodes_base.py -v +``` +Expected: 3 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/__init__.py stock-lab/app/screener/nodes/base.py stock-lab/app/test_screener_nodes_base.py +git commit -m "feat(stock-lab): ScoreNode/GateNode 추상 + percentile_rank 유틸" +``` + +--- + +## Task 1.2: HygieneGate + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/hygiene.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_hygiene.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_hygiene.py`: +```python +import datetime as dt +from app.screener.nodes.hygiene import HygieneGate +from app.screener.engine import ScreenContext +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=__import__("pandas").Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_filter_excludes_small_cap(): + g = HygieneGate() + ctx = _ctx( + make_master(["A", "B"], market_caps={"A": 1_000_000_000, "B": 100_000_000_000}), + make_prices(["A", "B"], days=30), + make_flow(["A", "B"], days=30), + ) + out = g.filter(ctx, {**g.default_params, "min_listed_days": 0}) + assert list(out) == ["B"] + + +def test_filter_excludes_preferred(): + g = HygieneGate() + ctx = _ctx( + make_master(["A", "B"], preferred={"B"}), + make_prices(["A", "B"], days=30), + make_flow(["A", "B"], days=30), + ) + out = g.filter(ctx, {**g.default_params, "min_listed_days": 0}) + assert list(out) == ["A"] + + +def test_filter_excludes_low_value(): + g = HygieneGate() + prices = make_prices(["A", "B"], days=30) + prices.loc[prices["ticker"] == "A", "value"] = 100_000 # 매우 작음 + ctx = _ctx(make_master(["A", "B"]), prices, make_flow(["A", "B"], days=30)) + out = g.filter(ctx, {**g.default_params, "min_listed_days": 0}) + assert list(out) == ["B"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_hygiene.py -v` +Expected: FAIL — module not found. + +- [ ] **Step 3: hygiene.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/hygiene.py`: +```python +"""HygieneGate — pre-filter for screener.""" + +from __future__ import annotations + +import pandas as pd + +from .base import GateNode + + +class HygieneGate(GateNode): + name = "hygiene" + label = "위생 게이트" + default_params = { + "min_market_cap_won": 50_000_000_000, + "min_avg_value_won": 500_000_000, + "min_listed_days": 60, + "skip_managed": True, + "skip_preferred": True, + "skip_spac": True, + "skip_halted_days": 3, + } + param_schema = { + "type": "object", + "properties": { + "min_market_cap_won": {"type": "integer", "minimum": 0}, + "min_avg_value_won": {"type": "integer", "minimum": 0}, + "min_listed_days": {"type": "integer", "minimum": 0}, + "skip_managed": {"type": "boolean"}, + "skip_preferred": {"type": "boolean"}, + "skip_spac": {"type": "boolean"}, + "skip_halted_days": {"type": "integer", "minimum": 0}, + }, + } + + def filter(self, ctx, params: dict) -> pd.Index: + master = ctx.master.copy() + prices = ctx.prices + + # 시총 + master = master[master["market_cap"].fillna(0) >= params["min_market_cap_won"]] + + # 우선주·관리·스팩 + if params.get("skip_preferred", True): + master = master[master["is_preferred"] == 0] + if params.get("skip_managed", True): + master = master[master["is_managed"] == 0] + if params.get("skip_spac", True): + master = master[master["is_spac"] == 0] + + candidates = master.index + + # 20일 평균 거래대금 + if not prices.empty: + recent20 = ( + prices[prices["ticker"].isin(candidates)] + .sort_values("date") + .groupby("ticker") + .tail(20) + ) + avg_value = recent20.groupby("ticker")["value"].mean() + ok = avg_value[avg_value >= params["min_avg_value_won"]].index + candidates = candidates.intersection(ok) + + # 최근 N일 거래정지 (volume==0 N일 이상) + halted_days = params.get("skip_halted_days", 3) + if halted_days > 0 and not prices.empty: + recent = ( + prices[prices["ticker"].isin(candidates)] + .sort_values("date") + .groupby("ticker") + .tail(halted_days) + ) + zero_count = recent.assign(z=lambda d: (d["volume"] == 0).astype(int)) \ + .groupby("ticker")["z"].sum() + healthy = zero_count[zero_count < halted_days].index + candidates = candidates.intersection(healthy) + + # 상장 N일 — MVP에선 listed_date null 허용, null이면 통과 + return pd.Index(candidates) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_hygiene.py -v` +Expected: 3 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/hygiene.py stock-lab/app/test_screener_nodes_hygiene.py +git commit -m "feat(stock-lab): HygieneGate — 위생 필터 (시총/거래대금/우선주/관리종목)" +``` + +--- + +## Task 1.3: ForeignBuy + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/foreign_buy.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_foreign_buy.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_foreign_buy.py`: +```python +import datetime as dt +import pandas as pd +import pytest + +from app.screener.engine import ScreenContext +from app.screener.nodes.foreign_buy import ForeignBuy +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_higher_foreign_buy_gets_higher_score(): + asof = dt.date(2026, 5, 12) + master = make_master(["A", "B"]) + prices = make_prices(["A", "B"], days=30, asof=asof) + flow = make_flow(["A", "B"], days=30, asof=asof, + foreign_per_day={"A": 100_000_000, "B": 0}) + out = ForeignBuy().compute(_ctx(master, prices, flow), {"window_days": 5}) + assert out["A"] > out["B"] + assert 0 <= out.min() <= out.max() <= 100 + + +def test_all_zero_returns_50(): + asof = dt.date(2026, 5, 12) + master = make_master(["A", "B"]) + prices = make_prices(["A", "B"], days=30, asof=asof) + flow = make_flow(["A", "B"], days=30, asof=asof, foreign_per_day={"A": 0, "B": 0}) + out = ForeignBuy().compute(_ctx(master, prices, flow), {"window_days": 5}) + assert (out == 50.0).all() +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_foreign_buy.py -v` +Expected: FAIL. + +- [ ] **Step 3: foreign_buy.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/foreign_buy.py`: +```python +"""외국인 N일 누적 순매수 강도 (시총 대비).""" + +import pandas as pd + +from .base import ScoreNode, percentile_rank + + +class ForeignBuy(ScoreNode): + name = "foreign_buy" + label = "외국인 누적 순매수" + default_params = {"window_days": 5} + param_schema = { + "type": "object", + "properties": { + "window_days": {"type": "integer", "minimum": 1, "maximum": 60, "default": 5} + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + window = int(params.get("window_days", 5)) + flow = ctx.flow + if flow.empty: + return pd.Series(dtype=float) + + last_dates = ( + flow.sort_values("date").groupby("ticker").tail(window) + ) + net_sum = last_dates.groupby("ticker")["foreign_net"].sum() + + market_cap = ctx.master["market_cap"].fillna(0).reindex(net_sum.index) + raw = (net_sum / market_cap.replace(0, pd.NA)).astype(float) + + return percentile_rank(raw).fillna(50.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_foreign_buy.py -v` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/foreign_buy.py stock-lab/app/test_screener_nodes_foreign_buy.py +git commit -m "feat(stock-lab): ForeignBuy 노드 — 외국인 N일 누적 순매수 강도" +``` + +--- + +## Task 1.4: VolumeSurge + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/volume_surge.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_volume_surge.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_volume_surge.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.volume_surge import VolumeSurge +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_recent_volume_surge_gets_higher_score(): + asof = dt.date(2026, 5, 12) + master = make_master(["A", "B"]) + prices = make_prices(["A", "B"], days=30, asof=asof) + # A는 최근 3일 거래량 10배로 + mask = (prices["ticker"] == "A") & (prices["date"] >= (asof - dt.timedelta(days=3)).isoformat()) + prices.loc[mask, "volume"] *= 10 + flow = make_flow(["A", "B"], days=30, asof=asof) + + out = VolumeSurge().compute( + _ctx(master, prices, flow), + {"baseline_days": 20, "eval_days": 3}, + ) + assert out["A"] > out["B"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_volume_surge.py -v` + +- [ ] **Step 3: volume_surge.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/volume_surge.py`: +```python +"""거래량 급증 — log1p(recent/baseline).""" + +import numpy as np +import pandas as pd + +from .base import ScoreNode, percentile_rank + + +class VolumeSurge(ScoreNode): + name = "volume_surge" + label = "거래량 급증" + default_params = {"baseline_days": 20, "eval_days": 3} + param_schema = { + "type": "object", + "properties": { + "baseline_days": {"type": "integer", "minimum": 5, "maximum": 60, "default": 20}, + "eval_days": {"type": "integer", "minimum": 1, "maximum": 10, "default": 3}, + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + baseline = int(params.get("baseline_days", 20)) + eval_d = int(params.get("eval_days", 3)) + prices = ctx.prices + if prices.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date") + last_recent = ordered.groupby("ticker").tail(eval_d).groupby("ticker")["volume"].mean() + last_baseline = ( + ordered.groupby("ticker") + .tail(baseline + eval_d) + .groupby("ticker") + .head(baseline) + .groupby("ticker")["volume"] + .mean() + ) + ratio = last_recent / last_baseline.replace(0, pd.NA) + raw = np.log1p(ratio.astype(float)) + return percentile_rank(raw).fillna(50.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_volume_surge.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/volume_surge.py stock-lab/app/test_screener_nodes_volume_surge.py +git commit -m "feat(stock-lab): VolumeSurge 노드 — log(최근/평균) 거래량 급증" +``` + +--- + +## Task 1.5: Momentum20 + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/momentum.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_momentum.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_momentum.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.momentum import Momentum20 +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_higher_momentum_gets_higher_score(): + asof = dt.date(2026, 5, 12) + master = make_master(["UP", "DN"]) + up = make_prices(["UP"], days=30, asof=asof, trend_pct=0.5) + dn = make_prices(["DN"], days=30, asof=asof, trend_pct=-0.3) + prices = pd.concat([up, dn], ignore_index=True) + flow = make_flow(["UP", "DN"], days=30, asof=asof) + + out = Momentum20().compute(_ctx(master, prices, flow), {"window_days": 20}) + assert out["UP"] > out["DN"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_momentum.py -v` + +- [ ] **Step 3: momentum.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/momentum.py`: +```python +"""20일 모멘텀.""" + +import pandas as pd + +from .base import ScoreNode, percentile_rank + + +class Momentum20(ScoreNode): + name = "momentum" + label = "20일 모멘텀" + default_params = {"window_days": 20} + param_schema = { + "type": "object", + "properties": { + "window_days": {"type": "integer", "minimum": 5, "maximum": 120, "default": 20} + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + window = int(params.get("window_days", 20)) + prices = ctx.prices + if prices.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date") + last = ordered.groupby("ticker").tail(window + 1) + + def _ret(g): + if len(g) < window + 1: + return float("nan") + return g["close"].iloc[-1] / g["close"].iloc[0] - 1 + + raw = last.groupby("ticker").apply(_ret) + return percentile_rank(raw).fillna(50.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_momentum.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/momentum.py stock-lab/app/test_screener_nodes_momentum.py +git commit -m "feat(stock-lab): Momentum20 노드 — N일 수익률 백분위" +``` + +--- + +## Task 1.6: High52WProximity + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/high52w.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_high52w.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_high52w.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.high52w import High52WProximity +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_proximity_at_high_returns_100(): + asof = dt.date(2026, 5, 12) + master = make_master(["A"]) + prices = make_prices(["A"], days=260, asof=asof, trend_pct=0.05) + flow = make_flow(["A"], days=260, asof=asof) + + out = High52WProximity().compute(_ctx(master, prices, flow), {"window_days": 252}) + assert out["A"] >= 95 + + +def test_proximity_below_70pct_returns_0(): + asof = dt.date(2026, 5, 12) + master = make_master(["A"]) + prices = make_prices(["A"], days=260, asof=asof, start_close=100000, trend_pct=-0.5) + flow = make_flow(["A"], days=260, asof=asof) + + out = High52WProximity().compute(_ctx(master, prices, flow), {"window_days": 252}) + assert out["A"] == 0 +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_high52w.py -v` + +- [ ] **Step 3: high52w.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/high52w.py`: +```python +"""52주 신고가 근접도 (룰 기반: 70% 미만 0점, 100% 도달 100점, 선형).""" + +import pandas as pd + +from .base import ScoreNode + + +class High52WProximity(ScoreNode): + name = "high52w" + label = "52주 신고가 근접도" + default_params = {"window_days": 252} + param_schema = { + "type": "object", + "properties": { + "window_days": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252} + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + window = int(params.get("window_days", 252)) + prices = ctx.prices + if prices.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date") + last = ordered.groupby("ticker").tail(window) + agg = last.groupby("ticker").agg(close=("close", "last"), high=("high", "max")) + proximity = (agg["close"] / agg["high"]).clip(upper=1.0) + score = ((proximity - 0.7) / 0.3).clip(lower=0.0, upper=1.0) * 100.0 + return score.fillna(0.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_high52w.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/high52w.py stock-lab/app/test_screener_nodes_high52w.py +git commit -m "feat(stock-lab): High52WProximity 노드 — 신고가 대비 근접도 룰 점수" +``` + +--- + +## Task 1.7: RsRating + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/rs_rating.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_rs_rating.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_rs_rating.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.rs_rating import RsRating +from app.screener._test_fixtures import make_master, make_prices, make_flow, make_kospi + + +def _ctx(master, prices, flow, kospi): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=kospi, asof=dt.date(2026, 5, 12)) + + +def test_outperformer_gets_higher_score(): + asof = dt.date(2026, 5, 12) + master = make_master(["UP", "DN"]) + up = make_prices(["UP"], days=260, asof=asof, trend_pct=0.3) + dn = make_prices(["DN"], days=260, asof=asof, trend_pct=-0.1) + prices = pd.concat([up, dn], ignore_index=True) + flow = make_flow(["UP", "DN"], days=260, asof=asof) + kospi = make_kospi(days=260, asof=asof, trend_pct=0.0) + + out = RsRating().compute(_ctx(master, prices, flow, kospi), + RsRating.default_params) + assert out["UP"] > out["DN"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_rs_rating.py -v` + +- [ ] **Step 3: rs_rating.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/rs_rating.py`: +```python +"""RS Rating — IBD 가중 (3m=2,6m=1,9m=1,12m=1).""" + +import pandas as pd + +from .base import ScoreNode, percentile_rank + + +_PERIOD_TO_DAYS = {"3m": 63, "6m": 126, "9m": 189, "12m": 252} + + +class RsRating(ScoreNode): + name = "rs_rating" + label = "RS Rating (시장 대비 상대강도)" + default_params = {"weights": {"3m": 2, "6m": 1, "9m": 1, "12m": 1}} + param_schema = { + "type": "object", + "properties": { + "weights": {"type": "object"} + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + weights: dict = params.get("weights", self.default_params["weights"]) + prices = ctx.prices + kospi = ctx.kospi + if prices.empty or kospi.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date") + + def _excess_for_ticker(g: pd.DataFrame) -> float: + closes = g.set_index("date")["close"] + total = 0.0 + wsum = 0.0 + for period, w in weights.items(): + k = _PERIOD_TO_DAYS.get(period, 0) + if len(closes) <= k or len(kospi) <= k: + continue + r_stock = closes.iloc[-1] / closes.iloc[-(k + 1)] - 1 + r_market = kospi.iloc[-1] / kospi.iloc[-(k + 1)] - 1 + total += w * (r_stock - r_market) + wsum += w + return total / wsum if wsum else float("nan") + + raw = ordered.groupby("ticker").apply(_excess_for_ticker) + return percentile_rank(raw).fillna(50.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_rs_rating.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/rs_rating.py stock-lab/app/test_screener_nodes_rs_rating.py +git commit -m "feat(stock-lab): RsRating 노드 — IBD 가중 시장초과수익 백분위" +``` + +--- + +## Task 1.8: MaAlignment + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/ma_alignment.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_ma_alignment.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_ma_alignment.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.ma_alignment import MaAlignment +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_strong_uptrend_returns_100(): + asof = dt.date(2026, 5, 12) + master = make_master(["UP"]) + prices = make_prices(["UP"], days=260, asof=asof, start_close=50000, trend_pct=0.2) + flow = make_flow(["UP"], days=260, asof=asof) + out = MaAlignment().compute(_ctx(master, prices, flow), MaAlignment.default_params) + assert out["UP"] == 100.0 + + +def test_downtrend_returns_low(): + asof = dt.date(2026, 5, 12) + master = make_master(["DN"]) + prices = make_prices(["DN"], days=260, asof=asof, start_close=100000, trend_pct=-0.1) + flow = make_flow(["DN"], days=260, asof=asof) + out = MaAlignment().compute(_ctx(master, prices, flow), MaAlignment.default_params) + assert out["DN"] <= 20.0 +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_ma_alignment.py -v` + +- [ ] **Step 3: ma_alignment.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/ma_alignment.py`: +```python +"""이평선 정배열 점수 — 5개 조건 충족 개수 / 5 × 100.""" + +import pandas as pd + +from .base import ScoreNode + + +class MaAlignment(ScoreNode): + name = "ma_alignment" + label = "이평선 정배열" + default_params = {"ma_periods": [50, 150, 200]} + param_schema = { + "type": "object", + "properties": { + "ma_periods": {"type": "array", "items": {"type": "integer"}} + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + prices = ctx.prices + if prices.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date") + + def _score(g: pd.DataFrame) -> float: + closes = g["close"].astype(float) + if len(closes) < 252: + return float("nan") + close = closes.iloc[-1] + ma50 = closes.rolling(50).mean().iloc[-1] + ma150 = closes.rolling(150).mean().iloc[-1] + ma200 = closes.rolling(200).mean().iloc[-1] + low52 = closes.iloc[-252:].min() + conds = [ + close > ma50, + ma50 > ma150, + ma150 > ma200, + close > ma200, + close >= low52 * 1.25, + ] + return sum(conds) / 5 * 100.0 + + raw = ordered.groupby("ticker").apply(_score) + return raw.fillna(0.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_ma_alignment.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/ma_alignment.py stock-lab/app/test_screener_nodes_ma_alignment.py +git commit -m "feat(stock-lab): MaAlignment 노드 — 이평선 정배열 5조건 룰 점수" +``` + +--- + +## Task 1.9: VcpLite + +**Files:** +- Create: `web-backend/stock-lab/app/screener/nodes/vcp_lite.py` +- Create: `web-backend/stock-lab/app/test_screener_nodes_vcp_lite.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_nodes_vcp_lite.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.nodes.vcp_lite import VcpLite +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_contracting_stock_scores_higher_than_expanding(): + asof = dt.date(2026, 5, 12) + master = make_master(["CON", "EXP"]) + prices = make_prices(["CON", "EXP"], days=260, asof=asof) + + # CON: 최근 40일 변동성 축소 (high/low 좁힘) + mask_recent_con = (prices["ticker"] == "CON") & ( + prices["date"] >= (asof - dt.timedelta(days=40)).isoformat() + ) + prices.loc[mask_recent_con, "high"] = (prices.loc[mask_recent_con, "close"] * 1.003).astype(int) + prices.loc[mask_recent_con, "low"] = (prices.loc[mask_recent_con, "close"] * 0.997).astype(int) + + # EXP: 최근 40일 변동성 확대 + mask_recent_exp = (prices["ticker"] == "EXP") & ( + prices["date"] >= (asof - dt.timedelta(days=40)).isoformat() + ) + prices.loc[mask_recent_exp, "high"] = (prices.loc[mask_recent_exp, "close"] * 1.05).astype(int) + prices.loc[mask_recent_exp, "low"] = (prices.loc[mask_recent_exp, "close"] * 0.95).astype(int) + + flow = make_flow(["CON", "EXP"], days=260, asof=asof) + out = VcpLite().compute(_ctx(master, prices, flow), VcpLite.default_params) + assert out["CON"] > out["EXP"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_nodes_vcp_lite.py -v` + +- [ ] **Step 3: vcp_lite.py 구현** + +Create `web-backend/stock-lab/app/screener/nodes/vcp_lite.py`: +```python +"""VCP-lite — 단기/장기 일중 변동성 비율 기반 수축률.""" + +import pandas as pd + +from .base import ScoreNode, percentile_rank + + +class VcpLite(ScoreNode): + name = "vcp_lite" + label = "VCP-lite (변동성 수축)" + default_params = {"short_window": 40, "long_window": 252} + param_schema = { + "type": "object", + "properties": { + "short_window": {"type": "integer", "minimum": 10, "maximum": 120, "default": 40}, + "long_window": {"type": "integer", "minimum": 60, "maximum": 504, "default": 252}, + }, + } + + def compute(self, ctx, params: dict) -> pd.Series: + short_w = int(params.get("short_window", 40)) + long_w = int(params.get("long_window", 252)) + prices = ctx.prices + if prices.empty: + return pd.Series(dtype=float) + + ordered = prices.sort_values("date").copy() + ordered["range_pct"] = (ordered["high"] - ordered["low"]) / ordered["close"] + + def _ratio(g: pd.DataFrame) -> float: + if len(g) < long_w: + return float("nan") + short_vol = g["range_pct"].tail(short_w).mean() + long_vol = g["range_pct"].tail(long_w).mean() + if long_vol == 0 or pd.isna(long_vol): + return float("nan") + return 1 - (short_vol / long_vol) + + raw = ordered.groupby("ticker").apply(_ratio) + return percentile_rank(raw).fillna(50.0) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_nodes_vcp_lite.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/nodes/vcp_lite.py stock-lab/app/test_screener_nodes_vcp_lite.py +git commit -m "feat(stock-lab): VcpLite 노드 — 변동성 수축률 백분위" +``` + +--- + +# Phase 2 — 엔진 · 포지션 사이저 · 텔레그램 · 레지스트리 + +## Task 2.1: position_sizer.py — ATR Wilder + entry/stop/target + +**Files:** +- Create: `web-backend/stock-lab/app/screener/position_sizer.py` +- Create: `web-backend/stock-lab/app/test_screener_position_sizer.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_position_sizer.py`: +```python +import datetime as dt +import pandas as pd + +from app.screener.engine import ScreenContext +from app.screener.position_sizer import compute_atr_wilder, plan_positions +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=pd.Series(dtype=float, name="kospi"), + asof=dt.date(2026, 5, 12)) + + +def test_atr_wilder_positive_and_smooth(): + df = make_prices(["A"], days=30) + atr = compute_atr_wilder(df[df["ticker"] == "A"], window=14) + assert atr > 0 + + +def test_plan_positions_returns_entry_stop_target(): + asof = dt.date(2026, 5, 12) + master = make_master(["A"]) + prices = make_prices(["A"], days=30, asof=asof, start_close=50000) + flow = make_flow(["A"], days=30, asof=asof) + ctx = _ctx(master, prices, flow) + sizing = plan_positions(ctx, ["A"], {"atr_window": 14, "atr_stop_mult": 2.0, "rr_ratio": 2.0}) + + row = sizing["A"] + assert row["entry_price"] > 0 + assert row["stop_price"] < row["entry_price"] + assert row["target_price"] > row["entry_price"] + assert row["atr14"] > 0 +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_position_sizer.py -v` + +- [ ] **Step 3: position_sizer.py 구현** + +Create `web-backend/stock-lab/app/screener/position_sizer.py`: +```python +"""ATR Wilder smoothing + entry/stop/target 계산.""" + +import pandas as pd + + +def compute_atr_wilder(df_one_ticker: pd.DataFrame, window: int = 14) -> float: + """단일 종목 DataFrame(date·open·high·low·close)에 대해 Wilder ATR 마지막 값.""" + g = df_one_ticker.sort_values("date").copy() + high = g["high"].astype(float) + low = g["low"].astype(float) + close = g["close"].astype(float) + prev_close = close.shift(1) + tr = pd.concat([ + (high - low), + (high - prev_close).abs(), + (low - prev_close).abs(), + ], axis=1).max(axis=1) + atr = tr.ewm(alpha=1 / window, adjust=False).mean() + return float(atr.iloc[-1]) + + +def round_won(x: float) -> int: + return int(round(x)) + + +def plan_positions(ctx, tickers: list[str], params: dict) -> dict: + """각 ticker 에 대해 entry/stop/target/atr14 반환.""" + atr_window = int(params.get("atr_window", 14)) + stop_mult = float(params.get("atr_stop_mult", 2.0)) + rr = float(params.get("rr_ratio", 2.0)) + + prices = ctx.prices.sort_values("date") + out: dict = {} + for t in tickers: + sub = prices[prices["ticker"] == t] + if sub.empty: + continue + close = float(sub["close"].iloc[-1]) + atr14 = compute_atr_wilder(sub, window=atr_window) + entry = round_won(close * 1.005) + stop = round_won(close - stop_mult * atr14) + target = round_won(entry + rr * (entry - stop)) + r_pct = (entry - stop) / entry * 100 if entry else 0.0 + out[t] = { + "entry_price": entry, + "stop_price": stop, + "target_price": target, + "atr14": atr14, + "r_pct": r_pct, + } + return out +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_position_sizer.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/position_sizer.py stock-lab/app/test_screener_position_sizer.py +git commit -m "feat(stock-lab): position_sizer — ATR Wilder + entry/stop/target" +``` + +--- + +## Task 2.2: registry.py + engine.Screener + combine + +**Files:** +- Create: `web-backend/stock-lab/app/screener/registry.py` +- Modify: `web-backend/stock-lab/app/screener/engine.py` (Screener·combine 추가) +- Create: `web-backend/stock-lab/app/test_screener_engine.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_engine.py`: +```python +import datetime as dt +import pandas as pd +import pytest + +from app.screener.engine import ScreenContext, Screener, combine +from app.screener.nodes.hygiene import HygieneGate +from app.screener.nodes.foreign_buy import ForeignBuy +from app.screener.nodes.momentum import Momentum20 +from app.screener._test_fixtures import make_master, make_prices, make_flow, make_kospi + + +def _ctx(master, prices, flow): + return ScreenContext(master=master, prices=prices, flow=flow, + kospi=make_kospi(days=260), + asof=dt.date(2026, 5, 12)) + + +def test_combine_weighted_average(): + scores = { + "foreign_buy": pd.Series({"A": 80, "B": 20}), + "momentum": pd.Series({"A": 60, "B": 40}), + } + weights = {"foreign_buy": 2.0, "momentum": 1.0} + out = combine(scores, weights) + # A: (80*2 + 60*1)/3 = 73.33 + assert abs(out["A"] - 73.333) < 0.1 + assert abs(out["B"] - 26.666) < 0.1 + + +def test_combine_all_zero_weight_raises(): + scores = {"foreign_buy": pd.Series({"A": 80})} + with pytest.raises(ValueError, match="no active"): + combine(scores, {"foreign_buy": 0}) + + +def test_screener_run_end_to_end(): + asof = dt.date(2026, 5, 12) + master = make_master(["GOOD", "SMALL"], + market_caps={"GOOD": 200_000_000_000, "SMALL": 1_000_000_000}) + prices = make_prices(["GOOD", "SMALL"], days=260, asof=asof, trend_pct=0.1) + flow = make_flow(["GOOD", "SMALL"], days=260, asof=asof, + foreign_per_day={"GOOD": 100_000_000, "SMALL": 0}) + ctx = _ctx(master, prices, flow) + + screener = Screener( + gate=HygieneGate(), + score_nodes=[ForeignBuy(), Momentum20()], + weights={"foreign_buy": 1.0, "momentum": 1.0}, + node_params={"foreign_buy": {"window_days": 5}, "momentum": {"window_days": 20}}, + gate_params={**HygieneGate.default_params, "min_listed_days": 0}, + top_n=10, + ) + result = screener.run(ctx) + assert result.survivors_count == 1 # SMALL은 게이트 탈락 + assert result.ranked.index[0] == "GOOD" +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_engine.py -v` +Expected: FAIL — Screener / combine 없음. + +- [ ] **Step 3: registry.py 작성** + +Create `web-backend/stock-lab/app/screener/registry.py`: +```python +"""Registry of node classes (single source of truth for /nodes endpoint).""" + +from .nodes.hygiene import HygieneGate +from .nodes.foreign_buy import ForeignBuy +from .nodes.volume_surge import VolumeSurge +from .nodes.momentum import Momentum20 +from .nodes.high52w import High52WProximity +from .nodes.rs_rating import RsRating +from .nodes.ma_alignment import MaAlignment +from .nodes.vcp_lite import VcpLite + +NODE_REGISTRY: dict[str, type] = { + "foreign_buy": ForeignBuy, + "volume_surge": VolumeSurge, + "momentum": Momentum20, + "high52w": High52WProximity, + "rs_rating": RsRating, + "ma_alignment": MaAlignment, + "vcp_lite": VcpLite, +} + +GATE_REGISTRY: dict[str, type] = { + "hygiene": HygieneGate, +} +``` + +- [ ] **Step 4: engine.py 확장** + +Append to `web-backend/stock-lab/app/screener/engine.py`: +```python + + +# ---- combine + Screener (Phase 2) ---- + +from dataclasses import field as _field + +import pandas as _pd + +from . import position_sizer as _ps + + +def combine(scores: dict, weights: dict) -> pd.Series: + """Weighted average across score nodes. ValueError if all weights = 0.""" + active = {k: w for k, w in weights.items() if w > 0 and k in scores} + if not active: + raise ValueError("no active score nodes (all weights = 0)") + + df = pd.DataFrame({k: scores[k] for k in active}) + w = pd.Series(active) + weighted = (df.fillna(0).multiply(w, axis=1)).sum(axis=1) / w.sum() + return weighted + + +@dataclass +class ScreenerResult: + asof: dt.date + survivors_count: int + scores: dict # node name → pd.Series + weights: dict + ranked: pd.Series # ticker → total_score (sorted desc, head=top_n) + rows: list # list of dicts (for serialization) + warnings: list + + +class Screener: + def __init__(self, gate, score_nodes, weights: dict, node_params: dict, + gate_params: dict, top_n: int = 20, sizer_params: dict | None = None): + self.gate = gate + self.score_nodes = score_nodes + self.weights = weights + self.node_params = node_params + self.gate_params = gate_params + self.top_n = top_n + self.sizer_params = sizer_params or {"atr_window": 14, "atr_stop_mult": 2.0, "rr_ratio": 2.0} + + def run(self, ctx: ScreenContext) -> ScreenerResult: + warnings: list[str] = [] + + survivors = self.gate.filter(ctx, self.gate_params) + if len(survivors) == 0: + raise ValueError("no survivors after hygiene gate") + if len(survivors) < 100: + warnings.append(f"survivors_count={len(survivors)} < 100 — 백분위 정규화 신뢰도 낮음") + + scoped = ctx.restrict(survivors) + scores: dict = {} + for n in self.score_nodes: + w = self.weights.get(n.name, 0) + if w <= 0: + continue + try: + scores[n.name] = n.compute(scoped, self.node_params.get(n.name, {})) + except Exception as e: + warnings.append(f"node '{n.name}' failed: {e}") + scores[n.name] = pd.Series(0.0, index=scoped.master.index) + + total = combine(scores, self.weights) + ranked = total.sort_values(ascending=False).head(self.top_n) + + sizing = _ps.plan_positions(scoped, list(ranked.index), self.sizer_params) + latest_close = scoped.latest_close() + + rows = [] + for rank_idx, ticker in enumerate(ranked.index, start=1): + s = sizing.get(ticker, {}) + row = { + "rank": rank_idx, + "ticker": ticker, + "name": str(scoped.master.loc[ticker, "name"]), + "total_score": float(ranked.loc[ticker]), + "scores": {k: float(v.get(ticker, 0.0)) for k, v in scores.items()}, + "close": int(latest_close.get(ticker, 0)), + "market_cap": int(scoped.master.loc[ticker, "market_cap"] or 0), + "entry_price": s.get("entry_price"), + "stop_price": s.get("stop_price"), + "target_price": s.get("target_price"), + "atr14": s.get("atr14"), + "r_pct": s.get("r_pct"), + } + rows.append(row) + + return ScreenerResult( + asof=ctx.asof, survivors_count=len(survivors), + scores=scores, weights=self.weights, + ranked=ranked, rows=rows, warnings=warnings, + ) +``` + +- [ ] **Step 5: __init__.py 활성화** + +Modify `web-backend/stock-lab/app/screener/__init__.py`: +```python +"""Stock screener — KRX 강세주 분석 노드 기반 보드.""" + +from .engine import Screener, ScreenContext, ScreenerResult +from .registry import NODE_REGISTRY, GATE_REGISTRY + +__all__ = [ + "Screener", "ScreenContext", "ScreenerResult", + "NODE_REGISTRY", "GATE_REGISTRY", +] +``` + +- [ ] **Step 6: Run test, expect pass** + +Run: `pytest app/test_screener_engine.py -v` +Expected: 3 passed. + +- [ ] **Step 7: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/registry.py stock-lab/app/screener/engine.py stock-lab/app/screener/__init__.py stock-lab/app/test_screener_engine.py +git commit -m "feat(stock-lab): Screener 엔진 + combine + ScreenerResult + 노드 레지스트리" +``` + +--- + +## Task 2.3: telegram.py — 메시지 빌더 + +**Files:** +- Create: `web-backend/stock-lab/app/screener/telegram.py` +- Create: `web-backend/stock-lab/app/test_screener_telegram.py` + +- [ ] **Step 1: 테스트** + +Create `web-backend/stock-lab/app/test_screener_telegram.py`: +```python +import datetime as dt +from app.screener.telegram import build_telegram_payload + + +def test_build_payload_includes_top10_and_link(): + rows = [ + { + "rank": i, "ticker": f"00{i:04}", "name": f"종목{i}", + "total_score": 90 - i, + "scores": {"foreign_buy": 80 + i, "volume_surge": 60, "momentum": 70, + "high52w": 75, "rs_rating": 85, "ma_alignment": 80, "vcp_lite": 30}, + "close": 50000, "entry_price": 50250, "stop_price": 48500, + "target_price": 53750, "r_pct": 3.5, + } + for i in range(1, 21) + ] + p = build_telegram_payload( + asof=dt.date(2026, 5, 12), + mode="auto", + survivors_count=612, + top_n=20, + rows=rows, + run_id=42, + ) + assert p["parse_mode"] == "MarkdownV2" + text = p["text"] + assert "2026-05-12" in text + assert "종목1" in text + assert "종목10" in text + assert "종목11" not in text # 본문 1-10만 + assert "run_id=42" in text + + +def test_score_threshold_filters_icons(): + rows = [{ + "rank": 1, "ticker": "A", "name": "A주", + "total_score": 80, + "scores": {"foreign_buy": 90, "volume_surge": 50, "momentum": 70, + "high52w": 30, "rs_rating": 80, "ma_alignment": 80, "vcp_lite": 60}, + "close": 50000, "entry_price": 50250, "stop_price": 48500, + "target_price": 53750, "r_pct": 3.5, + }] + p = build_telegram_payload(dt.date(2026, 5, 12), "auto", 100, 1, rows, run_id=1) + # foreign_buy(90), momentum(70), rs_rating(80), ma_alignment(80) 만 표시 (≥70) + assert "👤외" in p["text"] + assert "🚀모" in p["text"] + assert "💪RS" in p["text"] + assert "📈MA" in p["text"] + assert "⚡거" not in p["text"] + assert "🆙고" not in p["text"] + assert "🌀VCP" not in p["text"] +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_telegram.py -v` + +- [ ] **Step 3: telegram.py 구현** + +Create `web-backend/stock-lab/app/screener/telegram.py`: +```python +"""Telegram payload builder. Caller (agent-office) handles actual delivery.""" + +from __future__ import annotations + +import datetime as dt + +NODE_ICONS = { + "foreign_buy": "👤외", + "volume_surge": "⚡거", + "momentum": "🚀모", + "high52w": "🆙고", + "rs_rating": "💪RS", + "ma_alignment": "📈MA", + "vcp_lite": "🌀VCP", +} + +PAGE_BASE = "https://gahusb.synology.me/stock/screener" + + +def _escape_md(s: str) -> str: + """Minimal MarkdownV2 escape — extend if formatting breaks.""" + for ch in r"\_*[]()~`>#+-=|{}.!": + s = s.replace(ch, "\\" + ch) + return s + + +def _format_won(n: int | None) -> str: + if n is None: + return "-" + return f"{int(n):,}" + + +def build_telegram_payload(asof: dt.date, mode: str, survivors_count: int, + top_n: int, rows: list[dict], run_id: int | None) -> dict: + title = "*KRX 강세주 스크리너*" + header = ( + f"🎯 {title} — {asof.isoformat()} \\({mode}\\)\n" + f"통과 {survivors_count}종 / Top {top_n} / 본문 1\\-10" + ) + + lines = [] + for r in rows[:10]: + icons = " ".join( + NODE_ICONS[name] for name, sc in r["scores"].items() + if sc >= 70 and name in NODE_ICONS + ) + score_str = f"{r['total_score']:.1f}" + lines.append( + f"{r['rank']}\\. *{_escape_md(r['name'])}* `{r['ticker']}` " + f"⭐ {_escape_md(score_str)}\n" + f" {icons}\n" + f" 진입 {_format_won(r.get('entry_price'))} " + f"손절 {_format_won(r.get('stop_price'))} " + f"익절 {_format_won(r.get('target_price'))} " + f"\\(R {_escape_md(f'{r.get('r_pct', 0):.1f}')}%\\)" + ) + + link = f"🔗 전체 결과·11~20위:\n{PAGE_BASE}?run\\_id={run_id}" if run_id else "" + + text = header + "\n\n" + "\n\n".join(lines) + ("\n\n" + link if link else "") + + return { + "chat_target": "default", + "parse_mode": "MarkdownV2", + "text": text, + } +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_telegram.py -v` +Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/telegram.py stock-lab/app/test_screener_telegram.py +git commit -m "feat(stock-lab): telegram.py 메시지 빌더 (Top10 + 아이콘 + 페이지 링크)" +``` + +--- + +# Phase 3 — FastAPI 라우터 + +## Task 3.1: schemas.py (Pydantic) + +**Files:** +- Create: `web-backend/stock-lab/app/screener/schemas.py` + +- [ ] **Step 1: schemas.py 작성** + +Create `web-backend/stock-lab/app/screener/schemas.py`: +```python +from __future__ import annotations +from typing import Literal, Optional +from pydantic import BaseModel, Field + + +class NodeMeta(BaseModel): + name: str + label: str + default_params: dict + param_schema: dict + + +class NodesResponse(BaseModel): + score_nodes: list[NodeMeta] + gate_nodes: list[NodeMeta] + + +class SettingsBody(BaseModel): + weights: dict[str, float] + node_params: dict[str, dict] = Field(default_factory=dict) + gate_params: dict + top_n: int = 20 + rr_ratio: float = 2.0 + atr_window: int = 14 + atr_stop_mult: float = 2.0 + + +class SettingsResponse(SettingsBody): + updated_at: str + + +class RunRequest(BaseModel): + mode: Literal["preview", "manual_save", "auto"] = "preview" + asof: Optional[str] = None + weights: Optional[dict[str, float]] = None + node_params: Optional[dict[str, dict]] = None + gate_params: Optional[dict] = None + top_n: Optional[int] = None + + +class ResultRow(BaseModel): + rank: int + ticker: str + name: str + total_score: float + scores: dict[str, float] + close: int + market_cap: int + entry_price: Optional[int] = None + stop_price: Optional[int] = None + target_price: Optional[int] = None + atr14: Optional[float] = None + r_pct: Optional[float] = None + + +class TelegramPayload(BaseModel): + chat_target: str + parse_mode: str + text: str + + +class RunResponse(BaseModel): + asof: str + mode: str + status: Literal["success", "failed", "skipped_holiday"] + run_id: Optional[int] = None + survivors_count: Optional[int] = None + weights: dict[str, float] + top_n: int + results: list[ResultRow] = Field(default_factory=list) + telegram_payload: Optional[TelegramPayload] = None + warnings: list[str] = Field(default_factory=list) + error: Optional[str] = None + + +class RunSummary(BaseModel): + id: int + asof: str + mode: str + status: str + started_at: str + finished_at: Optional[str] = None + top_n: int + survivors_count: Optional[int] = None + telegram_sent: bool +``` + +- [ ] **Step 2: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/schemas.py +git commit -m "feat(stock-lab): screener Pydantic 스키마" +``` + +--- + +## Task 3.2: router.py — /nodes + /settings + +**Files:** +- Create: `web-backend/stock-lab/app/screener/router.py` +- Create: `web-backend/stock-lab/app/test_screener_router.py` + +- [ ] **Step 1: 테스트 (nodes + settings round-trip)** + +Create `web-backend/stock-lab/app/test_screener_router.py`: +```python +import os +import sqlite3 +import pytest +from fastapi.testclient import TestClient + +from app.main import app # 가정: app.main 에 FastAPI 인스턴스 노출 +from app.screener.schema import ensure_screener_schema + + +@pytest.fixture(autouse=True) +def isolated_db(tmp_path, monkeypatch): + db_path = tmp_path / "screener_router.db" + c = sqlite3.connect(db_path) + ensure_screener_schema(c) + c.close() + monkeypatch.setenv("STOCK_DB_PATH", str(db_path)) + + +@pytest.fixture +def client(): + return TestClient(app) + + +def test_get_nodes_lists_7_score_and_1_gate(client): + r = client.get("/api/stock/screener/nodes") + assert r.status_code == 200 + body = r.json() + assert len(body["score_nodes"]) == 7 + assert len(body["gate_nodes"]) == 1 + assert {n["name"] for n in body["score_nodes"]} == { + "foreign_buy", "volume_surge", "momentum", + "high52w", "rs_rating", "ma_alignment", "vcp_lite", + } + + +def test_settings_get_returns_defaults(client): + r = client.get("/api/stock/screener/settings") + assert r.status_code == 200 + body = r.json() + assert body["weights"]["foreign_buy"] == 1.0 + assert body["top_n"] == 20 + + +def test_settings_put_then_get_round_trip(client): + new_settings = { + "weights": {"foreign_buy": 2.5, "momentum": 1.0, "volume_surge": 1.0, + "high52w": 1.2, "rs_rating": 1.2, "ma_alignment": 1.0, "vcp_lite": 0.8}, + "node_params": {"foreign_buy": {"window_days": 7}}, + "gate_params": {"min_market_cap_won": 100_000_000_000, + "min_avg_value_won": 500_000_000, + "min_listed_days": 60, + "skip_managed": True, "skip_preferred": True, "skip_spac": True, + "skip_halted_days": 3}, + "top_n": 30, + "rr_ratio": 2.5, + "atr_window": 14, + "atr_stop_mult": 2.0, + } + r = client.put("/api/stock/screener/settings", json=new_settings) + assert r.status_code == 200 + r2 = client.get("/api/stock/screener/settings") + body = r2.json() + assert body["weights"]["foreign_buy"] == 2.5 + assert body["top_n"] == 30 +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_router.py -v` +Expected: FAIL (router 없음). + +- [ ] **Step 3: router.py — /nodes + /settings 구현** + +Create `web-backend/stock-lab/app/screener/router.py`: +```python +"""FastAPI router for /api/stock/screener/*""" + +from __future__ import annotations + +import datetime as dt +import json +import os +import sqlite3 +from typing import Optional + +from fastapi import APIRouter, HTTPException + +from . import schemas +from .registry import NODE_REGISTRY, GATE_REGISTRY + + +router = APIRouter(prefix="/api/stock/screener") + + +def _db_path() -> str: + return os.environ.get("STOCK_DB_PATH", "/data/stock.db") + + +def _conn() -> sqlite3.Connection: + return sqlite3.connect(_db_path()) + + +# ---------- /nodes ---------- + +@router.get("/nodes", response_model=schemas.NodesResponse) +def get_nodes(): + score_nodes = [ + schemas.NodeMeta( + name=cls.name, label=cls.label, + default_params=cls.default_params, param_schema=cls.param_schema, + ) + for cls in NODE_REGISTRY.values() + ] + gate_nodes = [ + schemas.NodeMeta( + name=cls.name, label=cls.label, + default_params=cls.default_params, param_schema=cls.param_schema, + ) + for cls in GATE_REGISTRY.values() + ] + return schemas.NodesResponse(score_nodes=score_nodes, gate_nodes=gate_nodes) + + +# ---------- /settings ---------- + +@router.get("/settings", response_model=schemas.SettingsResponse) +def get_settings(): + with _conn() as c: + row = c.execute( + "SELECT weights_json, node_params_json, gate_params_json, " + "top_n, rr_ratio, atr_window, atr_stop_mult, updated_at " + "FROM screener_settings WHERE id=1" + ).fetchone() + if row is None: + raise HTTPException(503, "settings not initialized") + return schemas.SettingsResponse( + weights=json.loads(row[0]), + node_params=json.loads(row[1]), + gate_params=json.loads(row[2]), + top_n=row[3], rr_ratio=row[4], atr_window=row[5], atr_stop_mult=row[6], + updated_at=row[7], + ) + + +@router.put("/settings", response_model=schemas.SettingsResponse) +def put_settings(body: schemas.SettingsBody): + now = dt.datetime.utcnow().isoformat() + with _conn() as c: + c.execute( + """UPDATE screener_settings SET + weights_json=?, node_params_json=?, gate_params_json=?, + top_n=?, rr_ratio=?, atr_window=?, atr_stop_mult=?, updated_at=? + WHERE id=1""", + ( + json.dumps(body.weights), json.dumps(body.node_params), + json.dumps(body.gate_params), + body.top_n, body.rr_ratio, body.atr_window, body.atr_stop_mult, now, + ), + ) + c.commit() + return schemas.SettingsResponse(**body.model_dump(), updated_at=now) +``` + +- [ ] **Step 4: app/main.py 에 router 등록** + +Read `web-backend/stock-lab/app/main.py` 파일 확인. FastAPI 앱이 `app = FastAPI(...)`로 만들어지는 위치에 다음 라인 추가: + +```python +from app.screener.router import router as screener_router +app.include_router(screener_router) +``` + +- [ ] **Step 5: Run test, expect pass** + +Run: `pytest app/test_screener_router.py -v` +Expected: 3 passed. + +- [ ] **Step 6: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/router.py stock-lab/app/main.py stock-lab/app/test_screener_router.py +git commit -m "feat(stock-lab): /nodes + /settings 라우터 + main.py include" +``` + +--- + +## Task 3.3: /run 엔드포인트 + +**Files:** +- Modify: `web-backend/stock-lab/app/screener/router.py` (append /run) +- Modify: `web-backend/stock-lab/app/test_screener_router.py` (append tests) + +- [ ] **Step 1: 테스트 추가** + +Append to `web-backend/stock-lab/app/test_screener_router.py`: +```python +from app.screener._test_fixtures import make_master, make_prices, make_flow + + +def _seed_min(conn, asof_iso="2026-05-12"): + import datetime as dt + now = dt.datetime.utcnow().isoformat() + # 시총 큰 종목 2개 + 작은 종목 1개 + rows = [ + ("BIG1", "큰주식1", "KOSPI", 200_000_000_000, 0, 0, 0, None, now), + ("BIG2", "큰주식2", "KOSPI", 100_000_000_000, 0, 0, 0, None, now), + ("SMALL", "작은주식", "KOSPI", 1_000_000_000, 0, 0, 0, None, now), + ] + for r in rows: + conn.execute("""INSERT INTO krx_master (ticker,name,market,market_cap, + is_managed,is_preferred,is_spac,listed_date,updated_at) + VALUES (?,?,?,?,?,?,?,?,?)""", r) + asof = dt.date(2026, 5, 12) + p = make_prices(["BIG1", "BIG2", "SMALL"], days=260, asof=asof) + f = make_flow(["BIG1", "BIG2", "SMALL"], days=260, asof=asof, + foreign_per_day={"BIG1": 100_000_000, "BIG2": 50_000_000, "SMALL": 0}) + p.to_sql("krx_daily_prices", conn, if_exists="append", index=False) + f.to_sql("krx_flow", conn, if_exists="append", index=False) + conn.commit() + + +def test_run_preview_no_save(client, tmp_path): + db_path = os.environ["STOCK_DB_PATH"] + c = sqlite3.connect(db_path) + _seed_min(c) + c.close() + + r = client.post("/api/stock/screener/run", json={"mode": "preview", "asof": "2026-05-12"}) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "success" + assert body["run_id"] is None + assert body["telegram_payload"] is not None + + c = sqlite3.connect(db_path) + cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0] + assert cnt == 0 + + +def test_run_manual_save_writes_row(client): + db_path = os.environ["STOCK_DB_PATH"] + c = sqlite3.connect(db_path) + _seed_min(c) + c.close() + + r = client.post("/api/stock/screener/run", + json={"mode": "manual_save", "asof": "2026-05-12"}) + assert r.status_code == 200 + assert r.json()["run_id"] is not None + + c = sqlite3.connect(db_path) + cnt = c.execute("SELECT count(*) FROM screener_runs").fetchone()[0] + assert cnt == 1 + + +def test_run_holiday_returns_skipped(): + """공휴일 처리 — TODO Phase 6에서 holidays.json 통합 시 활성화.""" + pytest.skip("holidays.json integration is part of Task 6.x") +``` + +- [ ] **Step 2: Run test, expect failure** + +Run: `pytest app/test_screener_router.py -k run -v` +Expected: FAIL — /run 없음. + +- [ ] **Step 3: /run 구현** + +Append to `web-backend/stock-lab/app/screener/router.py`: +```python + + +# ---------- /run ---------- + +from . import telegram as _tg +from .engine import Screener, ScreenContext +from .registry import NODE_REGISTRY, GATE_REGISTRY + + +def _resolve_asof(asof_str: Optional[str], conn: sqlite3.Connection) -> dt.date: + if asof_str: + return dt.date.fromisoformat(asof_str) + row = conn.execute("SELECT max(date) FROM krx_daily_prices").fetchone() + if not row or row[0] is None: + raise HTTPException(503, "no snapshot available — run /snapshot/refresh first") + return dt.date.fromisoformat(row[0]) + + +def _load_settings(conn) -> dict: + row = conn.execute( + "SELECT weights_json,node_params_json,gate_params_json,top_n," + "rr_ratio,atr_window,atr_stop_mult FROM screener_settings WHERE id=1" + ).fetchone() + return { + "weights": json.loads(row[0]), + "node_params": json.loads(row[1]), + "gate_params": json.loads(row[2]), + "top_n": row[3], + "rr_ratio": row[4], + "atr_window": row[5], + "atr_stop_mult": row[6], + } + + +def _persist_run(conn, asof, mode, weights, node_params, gate_params, top_n, + result, started_at, finished_at) -> int: + cur = conn.execute( + """INSERT INTO screener_runs (asof,mode,status,started_at,finished_at, + weights_json,node_params_json,gate_params_json,top_n,survivors_count,telegram_sent) + VALUES (?,?,?,?,?,?,?,?,?,?,0)""", + (asof.isoformat(), mode, "success", started_at, finished_at, + json.dumps(weights), json.dumps(node_params), json.dumps(gate_params), + top_n, result.survivors_count), + ) + run_id = cur.lastrowid + for row in result.rows: + conn.execute( + """INSERT INTO screener_results (run_id,rank,ticker,name,total_score, + scores_json,close,market_cap,entry_price,stop_price,target_price,atr14) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", + (run_id, row["rank"], row["ticker"], row["name"], row["total_score"], + json.dumps(row["scores"]), row["close"], row["market_cap"], + row["entry_price"], row["stop_price"], row["target_price"], row["atr14"]), + ) + conn.commit() + return run_id + + +@router.post("/run", response_model=schemas.RunResponse) +def post_run(body: schemas.RunRequest): + started_at = dt.datetime.utcnow().isoformat() + with _conn() as c: + try: + asof = _resolve_asof(body.asof, c) + except HTTPException: + raise + + defaults = _load_settings(c) + + if body.mode == "auto": + weights = defaults["weights"] + node_params = defaults["node_params"] + gate_params = defaults["gate_params"] + top_n = defaults["top_n"] + else: + weights = body.weights if body.weights is not None else defaults["weights"] + node_params = body.node_params if body.node_params is not None else defaults["node_params"] + gate_params = body.gate_params if body.gate_params is not None else defaults["gate_params"] + top_n = body.top_n if body.top_n is not None else defaults["top_n"] + + sizer_params = { + "atr_window": defaults["atr_window"], + "atr_stop_mult": defaults["atr_stop_mult"], + "rr_ratio": defaults["rr_ratio"], + } + + ctx = ScreenContext.load(c, asof) + score_nodes = [cls() for name, cls in NODE_REGISTRY.items() if weights.get(name, 0) > 0] + gate = GATE_REGISTRY["hygiene"]() + + try: + screener = Screener( + gate=gate, score_nodes=score_nodes, weights=weights, + node_params=node_params, gate_params=gate_params, + top_n=top_n, sizer_params=sizer_params, + ) + result = screener.run(ctx) + except ValueError as e: + raise HTTPException(422, str(e)) + + finished_at = dt.datetime.utcnow().isoformat() + run_id = None + if body.mode in ("manual_save", "auto"): + run_id = _persist_run(c, asof, body.mode, weights, node_params, gate_params, + top_n, result, started_at, finished_at) + + payload = _tg.build_telegram_payload( + asof=asof, mode=body.mode, survivors_count=result.survivors_count, + top_n=top_n, rows=result.rows, run_id=run_id, + ) + + return schemas.RunResponse( + asof=asof.isoformat(), mode=body.mode, status="success", + run_id=run_id, survivors_count=result.survivors_count, + weights=weights, top_n=top_n, + results=result.rows, + telegram_payload=schemas.TelegramPayload(**payload), + warnings=result.warnings, + ) +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_router.py -k run -v` +Expected: 2 passed (third = skipped). + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py +git commit -m "feat(stock-lab): /run 엔드포인트 — preview/manual_save/auto 모드 매트릭스" +``` + +--- + +## Task 3.4: /snapshot/refresh + /runs + +**Files:** +- Modify: `web-backend/stock-lab/app/screener/router.py` + +- [ ] **Step 1: 라우터 추가** + +Append to `web-backend/stock-lab/app/screener/router.py`: +```python + + +# ---------- /snapshot/refresh ---------- + +from . import snapshot as _snap + + +@router.post("/snapshot/refresh") +def post_snapshot_refresh(asof: Optional[str] = None): + asof_date = dt.date.fromisoformat(asof) if asof else dt.date.today() + if asof_date.weekday() >= 5: + return {"asof": asof_date.isoformat(), "status": "skipped_weekend"} + with _conn() as c: + summary = _snap.refresh_daily(c, asof_date) + return summary + + +# ---------- /runs ---------- + +@router.get("/runs", response_model=list[schemas.RunSummary]) +def list_runs(limit: int = 30): + with _conn() as c: + rows = c.execute( + "SELECT id,asof,mode,status,started_at,finished_at,top_n," + "survivors_count,telegram_sent FROM screener_runs " + "ORDER BY asof DESC, id DESC LIMIT ?", (limit,), + ).fetchall() + return [ + schemas.RunSummary( + id=r[0], asof=r[1], mode=r[2], status=r[3], + started_at=r[4], finished_at=r[5], top_n=r[6], + survivors_count=r[7], telegram_sent=bool(r[8]), + ) + for r in rows + ] + + +@router.get("/runs/{run_id}") +def get_run(run_id: int): + with _conn() as c: + meta = c.execute( + "SELECT id,asof,mode,status,started_at,finished_at,top_n," + "survivors_count,telegram_sent,weights_json,node_params_json,gate_params_json " + "FROM screener_runs WHERE id=?", + (run_id,), + ).fetchone() + if not meta: + raise HTTPException(404, "run not found") + rows = c.execute( + "SELECT rank,ticker,name,total_score,scores_json,close,market_cap," + "entry_price,stop_price,target_price,atr14 " + "FROM screener_results WHERE run_id=? ORDER BY rank", + (run_id,), + ).fetchall() + + return { + "meta": { + "id": meta[0], "asof": meta[1], "mode": meta[2], "status": meta[3], + "started_at": meta[4], "finished_at": meta[5], "top_n": meta[6], + "survivors_count": meta[7], "telegram_sent": bool(meta[8]), + "weights": json.loads(meta[9]), + "node_params": json.loads(meta[10]), + "gate_params": json.loads(meta[11]), + }, + "results": [ + { + "rank": r[0], "ticker": r[1], "name": r[2], + "total_score": r[3], "scores": json.loads(r[4]), + "close": r[5], "market_cap": r[6], + "entry_price": r[7], "stop_price": r[8], "target_price": r[9], + "atr14": r[10], + } + for r in rows + ], + } +``` + +- [ ] **Step 2: 테스트 추가** + +Append to `web-backend/stock-lab/app/test_screener_router.py`: +```python + + +def test_runs_list_and_detail(client): + db_path = os.environ["STOCK_DB_PATH"] + c = sqlite3.connect(db_path) + _seed_min(c) + c.close() + + saved = client.post( + "/api/stock/screener/run", + json={"mode": "manual_save", "asof": "2026-05-12"}, + ).json() + run_id = saved["run_id"] + + list_r = client.get("/api/stock/screener/runs?limit=5") + assert list_r.status_code == 200 + assert any(r["id"] == run_id for r in list_r.json()) + + detail = client.get(f"/api/stock/screener/runs/{run_id}") + assert detail.status_code == 200 + assert detail.json()["meta"]["id"] == run_id + assert isinstance(detail.json()["results"], list) +``` + +- [ ] **Step 3: Run test, expect pass** + +Run: `pytest app/test_screener_router.py -v` +Expected: 모든 테스트 pass. + +- [ ] **Step 4: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py +git commit -m "feat(stock-lab): /snapshot/refresh + /runs 리스트·상세 라우터" +``` + +--- + +## Task 3.5: 공휴일 처리 + asof 자동 추정 + skipped_holiday + +**Files:** +- Modify: `web-backend/stock-lab/app/screener/router.py` +- Modify: `web-backend/stock-lab/app/test_screener_router.py` (holiday 테스트 활성화) + +- [ ] **Step 1: holidays.json 로딩 헬퍼 추가** + +Append to `web-backend/stock-lab/app/screener/router.py` (파일 상단의 imports 다음에): +```python +import json as _json +import pathlib as _pathlib + +_HOLIDAYS_CACHE: set[str] | None = None + + +def _holidays() -> set[str]: + global _HOLIDAYS_CACHE + if _HOLIDAYS_CACHE is None: + path = _pathlib.Path(__file__).resolve().parent.parent / "holidays.json" + try: + with path.open(encoding="utf-8") as f: + data = _json.load(f) + _HOLIDAYS_CACHE = set(data) if isinstance(data, list) else set(data.keys()) + except FileNotFoundError: + _HOLIDAYS_CACHE = set() + return _HOLIDAYS_CACHE + + +def _is_holiday(d: dt.date) -> bool: + return d.weekday() >= 5 or d.isoformat() in _holidays() +``` + +- [ ] **Step 2: post_run에 holiday 분기 추가** + +`post_run` 함수의 `started_at = dt.datetime.utcnow().isoformat()` 다음, `with _conn() as c:` 진입 직후: +```python + try: + asof = _resolve_asof(body.asof, c) + except HTTPException: + raise + + # Skipped holiday handling for mode='auto' + if body.mode == "auto" and _is_holiday(asof): + return schemas.RunResponse( + asof=asof.isoformat(), mode="auto", status="skipped_holiday", + run_id=None, survivors_count=None, + weights={}, top_n=0, + results=[], telegram_payload=None, + warnings=[f"{asof.isoformat()} is a holiday — skipped"], + ) +``` + +- [ ] **Step 3: 테스트 활성화** + +Replace the skipped test in `app/test_screener_router.py`: +```python +def test_run_holiday_returns_skipped(client): + # 토요일 + r = client.post("/api/stock/screener/run", + json={"mode": "auto", "asof": "2026-05-09"}) # 2026-05-09 is Saturday + assert r.status_code == 200 + assert r.json()["status"] == "skipped_holiday" +``` + +- [ ] **Step 4: Run test, expect pass** + +Run: `pytest app/test_screener_router.py -v` + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add stock-lab/app/screener/router.py stock-lab/app/test_screener_router.py +git commit -m "feat(stock-lab): /run mode=auto 공휴일·주말 skipped_holiday 처리" +``` + +--- + +# Phase 4 — 프론트엔드 (web-ui) + +**Phase 전체에서 commit은 `web-ui` 디렉토리에서 수행.** + +## Task 4.1: api.js 헬퍼 7개 추가 + +**Files:** +- Modify: `web-ui/src/api.js` + +- [ ] **Step 1: 헬퍼 추가** + +Read `web-ui/src/api.js` 끝부분에 다음 7개 헬퍼 추가: +```javascript +// ---- Stock Screener ---- +export const getScreenerNodes = () => apiGet ('/api/stock/screener/nodes'); +export const getScreenerSettings = () => apiGet ('/api/stock/screener/settings'); +export const saveScreenerSettings = (body) => apiPut ('/api/stock/screener/settings', body); +export const runScreener = (body) => apiPost('/api/stock/screener/run', body); +export const refreshScreenerSnap = () => apiPost('/api/stock/screener/snapshot/refresh'); +export const listScreenerRuns = (limit = 30) => apiGet (`/api/stock/screener/runs?limit=${limit}`); +export const getScreenerRun = (id) => apiGet (`/api/stock/screener/runs/${id}`); +``` + +- [ ] **Step 2: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/api.js +git commit -m "feat(stock): screener API 헬퍼 7개" +``` + +--- + +## Task 4.2: 라우트 + 네비게이션 등록 + +**Files:** +- Modify: `web-ui/src/routes.jsx` +- Modify: `web-ui/src/Router.jsx` + +- [ ] **Step 1: routes.jsx 수정** + +Read `web-ui/src/routes.jsx` 검토 후 라우트 배열에 추가 (기존 `/stock` 항목 근처): +```jsx +{ path: '/stock/screener', label: '스크리너', element: lazy(() => import('./pages/stock/screener/Screener')) }, +``` + +(파일의 정확한 패턴은 기존 항목을 따라가세요. lazy import 사용 여부 등.) + +- [ ] **Step 2: Router.jsx 수정** + +Read `web-ui/src/Router.jsx` 검토. 라우트 추가 패턴이 별도로 있으면 그곳에 동일하게 `/stock/screener` 등록. + +- [ ] **Step 3: 빌드 확인** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +npm run dev +``` + +브라우저에서 `/stock/screener` 접근. (페이지는 아직 비어 있어도 무방, 라우트만 매칭 확인) + +- [ ] **Step 4: Commit (페이지 빈 컴포넌트 임시 작성 후)** + +임시 placeholder 페이지를 위해 4.3을 먼저 작성하고 함께 commit. **이 단계는 4.3 commit에 통합.** + +--- + +## Task 4.3: Screener.jsx 페이지 골격 + 빈 컴포넌트 + +**Files:** +- Create: `web-ui/src/pages/stock/screener/Screener.jsx` +- Create: `web-ui/src/pages/stock/screener/Screener.css` + +- [ ] **Step 1: 페이지 골격** + +Create `web-ui/src/pages/stock/screener/Screener.jsx`: +```jsx +import React from 'react'; +import { Link } from 'react-router-dom'; +import './Screener.css'; + +import { useScreenerMeta } from './hooks/useScreenerMeta'; +import { useScreenerSettings } from './hooks/useScreenerSettings'; +import { useScreenerRun } from './hooks/useScreenerRun'; +import { useScreenerHistory } from './hooks/useScreenerHistory'; + +import GatePanel from './components/GatePanel'; +import NodePanel from './components/NodePanel'; +import GlobalControls from './components/GlobalControls'; +import ResultTable from './components/ResultTable'; +import TelegramPreview from './components/TelegramPreview'; +import RunHistoryList from './components/RunHistoryList'; + +export default function Screener() { + const { meta, loading: metaLoading } = useScreenerMeta(); + const { settings, dirty, setLocal, save } = useScreenerSettings(); + const { result, running, runPreview, runSave } = useScreenerRun(); + const { runs, runs_loading, selectRun, selectedRun } = useScreenerHistory(); + + const activeResult = selectedRun || result; + + if (metaLoading || !meta || !settings) { + return
로딩 중…
; + } + + return ( +
+
+
+

스크리너

+

+ 최근 자동 잡: {runs?.find(r => r.mode === 'auto')?.asof ?? '-'} + · 분석 기준일: {activeResult?.asof ?? settings.asof ?? '-'} +

+
+ +
+ +
+ + +
+ + +
+ + +
+
+ ); +} +``` + +- [ ] **Step 2: Screener.css 골격** + +Create `web-ui/src/pages/stock/screener/Screener.css`: +```css +.screener-page { + padding: 24px; + color: var(--text, #e5e7eb); + background: var(--bg, #0b0f17); + min-height: 100vh; +} + +.screener-header { + display: flex; + align-items: flex-end; + justify-content: space-between; + margin-bottom: 24px; +} + +.screener-header h1 { + font-size: 28px; + margin: 0 0 4px 0; +} + +.screener-header .meta { + color: #9ca3af; + font-size: 13px; + margin: 0; +} + +.screener-header nav a { + margin-left: 12px; + color: #9ca3af; + text-decoration: none; +} + +.screener-grid { + display: grid; + grid-template-columns: 320px 1fr 280px; + gap: 24px; +} + +@media (max-width: 1023px) { + .screener-grid { grid-template-columns: 1fr; } +} + +.screener-loading { padding: 80px; text-align: center; color: #9ca3af; } +``` + +- [ ] **Step 3: 빈 hooks 4개 stub 작성 (다음 task에서 채움)** + +Create `web-ui/src/pages/stock/screener/hooks/useScreenerMeta.js`: +```javascript +import { useEffect, useState } from 'react'; +import { getScreenerNodes } from '../../../../api'; + +export function useScreenerMeta() { + const [meta, setMeta] = useState(null); + const [loading, setLoading] = useState(true); + useEffect(() => { + getScreenerNodes().then((m) => { setMeta(m); setLoading(false); }); + }, []); + return { meta, loading }; +} +``` + +Create `useScreenerSettings.js`: +```javascript +import { useEffect, useState } from 'react'; +import { getScreenerSettings, saveScreenerSettings } from '../../../../api'; + +export function useScreenerSettings() { + const [remote, setRemote] = useState(null); + const [local, setLocal] = useState(null); + + useEffect(() => { + getScreenerSettings().then((s) => { setRemote(s); setLocal(s); }); + }, []); + + const dirty = remote && local && JSON.stringify(remote) !== JSON.stringify(local); + + async function save() { + if (!local) return; + const saved = await saveScreenerSettings({ + weights: local.weights, node_params: local.node_params, gate_params: local.gate_params, + top_n: local.top_n, rr_ratio: local.rr_ratio, + atr_window: local.atr_window, atr_stop_mult: local.atr_stop_mult, + }); + setRemote(saved); + setLocal(saved); + } + + return { settings: local, dirty, setLocal, save }; +} +``` + +Create `useScreenerRun.js`: +```javascript +import { useState } from 'react'; +import { runScreener } from '../../../../api'; + +export function useScreenerRun() { + const [result, setResult] = useState(null); + const [running, setRunning] = useState(false); + + async function call(mode, settings) { + setRunning(true); + try { + const body = { + mode, + weights: settings.weights, + node_params: settings.node_params, + gate_params: settings.gate_params, + top_n: settings.top_n, + }; + const r = await runScreener(body); + setResult(r); + return r; + } finally { + setRunning(false); + } + } + + return { + result, running, + runPreview: (s) => call('preview', s), + runSave: (s) => call('manual_save', s), + }; +} +``` + +Create `useScreenerHistory.js`: +```javascript +import { useEffect, useState } from 'react'; +import { listScreenerRuns, getScreenerRun } from '../../../../api'; + +export function useScreenerHistory() { + const [runs, setRuns] = useState([]); + const [loading, setLoading] = useState(true); + const [selectedRun, setSelectedRun] = useState(null); + + useEffect(() => { + listScreenerRuns(30).then((r) => { setRuns(r); setLoading(false); }); + }, []); + + async function selectRun(id) { + if (!id) { setSelectedRun(null); return; } + const detail = await getScreenerRun(id); + setSelectedRun({ + asof: detail.meta.asof, + mode: detail.meta.mode, + status: detail.meta.status, + run_id: detail.meta.id, + survivors_count: detail.meta.survivors_count, + weights: detail.meta.weights, + top_n: detail.meta.top_n, + results: detail.results, + telegram_payload: null, // history detail은 메시지 미리보기 비포함 (MVP) + warnings: [], + meta: detail.meta, + }); + } + + return { runs, runs_loading: loading, selectedRun, selectRun }; +} +``` + +- [ ] **Step 4: 빈 컴포넌트 5개 stub** + +Create `web-ui/src/pages/stock/screener/components/GatePanel.jsx`: +```jsx +export default function GatePanel({ meta, value, onChange }) { + return

{meta?.label ?? '게이트'}

TODO: 게이트 파라미터 폼

; +} +``` + +Create `NodePanel.jsx`: +```jsx +export default function NodePanel({ meta, weights, params, onWeights, onParams }) { + return

점수 노드 ({meta.length})

TODO: 노드별 카드

; +} +``` + +Create `GlobalControls.jsx`: +```jsx +export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) { + return ( +
+ + + +
+ ); +} +``` + +Create `ResultTable.jsx`: +```jsx +export default function ResultTable({ result }) { + if (!result) return

아직 결과 없음. "지금 실행"을 눌러보세요.

; + return ( +
+

Top {result.top_n} · 통과 {result.survivors_count}

+ + + + + + {(result.results || []).map((r) => ( + + + + + + + + + ))} + +
#종목총점진입손절익절R%
{r.rank}{r.name} ({r.ticker}){r.total_score?.toFixed?.(1)}{r.entry_price?.toLocaleString?.()}{r.stop_price?.toLocaleString?.()}{r.target_price?.toLocaleString?.()}{r.r_pct?.toFixed?.(1)}
+
+ ); +} +``` + +Create `TelegramPreview.jsx`: +```jsx +export default function TelegramPreview({ payload }) { + if (!payload) return null; + return ( +
+

텔레그램 미리보기

+ 
{payload.text}
+
+ ); +} +``` + +Create `RunHistoryList.jsx`: +```jsx +export default function RunHistoryList({ runs, loading, onSelect, selectedId }) { + if (loading) return

로딩…

; + return ( +
+

최근 실행

+
    + {(runs || []).map((r) => ( +
  • onSelect(r.id)}> + {r.asof} · {r.mode} +
    • + ))} +
+
+ ); +} +``` + +추가 css to Screener.css: +```css +.screener-card { + background: #0f1623; + border: 1px solid #1f2937; + border-radius: 8px; + padding: 16px; + margin-bottom: 16px; +} +.screener-card h3 { margin: 0 0 12px 0; font-size: 15px; } +``` + +- [ ] **Step 5: 빌드 확인** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +npm run dev +``` + +브라우저: `/stock/screener` 접속, 페이지 로드 확인. (백엔드 미동작이어도 hook 호출 실패 메시지 정도가 보임) + +- [ ] **Step 6: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/pages/stock/screener src/routes.jsx src/Router.jsx +git commit -m "feat(stock): /stock/screener 페이지 골격 + hooks 4개 + 컴포넌트 stub 6개" +``` + +--- + +## Task 4.4: NodeCard — param_schema 기반 자동 폼 + +**Files:** +- Create: `web-ui/src/pages/stock/screener/components/NodeCard.jsx` +- Modify: `web-ui/src/pages/stock/screener/components/NodePanel.jsx` + +- [ ] **Step 1: NodeCard 작성** + +Create `web-ui/src/pages/stock/screener/components/NodeCard.jsx`: +```jsx +import React from 'react'; + +export default function NodeCard({ meta, weight, params, onWeightChange, onParamsChange }) { + const enabled = (weight ?? 0) > 0; + + return ( +
+
+ +
+
+
+ 가중치 + onWeightChange(parseFloat(e.target.value))} + style={{ flex: 1 }} + /> + {(weight ?? 0).toFixed(1)} +
+ {Object.entries(meta.param_schema?.properties || {}).map(([key, prop]) => ( + onParamsChange({ ...params, [key]: v })} + /> + ))} +
+
+ ); +} + +function ParamRow({ paramKey, prop, value, disabled, onChange }) { + const type = prop.type; + if (type === 'integer' || type === 'number') { + return ( +
+ {paramKey} + onChange(type === 'integer' ? parseInt(e.target.value, 10) : parseFloat(e.target.value))} + style={{ width: 80 }} + /> +
+ ); + } + if (type === 'boolean') { + return ( +
+ +
+ ); + } + // object/array는 MVP에서 read-only JSON 표시 (RsRating의 weights 등) + return ( +
+ {paramKey}: {JSON.stringify(value)} +
+ ); +} +``` + +- [ ] **Step 2: NodePanel 갱신** + +Replace `web-ui/src/pages/stock/screener/components/NodePanel.jsx`: +```jsx +import NodeCard from './NodeCard'; + +export default function NodePanel({ meta, weights, params, onWeights, onParams }) { + return ( +
+

점수 노드 ({meta.length})

+
+ {meta.map((m) => ( + onWeights({ ...weights, [m.name]: w })} + onParamsChange={(p) => onParams({ ...params, [m.name]: p })} + /> + ))} +
+
+ ); +} +``` + +- [ ] **Step 3: CSS 추가** + +Append to `Screener.css`: +```css +.node-card { + background: #0a0f1a; + border: 1px solid #1f2937; + border-radius: 6px; + padding: 10px; + font-size: 13px; +} +.node-card-header { font-weight: 500; margin-bottom: 6px; } +.weight-row, .param-row { display: flex; align-items: center; gap: 6px; margin-top: 6px; } +``` + +- [ ] **Step 4: 빌드 확인** + +`npm run dev`로 노드 7개 카드가 슬라이더·체크박스·숫자 입력으로 렌더링되는지 확인. + +- [ ] **Step 5: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/pages/stock/screener/components/NodeCard.jsx src/pages/stock/screener/components/NodePanel.jsx src/pages/stock/screener/Screener.css +git commit -m "feat(stock): NodeCard 자동 폼 (param_schema 기반) + NodePanel 통합" +``` + +--- + +## Task 4.5: GatePanel + GlobalControls 본구현 + +**Files:** +- Modify: `web-ui/src/pages/stock/screener/components/GatePanel.jsx` +- Modify: `web-ui/src/pages/stock/screener/components/GlobalControls.jsx` + +- [ ] **Step 1: GatePanel 갱신** + +Replace `GatePanel.jsx`: +```jsx +export default function GatePanel({ meta, value, onChange }) { + if (!meta) return null; + const props = meta.param_schema?.properties || {}; + return ( +
+

{meta.label}

+

+ 통과 조건 — 통과한 종목만 점수 노드에 전달 +

+ {Object.entries(props).map(([key, prop]) => ( + onChange({ ...value, [key]: v })} /> + ))} +
+ ); +} + +function GateField({ paramKey, prop, value, onChange }) { + if (prop.type === 'integer') { + return ( +
+ + onChange(parseInt(e.target.value, 10))} + style={{ flex: 1 }} /> +
+ ); + } + if (prop.type === 'boolean') { + return ( +
+ +
+ ); + } + return null; +} +``` + +- [ ] **Step 2: GlobalControls 갱신** + +Replace `GlobalControls.jsx`: +```jsx +export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) { + return ( +
+

실행 옵션

+
+ + setSettings({ ...settings, top_n: parseInt(e.target.value, 10) })} + min={5} max={100} style={{ width: 80 }} /> +
+
+ + setSettings({ ...settings, atr_window: parseInt(e.target.value, 10) })} + min={5} max={50} style={{ width: 80 }} /> +
+
+ + setSettings({ ...settings, atr_stop_mult: parseFloat(e.target.value) })} + min={0.5} max={5} style={{ width: 80 }} /> +
+
+ + setSettings({ ...settings, rr_ratio: parseFloat(e.target.value) })} + min={1} max={10} style={{ width: 80 }} /> +
+ + + +
+ ); +} +``` + +- [ ] **Step 3: 빌드 확인** + +`npm run dev`로 게이트 폼·글로벌 옵션 렌더링 확인. + +- [ ] **Step 4: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/pages/stock/screener/components/GatePanel.jsx src/pages/stock/screener/components/GlobalControls.jsx +git commit -m "feat(stock): GatePanel 자동 폼 + GlobalControls (TopN/ATR/RR + 3버튼)" +``` + +--- + +## Task 4.6: ResultTable 본구현 (ScoreChips) + +**Files:** +- Create: `web-ui/src/pages/stock/screener/components/ScoreChips.jsx` +- Modify: `web-ui/src/pages/stock/screener/components/ResultTable.jsx` + +- [ ] **Step 1: ScoreChips** + +Create `ScoreChips.jsx`: +```jsx +const NODE_ICONS = { + foreign_buy: { icon: '👤', label: '외국인' }, + volume_surge: { icon: '⚡', label: '거래량' }, + momentum: { icon: '🚀', label: '모멘텀' }, + high52w: { icon: '🆙', label: '52w고' }, + rs_rating: { icon: '💪', label: 'RS' }, + ma_alignment: { icon: '📈', label: '정배열' }, + vcp_lite: { icon: '🌀', label: 'VCP' }, +}; + +export default function ScoreChips({ scores }) { + return ( +
+ {Object.entries(scores || {}).map(([name, s]) => { + const meta = NODE_ICONS[name]; + if (!meta) return null; + const active = s >= 70; + return ( + + {meta.icon}{Math.round(s)} + + ); + })} +
+ ); +} +``` + +- [ ] **Step 2: ResultTable 갱신** + +Replace `ResultTable.jsx`: +```jsx +import ScoreChips from './ScoreChips'; + +export default function ResultTable({ result }) { + if (!result) { + return ( +
+

아직 결과 없음. "지금 실행"을 눌러보세요.

+
+ ); + } + if (result.warnings?.length) { + // 경고 배너만 노출하고 표는 계속 렌더 + } + + return ( +
+
+

+ Top {result.top_n} · 통과 {result.survivors_count} · {result.asof} +

+ {result.warnings?.length > 0 && ( +
+ ⚠ {result.warnings.join(' · ')} +
+ )} +
+ +
+ + + + + + + + + {(result.results || []).map((r) => ( + + + + + + + + + + + ))} + +
#종목총점노드진입손절익절R%
{r.rank}{r.name}
{r.ticker}		{r.total_score?.toFixed(1)}{r.entry_price?.toLocaleString?.()}{r.stop_price?.toLocaleString?.()}{r.target_price?.toLocaleString?.()}{r.r_pct?.toFixed?.(1)}
+
+
+ ); +} +``` + +- [ ] **Step 3: CSS 추가** + +Append to `Screener.css`: +```css +.screener-table { + width: 100%; + font-size: 13px; + border-collapse: collapse; +} +.screener-table th { text-align: left; padding: 8px; background: #0a0f1a; color: #9ca3af; font-weight: 500; border-bottom: 1px solid #1f2937; } +.screener-table td { padding: 8px; border-bottom: 1px solid #1a2230; vertical-align: middle; } +.screener-table tr:hover { background: #0a0f1a; } +``` + +- [ ] **Step 4: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/pages/stock/screener/components/ScoreChips.jsx src/pages/stock/screener/components/ResultTable.jsx src/pages/stock/screener/Screener.css +git commit -m "feat(stock): ResultTable 본구현 + ScoreChips (노드 칩 + 70점 강조)" +``` + +--- + +## Task 4.7: CLAUDE.md API 표 갱신 + Stock 페이지에 스크리너 링크 + +**Files:** +- Modify: `web-ui/CLAUDE.md` +- Modify: `web-ui/src/pages/stock/Stock.jsx` (또는 StockTrade.jsx — 스크리너 링크 위치) + +- [ ] **Step 1: CLAUDE.md API 표에 7행 추가** + +Read `web-ui/CLAUDE.md`의 "API 엔드포인트 목록" 표를 찾고, 마지막에 다음 7행 추가: + +```markdown +| 스크리너 | GET | `/api/stock/screener/nodes` | +| 스크리너 | GET/PUT | `/api/stock/screener/settings` | +| 스크리너 | POST | `/api/stock/screener/run` — body: `{ mode, asof?, weights?, ... }` | +| 스크리너 | POST | `/api/stock/screener/snapshot/refresh` | +| 스크리너 | GET | `/api/stock/screener/runs?limit=N` | +| 스크리너 | GET | `/api/stock/screener/runs/:id` | +``` + +또한 "페이지 구조" 표에 `/stock/screener` 행 추가: +```markdown +| `/stock/screener` | `Screener` | 노드 기반 강세주 스크리너 (점수 노드 7 + 위생 게이트 + ATR 포지션 사이저) | +``` + +- [ ] **Step 2: Stock 페이지에 스크리너 링크** + +Read `web-ui/src/pages/stock/Stock.jsx` 상단 헤더 또는 nav 영역에 다음 링크 추가 (기존 `` 패턴 옆에): + +```jsx +스크리너 +``` + +- [ ] **Step 3: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add CLAUDE.md src/pages/stock/Stock.jsx +git commit -m "docs(stock): CLAUDE.md 스크리너 API 표 추가 + Stock 페이지 링크" +``` + +--- + +## Task 4.8: 모바일 레이아웃 점검 + 작은 화면 적층 + +**Files:** +- Modify: `web-ui/src/pages/stock/screener/Screener.css` + +- [ ] **Step 1: 모바일 CSS 추가** + +Append to `Screener.css`: +```css +@media (max-width: 1023px) { + .screener-page { padding: 16px; } + .screener-header { flex-direction: column; align-items: flex-start; gap: 8px; } + .screener-grid { gap: 16px; } + .screener-left { order: 1; } + .screener-center { order: 2; } + .screener-right { order: 3; } + .screener-table { font-size: 12px; } + .screener-table th, .screener-table td { padding: 6px 4px; } +} + +@media (max-width: 640px) { + .screener-page { padding: 12px; } + .screener-card { padding: 12px; } +} +``` + +- [ ] **Step 2: 브라우저 모바일 폭(<768px) 확인** + +`npm run dev`로 Chrome DevTools에서 iPhone 사이즈 토글 → 카드 세로 적층, 표 가로 스크롤 확인. + +- [ ] **Step 3: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +git add src/pages/stock/screener/Screener.css +git commit -m "style(stock): 스크리너 모바일 적층 + 표 가로 스크롤" +``` + +--- + +## Task 4.9: ESLint + 빌드 검증 + +**Files:** (변경 없음, 검증만) + +- [ ] **Step 1: lint** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +npm run lint +``` + +오류가 있으면 fix 후 재실행. + +- [ ] **Step 2: build** + +```bash +npm run build +``` + +`dist/` 생성 확인. + +- [ ] **Step 3: fix가 있었으면 commit** + +```bash +git add -A +git commit -m "chore(stock): 스크리너 lint/build 정리" +``` + +(변경 없으면 step 건너뜀) + +--- + +# Phase 5 — agent-office 자동 잡 통합 + +## Task 5.1: agent-office에 stock_screener 잡 추가 + +**Files:** (`web-backend/agent-office/` 구조에 따라 위치 결정 — 기존 realestate-lab/뉴스 알림 잡 패턴을 그대로 따라가세요) + +agent-office가 어떤 형태로 잡을 정의하는지(YAML config, Python module, cron entry 등)를 먼저 확인해야 합니다. 가장 가능성 있는 패턴 3가지: + +**Patten A — 잡 module 추가**: `web-backend/agent-office/app/jobs/stock_screener.py` +**Pattern B — config 항목 추가**: `web-backend/agent-office/jobs.yaml`에 항목 1개 +**Pattern C — agent 명령**: `agent-office` 내부 명령 핸들러에 `stock_agent screener` 추가 + +이 task는 **반드시 기존 코드를 먼저 읽어 패턴을 확인한 뒤** 그 패턴에 맞춰 잡을 추가하세요. + +- [ ] **Step 1: 기존 자동 잡 패턴 식별 (로컬 파일 시스템 탐색)** + +로컬 파일을 직접 읽어 패턴 파악 (Grep/Glob 도구 사용 — Docker 불필요): +- `web-backend/agent-office/app/` 디렉토리 트리 확인 +- "telegram" 키워드 grep — 어떤 모듈에서 메시지를 보내는지 +- "schedule|cron|APScheduler|tick" grep — 잡을 어떻게 등록하는지 +- realestate-lab 매칭 알림이 가장 유사한 기존 잡이므로 그 코드를 통째로 읽기 + +가장 비슷한 기존 잡 1개의 코드를 통째로 읽고, 그 구조를 그대로 따라갑니다. + +- [ ] **Step 2: stock_screener 잡 작성** + +기존 패턴이 함수 기반(예: `def realestate_match_job(): ...`)이라면 다음을 같은 위치에 추가: + +```python +"""Stock screener auto job — 평일 16:30 KST.""" + +import logging +import os +from datetime import datetime, timezone, timedelta + +import httpx + +log = logging.getLogger(__name__) + +STOCK_LAB_URL = os.environ.get("STOCK_LAB_URL", "http://stock-lab:18500") + + +async def run_stock_screener_job(): + """Refresh KRX snapshot, run screener auto, dispatch telegram.""" + today = datetime.now(timezone(timedelta(hours=9))).date() + asof = today.isoformat() + + try: + async with httpx.AsyncClient(timeout=180) as client: + # 1. Snapshot refresh (실패해도 다음 단계 계속) + try: + await client.post(f"{STOCK_LAB_URL}/api/stock/screener/snapshot/refresh", + json={"asof": asof}) + except Exception as e: + log.warning("snapshot refresh failed: %s", e) + + # 2. Run auto + r = await client.post( + f"{STOCK_LAB_URL}/api/stock/screener/run", + json={"mode": "auto", "asof": asof}, + ) + r.raise_for_status() + body = r.json() + + status = body.get("status") + if status == "skipped_holiday": + log.info("stock_screener skipped (holiday/weekend) %s", asof) + return + if status != "success": + log.error("stock_screener failed: %s", body.get("error")) + _notify_self("stock_screener 잡 실패: " + (body.get("error") or "unknown")) + return + + payload = body.get("telegram_payload") + if payload: + # 기존 agent-office 텔레그램 전송 함수 호출 — 이름은 기존 코드 따름 + from .telegram import send_message # 예시 + await send_message(payload["text"], parse_mode=payload["parse_mode"]) + log.info("stock_screener sent telegram for %s", asof) + + except Exception as e: + log.exception("stock_screener job crashed") + _notify_self(f"stock_screener 잡 예외: {e}") + + +def _notify_self(msg: str): + """운영자에게 알림 — 기존 패턴 차용.""" + # TODO: 기존 운영 알림 채널 함수로 연결 + log.warning(msg) +``` + +기존 스케줄러에 등록 (APScheduler·cron entry·기존 main loop에 추가): +- 매일 한국 시간 16:30 트리거 +- 평일만 (스케줄러가 weekday 필터 지원하면 거기서, 아니면 잡 내부에서 분기) + +- [ ] **Step 3: 환경변수 점검** + +`web-backend/agent-office/.env`(또는 docker-compose.yml의 environment 섹션)에서 stock-lab으로의 내부 URL 확인. 보통 도커 네트워크에서 서비스명으로 접근: +``` +STOCK_LAB_URL=http://stock-lab:18500 +``` + +- [ ] **Step 4: 단위 테스트 (로컬 venv, 함수 직접 호출)** + +가능하면 `run_stock_screener_job`에 대한 단위 테스트를 추가 (httpx mock으로 stock-lab 응답 stub). 기존 agent-office가 venv로 로컬 실행 가능하면 그것으로 실행. 그렇지 않으면 함수 시그니처와 로직 코드 리뷰만 하고 통과. + +- [ ] **Step 5: 배포 후 NAS에서 실전 발동 확인 (Task 6.3에서)** + +agent-office 컨테이너 재시작·로그 확인은 NAS SSH에서 수행 (Task 6.3 참조). 로컬에서는 코드만 작성하고 commit. + +- [ ] **Step 6: Commit** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git add agent-office/ +git commit -m "feat(agent-office): stock screener 평일 16:30 KST 자동 잡 + 텔레그램 전송" +``` + +--- + +# Phase 6 — 초기 백필 · 수동 검증 · 배포 + +## Task 6.1: 초기 KRX 데이터 백필 (2년치) + +**Files:** (스크립트 1회용 실행, 코드 신규 없음) + +- [ ] **Step 1: NAS SSH로 접속 후 백필 (운영 DB 대상)** + +```bash +ssh user@gahusb.synology.me +cd /volume1/docker/webpage +sudo docker compose exec stock-lab python -c " +import datetime as dt, sqlite3 +from app.screener.snapshot import backfill +start = dt.date(2024, 5, 12) +end = dt.date(2026, 5, 11) +conn = sqlite3.connect('/data/stock.db') +results = backfill(conn, start, end) +ok = sum(1 for r in results if 'error' not in r) +print(f'backfill done: {ok}/{len(results)} days') +" +``` + +⚠ 시간이 오래 걸립니다 (~30분~수 시간, 2,700종목 × ~500 거래일). NAS 안에서 백그라운드(`nohup`/`screen`) 실행 권장. 끝나기를 기다리고 결과 로그 확인. + +- [ ] **Step 2: DB 행 수 확인 (NAS)** + +```bash +sudo docker compose exec stock-lab sqlite3 /data/stock.db " +SELECT count(*) AS prices FROM krx_daily_prices; +SELECT count(*) AS flows FROM krx_flow; +SELECT count(*) AS master FROM krx_master; +" +``` + +prices·flow 둘 다 100만+ 행 정도 나와야 정상. + +- [ ] **Step 3: (선택) DB 백업 (NAS)** + +```bash +sudo docker compose exec stock-lab sh -c "cp /data/stock.db /data/stock-pre-screener-backup.db" +``` + +--- + +## Task 6.2: 수동 검증 — Top 20이 합리적인가 + +- [ ] **Step 1: 프론트에서 1회 미리보기 실행** + +`/stock/screener`로 이동 → "지금 실행 (미리보기)" 클릭. 결과 Top 20 표 확인. + +- [ ] **Step 2: 합리성 체크리스트 (사용자 눈으로)** + +다음 중 다수가 "예"여야 정상: +- [ ] 상위 종목들이 최근 1~3개월간 외국인 누적 매수 + 거래량 증가 + 우상향 추세인가? +- [ ] 작전주·관리종목·우선주가 포함되어 있지 않은가? +- [ ] 시총이 모두 500억 이상인가? +- [ ] 각 종목의 ATR 기반 손절가가 진입가 대비 -2~-7% 범위인가? (R% 컬럼) +- [ ] 노드 칩(👤외 ⚡거 등)이 종목마다 0~5개 정도 점등되는가? (모두 점등 또는 모두 비점등은 의심) + +이상치가 보이면 노드 알고리즘 또는 데이터 캐시 문제 → 개별 디버그. + +- [ ] **Step 3: 모바일 화면 확인** + +iPhone 사이즈로 좌측 패널 접힘·결과 표 가로 스크롤 OK 확인. + +--- + +## Task 6.3: 자동 잡 실전 발동 & 텔레그램 도착 확인 + +- [ ] **Step 1: NAS SSH에서 자동 잡 다음 평일 16:30 모니터링** + +```bash +ssh user@gahusb.synology.me +cd /volume1/docker/webpage +sudo docker compose logs -f agent-office | grep stock_screener +``` + +- [ ] **Step 2: 텔레그램 메시지 도착 확인** + +- 헤더에 분석 기준일·통과 종목 수 표시되는가 +- Top 10이 한 메시지에 다 들어왔는가 (텔레그램 4096자 한도 위반 없음) +- 노드 아이콘이 70점 이상에서만 점등됐는가 +- 진입가/손절가/익절가/R% 모두 표시됐는가 +- 페이지 링크 클릭 시 해당 `run_id` 결과가 프론트에 로드되는가 + +- [ ] **Step 3: 운영 상황에서 누적 검증 (1주일)** + +매일 텔레그램 도착 + `/stock/screener` 히스토리에 자동 잡 row가 매일 1개씩 추가되는지 확인. + +--- + +## 최종 commit & 배포 + +- [ ] **Step 1: 프론트 배포** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-ui +npm run release:nas +``` + +Z 드라이브 마운트 필수 (CLAUDE.md 참조). + +- [ ] **Step 2: 백엔드 배포 (이미 docker compose up이면 자동, 아니면 git push로 webhook 트리거)** + +```bash +cd C:\Users\jaeoh\Desktop\workspace\web-backend +git push +``` + +Gitea webhook → NAS deployer가 stock-lab + agent-office 컨테이너 자동 재시작. + +- [ ] **Step 3: 운영 사이트에서 1회 확인** + +`https://gahusb.synology.me/stock/screener` 접근 → 데이터 정상 로드 확인. + +--- + +# 끝