Replay Engine — Internals¶

Overview¶

The APS Replay Engine iteratively re-runs the supply + allocation planning pipeline over a configurable historical horizon. This page documents the internal architecture, data model, code structure, and integration points for developers extending or debugging the replay system.

Architecture¶

flowchart TD
    API["/api/replay/runs/{id}/start"]
    BG["Background Thread"]
    ENGINE["ReplayEngine\nfiles/replay/replay_engine.py"]
    PD["planning_today()\nfiles/utils/planning_date.py"]
    SUPPLY["Supply Subprocess\nrun_pipeline.py --only supply"]
    ALLOC["Allocation Runner\nallocation_runner.py"]
    PG["PostgreSQL\nreplay_run + replay_iteration"]
    CH["ClickHouse\nPIPE_supply_* + PIPE_allocation_*"]
    OH["master.on_hand\n(backed up before replay)"]

    API --> BG --> ENGINE
    ENGINE --> PD
    ENGINE -->|"receive past orders"| OH
    ENGINE -->|"REPLAY_PLANNING_DATE env"| SUPPLY
    ENGINE -->|"REPLAY_PLANNING_DATE env"| ALLOC
    SUPPLY --> CH
    ALLOC --> CH
    ENGINE --> PG

Planning Date Override¶

The central mechanism: REPLAY_PLANNING_DATE environment variable.

`utils/planning_date.py`¶

def planning_today(pipeline_id=None) -> date:
    env = os.environ.get("REPLAY_PLANNING_DATE", "").strip()
    if env:
        return date.fromisoformat(env)
    # ... DB lookup chain omitted ...
    return datetime.now(timezone.utc).date()

When set, all downstream runners see the replay iteration's planning date as "today" instead of the real calendar date. This is a process-level override — the supply subprocess inherits the env var from its parent.

Patched Call Sites¶

File	Original	Replacement
`allocation_runner.py:_today_epoch()`	`date.today()`	`planning_today()`
`allocation_runner.py:load_demands()`	`date.today()`	`planning_today()`
`supply_runner.py:run_supply_plan()` (supersession filter)	`_date_cls_ss.today()`	`planning_today()`
`supply_runner.py:run_supply_plan()` (horizon start)	`_date_cls.today()`	`planning_today()`
`supply_runner.py:_load_sku_plans()` (firm orders)	`_date.today()`	`planning_today()`

All patches are local imports (from utils.planning_date import planning_today) — no module-level side effects when the env var is unset.

Database Schema¶

`replay_run` (PostgreSQL, `{schema}` schema)¶

Column	Type	Description
`id`	`BIGSERIAL PK`	Run identifier
`pipeline_id`	`BIGINT NOT NULL`	The replay pipeline (cloned from source)
`name`	`TEXT NOT NULL`	User-provided name
`description`	`TEXT`	Optional description
`source_pipeline_id`	`BIGINT NOT NULL`	The pipeline being replayed
`start_date`	`DATE NOT NULL`	First iteration planning date
`end_date`	`DATE NOT NULL`	Last iteration planning date
`current_date`	`DATE`	Currently executing iteration date
`granularity`	`VARCHAR(10)`	`'daily'` or `'weekly'` (default)
`status`	`VARCHAR(20)`	`pending`, `running`, `completed`, `failed`, `cancelled`
`iteration_count`	`INTEGER`	Total iterations (computed from date range)
`current_iteration`	`INTEGER`	Current iteration number
`config`	`JSONB`	Replay parameters (return_rate overrides, etc.)
`created_at`	`TIMESTAMPTZ`	Creation timestamp
`updated_at`	`TIMESTAMPTZ`	Last update timestamp
`completed_at`	`TIMESTAMPTZ`	Completion timestamp
`error_msg`	`TEXT`	Error message if failed

`replay_iteration` (PostgreSQL)¶

