feat(stock-lab): HygieneGate — 위생 필터 (시총/거래대금/우선주/관리종목)

stock-lab/app/screener/nodes/hygiene.py

@@ -0,0 +1,81 @@
+"""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)

stock-lab/app/test_screener_nodes_hygiene.py

@@ -0,0 +1,46 @@
+import datetime as dt
+
+import pandas as pd
+
+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=pd.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"]