# Canonical schemas

Three tables, three grains. Every file is append-only with a `snapshot_date`
column (UTC calendar date); one block of rows per run. Brand-specific attributes
with no canonical column go into `extra_attributes` as JSON so the header never
widens. `IncrementalCsv` fails loud on header drift and flushes every row.

## Products (`output/brands/<brand>.csv`, 35 columns)

```
snapshot_date, brand, sku, name, family, collection, category, gender,
price_value, price_currency, list_price_value, is_discounted, stock_status,
movement_type, caliber, power_reserve,
case_material, case_diameter_mm, case_thickness, case_shape, crystal,
water_resistance, dial_color, bracelet_material, bracelet_color, buckle,
is_limited_edition, rating, review_count,
description, image_url, image_count, product_url, extra_attributes
```

- `sku` = the site's own reference (`T1424621605200`, `M79030N-0001`, `TW2V68000`).
  One row per SKU; a model with variant SKUs gets one row per variant that has
  its own URL and price.
- `collection` = the model line (Seastar, PRX, BOLD). `category` = the site-nav
  bucket. Deliberately distinct; `category` is usually empty on a PDP.
- `gender` ∈ {`men's`, `women's`, `unisex`, `""`}.
- `price_value` is a number in `price_currency`; 0 or missing -> empty (a gap,
  never a zero price). `list_price_value` only when a strike-through exists;
  `is_discounted` = list > price.
- `stock_status` = the schema.org availability tail (`InStock`, `OutOfStock`,
  `PreOrder`) or the site's own word.
- `case_diameter_mm` / `case_thickness` are floats (mm, 1 decimal).
- `extra_attributes` = JSON of every spec label the page printed that was not
  promoted. Audit it after the first full run and promote frequent keys.

## Stores (`output/stores/<brand>.csv`, 15 columns)

```
snapshot_date, brand, store_name, store_type, address, city, state_region,
postal_code, country_code, phone, email, latitude, longitude, website, store_url
```

`store_type` keeps the site's own vocabulary (boutique, authorized dealer,
department store, service center). Counts from an N-nearest-capped locator
are a floor and must be labelled as such wherever they are compared.

## Retailer listings (`output/retailers/<retailer>.csv`, 20 columns)

```
snapshot_date, retailer, brand, matched_sku, match_method, match_score,
retailer_product_id, manufacturer_ref, name, gender,
price_value, list_price_value, is_on_sale, price_currency,
availability, category, collection,
image_url, product_url, raw_identifiers
```

- One row per retailer page (= one colourway at most retailers).
- `match_method` ∈ {`exact_ref`, `composite`, `composite_ambiguous`, `unmatched`};
  never silently guess. `raw_identifiers` keeps every retailer-side id verbatim
  so matching can be re-run offline without re-scraping.

## Derived tables (built from the per-brand files, never scraped)

- **Active catalog** (`products_all.csv`): one row per (brand, sku) from each
  brand's newest *substantial* snapshot window. Window rule: anchor at the
  newest date whose block holds >= 50% of the largest block on record (a full
  refresh), every later date joins it, latest row per SKU wins inside the
  window. A `--mode new` top-up alone must never collapse the active catalog
  (measured: a 58-SKU top-up once shrank Seiko from 307 to 58 rows under a
  "newest date wins" rule).
- **SKU lifecycle** (`sku_lifecycle.csv`): one row per (brand, sku) ever seen:
  `first_seen`, `last_seen`, `snapshot_count`, `is_active`, `first_price`,
  `latest_price`. Delistings and price movement live here.
- Never derive "latest row per SKU across all history": it resurrects
  delisted products at stale prices (Hamilton: 182 dropped models reappeared
  priced from a month earlier).

```python
# active window + lifecycle in ~20 lines
import pandas as pd
df = pd.read_csv("output/brands/hamilton.csv", encoding="utf-8-sig")
counts = df.groupby("snapshot_date").size().sort_index()
anchor = max(d for d, n in counts.items() if n >= 0.5 * counts.max())
active = (df[df.snapshot_date >= anchor].sort_values("snapshot_date")
            .drop_duplicates("sku", keep="last"))
life = (df.sort_values("snapshot_date").groupby("sku")
          .agg(first_seen=("snapshot_date", "first"), last_seen=("snapshot_date", "last"),
               snapshot_count=("snapshot_date", "nunique"),
               first_price=("price_value", "first"), latest_price=("price_value", "last")))
life["is_active"] = life.index.isin(active.sku)
```