Files

gahusb f7323a5b72 docs(plan): Stock Screener Board MVP 구현 plan

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) <noreply@anthropic.com>

2026-05-12 03:48:20 +09:00

137 KiB

Raw Blame History

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로 직접 실행한다.

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:

"""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 끝에 다시 활성화.

수정:

"""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 설치

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)

python -c "from pykrx import stock; print(stock.get_market_ticker_list('20260512', market='KOSPI')[:5])"

Expected: 5개 KOSPI 종목 코드 출력. 실패 시 IP 차단/네트워크 확인.

Step 5: Commit

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:

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

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:

"""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

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)를 찾으세요. 그 함수 마지막에 다음을 추가:

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

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:

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

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:

"""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

pytest app/test_screener_snapshot.py -v

Expected: 2 passed.

Step 5: Commit

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:

"""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:

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

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:

"""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

pytest app/test_screener_context.py -v

Expected: 2 passed.

Step 5: Commit

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:

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

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:

"""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

pytest app/test_screener_nodes_base.py -v

Expected: 3 passed.

Step 5: Commit

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:

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:

"""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

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:

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:

"""외국인 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

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:

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:

"""거래량 급증 — 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

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:

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:

"""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

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:

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:

"""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

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:

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:

"""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

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:

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:

"""이평선 정배열 점수 — 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

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:

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:

"""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

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:

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:

"""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

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:

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:

"""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:



# ---- 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:

"""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

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:

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:

"""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

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:

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

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:

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:

"""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(...)로 만들어지는 위치에 다음 라인 추가:

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

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:

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:



# ---------- /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

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:



# ---------- /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:



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

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 다음에):

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: 진입 직후:

        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:

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

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개 헬퍼 추가:

// ---- 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

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 항목 근처):

{ 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: 빌드 확인

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:

import React from 'react';
import { Link } from 'react-router-dom';
import './Screener.css';

import { useScreenerMeta }     from './hooks/useScreenerMeta';
import { useScreenerSettings } from './hooks/useScreenerSettings';
import { useScreenerRun }      from './hooks/useScreenerRun';
import { useScreenerHistory }  from './hooks/useScreenerHistory';

import GatePanel        from './components/GatePanel';
import NodePanel        from './components/NodePanel';
import GlobalControls   from './components/GlobalControls';
import ResultTable      from './components/ResultTable';
import TelegramPreview  from './components/TelegramPreview';
import RunHistoryList   from './components/RunHistoryList';

export default function Screener() {
  const { meta, loading: metaLoading } = useScreenerMeta();
  const { settings, dirty, setLocal, save }   = useScreenerSettings();
  const { result, running, runPreview, runSave } = useScreenerRun();
  const { runs, runs_loading, selectRun, selectedRun } = useScreenerHistory();

  const activeResult = selectedRun || result;

  if (metaLoading || !meta || !settings) {
    return <div className="screener-loading">로딩 중…</div>;
  }

  return (
    <div className="screener-page">
      <header className="screener-header">
        <div>
          <h1>스크리너</h1>
          <p className="meta">
            최근 자동 잡: {runs?.find(r => r.mode === 'auto')?.asof ?? '-'}
            · 분석 기준일: {activeResult?.asof ?? settings.asof ?? '-'}
          </p>
        </div>
        <nav>
          <Link to="/stock">시장</Link>
          <Link to="/stock/trade">트레이드</Link>
        </nav>
      </header>

      <div className="screener-grid">
        <aside className="screener-left">
          <GatePanel meta={meta.gate_nodes[0]} value={settings.gate_params} onChange={(p) => setLocal({...settings, gate_params: p})} />
          <NodePanel meta={meta.score_nodes} weights={settings.weights} params={settings.node_params}
                     onWeights={(w) => setLocal({...settings, weights: w})}
                     onParams={(p)  => setLocal({...settings, node_params: p})} />
          <GlobalControls settings={settings} setSettings={setLocal}
                          onRun={() => runPreview(settings)}
                          onSave={() => runSave(settings)}
                          onPersist={save}
                          dirty={dirty}
                          running={running} />
        </aside>

        <main className="screener-center">
          <ResultTable result={activeResult} />
          <TelegramPreview payload={activeResult?.telegram_payload} />
        </main>

        <aside className="screener-right">
          <RunHistoryList runs={runs} loading={runs_loading} onSelect={selectRun}
                          selectedId={selectedRun?.meta?.id} />
        </aside>
      </div>
    </div>
  );
}

Step 2: Screener.css 골격

Create web-ui/src/pages/stock/screener/Screener.css:

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

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:

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:

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:

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:

export default function GatePanel({ meta, value, onChange }) {
  return <section className="screener-card"><h3>{meta?.label ?? '게이트'}</h3><p style={{fontSize: 12, color:'#9ca3af'}}>TODO: 게이트 파라미터 폼</p></section>;
}

Create NodePanel.jsx:

export default function NodePanel({ meta, weights, params, onWeights, onParams }) {
  return <section className="screener-card"><h3>점수 노드 ({meta.length})</h3><p style={{fontSize: 12, color:'#9ca3af'}}>TODO: 노드별 카드</p></section>;
}

Create GlobalControls.jsx:

export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
  return (
    <section className="screener-card">
      <button onClick={onRun} disabled={running}>{running ? '실행 중…' : '지금 실행 (미리보기)'}</button>
      <button onClick={onSave} disabled={running} style={{ marginTop: 8 }}>스냅샷 저장</button>
      <button onClick={onPersist} disabled={!dirty} style={{ marginTop: 8 }}>설정 저장</button>
    </section>
  );
}

