Price Signal Collection — 竞品价格信号实时采集与结构化

1. 解决的问题

在于：数据异构性（多平台格式差异）、反爬对抗（动态 JS 渲染、验证码）、实时性需求（价格窗口窄、竞品调价响应快）三重矛盾。

2. 核心算法逻辑

竞品价格信号采集的核心挑战在于：数据异构性（多平台格式差异）、反爬对抗（动态 JS 渲染、验证码）、实时性需求（价格窗口窄、竞品调价响应快）三重矛盾。

3. 业务应用场景

业务背景：某母婴 DTC 品牌旗下安抚奶嘴（ASIN B0XXXX001）面对 Top5 竞品的价格战。需要在竞品降价 30 分钟内做出调价响应，否则 Buy Box 获得率从 82% 跌至 41%。

效果量化： - 竞品降价捕获率：71% → 94%（+23 pp） - 平均响应延迟：47 min → 18 min（-62%） - Buy Box 获得率月均：41% → 79%（+38 pp） - 月度 GMV 增量：约 +$32,000（基于 Buy Box 转化率提升测算）

ROI：爬取基础设施成本 $800/月，GMV 增量带来毛利约 $9,600/月，ROI ≈ 12x。

4. 输入数据要求

• `dt_min=15`（促销期可调至 5 min）
• `dt_max=1440`（非活跃 SKU 每天一次）
• `lam=3.0`（调高 → 更激进高频；调低 → 更平滑）
• 反爬应对：建议使用住宅 IP 代理池 + User-Agent 轮换 + 随机延迟 [0.5, 3.0]s
• 汇率更新：`CURRENCY_RATES` 应每小时从 ECB/OpenExchangeRates 刷新

5. 输出结果

• `dt_min=15`（促销期可调至 5 min）
• `dt_max=1440`（非活跃 SKU 每天一次）
• `lam=3.0`（调高 → 更激进高频；调低 → 更平滑）
• 反爬应对：建议使用住宅 IP 代理池 + User-Agent 轮换 + 随机延迟 [0.5, 3.0]s
• 汇率更新：`CURRENCY_RATES` 应每小时从 ECB/OpenExchangeRates 刷新

6. 业务价值 / ROI

73.2 万

7. 代码模板

代码块数量：4 · 路径：未检测到

"""
Price Signal Collection Pipeline
整合 PriceHunter (DOM抽取) + DART-Price (自适应调度) + SignalFusion (Kalman融合)
使用 mock 数据，可直接运行
"""

import re
import math
import time
import random
import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta


# ── 数据结构 ────────────────────────────────────────────────────────────

@dataclass
class PriceRecord:
    """单次价格采集记录"""
    sku_id: str
    source: str          # amazon / walmart / 1688
    raw_price: float     # 原始价格（本地货币）
    currency: str        # USD / CNY / EUR
    timestamp: datetime
    stock_status: str    # in_stock / out_of_stock / limited
    seller: str = ""


@dataclass
class PriceSignal:
    """融合后的价格信号"""
    sku_id: str
    fused_price: float          # Kalman 融合价格（USD）
    price_change_pct: float     # 相对上次融合值的变化百分比
    volatility: float           # 历史波动率 σ
    anomaly: bool               # 是否异常（价格突变）
    sources_count: int          # 本次融合的数据源数量
    updated_at: datetime


# ── PriceHunter：DOM 语义价格抽取 ───────────────────────────────────────

class PriceHunter:
    """
    模拟 DOM 树剪枝 + 语义价格抽取
    真实环境中使用 playwright + BeautifulSoup 替换 mock_html_fetch
    """

    # 货币符号 + 数字模式
    PRICE_PATTERN = re.compile(
        r'([$€¥£₩])\s*(\d{1,6}(?:[,，]\d{3})*(?:\.\d{1,2})?)'
        r'|(\d{1,6}(?:[,，]\d{3})*(?:\.\d{1,2})?)\s*(USD|CNY|EUR|GBP)'
    )
    CURRENCY_RATES = {"USD": 1.0, "CNY": 0.138, "EUR": 1.08, "GBP": 1.27, "$": 1.0, "¥": 0.138, "€": 1.08, "£": 1.27, "₩": 0.00073}

    def extract_from_html(self, html: str, sku_id: str, source: str) -> Optional[PriceRecord]:
        """从 HTML 字符串中抽取价格"""
        matches = self.PRICE_PATTERN.findall(html)

8. 论文来源

• 2412.09883
• 2501.14423
• 2503.07612