#!/usr/bin/env python3 """Reproduce the published mushroom-lamp case-study aggregates.""" from __future__ import annotations import argparse import csv import hashlib import json import statistics from collections import Counter, defaultdict from pathlib import Path EXPECTED_COUNTS = { "captured_rows": 1275, "unique_listings": 1090, "duplicate_rows": 185, "unique_shops": 561, } REQUIRED_FIELDS = { "discount_pct", "is_ad", "is_bestseller", "is_etsys_pick", "is_on_sale", "is_popular_now", "is_star_seller", "listing_id", "price", "rank", "shop_id", "shop_name", } def percentage(count: int, total: int) -> float: return round(100 * count / total, 1) def is_true(row: dict[str, str], field: str) -> bool: return row[field].lower() == "true" def summarize_group(rows: list[dict[str, str]]) -> dict[str, float | int]: return { "listings": len(rows), "median_price": round(statistics.median(float(row["price"]) for row in rows), 2), "on_sale_pct": percentage(sum(is_true(row, "is_on_sale") for row in rows), len(rows)), "star_seller_pct": percentage(sum(is_true(row, "is_star_seller") for row in rows), len(rows)), } def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Reproduce the published Astrava Labs mushroom-lamp case-study aggregates.", ) parser.add_argument("dataset", type=Path, help="Path to the original Etsy Listings Scraper CSV export.") return parser.parse_args() def main() -> None: dataset = parse_args().dataset with dataset.open(encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) missing_fields = REQUIRED_FIELDS.difference(reader.fieldnames or []) if missing_fields: raise SystemExit(f"Dataset is missing required fields: {', '.join(sorted(missing_fields))}") captured = list(reader) by_listing: dict[str, list[dict[str, str]]] = defaultdict(list) for row in captured: by_listing[row["listing_id"]].append(row) first_seen = { listing_id: min(rows, key=lambda row: int(row["rank"])) for listing_id, rows in by_listing.items() } unique = list(first_seen.values()) prices = [float(row["price"]) for row in unique] parsed_discounts = [float(row["discount_pct"]) for row in unique if row["discount_pct"]] placement_groups: dict[str, list[dict[str, str]]] = defaultdict(list) for listing_id, rows in by_listing.items(): has_ad = any(is_true(row, "is_ad") for row in rows) has_organic = any(not is_true(row, "is_ad") for row in rows) group = "both" if has_ad and has_organic else "sponsored_only" if has_ad else "organic_only" placement_groups[group].append(first_seen[listing_id]) price_band_tests = { "under_25": lambda price: price < 25, "25_to_49": lambda price: 25 <= price < 50, "50_to_74": lambda price: 50 <= price < 75, "75_to_99": lambda price: 75 <= price < 100, "100_to_249": lambda price: 100 <= price < 250, "250_plus": lambda price: price >= 250, } discount_values = [float(row["discount_pct"]) if row["discount_pct"] else 0 for row in unique] discount_bands = { "none": sum(value == 0 for value in discount_values), "1_to_20": sum(0 < value <= 20 for value in discount_values), "21_to_40": sum(20 < value <= 40 for value in discount_values), "41_to_60": sum(40 < value <= 60 for value in discount_values), "61_plus": sum(value > 60 for value in discount_values), } rank_groups = { "first_100_rows": [row for row in unique if int(row["rank"]) <= 100], "remaining_rows": [row for row in unique if int(row["rank"]) > 100], } badge_fields = ["is_bestseller", "is_popular_now", "is_etsys_pick", "is_star_seller"] badge_rates = { group: { field: { "count": sum(is_true(row, field) for row in rows), "pct": percentage(sum(is_true(row, field) for row in rows), len(rows)), } for field in badge_fields } for group, rows in rank_groups.items() } shops = Counter((row["shop_id"], row["shop_name"]) for row in unique) summary = { "dataset_sha256": sha256(dataset), "captured_rows": len(captured), "unique_listings": len(unique), "duplicate_rows": len(captured) - len(unique), "repeated_listing_ids": sum(len(rows) > 1 for rows in by_listing.values()), "unique_shops": len(shops), "median_price": round(statistics.median(prices), 2), "mean_price": round(statistics.mean(prices), 2), "on_sale": { "count": sum(is_true(row, "is_on_sale") for row in unique), "pct": percentage(sum(is_true(row, "is_on_sale") for row in unique), len(unique)), }, "parsed_discounts": { "count": len(parsed_discounts), "pct": percentage(len(parsed_discounts), len(unique)), "median": statistics.median(parsed_discounts), }, "sponsored_classified_rows": { "count": sum(is_true(row, "is_ad") for row in captured), "pct": percentage(sum(is_true(row, "is_ad") for row in captured), len(captured)), }, "dual_mode_listing_ids": len(placement_groups["both"]), "price_bands": { name: {"count": sum(test(price) for price in prices), "pct": percentage(sum(test(price) for price in prices), len(unique))} for name, test in price_band_tests.items() }, "discount_bands": { name: {"count": count, "pct": percentage(count, len(unique))} for name, count in discount_bands.items() }, "placement_groups": { name: summarize_group(rows) for name, rows in placement_groups.items() }, "badge_rates": badge_rates, "top_shops": [ {"shop": name, "listings": count, "share_pct": percentage(count, len(unique))} for (_shop_id, name), count in shops.most_common(10) ], "top_10_shop_share_pct": percentage(sum(count for _, count in shops.most_common(10)), len(unique)), } for field, expected in EXPECTED_COUNTS.items(): actual = summary[field] if actual != expected: raise SystemExit(f"Unexpected {field}: expected {expected}, received {actual}") print(json.dumps(summary, indent=2)) if __name__ == "__main__": main()