diff --git a/CHANGELOG.md b/CHANGELOG.md
index 06d30f8..dd27fd1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,48 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.7.0] — 2026-06-03
+
+Generalized, named, configurable shift/period splitting — so usage can be
+compared across operator-defined shifts (day vs night, A/B/C), with windows
+that may wrap past midnight. Parity-tested Python↔JS.
+
+### Added — shift splitting (`--split-by shifts`)
+- **`--split-by shifts`** activates named shift mode, with **`--shifts
+  "name=HH:MM-HH:MM,..."`** (comma-separated; a window where `end<=start` wraps
+  past midnight, e.g. `night=18:00-06:00`) or the JSON **`--shifts-file`**
+  (`{"shifts":[{"name","start","end"}]}`, mirroring `--rules-file`). With neither
+  flag the default is `day=06:00-18:00,night=18:00-06:00`.
+- **Timezone-correct windows:** shift windows are evaluated in the report
+  timezone (`--tz`), localizing each UTC record before applying the `HH:MM`
+  rule (UTC if `--tz` unset, and the output says so). Validated on a real
+  Central-time session: day/night split lands exactly on 06:00 / 18:00 Central.
+- **Two outputs:**
+  - `shift_comparison.csv` + `shift_comparison.json` — the headline per-shift
+    aggregate (records, hours, kWh, P avg/min/max, peak rolling demand, PF avg,
+    V_LN avg/p5/p95, V_THD p95, event counts, outage minutes), gathering each
+    shift's non-contiguous records across the whole session.
+  - Per-occurrence contiguous buckets under `<out>/shifts/<name>_<date>/`
+    (session.csv, events.json+ITIC, summary.txt); a midnight-spanning night is
+    one occurrence labeled by its start date.
+- Gap/overlap validation (warns when windows don't tile 24 h), first-matching
+  window wins on overlap, and an `unassigned` shift for records matching none.
+- Shift-comparison table embedded in `summary.txt`.
+
+### Added — model + parity
+- `analysis.py`: `Shift` / `ShiftSet` (parse / from_spec / default /
+  coverage_issues), `gather_store` (non-contiguous index slicing),
+  `aggregate_shifts`, `shift_occurrences`, `shift_comparison_rows`; new
+  `shifts_file.py` loader.
+- `web/analysis.js`: full JS port (`ShiftSet`, `localMinuteOfDay`,
+  `aggregateShifts`, `shiftOccurrences`, `shiftComparisonRows`) with a
+  Python↔JS parity test (`web/tests/shifts_parity.test.js`) on a shared
+  synthetic multi-day fixture.
+- Docs: new [`docs/SHIFTS.md`](docs/SHIFTS.md); README options table updated.
+
+### Tests
+- Python 251 passing (was 223); web 120 passing (was 114).
+
 ## [0.6.0] — 2026-06-03
 
 Web parity + analysis depth. Closes the web memory/parity gap from 0.5.0 and
diff --git a/README.md b/README.md
index dcacdd3..02139ca 100644
--- a/README.md
+++ b/README.md
@@ -91,7 +91,9 @@ fluke-analyze path/to/ES.NNN -o output/ \
 | `--max-csv-rows N` | off | Cap the full CSV at ~N rows for week-long sessions; stride is auto-raised and the downsampling is logged. 1-min CSV + analysis keep full resolution. |
 | `--anchor-start ISO_TIME` | off | Pin the real wall-clock **start** to correct a wrong meter RTC (shifts every timestamp). Mutually exclusive with `--anchor-end`. |
 | `--anchor-end ISO_TIME` | off | Pin the real wall-clock **end** (e.g. a known shutdown time). |
-| `--split-by PERIOD` | off | Partition into time buckets (`hour`\|`day`\|`week` or a duration like `30m`/`6h`/`2d`). Emits a full per-bucket report (`<label>/`) + `buckets_summary.csv` roll-up. |
+| `--split-by PERIOD` | off | Partition into time buckets (`hour`\|`day`\|`week`, a duration like `30m`/`6h`/`2d`, or `shifts` for named shift windows). Emits a full per-bucket report (`<label>/`) + `buckets_summary.csv` roll-up. |
+| `--shifts SPEC` | `day=06:00-18:00,night=18:00-06:00` | With `--split-by shifts`: named windows `name=HH:MM-HH:MM,...`. `end<=start` wraps past midnight. Evaluated in `--tz`. Emits `shift_comparison.csv`/`.json` + per-occurrence reports. See [`docs/SHIFTS.md`](docs/SHIFTS.md). |
+| `--shifts-file FILE` | off | JSON shift schedule (`{"shifts":[{"name","start","end"}]}`); alternative to `--shifts`. |
 | `--mark "ISO=LABEL"` | | Add an event marker (repeatable); cross-referenced to nearest events in `markers.json`. |
 | `--marks FILE.csv` | | Load markers from a CSV (`time,label`). |
 | `--tod-profile [HH:MM-HH:MM]` | off | Time-of-day (diurnal) profile — avg/min/max envelope per bin across all days. Bare flag = 24 h. Writes `time_of_day_profile.csv` + an XLSX sheet. |
diff --git a/ROADMAP.md b/ROADMAP.md
index 09aa8ee..0d127c8 100644
--- a/ROADMAP.md
+++ b/ROADMAP.md
@@ -4,6 +4,27 @@ Shipped releases live in [CHANGELOG.md](CHANGELOG.md). This file is the
 backlog of ideas we've discussed but not yet committed to a specific
 release.
 
+## Shipped in v0.7 — generalized shift splitting
+
+Operator-defined, named, midnight-wrapping shift windows so usage can be
+compared across shifts (day vs night, A/B/C), parity-tested Python↔JS:
+
+- **`--split-by shifts`** + `--shifts "name=HH:MM-HH:MM,..."` / `--shifts-file`
+  (JSON). Windows that wrap past midnight; default day/night.
+- Windows evaluated in the **report timezone** (`--tz`), not raw UTC — verified
+  on a real Central-time session.
+- Headline `shift_comparison.csv`/`.json` aggregate per shift name (energy,
+  power, peak demand, PF, V_LN/THD percentiles, event counts, outage minutes)
+  plus per-occurrence contiguous buckets. This delivers the shift slice of the
+  open **comparison polish (Theme C)** backlog.
+- New model in `analysis.py` (`Shift`/`ShiftSet`, `gather_store`,
+  `aggregate_shifts`, `shift_occurrences`, `shift_comparison_rows`) + JS port +
+  parity test + `docs/SHIFTS.md`.
+
+Still open from comparison polish: cross-session / period-over-period diffing
+and a web UI surface for shifts (the core logic + parity are shipped; web
+rendering is best-effort).
+
 ## Shipped in v0.6 — web parity + analysis depth
 
 Closed the v0.5 web memory/parity gap and added standards-grade analysis, all
diff --git a/docs/SHIFTS.md b/docs/SHIFTS.md
new file mode 100644
index 0000000..6c76af1
--- /dev/null
+++ b/docs/SHIFTS.md
@@ -0,0 +1,112 @@
+# Generalized shift splitting (`--split-by shifts`)
+
+Define **named time periods within a day** — which may wrap past midnight — and
+compare power usage across them (day vs night, or A/B/C shifts). This is the
+generalized successor to the clock-aligned `--split-by hour|day|week`: instead of
+fixed grid buckets, you define the windows that match how the plant actually
+runs.
+
+```bash
+# Default day/night
+fluke-analyze ES.004 --split-by shifts --tz America/Chicago
+
+# Explicit two-shift
+fluke-analyze ES.004 --split-by shifts \
+  --shifts "day=06:00-18:00,night=18:00-06:00" --tz America/Chicago
+
+# Three 8-hour shifts
+fluke-analyze ES.004 --split-by shifts \
+  --shifts "A=06:00-14:00,B=14:00-22:00,C=22:00-06:00" --tz America/Chicago
+
+# From a committed schedule file
+fluke-analyze ES.004 --split-by shifts --shifts-file shifts.json --tz America/Chicago
+```
+
+## Defining shifts
+
+A shift is `name=HH:MM-HH:MM`. Multiple shifts are comma-separated. **A window
+where `end <= start` wraps past midnight**: `night=18:00-06:00` covers
+`[18:00, 24:00)` plus `[00:00, 06:00)`. Windows are half-open: start inclusive,
+end exclusive.
+
+- **Two shifts:** `day=06:00-18:00,night=18:00-06:00` (the default when
+  `--split-by shifts` is given with no `--shifts`/`--shifts-file`).
+- **Three shifts:** `A=06:00-14:00,B=14:00-22:00,C=22:00-06:00`.
+
+### File form (`--shifts-file`)
+
+```json
+{
+  "shifts": [
+    {"name": "day",   "start": "06:00", "end": "18:00"},
+    {"name": "night", "start": "18:00", "end": "06:00"}
+  ]
+}
+```
+
+A bare top-level list of `{name, start, end}` objects is also accepted.
+
+## Timezone contract (important)
+
+Captured timestamps are stored in **UTC**. Shift windows are interpreted in the
+**report timezone** set by `--tz` (e.g. `America/Chicago`). Each record's
+timestamp is localized to that zone *before* the `HH:MM` rule is applied. If you
+omit `--tz`, windows are evaluated in **UTC** and the output says so. Always pass
+`--tz` so a `06:00-18:00` shift means 6 AM local, not 6 AM UTC.
+
+Validated on a real Central-time session: `day=06:00-18:00,night=18:00-06:00`
+splits exactly on 06:00 / 18:00 **Central**, 43 200 one-second records per
+half-day.
+
+## Assignment rules
+
+- Each record is assigned to the **first** window it matches (so overlapping
+  windows resolve deterministically).
+- Records that match **no** window go to an `unassigned` shift, with a printed
+  notice.
+- If the windows don't tile 24 h, a gap/overlap **warning** is printed to
+  stderr (and recorded in `shift_comparison.json → coverage_issues`). Shifts
+  ideally tile the day with no gaps or overlaps.
+
+## Outputs
+
+### 1. `shift_comparison.csv` / `shift_comparison.json` — the headline
+
+One row **per shift name**, aggregating **all** records of that shift across the
+whole session (a "night" recurs daily; those non-contiguous records are gathered
+into one row). This is what lets you compare day vs night usage.
+
+| Column | Meaning |
+|---|---|
+| `shift` | shift name (or `unassigned`) |
+| `window` | the `HH:MM-HH:MM` window |
+| `records` | number of records in the shift |
+| `hours` | hours covered (1 record = 1 s) |
+| `kWh` | energy = mean power × hours |
+| `P_total_avg_W` / `P_total_min_W` / `P_total_max_W` | real-power avg/min/max |
+| `peak_demand_kW` | peak rolling demand within the shift (15 min default; from `--demand-window`) |
+| `peak_demand_window_secs` | the demand window used |
+| `PF_avg` | average power factor |
+| `V_LN_avg_V` / `V_LN_p5_V` / `V_LN_p95_V` | L-N voltage avg + p5/p95 (outage zeros excluded) |
+| `V_THD_p95_pct` | 95th-percentile voltage THD |
+| `n_outages` / `n_dips` / `n_swells` | event counts (filed by event start time) |
+| `outage_minutes` | total outage minutes in the shift |
+
+The JSON also carries `tz`, `spec`, `demand_window_secs`, and `coverage_issues`.
+
+### 2. Per-occurrence buckets — `<out>/shifts/<name>_<date>/`
+
+Each **individual** shift instance becomes its own contiguous, time-ordered
+bucket reusing the standard per-bucket report machinery (`session.csv`,
+`events.json` with ITIC, `summary.txt`). A night spanning midnight is **one**
+occurrence labeled by its **start date** (e.g. `night 2026-05-29`).
+
+A shift-comparison table is also embedded in the top-level `summary.txt`.
+
+## Parity
+
+The shift model (parse, midnight-wrap, tz-localized assignment, aggregate,
+occurrences, comparison) is ported to the web side (`web/analysis.js`:
+`ShiftSet`, `aggregateShifts`, `shiftOccurrences`, `shiftComparisonRows`) and
+covered by a Python↔JS parity test on a shared synthetic multi-day fixture
+(`web/tests/shifts_parity.test.js` vs `analysis_golden.json`).
diff --git a/python/pyproject.toml b/python/pyproject.toml
index f9d7139..50bb0bb 100644
--- a/python/pyproject.toml
+++ b/python/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "fluke-3540-analyzer"
-version = "0.6.0"
+version = "0.7.0"
 description = "Parser, event detector, and chart generator for Fluke 3540 FC three-phase power-quality sessions. See https://github.com/GrumpyTanker/fluke-3540-analyzer."
 requires-python = ">=3.10"
 license = {text = "MIT"}
diff --git a/python/src/fluke_3540/__init__.py b/python/src/fluke_3540/__init__.py
index 4adaa41..7350e37 100644
--- a/python/src/fluke_3540/__init__.py
+++ b/python/src/fluke_3540/__init__.py
@@ -4,4 +4,4 @@
 The canonical field map and binary layout live in spec/field_map.json at
 the repo root and are shared with the JavaScript port. See README.md.
 """
-__version__ = "0.6.0"
+__version__ = "0.7.0"
diff --git a/python/src/fluke_3540/analysis.py b/python/src/fluke_3540/analysis.py
index d99de0a..8b01ee4 100644
--- a/python/src/fluke_3540/analysis.py
+++ b/python/src/fluke_3540/analysis.py
@@ -610,6 +610,345 @@ def slice_store(store: ColumnStore, lo: int, hi: int) -> ColumnStore:
     return sub
 
 
+def gather_store(store: ColumnStore, indices: Sequence[int]) -> ColumnStore:
+    """Like :func:`slice_store` but over an arbitrary (possibly non-contiguous)
+    list of record indices.
+
+    Used by shift aggregation: a "night" shift recurs daily, so its records
+    are scattered through the store rather than forming one contiguous range.
+    ``indices`` are taken in the given order (callers pass them ascending so
+    timestamps stay monotonic, which keeps kWh/demand roll-ups well-defined).
+    """
+    sub = ColumnStore(time_shift=store.time_shift)
+    cols = store.columns
+    src = {name: store._cols[name] for name in cols}
+    dst = {name: sub._cols[name] for name in cols}
+    s_ticks = store._start_ticks
+    e_ticks = store._end_ticks
+    for i in indices:
+        for name in cols:
+            dst[name].append(src[name][i])
+        sub._start_ticks.append(s_ticks[i])
+        sub._end_ticks.append(e_ticks[i])
+    sub._n = len(sub._start_ticks)
+    return sub
+
+
+# --- Generalized named shift/period splitting (--split-by shifts) ------------
+
+UNASSIGNED_SHIFT = "unassigned"
+
+
+def _parse_hhmm(text: str) -> int:
+    """Parse 'HH:MM' (colon required) to minute-of-day in [0, 1440].
+
+    '24:00' is accepted as end-of-day (1440). Raises ValueError otherwise.
+    """
+    s = text.strip()
+    if ":" not in s:
+        raise ValueError(f"shift time must be HH:MM (with a colon): {text!r}")
+    hh, _, mm = s.partition(":")
+    if not (hh.isdigit() and mm.isdigit()):
+        raise ValueError(f"shift time must be numeric HH:MM: {text!r}")
+    h = int(hh)
+    m = int(mm)
+    if m >= 60:
+        raise ValueError(f"shift minutes out of range in {text!r}")
+    total = h * 60 + m
+    if total < 0 or total > 1440:
+        raise ValueError(f"shift time out of range (00:00..24:00): {text!r}")
+    return total
+
+
+@dataclass(frozen=True)
+class Shift:
+    """One named shift window, in minutes-of-day (report-tz wall clock).
+
+    ``end_min <= start_min`` means the window wraps past midnight, e.g.
+    18:00-06:00 covers [18:00, 24:00) plus [00:00, 06:00).
+    """
+    name: str
+    start_min: int
+    end_min: int
+
+    @property
+    def wraps(self) -> bool:
+        return self.end_min <= self.start_min
+
+    def contains_minute(self, mod: int) -> bool:
+        """Is minute-of-day ``mod`` inside this window? start inclusive, end
+        exclusive; wrap-aware."""
+        if not self.wraps:
+            return self.start_min <= mod < self.end_min
+        # Wrapping: [start, 1440) U [0, end)
+        return mod >= self.start_min or mod < self.end_min
+
+    def length_minutes(self) -> int:
+        if self.wraps:
+            return (1440 - self.start_min) + self.end_min
+        return self.end_min - self.start_min
+
+    @property
+    def window_str(self) -> str:
+        def fmt(m: int) -> str:
+            return f"{(m // 60) % 24:02d}:{m % 60:02d}" if m != 1440 else "24:00"
+        return f"{fmt(self.start_min)}-{fmt(self.end_min)}"
+
+
+@dataclass(frozen=True)
+class ShiftSet:
+    """An ordered collection of named shift windows.
+
+    Order matters: each record is assigned to the FIRST window it matches, so
+    overlaps resolve deterministically.
+    """
+    shifts: tuple[Shift, ...]
+
+    @classmethod
+    def parse(cls, text: str) -> "ShiftSet":
+        """Parse 'name=HH:MM-HH:MM,name=HH:MM-HH:MM,...'."""
+        out: list[Shift] = []
+        seen: set[str] = set()
+        parts = [p.strip() for p in text.split(",") if p.strip()]
+        if not parts:
+            raise ValueError(f"no shifts parsed from {text!r}")
+        for part in parts:
+            name, sep, window = part.partition("=")
+            name = name.strip()
+            if not sep or not name:
+                raise ValueError(
+                    f"shift must be name=HH:MM-HH:MM: {part!r}")
+            a, dash, b = window.partition("-")
+            if not dash:
+                raise ValueError(
+                    f"shift window must be HH:MM-HH:MM: {window!r}")
+            start = _parse_hhmm(a)
+            end = _parse_hhmm(b)
+            if start == end:
+                raise ValueError(
+                    f"shift {name!r} has a zero-length window {window!r}")
+            if name in seen:
+                raise ValueError(f"duplicate shift name {name!r}")
+            seen.add(name)
+            out.append(Shift(name, start, end))
+        return cls(tuple(out))
+
+    @classmethod
+    def from_spec(cls, spec: Sequence[dict]) -> "ShiftSet":
+        """Build from a list of {name, start, end} dicts (the --shifts-file form)."""
+        text = ",".join(
+            f"{s['name']}={s['start']}-{s['end']}" for s in spec)
+        return cls.parse(text)
+
+    @classmethod
+    def default(cls) -> "ShiftSet":
+        return cls.parse("day=06:00-18:00,night=18:00-06:00")
+
+    # --- validation --------------------------------------------------------
+    def coverage_issues(self) -> list[str]:
+        """Warn if the windows don't tile 24 h (gaps or overlaps).
+
+        Walks minute-of-day 0..1439 counting how many windows cover each
+        minute. Reports the count of uncovered minutes (gap) and doubly-covered
+        minutes (overlap). Empty list = clean 24 h tiling.
+        """
+        cover = [0] * 1440
+        for sh in self.shifts:
+            for m in range(1440):
+                if sh.contains_minute(m):
+                    cover[m] += 1
+        gap = sum(1 for c in cover if c == 0)
+        overlap = sum(1 for c in cover if c > 1)
+        issues: list[str] = []
+        if gap:
+            issues.append(
+                f"{gap} minute(s)/day fall in NO shift window (gap); those "
+                f"records go to '{UNASSIGNED_SHIFT}'.")
+        if overlap:
+            issues.append(
+                f"{overlap} minute(s)/day are covered by MORE THAN ONE shift "
+                "(overlap); the first matching window wins.")
+        return issues
+
+
+def _localize_minute(t: dt.datetime, tz) -> int:
+    """Minute-of-day for ``t`` in the report timezone ``tz`` (UTC if None)."""
+    if tz is not None:
+        t = t.astimezone(tz)
+    return t.hour * 60 + t.minute
+
+
+def aggregate_shifts(store: ColumnStore, ss: ShiftSet, tz=None) -> dict[str, list[int]]:
+    """Group record indices by shift NAME, evaluating windows in ``tz``.
+
+    Returns {shift_name: [ascending indices]}. Each named shift present in
+    ``ss`` always has a key (possibly empty). Records matching no window are
+    collected under ``UNASSIGNED_SHIFT`` (only added when non-empty).
+    """
+    out: dict[str, list[int]] = {sh.name: [] for sh in ss.shifts}
+    for i in range(store.n):
+        mod = _localize_minute(store.start(i), tz)
+        for sh in ss.shifts:
+            if sh.contains_minute(mod):
+                out[sh.name].append(i)
+                break
+        else:
+            out.setdefault(UNASSIGNED_SHIFT, []).append(i)
+    return out
+
+
+def shift_occurrences(store: ColumnStore, ss: ShiftSet, tz=None
+                      ) -> list[tuple[str, str, int, int]]:
+    """Partition records into contiguous per-OCCURRENCE shift buckets.
+
+    Returns a time-ordered list of (label, shift_name, lo, hi_exclusive). A
+    new occurrence starts whenever the matched shift name changes between
+    consecutive records. A wrap-past-midnight window stays ONE occurrence
+    (the name does not change), labeled by the occurrence's START date in the
+    report timezone, e.g. ``"night 2026-05-29"``.
+    """
+    n = store.n
+    if n == 0:
+        return []
+
+    def name_at(i: int) -> str:
+        mod = _localize_minute(store.start(i), tz)
+        for sh in ss.shifts:
+            if sh.contains_minute(mod):
+                return sh.name
+        return UNASSIGNED_SHIFT
+
+    def start_date(i: int) -> str:
+        t = store.start(i)
+        if tz is not None:
+            t = t.astimezone(tz)
+        return t.strftime("%Y-%m-%d")
+
+    out: list[tuple[str, str, int, int]] = []
+    cur_name = name_at(0)
+    lo = 0
+    for i in range(1, n):
+        nm = name_at(i)
+        if nm != cur_name:
+            out.append((f"{cur_name} {start_date(lo)}", cur_name, lo, i))
+            cur_name = nm
+            lo = i
+    out.append((f"{cur_name} {start_date(lo)}", cur_name, lo, n))
+    return out
+
+
+def shift_comparison_rows(store: ColumnStore, ss: ShiftSet, events: Sequence,
+                          tz=None, nominal_ln_v: float = 277.0,
+                          demand_window: int = 15) -> list[dict]:
+    """The headline per-shift-name aggregate comparison.
+
+    One row per named shift (plus ``unassigned`` if any records land there),
+    aggregating ALL records of that shift across the whole session. ``events``
+    are filed to a shift by the shift their ``t_start`` falls in.
+
+    ``demand_window`` is in MINUTES (peak rolling demand within the shift's
+    records). Per-row energy/avg/percentiles are memory-bounded.
+    """
+    by_name = aggregate_shifts(store, ss, tz=tz)
+    win_by_name = {sh.name: sh for sh in ss.shifts}
+
+    # File events to a shift by their start time's window.
+    ev_by_name: dict[str, list] = {nm: [] for nm in by_name}
+    for e in events:
+        mod = _localize_minute(e.t_start, tz)
+        placed = False
+        for sh in ss.shifts:
+            if sh.contains_minute(mod):
+                ev_by_name.setdefault(sh.name, []).append(e)
+                placed = True
+                break
+        if not placed:
+            ev_by_name.setdefault(UNASSIGNED_SHIFT, []).append(e)
+
+    rows: list[dict] = []
+    for name, idxs in by_name.items():
+        sub = gather_store(store, idxs)
+        sh = win_by_name.get(name)
+        window = sh.window_str if sh is not None else "—"
+        bucket_events = ev_by_name.get(name, [])
+        rows.append(_shift_row(name, window, sub, bucket_events,
+                               nominal_ln_v, demand_window))
+    return rows
+
+
+def _shift_row(name: str, window: str, sub: ColumnStore, bucket_events,
+               nominal_ln_v: float, demand_window: int) -> dict:
+    """Aggregate one shift's records into a comparison row."""
+    nrec = sub.n
+    p = sub.col("P_total_avg_W")
+    pf = sub.col("PF_total_avg")
+    va = sub.col("V_LN_a_avg_V"); vb = sub.col("V_LN_b_avg_V"); vc = sub.col("V_LN_c_avg_V")
+    vth_a = sub.col("V_THD_pct_a_avg")
+    vth_b = sub.col("V_THD_pct_b_avg")
+    vth_c = sub.col("V_THD_pct_c_avg")
+
+    p_mom = _RunningMoments()
+    p_min = math.inf; p_max = -math.inf
+    pf_mom = _RunningMoments()
+    v_mom = _RunningMoments()
+    v_sketch = _PercentileSketch(0.0, 400.0)
+    vthd_sketch = _PercentileSketch(0.0, 50.0)
+    for i in range(nrec):
+        pv = p[i]
+        if math.isfinite(pv):
+            p_mom.add(pv)
+            p_min = min(p_min, pv); p_max = max(p_max, pv)
+        pfi = pf[i]
+        if math.isfinite(pfi):
+            pf_mom.add(pfi)
+        for vv in (va[i], vb[i], vc[i]):
+            if math.isfinite(vv) and vv > 50.0:  # ignore outage zeros
+                v_mom.add(vv); v_sketch.add(vv)
+        for tv in (vth_a[i], vth_b[i], vth_c[i]):
+            if math.isfinite(tv):
+                vthd_sketch.add(tv)
+
+    p_mean = p_mom.mean if p_mom.n else 0.0
+    hours = p_mom.n / 3600.0  # 1 record == 1 s
+    kwh = p_mean / 1000.0 * hours
+
+    # Peak rolling demand within the shift's gathered records.
+    demand = demand_analysis(sub, window_secs=max(1, demand_window * 60))
+
+    n_out = sum(1 for e in bucket_events if e.kind == "outage")
+    n_dip = sum(1 for e in bucket_events if e.kind == "dip")
+    n_swell = sum(1 for e in bucket_events if e.kind == "swell")
+    outage_minutes = sum(
+        (e.t_end - e.t_start).total_seconds() for e in bucket_events
+        if e.kind == "outage") / 60.0
+
+    def q(sketch, p_):
+        v = sketch.quantile(p_)
+        return 0.0 if (v != v) else v
+
+    return {
+        "shift": name,
+        "window": window,
+        "records": nrec,
+        "hours": hours,
+        "kWh": kwh,
+        "P_total_avg_W": p_mean,
+        "P_total_min_W": (p_min if p_min != math.inf else 0.0),
+        "P_total_max_W": (p_max if p_max != -math.inf else 0.0),
+        "peak_demand_kW": demand["peak_demand_kw"],
+        "peak_demand_window_secs": demand["window_secs"],
+        "PF_avg": (pf_mom.mean if pf_mom.n else 0.0),
+        "V_LN_avg_V": (v_mom.mean if v_mom.n else 0.0),
+        "V_LN_p5_V": q(v_sketch, 0.05),
+        "V_LN_p95_V": q(v_sketch, 0.95),
+        "V_THD_p95_pct": q(vthd_sketch, 0.95),
+        "n_outages": n_out,
+        "n_dips": n_dip,
+        "n_swells": n_swell,
+        "outage_minutes": outage_minutes,
+    }
+
+
 # --- Event markers / correlation (--mark / --marks) -------------------------
 
 @dataclass(frozen=True)
diff --git a/python/src/fluke_3540/cli.py b/python/src/fluke_3540/cli.py
index 628b7e1..33d20fd 100644
--- a/python/src/fluke_3540/cli.py
+++ b/python/src/fluke_3540/cli.py
@@ -163,8 +163,23 @@ def build_argparser() -> argparse.ArgumentParser:
     ap.add_argument("--split-by", dest="split_by", type=str, default=None,
                     metavar="PERIOD",
                     help="Partition the session into time buckets, emitting a full "
-                         "per-bucket report plus a roll-up. PERIOD: hour|day|week or "
-                         "a duration like 30m, 6h, 2d.")
+                         "per-bucket report plus a roll-up. PERIOD: hour|day|week, "
+                         "a duration like 30m, 6h, 2d, OR 'shifts' for named "
+                         "shift windows (see --shifts / --shifts-file).")
+
+    # Named, configurable shift windows (--split-by shifts)
+    ap.add_argument("--shifts", dest="shifts", type=str, default=None,
+                    metavar="SPEC",
+                    help="Shift windows as 'name=HH:MM-HH:MM,...' (comma-separated). "
+                         "A window where end<=start wraps past midnight "
+                         "(e.g. night=18:00-06:00). Interpreted in --tz (UTC if "
+                         "unset). Default day=06:00-18:00,night=18:00-06:00. "
+                         "Requires --split-by shifts.")
+    ap.add_argument("--shifts-file", dest="shifts_file", type=Path, default=None,
+                    metavar="FILE",
+                    help="JSON file of shift windows "
+                         "({\"shifts\":[{\"name\",\"start\",\"end\"}]}); an "
+                         "alternative to --shifts. See docs/SHIFTS.md.")
 
     # Event markers / correlation
     ap.add_argument("--mark", action="append", default=None, metavar="ISO=LABEL",
@@ -592,6 +607,113 @@ def _run_split_by(args: argparse.Namespace, outdir: Path, store: ColumnStore,
           f"per-bucket reports under {outdir}/<label>/")
 
 
+def _resolve_shift_set(args: argparse.Namespace):
+    """Build the ShiftSet from --shifts / --shifts-file, or the default."""
+    from .analysis import ShiftSet
+    if getattr(args, "shifts_file", None):
+        from .shifts_file import load_shifts
+        return load_shifts(args.shifts_file), f"file:{args.shifts_file}"
+    if getattr(args, "shifts", None):
+        return ShiftSet.parse(args.shifts), args.shifts
+    return ShiftSet.default(), "default(day=06:00-18:00,night=18:00-06:00)"
+
+
+def _run_shifts(args: argparse.Namespace, outdir: Path, store: ColumnStore,
+                events: Sequence[Event], config: dict,
+                nominal_ln_v: float) -> None:
+    """Generalized named-shift splitting (--split-by shifts).
+
+    Emits (1) the headline per-shift-name aggregate comparison
+    (shift_comparison.csv + .json) and (2) per-occurrence buckets reusing the
+    standard per-bucket report machinery. Windows are evaluated in --tz.
+    """
+    from .analysis import (event_itic, shift_comparison_rows,
+                           shift_occurrences, slice_store)
+    from .tzutil import tz_label
+
+    tz = getattr(args, "_tz", None)
+    tz_name = getattr(args, "tz", None)
+    ss, spec_label = _resolve_shift_set(args)
+    demand_min = max(1, getattr(args, "demand_window", 900) // 60)
+
+    print(f"[shifts] --split-by shifts ({spec_label}); windows in "
+          f"{tz_label(tz, tz_name)}")
+    for issue in ss.coverage_issues():
+        print(f"[shifts]   WARNING: {issue}", file=sys.stderr)
+
+    # (1) Headline aggregate comparison ------------------------------------
+    rows = shift_comparison_rows(store, ss, events, tz=tz,
+                                 nominal_ln_v=nominal_ln_v,
+                                 demand_window=demand_min)
+    cols = ["shift", "window", "records", "hours", "kWh", "P_total_avg_W",
+            "P_total_min_W", "P_total_max_W", "peak_demand_kW",
+            "peak_demand_window_secs", "PF_avg", "V_LN_avg_V", "V_LN_p5_V",
+            "V_LN_p95_V", "V_THD_p95_pct", "n_outages", "n_dips", "n_swells",
+            "outage_minutes"]
+    with (outdir / "shift_comparison.csv").open("w", newline="", encoding="utf-8") as fh:
+        w = _csv.writer(fh)
+        w.writerow(cols)
+        for r in rows:
+            w.writerow([_fmt_shift_cell(c, r[c]) for c in cols])
+    payload = {
+        "tz": tz_label(tz, tz_name),
+        "spec": spec_label,
+        "demand_window_secs": demand_min * 60,
+        "coverage_issues": ss.coverage_issues(),
+        "shifts": rows,
+    }
+    (outdir / "shift_comparison.json").write_text(
+        json.dumps(payload, indent=2), encoding="utf-8")
+    for r in rows:
+        print(f"[shifts]   {r['shift']:10s} {r['window']:>13s}  "
+              f"{r['records']:>7d} rec  {r['hours']:6.2f} h  "
+              f"{r['kWh']:9.2f} kWh  Pavg={r['P_total_avg_W']/1000:7.2f} kW  "
+              f"peak={r['peak_demand_kW']:7.2f} kW")
+
+    # (2) Per-occurrence buckets (contiguous) ------------------------------
+    occurrences = shift_occurrences(store, ss, tz=tz)
+    for (label, name, lo, hi) in occurrences:
+        sub = slice_store(store, lo, hi)
+        safe = label.replace(" ", "_").replace(":", "")
+        bdir = outdir / "shifts" / safe
+        bdir.mkdir(parents=True, exist_ok=True)
+        b_start = store.start(lo)
+        b_end = store.end(hi - 1)
+        bucket_events = [e for e in events if b_start <= e.t_start < b_end]
+        _write_bucket_csv(bdir / "session.csv", sub)
+        ev_payload = []
+        for e in bucket_events:
+            d = _event_to_json(e)
+            info = event_itic(e, nominal_ln_v)
+            if info:
+                d["itic"] = info
+            ev_payload.append(d)
+        (bdir / "events.json").write_text(json.dumps(ev_payload, indent=2),
+                                          encoding="utf-8")
+        b_findings = analyze_insights(sub, bucket_events, [], config)
+        b_snaps = pick_snapshots(sub, bucket_events, n=1)
+        _write_summary_txt(bdir, bucket_events, b_snaps, b_findings, config,
+                           tz=tz, tz_name=tz_name, store=sub)
+    print(f"[shifts] wrote shift_comparison.csv/json + "
+          f"{len(occurrences)} per-occurrence report(s) under {outdir}/shifts/")
+    return rows
+
+
+def _fmt_shift_cell(col: str, val):
+    """Format one comparison-CSV cell (round floats; pass ints/strings)."""
+    if col in ("shift", "window") or isinstance(val, int):
+        return val
+    if col == "kWh":
+        return f"{val:.2f}"
+    if col in ("hours", "outage_minutes"):
+        return f"{val:.2f}"
+    if col == "PF_avg":
+        return f"{val:.3f}"
+    if col == "peak_demand_kW":
+        return f"{val:.2f}"
+    return f"{val:.1f}"
+
+
 def _write_bucket_csv(path: Path, sub: ColumnStore) -> None:
     """Write a CSV for a bucket slice from the store's retained columns."""
     cols = list(sub.columns)
@@ -660,13 +782,18 @@ def _run_extra_analyses(args: argparse.Namespace, outdir: Path,
     if getattr(args, "tod_profile", None):
         tod_rows = _write_tod(outdir, store, args)
 
+    shift_rows = None
     if getattr(args, "split_by", None):
-        _run_split_by(args, outdir, store, events, config, nominal_ln_v)
+        if str(args.split_by).strip().lower() == "shifts":
+            shift_rows = _run_shifts(args, outdir, store, events, config,
+                                     nominal_ln_v)
+        else:
+            _run_split_by(args, outdir, store, events, config, nominal_ln_v)
 
     # Auto-narrative / executive summary (Feature E) — needs stats + ct.
     narrative = _write_narrative(outdir, store, events, findings, stats, ct, config)
 
-    return stats, tod_rows, narrative, demand
+    return stats, tod_rows, narrative, demand, shift_rows
 
 
 def _write_narrative(outdir: Path, store: ColumnStore, events, findings,
@@ -692,7 +819,8 @@ def _write_summary_txt(outdir: Path, events: Sequence[Event],
                        config: dict,
                        narrative: str | None = None,
                        tz=None, tz_name: str | None = None,
-                       store: "ColumnStore | None" = None) -> None:
+                       store: "ColumnStore | None" = None,
+                       shift_rows: "list[dict] | None" = None) -> None:
     lines: list[str] = ["Fluke 3540 FC Session Summary", "=" * 32, ""]
     if narrative:
         lines.append("Executive Summary")
@@ -720,6 +848,17 @@ def _write_summary_txt(outdir: Path, events: Sequence[Event],
         for f in findings:
             lines.append(f"  [{f.severity:5s}] {f.kind:25s}  {f.headline}")
         lines.append("")
+    if shift_rows:
+        lines.append("Shift comparison")
+        lines.append("-" * 16)
+        lines.append(f"  {'shift':<10} {'window':>13}  {'records':>8}  "
+                     f"{'kWh':>9}  {'Pavg_kW':>8}  {'peak_kW':>8}  {'PF':>5}")
+        for r in shift_rows:
+            lines.append(
+                f"  {r['shift']:<10} {r['window']:>13}  {r['records']:>8d}  "
+                f"{r['kWh']:>9.2f}  {r['P_total_avg_W']/1000:>8.2f}  "
+                f"{r['peak_demand_kW']:>8.2f}  {r['PF_avg']:>5.3f}")
+        lines.append("")
     lines.append(f"Events detected: {len(events)}")
     for ev in events:
         phases = "/".join(ev.affected_phases) or "—"
@@ -978,12 +1117,13 @@ def print(*a, **kw):  # noqa: A001 — intentional shadow
 
         # Post-detection analysis features (markers, stats, tod, split) run on
         # the in-memory store + on-disk CSVs.
-        stats, tod_rows, narrative, demand = _run_extra_analyses(
+        stats, tod_rows, narrative, demand, shift_rows = _run_extra_analyses(
             args, outdir, store, events, findings, config, full_csv, min_csv)
         # Re-write summary.txt with the executive narrative + tz-aware time range.
         _write_summary_txt(outdir, events, snaps, findings, config,
                            narrative=narrative, tz=getattr(args, "_tz", None),
-                           tz_name=getattr(args, "tz", None), store=store)
+                           tz_name=getattr(args, "tz", None), store=store,
+                           shift_rows=shift_rows)
 
         if getattr(args, "json_mode", False):
             _emit_json(events, snaps, findings, config)
diff --git a/python/src/fluke_3540/shifts_file.py b/python/src/fluke_3540/shifts_file.py
new file mode 100644
index 0000000..5545857
--- /dev/null
+++ b/python/src/fluke_3540/shifts_file.py
@@ -0,0 +1,45 @@
+"""Shift-window definitions loaded from a JSON file (``--shifts-file``).
+
+Mirrors the ``--rules-file`` pattern: an alternative to the inline ``--shifts``
+string for committing a reusable shift schedule alongside an asset.
+
+JSON form::
+
+    {
+      "shifts": [
+        {"name": "day",   "start": "06:00", "end": "18:00"},
+        {"name": "night", "start": "18:00", "end": "06:00"}
+      ]
+    }
+
+A bare top-level list of ``{name,start,end}`` dicts is also accepted. The same
+HH:MM rules and wrap-past-midnight semantics as :class:`ShiftSet` apply.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from .analysis import ShiftSet
+
+
+def load_shifts(path: Path) -> ShiftSet:
+    """Build a :class:`ShiftSet` from a JSON shifts file."""
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(raw, dict):
+        spec = raw.get("shifts")
+        if spec is None:
+            raise ValueError(
+                f"{path}: shifts file must have a top-level 'shifts' list "
+                "(or be a bare list of {name,start,end} objects)")
+    elif isinstance(raw, list):
+        spec = raw
+    else:
+        raise ValueError(f"{path}: shifts file must be an object or a list")
+    if not isinstance(spec, list) or not spec:
+        raise ValueError(f"{path}: 'shifts' must be a non-empty list")
+    for i, s in enumerate(spec):
+        if not isinstance(s, dict) or not {"name", "start", "end"} <= set(s):
+            raise ValueError(
+                f"{path}: shift #{i} must be an object with name/start/end")
+    return ShiftSet.from_spec(spec)
diff --git a/python/tests/fixtures/analysis_golden.json b/python/tests/fixtures/analysis_golden.json
index 2a3ec33..a1aed2e 100644
--- a/python/tests/fixtures/analysis_golden.json
+++ b/python/tests/fixtures/analysis_golden.json
@@ -429,6 +429,163 @@
       }
     ]
   },
+  "shifts": {
+    "base_epoch_ms": 1779840000000,
+    "n_records": 4320,
+    "comparison_utc": [
+      {
+        "shift": "day",
+        "window": "06:00-18:00",
+        "records": 2160,
+        "hours": 0.6,
+        "kWh": 22.46690277777777,
+        "P_total_avg_W": 37444.83796296295,
+        "P_total_min_W": 35000.0,
+        "P_total_max_W": 39800.0,
+        "peak_demand_kW": 37.52005555555556,
+        "peak_demand_window_secs": 900,
+        "PF_avg": 0.9399999976158142,
+        "V_LN_avg_V": 277.00500000847757,
+        "V_LN_p5_V": 275.05,
+        "V_LN_p95_V": 279.05,
+        "V_THD_p95_pct": 4.5062500000000005,
+        "n_outages": 0,
+        "n_dips": 0,
+        "n_swells": 0,
+        "outage_minutes": 0.0
+      },
+      {
+        "shift": "night",
+        "window": "18:00-06:00",
+        "records": 2160,
+        "hours": 0.6,
+        "kWh": 19.396847222222263,
+        "P_total_avg_W": 32328.078703703777,
+        "P_total_min_W": 30000.0,
+        "P_total_max_W": 34800.0,
+        "peak_demand_kW": 32.431444444444445,
+        "peak_demand_window_secs": 900,
+        "PF_avg": 0.9399999976158142,
+        "V_LN_avg_V": 276.9927777608227,
+        "V_LN_p5_V": 275.05,
+        "V_LN_p95_V": 279.05,
+        "V_THD_p95_pct": 4.5062500000000005,
+        "n_outages": 0,
+        "n_dips": 0,
+        "n_swells": 0,
+        "outage_minutes": 0.0
+      }
+    ],
+    "occurrences_utc": [
+      {
+        "label": "night 2026-05-27",
+        "name": "night",
+        "lo": 0,
+        "hi": 360
+      },
+      {
+        "label": "day 2026-05-27",
+        "name": "day",
+        "lo": 360,
+        "hi": 1080
+      },
+      {
+        "label": "night 2026-05-27",
+        "name": "night",
+        "lo": 1080,
+        "hi": 1800
+      },
+      {
+        "label": "day 2026-05-28",
+        "name": "day",
+        "lo": 1800,
+        "hi": 2520
+      },
+      {
+        "label": "night 2026-05-28",
+        "name": "night",
+        "lo": 2520,
+        "hi": 3240
+      },
+      {
+        "label": "day 2026-05-29",
+        "name": "day",
+        "lo": 3240,
+        "hi": 3960
+      },
+      {
+        "label": "night 2026-05-29",
+        "name": "night",
+        "lo": 3960,
+        "hi": 4320
+      }
+    ],
+    "comparison_abc_chicago": [
+      {
+        "shift": "A",
+        "window": "06:00-14:00",
+        "records": 1440,
+        "hours": 0.4,
+        "kWh": 14.70750000000002,
+        "P_total_avg_W": 36768.750000000044,
+        "P_total_min_W": 30000.0,
+        "P_total_max_W": 39800.0,
+        "peak_demand_kW": 37.11338888888889,
+        "peak_demand_window_secs": 900,
+        "PF_avg": 0.939999997615815,
+        "V_LN_avg_V": 276.99166666666616,
+        "V_LN_p5_V": 275.05,
+        "V_LN_p95_V": 279.05,
+        "V_THD_p95_pct": 4.5062500000000005,
+        "n_outages": 0,
+        "n_dips": 0,
+        "n_swells": 0,
+        "outage_minutes": 0.0
+      },
+      {
+        "shift": "B",
+        "window": "14:00-22:00",
+        "records": 1440,
+        "hours": 0.4,
+        "kWh": 12.94979166666667,
+        "P_total_avg_W": 32374.479166666675,
+        "P_total_min_W": 30000.0,
+        "P_total_max_W": 34800.0,
+        "peak_demand_kW": 32.44583333333333,
+        "peak_demand_window_secs": 900,
+        "PF_avg": 0.939999997615815,
+        "V_LN_avg_V": 276.9991666369966,
+        "V_LN_p5_V": 275.05,
+        "V_LN_p95_V": 279.05,
+        "V_THD_p95_pct": 4.5062500000000005,
+        "n_outages": 0,
+        "n_dips": 0,
+        "n_swells": 0,
+        "outage_minutes": 0.0
+      },
+      {
+        "shift": "C",
+        "window": "22:00-06:00",
+        "records": 1440,
+        "hours": 0.4,
+        "kWh": 14.206458333333323,
+        "P_total_avg_W": 35516.14583333331,
+        "P_total_min_W": 30000.0,
+        "P_total_max_W": 39800.0,
+        "peak_demand_kW": 35.76972222222222,
+        "peak_demand_window_secs": 900,
+        "PF_avg": 0.939999997615815,
+        "V_LN_avg_V": 277.0058333502876,
+        "V_LN_p5_V": 275.05,
+        "V_LN_p95_V": 279.05,
+        "V_THD_p95_pct": 4.5062500000000005,
+        "n_outages": 0,
+        "n_dips": 0,
+        "n_swells": 0,
+        "outage_minutes": 0.0
+      }
+    ]
+  },
   "timezone": {
     "epoch_ms": 1705158000000,
     "utc": "2024-01-13T15:00:00+00:00",
diff --git a/python/tests/test_analysis_parity_golden.py b/python/tests/test_analysis_parity_golden.py
index 9fb1419..0eca3b4 100644
--- a/python/tests/test_analysis_parity_golden.py
+++ b/python/tests/test_analysis_parity_golden.py
@@ -16,12 +16,14 @@
 from pathlib import Path
 
 from fluke_3540.analysis import (
-    classify_itic, demand_analysis, detect_ct_reversal, event_itic,
-    ieee519_compliance, sarfi_indices, time_of_day_profile, whole_session_stats,
+    ShiftSet, classify_itic, demand_analysis, detect_ct_reversal, event_itic,
+    ieee519_compliance, sarfi_indices, shift_comparison_rows, shift_occurrences,
+    time_of_day_profile, whole_session_stats,
 )
 from fluke_3540.events import Event
 from fluke_3540.insights import Finding
 from fluke_3540.narrative import build_narrative
+from fluke_3540.parser import Record
 from fluke_3540.store import ColumnStore
 
 from conftest import make_records, plant_window
@@ -121,6 +123,48 @@ def test_emit_analysis_golden():
     ramp_store = ColumnStore.from_records(make_records(600, overrides=ramp_over))
     demand = demand_analysis(ramp_store, window_secs=120, series_step_secs=120)
 
+    # --- Shifts golden (Feature: generalized shift splitting) -------------
+    # A deterministic 3-day, MINUTE-resolution session crossing day/night
+    # boundaries. The JS parity test recreates this record set exactly
+    # (startMs = SHIFT_BASE + n*60000) and compares aggregate/occurrence/
+    # comparison outputs. Records are minute-spaced because per-second records
+    # all collapse to one minute-of-day, which the shift logic keys on.
+    shift_base = dt.datetime(2026, 5, 27, 0, 0, 0, tzinfo=dt.timezone.utc)
+    n_min = 3 * 1440  # 3 full days of one-minute records
+    shift_recs: list[Record] = []
+    for n in range(n_min):
+        start = shift_base + dt.timedelta(minutes=n)
+        # Deterministic, day/night-distinguishable power + voltage shapes.
+        mod = (n % 1440)
+        p_w = 30_000.0 + 5_000.0 * (1 if (360 <= mod < 1080) else 0) + (n % 97) * 50.0
+        v = 277.0 + ((n % 11) - 5) * 0.4
+        floats = [0.0] * len(make_records(1)[0].floats)
+        from conftest import FIELD_INDEX
+        for ph in ("a", "b", "c"):
+            floats[FIELD_INDEX[f"V_LN_{ph}_avg_V"]] = v
+            floats[FIELD_INDEX[f"I_{ph}_avg_A"]] = 100.0 + (n % 13)
+        floats[FIELD_INDEX["P_total_avg_W"]] = p_w
+        floats[FIELD_INDEX["PF_total_avg"]] = 0.92 + (n % 5) * 0.01
+        floats[FIELD_INDEX["V_THD_pct_a_avg"]] = 3.0 + (n % 7) * 0.3
+        floats[FIELD_INDEX["V_THD_pct_b_avg"]] = 2.5 + (n % 5) * 0.2
+        floats[FIELD_INDEX["V_THD_pct_c_avg"]] = 4.0 + (n % 3) * 0.25
+        shift_recs.append(Record(index=n, start=start,
+                                 end=start + dt.timedelta(minutes=1),
+                                 floats=tuple(floats)))
+    shift_store = ColumnStore.from_records(shift_recs)
+    shift_ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    shift_events_utc = shift_comparison_rows(
+        shift_store, shift_ss, events=[], tz=None,
+        nominal_ln_v=277.0, demand_window=15)
+    shift_occ_utc = shift_occurrences(shift_store, shift_ss, tz=None)
+    # 3-shift A/B/C in America/Chicago to exercise the tz-localized path.
+    shift_abc = ShiftSet.parse("A=06:00-14:00,B=14:00-22:00,C=22:00-06:00")
+    from fluke_3540.tzutil import resolve_tz
+    chi = resolve_tz("America/Chicago")
+    shift_abc_chi = shift_comparison_rows(
+        shift_store, shift_abc, events=[], tz=chi,
+        nominal_ln_v=277.0, demand_window=15)
+
     golden = {
         "stats": stats,
         "tod_rows": tod,
@@ -132,6 +176,16 @@ def test_emit_analysis_golden():
         "ieee519": ieee519,
         "sarfi": sarfi,
         "demand": demand,
+        "shifts": {
+            "base_epoch_ms": int(shift_base.timestamp() * 1000),
+            "n_records": n_min,
+            "comparison_utc": shift_events_utc,
+            "occurrences_utc": [
+                {"label": l, "name": nm, "lo": lo, "hi": hi}
+                for (l, nm, lo, hi) in shift_occ_utc
+            ],
+            "comparison_abc_chicago": shift_abc_chi,
+        },
     }
 
     # Timezone formatting golden (Feature H): a fixed UTC instant rendered in
@@ -155,3 +209,10 @@ def test_emit_analysis_golden():
     assert len(tod) > 0
     assert ct["reversed"] is True
     assert abs(ct["frac_negative"] - 0.70) < 1e-9
+    # Shifts: day window 06:00-18:00 = 720 min/day × 3 days = 2160 records.
+    by = {r["shift"]: r for r in shift_events_utc}
+    assert by["day"]["records"] == 720 * 3
+    assert by["night"]["records"] == 720 * 3
+    # 3 days × 2 shifts, but the first record (00:00) is night and the run
+    # continues; occurrences = day/night alternation. Expect 6-7 occurrences.
+    assert len(shift_occ_utc) >= 6
diff --git a/python/tests/test_cli_features.py b/python/tests/test_cli_features.py
index f21228b..9f4b186 100644
--- a/python/tests/test_cli_features.py
+++ b/python/tests/test_cli_features.py
@@ -109,6 +109,90 @@ def test_split_by_day(tmp_path: Path):
     assert rows[0]["bucket"] == "2024-01-13"
 
 
+def test_split_by_shifts_default(tmp_path: Path):
+    # base 22:00 UTC → all records in the default night shift.
+    base = dt.datetime(2024, 1, 13, 22, 0, 0, tzinfo=dt.timezone.utc)
+    d = _session(tmp_path, 300, base=base)
+    out = tmp_path / "out"
+    rc = main([str(d), "-o", str(out), "--parse-only", "--split-by", "shifts",
+               "--no-xlsx"])
+    assert rc == 0
+    assert (out / "shift_comparison.csv").is_file()
+    assert (out / "shift_comparison.json").is_file()
+    payload = json.loads((out / "shift_comparison.json").read_text())
+    names = {r["shift"] for r in payload["shifts"]}
+    assert {"day", "night"} <= names
+    by = {r["shift"]: r for r in payload["shifts"]}
+    assert by["night"]["records"] == 300
+    assert by["day"]["records"] == 0
+    assert payload["tz"] == "UTC"
+    # per-occurrence buckets exist
+    assert (out / "shifts").is_dir()
+
+
+def test_split_by_shifts_crosses_boundary(tmp_path: Path):
+    # base 17:55:00 UTC, 700 s (~11.6 min) → crosses 18:00 day→night.
+    base = dt.datetime(2024, 1, 13, 17, 55, 0, tzinfo=dt.timezone.utc)
+    d = _session(tmp_path, 700, base=base)
+    out = tmp_path / "out"
+    rc = main([str(d), "-o", str(out), "--parse-only", "--split-by", "shifts",
+               "--shifts", "day=06:00-18:00,night=18:00-06:00", "--no-xlsx"])
+    assert rc == 0
+    rows = list(csv.DictReader((out / "shift_comparison.csv").open()))
+    by = {r["shift"]: r for r in rows}
+    # 17:55:00..17:59:59 = 300 records day; 18:00:00.. = 400 records night.
+    assert int(by["day"]["records"]) == 300
+    assert int(by["night"]["records"]) == 400
+    assert by["day"]["window"] == "06:00-18:00"
+
+
+def test_split_by_shifts_tz_central(tmp_path: Path):
+    # base 10:25 UTC = 05:25 America/Chicago (winter, UTC-6) → night (wrap leg)
+    # in Central but day in UTC. With --tz the records must land in night.
+    base = dt.datetime(2024, 1, 13, 10, 25, 0, tzinfo=dt.timezone.utc)
+    d = _session(tmp_path, 120, base=base)
+    out = tmp_path / "out"
+    rc = main([str(d), "-o", str(out), "--parse-only", "--split-by", "shifts",
+               "--tz", "America/Chicago", "--no-xlsx"])
+    assert rc == 0
+    payload = json.loads((out / "shift_comparison.json").read_text())
+    by = {r["shift"]: r for r in payload["shifts"]}
+    assert payload["tz"] == "America/Chicago"
+    assert by["night"]["records"] == 120   # Central 05:25 → night wrap leg
+    assert by["day"]["records"] == 0
+
+
+def test_split_by_shifts_file(tmp_path: Path):
+    base = dt.datetime(2024, 1, 13, 22, 0, 0, tzinfo=dt.timezone.utc)
+    d = _session(tmp_path, 100, base=base)
+    out = tmp_path / "out"
+    sf = tmp_path / "shifts.json"
+    sf.write_text(json.dumps({"shifts": [
+        {"name": "A", "start": "06:00", "end": "14:00"},
+        {"name": "B", "start": "14:00", "end": "22:00"},
+        {"name": "C", "start": "22:00", "end": "06:00"},
+    ]}), encoding="utf-8")
+    rc = main([str(d), "-o", str(out), "--parse-only", "--split-by", "shifts",
+               "--shifts-file", str(sf), "--no-xlsx"])
+    assert rc == 0
+    payload = json.loads((out / "shift_comparison.json").read_text())
+    by = {r["shift"]: r for r in payload["shifts"]}
+    assert set(by) >= {"A", "B", "C"}
+    assert by["C"]["records"] == 100   # 22:00 UTC → shift C
+
+
+def test_split_by_shifts_summary_table(tmp_path: Path):
+    base = dt.datetime(2024, 1, 13, 22, 0, 0, tzinfo=dt.timezone.utc)
+    d = _session(tmp_path, 120, base=base)
+    out = tmp_path / "out"
+    rc = main([str(d), "-o", str(out), "--parse-only", "--split-by", "shifts",
+               "--no-xlsx"])
+    assert rc == 0
+    summary = (out / "summary.txt").read_text(encoding="utf-8")
+    assert "Shift comparison" in summary
+    assert "night" in summary
+
+
 def test_split_by_union_of_events_equals_whole(tmp_path: Path):
     # Whole-session events should equal the union of per-bucket events.
     from conftest import make_records, plant_window
diff --git a/python/tests/test_shifts.py b/python/tests/test_shifts.py
new file mode 100644
index 0000000..85b6e89
--- /dev/null
+++ b/python/tests/test_shifts.py
@@ -0,0 +1,344 @@
+"""Tests for generalized, named, configurable shift/period splitting.
+
+The shift model lets users define multiple named time windows within a day
+(which may wrap past midnight) and:
+  * AGGREGATE all records of each named shift across the whole session into a
+    single comparison row (day vs night, A/B/C shifts, …);
+  * bucket each individual shift OCCURRENCE as its own contiguous time-ordered
+    range (a "night" spanning midnight is ONE occurrence labeled by its start
+    date).
+
+CRITICAL tz contract: the store holds UTC timestamps; shift windows are
+evaluated in the *report* timezone (``--tz``). The localize step happens before
+the HH:MM window rule is applied.
+"""
+from __future__ import annotations
+
+import datetime as dt
+
+import pytest
+
+try:  # Python 3.9+
+    from zoneinfo import ZoneInfo
+except ImportError:  # pragma: no cover
+    ZoneInfo = None
+
+from fluke_3540.analysis import (
+    Shift,
+    ShiftSet,
+    aggregate_shifts,
+    gather_store,
+    shift_comparison_rows,
+    shift_occurrences,
+)
+from fluke_3540.parser import Record
+from fluke_3540.store import ColumnStore
+
+from conftest import make_records
+
+
+def make_minute_records(count, base, overrides=None, defaults=None):
+    """make_records spaced one MINUTE apart (the per-second helper packs all
+    records into the same minute-of-day, which the shift logic keys on)."""
+    recs = make_records(count, base=base, overrides=overrides, defaults=defaults)
+    out = []
+    for n, r in enumerate(recs):
+        start = base + dt.timedelta(minutes=n)
+        out.append(Record(index=n, start=start,
+                          end=start + dt.timedelta(minutes=1), floats=r.floats))
+    return out
+
+
+# --- ShiftSet.parse ----------------------------------------------------------
+
+def test_parse_two_shifts():
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    assert [s.name for s in ss.shifts] == ["day", "night"]
+    day, night = ss.shifts
+    assert (day.start_min, day.end_min) == (360, 1080)
+    assert day.wraps is False
+    assert (night.start_min, night.end_min) == (1080, 360)
+    assert night.wraps is True
+
+
+def test_parse_three_shifts():
+    ss = ShiftSet.parse("A=06:00-14:00,B=14:00-22:00,C=22:00-06:00")
+    assert [s.name for s in ss.shifts] == ["A", "B", "C"]
+    assert ss.shifts[2].wraps is True  # C crosses midnight
+
+
+def test_parse_default():
+    ss = ShiftSet.default()
+    assert [s.name for s in ss.shifts] == ["day", "night"]
+    assert ss.shifts[0].start_min == 360
+
+
+def test_parse_bad_time():
+    with pytest.raises(ValueError):
+        ShiftSet.parse("day=06:00-25:00")
+    with pytest.raises(ValueError):
+        ShiftSet.parse("day=0600-1800")  # missing colon → not HH:MM
+    with pytest.raises(ValueError):
+        ShiftSet.parse("garbage")
+
+
+def test_parse_duplicate_names():
+    with pytest.raises(ValueError):
+        ShiftSet.parse("day=06:00-12:00,day=12:00-18:00")
+
+
+def test_parse_zero_length_window_rejected():
+    with pytest.raises(ValueError):
+        ShiftSet.parse("x=06:00-06:00")
+
+
+def test_load_shifts_file(tmp_path):
+    from fluke_3540.shifts_file import load_shifts
+    p = tmp_path / "s.json"
+    p.write_text('{"shifts":[{"name":"day","start":"06:00","end":"18:00"},'
+                 '{"name":"night","start":"18:00","end":"06:00"}]}',
+                 encoding="utf-8")
+    ss = load_shifts(p)
+    assert [s.name for s in ss.shifts] == ["day", "night"]
+
+
+def test_load_shifts_file_bare_list(tmp_path):
+    from fluke_3540.shifts_file import load_shifts
+    p = tmp_path / "s.json"
+    p.write_text('[{"name":"A","start":"06:00","end":"18:00"}]', encoding="utf-8")
+    ss = load_shifts(p)
+    assert [s.name for s in ss.shifts] == ["A"]
+
+
+def test_load_shifts_file_bad(tmp_path):
+    from fluke_3540.shifts_file import load_shifts
+    p = tmp_path / "s.json"
+    p.write_text('{"nope": 1}', encoding="utf-8")
+    with pytest.raises(ValueError):
+        load_shifts(p)
+
+
+def test_from_spec_dicts():
+    ss = ShiftSet.from_spec([
+        {"name": "day", "start": "06:00", "end": "18:00"},
+        {"name": "night", "start": "18:00", "end": "06:00"},
+    ])
+    assert [s.name for s in ss.shifts] == ["day", "night"]
+    assert ss.shifts[1].wraps is True
+
+
+# --- Shift.contains_minute (wrap logic) --------------------------------------
+
+def test_contains_minute_non_wrapping():
+    day = Shift("day", 360, 1080)
+    assert day.contains_minute(360) is True   # 06:00 inclusive
+    assert day.contains_minute(1079) is True  # 17:59
+    assert day.contains_minute(1080) is False  # 18:00 exclusive
+    assert day.contains_minute(300) is False
+
+
+def test_contains_minute_wrapping():
+    night = Shift("night", 1080, 360)
+    assert night.contains_minute(1080) is True   # 18:00
+    assert night.contains_minute(1439) is True   # 23:59
+    assert night.contains_minute(0) is True      # 00:00
+    assert night.contains_minute(359) is True    # 05:59
+    assert night.contains_minute(360) is False   # 06:00 → day
+    assert night.contains_minute(720) is False   # noon
+
+
+# --- tz-localized assignment -------------------------------------------------
+
+@pytest.mark.skipif(ZoneInfo is None, reason="zoneinfo unavailable")
+def test_assignment_localizes_to_report_tz():
+    # A record stored at 10:30 UTC is 05:30 America/Chicago (CDT, UTC-5).
+    # In Central that lands in the night shift (the 00:00-06:00 wrap leg); in
+    # UTC (10:30, mod 630) it would land in the day shift. This pins the tz
+    # contract: windows are evaluated in the report tz, not raw UTC.
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    central = ZoneInfo("America/Chicago")
+    base = dt.datetime(2026, 5, 29, 10, 30, 0, tzinfo=dt.timezone.utc)
+    recs = make_records(1, base=base)
+    store = ColumnStore.from_records(recs)
+
+    by_name_central = aggregate_shifts(store, ss, tz=central)
+    assert by_name_central["night"] == [0]
+    assert by_name_central.get("day", []) == []
+
+    by_name_utc = aggregate_shifts(store, ss, tz=None)
+    assert by_name_utc["day"] == [0]
+    assert by_name_utc.get("night", []) == []
+
+
+def test_assignment_utc_default():
+    # 08:00 UTC, no tz → day shift.
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    base = dt.datetime(2026, 5, 29, 8, 0, 0, tzinfo=dt.timezone.utc)
+    recs = make_records(1, base=base)
+    store = ColumnStore.from_records(recs)
+    by_name = aggregate_shifts(store, ss, tz=None)
+    assert by_name["day"] == [0]
+
+
+# --- aggregate_shifts grouping (non-contiguous) ------------------------------
+
+def test_aggregate_groups_noncontiguous_indices():
+    # 06:00 + a few hours of records straddling into night, all 1/sec. Build a
+    # compact multi-window day: 11 records starting 17:58:00 UTC → 17:58..18:08.
+    # Records before 18:00 → day; 18:00 onward → night. So indices are split
+    # into two contiguous chunks here, but the API returns lists by name.
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    base = dt.datetime(2026, 5, 29, 17, 58, 0, tzinfo=dt.timezone.utc)
+    recs = make_minute_records(11, base=base)  # 17:58 .. 18:08 starts
+    store = ColumnStore.from_records(recs)
+    by_name = aggregate_shifts(store, ss, tz=None)
+    # 17:58:00, 17:59:00 (mins 1078,1079) are day; 18:00:00.. are night.
+    assert by_name["day"] == [0, 1]
+    assert by_name["night"] == list(range(2, 11))
+
+
+def test_gather_store_picks_indices():
+    recs = make_records(50, defaults={"P_total_avg_W": 1000.0})
+    store = ColumnStore.from_records(recs, time_shift=dt.timedelta(hours=2))
+    sub = gather_store(store, [3, 7, 40])
+    assert sub.n == 3
+    assert sub.time_shift == dt.timedelta(hours=2)
+    assert sub.start(0) == store.start(3)
+    assert sub.start(1) == store.start(7)
+    assert sub.col("P_total_avg_W")[2] == store.col("P_total_avg_W")[40]
+
+
+# --- unassigned + overlap/gap validation -------------------------------------
+
+def test_unassigned_when_window_does_not_tile():
+    # Only a morning shift; afternoon records have no home → "unassigned".
+    ss = ShiftSet.parse("morning=06:00-12:00")
+    base = dt.datetime(2026, 5, 29, 11, 58, 0, tzinfo=dt.timezone.utc)
+    recs = make_minute_records(5, base=base)  # 11:58..12:02
+    store = ColumnStore.from_records(recs)
+    by_name = aggregate_shifts(store, ss, tz=None)
+    assert by_name["morning"] == [0, 1]            # 11:58, 11:59
+    assert by_name["unassigned"] == [2, 3, 4]      # 12:00, 12:01, 12:02
+
+
+def test_first_matching_window_wins_on_overlap():
+    # Overlapping windows: record at 10:00 matches both; the first listed wins.
+    ss = ShiftSet.parse("early=06:00-12:00,late=10:00-18:00")
+    base = dt.datetime(2026, 5, 29, 10, 0, 0, tzinfo=dt.timezone.utc)
+    recs = make_records(1, base=base)
+    store = ColumnStore.from_records(recs)
+    by_name = aggregate_shifts(store, ss, tz=None)
+    assert by_name["early"] == [0]
+    assert by_name.get("late", []) == []
+
+
+def test_coverage_gaps_and_overlaps_reported():
+    full = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    assert full.coverage_issues() == []  # tiles 24 h cleanly
+
+    gapped = ShiftSet.parse("morning=06:00-12:00,evening=14:00-20:00")
+    issues = gapped.coverage_issues()
+    assert any("gap" in s.lower() for s in issues)
+
+    overlapped = ShiftSet.parse("early=06:00-13:00,late=12:00-18:00")
+    issues = overlapped.coverage_issues()
+    assert any("overlap" in s.lower() for s in issues)
+
+
+# --- occurrences (contiguous per-instance buckets) ---------------------------
+
+def test_shift_occurrences_labels_by_start_date():
+    # Night window 18:00-06:00; records 17:58:00..18:03:00 UTC → the records at
+    # 18:00+ are one "night" occurrence labeled by its start date.
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    base = dt.datetime(2026, 5, 29, 17, 58, 0, tzinfo=dt.timezone.utc)
+    recs = make_minute_records(6, base=base)  # 17:58..18:03 starts
+    store = ColumnStore.from_records(recs)
+    occ = shift_occurrences(store, ss, tz=None)
+    # occurrences are (label, name, lo, hi) contiguous ranges, time-ordered.
+    labels = [o[0] for o in occ]
+    names = [o[1] for o in occ]
+    assert names == ["day", "night"]
+    assert labels[1] == "night 2026-05-29"
+    # contiguous & covering
+    assert occ[0][2] == 0 and occ[-1][3] == store.n
+    for a, b in zip(occ, occ[1:]):
+        assert a[3] == b[2]
+
+
+def test_shift_occurrences_midnight_span_is_one_bucket():
+    # Build records crossing local midnight inside the night shift. Use UTC tz
+    # so wall = stored. 23:58:00 .. 00:02:00 (next day) → all night, ONE
+    # occurrence labeled by the START date (the 29th).
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    base = dt.datetime(2026, 5, 29, 23, 58, 0, tzinfo=dt.timezone.utc)
+    recs = make_minute_records(5, base=base)  # 23:58,23:59,00:00,00:01,00:02
+    store = ColumnStore.from_records(recs)
+    occ = shift_occurrences(store, ss, tz=None)
+    night_occ = [o for o in occ if o[1] == "night"]
+    assert len(night_occ) == 1
+    assert night_occ[0][0] == "night 2026-05-29"
+    assert night_occ[0][2] == 0 and night_occ[0][3] == 5
+
+
+# --- comparison rows (the headline output) -----------------------------------
+
+def test_shift_comparison_rows_schema_and_values():
+    # Two shifts, distinct power levels so the comparison is meaningful.
+    # Build 4 minutes: 2 min in day (05:59 → no, use 06:00..) and 2 in night.
+    # Easier: 17:59:00 (day) ×60 then 18:00:00 (night) ×60, each 1/sec.
+    overrides = {}
+    for i in range(60):
+        overrides[i] = {"P_total_avg_W": 10_000.0}        # day records
+    for i in range(60, 120):
+        overrides[i] = {"P_total_avg_W": 20_000.0}        # night records
+    base = dt.datetime(2026, 5, 29, 17, 59, 0, tzinfo=dt.timezone.utc)
+    recs = make_records(120, base=base, overrides=overrides)
+    store = ColumnStore.from_records(recs)
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    rows = shift_comparison_rows(store, ss, events=[], tz=None,
+                                 nominal_ln_v=277.0, demand_window=15)
+    by = {r["shift"]: r for r in rows}
+    assert set(by) == {"day", "night"}
+    d, ngt = by["day"], by["night"]
+    assert d["records"] == 60
+    assert ngt["records"] == 60
+    assert d["window"] == "06:00-18:00"
+    assert ngt["window"] == "18:00-06:00"
+    assert d["P_total_avg_W"] == pytest.approx(10_000.0)
+    assert ngt["P_total_avg_W"] == pytest.approx(20_000.0)
+    assert ngt["P_total_avg_W"] > d["P_total_avg_W"]
+    # energy = mean power * hours; 60 records = 60 s = 1/60 h.
+    assert d["kWh"] == pytest.approx(10.0 / 1000.0 * (60 / 3600.0) * 1000)  # 10 kW * (60s/3600) h
+    # required schema keys present
+    for k in ("shift", "window", "records", "hours", "kWh", "P_total_avg_W",
+              "P_total_min_W", "P_total_max_W", "peak_demand_kW", "PF_avg",
+              "V_LN_avg_V", "V_LN_p5_V", "V_LN_p95_V", "V_THD_p95_pct",
+              "n_outages", "n_dips", "n_swells", "outage_minutes"):
+        assert k in d, f"missing schema key {k}"
+
+
+def test_shift_comparison_multi_day_aggregates_across_occurrences():
+    # Two day-shift windows on two different dates aggregate into ONE day row.
+    # Day A: 2026-05-29 08:00 ×30 @ 5kW ; Day B: 2026-05-30 08:00 ×30 @ 15kW.
+    # Using a 1-day gap of night records between them so they're non-contiguous.
+    recs = []
+    recs += make_records(30, base=dt.datetime(2026, 5, 29, 8, 0, tzinfo=dt.timezone.utc),
+                         defaults={"P_total_avg_W": 5_000.0})
+    recs += make_records(30, base=dt.datetime(2026, 5, 29, 20, 0, tzinfo=dt.timezone.utc),
+                         defaults={"P_total_avg_W": 1_000.0})  # night
+    recs += make_records(30, base=dt.datetime(2026, 5, 30, 8, 0, tzinfo=dt.timezone.utc),
+                         defaults={"P_total_avg_W": 15_000.0})
+    # Re-index the records so indices are contiguous 0..89 with correct times.
+    from fluke_3540.parser import Record
+    fixed = [Record(index=i, start=r.start, end=r.end, floats=r.floats)
+             for i, r in enumerate(recs)]
+    store = ColumnStore.from_records(fixed)
+    ss = ShiftSet.parse("day=06:00-18:00,night=18:00-06:00")
+    rows = shift_comparison_rows(store, ss, events=[], tz=None,
+                                 nominal_ln_v=277.0, demand_window=15)
+    by = {r["shift"]: r for r in rows}
+    assert by["day"]["records"] == 60       # both day windows merged
+    assert by["night"]["records"] == 30
+    # day avg power = mean of 30×5k + 30×15k = 10k
+    assert by["day"]["P_total_avg_W"] == pytest.approx(10_000.0)
diff --git a/web/analysis.js b/web/analysis.js
index 11e75e7..2d43506 100644
--- a/web/analysis.js
+++ b/web/analysis.js
@@ -502,3 +502,308 @@ export function correlateMarkers(markers, events) {
   }
   return out;
 }
+
+// --- Generalized named shift/period splitting (--split-by shifts) ----------
+// JS port of python/.../analysis.py Shift / ShiftSet + aggregate / occurrences
+// / comparison. The store holds UTC epoch ms; shift windows are evaluated in
+// the REPORT timezone (tzName, IANA) — localize minute-of-day BEFORE applying
+// the HH:MM rule, exactly like the Python side. tzName null/'UTC' = UTC.
+
+export const UNASSIGNED_SHIFT = 'unassigned';
+
+function parseHhmm(text) {
+  const s = String(text).trim();
+  if (!s.includes(':')) throw new Error(`shift time must be HH:MM (with a colon): ${text}`);
+  const [hh, mm] = s.split(':');
+  if (!/^\d+$/.test(hh) || !/^\d+$/.test(mm)) {
+    throw new Error(`shift time must be numeric HH:MM: ${text}`);
+  }
+  const h = parseInt(hh, 10);
+  const m = parseInt(mm, 10);
+  if (m >= 60) throw new Error(`shift minutes out of range in ${text}`);
+  const total = h * 60 + m;
+  if (total < 0 || total > 1440) throw new Error(`shift time out of range (00:00..24:00): ${text}`);
+  return total;
+}
+
+export class Shift {
+  constructor(name, startMin, endMin) {
+    this.name = name;
+    this.startMin = startMin;
+    this.endMin = endMin;
+  }
+
+  get wraps() { return this.endMin <= this.startMin; }
+
+  containsMinute(mod) {
+    if (!this.wraps) return this.startMin <= mod && mod < this.endMin;
+    return mod >= this.startMin || mod < this.endMin;
+  }
+
+  lengthMinutes() {
+    return this.wraps ? (1440 - this.startMin) + this.endMin : this.endMin - this.startMin;
+  }
+
+  get windowStr() {
+    const fmt = (m) => (m === 1440
+      ? '24:00'
+      : `${String(Math.floor(m / 60) % 24).padStart(2, '0')}:${String(m % 60).padStart(2, '0')}`);
+    return `${fmt(this.startMin)}-${fmt(this.endMin)}`;
+  }
+}
+
+export class ShiftSet {
+  constructor(shifts) { this.shifts = shifts; }
+
+  static parse(text) {
+    const out = [];
+    const seen = new Set();
+    const parts = String(text).split(',').map((p) => p.trim()).filter(Boolean);
+    if (!parts.length) throw new Error(`no shifts parsed from ${text}`);
+    for (const part of parts) {
+      const eq = part.indexOf('=');
+      if (eq <= 0) throw new Error(`shift must be name=HH:MM-HH:MM: ${part}`);
+      const name = part.slice(0, eq).trim();
+      const window = part.slice(eq + 1);
+      const dash = window.indexOf('-');
+      if (dash < 0) throw new Error(`shift window must be HH:MM-HH:MM: ${window}`);
+      const start = parseHhmm(window.slice(0, dash));
+      const end = parseHhmm(window.slice(dash + 1));
+      if (start === end) throw new Error(`shift ${name} has a zero-length window ${window}`);
+      if (seen.has(name)) throw new Error(`duplicate shift name ${name}`);
+      seen.add(name);
+      out.push(new Shift(name, start, end));
+    }
+    return new ShiftSet(out);
+  }
+
+  static fromSpec(spec) {
+    return ShiftSet.parse(spec.map((s) => `${s.name}=${s.start}-${s.end}`).join(','));
+  }
+
+  static default() {
+    return ShiftSet.parse('day=06:00-18:00,night=18:00-06:00');
+  }
+
+  coverageIssues() {
+    const cover = new Array(1440).fill(0);
+    for (const sh of this.shifts) {
+      for (let m = 0; m < 1440; m++) if (sh.containsMinute(m)) cover[m] += 1;
+    }
+    const gap = cover.filter((c) => c === 0).length;
+    const overlap = cover.filter((c) => c > 1).length;
+    const issues = [];
+    if (gap) {
+      issues.push(`${gap} minute(s)/day fall in NO shift window (gap); those `
+        + `records go to '${UNASSIGNED_SHIFT}'.`);
+    }
+    if (overlap) {
+      issues.push(`${overlap} minute(s)/day are covered by MORE THAN ONE shift `
+        + '(overlap); the first matching window wins.');
+    }
+    return issues;
+  }
+}
+
+/**
+ * Minute-of-day for epoch `ms` in the report timezone (UTC if tzName falsy).
+ * Uses Intl (same approach as tzutil.isoInZone) so it matches Python's
+ * datetime.astimezone(ZoneInfo(tz)) wall clock.
+ */
+export function localMinuteOfDay(ms, tzName) {
+  if (!tzName || String(tzName).toUpperCase() === 'UTC') {
+    const d = new Date(ms);
+    return d.getUTCHours() * 60 + d.getUTCMinutes();
+  }
+  const fmt = new Intl.DateTimeFormat('en-CA', {
+    timeZone: tzName, hour: '2-digit', minute: '2-digit', hour12: false,
+  });
+  const parts = {};
+  for (const p of fmt.formatToParts(new Date(ms))) parts[p.type] = p.value;
+  const hour = parts.hour === '24' ? 0 : parseInt(parts.hour, 10);
+  return hour * 60 + parseInt(parts.minute, 10);
+}
+
+function localDateStr(ms, tzName) {
+  if (!tzName || String(tzName).toUpperCase() === 'UTC') {
+    return new Date(ms).toISOString().slice(0, 10);
+  }
+  const fmt = new Intl.DateTimeFormat('en-CA', {
+    timeZone: tzName, year: 'numeric', month: '2-digit', day: '2-digit',
+  });
+  const parts = {};
+  for (const p of fmt.formatToParts(new Date(ms))) parts[p.type] = p.value;
+  return `${parts.year}-${parts.month}-${parts.day}`;
+}
+
+/**
+ * Group record indices by shift NAME, evaluated in tzName. Returns a plain
+ * object {name: [indices]}; every named shift key is present (possibly empty);
+ * unmatched records collect under UNASSIGNED_SHIFT (only when non-empty).
+ */
+export function aggregateShifts(source, spec, ss, opts = {}) {
+  const tzName = opts.tz ?? null;
+  const src = asColumnSource(source, spec);
+  const out = {};
+  for (const sh of ss.shifts) out[sh.name] = [];
+  for (let i = 0; i < src.length; i++) {
+    const mod = localMinuteOfDay(src.startMs(i), tzName);
+    let placed = false;
+    for (const sh of ss.shifts) {
+      if (sh.containsMinute(mod)) { out[sh.name].push(i); placed = true; break; }
+    }
+    if (!placed) {
+      (out[UNASSIGNED_SHIFT] ??= []).push(i);
+    }
+  }
+  return out;
+}
+
+/**
+ * Contiguous per-occurrence buckets: [{label, name, lo, hi}], time-ordered. A
+ * wrap-past-midnight occurrence stays one bucket, labeled by its start date.
+ */
+export function shiftOccurrences(source, spec, ss, opts = {}) {
+  const tzName = opts.tz ?? null;
+  const src = asColumnSource(source, spec);
+  const n = src.length;
+  if (n === 0) return [];
+  const nameAt = (i) => {
+    const mod = localMinuteOfDay(src.startMs(i), tzName);
+    for (const sh of ss.shifts) if (sh.containsMinute(mod)) return sh.name;
+    return UNASSIGNED_SHIFT;
+  };
+  const out = [];
+  let curName = nameAt(0);
+  let lo = 0;
+  for (let i = 1; i < n; i++) {
+    const nm = nameAt(i);
+    if (nm !== curName) {
+      out.push({ label: `${curName} ${localDateStr(src.startMs(lo), tzName)}`, name: curName, lo, hi: i });
+      curName = nm; lo = i;
+    }
+  }
+  out.push({ label: `${curName} ${localDateStr(src.startMs(lo), tzName)}`, name: curName, lo, hi: n });
+  return out;
+}
+
+/**
+ * The headline per-shift-name aggregate comparison. One row per named shift
+ * (plus unassigned if any). `events` are filed by the shift their tStartMs
+ * falls in. demandWindow is in MINUTES.
+ * @returns {Array<object>} rows mirroring python shift_comparison_rows
+ */
+export function shiftComparisonRows(source, spec, ss, opts = {}) {
+  const tzName = opts.tz ?? null;
+  const events = opts.events ?? [];
+  const demandMin = Math.max(1, Math.floor(opts.demandWindow ?? 15));
+  const src = asColumnSource(source, spec);
+  const byName = aggregateShifts(source, spec, ss, { tz: tzName });
+  const winByName = new Map(ss.shifts.map((sh) => [sh.name, sh]));
+
+  const evByName = {};
+  for (const nm of Object.keys(byName)) evByName[nm] = [];
+  for (const e of events) {
+    const mod = localMinuteOfDay(e.tStartMs, tzName);
+    let placed = false;
+    for (const sh of ss.shifts) {
+      if (sh.containsMinute(mod)) { (evByName[sh.name] ??= []).push(e); placed = true; break; }
+    }
+    if (!placed) (evByName[UNASSIGNED_SHIFT] ??= []).push(e);
+  }
+
+  const p = src.column('P_total_avg_W');
+  const pf = src.column('PF_total_avg');
+  const va = src.column('V_LN_a_avg_V');
+  const vb = src.column('V_LN_b_avg_V');
+  const vc = src.column('V_LN_c_avg_V');
+  const vthA = src.column('V_THD_pct_a_avg');
+  const vthB = src.column('V_THD_pct_b_avg');
+  const vthC = src.column('V_THD_pct_c_avg');
+
+  const rows = [];
+  for (const [name, idxs] of Object.entries(byName)) {
+    const sh = winByName.get(name);
+    const window = sh ? sh.windowStr : '—';
+    const evs = evByName[name] ?? [];
+
+    const pMom = new RunningMoments();
+    let pMin = Infinity; let pMax = -Infinity;
+    const pfMom = new RunningMoments();
+    const vMom = new RunningMoments();
+    const vSketch = new PercentileSketch(0.0, 400.0);
+    const vthdSketch = new PercentileSketch(0.0, 50.0);
+    const gatheredP = new Float64Array(idxs.length);
+    for (let k = 0; k < idxs.length; k++) {
+      const i = idxs[k];
+      const pv = p[i];
+      gatheredP[k] = Number.isFinite(pv) ? pv : 0.0;
+      if (Number.isFinite(pv)) { pMom.add(pv); pMin = Math.min(pMin, pv); pMax = Math.max(pMax, pv); }
+      const pfi = pf[i];
+      if (Number.isFinite(pfi)) pfMom.add(pfi);
+      for (const vv of [va[i], vb[i], vc[i]]) {
+        if (Number.isFinite(vv) && vv > 50.0) { vMom.add(vv); vSketch.add(vv); }
+      }
+      for (const tv of [vthA[i], vthB[i], vthC[i]]) {
+        if (Number.isFinite(tv)) vthdSketch.add(tv);
+      }
+    }
+
+    const pMean = pMom.n ? pMom.mean : 0.0;
+    const hours = pMom.n / 3600.0;
+    const kwh = (pMean / 1000.0) * hours;
+    const peakKw = peakRollingDemandKw(gatheredP, demandMin * 60);
+
+    let nOut = 0; let nDip = 0; let nSwell = 0; let outageSecs = 0.0;
+    for (const e of evs) {
+      if (e.kind === 'outage') { nOut += 1; outageSecs += (e.tEndMs - e.tStartMs) / 1000; }
+      else if (e.kind === 'dip') nDip += 1;
+      else if (e.kind === 'swell') nSwell += 1;
+    }
+    const q = (sk, qq) => { const v = sk.quantile(qq); return Number.isNaN(v) ? 0.0 : v; };
+
+    rows.push({
+      shift: name,
+      window,
+      records: idxs.length,
+      hours,
+      kWh: kwh,
+      P_total_avg_W: pMean,
+      P_total_min_W: pMin === Infinity ? 0.0 : pMin,
+      P_total_max_W: pMax === -Infinity ? 0.0 : pMax,
+      peak_demand_kW: peakKw,
+      peak_demand_window_secs: demandMin * 60,
+      PF_avg: pfMom.n ? pfMom.mean : 0.0,
+      V_LN_avg_V: vMom.n ? vMom.mean : 0.0,
+      V_LN_p5_V: q(vSketch, 0.05),
+      V_LN_p95_V: q(vSketch, 0.95),
+      V_THD_p95_pct: q(vthdSketch, 0.95),
+      n_outages: nOut,
+      n_dips: nDip,
+      n_swells: nSwell,
+      outage_minutes: outageSecs / 60.0,
+    });
+  }
+  return rows;
+}
+
+// Trailing rolling-mean peak over a 1-Hz power array, mirroring
+// demand_analysis(window_secs).peak_demand_kw on a gathered sub-store.
+function peakRollingDemandKw(pArr, windowSecs) {
+  const n = pArr.length;
+  const w = Math.max(1, Math.floor(windowSecs));
+  if (n === 0) return 0.0;
+  let running = 0.0;
+  let peak = -Infinity;
+  let any = false;
+  for (let i = 0; i < n; i++) {
+    running += pArr[i];
+    if (i >= w) running -= pArr[i - w];
+    if (i >= w - 1) {
+      const demand = running / w;
+      if (demand > peak) peak = demand;
+      any = true;
+    }
+  }
+  return any ? peak / 1000.0 : 0.0;
+}
diff --git a/web/tests/shifts_parity.test.js b/web/tests/shifts_parity.test.js
new file mode 100644
index 0000000..9568a8e
--- /dev/null
+++ b/web/tests/shifts_parity.test.js
@@ -0,0 +1,147 @@
+// Shifts parity — web/analysis.js shift functions must produce the same
+// numbers as python/.../analysis.py. We recreate the exact deterministic
+// minute-resolution multi-day session the Python golden generator builds
+// (test_analysis_parity_golden.py `shifts` block), run the JS shift analysis,
+// and compare against fixtures/analysis_golden.json["shifts"].
+import { strict as assert } from 'node:assert';
+import { readFileSync } from 'node:fs';
+import { fileURLToPath } from 'node:url';
+import { dirname, resolve } from 'node:path';
+import { test } from 'node:test';
+
+import {
+  ShiftSet, aggregateShifts, shiftOccurrences, shiftComparisonRows,
+} from '../analysis.js';
+
+const __dirname = dirname(fileURLToPath(import.meta.url));
+const repoRoot = resolve(__dirname, '..', '..');
+const spec = JSON.parse(readFileSync(resolve(repoRoot, 'spec', 'field_map.json'), 'utf8'));
+const golden = JSON.parse(readFileSync(
+  resolve(repoRoot, 'python', 'tests', 'fixtures', 'analysis_golden.json'), 'utf8'));
+
+const FI = new Map(spec.fields.map((f) => [f.name, f.index]));
+const G = golden.shifts;
+
+// Mirror the Python `shift_recs` builder exactly.
+function buildShiftSession() {
+  const baseMs = G.base_epoch_ms;
+  const nMin = G.n_records;
+  const records = [];
+  for (let n = 0; n < nMin; n++) {
+    const startMs = baseMs + n * 60000;
+    const mod = n % 1440;
+    const pW = 30000.0 + 5000.0 * ((mod >= 360 && mod < 1080) ? 1 : 0) + (n % 97) * 50.0;
+    const v = 277.0 + ((n % 11) - 5) * 0.4;
+    const floats = new Float32Array(spec.data_floats);
+    for (const ph of ['a', 'b', 'c']) {
+      floats[FI.get(`V_LN_${ph}_avg_V`)] = v;
+      floats[FI.get(`I_${ph}_avg_A`)] = 100.0 + (n % 13);
+    }
+    floats[FI.get('P_total_avg_W')] = pW;
+    floats[FI.get('PF_total_avg')] = 0.92 + (n % 5) * 0.01;
+    floats[FI.get('V_THD_pct_a_avg')] = 3.0 + (n % 7) * 0.3;
+    floats[FI.get('V_THD_pct_b_avg')] = 2.5 + (n % 5) * 0.2;
+    floats[FI.get('V_THD_pct_c_avg')] = 4.0 + (n % 3) * 0.25;
+    records.push({ index: n, startMs, endMs: startMs + 60000, floats });
+  }
+  return records;
+}
+
+const REL = 1e-4;
+const PCT_ABS = 1.0;  // percentile sketch is bin-width bounded
+
+function approx(actual, expected, absTol, msg) {
+  const tol = Math.max(absTol, Math.abs(expected) * REL);
+  assert.ok(Math.abs(actual - expected) <= tol,
+    `${msg}: got ${actual}, expected ${expected} (tol ${tol})`);
+}
+
+test('aggregateShifts: index grouping matches occurrences (UTC)', () => {
+  const recs = buildShiftSession();
+  const ss = ShiftSet.parse('day=06:00-18:00,night=18:00-06:00');
+  const by = aggregateShifts(recs, spec, ss, { tz: null });
+  // day total must equal sum of day-occurrence lengths from the golden.
+  const dayLen = G.occurrences_utc.filter((o) => o.name === 'day')
+    .reduce((s, o) => s + (o.hi - o.lo), 0);
+  const nightLen = G.occurrences_utc.filter((o) => o.name === 'night')
+    .reduce((s, o) => s + (o.hi - o.lo), 0);
+  assert.equal(by.day.length, dayLen);
+  assert.equal(by.night.length, nightLen);
+});
+
+test('shiftOccurrences: matches Python golden row-for-row (UTC)', () => {
+  const recs = buildShiftSession();
+  const ss = ShiftSet.parse('day=06:00-18:00,night=18:00-06:00');
+  const occ = shiftOccurrences(recs, spec, ss, { tz: null });
+  assert.equal(occ.length, G.occurrences_utc.length, 'occurrence count');
+  for (let i = 0; i < occ.length; i++) {
+    const js = occ[i];
+    const g = G.occurrences_utc[i];
+    assert.equal(js.label, g.label, `occ[${i}].label`);
+    assert.equal(js.name, g.name, `occ[${i}].name`);
+    assert.equal(js.lo, g.lo, `occ[${i}].lo`);
+    assert.equal(js.hi, g.hi, `occ[${i}].hi`);
+  }
+});
+
+test('shiftComparisonRows: matches Python golden (UTC, day vs night)', () => {
+  const recs = buildShiftSession();
+  const ss = ShiftSet.parse('day=06:00-18:00,night=18:00-06:00');
+  const rows = shiftComparisonRows(recs, spec, ss,
+    { tz: null, events: [], demandWindow: 15 });
+  const by = Object.fromEntries(rows.map((r) => [r.shift, r]));
+  const gby = Object.fromEntries(G.comparison_utc.map((r) => [r.shift, r]));
+  assert.deepEqual(Object.keys(by).sort(), Object.keys(gby).sort());
+  for (const name of Object.keys(gby)) {
+    const j = by[name];
+    const g = gby[name];
+    assert.equal(j.window, g.window, `${name}.window`);
+    assert.equal(j.records, g.records, `${name}.records`);
+    approx(j.hours, g.hours, 1e-6, `${name}.hours`);
+    approx(j.kWh, g.kWh, 1e-3, `${name}.kWh`);
+    approx(j.P_total_avg_W, g.P_total_avg_W, 1e-3, `${name}.P_total_avg_W`);
+    approx(j.P_total_min_W, g.P_total_min_W, 1e-3, `${name}.P_total_min_W`);
+    approx(j.P_total_max_W, g.P_total_max_W, 1e-3, `${name}.P_total_max_W`);
+    approx(j.peak_demand_kW, g.peak_demand_kW, 1e-3, `${name}.peak_demand_kW`);
+    approx(j.PF_avg, g.PF_avg, 1e-4, `${name}.PF_avg`);
+    approx(j.V_LN_avg_V, g.V_LN_avg_V, 1e-3, `${name}.V_LN_avg_V`);
+    approx(j.V_LN_p5_V, g.V_LN_p5_V, PCT_ABS, `${name}.V_LN_p5_V`);
+    approx(j.V_LN_p95_V, g.V_LN_p95_V, PCT_ABS, `${name}.V_LN_p95_V`);
+    approx(j.V_THD_p95_pct, g.V_THD_p95_pct, PCT_ABS, `${name}.V_THD_p95_pct`);
+    assert.equal(j.n_outages, g.n_outages, `${name}.n_outages`);
+    assert.equal(j.n_dips, g.n_dips, `${name}.n_dips`);
+    assert.equal(j.n_swells, g.n_swells, `${name}.n_swells`);
+    approx(j.outage_minutes, g.outage_minutes, 1e-6, `${name}.outage_minutes`);
+  }
+});
+
+test('shiftComparisonRows: tz-localized A/B/C matches Python golden (America/Chicago)', () => {
+  const recs = buildShiftSession();
+  const ss = ShiftSet.parse('A=06:00-14:00,B=14:00-22:00,C=22:00-06:00');
+  const rows = shiftComparisonRows(recs, spec, ss,
+    { tz: 'America/Chicago', events: [], demandWindow: 15 });
+  const by = Object.fromEntries(rows.map((r) => [r.shift, r]));
+  const gby = Object.fromEntries(G.comparison_abc_chicago.map((r) => [r.shift, r]));
+  assert.deepEqual(Object.keys(by).sort(), Object.keys(gby).sort());
+  for (const name of Object.keys(gby)) {
+    assert.equal(by[name].records, gby[name].records, `${name}.records (tz)`);
+    approx(by[name].P_total_avg_W, gby[name].P_total_avg_W, 1e-3, `${name}.P_avg (tz)`);
+    approx(by[name].kWh, gby[name].kWh, 1e-3, `${name}.kWh (tz)`);
+  }
+});
+
+test('ShiftSet.parse: rejects bad/duplicate/zero-length specs', () => {
+  assert.throws(() => ShiftSet.parse('day=06:00-25:00'));
+  assert.throws(() => ShiftSet.parse('day=0600-1800'));
+  assert.throws(() => ShiftSet.parse('garbage'));
+  assert.throws(() => ShiftSet.parse('day=06:00-12:00,day=12:00-18:00'));
+  assert.throws(() => ShiftSet.parse('x=06:00-06:00'));
+});
+
+test('ShiftSet.coverageIssues: clean tiling vs gap vs overlap', () => {
+  assert.deepEqual(ShiftSet.parse('day=06:00-18:00,night=18:00-06:00').coverageIssues(), []);
+  assert.ok(ShiftSet.parse('m=06:00-12:00,e=14:00-20:00').coverageIssues()
+    .some((s) => s.toLowerCase().includes('gap')));
+  assert.ok(ShiftSet.parse('a=06:00-13:00,b=12:00-18:00').coverageIssues()
+    .some((s) => s.toLowerCase().includes('overlap')));
+});