Historization — Archive Date Model¶

Overview¶

ForecastAI historizes pipeline snapshots using archive_date — always the Monday of the calendar week when the snapshot was created. This replaces the former opaque pipeline_version_id integer.

Figure: planning_date resolution chain — replay env always wins, then the per-pipeline row, then the global pipeline_id=-1 row, finally real UTC today; planning_monday() derives the archive_date.

flowchart TD
    A["REPLAY_PLANNING_DATE env"] -->|"set?"| B["Use replay date"]
    A -->|"unset"| C["scenario.planning_date<br/>WHERE pipeline_id = ?"]
    C --> D{"row for this pipeline_id?"}
    D -->|"yes"| E["Use that date"]
    D -->|"no"| F{"row for pipeline_id = -1<br/>(global)?"}
    F -->|"yes"| G["Use global date"]
    F -->|"no"| H["datetime.now(UTC).date()"]
    B --> I["planning_today(pipeline_id)"]
    E --> I
    G --> I
    H --> I
    I --> J["planning_monday(pipeline_id)<br/>= Monday of that week"]
    J --> K["archive_date"]

Every frozen snapshot is identified by (pipeline_id, archive_date). If the same pipeline is frozen twice in the same week, the second freeze replaces the previous one (unless the previous one is pinned).

Key Concepts¶

Figure: how archive_date is derived from the effective planning date — the live partition always uses planning_monday(), frozen snapshots tag an immutable partition with the same Monday.

flowchart LR
    A["planning_today(pipeline_id)"] --> B["planning_monday(pipeline_id)"]
    B --> C["archive_date (Monday of freeze week)"]
    C --> D["Live partition<br/>archive_date = planning_monday()"]
    C --> E["Frozen snapshot<br/>immutable tagged partition"]

Concept	Description
`archive_date`	ISO date of the Monday of the week the freeze ran (e.g. `2026-06-02`)
Live partition	The working state: `archive_date = planning_monday()`. Overwritten on every pipeline run.
Frozen snapshot	An immutable copy of the live partition tagged with the week's Monday date.
Pinned snapshot	Exempt from auto-replacement and retention cleanup. Multiple pinned snapshots can coexist for the same `archive_date`.
Dedup rule	One un-pinned frozen snapshot per `(pipeline_id, archive_date)`. Re-freezing the same week replaces the prior un-pinned one.

Data Model¶

PostgreSQL — `master.pipeline_version_index`¶

id                   BIGSERIAL PRIMARY KEY   -- internal surrogate
pipeline_id          BIGINT NOT NULL
scenario_id          BIGINT
archive_date         DATE NOT NULL           -- Monday of freeze week
snapshot_ts          TIMESTAMPTZ             -- exact freeze timestamp
created_by           TEXT
comment              TEXT
frozen               BOOLEAN DEFAULT FALSE
pinned               BOOLEAN DEFAULT FALSE
parent_archive_date  DATE                    -- lineage FK
replaced_archive_date DATE                   -- which previous archive was replaced
row_count_total      BIGINT
table_counts         JSONB                   -- per-table row counts
kpi_*                DOUBLE PRECISION        -- pre-computed KPI headline values

Unique index: (pipeline_id, archive_date) WHERE frozen = TRUE AND pinned = FALSE
Pinned rows for the same date coexist alongside the current un-pinned snapshot.

ClickHouse — all `PIPE_*` tables¶

Every PIPE_* table has:

archive_date  Date  DEFAULT toMonday(today())

Partition keys: - Arity-3 tables: PARTITION BY (pipeline_id, scenario_id, archive_date) - Arity-2 tables: PARTITION BY (pipeline_id, archive_date) - PIPE_on_hand_snapshot: PARTITION BY archive_date

Live rows use archive_date = planning_monday(). They are overwritten on each pipeline run via DROP PARTITION + INSERT.

Frozen rows use the archive_date of their freeze week. They are immutable partitions that are never overwritten by regular runs.

Freeze Process¶

Figure: weekly freeze/archive timeline — the live partition is overwritten every run; one immutable frozen snapshot per week (replaced unless pinned).

