Real Excel test 2: bug fixes in src, new example, text

.gitattributes

@@ -34,3 +34,5 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
 media/weekly-analyst-demo.mp4 filter=lfs diff=lfs merge=lfs -text
+examples/online-retail-ii-p4.png filter=lfs diff=lfs merge=lfs -text
+examples/online-retail-ii-p6.png filter=lfs diff=lfs merge=lfs -text

README.md

@@ -49,7 +49,7 @@ MiniCPM-V 4.7 1B (f16) runs locally through llama.cpp. It selects tools and para
 | Real model, quiet week | silent in about 1 s |
 | Real model, spike or drop week | PDF draft in about 195 s on a 2-vCPU Colab, about 5 min on the Space; approve re-renders, re-checks and writes it to the sent folder |
 | Peak memory | 5.2 GB |
-| Test suite (full project) | 656 passed, 2 skipped, 0 failed on 2026-11-12 (658 collected; includes 2 new tests for the number-check fix) |
+| Test suite (full project) | 659 passed, 2 skipped, 0 failed on 2026-11-13 (includes 5 new tests for bugs found on real data) |
 | First weight edit (16,384 values, no training; superseded by the LoRA), held-out plans, 80 tasks, no grammar | 64/80 to 80/80; a random edit of the same size stayed at 64/80 |
 | Small LoRA, round 2 (`lora3`), scored once on held-out seeds 400 to 439 (deployed on the Space) | plans 80/80, drafts 80/80 and 40/40 on each of six variants without a grammar, including two held-out reworded instructions (unedited model: 65/80, 0/80, 0/40 each) |
 | First real-data test (UCI Online Retail, about 540,000 invoice rows, T4, model output unconstrained) | valid plan and draft each time; after we fixed a bug in our own number check, a full report for 3 of 3 weeks (it halted on all 3 before the fix) |
@@ -110,6 +110,19 @@ Example output for the week of 2011-12-04 (ISO week 2011-W48), made with `script
 |---|---|---|
  An earlier try on the Tableau Superstore sample stayed silent because its segments were far below the size minimum.
 
+### A second real Excel file (UCI Online Retail II, 2026-11-13)
+
+We pointed the app at a second real workbook, the UCI Online Retail II sheet for 2009-2010 (525,461 invoice lines, 53 weeks), on a T4 with `lora3` and unconstrained model calls. It found three bugs of ours, now fixed and covered by tests:
+
+- A raw `.xlsx` halted with "file is not valid UTF-8". It now says the file is an Excel workbook and must be exported as CSV first.
+- A first run crashed with `unable to open database file` when the work folder did not exist yet. The store now creates it.
+- Our number check and the model-path calculation looked for a column literally named `date` and ignored `date_column` from the metrics file. All four weeks failed until we fixed it. After the fix, 4 of 4 weeks produced a full report.
+
+I recomputed the headline numbers of one report from the CSV and they match the PDF exactly. The draft text is as formulaic as before. Details: [results/real-excel-online-retail-ii.md](results/real-excel-online-retail-ii.md). Example: [report PDF](examples/online-retail-ii-2010-w47.pdf), the [model's replies](examples/online-retail-ii-model-replies.json), the converter `scripts/online_retail_ii_to_csv.py` and the metrics file `examples/metrics_online_retail_ii.yaml`.
+
+| ![page 1](examples/online-retail-ii-p1.png) | ![page 2](examples/online-retail-ii-p2.png) | ![page 3](examples/online-retail-ii-p3.png) |
+|---|---|---|
+
 ## Honest limits
 
 - Detection: recall 0.21 on the holdout replay at the quiet gate (the bar was 0.80), and 14% of quiet weeks raised an alert. We know why, we have not fixed it.
@@ -117,13 +130,13 @@ Example output for the week of 2011-12-04 (ISO week 2011-W48), made with `script
 - All scores are on synthetic tasks from one generator, without a JSON grammar, greedy, 80 tasks scored once per round. In production the agent uses a grammar and retries, so the real gain is probably smaller. The first LoRA failed an instruction that did not name the JSON fields (0/40); round 2 fixed that. Dev wordings were written in the same session as the training wordings, so dev was optimistic; the fresh-seed score is the clean number.
 - The Space runs the round 2 GGUF; its llama.cpp check used unspent seeds 500 to 511 plus a held-out wording. The real-data test used one dataset and was not run on the Space.
 - Free SQL: the unedited model got 21 of 30 under a grammar (bar 24), so named queries are the main path. This was not re-measured with the LoRA.
-- Real data: one public dataset (UCI Online Retail). The draft text there is formulaic; the numbers come from checked code. The product reads CSV only, so an Excel sheet must be converted first (`scripts/online_retail_to_csv.py` is one example).
+- Real data: two public datasets (UCI Online Retail and Online Retail II). The draft text there is formulaic; the numbers come from checked code. The product reads CSV only, so an Excel sheet must be converted first (`scripts/online_retail_to_csv.py` is one example).
 - Chart reading by the vision encoder is not claimed (the vision projector would not load in the runtime used). CPU only on the Space; Apple Silicon speed is untested.
 - The scenarios on the Space are synthetic. Nothing is uploaded, and the source file is never modified.
 
 ## Run the tests
 
-The full test suite is in `tests/` of the GitHub repo (with `eval/` and `pyproject.toml`): `pip install -e ".[dev]" && pytest`. It uses a stand-in model, not the real one, and takes about 10 minutes on a 12-core machine when split across processes (about 25 minutes in one process). On 2026-11-12: 656 passed, 2 skipped, 0 failed.
+The full test suite is in `tests/` of the GitHub repo (with `eval/` and `pyproject.toml`): `pip install -e ".[dev]" && pytest`. It uses a stand-in model, not the real one, and takes about 10 minutes on a 12-core machine when split across processes (about 25 minutes in one process). On 2026-11-13: 659 passed, 2 skipped, 0 failed.
 
 ## Run it
 

app.py

@@ -272,7 +272,7 @@ Set `WEEKLY_ANALYST_LORA=0` for the older 16,384-value plan edit.
 Synthetic tasks from one generator. Reword b and c were never trained on.
 
 ### Honest limits
-- The scenarios on this Space and the reliability figures are synthetic. The one real-data test is described below.
+- The scenarios on this Space and the reliability figures are synthetic. The real-data tests are described below.
 - For free SQL the unedited 1B model got 21 of 30 questions right under a grammar, short of the 24 of 30 bar, so named,
   parameterized queries are the main path and free SQL is a last resort. That was measured before the LoRA; free SQL
   was not re-measured with the LoRA model this Space runs.
@@ -281,8 +281,10 @@ Synthetic tasks from one generator. Reword b and c were never trained on.
 - The change detector's recall on the holdout replay is 0.21 (the bar was 0.80), and 14% of quiet weeks raised an alert.
 - The LoRA and weight-edit scores are synthetic, from one task generator, scored once. A first real-data test (the UCI Online
   Retail sheet, about 540,000 invoice rows, run on a T4 and not on this Space) gave a valid plan and draft each time, and
