不同地区亚马逊价格对比：同款商品为何相差 40%，以及如何系统监控

import requests
import json
import time
import logging
from datetime import datetime, timezone
from typing import Optional
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(message)s")

PANGOLINFO_API_KEY = "your_api_key_here"
EXCHANGE_API_KEY = "your_exchangerate_api_key"   # exchangerate-api.com 或同类服务

# 各站点 VAT 税率（欧洲站含税，美国站不含税）
VAT_RATES = {
    "US": 0.0,    # 美国不在页面显示税
    "UK": 0.20,   # 英国 VAT 20%
    "DE": 0.19,   # 德国 VAT 19%
    "FR": 0.20,   # 法国 VAT 20%
    "IT": 0.22,   # 意大利 VAT 22%
    "ES": 0.21,   # 西班牙 VAT 21%
    "JP": 0.10,   # 日本消费税 10%
    "CA": 0.0,    # 加拿大 GST 结账时计，页面不含
    "MX": 0.16,   # 墨西哥 IVA 16%
}

# 各站点货币代码
MARKETPLACE_CURRENCIES = {
    "US": "USD", "UK": "GBP", "DE": "EUR", "FR": "EUR",
    "IT": "EUR", "ES": "EUR", "JP": "JPY", "CA": "CAD", "MX": "MXN"
}