gantt
    title Archive date timeline (one freeze per ISO week)
    dateFormat YYYY-MM-DD
    axisFormat %m-%d
    section Live partition
    Live week 23 :active, l23, 2026-06-02, 7d
    Live week 24 :active, l24, 2026-06-09, 7d
    Live week 25 :active, l25, 2026-06-16, 7d
    section Frozen snapshots
    Freeze week 23 :f23, 2026-06-02, 1d
    Freeze week 24 (pinned) :crit, f24, 2026-06-09, 1d
    Freeze week 25 :f25, 2026-06-16, 1d

POST /api/history/freeze
{
    "pipeline_id": 4,
    "comment":     "Week 42 baseline",
    "pin":         false
}

Steps executed by historization/freeze.py:

Compute archive_date = planning_monday(pipeline_id)
Check for an existing un-pinned frozen snapshot for (pipeline_id, archive_date)
If found: drop its CH partitions and delete its registry row
Insert a new frozen=FALSE registry row in master.pipeline_version_index

For each PIPE_* table: copy all live rows (archive_date = planning_monday(pipeline_id)) into a new partition via:

INSERT INTO PIPE_xxx
SELECT * REPLACE (toDate('YYYY-MM-DD') AS archive_date)
FROM PIPE_xxx
WHERE pipeline_id = X AND archive_date = toMonday(today())

Snapshot master.on_hand → PIPE_on_hand_snapshot tagged with archive_date
Set frozen = TRUE on the registry row
Compute and store headline KPIs (fill_rate, inventory_value, shortage_qty, etc.)
Invalidate the latest_archive_date() cache

Returns:

{
    "archive_date":           "2026-06-02",
    "snapshot_ts":            "2026-06-07T10:00:00Z",
    "replaced_archive_date":  null,
    "table_counts":           {"PIPE_forecast_point_values": 12500, ...},
    "kpis":                   {"kpi_fill_rate": 0.94, ...},
    "comment":                "Week 42 baseline",
    "pinned":                 false
}

API Endpoints¶

All history endpoints are in api/history_router.py.

Freeze¶

POST /api/history/freeze
Body: { pipeline_id, comment?, pin? }

List archives¶

GET /api/history/versions?pipeline_id=4&frozen_only=true
GET /api/history/versions?pipeline_id=4&from=2026-01-01&to=2026-06-30

Get single archive¶

GET /api/history/versions/2026-06-02?pipeline_id=4

Update comment / pin¶

PATCH /api/history/versions/2026-06-02?pipeline_id=4
Body: { comment?: "new comment", pinned?: true }

Delete (un-pinned only)¶

DELETE /api/history/versions/2026-06-02?pipeline_id=4

Compare two archives¶

GET /api/compare/versions?base=2026-05-26&target=2026-06-02&dimension=kpi
GET /api/compare/versions?base=2026-05-26&target=2026-06-02&dimension=forecast
GET /api/compare/versions?base=2026-05-26&target=2026-06-02&dimension=order

Supported dimensions: kpi, forecast, order, inventory, meio_buffer, exceptions, best_method

Plan vs Actual¶

GET /api/compare/plan-vs-actual?pipeline_id=4&archive_date=2026-06-02&dimension=demand

Timeline¶

GET /api/history/timeline?pipeline_id=4&from=2026-01-01

Replay (read archived data)¶

GET /api/history/replay/2026-06-02?pipeline_id=4&view=summary
GET /api/history/replay/2026-06-02?pipeline_id=4&view=series&unique_id=5_301
GET /api/history/replay/2026-06-02?pipeline_id=4&view=supply
GET /api/history/replay/2026-06-02?pipeline_id=4&view=meio

Python API¶

`db.db.latest_archive_date(pipeline_id, scenario_id=None) -> date | None`¶

Returns the latest frozen archive_date for a pipeline. Returns None if no frozen snapshot exists (callers should then use the live partition without an archive_date filter).

from db.db import latest_archive_date

ad = latest_archive_date(pipeline_id=4)
if ad is not None:
    # query frozen snapshot: archive_date = ad
else:
    # query live: archive_date = toMonday(today())

`_archive_date_or_live(pipeline_id, scenario_id=None) -> date`¶

Helper defined in api/main.py — always returns a date. Falls back to toMonday(today()) when no frozen snapshot exists.

`historization.freeze.freeze_pipeline(pipeline_id, ...)`¶

Main entry point. Returns a dict with archive_date, snapshot_ts, kpis, etc.