-  after we fixed a bug in our own number check the pipeline produced a full report for 3 of 3 weeks. Nothing has been
-  measured for general capabilities beyond the agent's task.
+  after we fixed a bug in our own number check the pipeline produced a full report for 3 of 3 weeks. A second workbook
+  (Online Retail II, 525,461 lines) produced a full report for 4 of 4 weeks after we fixed three more bugs of ours: a
+  misleading error for a raw .xlsx, a crash when the work folder did not exist, and a hard-coded `date` column name.
+  The agent still reads CSV only. Nothing has been measured for general capabilities beyond the agent's task.
 """
 
 with gr.Blocks(title="Weekly Analyst") as demo_app:
@@ -308,10 +310,10 @@ with gr.Blocks(title="Weekly Analyst") as demo_app:
     with gr.Tab("Example reports (already run)"):
         gr.Markdown(
             "The first two reports (synthetic data) were produced by this same code with the real model and the earlier "
-            "plan edit, on a 2-core CPU, then approved with the real approve command. The third is real public data "
-            "(UCI Online Retail, week of 2011-12-04) made with the current LoRA model on a T4; its draft text is formulaic."
+            "plan edit, on a 2-core CPU, then approved with the real approve command. The last two are real public Excel data "
+            "(UCI Online Retail, week of 2011-12-04, and Online Retail II, 2010-W47) made with the current LoRA model on a T4; their draft text is formulaic."
         )
-        for prefix, title in (("drop", "Sales drop"), ("spike", "Sales spike"), ("online-retail", "Real data: Online Retail, 2011-W48")):
+        for prefix, title in (("drop", "Sales drop"), ("spike", "Sales spike"), ("online-retail", "Real data: Online Retail, 2011-W48"), ("online-retail-ii", "Real data: Online Retail II, 2010-W47")):
             pages, pdf = example_files(prefix)
             gr.Markdown(f"#### {title}")
             gr.Gallery(value=pages, label=title, columns=3, height=420, object_fit="contain")

examples/online-retail-ii-model-replies.json

@@ -0,0 +1,71 @@
+{
+ "note": "The model's raw replies for the week 2010-W47 (cut 2010-11-28), lora3 GGUF, T4, no JSON schema or grammar. plan = first reply, draft = second.",
+ "plan": {
+  "steps": [
+   {
+    "item_key": "return_rate|2010-W47|Country=EIRE,CustomerType=registered",
+    "tool": "named_query",
+    "args": {
+     "name": "by_segment",
+     "params": {
+      "metric": "return_rate",
+      "segment_col": "Country",
+      "period": "2010-W47"
+     }
+    }
+   },
+   {
+    "item_key": "return_rate|2010-W47|Country=Germany,CustomerType=registered",
+    "tool": "named_query",
+    "args": {
+     "name": "by_segment",
+     "params": {
+      "metric": "return_rate",
+      "segment_col": "Country",
+      "period": "2010-W47"
+     }
+    }
+   },
+   {
+    "item_key": "revenue|2010-W47|Country=EIRE,CustomerType=registered",
+    "tool": "named_query",
+    "args": {
+     "name": "by_segment",
+     "params": {
+      "metric": "revenue",
+      "segment_col": "Country",
+      "period": "2010-W47"
+     }
+    }
+   }
+  ]
+ },
+ "draft": {
+  "headline": "Return rate moved up led by EIRE registered",
+  "recommended_action": "Review the affected segments with the owning team",
+  "findings": [
+   {
+    "item_key": "return_rate|2010-W47|Country=EIRE,CustomerType=registered",
+    "so_what": "Return rate moved up in EIRE registered",
+    "metric_refs": [
+     "return_rate"
+    ]
+   },
+   {
+    "item_key": "return_rate|2010-W47|Country=Germany,CustomerType=registered",
+    "so_what": "Return rate moved down in Germany registered",
+    "metric_refs": [
+     "return_rate"
+    ]
+   },
+   {
+    "item_key": "revenue|2010-W47|Country=EIRE,CustomerType=registered",
+    "so_what": "Revenue moved down in EIRE registered",
+    "metric_refs": [
+     "revenue"
+    ]
+   }
+  ],
+  "caveats": []
+ }
+}

src/weekly_analyst/analysis/checks.py

@@ -1,87 +1,87 @@
-"""Deterministic report checks on the rendered PDF text (spec 6.3)."""
-
-from collections.abc import Sequence
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Literal
-
-from weekly_analyst.analysis.analyst import wording_check
-from weekly_analyst.core.text import numerals
-from weekly_analyst.data.metrics_file import MetricDef, MetricsFile
-from weekly_analyst.data.recompute import matches_displayed, recompute
-
-
-@dataclass
-class ShownNumber:
-    text: str
-    metric: str
-    segment: dict[str, str] | None
-    period: str
-    prev_period: str | None = None
-
-
-@dataclass
-class Failure:
-    kind: Literal["wrong_number", "no_metric", "wording", "unlisted_number"]
-    detail: str
-
-
-def _recomputed(snapshot_path: Path, metric: MetricDef, shown: ShownNumber) -> float | None:
-    """The value a shown number must match: its level, or the signed change from prev_period to period."""
-    now = recompute(snapshot_path, metric, shown.segment, shown.period)
-    if shown.prev_period is None:
-        return now
-    before = recompute(snapshot_path, metric, shown.segment, shown.prev_period)
-    if now is None or before is None:
-        return None
-    return now - before
-
-
-def check_report(
-    pdf_text: str,
-    shown_numbers: list[ShownNumber],
-    metrics: MetricsFile,
-    snapshot_path: Path,
-    exempt: Sequence[str] = (),
-    model_text: str = "",
-) -> list[Failure]:
-    """Check each shown number, the wording, and every numeral on the page.
-
-    exempt holds the numerals the renderer writes itself (dates, page numbers, indices). model_text is
-    the model-written text: every numeral in it must be a shown number, with no exemption.
-    """
-    failures: list[Failure] = []
-    by_name = {m.name: m for m in metrics.metrics}
-
-    for shown in shown_numbers:
-        metric = by_name.get(shown.metric)
-        if metric is None:
-            failures.append(Failure("no_metric", f"{shown.text!r} names no known metric {shown.metric!r}"))
-            continue
-        try:
-            value = _recomputed(snapshot_path, metric, shown)
-            matched = value is not None and matches_displayed(value, shown.text)
-        except ValueError as err:
-            failures.append(Failure("wrong_number", f"{shown.text!r} cannot be verified: {err}"))
-            continue
-        if not matched:
-            failures.append(
-                Failure("wrong_number", f"{shown.text!r} not confirmed by recompute (value {value!r})")
-            )
-
-    for flag in wording_check(pdf_text):
-        failures.append(Failure("wording", flag))
-
-    listed = {shown.text for shown in shown_numbers}
-    model_tokens = set(numerals(model_text))
-    exempt_tokens = set(exempt)
-    for token in dict.fromkeys(numerals(pdf_text)):
-        # A numeral from model text is judged by the stricter model_text rule below, so it is not reported twice.
-        if token not in listed and token not in exempt_tokens and token not in model_tokens:
-            failures.append(Failure("unlisted_number", token))
-
-    for token in dict.fromkeys(numerals(model_text)):
-        if token not in listed:
-            failures.append(Failure("unlisted_number", token))
-
-    return failures
+"""Deterministic report checks on the rendered PDF text (spec 6.3)."""
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Literal
+
+from weekly_analyst.analysis.analyst import wording_check
+from weekly_analyst.core.text import numerals
+from weekly_analyst.data.metrics_file import MetricDef, MetricsFile
+from weekly_analyst.data.recompute import matches_displayed, recompute
+
+
+@dataclass
+class ShownNumber:
+    text: str
+    metric: str
+    segment: dict[str, str] | None
+    period: str
+    prev_period: str | None = None
+
+
+@dataclass
+class Failure:
+    kind: Literal["wrong_number", "no_metric", "wording", "unlisted_number"]
+    detail: str
+
+
+def _recomputed(snapshot_path: Path, metric: MetricDef, shown: ShownNumber, date_column: str) -> float | None:
+    """The value a shown number must match: its level, or the signed change from prev_period to period."""
+    now = recompute(snapshot_path, metric, shown.segment, shown.period, date_column)
+    if shown.prev_period is None:
+        return now
+    before = recompute(snapshot_path, metric, shown.segment, shown.prev_period, date_column)
+    if now is None or before is None:
+        return None
+    return now - before
+
+
+def check_report(
+    pdf_text: str,
+    shown_numbers: list[ShownNumber],
+    metrics: MetricsFile,
+    snapshot_path: Path,
+    exempt: Sequence[str] = (),
+    model_text: str = "",
+) -> list[Failure]:
+    """Check each shown number, the wording, and every numeral on the page.
+
+    exempt holds the numerals the renderer writes itself (dates, page numbers, indices). model_text is
+    the model-written text: every numeral in it must be a shown number, with no exemption.
+    """
+    failures: list[Failure] = []
+    by_name = {m.name: m for m in metrics.metrics}
+
+    for shown in shown_numbers:
+        metric = by_name.get(shown.metric)
+        if metric is None:
+            failures.append(Failure("no_metric", f"{shown.text!r} names no known metric {shown.metric!r}"))
+            continue
+        try:
+            value = _recomputed(snapshot_path, metric, shown, metrics.date_column)
+            matched = value is not None and matches_displayed(value, shown.text)
+        except ValueError as err:
+            failures.append(Failure("wrong_number", f"{shown.text!r} cannot be verified: {err}"))
+            continue
+        if not matched:
+            failures.append(
+                Failure("wrong_number", f"{shown.text!r} not confirmed by recompute (value {value!r})")
+            )
+
+    for flag in wording_check(pdf_text):
+        failures.append(Failure("wording", flag))
+
+    listed = {shown.text for shown in shown_numbers}
+    model_tokens = set(numerals(model_text))
+    exempt_tokens = set(exempt)
+    for token in dict.fromkeys(numerals(pdf_text)):
+        # A numeral from model text is judged by the stricter model_text rule below, so it is not reported twice.
+        if token not in listed and token not in exempt_tokens and token not in model_tokens:
+            failures.append(Failure("unlisted_number", token))
+
+    for token in dict.fromkeys(numerals(model_text)):
+        if token not in listed:
+            failures.append(Failure("unlisted_number", token))
+
+    return failures

src/weekly_analyst/core/store.py

@@ -1,78 +1,79 @@
-"""SQLite state store: weekly alert counts, reported items, run states (spec 6.6)."""
-
-import json
-import sqlite3
-from collections.abc import Iterator
-from contextlib import contextmanager
-from datetime import datetime
-from pathlib import Path
-
-
-def iso_week_of(ts: datetime) -> str:
-    iso = ts.isocalendar()
-    return f"{iso.year}-W{iso.week:02d}"
-
-
-_SCHEMA = """
-CREATE TABLE IF NOT EXISTS alerts (
-    item_key TEXT NOT NULL,
-    iso_week TEXT NOT NULL,
-    run_id TEXT NOT NULL
-);
-CREATE TABLE IF NOT EXISTS runs (
-    run_id TEXT PRIMARY KEY,
-    state TEXT NOT NULL,
-    payload TEXT NOT NULL
-);
-"""
-
-
-class StateStore:
-    def __init__(self, path: Path) -> None:
-        self._path = Path(path)
-        with self._transaction() as conn:
-            conn.executescript(_SCHEMA)
-
-    @contextmanager
-    def _transaction(self) -> Iterator[sqlite3.Connection]:
-        conn = sqlite3.connect(self._path)
-        try:
-            with conn:
-                yield conn
-        finally:
-            conn.close()
-
-    def alerts_sent_in_week(self, iso_week: str) -> int:
-        with self._transaction() as conn:
-            row = conn.execute(
-                "SELECT COUNT(*) FROM alerts WHERE iso_week = ?", (iso_week,)
-            ).fetchone()
-        return int(row[0])
-
-    def record_alert(self, item_key: str, iso_week: str, run_id: str) -> None:
-        with self._transaction() as conn:
-            conn.execute(
-                "INSERT INTO alerts (item_key, iso_week, run_id) VALUES (?, ?, ?)",
-                (item_key, iso_week, run_id),
-            )
-
-    def reported_keys(self) -> set[str]:
-        with self._transaction() as conn:
-            rows = conn.execute("SELECT DISTINCT item_key FROM alerts").fetchall()
-        return {row[0] for row in rows}
-
-    def save_state(self, run_id: str, state: str, payload: dict) -> None:
-        with self._transaction() as conn:
-            conn.execute(
-                "INSERT OR REPLACE INTO runs (run_id, state, payload) VALUES (?, ?, ?)",
-                (run_id, state, json.dumps(payload, sort_keys=True)),
-            )
-
-    def load_run(self, run_id: str) -> dict | None:
-        with self._transaction() as conn:
-            row = conn.execute(
-                "SELECT state, payload FROM runs WHERE run_id = ?", (run_id,)
-            ).fetchone()
-        if row is None:
-            return None
-        return {"run_id": run_id, "state": row[0], "payload": json.loads(row[1])}
+"""SQLite state store: weekly alert counts, reported items, run states (spec 6.6)."""
+
+import json
+import sqlite3
+from collections.abc import Iterator
+from contextlib import contextmanager
+from datetime import datetime
+from pathlib import Path
+
+
+def iso_week_of(ts: datetime) -> str:
+    iso = ts.isocalendar()
+    return f"{iso.year}-W{iso.week:02d}"
+
+
+_SCHEMA = """
+CREATE TABLE IF NOT EXISTS alerts (
+    item_key TEXT NOT NULL,
+    iso_week TEXT NOT NULL,
+    run_id TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS runs (
+    run_id TEXT PRIMARY KEY,
+    state TEXT NOT NULL,
+    payload TEXT NOT NULL
+);
+"""
+
+
+class StateStore:
+    def __init__(self, path: Path) -> None:
+        self._path = Path(path)
+        self._path.parent.mkdir(parents=True, exist_ok=True)  # a first run has no work folder yet
+        with self._transaction() as conn:
+            conn.executescript(_SCHEMA)
+
+    @contextmanager
+    def _transaction(self) -> Iterator[sqlite3.Connection]:
+        conn = sqlite3.connect(self._path)
+        try:
+            with conn:
+                yield conn
+        finally:
+            conn.close()
+
+    def alerts_sent_in_week(self, iso_week: str) -> int:
+        with self._transaction() as conn:
+            row = conn.execute(
+                "SELECT COUNT(*) FROM alerts WHERE iso_week = ?", (iso_week,)
+            ).fetchone()
+        return int(row[0])
+
+    def record_alert(self, item_key: str, iso_week: str, run_id: str) -> None:
+        with self._transaction() as conn:
+            conn.execute(
+                "INSERT INTO alerts (item_key, iso_week, run_id) VALUES (?, ?, ?)",
+                (item_key, iso_week, run_id),
+            )
+
+    def reported_keys(self) -> set[str]:
+        with self._transaction() as conn:
+            rows = conn.execute("SELECT DISTINCT item_key FROM alerts").fetchall()
+        return {row[0] for row in rows}
+
+    def save_state(self, run_id: str, state: str, payload: dict) -> None:
+        with self._transaction() as conn:
+            conn.execute(
+                "INSERT OR REPLACE INTO runs (run_id, state, payload) VALUES (?, ?, ?)",
+                (run_id, state, json.dumps(payload, sort_keys=True)),
+            )
+
+    def load_run(self, run_id: str) -> dict | None:
+        with self._transaction() as conn:
+            row = conn.execute(
+                "SELECT state, payload FROM runs WHERE run_id = ?", (run_id,)
+            ).fetchone()
+        if row is None:
+            return None
+        return {"run_id": run_id, "state": row[0], "payload": json.loads(row[1])}

src/weekly_analyst/data/extract.py

@@ -1,158 +1,162 @@
-"""Read-only CSV extract to a Parquet snapshot (spec 6.2 state 2, 6.5)."""
-
-import codecs
-import csv
-import hashlib
-import logging
-import os
-import re
-import tempfile
-from dataclasses import dataclass
-from datetime import date
-from pathlib import Path
-
-import polars as pl
-
-from weekly_analyst.core.config import Settings
-from weekly_analyst.data.metrics_file import MetricsFile
-
-log = logging.getLogger(__name__)
-
-ISO_DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")
-CHUNK_BYTES = 1 << 20
-
-
-class ExtractError(Exception):
-    def __init__(self, code: str, detail: str):
-        super().__init__(f"{code}: {detail}")
-        self.code = code
-
-
-@dataclass(frozen=True)
-class ExtractResult:
-    snapshot_hash: str
-    parquet_path: Path
-    rows_read: int
-    rows_in_file: int
-
-
-def _halt(code: str, detail: str, rows_read: int, rows_in_file: int | str) -> ExtractError:
-    log.error(
-        "extract halted code=%s rows_read=%s rows_in_file=%s detail=%s",
-        code,
-        rows_read,
-        rows_in_file,
-        detail,
-    )
-    return ExtractError(code, detail)
-
-
-def _hash_and_check_utf8(csv_path: Path) -> str:
-    digest = hashlib.sha256()
-    decoder = codecs.getincrementaldecoder("utf-8")()
-    with open(csv_path, "rb") as fh:
-        for chunk in iter(lambda: fh.read(CHUNK_BYTES), b""):
-            digest.update(chunk)
-            try:
-                decoder.decode(chunk)
-            except UnicodeDecodeError:
-                raise _halt("bad_encoding", "file is not valid UTF-8", 0, "unknown") from None
-    try:
-        decoder.decode(b"", final=True)
-    except UnicodeDecodeError:
-        raise _halt("bad_encoding", "file is not valid UTF-8", 0, "unknown") from None
-    return digest.hexdigest()
-
-
-def _is_iso_date(value: str) -> bool:
-    if not ISO_DATE.fullmatch(value):
-        return False
-    try:
-        date.fromisoformat(value)
-    except ValueError:
-        return False
-    return True
-
-
-def _scan_source(csv_path: Path, metrics: MetricsFile) -> tuple[list[str], int, int | None]:
-    """Stream the file once: check the header, count data rows, find the first bad date."""
-    with open(csv_path, "r", encoding="utf-8-sig", newline="") as fh:
-        reader = csv.reader(fh)
-        header = next(reader, None)
-        if header is None:
-            return [], 0, None
-        if not header:
-            raise _halt("bad_header", "header row is blank", 0, "unknown")
-        if any(";" in cell for cell in header):
-            raise _halt("bad_header", "header uses a semicolon delimiter; use commas", 0, "unknown")
-        if len(set(header)) != len(header):
-            raise _halt("duplicate_columns", "header has duplicate column names", 0, "unknown")
-        if metrics.date_column not in header:
-            raise _halt("bad_header", f"header has no date column {metrics.date_column!r}", 0, "unknown")
-        date_index = header.index(metrics.date_column)
-        ncols = len(header)
-        rows = 0
-        first_bad_date: int | None = None
-        for record in reader:
-            if len(record) != ncols:
-                raise _halt(
-                    "bad_header",
-                    f"data row {rows + 1} has {len(record)} fields, header has {ncols}",
-                    0,
-                    "unknown",
-                )
-            rows += 1
-            if first_bad_date is None and not _is_iso_date(record[date_index]):
-                first_bad_date = rows
-    return header, rows, first_bad_date
-
-
-def _write_atomic(df: pl.DataFrame, final: Path) -> None:
-    fd, tmp_name = tempfile.mkstemp(dir=final.parent, prefix=".extract-", suffix=".tmp")
-    tmp = Path(tmp_name)
-    try:
-        with os.fdopen(fd, "wb") as fh:
-            df.write_parquet(fh)
-            fh.flush()
-            os.fsync(fh.fileno())
-        os.replace(tmp, final)
-    except BaseException:
-        tmp.unlink(missing_ok=True)
-        raise
-
-
-def extract(csv_path: Path, metrics: MetricsFile, out_dir: Path, settings: Settings) -> ExtractResult:
-    size = csv_path.stat().st_size
-    if size > settings.max_bytes:
-        raise _halt("too_big", f"file is {size} bytes, cap is {settings.max_bytes}", 0, "unknown")
-
-    snapshot_hash = _hash_and_check_utf8(csv_path)
-    header, rows, first_bad_date = _scan_source(csv_path, metrics)
-    if rows == 0:
-        raise _halt("empty", "no data rows after the header", 0, 0)
-    if rows > settings.max_rows:
-        raise _halt("too_many_rows", f"{rows} rows, cap is {settings.max_rows}", 0, rows)
-    cells = rows * len(header)
-    if cells > settings.max_cells:
-        raise _halt("too_many_cells", f"{cells} cells, cap is {settings.max_cells}", 0, rows)
-    if first_bad_date is not None:
-        raise _halt("bad_date", f"data row {first_bad_date} is not an ISO date (YYYY-MM-DD)", 0, rows)
-
-    metric_columns = metrics.columns()
-    present = [name for name in header if name in metric_columns]
-    df = pl.read_csv(
-        csv_path,
-        columns=present,
-        schema_overrides={metrics.date_column: pl.Utf8},
-        infer_schema_length=None,
-    )
-    df = df.with_columns(pl.col(metrics.date_column).str.to_date("%Y-%m-%d", strict=True))
-
-    final = out_dir / f"snapshot-{snapshot_hash}.parquet"
-    _write_atomic(df, final)
-    return ExtractResult(
-        snapshot_hash=snapshot_hash,
-        parquet_path=final,
-        rows_read=df.height,
-        rows_in_file=rows,
-    )
+"""Read-only CSV extract to a Parquet snapshot (spec 6.2 state 2, 6.5)."""
+
+import codecs
+import csv
+import hashlib
+import logging
+import os
+import re
+import tempfile
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+
+import polars as pl
+
+from weekly_analyst.core.config import Settings
+from weekly_analyst.data.metrics_file import MetricsFile
+
+log = logging.getLogger(__name__)
+
+ISO_DATE = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")
+CHUNK_BYTES = 1 << 20
+WORKBOOK_MAGIC = (b"PK\x03\x04", b"\xd0\xcf\x11\xe0")  # .xlsx (zip) and old .xls (OLE)
+
+
+class ExtractError(Exception):
+    def __init__(self, code: str, detail: str):
+        super().__init__(f"{code}: {detail}")
+        self.code = code
+
+
+@dataclass(frozen=True)
+class ExtractResult:
+    snapshot_hash: str
+    parquet_path: Path
+    rows_read: int
+    rows_in_file: int
+
+
+def _halt(code: str, detail: str, rows_read: int, rows_in_file: int | str) -> ExtractError:
+    log.error(
+        "extract halted code=%s rows_read=%s rows_in_file=%s detail=%s",
+        code,
+        rows_read,
+        rows_in_file,
+        detail,
+    )
+    return ExtractError(code, detail)
+
+
+def _hash_and_check_utf8(csv_path: Path) -> str:
+    digest = hashlib.sha256()
+    decoder = codecs.getincrementaldecoder("utf-8")()
+    with open(csv_path, "rb") as fh:
+        if fh.read(4) in WORKBOOK_MAGIC:
+            raise _halt("not_csv", "this is an Excel workbook, not a CSV; export the sheet as CSV (UTF-8) first", 0, "unknown")
+        fh.seek(0)
+        for chunk in iter(lambda: fh.read(CHUNK_BYTES), b""):
+            digest.update(chunk)
+            try:
+                decoder.decode(chunk)
+            except UnicodeDecodeError:
+                raise _halt("bad_encoding", "file is not valid UTF-8", 0, "unknown") from None
+    try:
+        decoder.decode(b"", final=True)
+    except UnicodeDecodeError:
+        raise _halt("bad_encoding", "file is not valid UTF-8", 0, "unknown") from None
+    return digest.hexdigest()
+
+
+def _is_iso_date(value: str) -> bool:
+    if not ISO_DATE.fullmatch(value):
+        return False
+    try:
+        date.fromisoformat(value)
+    except ValueError:
+        return False
+    return True
+
+
+def _scan_source(csv_path: Path, metrics: MetricsFile) -> tuple[list[str], int, int | None]:
+    """Stream the file once: check the header, count data rows, find the first bad date."""
+    with open(csv_path, "r", encoding="utf-8-sig", newline="") as fh:
+        reader = csv.reader(fh)
+        header = next(reader, None)
+        if header is None:
+            return [], 0, None
+        if not header:
+            raise _halt("bad_header", "header row is blank", 0, "unknown")
+        if any(";" in cell for cell in header):
+            raise _halt("bad_header", "header uses a semicolon delimiter; use commas", 0, "unknown")
+        if len(set(header)) != len(header):
+            raise _halt("duplicate_columns", "header has duplicate column names", 0, "unknown")
+        if metrics.date_column not in header:
+            raise _halt("bad_header", f"header has no date column {metrics.date_column!r}", 0, "unknown")
+        date_index = header.index(metrics.date_column)
+        ncols = len(header)
+        rows = 0
+        first_bad_date: int | None = None
+        for record in reader:
+            if len(record) != ncols:
+                raise _halt(
+                    "bad_header",
+                    f"data row {rows + 1} has {len(record)} fields, header has {ncols}",
+                    0,
+                    "unknown",
+                )
+            rows += 1
+            if first_bad_date is None and not _is_iso_date(record[date_index]):
+                first_bad_date = rows
+    return header, rows, first_bad_date
+
+
+def _write_atomic(df: pl.DataFrame, final: Path) -> None:
+    fd, tmp_name = tempfile.mkstemp(dir=final.parent, prefix=".extract-", suffix=".tmp")
+    tmp = Path(tmp_name)
+    try:
+        with os.fdopen(fd, "wb") as fh:
+            df.write_parquet(fh)
+            fh.flush()
+            os.fsync(fh.fileno())
+        os.replace(tmp, final)
+    except BaseException:
+        tmp.unlink(missing_ok=True)
+        raise
+
+
+def extract(csv_path: Path, metrics: MetricsFile, out_dir: Path, settings: Settings) -> ExtractResult:
+    size = csv_path.stat().st_size
+    if size > settings.max_bytes:
+        raise _halt("too_big", f"file is {size} bytes, cap is {settings.max_bytes}", 0, "unknown")
+
+    snapshot_hash = _hash_and_check_utf8(csv_path)
+    header, rows, first_bad_date = _scan_source(csv_path, metrics)
+    if rows == 0:
+        raise _halt("empty", "no data rows after the header", 0, 0)
+    if rows > settings.max_rows:
+        raise _halt("too_many_rows", f"{rows} rows, cap is {settings.max_rows}", 0, rows)
+    cells = rows * len(header)
+    if cells > settings.max_cells:
+        raise _halt("too_many_cells", f"{cells} cells, cap is {settings.max_cells}", 0, rows)
+    if first_bad_date is not None:
+        raise _halt("bad_date", f"data row {first_bad_date} is not an ISO date (YYYY-MM-DD)", 0, rows)
+
+    metric_columns = metrics.columns()
+    present = [name for name in header if name in metric_columns]
+    df = pl.read_csv(
+        csv_path,
+        columns=present,
+        schema_overrides={metrics.date_column: pl.Utf8},
+        infer_schema_length=None,
+    )
+    df = df.with_columns(pl.col(metrics.date_column).str.to_date("%Y-%m-%d", strict=True))
+
+    final = out_dir / f"snapshot-{snapshot_hash}.parquet"
+    _write_atomic(df, final)
+    return ExtractResult(
+        snapshot_hash=snapshot_hash,
+        parquet_path=final,
+        rows_read=df.height,
+        rows_in_file=rows,
+    )

src/weekly_analyst/data/metrics_calc.py

@@ -1,38 +1,38 @@
-"""Model-path metric computation with polars (spec 6.3, 6.8).
-
-Independent of recompute.py by design: no shared code or engine.
-"""
-
-import math
-
-import polars as pl
-
-from weekly_analyst.core.config import Settings
-from weekly_analyst.data.metrics_file import MetricDef
-
-DATE_COLUMN = "date"
-
-
-def compute(
-    df: pl.DataFrame, m: MetricDef, segment: dict[str, str] | None, period: str
-) -> float | None:
-    """Value of metric m for one ISO week (e.g. "2026-W46") and segment.
-
-    Returns None when the slice has fewer rows than Settings().min_rows_per_segment,
-    or when there are no non-null values to aggregate.
-    """
-    predicate = pl.col(DATE_COLUMN).cast(pl.Date).dt.strftime("%G-W%V") == period
-    for key, value in (segment or {}).items():
-        predicate = predicate & (pl.col(key) == value)
-    rows = df.filter(predicate)
-    if rows.height < Settings().min_rows_per_segment:
-        return None
-    values = rows.get_column(m.column).drop_nulls()
-    if m.agg == "count":
-        return float(values.len())
-    if values.len() == 0:
-        return None
-    result = values.sum() if m.agg == "sum" else values.mean()
-    if result is None or math.isnan(result):
-        return None
-    return float(result)
+"""Model-path metric computation with polars (spec 6.3, 6.8).
+
+Independent of recompute.py by design: no shared code or engine.
+"""
+
+import math
+
+import polars as pl
+
+from weekly_analyst.core.config import Settings
+from weekly_analyst.data.metrics_file import MetricDef
+
+DATE_COLUMN = "date"
+
+
+def compute(
+    df: pl.DataFrame, m: MetricDef, segment: dict[str, str] | None, period: str, date_column: str = DATE_COLUMN
+) -> float | None:
+    """Value of metric m for one ISO week (e.g. "2026-W46") and segment.
+
+    Returns None when the slice has fewer rows than Settings().min_rows_per_segment,
+    or when there are no non-null values to aggregate.
+    """
+    predicate = pl.col(date_column).cast(pl.Date).dt.strftime("%G-W%V") == period
+    for key, value in (segment or {}).items():
+        predicate = predicate & (pl.col(key) == value)
+    rows = df.filter(predicate)
+    if rows.height < Settings().min_rows_per_segment:
+        return None
+    values = rows.get_column(m.column).drop_nulls()
+    if m.agg == "count":
+        return float(values.len())
+    if values.len() == 0:
+        return None
+    result = values.sum() if m.agg == "sum" else values.mean()
+    if result is None or math.isnan(result):
+        return None
+    return float(result)

src/weekly_analyst/data/recompute.py

@@ -1,78 +1,80 @@
-"""Independent recompute of a metric from the Parquet snapshot (spec 6.3).
-
-Deliberately separate from metrics_calc: pyarrow and pyarrow.compute only, so the
-two implementations share neither engine nor code.
-"""
-
-import math
-import re
-from decimal import ROUND_HALF_UP, Decimal
-from pathlib import Path
-
-import pyarrow as pa
-import pyarrow.compute as pc
-import pyarrow.parquet as pq
-
-from weekly_analyst.core.config import Settings
-from weekly_analyst.data.metrics_file import MetricDef
-
-# The snapshot's date column. recompute's signature carries no date column name,
-# and the simulator and metrics fixture both use "date".
-DATE_COLUMN = "date"
-
-_PERIOD = re.compile(r"^(\d{4})-W(\d{2})$")
-_SHOWN = re.compile(
-    r"^(?P<sign>[+-]?)(?P<int>\d{1,3}(?:,\d{3})+|\d+)(?:\.(?P<frac>\d+))?(?P<suffix>[KMB%]?)$"
-)
-# Divisor that turns the raw value into the unit shown ("%" shows a fraction as percent).
-_SCALE = {
-    "": Decimal(1),
-    "K": Decimal(10**3),
-    "M": Decimal(10**6),
-    "B": Decimal(10**9),
-    "%": Decimal("0.01"),
-}
-
-
-def _parse_period(period: str) -> tuple[int, int]:
-    match = _PERIOD.match(period)
-    if match is None:
-        raise ValueError(f"period must look like 2026-W46, got {period!r}")
-    return int(match.group(1)), int(match.group(2))
-
-
-def recompute(snapshot_path: Path, m: MetricDef, segment: dict[str, str] | None, period: str) -> float | None:
-    year, week = _parse_period(period)
-    segment = segment or {}
-    table = pq.read_table(snapshot_path, columns=[DATE_COLUMN, m.column, *segment])
-
-    dates = pc.cast(table[DATE_COLUMN], pa.date32())
-    mask = pc.and_(pc.equal(pc.iso_year(dates), year), pc.equal(pc.iso_week(dates), week))
-    for key, value in segment.items():
-        mask = pc.and_(mask, pc.equal(pc.cast(table[key], pa.string()), value))
-    rows = table.filter(mask, null_selection_behavior="drop")
-
-    if rows.num_rows < Settings().min_rows_per_segment:
-        return None
-    values = rows[m.column]
-    if m.agg == "sum":
-        result = pc.sum(values)
-    elif m.agg == "mean":
-        result = pc.mean(values)
-    else:
-        result = pc.count(values)
-    result = result.as_py()
-    return None if result is None else float(result)
-
-
-def matches_displayed(value: float, shown: str) -> bool:
-    match = _SHOWN.match(shown.strip())
-    if match is None:
-        raise ValueError(f"unrecognised displayed number: {shown!r}")
-    if not math.isfinite(value):
-        return False
-    frac = match.group("frac") or ""
-    expected = Decimal(match.group("sign") + match.group("int").replace(",", "") + ("." + frac if frac else ""))
-    scaled = Decimal(repr(float(value))) / _SCALE[match.group("suffix")]
-    rounded = scaled.quantize(Decimal(1).scaleb(-len(frac)), rounding=ROUND_HALF_UP)
-    return rounded == expected
+"""Independent recompute of a metric from the Parquet snapshot (spec 6.3).
+
+Deliberately separate from metrics_calc: pyarrow and pyarrow.compute only, so the
+two implementations share neither engine nor code.
+"""
+
+import math
+import re
+from decimal import ROUND_HALF_UP, Decimal
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.compute as pc
+import pyarrow.parquet as pq
+
+from weekly_analyst.core.config import Settings
+from weekly_analyst.data.metrics_file import MetricDef
+
+# The snapshot's date column. recompute's signature carries no date column name,
+# and the simulator and metrics fixture both use "date".
+DATE_COLUMN = "date"
+
+_PERIOD = re.compile(r"^(\d{4})-W(\d{2})$")
+_SHOWN = re.compile(
+    r"^(?P<sign>[+-]?)(?P<int>\d{1,3}(?:,\d{3})+|\d+)(?:\.(?P<frac>\d+))?(?P<suffix>[KMB%]?)$"
+)
+# Divisor that turns the raw value into the unit shown ("%" shows a fraction as percent).
+_SCALE = {
+    "": Decimal(1),
+    "K": Decimal(10**3),
+    "M": Decimal(10**6),
+    "B": Decimal(10**9),
+    "%": Decimal("0.01"),
+}
+
+
+def _parse_period(period: str) -> tuple[int, int]:
+    match = _PERIOD.match(period)
+    if match is None:
+        raise ValueError(f"period must look like 2026-W46, got {period!r}")
+    return int(match.group(1)), int(match.group(2))
+
+
+def recompute(
+    snapshot_path: Path, m: MetricDef, segment: dict[str, str] | None, period: str, date_column: str = DATE_COLUMN
+) -> float | None:
+    year, week = _parse_period(period)
+    segment = segment or {}
+    table = pq.read_table(snapshot_path, columns=[date_column, m.column, *segment])
+
+    dates = pc.cast(table[date_column], pa.date32())
+    mask = pc.and_(pc.equal(pc.iso_year(dates), year), pc.equal(pc.iso_week(dates), week))
+    for key, value in segment.items():
+        mask = pc.and_(mask, pc.equal(pc.cast(table[key], pa.string()), value))
+    rows = table.filter(mask, null_selection_behavior="drop")
+
+    if rows.num_rows < Settings().min_rows_per_segment:
+        return None
+    values = rows[m.column]
+    if m.agg == "sum":
+        result = pc.sum(values)
+    elif m.agg == "mean":
+        result = pc.mean(values)
+    else:
+        result = pc.count(values)
+    result = result.as_py()
+    return None if result is None else float(result)
+
+
+def matches_displayed(value: float, shown: str) -> bool:
+    match = _SHOWN.match(shown.strip())
+    if match is None:
+        raise ValueError(f"unrecognised displayed number: {shown!r}")
+    if not math.isfinite(value):
+        return False
+    frac = match.group("frac") or ""
+    expected = Decimal(match.group("sign") + match.group("int").replace(",", "") + ("." + frac if frac else ""))
+    scaled = Decimal(repr(float(value))) / _SCALE[match.group("suffix")]
+    rounded = scaled.quantize(Decimal(1).scaleb(-len(frac)), rounding=ROUND_HALF_UP)
+    return rounded == expected

src/weekly_analyst/systems/detect.py

@@ -1,59 +1,59 @@
-"""Detect state: score and rank candidate changes, then select alerts within the weekly budget (spec 6.2, 6.6)."""
-
-import math
-from dataclasses import replace
-
-from weekly_analyst.core.components import ItemsC, SnapshotC, ValidationC
-from weekly_analyst.core.context import RunContext
-from weekly_analyst.core.entity import Entity
-from weekly_analyst.core.items import Item
-from weekly_analyst.core.system import System
-from weekly_analyst.core.world import World
-from weekly_analyst.data.metrics_calc import compute
-from weekly_analyst.detection import remaining_budget, score_items, select_alerts
-from weekly_analyst.systems.common import frame, metric_map, prev_period, split
-
-
-class DetectSystem(System):
-    name = "detect"
-    requires = (ValidationC, SnapshotC)
-    produces = (ItemsC,)
-
-    def run(self, world: World, entity: Entity, ctx: RunContext) -> None:
-        validation = world.get(entity, ValidationC)
-        df = frame(world, entity, ctx)
-        history, current, period = split(df, ctx.metrics.date_column)
-        selected: list[Item] = []
-        if not validation.cold_start:
-            items = score_items(history, current, ctx.metrics, ctx.settings)
-            prev = prev_period(period)
-            priced = [
-                replace(item, detail={**item.detail, "abs_change": _abs_change(ctx, df, item, prev)})
-                for item in items
-            ]
-            min_change = {m.name: m.min_change for m in ctx.metrics.metrics}
-            budget = remaining_budget(ctx.store, period, ctx.settings)
-            selected = select_alerts(priced, budget, min_change, ctx.store.reported_keys())
-        schema_change = any(f["kind"] == "schema_change" for f in validation.findings)
-        world.add(
-            entity,
-            ItemsC(
-                selected=selected,
-                material=bool(selected) or schema_change,
-                alerts_before=ctx.store.alerts_sent_in_week(period),
-            ),
-        )
-
-    def ends_run(self, world: World, entity: Entity) -> bool:
-        """A quiet snapshot (no material item, no schema change) ends after detect; nothing is sent."""
-        return not world.get(entity, ItemsC).material
-
-
-def _abs_change(ctx: RunContext, df, item: Item, prev: str) -> float:
-    metric = metric_map(ctx)[item.metric]
-    segment = dict(item.segment)
-    now = compute(df, metric, segment, item.period)
-    before = compute(df, metric, segment, prev)
-    if now is None or before is None:
-        return math.nan  # a NaN change fails the minimum-change gate
-    return abs(now - before)
+"""Detect state: score and rank candidate changes, then select alerts within the weekly budget (spec 6.2, 6.6)."""
+
+import math
+from dataclasses import replace
+
+from weekly_analyst.core.components import ItemsC, SnapshotC, ValidationC
+from weekly_analyst.core.context import RunContext
+from weekly_analyst.core.entity import Entity
+from weekly_analyst.core.items import Item
+from weekly_analyst.core.system import System
+from weekly_analyst.core.world import World
+from weekly_analyst.data.metrics_calc import compute
+from weekly_analyst.detection import remaining_budget, score_items, select_alerts
+from weekly_analyst.systems.common import frame, metric_map, prev_period, split
+
+
+class DetectSystem(System):
+    name = "detect"
+    requires = (ValidationC, SnapshotC)
+    produces = (ItemsC,)
+
+    def run(self, world: World, entity: Entity, ctx: RunContext) -> None:
+        validation = world.get(entity, ValidationC)
+        df = frame(world, entity, ctx)
+        history, current, period = split(df, ctx.metrics.date_column)
+        selected: list[Item] = []
+        if not validation.cold_start:
+            items = score_items(history, current, ctx.metrics, ctx.settings)
+            prev = prev_period(period)
+            priced = [
+                replace(item, detail={**item.detail, "abs_change": _abs_change(ctx, df, item, prev)})
+                for item in items
+            ]
+            min_change = {m.name: m.min_change for m in ctx.metrics.metrics}
+            budget = remaining_budget(ctx.store, period, ctx.settings)
+            selected = select_alerts(priced, budget, min_change, ctx.store.reported_keys())
+        schema_change = any(f["kind"] == "schema_change" for f in validation.findings)
+        world.add(
+            entity,
+            ItemsC(
+                selected=selected,
+                material=bool(selected) or schema_change,
+                alerts_before=ctx.store.alerts_sent_in_week(period),
+            ),
+        )
+
+    def ends_run(self, world: World, entity: Entity) -> bool:
+        """A quiet snapshot (no material item, no schema change) ends after detect; nothing is sent."""
+        return not world.get(entity, ItemsC).material
+
+
+def _abs_change(ctx: RunContext, df, item: Item, prev: str) -> float:
+    metric = metric_map(ctx)[item.metric]
+    segment = dict(item.segment)
+    now = compute(df, metric, segment, item.period, ctx.metrics.date_column)
+    before = compute(df, metric, segment, prev, ctx.metrics.date_column)
+    if now is None or before is None:
+        return math.nan  # a NaN change fails the minimum-change gate
+    return abs(now - before)

src/weekly_analyst/systems/draft.py

@@ -38,8 +38,8 @@ def _direction(df, ctx: RunContext, item) -> str:
     """up, down or unchanged for the item's metric against the prior week: words only, no numbers for the model."""
     metric = metric_map(ctx)[item.metric]
     segment = dict(item.segment)
-    now = compute(df, metric, segment, item.period)
-    before = compute(df, metric, segment, prev_period(item.period))
+    now = compute(df, metric, segment, item.period, ctx.metrics.date_column)
+    before = compute(df, metric, segment, prev_period(item.period), ctx.metrics.date_column)
     if now is None or before is None:
         return "unknown"
     return "up" if now > before else "down" if now < before else "unchanged"

src/weekly_analyst/systems/report_findings.py

@@ -23,7 +23,7 @@ def finding_for(ctx: RunContext, df: pl.DataFrame, item: Item, text) -> tuple[Fi
     shown = public_item(item)  # what the report prints
     label = segment_label(dict(shown.segment))
     key = plan_item_key(shown)
-    level = compute(df, metric, segment, item.period)
+    level = compute(df, metric, segment, item.period, ctx.metrics.date_column)
     if level is None:
         raise Halt(f"{key}: too few rows to report")
     points = series(ctx, df, metric, segment)
@@ -66,7 +66,7 @@ def attribution_rows(df: pl.DataFrame, date_col: str, metric, col: str, period:
     rows = []
     for r in attribute_change(week(df, date_col, prev), week(df, date_col, period), metric, col).iter_rows(named=True):
         segment = {col: str(r["segment"])}
-        now, before = compute(df, metric, segment, period), compute(df, metric, segment, prev)
+        now, before = compute(df, metric, segment, period, date_col), compute(df, metric, segment, prev, date_col)
         if now is not None and before is not None:
             rows.append(AttributionRow(segment=sanitize.label(str(r["segment"])), change=ShownNumber(change_text(now - before), metric.name, segment, period)))
     return rows
@@ -76,7 +76,7 @@ def series(ctx: RunContext, df: pl.DataFrame, metric, segment: dict) -> list[tup
     periods = df.select(period_expr(ctx.metrics.date_column).unique().sort()).to_series().to_list()
     points = []
     for period in periods[-SAMPLE_WEEKS:]:
-        value = compute(df, metric, segment, period)
+        value = compute(df, metric, segment, period, ctx.metrics.date_column)
         if value is not None:
             points.append((period, "W" + period.split("-W", 1)[1], value))
     return points

src/weekly_analyst/systems/report_model.py

@@ -1,94 +1,94 @@
-"""Report model: the input the fixed PDF template renders, built from the draft and the run's components (spec 6.8)."""
-
-import polars as pl
-
-from weekly_analyst.analysis.draft import DraftFields
-from weekly_analyst.analysis.plan import _item_key as plan_item_key
-from weekly_analyst.core.components import EvidenceC, SourceC, ValidationC
-from weekly_analyst.core.context import RunContext
-from weekly_analyst.core.entity import Entity
-from weekly_analyst.core.items import Item
-from weekly_analyst.core.world import World
-from weekly_analyst.data.metrics_calc import compute
-from weekly_analyst.report.model import DataTable, Finding, Kpi, MetricDefinition, ReportModel, ShownNumber
-from weekly_analyst.systems.common import frame, metric_map, prev_period, public_item, raw_items
-from weekly_analyst.systems.report_findings import change_text, finding_for, level_text
-
-
-def build_report(world: World, entity: Entity, ctx: RunContext, draft: DraftFields) -> ReportModel:
-    df = frame(world, entity, ctx)
-    validation = world.get(entity, ValidationC)
-    period = validation.period
-    prev = prev_period(period)
-    items = raw_items(world, entity)
-    texts = {f.item_key: f for f in draft.findings}
-    findings: list[Finding] = []
-    tables: list[DataTable] = []
-    for item in items:
-        found, table = finding_for(ctx, df, item, texts[plan_item_key(public_item(item))])
-        findings.append(found)
-        tables.append(table)
-    snapshot_date = df[ctx.metrics.date_column].max().isoformat()
-    risks = list(draft.caveats) + [
-        f"{f['kind'].replace('_', ' ')}: {f['detail']}" for f in validation.findings
-    ]
-    failed = [r for r in world.get(entity, EvidenceC).results if "error" in r]
-    if failed:
-        risks.append(f"{len(failed)} planned follow-up check(s) could not be run and are not reflected in this report")
-    if validation.cold_start:
-        risks.append(
-            f"insufficient history: fewer than {ctx.settings.cold_start_periods} prior weeks, "
-            "so no change alerts were sent"
-        )
-    next_checks = [f"Check the source rows behind {plan_item_key(public_item(item))}" for item in items]
-    if any(f["kind"] == "schema_change" for f in validation.findings):
-        next_checks.append("Confirm the schema change with the owner of the source file")
-    return ReportModel(
-        report_name="Weekly analysis",
-        reporting_period=period,
-        prior_period=prev,
-        data_as_of=snapshot_date,
-        prepared_for=", ".join(ctx.recipients) or "Recipients not set",
-        report_version="1",
-        run_id=entity.run_id,
-        snapshot_date=snapshot_date,
-        snapshot_hash=world.get(entity, SourceC).snapshot_hash,
-        headline=draft.headline,
-        recommended_action=draft.recommended_action,
-        kpis=_kpis(ctx, df, period, prev, items),
-        findings=findings,
-        risks_caveats=risks,
-        next_checks=next_checks,
-        metric_definitions=[
-            MetricDefinition(
-                name=m.name,
-                definition=m.definition,
-                owner=m.owner,
-                includes=m.inclusions,
-                excludes=m.exclusions,
-                version=str(m.version),
-                checksum=ctx.metrics.checksum,
-            )
-            for m in ctx.metrics.metrics
-        ],
-        data_tables=tables,
-    )
-
-
-def _kpis(ctx: RunContext, df: pl.DataFrame, period: str, prev: str, items: list[Item]) -> list[Kpi]:
-    kpis = []
-    by_name = metric_map(ctx)
-    for name in dict.fromkeys(item.metric for item in items):
-        metric = by_name[name]
-        now = compute(df, metric, None, period)
-        before = compute(df, metric, None, prev)
-        if now is None or before is None:
-            continue
-        kpis.append(
-            Kpi(
-                label=metric.name,
-                value=ShownNumber(level_text(now), metric.name, None, period),
-                vs_prior=ShownNumber(change_text(now - before), metric.name, None, period),
-            )
-        )
-    return kpis
+"""Report model: the input the fixed PDF template renders, built from the draft and the run's components (spec 6.8)."""
+
+import polars as pl
+
+from weekly_analyst.analysis.draft import DraftFields
+from weekly_analyst.analysis.plan import _item_key as plan_item_key
+from weekly_analyst.core.components import EvidenceC, SourceC, ValidationC
+from weekly_analyst.core.context import RunContext
+from weekly_analyst.core.entity import Entity
+from weekly_analyst.core.items import Item
+from weekly_analyst.core.world import World
+from weekly_analyst.data.metrics_calc import compute
+from weekly_analyst.report.model import DataTable, Finding, Kpi, MetricDefinition, ReportModel, ShownNumber
+from weekly_analyst.systems.common import frame, metric_map, prev_period, public_item, raw_items
+from weekly_analyst.systems.report_findings import change_text, finding_for, level_text
+
+
+def build_report(world: World, entity: Entity, ctx: RunContext, draft: DraftFields) -> ReportModel:
+    df = frame(world, entity, ctx)
+    validation = world.get(entity, ValidationC)
+    period = validation.period
+    prev = prev_period(period)
+    items = raw_items(world, entity)
+    texts = {f.item_key: f for f in draft.findings}
+    findings: list[Finding] = []
+    tables: list[DataTable] = []
+    for item in items:
+        found, table = finding_for(ctx, df, item, texts[plan_item_key(public_item(item))])
+        findings.append(found)
+        tables.append(table)
+    snapshot_date = df[ctx.metrics.date_column].max().isoformat()
+    risks = list(draft.caveats) + [
+        f"{f['kind'].replace('_', ' ')}: {f['detail']}" for f in validation.findings
+    ]
+    failed = [r for r in world.get(entity, EvidenceC).results if "error" in r]
+    if failed:
+        risks.append(f"{len(failed)} planned follow-up check(s) could not be run and are not reflected in this report")
+    if validation.cold_start:
+        risks.append(
+            f"insufficient history: fewer than {ctx.settings.cold_start_periods} prior weeks, "
+            "so no change alerts were sent"
+        )
+    next_checks = [f"Check the source rows behind {plan_item_key(public_item(item))}" for item in items]
+    if any(f["kind"] == "schema_change" for f in validation.findings):
+        next_checks.append("Confirm the schema change with the owner of the source file")
+    return ReportModel(
+        report_name="Weekly analysis",
+        reporting_period=period,
+        prior_period=prev,
+        data_as_of=snapshot_date,
+        prepared_for=", ".join(ctx.recipients) or "Recipients not set",
+        report_version="1",
+        run_id=entity.run_id,
+        snapshot_date=snapshot_date,
+        snapshot_hash=world.get(entity, SourceC).snapshot_hash,
+        headline=draft.headline,
+        recommended_action=draft.recommended_action,
+        kpis=_kpis(ctx, df, period, prev, items),
+        findings=findings,
+        risks_caveats=risks,
+        next_checks=next_checks,
+        metric_definitions=[
+            MetricDefinition(
+                name=m.name,
+                definition=m.definition,
+                owner=m.owner,
+                includes=m.inclusions,
+                excludes=m.exclusions,
+                version=str(m.version),
+                checksum=ctx.metrics.checksum,
+            )
+            for m in ctx.metrics.metrics
+        ],
+        data_tables=tables,
+    )
+
+
+def _kpis(ctx: RunContext, df: pl.DataFrame, period: str, prev: str, items: list[Item]) -> list[Kpi]:
+    kpis = []
+    by_name = metric_map(ctx)
+    for name in dict.fromkeys(item.metric for item in items):
+        metric = by_name[name]
+        now = compute(df, metric, None, period, ctx.metrics.date_column)
+        before = compute(df, metric, None, prev, ctx.metrics.date_column)
+        if now is None or before is None:
+            continue
+        kpis.append(
+            Kpi(
+                label=metric.name,
+                value=ShownNumber(level_text(now), metric.name, None, period),
+                vs_prior=ShownNumber(change_text(now - before), metric.name, None, period),
+            )
+        )
+    return kpis