diff --git a/stock-lab/app/screener/nodes/hygiene.py b/stock-lab/app/screener/nodes/hygiene.py new file mode 100644 index 0000000..066fafe --- /dev/null +++ b/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) diff --git a/stock-lab/app/test_screener_nodes_hygiene.py b/stock-lab/app/test_screener_nodes_hygiene.py new file mode 100644 index 0000000..b43db2e --- /dev/null +++ b/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"]