Create ResultTable.jsx:

export default function ResultTable({ result }) {
  if (!result) return <section className="screener-card"><p style={{color:'#9ca3af'}}>아직 결과 없음. "지금 실행"을 눌러보세요.</p></section>;
  return (
    <section className="screener-card">
      <h3>Top {result.top_n} · 통과 {result.survivors_count}</h3>
      <table style={{ width: '100%', fontSize: 13 }}>
        <thead>
          <tr><th>#</th><th>종목</th><th>총점</th><th>진입</th><th>손절</th><th>익절</th><th>R%</th></tr>
        </thead>
        <tbody>
          {(result.results || []).map((r) => (
            <tr key={r.ticker}>
              <td>{r.rank}</td><td>{r.name} ({r.ticker})</td>
              <td>{r.total_score?.toFixed?.(1)}</td>
              <td>{r.entry_price?.toLocaleString?.()}</td>
              <td>{r.stop_price?.toLocaleString?.()}</td>
              <td>{r.target_price?.toLocaleString?.()}</td>
              <td>{r.r_pct?.toFixed?.(1)}</td>
            </tr>
          ))}
        </tbody>
      </table>
    </section>
  );
}

Create TelegramPreview.jsx:

export default function TelegramPreview({ payload }) {
  if (!payload) return null;
  return (
    <section className="screener-card">
      <h3>텔레그램 미리보기</h3>
      <pre style={{whiteSpace:'pre-wrap', fontFamily:'monospace', fontSize:12}}>{payload.text}</pre>
    </section>
  );
}

Create RunHistoryList.jsx:

export default function RunHistoryList({ runs, loading, onSelect, selectedId }) {
  if (loading) return <section className="screener-card"><p>로딩…</p></section>;
  return (
    <section className="screener-card">
      <h3>최근 실행</h3>
      <ul style={{listStyle:'none', padding:0, margin:0, fontSize:13}}>
        {(runs || []).map((r) => (
          <li key={r.id} style={{padding:'6px 0', borderBottom:'1px solid #1f2937', cursor:'pointer',
                                  color: selectedId === r.id ? '#fbbf24' : '#e5e7eb'}}
              onClick={() => onSelect(r.id)}>
            {r.asof} · {r.mode}
          </li>
        ))}
      </ul>
    </section>
  );
}

추가 css to Screener.css:

.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: 빌드 확인

cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run dev

브라우저: /stock/screener 접속, 페이지 로드 확인. (백엔드 미동작이어도 hook 호출 실패 메시지 정도가 보임)

Step 6: Commit

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:

import React from 'react';

export default function NodeCard({ meta, weight, params, onWeightChange, onParamsChange }) {
  const enabled = (weight ?? 0) > 0;

  return (
    <div className="node-card" style={{ opacity: enabled ? 1 : 0.6 }}>
      <div className="node-card-header">
        <label style={{ display: 'flex', alignItems: 'center', gap: 8 }}>
          <input
            type="checkbox"
            checked={enabled}
            onChange={(e) => onWeightChange(e.target.checked ? (weight || 1) : 0)}
          />
          <span>{meta.label}</span>
        </label>
      </div>
      <div className="node-card-body">
        <div className="weight-row">
          <span style={{ width: 50, fontSize: 12, color: '#9ca3af' }}>가중치</span>
          <input
            type="range" min="0" max="3" step="0.1"
            value={weight ?? 0}
            disabled={!enabled}
            onChange={(e) => onWeightChange(parseFloat(e.target.value))}
            style={{ flex: 1 }}
          />
          <span style={{ width: 32, textAlign: 'right', fontSize: 12 }}>{(weight ?? 0).toFixed(1)}</span>
        </div>
        {Object.entries(meta.param_schema?.properties || {}).map(([key, prop]) => (
          <ParamRow
            key={key}
            paramKey={key}
            prop={prop}
            value={params?.[key] ?? meta.default_params?.[key]}
            disabled={!enabled}
            onChange={(v) => onParamsChange({ ...params, [key]: v })}
          />
        ))}
      </div>
    </div>
  );
}

