Building a price comparison agent across product feeds

A price comparison agent product feeds system sounds straightforward — fetch a few CSVs, join on SKU, sort by price. Reality is messier: feeds arrive as XML, JSON Lines, gzipped CSVs, and proprietary formats; fields use different names for the same concept; availability signals conflict; and the “same” product has different identifiers across merchants. This tutorial builds a working agent that handles all of it, with code you can run and extend.

Prerequisites

• Python 3.10+
• pip install httpx pandas lxml tenacity pydantic pydantic-settings python-dotenv
• Access to at least two product feeds (sample feeds provided below if you don’t have your own)
• Optional: an LLM API key if you want the agent to generate natural-language summaries

Create a project structure:

mkdir price-comparison-agent && cd price-comparison-agent
mkdir feeds src tests
touch src/__init__.py

Feed ingestion layer

Product feeds share nothing but intent. Some are FTP drops, some HTTP endpoints with auth, some S3 presigned URLs. Start with a protocol-agnostic fetcher that handles compression, encoding detection, and retries.

# src/fetcher.py
from __future__ import annotations
import gzip
import io
import os
from pathlib import Path
from typing import BinaryIO
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

class FeedFetchError(Exception):
    pass

class Fetcher:
    def __init__(self, timeout: float = 30.0, max_retries: int = 3):
        self.client = httpx.Client(timeout=timeout, follow_redirects=True)
        self._retry_policy = retry(
            wait=wait_exponential_jitter(initial=1, max=10),
            stop=stop_after_attempt(max_retries),
            reraise=True,
        )

    @_retry_policy
    def fetch(self, url: str) -> bytes:
        if url.startswith(("http://", "https://")):
            resp = self.client.get(url)
            resp.raise_for_status()
            return resp.content
        elif url.startswith("file://"):
            path = Path(url[7:])
            return path.read_bytes()
        else:
            # Treat as local path
            return Path(url).read_bytes()

    def open_stream(self, url: str) -> BinaryIO:
        raw = self.fetch(url)
        if url.endswith(".gz") or raw[:2] == b"\x1f\x8b":
            return io.BytesIO(gzip.decompress(raw))
        return io.BytesIO(raw)

Test it with a sample feed. Create feeds/sample_merchant_a.xml.gz:

<?xml version="1.0" encoding="UTF-8"?>
<products>
  <product>
    <id>MA-1001</id>
    <title>Wireless Noise-Cancelling Headphones</title>
    <brand>AudioTech</brand>
    <category>Electronics > Audio > Headphones</category>
    <price currency="USD">249.99</price>
    <sale_price currency="USD">199.99</price>
    <availability>in_stock</availability>
    <url>https://merchant-a.example.com/p/MA-1001</url>
    <image_url>https://cdn.merchant-a.example.com/MA-1001.jpg</image_url>
    <gtin>0840123456789</gtin>
    <mpn>AT-WH1000XM5</mpn>
    <condition>new</condition>
  </product>
  <product>
    <id>MA-1002</id>
    <title>Mechanical Gaming Keyboard RGB</title>
    <brand>KeyMaster</brand>
    <category>Electronics > Computers > Keyboards</category>
    <price currency="USD">129.99</price>
    <availability>in_stock</availability>
    <url>https://merchant-a.example.com/p/MA-1002</url>
    <gtin>0840123456796</gtin>
    <mpn>KM-K70-RGB</mpn>
    <condition>new</condition>
  </product>
</products>

And feeds/sample_merchant_b.jsonl.gz (JSON Lines, gzipped):

{"sku": "MB-2001", "name": "Wireless Noise-Cancelling Headphones", "manufacturer": "AudioTech", "product_type": "Headphones", "list_price": 279.99, "current_price": 219.99, "stock_status": "available", "product_url": "https://merchant-b.example.com/item/MB-2001", "upc": "0840123456789", "part_number": "AT-WH1000XM5", "condition": "New"}
{"sku": "MB-2002", "name": "27\" 4K IPS Monitor", "manufacturer": "ViewSonic", "product_type": "Monitors", "list_price": 449.99, "current_price": 399.99, "stock_status": "available", "product_url": "https://merchant-b.example.com/item/MB-2002", "upc": "0766907123456", "part_number": "VP2785-4K", "condition": "New"}
{"sku": "MB-2003", "name": "Mechanical Gaming Keyboard RGB", "manufacturer": "KeyMaster", "product_type": "Keyboards", "list_price": 149.99, "current_price": 119.99, "stock_status": "limited", "product_url": "https://merchant-b.example.com/item/MB-2003", "upc": "0840123456796", "part_number": "KM-K70-RGB", "condition": "New"}