`historization.freeze.compute_archive_date(ts=None) -> date`¶

Returns toMonday(ts) — the archive_date for any given timestamp.

ClickHouse Query Patterns¶

Query the live partition (current week)¶

SELECT * FROM PIPE_forecast_point_values
WHERE pipeline_id = 4 AND archive_date = toMonday(today())

Query a specific frozen archive¶

SELECT * FROM PIPE_forecast_point_values
WHERE pipeline_id = 4 AND archive_date = toDate('2026-06-02')

Drop the live partition before a new run¶

ALTER TABLE PIPE_forecast_point_values
DROP PARTITION (4, 1, toDate('2026-06-02'))

Compare two archives¶

SELECT
    t.item_id, t.site_id,
    b.point_forecast AS base_forecast,
    t.point_forecast AS target_forecast,
    t.point_forecast - b.point_forecast AS delta
FROM PIPE_forecast_point_values t
FULL OUTER JOIN PIPE_forecast_point_values b
    ON b.item_id = t.item_id AND b.site_id = t.site_id AND b.method = t.method
    AND b.forecast_week = t.forecast_week
    AND b.pipeline_id = t.pipeline_id
    AND b.archive_date = toDate('2026-05-26')
WHERE t.pipeline_id = 4
  AND t.archive_date = toDate('2026-06-02')
  AND t.method = t.best_method

Retention¶

Retention is governed by config_parameters (type history):

Parameter	Default	Description
`retention_versions`	52	Keep last N frozen archives per pipeline
`retention_min_age_days`	0	Don't delete archives younger than N days

Pinned archives are exempt from deletion.

Run manually:

python -m historization.retention --pipeline-id 4 --dry-run
python -m historization.retention

Migration History¶

Date	Migration	Description
2026-05-15	`ch_migration_v2.sql`	Added `pipeline_version_id Int64 DEFAULT 0` to all PIPE_* tables
2026-06-07	`ch_migrate_archive_date.py`	Replaced `pipeline_version_id` with `archive_date Date` in all PIPE_* tables
2026-06-07	`pg_migrate_archive_date.sql`	Migrated `master.pipeline_version_index` to use `archive_date DATE` as primary key

Re-running the migration on a new tenant¶

Run files/DDL/ch_migrate_archive_date.py --db <tenant_name>
Run files/DDL/pg_migrate_archive_date.sql against the PostgreSQL database

Pinning¶

Pinned archives are never auto-replaced and never deleted by retention. Use them to mark significant baselines:

PATCH /api/history/versions/2026-06-02?pipeline_id=4
Body: { "pinned": true, "comment": "Q2 2026 official plan" }

A pinned archive and a current (un-pinned) archive can coexist for the same archive_date. The un-pinned one will be replaced on next freeze; the pinned one persists indefinitely.

Historization — Archive Date Model¶

Overview¶

Key Concepts¶

Data Model¶

PostgreSQL — master.pipeline_version_index¶

ClickHouse — all PIPE_* tables¶

Freeze Process¶

API Endpoints¶

Freeze¶

List archives¶

Get single archive¶

Update comment / pin¶

Delete (un-pinned only)¶

Compare two archives¶

Plan vs Actual¶

Timeline¶

Replay (read archived data)¶

Python API¶

db.db.latest_archive_date(pipeline_id, scenario_id=None) -> date | None¶

_archive_date_or_live(pipeline_id, scenario_id=None) -> date¶

historization.freeze.freeze_pipeline(pipeline_id, ...)¶

historization.freeze.compute_archive_date(ts=None) -> date¶

ClickHouse Query Patterns¶

Query the live partition (current week)¶

Query a specific frozen archive¶

Drop the live partition before a new run¶

Compare two archives¶

Retention¶

Migration History¶

Re-running the migration on a new tenant¶

Pinning¶

PostgreSQL — `master.pipeline_version_index`¶

ClickHouse — all `PIPE_*` tables¶

`db.db.latest_archive_date(pipeline_id, scenario_id=None) -> date | None`¶

`_archive_date_or_live(pipeline_id, scenario_id=None) -> date`¶

`historization.freeze.freeze_pipeline(pipeline_id, ...)`¶

`historization.freeze.compute_archive_date(ts=None) -> date`¶