function ParamRow({ paramKey, prop, value, disabled, onChange }) {
  const type = prop.type;
  if (type === 'integer' || type === 'number') {
    return (
      <div className="param-row">
        <span style={{ width: 100, fontSize: 12 }}>{paramKey}</span>
        <input
          type="number"
          min={prop.minimum} max={prop.maximum}
          step={type === 'integer' ? 1 : 0.1}
          value={value ?? ''}
          disabled={disabled}
          onChange={(e) => onChange(type === 'integer' ? parseInt(e.target.value, 10) : parseFloat(e.target.value))}
          style={{ width: 80 }}
        />
      </div>
    );
  }
  if (type === 'boolean') {
    return (
      <div className="param-row">
        <label>
          <input type="checkbox" checked={!!value} disabled={disabled}
                 onChange={(e) => onChange(e.target.checked)} />
          <span style={{ marginLeft: 6, fontSize: 12 }}>{paramKey}</span>
        </label>
      </div>
    );
  }
  // object/array는 MVP에서 read-only JSON 표시 (RsRating의 weights 등)
  return (
    <div className="param-row" style={{ fontSize: 11, color: '#9ca3af' }}>
      {paramKey}: <code>{JSON.stringify(value)}</code>
    </div>
  );
}

Step 2: NodePanel 갱신

Replace web-ui/src/pages/stock/screener/components/NodePanel.jsx:

import NodeCard from './NodeCard';

export default function NodePanel({ meta, weights, params, onWeights, onParams }) {
  return (
    <section className="screener-card">
      <h3>점수 노드 ({meta.length})</h3>
      <div style={{ display: 'flex', flexDirection: 'column', gap: 12 }}>
        {meta.map((m) => (
          <NodeCard
            key={m.name}
            meta={m}
            weight={weights[m.name]}
            params={params[m.name]}
            onWeightChange={(w) => onWeights({ ...weights, [m.name]: w })}
            onParamsChange={(p) => onParams({ ...params, [m.name]: p })}
          />
        ))}
      </div>
    </section>
  );
}

Step 3: CSS 추가

Append to Screener.css:

.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

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:

export default function GatePanel({ meta, value, onChange }) {
  if (!meta) return null;
  const props = meta.param_schema?.properties || {};
  return (
    <section className="screener-card">
      <h3>{meta.label}</h3>
      <p style={{ fontSize: 11, color: '#9ca3af', marginTop: 0 }}>
        통과 조건 — 통과한 종목만 점수 노드에 전달
      </p>
      {Object.entries(props).map(([key, prop]) => (
        <GateField key={key} paramKey={key} prop={prop}
                   value={value?.[key] ?? meta.default_params?.[key]}
                   onChange={(v) => onChange({ ...value, [key]: v })} />
      ))}
    </section>
  );
}

function GateField({ paramKey, prop, value, onChange }) {
  if (prop.type === 'integer') {
    return (
      <div className="param-row">
        <label style={{ width: 160, fontSize: 12 }}>{paramKey}</label>
        <input type="number" value={value ?? ''}
               min={prop.minimum} onChange={(e) => onChange(parseInt(e.target.value, 10))}
               style={{ flex: 1 }} />
      </div>
    );
  }
  if (prop.type === 'boolean') {
    return (
      <div className="param-row">
        <label>
          <input type="checkbox" checked={!!value} onChange={(e) => onChange(e.target.checked)} />
          <span style={{ marginLeft: 6, fontSize: 12 }}>{paramKey}</span>
        </label>
      </div>
    );
  }
  return null;
}

Step 2: GlobalControls 갱신

Replace GlobalControls.jsx:

export default function GlobalControls({ settings, setSettings, onRun, onSave, onPersist, dirty, running }) {
  return (
    <section className="screener-card">
      <h3>실행 옵션</h3>
      <div className="param-row">
        <label style={{ width: 80, fontSize: 12 }}>Top N</label>
        <input type="number" value={settings.top_n}
               onChange={(e) => setSettings({ ...settings, top_n: parseInt(e.target.value, 10) })}
               min={5} max={100} style={{ width: 80 }} />
      </div>
      <div className="param-row">
        <label style={{ width: 80, fontSize: 12 }}>ATR window</label>
        <input type="number" value={settings.atr_window}
               onChange={(e) => setSettings({ ...settings, atr_window: parseInt(e.target.value, 10) })}
               min={5} max={50} style={{ width: 80 }} />
      </div>
      <div className="param-row">
        <label style={{ width: 80, fontSize: 12 }}>손절 ×ATR</label>
        <input type="number" value={settings.atr_stop_mult} step={0.1}
               onChange={(e) => setSettings({ ...settings, atr_stop_mult: parseFloat(e.target.value) })}
               min={0.5} max={5} style={{ width: 80 }} />
      </div>
      <div className="param-row">
        <label style={{ width: 80, fontSize: 12 }}>R:R 비율</label>
        <input type="number" value={settings.rr_ratio} step={0.1}
               onChange={(e) => setSettings({ ...settings, rr_ratio: parseFloat(e.target.value) })}
               min={1} max={10} style={{ width: 80 }} />
      </div>
      <button onClick={onRun} disabled={running}
              style={{ marginTop: 16, width: '100%', padding: 10, background: '#fbbf24', color: '#0b0f17', border: 'none', borderRadius: 6, fontWeight: 600 }}>
        {running ? '실행 중…' : '지금 실행 (미리보기)'}
      </button>
      <button onClick={onSave} disabled={running}
              style={{ marginTop: 8, width: '100%', padding: 8 }}>
        스냅샷 저장
      </button>
      <button onClick={onPersist} disabled={!dirty}
              style={{ marginTop: 8, width: '100%', padding: 8, opacity: dirty ? 1 : 0.5 }}>
        설정 저장 (디폴트 갱신)
      </button>
    </section>
  );
}

Step 3: 빌드 확인

npm run dev로 게이트 폼·글로벌 옵션 렌더링 확인.

Step 4: Commit

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:

const NODE_ICONS = {
  foreign_buy:  { icon: '👤', label: '외국인' },
  volume_surge: { icon: '⚡', label: '거래량' },
  momentum:     { icon: '🚀', label: '모멘텀' },
  high52w:      { icon: '🆙', label: '52w고' },
  rs_rating:    { icon: '💪', label: 'RS' },
  ma_alignment: { icon: '📈', label: '정배열' },
  vcp_lite:     { icon: '🌀', label: 'VCP' },
};

export default function ScoreChips({ scores }) {
  return (
    <div style={{ display: 'flex', gap: 4, flexWrap: 'wrap' }}>
      {Object.entries(scores || {}).map(([name, s]) => {
        const meta = NODE_ICONS[name];
        if (!meta) return null;
        const active = s >= 70;
        return (
          <span key={name}
                title={`${meta.label}: ${s.toFixed?.(0) ?? s}`}
                style={{
                  padding: '2px 6px', borderRadius: 4, fontSize: 11,
                  background: active ? '#fbbf24' : '#1f2937',
                  color: active ? '#0b0f17' : '#9ca3af',
                }}>
            {meta.icon}{Math.round(s)}
          </span>
        );
      })}
    </div>
  );
}

Step 2: ResultTable 갱신

Replace ResultTable.jsx:

import ScoreChips from './ScoreChips';

export default function ResultTable({ result }) {
  if (!result) {
    return (
      <section className="screener-card">
        <p style={{ color: '#9ca3af' }}>아직 결과 없음. "지금 실행"을 눌러보세요.</p>
      </section>
    );
  }
  if (result.warnings?.length) {
    // 경고 배너만 노출하고 표는 계속 렌더
  }

  return (
    <section className="screener-card">
      <div style={{ display: 'flex', justifyContent: 'space-between', alignItems: 'center' }}>
        <h3 style={{ margin: 0 }}>
          Top {result.top_n} · 통과 {result.survivors_count} · {result.asof}
        </h3>
        {result.warnings?.length > 0 && (
          <div style={{
            background: '#7c2d12', color: '#fde68a', padding: '4px 10px',
            borderRadius: 4, fontSize: 12,
          }}>
            ⚠ {result.warnings.join(' · ')}
          </div>
        )}
      </div>

      <div style={{ overflowX: 'auto', marginTop: 12 }}>
        <table className="screener-table">
          <thead>
            <tr>
              <th>#</th><th>종목</th><th>총점</th><th>노드</th>
              <th>진입</th><th>손절</th><th>익절</th><th>R%</th>
            </tr>
          </thead>
          <tbody>
            {(result.results || []).map((r) => (
              <tr key={r.ticker}>
                <td>{r.rank}</td>
                <td>{r.name}<br /><span style={{ fontSize: 11, color: '#9ca3af' }}>{r.ticker}</span></td>
                <td style={{ fontWeight: 600 }}>{r.total_score?.toFixed(1)}</td>
                <td><ScoreChips scores={r.scores} /></td>
                <td>{r.entry_price?.toLocaleString?.()}</td>
                <td>{r.stop_price?.toLocaleString?.()}</td>
                <td>{r.target_price?.toLocaleString?.()}</td>
                <td>{r.r_pct?.toFixed?.(1)}</td>
              </tr>
            ))}
          </tbody>
        </table>
      </div>
    </section>
  );
}

