경쟁사 가격 모니터링: 자동 크롤링과 동적 가격 대응 전략

이커머스

가격 모니터링경쟁사 분석동적 가격크롤링가격 전략

이 글은 누구를 위한 것인가

  • 경쟁사보다 비싸게 팔고 있는지 파악하지 못하는 팀
  • 최저가 보장 정책을 자동으로 운영하고 싶은 이커머스 팀
  • 가격 조정을 수동으로 하다가 지친 MD(상품기획자)

들어가며

가격 경쟁력은 이커머스 전환율의 핵심 요소다. 같은 상품이 경쟁사에서 10% 싸게 팔린다면 고객은 그리로 간다. 실시간 가격 모니터링과 자동 대응이 없으면 뒤늦게 매출 하락으로 알게 된다.

이 글은 bluefoxdev.kr의 이커머스 가격 전략 가이드 를 참고하여 작성했습니다.


1. 가격 수집 방법

[경쟁사 가격 수집 옵션]

1. 공식 API (권장):
   - 네이버쇼핑, 다나와, 에누리 API
   - 정확하고 합법적
   - 일부 플랫폼만 지원

2. RSS/사이트맵:
   - 일부 쇼핑몰이 가격 피드 제공
   - 단순하고 안정적

3. 웹 크롤링:
   - 직접 수집
   - robots.txt 준수 필요
   - 차단 위험, 유지보수 비용

4. 가격 비교 서비스 API:
   - 다나와, 에누리 비교 API
   - 이미 집계된 데이터 활용

2. 가격 모니터링 파이프라인

import asyncio
import httpx
from bs4 import BeautifulSoup
from datetime import datetime

class PriceMonitor:
    
    async def scrape_competitor_price(
        self,
        competitor: str,
        product_url: str,
        price_selector: str,
    ) -> float | None:
        """경쟁사 페이지에서 가격 추출"""
        
        headers = {
            "User-Agent": "Mozilla/5.0 (compatible; PriceMonitor/1.0)",
        }
        
        async with httpx.AsyncClient(follow_redirects=True, timeout=10.0) as client:
            try:
                response = await client.get(product_url, headers=headers)
                soup = BeautifulSoup(response.text, 'html.parser')
                
                price_element = soup.select_one(price_selector)
                if not price_element:
                    return None
                
                price_text = price_element.get_text().replace(",", "").strip()
                price = float(''.join(filter(str.isdigit, price_text)))
                return price
                
            except Exception as e:
                print(f"크롤링 오류 {competitor}: {e}")
                return None
    
    async def monitor_all_products(self, products: list[dict]):
        """전체 상품 가격 모니터링"""
        
        tasks = []
        for product in products:
            for competitor_config in product.get("competitor_urls", []):
                tasks.append(self.check_and_store(product, competitor_config))
        
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def check_and_store(self, product: dict, competitor_config: dict):
        price = await self.scrape_competitor_price(
            competitor=competitor_config["name"],
            product_url=competitor_config["url"],
            price_selector=competitor_config["price_selector"],
        )
        
        if price is None:
            return
        
        # DB 저장
        await db.execute("""
            INSERT INTO competitor_prices (product_id, competitor, price, scraped_at)
            VALUES ($1, $2, $3, NOW())
        """, product["id"], competitor_config["name"], price)
        
        # 가격 변동 감지 및 알림
        await self.check_price_alert(product, competitor_config["name"], price)
    
    async def check_price_alert(self, product: dict, competitor: str, competitor_price: float):
        """우리 가격이 경쟁사보다 N% 비싸면 알림"""
        our_price = product["current_price"]
        price_diff_pct = (our_price - competitor_price) / competitor_price * 100
        
        if price_diff_pct > 5:  # 5% 이상 비쌈
            await notify_slack(f"""
⚠️ 가격 경쟁력 경고
상품: {product['name']}
경쟁사({competitor}): {competitor_price:,.0f}원
우리: {our_price:,.0f}원
차이: +{price_diff_pct:.1f}%
            """)

3. 자동 가격 조정 규칙

from dataclasses import dataclass

@dataclass
class PricingRule:
    rule_type: str  # "match", "undercut", "maintain_margin"
    match_competitors: list[str]
    undercut_by: float = 0  # 원 단위
    undercut_pct: float = 0  # %
    min_price: float = 0    # 최저가 제한
    max_price: float = 0    # 최고가 제한

async def apply_pricing_rule(product_id: str, rule: PricingRule) -> float | None:
    """가격 조정 규칙 적용"""
    
    # 경쟁사 최저가 조회
    competitor_prices = await db.fetch("""
        SELECT MIN(price) as min_price
        FROM competitor_prices
        WHERE product_id = $1
          AND competitor = ANY($2)
          AND scraped_at > NOW() - INTERVAL '24 hours'
    """, product_id, rule.match_competitors)
    
    if not competitor_prices or not competitor_prices[0]["min_price"]:
        return None
    
    competitor_min = competitor_prices[0]["min_price"]
    
    if rule.rule_type == "match":
        new_price = competitor_min
    elif rule.rule_type == "undercut":
        new_price = competitor_min - rule.undercut_by - (competitor_min * rule.undercut_pct / 100)
    else:
        return None
    
    # 최저/최고가 제한 적용
    if rule.min_price > 0:
        new_price = max(new_price, rule.min_price)
    if rule.max_price > 0:
        new_price = min(new_price, rule.max_price)
    
    # 가격 업데이트
    await db.execute(
        "UPDATE products SET current_price=$1, price_updated_at=NOW() WHERE id=$2",
        new_price, product_id
    )
    
    return new_price

마무리

경쟁사 가격 모니터링은 "얼마나 자주"가 중요하다. 하루 1회 수집으로는 실시간 가격 전쟁에 대응할 수 없다. 핵심 경쟁 상품 100개는 1시간마다, 나머지는 하루 1-4회 수집이 현실적이다. 자동 가격 조정은 마진 하한선을 반드시 설정하고 시작해야 손실을 막을 수 있다.