Run a quick sanity check:

# tests/test_fetcher.py
from src.fetcher import Fetcher

fetcher = Fetcher()
xml_stream = fetcher.open_stream("file://feeds/sample_merchant_a.xml.gz")
print(xml_stream.read()[:200])
# b'<?xml version="1.0" encoding="UTF-8"?>\n<products>\n  <product>\n    <id>MA-1001</id>...'

Normalization schema

Every feed maps to a canonical Product model. Use Pydantic for validation and coercion — it catches malformed data early and gives you a single type downstream.

# src/models.py
from __future__ import annotations
from decimal import Decimal
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, field_validator

class Availability(str, Enum):
    IN_STOCK = "in_stock"
    LIMITED = "limited"
    OUT_OF_STOCK = "out_of_stock"
    PREORDER = "preorder"
    UNKNOWN = "unknown"

class Condition(str, Enum):
    NEW = "new"
    REFURBISHED = "refurbished"
    USED = "used"
    OPEN_BOX = "open_box"

class Product(BaseModel):
    source_id: str                    # Merchant's internal ID
    source_name: str                  # "merchant_a", "merchant_b", etc.
    title: str
    brand: Optional[str] = None
    category: Optional[str] = None
    price: Decimal                      # Current selling price
    list_price: Optional[Decimal] = None
    currency: str = "USD"
    availability: Availability = Availability.UNKNOWN
    condition: Condition = Condition.NEW
    product_url: str
    image_url: Optional[str] = None
    gtin: Optional[str] = None          # UPC/EAN/ISBN
    mpn: Optional[str] = None           # Manufacturer part number
    raw: dict = Field(default_factory=dict, exclude=True)  # Original row for debugging

    @field_validator("gtin", "mpn", mode="before")
    @classmethod
    def normalize_identifiers(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return None
        return v.strip().upper().replace("-", "").replace(" ", "")

    @field_validator("price", "list_price", mode="before")
    @classmethod
    def coerce_decimal(cls, v) -> Optional[Decimal]:
        if v is None or v == "":
            return None
        return Decimal(str(v).replace(",", "").replace("$", ""))

    def match_key(self) -> tuple[Optional[str], Optional[str]]:
        """Return the strongest identifier pair for cross-merchant matching."""
        return (self.gtin, self.mpn)

    def fingerprint(self) -> str:
        """Stable hash for deduplication within a feed."""
        parts = [self.source_name, self.source_id]
        if self.gtin:
            parts.append(self.gtin)
        if self.mpn:
            parts.append(self.mpn)
        return "|".join(parts)

Parsers per feed format

Each merchant gets a parser that yields Product objects. Keep parsers small and focused — they only know how to translate their format into the canonical model.

# src/parsers.py
from __future__ import annotations
import csv
import json
import xml.etree.ElementTree as ET
from io import TextIOWrapper
from typing import Iterator, BinaryIO
from decimal import Decimal

from src.models import Product, Availability, Condition
from src.fetcher import Fetcher

def parse_merchant_a_xml(stream: BinaryIO, source_name: str) -> Iterator[Product]:
    """Parse Merchant A's XML format."""
    tree = ET.parse(stream)
    root = tree.getroot()
    for elem in root.findall("product"):
        def text(tag: str) -> Optional[str]:
            node = elem.find(tag)
            return node.text.strip() if node is not None and node.text else None

        price_elem = elem.find("price")
        currency = price_elem.get("currency", "USD") if price_elem is not None else "USD"

        yield Product(
            source_id=text("id") or "",
            source_name=source_name,
            title=text("title") or "",
            brand=text("brand"),
            category=text("category"),
            price=Decimal(text("sale_price") or text("price") or "0"),
            list_price=Decimal(text("price")) if text("price") and text("sale_price") else None,
            currency=currency,
            availability=Availability(text("availability") or "unknown"),
            condition=Condition(text("condition") or "new"),
            product_url=text("url") or "",
            image_url=text("image_url"),
            gtin=text("gtin"),
            mpn=text("mpn"),
            raw={child.tag: child.text for child in elem},
        )

def parse_merchant_b_jsonl(stream: BinaryIO, source_name: str) -> Iterator[Product]:
    """Parse Merchant B's JSON Lines format."""
    wrapper = TextIOWrapper(stream, encoding="utf-8")
    for line in wrapper:
        line = line.strip()
        if not line:
            continue
        data = json.loads(line)
        yield Product(
            source_id=data.get("sku", ""),
            source_name=source_name,
            title=data.get("name", ""),
            brand=data.get("manufacturer"),
            category=data.get("product_type"),
            price=Decimal(str(data.get("current_price", 0))),
            list_price=Decimal(str(data["list_price"])) if data.get("list_price") else None,
            currency="USD",
            availability=Availability(data.get("stock_status", "unknown").lower()),
            condition=Condition(data.get("condition", "new").lower()),
            product_url=data.get("product_url", ""),
            gtin=data.get("upc"),
            mpn=data.get("part_number"),
            raw=data,
        )

PARSERS = {
    "merchant_a": parse_merchant_a_xml,
    "merchant_b": parse_merchant_b_jsonl,
}

def get_parser(source_name: str):
    parser = PARSERS.get(source_name)
    if not parser:
        raise ValueError(f"No parser registered for {source_name}")
    return parser

Matching engine

The core of a price comparison agent product feeds pipeline is matching the same physical product across merchants. GTIN (UPC/EAN) is the gold standard; MPN is the silver. When both are missing, fall back to fuzzy title+brand matching — but flag it as low confidence.

# src/matcher.py
from __future__ import annotations
from collections import defaultdict
from dataclasses import dataclass
from typing import Optional
from difflib import SequenceMatcher

from src.models import Product

@dataclass(frozen=True)
class MatchGroup:
    gtin: Optional[str]
    mpn: Optional[str]
    products: list[Product]
    confidence: str  # "high" | "medium" | "low"

    def best_price(self) -> Optional[Product]:
        in_stock = [p for p in self.products if p.availability.value == "in_stock"]
        candidates = in_stock if in_stock else self.products
        return min(candidates, key=lambda p: p.price) if candidates else None

    def price_spread(self) -> tuple[Decimal, Decimal]:
        prices = [p.price for p in self.products]
        return (min(prices), max(prices)) if prices else (Decimal("0"), Decimal("0"))

def similarity(a: str, b: str) -> float:
    return SequenceMatcher(None, a.lower(), b.lower()).ratio()

def group_products(products: list[Product]) -> list[MatchGroup]:
    # Primary index: GTIN
    by_gtin: dict[str, list[Product]] = defaultdict(list)
    no_gtin: list[Product] = []

    for p in products:
        if p.gtin:
            by_gtin[p.gtin].append(p)
        else:
            no_gtin.append(p)

    groups: list[MatchGroup] = []
    used = set()

    # High confidence: exact GTIN match
    for gtin, group in by_gtin.items():
        groups.append(MatchGroup(gtin=gtin, mpn=None, products=group, confidence="high"))
        for p in group:
            used.add(id(p))

    # Medium confidence: MPN match among remaining
    by_mpn: dict[str, list[Product]] = defaultdict(list)
    for p in no_gtin:
        if p.mpn:
            by_mpn[p.mpn].append(p)

    for mpn, group in by_mpn.items():
        # Check if any already matched via GTIN (shouldn't happen, but safe)
        if any(id(p) in used for p in group):
            continue
        groups.append(MatchGroup(gtin=None, mpn=mpn, products=group, confidence="medium"))
        for p in group:
            used.add(id(p))

    # Low confidence: fuzzy title+brand for leftovers
    remaining = [p for p in no_gtin if id(p) not in used and (not p.mpn or p.mpn not in by_mpn)]
    clustered: list[list[Product]] = []
    for p in remaining:
        placed = False
        for cluster in clustered:
            rep = cluster[0]
            if p.brand and rep.brand and p.brand.lower() == rep.brand.lower():
                if similarity(p.title, rep.title) > 0.85:
                    cluster.append(p)
                    placed = True
                    break
        if not placed:
            clustered.append([p])

    for cluster in clustered:
        if len(cluster) > 1:
            groups.append(MatchGroup(gtin=None, mpn=None, products=cluster, confidence="low"))

    return groups

Agent orchestration

Wire it together: fetch → parse → normalize → match → present. The agent exposes a clean query interface.

# src/agent.py
from __future__ import annotations
from dataclasses import dataclass
from typing import Optional
from decimal import Decimal

from src.fetcher import Fetcher
from src.parsers import get_parser
from src.matcher import group_products, MatchGroup
from src.models import Product, Availability

@dataclass
class ComparisonResult:
    match_group: MatchGroup
    best_deal: Product
    savings_vs_list: Optional[Decimal]
    all_offers: list[Product]

class PriceComparisonAgent:
    def __init__(self, fetcher: Optional[Fetcher] = None):
        self.fetcher = fetcher or Fetcher()
        self._products: list[Product] = []
        self._groups: list[MatchGroup] = []

    def ingest(self, feed_configs: list[dict]) -> int:
        """feed_configs: list of {"source_name": str, "url": str}"""
        all_products: list[Product] = []
        for cfg in feed_configs:
            source_name = cfg["source_name"]
            url = cfg["url"]
            parser = get_parser(source_name)
            stream = self.fetcher.open_stream(url)
            for product in parser(stream, source_name):
                all_products.append(product)
        self._products = all_products
        self._groups = group_products(all_products)
        return len(all_products)

    def compare(self, query: str, limit: int = 10) -> list[ComparisonResult]:
        """Simple text search over titles/brands, then return match groups."""
        query_lower = query.lower()
        matched_groups = []
        for group in self._groups:
            # Check if any product in group matches query
            for p in group.products:
                if query_lower in p.title.lower() or (p.brand and query_lower in p.brand.lower()):
                    matched_groups.append(group)
                    break
        matched_groups.sort(key=lambda g: g.best_price().price if g.best_price() else Decimal("Infinity"))
        return [self._build_result(g) for g in matched_groups[:limit]]

    def _build_result(self, group: MatchGroup) -> ComparisonResult:
        best = group.best_price()
        assert best is not None
        savings = None
        if best.list_price and best.list_price > best.price:
            savings = best.list_price - best.price
        return ComparisonResult(
            match_group=group,
            best_deal=best,
            savings_vs_list=savings,
            all_offers=sorted(group.products, key=lambda p: p.price),
        )

    def get_all_groups(self) -> list[MatchGroup]:
        return self._groups

CLI entry point

Make it runnable. A thin CLI lets you test feeds and iterate fast.

# src/cli.py
from __future__ import annotations
import argparse
import json
from decimal import Decimal

from src.agent import PriceComparisonAgent
from src.fetcher import Fetcher

def format_money(d: Decimal) -> str:
    return f"${d:.2f}"

def main():
    parser = argparse.ArgumentParser(description="Price Comparison Agent")
    parser.add_argument("--feeds", required=True, help="JSON file with feed configs")
    parser.add_argument("--query", help="Search query")
    parser.add_argument("--limit", type=int, default=10)
    parser.add_argument("--json", action="store_true", help="Output JSON")
    args = parser.parse_args()

    with open(args.feeds) as f:
        feed_configs = json.load(f)

    agent = PriceComparisonAgent(Fetcher())
    count = agent.ingest(feed_configs)
    print(f"Ingested {count} products from {len(feed_configs)} feeds", flush=True)

    if args.query:
        results = agent.compare(args.query, limit=args.limit)
        if args.json:
            print(json.dumps([{
                "query": args.query,
                "confidence": r.match_group.confidence,
                "best_deal": {
                    "merchant": r.best_deal.source_name,
                    "title": r.best_deal.title,
                    "price": format_money(r.best_deal.price),
                    "url": r.best_deal.product_url,
                },
                "all_offers": [{
                    "merchant": p.source_name,
                    "price": format_money(p.price),
                    "availability": p.availability.value,
                    "url": p.product_url,
                } for p in r.all_offers],
                "savings_vs_list": format_money(r.savings_vs_list) if r.savings_vs_list else None,
            } for r in results], indent=2))
        else:
            for r in results:
                print(f"\n=== {r.best_deal.title} ({r.match_group.confidence} confidence) ===")
                print(f"Best: {r.best_deal.source_name} @ {format_money(r.best_deal.price)}")
                if r.savings_vs_list:
                    print(f"  Savings vs list: {format_money(r.savings_vs_list)}")
                print("  All offers:")
                for p in r.all_offers:
                    print(f"    {p.source_name}: {format_money(p.price)} ({p.availability.value}) - {p.product_url}")
    else:
        # Show summary
        groups = agent.get_all_groups()
        print(f"Formed {len(groups)} match groups")
        for g in groups[:5]:
            best = g.best_price()
            print(f"  {g.confidence:6} | GTIN={g.gtin or 'N/A'} MPN={g.mpn or 'N/A'} | {len(g.products)} offers | Best: {best.source_name if best else 'N/A'} @ {format_money(best.price) if best else 'N/A'}")

if __name__ == "__main__":
    main()

Create a feed config file feeds/config.json:

[
  {"source_name": "merchant_a", "url": "file://feeds/sample_merchant_a.xml.gz"},
  {"source_name": "merchant_b", "url": "file://feeds/sample_merchant_b.jsonl.gz"}
]

Run it:

$ python -m src.cli --feeds feeds/config.json --query "headphones"
Ingested 5 products from 2 feeds
Formed 2 match groups

=== Wireless Noise-Cancelling Headphones (high confidence) ===
Best: merchant_a @ $199.99
  Savings vs list: $50.00
  All offers:
    merchant_a: $199.99 (in_stock) - https://merchant-a.example.com/p/MA-1001
    merchant_b: $219.99 (in_stock) - https://merchant-b.example.com/item/MB-2001

$ python -m src.cli --feeds feeds/config.json --query "keyboard" --json
{
  "query": "keyboard",
  "confidence": "high",
  "best_deal": {
    "merchant": "merchant_b",
    "title": "Mechanical Gaming Keyboard RGB",
    "price": "$119.99",
    "url": "https://merchant-b.example.com/item/MB-2003"
  },
  "all_offers": [
    {"merchant": "merchant_b", "price": "$119.99", "availability": "limited", "url": "https://merchant-b.example.com/item/MB-2003"},
    {"merchant": "merchant_a", "price": "$129.99", "availability": "in_stock", "url": "https://merchant-a.example.com/p/MA-1002"}
  ],
  "savings_vs_list": "$29.99"
}

Handling real-world messiness

The code above works for clean samples. Production feeds need more armor:

Availability normalization — merchants use “Y”, “N”, “In Stock”, “Out of Stock”, “1”, “0”, “Available”, “Backorder”. Extend Availability with a from_merchant_value(source_name, raw) classmethod that encapsulates per-merchant mappings.

Currency conversion — if feeds mix currencies, pull daily FX rates at ingest time and store price_usd alongside original. Don’t convert at query time; rates drift.

Feed freshness — add fetched_at to each Product. Expire groups where all products are older than your SLA (e.g., 24h). Schedule re-ingestion with a cron job or Airflow DAG.

Deduplication within a feed — some merchants list the same SKU multiple times (variants, bundles). The fingerprint() method helps; add a dedup pass before matching.

Variant handling — “Headphones Black” vs “Headphones White” share a GTIN but differ in color. Extend MatchGroup to cluster by GTIN+color_attribute when available.

Scaling to hundreds of feeds

The single-process design works up to ~50 feeds with modest size. Beyond that:

1. Parallel ingestion — use concurrent.futures.ThreadPoolExecutor in ingest() since fetching is I/O-bound.
2. Persistent storage — write normalized Product rows to Postgres or Parquet. Match on indexed GTIN/MPN columns instead of in-memory dicts.
3. Incremental updates — track feed ETags/Last-Modified; only re-parse changed feeds.
4. Search layer — index titles/brands in Elasticsearch or Typesense for sub-100ms query latency.

If you’re routing LLM calls for natural-language summaries (“Summarize the best deals for noise-cancelling headphones under $200”), n4n.ai’s single endpoint lets you swap models without rewriting the agent — useful when you need a cheaper model for high-volume summarization and a stronger one for complex reasoning.

Extending the agent

Three high-value additions:

Price history — store every price observation with timestamp. Surface “price dropped 15% in 7 days” signals.

Merchant reliability scoring — track fulfillment rate, return rate, shipping speed per merchant. Weight best_deal by reliability, not just price.

Alerting — when a match group’s best price crosses a user’s threshold, emit an event (webhook, email, push). The agent becomes a platform, not just a query tool.

Full file tree

price-comparison-agent/
├── feeds/
│   ├── config.json
│   ├── sample_merchant_a.xml.gz
│   └── sample_merchant_b.jsonl.gz
├── src/
│   ├── __init__.py
│   ├── agent.py
│   ├── cli.py
│   ├── fetcher.py
│   ├── matcher.py
│   ├── models.py
│   └── parsers.py
└── tests/
    └── test_fetcher.py

Run the full test suite:

$ python -m pytest tests/ -v

You now have a price comparison agent product feeds pipeline that ingests heterogeneous formats, normalizes to a strict schema, matches products across merchants with confidence scoring, and exposes a query interface. The architecture isolates format parsing, matching logic, and presentation — each replaceable without rewriting the others. Ship it, then iterate on the messiness your real feeds throw at you.