Step 3: CSS 추가

Append to Screener.css:

.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

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행 추가:

| 스크리너 | 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 행 추가:

| `/stock/screener` | `Screener` | 노드 기반 강세주 스크리너 (점수 노드 7 + 위생 게이트 + ATR 포지션 사이저) |

Step 2: Stock 페이지에 스크리너 링크

Read web-ui/src/pages/stock/Stock.jsx 상단 헤더 또는 nav 영역에 다음 링크 추가 (기존 <Link to="/stock/trade"> 패턴 옆에):

<Link to="/stock/screener">스크리너</Link>

Step 3: Commit

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:

@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

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

cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run lint

오류가 있으면 fix 후 재실행.

Step 2: build

npm run build

dist/ 생성 확인.

Step 3: fix가 있었으면 commit

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(): ...)이라면 다음을 같은 위치에 추가:

"""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

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 대상)

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)

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)

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 모니터링

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: 프론트 배포

cd C:\Users\jaeoh\Desktop\workspace\web-ui
npm run release:nas

Z 드라이브 마운트 필수 (CLAUDE.md 참조).

Step 2: 백엔드 배포 (이미 docker compose up이면 자동, 아니면 git push로 webhook 트리거)

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 접근 → 데이터 정상 로드 확인.

137 KiB Raw Blame History Unescape Escape

Stock Screener Board Implementation Plan

개요 — 6 Phase

로컬 개발 환경 설정 (1회 — 이후 모든 pytest는 이 venv에서)

Phase 0 — 백엔드 기반

Task 0.1: pykrx 의존성 + 스키마 함수 골격

Task 0.2: 스키마 마이그레이션 (SQL DDL 7테이블)

Task 0.3: snapshot.py — KRX 캐시 갱신

Task 0.4: ScreenContext.load() — DB → DataFrame

Phase 1 — 노드 8개 (TDD)

Task 1.1: ScoreNode / GateNode 추상 + percentile_rank 유틸

Task 1.2: HygieneGate

Task 1.3: ForeignBuy

Task 1.4: VolumeSurge

Task 1.5: Momentum20

Task 1.6: High52WProximity

Task 1.7: RsRating

Task 1.8: MaAlignment

Task 1.9: VcpLite

Phase 2 — 엔진 · 포지션 사이저 · 텔레그램 · 레지스트리

Task 2.1: position_sizer.py — ATR Wilder + entry/stop/target

Task 2.2: registry.py + engine.Screener + combine

Task 2.3: telegram.py — 메시지 빌더

Phase 3 — FastAPI 라우터

Task 3.1: schemas.py (Pydantic)

Task 3.2: router.py — /nodes + /settings

Task 3.3: /run 엔드포인트

Task 3.4: /snapshot/refresh + /runs

Task 3.5: 공휴일 처리 + asof 자동 추정 + skipped_holiday

Phase 4 — 프론트엔드 (web-ui)

Task 4.1: api.js 헬퍼 7개 추가

Task 4.2: 라우트 + 네비게이션 등록

Task 4.3: Screener.jsx 페이지 골격 + 빈 컴포넌트

Task 4.4: NodeCard — param_schema 기반 자동 폼

Task 4.5: GatePanel + GlobalControls 본구현

Task 4.6: ResultTable 본구현 (ScoreChips)

Task 4.7: CLAUDE.md API 표 갱신 + Stock 페이지에 스크리너 링크

Task 4.8: 모바일 레이아웃 점검 + 작은 화면 적층

Task 4.9: ESLint + 빌드 검증

Phase 5 — agent-office 자동 잡 통합

Task 5.1: agent-office에 stock_screener 잡 추가

Phase 6 — 초기 백필 · 수동 검증 · 배포

Task 6.1: 초기 KRX 데이터 백필 (2년치)

Task 6.2: 수동 검증 — Top 20이 합리적인가

Task 6.3: 자동 잡 실전 발동 & 텔레그램 도착 확인

최종 commit & 배포

끝

137 KiB

Raw Blame History