Column	Type	Description
`id`	`BIGSERIAL PK`	Iteration identifier
`replay_run_id`	`BIGINT NOT NULL FK`	→ `replay_run.id` (CASCADE)
`iteration_num`	`INTEGER NOT NULL`	1-based iteration index
`planning_date`	`DATE NOT NULL`	The date this iteration simulates
`status`	`VARCHAR(20)`	`pending`, `running`, `completed`, `failed`, `skipped`
`archive_date`	`DATE`	Written to CH as `archive_date` (= `planning_date`)
`supply_run_id`	`BIGINT`	FK to `pipe_supply_run` (if applicable)
`allocation_run_id`	`BIGINT`	FK to `allocation_run` (if applicable)
`started_at`	`TIMESTAMPTZ`	Iteration start timestamp
`completed_at`	`TIMESTAMPTZ`	Iteration end timestamp
`error_msg`	`TEXT`	Error details if failed
`metrics`	`JSONB`	`{fill_rate, total_stock, total_shortage, ...}`

Unique constraint: (replay_run_id, iteration_num)

`config_scenario_pipeline.is_replay`¶

A boolean column (DEFAULT FALSE) added to the pipeline registry. Replay pipelines are marked is_replay = TRUE so they can be filtered from normal pipeline lists and cleaned up on deletion.

`allocation_run.archive_date`¶

A DATE column added to allocation_run in PostgreSQL. For standard runs, it defaults to planning_today(). For replay iterations, it is set to the iteration's planning_date.

ClickHouse Schema Changes¶

`archive_date` added to allocation tables¶

Three PIPE_allocation_* tables now include an archive_date Date column:

Table	Column added	Purpose
`PIPE_allocation_results`	`archive_date Date`	Time-series queries across iterations
`PIPE_allocation_unfulfilled`	`archive_date Date`	Fill rate over time
`PIPE_allocation_explanations`	`archive_date Date`	Audit trail per iteration

The migration DDL is in files/DDL/allocation_ch_archive_date.sql.

For replay, each iteration writes with archive_date = planning_date. For standard runs, archive_date = planning_today(). This means the standard process also benefits from historization of allocation results.

ReplayEngine Class¶

`files/replay/replay_engine.py`¶

class ReplayEngine:
    def __init__(self, replay_run_id: int): ...
    def run(self, account_cfg: dict | None = None) -> dict: ...
    def cancel(self) -> None: ...

Iteration Loop¶

1. _load_run()         → load replay_run row from PG
2. _generate_iterations() → [start_date, start+step, ..., end_date]
3. _backup_on_hand()   → CREATE TABLE master._replay_oh_backup_{id} AS SELECT * FROM master.on_hand
4. For each planning_date:
   a. _check_cancelled()   → poll PG for status='cancelled'
   b. _receive_supply_orders() → close orders with arrival ≤ planning_date, add qty to on_hand
   c. _run_supply_subprocess() → subprocess: run_pipeline.py --only supply
   d. _run_allocation()     → call allocation_runner.run_allocation_pipeline()
   e. _compute_iteration_metrics() → query CH for fill_rate, stock, shortage
   f. _upsert_iteration()   → write metrics to replay_iteration
5. _update_run_status('completed')
6. Drop on_hand backup table

Stock Evolution: `_receive_supply_orders()`¶

Before each iteration, supply orders from previous iterations with arrival_date ≤ planning_date are "received":

Query master_supply_orders_firm for orders that have arrived
Add their quantity to master.on_hand
This mutated on_hand becomes the input to the next supply+allocation run

On-Hand Backup and Restore¶

Before replay: _backup_on_hand() creates master._replay_oh_backup_{run_id}
On success: the backup table is dropped
On failure/cancel: _restore_on_hand() truncates master.on_hand and re-inserts from the backup

Crash safety

If the API process crashes (kill -9, OOM) during a replay, the on_hand backup table remains in the master schema but the restore is not automatic. A manual cleanup script should check for orphaned _replay_oh_backup_* tables and restore from them if needed.

API Endpoints¶

