diff --git a/lps/services/pipeline/core.py b/lps/services/pipeline/core.py
new file mode 100644
index 0000000..169aa13
--- /dev/null
+++ b/lps/services/pipeline/core.py
@@ -0,0 +1,66 @@
+"""최저가 코어 파이프라인.
+
+검색 결과(NormalizedProduct[]) → 필터 → 이상치 제거 → 정렬 → top-N 최저가.
+각 STAGE 의 in/out 건수를 기록해 관측성을 확보한다(레퍼런스 pipeline_log 패턴 계승).
+
+AI 유사도 판정(같은 상품인지)은 키 준비 시 이 사이(이상치 제거 뒤, top-N 앞)에 끼운다 —
+지금은 슬롯만 비워두고 규칙 매칭을 하지 않는다(고정 파서 취약성 회피).
+"""
+
+from services.search.contract import NormalizedProduct
+from services.pipeline.filters import keep_only_mall, filter_out_malls, filter_by_price_band
+from services.pipeline.outliers import remove_price_outliers
+
+
+def run_price_pipeline(
+    products: list[NormalizedProduct],
+    *,
+    base_price: int | None = None,
+    keep_mall: str | None = None,
+    banned_malls=(),
+    band_tolerance: float = 0.7,
+    remove_outliers: bool = True,
+    top_n: int = 5,
+) -> dict:
+    stages: list[dict] = []
+    cur = list(products)
+
+    def stage(name: str, before: list, after: list):
+        stages.append({"stage": name, "in": len(before), "out": len(after)})
+
+    # 1) mall 필터 (쿠팡 단독이면 사실상 no-op, 네이버 결합 시 오픈마켓 정리용)
+    if keep_mall:
+        before = cur
+        cur = keep_only_mall(cur, keep_mall)
+        stage("keep_mall", before, cur)
+    elif banned_malls:
+        before = cur
+        cur = filter_out_malls(cur, banned_malls)
+        stage("filter_out_malls", before, cur)
+
+    # 2) 가격 밴드 (요청 현재가 기준 targeted 컷)
+    if base_price:
+        before = cur
+        cur = filter_by_price_band(cur, base_price, band_tolerance)
+        stage("price_band", before, cur)
+
+    # 3) 이상치 제거 (IQR)
+    if remove_outliers:
+        before = cur
+        cur, _ = remove_price_outliers(cur)
+        stage("outlier", before, cur)
+
+    # (AI 유사도 판정 슬롯 — 키 준비 시 여기)
+
+    # 4) 최저가순 정렬 + top-N
+    ranked = sorted(cur, key=lambda p: p.price)
+    top = ranked[:top_n]
+    stage("top_n", ranked, top)
+
+    return {
+        "total_found": len(products),
+        "kept": len(ranked),
+        "lowest": top[0].model_dump() if top else None,
+        "top": [p.model_dump() for p in top],
+        "stages": stages,
+    }
diff --git a/lps/services/pipeline/filters.py b/lps/services/pipeline/filters.py
new file mode 100644
index 0000000..00b4969
--- /dev/null
+++ b/lps/services/pipeline/filters.py
@@ -0,0 +1,26 @@
+"""가격 파이프라인 순수 필터(부작용 없음). 레퍼런스의 mall/price-band 필터를 NormalizedProduct 로 이식."""
+
+from services.search.contract import NormalizedProduct
+
+
+def keep_only_mall(products: list[NormalizedProduct], mall_name: str) -> list[NormalizedProduct]:
+    target = (mall_name or "").strip().lower()
+    return [p for p in products if (p.mall_name or "").strip().lower() == target]
+
+
+def filter_out_malls(products: list[NormalizedProduct], banned) -> list[NormalizedProduct]:
+    """banned mall(문자열/리스트) 제거. 네이버 결과에서 오픈마켓을 걷어낼 때 쓴다(쿠팡 단독이면 no-op)."""
+    names = [banned] if isinstance(banned, str) else list(banned or [])
+    ban = {m.strip().lower() for m in names if m}
+    if not ban:
+        return list(products)
+    return [p for p in products if (p.mall_name or "").strip().lower() not in ban]
+
+
+def filter_by_price_band(products: list[NormalizedProduct], base_price: int, tolerance: float = 0.7) -> list[NormalizedProduct]:
+    """기준가 ±(tolerance×100)% 밴드만 남긴다. base_price 가 없거나 0 이하면 무필터 통과.
+    요청의 현재가(price)를 기준으로 '엉뚱한 저가/고가'(액세서리·묶음 등)를 targeted 하게 컷."""
+    if not base_price or base_price <= 0:
+        return list(products)
+    lo, hi = base_price * (1 - tolerance), base_price * (1 + tolerance)
+    return [p for p in products if p.price is not None and lo <= p.price <= hi]
diff --git a/lps/services/pipeline/outliers.py b/lps/services/pipeline/outliers.py
new file mode 100644
index 0000000..354bd52
--- /dev/null
+++ b/lps/services/pipeline/outliers.py
@@ -0,0 +1,32 @@
+"""가격 이상치 제거.
+
+레퍼런스는 반복 z-score(정규분포 가정)를 썼으나, 가격 분포는 한쪽으로 치우친(로그정규) 경우가
+많아 취약하다. 대신 분포 가정이 없는 **IQR(사분위 범위) 방식**을 쓴다(stdlib 만, numpy 불필요).
+목적: '엉뚱한 초저가(액세서리/오매칭)·초고가(묶음)'가 최저가 산정을 오염시키는 것을 막는다.
+"""
+
+import statistics
+
+from services.search.contract import NormalizedProduct
+
+
+def remove_price_outliers(
+    products: list[NormalizedProduct], k: float = 1.5, min_items: int = 4
+) -> tuple[list[NormalizedProduct], list[NormalizedProduct]]:
+    """IQR 기반 이상치 제거. [Q1 - k·IQR, Q3 + k·IQR] 밖을 제거.
+    데이터가 min_items 미만이면 통계가 무의미하므로 그대로 통과. (kept, removed) 반환."""
+    prices = [p.price for p in products if p.price and p.price > 0]
+    if len(prices) < min_items:
+        return list(products), []
+
+    q1, _, q3 = statistics.quantiles(prices, n=4)
+    iqr = q3 - q1
+    lo, hi = q1 - k * iqr, q3 + k * iqr
+
+    kept, removed = [], []
+    for p in products:
+        if p.price is None or lo <= p.price <= hi:
+            kept.append(p)
+        else:
+            removed.append(p)
+    return kept, removed
diff --git a/lps/tests/test_pipeline.py b/lps/tests/test_pipeline.py
new file mode 100644
index 0000000..6281b2a
--- /dev/null
+++ b/lps/tests/test_pipeline.py
@@ -0,0 +1,64 @@
+"""코어 파이프라인 테스트 — 필터·IQR 이상치·top-N 최저가 (결정론적, 네트워크 불필요)."""
+
+from pathlib import Path
+
+from services.search.contract import NormalizedProduct
+from services.search.coupang.parser import parse_search_html
+from services.pipeline.filters import filter_by_price_band, filter_out_malls
+from services.pipeline.outliers import remove_price_outliers
+from services.pipeline.core import run_price_pipeline
+
+FIXTURE = Path(__file__).parent / "fixtures" / "coupang_search.html"
+
+
+def _p(price, name="p", mall="쿠팡"):
+    return NormalizedProduct(source="coupang", name=name, price=price, mall_name=mall)
+
+
+def test_top_n_is_lowest_price_sorted():
+    prods = [_p(x) for x in [3000, 1000, 2000, 5000, 4000]]
+    r = run_price_pipeline(prods, remove_outliers=False, top_n=3)
+    assert [p["price"] for p in r["top"]] == [1000, 2000, 3000]
+    assert r["lowest"]["price"] == 1000
+
+
+def test_outlier_removes_extreme_low_and_high():
+    normal = [_p(x) for x in [1000, 1010, 1020, 1030, 1040, 1050, 1060, 1070, 1080, 1090]]
+    kept, removed = remove_price_outliers(normal + [_p(5), _p(500000)])
+    prices_removed = {p.price for p in removed}
+    assert 5 in prices_removed and 500000 in prices_removed
+    assert all(1000 <= p.price <= 1090 for p in kept)
+
+
+def test_price_band_filter():
+    prods = [_p(x) for x in [8000, 10000, 12000, 30000, 1000]]
+    # 기준가 10000, ±70% → [3000, 17000] 만 통과
+    out = filter_by_price_band(prods, base_price=10000, tolerance=0.7)
+    assert {p.price for p in out} == {8000, 10000, 12000}
+
+
+def test_filter_out_malls():
+    prods = [_p(1000, mall="쿠팡"), _p(2000, mall="G마켓"), _p(3000, mall="옥션")]
+    out = filter_out_malls(prods, ["G마켓", "옥션"])
+    assert {p.mall_name for p in out} == {"쿠팡"}
+
+
+def test_empty_input_is_safe():
+    r = run_price_pipeline([], top_n=5)
+    assert r["lowest"] is None and r["top"] == [] and r["total_found"] == 0
+
+
+def test_stages_recorded():
+    prods = [_p(x) for x in [1000, 2000, 3000, 4000, 5000]]
+    r = run_price_pipeline(prods, base_price=3000, top_n=2)
+    names = [s["stage"] for s in r["stages"]]
+    assert "price_band" in names and "outlier" in names and names[-1] == "top_n"
+    assert r["stages"][-1]["out"] == 2  # top_n 결과 수
+
+
+def test_pipeline_on_real_fixture():
+    products = parse_search_html(FIXTURE.read_text())
+    r = run_price_pipeline(products, remove_outliers=False, top_n=3)
+    prices = [p["price"] for p in r["top"]]
+    assert prices == sorted(prices)                      # 최저가순
+    assert r["lowest"]["price"] == min(p.price for p in products)
diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py
index fda6e46..b80275f 100644
--- a/lps/worker/handlers.py
+++ b/lps/worker/handlers.py
@@ -1,14 +1,16 @@
 """잡 핸들러 — job_type 별 처리. 현재는 SEARCH(검색)만.
 
-검색 핸들러는 소스 어댑터로 검색해 정규화 결과를 반환한다.
-필터/이상치/AI 유사도(코어 파이프라인)는 다음 단계에서 이 핸들러 안에 결합한다.
+검색 핸들러: 소스 어댑터로 검색 → 코어 파이프라인(필터·이상치·top-N 최저가) → 결과.
+AI 유사도 판정은 파이프라인 슬롯에 키 준비 시 결합한다.
 """
 
 from common.enums import JobType
 from services.search.contract import SearchAdapter
+from services.search.util import parse_price
+from services.pipeline.core import run_price_pipeline
 
 
-def build_search_handler(adapters: dict[str, SearchAdapter], default_source: str = "coupang", limit: int = 40):
+def build_search_handler(adapters: dict[str, SearchAdapter], default_source: str = "coupang", limit: int = 40, top_n: int = 5):
     """검색 핸들러 생성. adapters = {source: SearchAdapter}."""
 
     async def handler(job: dict) -> dict:
@@ -25,11 +27,11 @@ def build_search_handler(adapters: dict[str, SearchAdapter], default_source: str
             raise ValueError(f"no adapter for source: {default_source}")
 
         products = await adapter.search(query, limit=limit)
-        return {
-            "source": adapter.source,
-            "query": query,
-            "count": len(products),
-            "products": [p.model_dump() for p in products],
-        }
+        # 요청 현재가(있으면)를 가격 밴드 기준으로 사용 → 엉뚱한 저가/고가 targeted 컷
+        base_price = parse_price(payload.get("price"))
+        result = run_price_pipeline(products, base_price=base_price, top_n=top_n)
+        result["source"] = adapter.source
+        result["query"] = query
+        return result
 
     return handler