def get_exchange_rates(base: str = "USD") -> dict:
    """获取当前汇率（以 USD 为基准）"""
    try:
        resp = requests.get(
            f"https://v6.exchangerate-api.com/v6/{EXCHANGE_API_KEY}/latest/{base}",
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        rates = data.get("conversion_rates", {})
        logging.info(f"汇率获取成功，时间戳：{data.get('time_last_update_utc')}")
        return rates
    except Exception as e:
        logging.error(f"汇率获取失败: {e}")
        # fallback 汇率（仅用于演示）
        return {"GBP": 1.27, "EUR": 1.08, "JPY": 0.0067, "CAD": 0.73, "MXN": 0.050}


def to_usd(price: float, currency: str, rates: dict) -> float:
    """将价格换算为 USD"""
    if currency == "USD":
        return price
    rate = rates.get(currency)
    if not rate:
        logging.warning(f"无 {currency} 汇率，跳过换算")
        return price
    return round(price / rate, 2)


def remove_vat(price: float, marketplace: str) -> float:
    """还原不含税价格"""
    vat_rate = VAT_RATES.get(marketplace, 0.0)
    if vat_rate == 0.0:
        return price
    return round(price / (1 + vat_rate), 2)


def fetch_price_snapshot(
    asin: str, marketplace: str
) -> Optional[dict]:
    """从 Pangolinfo API 获取单站点价格快照"""
    try:
        resp = requests.post(
            "https://api.pangolinfo.com/v1/amazon/product",
            headers={"Authorization": f"Bearer {PANGOLINFO_API_KEY}"},
            json={
                "asin": asin,
                "marketplace": marketplace,
                "fields": [
                    "buybox_price",
                    "buybox_seller_id",
                    "buybox_is_amazon",
                    "list_price",
                    "offer_count",
                    "coupon_type",
                    "coupon_value",
                    "effective_price",
                    "deal_price",
                ]
            },
            timeout=30
        )
        resp.raise_for_status()
        data = resp.json().get("data", {})
        data.update({
            "_asin": asin,
            "_marketplace": marketplace,
            "_ts": datetime.now(timezone.utc).isoformat()
        })
        return data
    except Exception as e:
        logging.error(f"[{asin}|{marketplace}] 采集失败: {e}")
        return None


def build_cross_region_comparison(
    asin: str,
    marketplaces: list[str],
    rates: dict,
    fetch_delay: float = 0.5
) -> dict:
    """
    对单个 ASIN 做跨站价格对比
    
    返回结构：
    {
      "asin": "...",
      "snapshots": {marketplace: {...raw data}},
      "comparison": [
        {
          "marketplace": "DE",
          "local_price": 34.99,
          "currency": "EUR",
          "price_excl_vat": 29.40,
          "usd_equivalent": 31.75,
          "offer_count": 8,
          ...
        },
        ...
      ],
      "cheapest_usd": {"marketplace": "JP", "usd": 24.59},
      "most_expensive_usd": {"marketplace": "FR", "usd": 32.39},
      "spread_usd": 7.80,   # 最贵与最便宜之差
      "spread_pct": 31.7    # 价差百分比
    }
    """
    snapshots = {}
    comparison = []

    for mkt in marketplaces:
        snap = fetch_price_snapshot(asin, mkt)
        if snap:
            snapshots[mkt] = snap
        time.sleep(fetch_delay)

    for mkt, snap in snapshots.items():
        local_price = snap.get("buybox_price") or snap.get("effective_price")
        if not local_price:
            continue

        currency = MARKETPLACE_CURRENCIES.get(mkt, "USD")
        price_excl_vat = remove_vat(local_price, mkt)
        usd_equiv = to_usd(price_excl_vat, currency, rates)

        comparison.append({
            "marketplace": mkt,
            "local_price": local_price,
            "currency": currency,
            "price_excl_vat": price_excl_vat,
            "vat_rate": VAT_RATES.get(mkt, 0),
            "usd_equivalent": usd_equiv,
            "offer_count": snap.get("offer_count"),
            "buybox_is_amazon": snap.get("buybox_is_amazon"),
            "has_coupon": bool(snap.get("coupon_type")),
            "deal_price": snap.get("deal_price"),
        })

    if not comparison:
        return {"asin": asin, "snapshots": snapshots, "comparison": [], "error": "no data"}

    # 排序：从低到高
    comparison.sort(key=lambda x: x["usd_equivalent"])
    cheapest = comparison[0]
    most_expensive = comparison[-1]
    spread_usd = round(most_expensive["usd_equivalent"] - cheapest["usd_equivalent"], 2)
    spread_pct = round(spread_usd / cheapest["usd_equivalent"] * 100, 1) if cheapest["usd_equivalent"] else 0

    return {
        "asin": asin,
        "snapshots": snapshots,
        "comparison": comparison,
        "cheapest_usd": {
            "marketplace": cheapest["marketplace"],
            "usd": cheapest["usd_equivalent"]
        },
        "most_expensive_usd": {
            "marketplace": most_expensive["marketplace"],
            "usd": most_expensive["usd_equivalent"]
        },
        "spread_usd": spread_usd,
        "spread_pct": spread_pct,
    }


def run_cross_region_monitor(
    asins: list[str],
    marketplaces: list[str] = None,
    alert_spread_pct: float = 20.0,
    output_file: str = "cross_region_prices.jsonl",
    interval_hours: int = 6
):
    """
    主监控循环：定期采集多站点价格，检测价差超阈值情况
    
    Args:
        alert_spread_pct: 价差超过此百分比时触发告警（默认 20%）
    """
    if marketplaces is None:
        marketplaces = ["US", "UK", "DE", "FR", "JP", "CA"]

    out = Path(output_file)
    cycle = 0

    while True:
        cycle += 1
        logging.info(f"\n{'='*60}\n第 {cycle} 轮 | {datetime.now().strftime('%Y-%m-%d %H:%M')}\n{'='*60}")

        # 每轮开始时统一获取汇率
        rates = get_exchange_rates("USD")

        batch_results = []
        for asin in asins:
            logging.info(f"  采集 {asin}：{len(marketplaces)} 个站点...")
            result = build_cross_region_comparison(asin, marketplaces, rates)
            batch_results.append(result)

            # 价差告警
            spread_pct = result.get("spread_pct", 0)
            if spread_pct >= alert_spread_pct:
                cheapest = result.get("cheapest_usd", {})
                expensive = result.get("most_expensive_usd", {})
                logging.warning(
                    f"  🚨 价差告警: {asin} | "
                    f"最便宜: {cheapest.get('marketplace')} ${cheapest.get('usd')} | "
                    f"最贵: {expensive.get('marketplace')} ${expensive.get('usd')} | "
                    f"价差: {spread_pct}%"
                )

            # 打印各站对比摘要
            for item in result.get("comparison", []):
                logging.info(
                    f"    {item['marketplace']}: "
                    f"{item['local_price']}{item['currency']} → "
                    f"不含税 {item['price_excl_vat']}{item['currency']} → "
                    f"USD ${item['usd_equivalent']} | "
                    f"Offers={item['offer_count']}"
                )

        # 持久化结果
        with open(out, "a", encoding="utf-8") as f:
            for r in batch_results:
                # 不存储 snapshots 原始数据（节省空间）
                r_slim = {k: v for k, v in r.items() if k != "snapshots"}
                r_slim["_cycle"] = cycle
                f.write(json.dumps(r_slim, ensure_ascii=False) + "\n")

        logging.info(f"本轮完成 | {interval_hours}h 后下一轮")
        time.sleep(interval_hours * 3600)


# ── 使用示例 ──────────────────────────────────────────────
if __name__ == "__main__":
    TARGET_ASINS = [
        "B0CHP7BPYQ",   # 替换为你的目标 ASIN
        "B09G9FPHY6",
    ]
    run_cross_region_monitor(
        asins=TARGET_ASINS,
        marketplaces=["US", "UK", "DE", "FR", "JP"],
        alert_spread_pct=20.0,
        output_file="cross_region_prices.jsonl",
        interval_hours=6
    )