All endpoints are under /api/replay and mounted from files/api/replay_router.py.

Method	Path	Description
`POST`	`/runs`	Create a new replay run + replay pipeline
`GET`	`/runs`	List all replay runs (with source pipeline names)
`GET`	`/runs/{id}`	Get run details + progress
`POST`	`/runs/{id}/start`	Start the replay (spawns background thread)
`POST`	`/runs/{id}/cancel`	Cancel a running replay
`DELETE`	`/runs/{id}`	Delete run + iterations + replay pipeline
`GET`	`/runs/{id}/iterations`	List iterations with metrics
`GET`	`/runs/{id}/stock-evolution`	Stock over time (from CH)
`GET`	`/runs/{id}/fill-rate-evolution`	Fill rate + shortfall over time (from CH)
`GET`	`/runs/{id}/supply-orders`	Supply orders across iterations
`GET`	`/runs/{id}/allocation-results`	Filled + unfilled orders across iterations

Background Execution¶

When /runs/{id}/start is called:

The run status is set to running in PG
A daemon thread is spawned: _run_replay_background(run_id, account_cfg)
The thread creates a ReplayEngine instance and calls .run()
Active replays are tracked in _active_replays: dict[int, dict]
The frontend polls /runs/{id} every 5 seconds for progress

Concurrency Guard¶

Only one replay can run per source pipeline. The API returns 409 if you try to start a second replay while one is already running.

Frontend: ReplayDashboard.jsx¶

Location¶

files/frontend/src/components/ReplayDashboard.jsx

Sections¶

Section	Component	Data Source
Replay runs list	HTML table	`GET /api/replay/runs`
Progress bar	Custom (div-based)	`current_iteration / iteration_count` from run row
Stock Over Time	Plotly scatter	`GET /api/replay/runs/{id}/stock-evolution`
Fill Rate Over Time	Plotly scatter + bar	`GET /api/replay/runs/{id}/fill-rate-evolution`
Iteration details	HTML table	`GET /api/replay/runs/{id}/iterations`

Auto-Polling¶

When any replay run has status = 'running', the dashboard polls every 5 seconds:

useEffect(() => {
  const activeRun = runs.find(r => r.status === 'running');
  if (activeRun) {
    pollRef.current = setInterval(() => {
      loadRuns();
      loadIterations(activeRun.id);
      loadChartData(activeRun.id);
    }, 5000);
  }
}, [runs]);

When no runs are active, polling stops and a final data refresh is performed.

DDL Files¶

File	Purpose
`files/DDL/replay_schema.sql`	Standalone DDL for `replay_run` + `replay_iteration` + `is_replay` + `allocation_run.archive_date`
`files/DDL/allocation_ch_archive_date.sql`	ClickHouse migration: add `archive_date` to `PIPE_allocation_*` tables

Both are also included in init_schema() in files/db/db.py so they are applied automatically when the API starts.

Return Logic¶

Returns are handled by the existing supply engine mechanism — no special replay code is needed:

master_route rows with source_type = 'Return' define return flows
The return_rate column on routes specifies the proportion of demand that returns
The Rust supply engine generates RETURN orders based on these rates
During replay, planning_today() ensures return lead times are computed relative to the iteration's planning date

Key Files¶

File	Role
`files/replay/replay_engine.py`	Core iteration loop, stock evolution, subprocess management
`files/utils/planning_date.py`	`planning_today()` — env-var-based date override
`files/api/replay_router.py`	FastAPI router (11 endpoints)
`files/frontend/src/components/ReplayDashboard.jsx`	React dashboard
`files/DDL/replay_schema.sql`	PG DDL for replay tables
`files/DDL/allocation_ch_archive_date.sql`	CH DDL for archive_date
`files/db/db.py`	`init_schema()` — creates replay tables on startup
`files/allocation_runner.py`	Patched: `planning_today()` + `archive_date`
`files/supply_runner.py`	Patched: `planning_today()` at 3 call sites