fb72ab625f
검색 결과를 실제 최저가로 정제하는 파이프라인을 핸들러에 결합한다. 검색→필터→이상치→정렬→top-N. AI 유사도 판정 슬롯은 비워둠(키 대기). - pipeline/filters: keep_only_mall·filter_out_malls·filter_by_price_band(요청 현재가 기준 targeted 컷) - pipeline/outliers: IQR 기반 이상치 제거(z-score 대신 — 분포 가정 없음, stdlib만) - pipeline/core: STAGE in/out 관측 로깅 + top-N 최저가(레퍼런스 pipeline_log 계승) - handler: 검색→run_price_pipeline 결합, 요청 price 를 밴드 기준으로 사용 - tests: 정렬/IQR(극단 저·고가 제거)/밴드/mall/빈입력/STAGE/실 fixture 7건 → 전체 27/27 - 라이브 확인: 스탠리 텀블러 40건→top-5 최저가, STAGE 카운트 노출 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
65 lines
2.6 KiB
Python
65 lines
2.6 KiB
Python
"""코어 파이프라인 테스트 — 필터·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)
|