Skip to content

Allocation Engine Internals

The allocation engine is a Rust-based, multi-pass deterministic algorithm that matches a finite supply pool to competing demands, weighted by priority, aging, lateness, strategic attributes, and corrected by routing / fairness / reservation policies. It runs as a native extension via PyO3, achieving sub-second performance on production-scale datasets.

Architecture Overview

The engine spans two language runtimes connected by a PyO3 bridge:

┌─────────────────────────────────────────────┐
│  Python (allocation_runner.py)              │
│                                             │
│  ┌─────────────┐   ┌──────────────────────┐ │
│  │ Demand load  │   │ Supply load          │ │
│  │ CH + PG      │   │ PG on_hand + CH SO   │ │
│  └──────┬──────┘   └──────────┬───────────┘ │
│         │                     │              │
│         └──────────┬──────────┘              │
│                    ▼                          │
│        ┌───────────────────────┐              │
│        │ AllocationParameter   │              │
│        │ Resolver              │              │
│        └───────────┬───────────┘              │
│                    │ JSON strings             │
│                    ▼                          │
│  ┌─────────────────────────────────────────┐ │
│  │  supply_engine.run_allocation()  (PyO3) │ │
│  └────────────────────┬────────────────────┘ │
└───────────────────────┼──────────────────────┘
                        │ py.allow_threads()
                        ▼
┌───────────────────────────────────────────────┐
│  Rust (score_alloc module)                    │
│                                               │
│  AllocationEngine::run()                      │
│    ├─ 8-pass algorithm (see below)            │
│    └─ Returns AllocationResult (JSON)         │
└───────────────────────┬───────────────────────┘
                        │
                        ▼
┌───────────────────────────────────────────────┐
│  Python — write_results_ch()                  │
│    ├─ PIPE_allocation_results                 │
│    ├─ PIPE_allocation_unfulfilled             │
│    └─ PIPE_allocation_explanations            │
└───────────────────────────────────────────────┘

PyO3 entry point

Figure: pipeline_id guard flow — the Rust entry asserts pipeline_id > 0 before releasing the GIL; results are then written to partitions keyed by pipeline_id (0 is never a valid partition).

flowchart TD
    A["Python: run_allocation(demands, supplies, config)"] --> B["Deserialize AllocationConfig"]
    B --> C{"pipeline_id > 0?"}
    C -->|"no"| D["Rust assert! abort<br/>(0 = orphan sentinel)"]
    C -->|"yes"| E["Release GIL<br/>py.allow_threads()"]
    E --> F["8-pass engine run"]
    F --> G["write_results_ch()"]
    G --> H["drop_ch_partition(pipeline_id)<br/>then bulk insert"]

The Rust crate exposes a single #[pyfunction] in files/supply/src/lib.rs:

#[pyfunction]
#[pyo3(signature = (demands_json, supplies_json, config_json))]
pub fn run_allocation(
    py: Python<'_>,
    demands_json: &str,
    supplies_json: &str,
    config_json: &str,
) -> PyResult<String> { ... }

All three inputs and the output are JSON strings, deserialized/serialized via serde_json. The GIL is released with py.allow_threads() before entering the engine to avoid deadlock with Rayon worker threads.

Module layout

files/supply/src/score_alloc/
├── mod.rs            # Public re-exports; entry point docs
├── config.rs         # AllocationConfig, PriorityScores, StarvationConfig
├── models.rs         # Demand, AllocationSupply, AllocationCandidate,
│                     # ScoreBreakdown, ScoredCandidate, Allocation,
│                     # UnfulfilledDemand, AllocationExplanation,
│                     # AllocationResult, AllocationMetrics
├── engine.rs         # AllocationEngine — 8-pass orchestrator
├── scoring.rs        # ScoreEngine — composite score + Rayon parallel scoring
├── starvation.rs     # StarvationEngine — anti-starvation bonus & escalation
└── policy/
    ├── mod.rs
    ├── aging.rs      # AgingPolicy (Linear / Exponential / None)
    ├── fairness.rs   # FairnessTracker, FairnessConfig
    ├── lateness.rs    # LatenessCurve (piecewise-linear)
    ├── reservation.rs # ReservationEngine, ReservationConfig, Fence
    └── routing.rs     # RoutingPolicy (hard/soft constraints)

Data Flow

Python side — demand loading

allocation_runner.load_demands() builds the demand vector from three sources:

Source Database Fields provided
PIPE_forecast_netting ClickHouse item_id, site_id, forecast_week, supply_plan_qty (= unconsumed forecast + firm demand)
master.demand_actuals (is_forecast=FALSE) PostgreSQL customer_id, priority_class per (item, site, week)
master.item PostgreSQL Fallback priority_class, strategic_weight, revenue_weight, criticality_weight

Resolution order for each demand record's priority: firm-order metadata → item-level default → "standard".

Firm-order priority matching (fixed)

Firm-order metadata (customer, priority class) is matched to a demand by week-index — (firm_order.date - planning_today).days // 7 — keyed against the demand's W{n} suffix. This is convention-independent: a tenant whose forecast weeks start on Sunday would break the older Monday-snap matching (date - date.weekday()), silently dropping every firm-order priority back to standard. The week-index match works for both Sunday- and Monday-start weeks.

Demand IDs follow the template D_{item_id}_{site_id}_W{week_index} for stable identification across runs.

Python side — supply loading

allocation_runner.load_supplies() combines two sources:

Source Database Notes
master.on_hand (planning_type IN ('good','new')) PostgreSQL Aggregated per (item, site, type); available_date = today
PIPE_supply_orders (status IN ('planned','firm','released','in_transit')) ClickHouse available_date = today + 7 × arrival_week; source_type mapped from order_type

Python side — parameter resolution

AllocationParameterResolver assembles the full AllocationConfig JSON. Seven parameter types are stored in {schema}.allocation_parameter_set:

Type Scope Config field
allocation_priority Global priority_scores, strategic_weight_factor, revenue_weight_factor, criticality_weight_factor
allocation_starvation Global starvation
allocation_reservation Global reservation
allocation_fairness Global fairness
allocation_routing Policy (many per scenario) routing_policies
allocation_aging Policy (many per scenario) aging_policies
allocation_lateness Policy (many per scenario) lateness_curves

Resolution order (highest priority first):

  1. Segment-specific parameter set (linked via allocation_parameter_set_segment)
  2. Scenario default (is_default = TRUE)

Global types return a single parameter set; policy types return a list that is keyed into a HashMap<String, Policy> by policy_id / curve_id.

Rust side — engine execution

The Rust AllocationEngine::run() receives the deserialized Vec<Demand>, Vec<AllocationSupply>, and AllocationConfig, runs the 8-pass algorithm, and returns AllocationResult containing allocations, unfulfilled demands, explanations, and aggregate metrics.

Python side — result persistence

write_results_ch() drops the pipeline partition from all three tables (atomic replace pattern), then bulk-inserts:

  • PIPE_allocation_results — one row per allocation
  • PIPE_allocation_unfulfilled — one row per demand with remaining shortfall
  • PIPE_allocation_explanations — one row per narrative fragment (step-based)

The 8-Pass Algorithm

Figure: the 8-pass pipeline — backlog enrichment, fairness seeding, reservation fences, candidate generation, hard filter, parallel scoring, sort, greedy allocation, then persistence to three PIPE_allocation_ tables.*

flowchart TD
    A["Demands + Supplies + Config"] --> P1["Pass 1: enrich backlog_days"]
    P1 --> P2["Pass 2: seed fairness groups"]
    P2 --> P3["Pass 3: build reservation fences"]
    P3 --> P4["Pass 4: generate candidates<br/>(item + location + date gates)"]
    P4 --> P5["Pass 5: filter hard routing violations"]
    P5 --> P6["Pass 6: score candidates (Rayon parallel)"]
    P6 --> P7["Pass 7: sort descending by total_score"]
    P7 --> P8["Pass 8: greedy allocation<br/>(reservation fences reduce supply)"]
    P8 --> R1["PIPE_allocation_results"]
    P8 --> R2["PIPE_allocation_unfulfilled"]
    P8 --> R3["PIPE_allocation_explanations"]

The engine is implemented in engine.rs:AllocationEngine::run(). Each pass is sequential and deterministic.

Pass 1 — Enrich backlog

for d in &mut demands {
    d.backlog_days = (self.config.current_date - d.required_date).max(0) as i32;
}

Adds backlog_days to every Demand, computed as the difference between current_date and required_date. This field drives aging and starvation scoring in later passes.

Pass 2 — Seed fairness

let mut fairness = FairnessTracker::new(self.config.fairness.clone());
for d in &demands {
    if let Some(g) = group_for_demand(d, &self.config.fairness) {
        fairness.register_demand(&g, d.qty);
    }
}

Initializes the FairnessTracker with total demand quantities per fairness group. The group field is configured via FairnessConfig.group_field (Channel, CustomerClass, Program, or Region). Tracking demand totals at seed time allows computing target shares before any allocation occurs.

Pass 3 — Build reservations

let mut reservation = ReservationEngine::new(self.config.reservation.clone());
reservation.build_fences(&demands, &supplies, self.config.current_date);

The ReservationEngine scans future demands within lookahead_days that are at or above min_protected_priority. For each, it creates a Fence on matching supply (same item_id + location_id), reserving qty × (1 + safety_margin). Reserved quantities are subtracted from the available supply pool before greedy allocation.

Pass 4 — Generate candidates

For each demand, the engine finds all compatible supplies by:

  1. Item gate — look up supply_by_item index for matching item_id
  2. Location gate — only pair demand with supply at the same location_id (prevents cartesian explosion)
  3. Date gate — supply.available_date - demand.required_date <= max_delay_days

For each surviving (demand, supply) pair, the RoutingPolicy (if assigned) is evaluated to determine feasibility and collect hard/soft violations.

if supply.location_id != demand.location_id { continue; }
let delay = supply.available_date - demand.required_date;
if delay > max_delay { continue; }

let (feasible, hard, soft) = if let Some(policy) = routing_policy {
    let r: RoutingCheckResult = policy.check(supply);
    (r.feasible, r.hard_violations, r.soft_violations)
} else {
    (true, Vec::new(), Vec::new())
};

Candidates are sorted deterministically by (demand_id, supply_id) to ensure repeatable results.

Pass 5 — Filter hard constraints

let (feasible, infeasible): (Vec<_>, Vec<_>) =
    all_candidates.into_iter().partition(|c| c.feasible);

Candidates with any hard routing violation are removed. These are tracked in infeasible_demand_ids to provide accurate "why unfulfilled?" reasons later.

Pass 6 — Score candidates (parallel)

The ScoreEngine computes a composite ScoreBreakdown for every feasible candidate. Scoring is parallelized via Rayon's par_iter():

candidates
    .par_iter()
    .filter_map(|c| {
        let d = demand_map.get(&c.demand_id)?;
        let s = supply_map.get(&c.supply_id)?;
        Some(self.score_one(c, d, s, fairness, reservation))
    })
    .collect()

FairnessTracker and ReservationEngine are read-only snapshot inputs during scoring — no mutation occurs. See Scoring Model for the full formula.

Pass 7 — Sort descending

scored.sort_unstable_by(|a, b| {
    b.total_score
        .partial_cmp(&a.total_score)
        .unwrap_or(Ordering::Equal)
        .then_with(|| a.candidate.demand_id.cmp(&b.candidate.demand_id))
        .then_with(|| a.candidate.supply_id.cmp(&b.candidate.supply_id))
});

Highest total score first. Ties are broken deterministically by (demand_id, supply_id) to guarantee reproducibility.

Pass 8 — Greedy allocation

Figure: how constrained supply is split between reservation fences (future high-priority demands) and greedy allocation; residual demand becomes unfulfilled.

flowchart TD
    SUP["Available supply pool"] --> RES["Reservation fences<br/>(future high-priority demands)"]
    SUP --> REM["Remaining = qty_available - reserved"]
    REM --> GREEDY["Greedy allocation<br/>(sorted by total_score desc)"]
    GREEDY --> ALLOC["Allocations<br/>-> PIPE_allocation_results"]
    GREEDY --> LEFT{"demand remaining > epsilon?"}
    LEFT -->|"yes"| UNFUL["UnfulfilledDemand<br/>-> PIPE_allocation_unfulfilled"]
    LEFT -->|"no"| DONE["Demand fully filled"]

Iterates sorted candidates. For each:

  1. Skip if supply remaining < ε or demand remaining < ε
  2. Skip if the demand's fairness group is hard-capped
  3. Compute alloc_qty = min(demand_remaining, supply_remaining, allocatable_qty)
  4. Deduct from both remaining maps
  5. Register allocation with FairnessTracker
  6. Build an AllocationExplanation with full score narrative

Reservation fences reduce available supply upfront:

let mut supply_remaining: HashMap<String, f64> = supplies
    .iter()
    .map(|s| {
        let reserved = reservation.reserved_qty(&s.supply_id);
        (s.supply_id.clone(), (s.qty_available - reserved).max(0.0))
    })
    .collect();

After the greedy pass, any demand with remaining quantity > ε becomes an UnfulfilledDemand with an explanatory reason string.

Scoring Model

Each feasible candidate receives a ScoreBreakdown computed by ScoreEngine::score_one(). The total score formula:

total = priority_component
      + aging_component
      + lateness_component
      + strategic_component
      + revenue_component
      + criticality_component
      − routing_penalty
      − fairness_penalty
      − reservation_penalty

Component details

Priority component

The StarvationEngine may escalate the demand's priority class (one level up) if auto_upgrade_priority = true and the demand has waited beyond escalation_threshold_days. The resulting effective_priority is looked up in the PriorityScores table:

PriorityClass Default score
AOG 10,000
CRITICAL 5,000
PREMIUM 2,000
STANDARD 1,000
LOW 100

All values are configurable via AllocationConfig.priority_scores.

Aging component

let aging_component = aging_policy.score(backlog_days) + starvation_bonus;

AgingPolicy supports three modes:

Mode Formula
Linear multiplier × days_late
Exponential multiplier × days_late^exponent
None 0

Default: Exponential with multiplier = 10.0, exponent = 1.5.

The starvation_bonus adds score_per_extra_day × (backlog_days − escalation_threshold) when the demand exceeds the threshold, capped at max_starvation_bonus.

Lateness component

Negative penalty based on expected delivery delay. Uses a LatenessCurve — a piecewise-linear function of (days_late, penalty) breakpoints with linear interpolation and flat extrapolation beyond the last point.

Default curve: [(0, 0), (5, 10), (10, 50), (20, 500)]

This is a penalty in the naming convention but is added to the score (negative lateness values make earlier supply more attractive relative to late supply within the same candidate set). Note: in the ScoreBreakdown::total() implementation, lateness_component is added (it is typically negative or zero).

Business weight components

strategic_component  = demand.strategic_weight  × config.strategic_weight_factor
revenue_component    = demand.revenue_weight    × config.revenue_weight_factor
criticality_component = demand.criticality_weight × config.criticality_weight_factor

Weights are sourced from master.item and scaled by global multipliers (default 1.0).

Routing penalty

routing_penalty = demand.routing_policy_id
    → routing_policies[id]
    → policy.check(supply).penalty

Each soft violation incurs routing_violation_penalty (default 100.0). Hard violations were already filtered in Pass 5.

Fairness penalty

fairness_penalty = fairness_tracker.penalty_for_group(group, allocatable_qty)

When a group's actual allocation share exceeds target_share + tolerance, the penalty is:

penalty = penalty_per_over_unit × overshoot_fraction × proposed_qty

If hard_cap = true, the group is entirely blocked from further allocation once the cap is exceeded.

Reservation penalty

(reservation_penalty, _) = reservation.penalty_for_demand(supply_id, demand)

If the supply has a fence for a higher-priority future demand, the lower-priority demand incurs ReservationConfig.penalty (default 2,000). This is a soft penalty — the allocation can still proceed if the score remains competitive.

Key Structs (Rust)

All structs are defined in models.rs unless noted.

Demand

pub struct Demand {
    pub demand_id: String,
    pub item_id: String,
    pub location_id: String,
    pub qty: f64,
    pub required_date: i64,          // Unix epoch days
    pub priority_class: PriorityClass,
    pub customer_class: Option<String>,
    pub channel: Option<String>,
    pub program: Option<String>,
    pub routing_policy_id: Option<String>,
    pub allocation_policy_id: Option<String>,
    pub max_delay_days: Option<i32>,
    pub aging_policy_id: Option<String>,
    pub lateness_curve_id: Option<String>,
    pub strategic_weight: f64,
    pub revenue_weight: f64,
    pub criticality_weight: f64,
    pub backlog_days: i32,           // enriched by engine
}

AllocationSupply

pub struct AllocationSupply {
    pub supply_id: String,
    pub item_id: String,
    pub location_id: String,
    pub qty_available: f64,
    pub available_date: i64,         // Unix epoch days
    pub source_type: SourceType,     // Buy | Make | Repair | Transfer | Inventory
    pub inventory_pool: Option<String>,
    pub supplier_id: Option<String>,
    pub program: Option<String>,
    pub reliability_score: f64,      // default 1.0
    pub routing_tags: Vec<String>,
}

AllocationCandidate

pub struct AllocationCandidate {
    pub demand_id: String,
    pub supply_id: String,
    pub allocatable_qty: f64,
    pub feasible: bool,
    pub hard_constraint_violations: Vec<String>,
    pub soft_constraint_violations: Vec<String>,
}

ScoreBreakdown

pub struct ScoreBreakdown {
    pub priority_component: f64,
    pub aging_component: f64,
    pub lateness_component: f64,
    pub strategic_component: f64,
    pub revenue_component: f64,
    pub criticality_component: f64,
    pub routing_penalty: f64,
    pub fairness_penalty: f64,
    pub reservation_penalty: f64,
}

impl ScoreBreakdown {
    pub fn total(&self) -> f64 {
        self.priority_component + self.aging_component + self.lateness_component
        + self.strategic_component + self.revenue_component + self.criticality_component
        - self.routing_penalty - self.fairness_penalty - self.reservation_penalty
    }
}

ScoredCandidate

pub struct ScoredCandidate {
    pub candidate: AllocationCandidate,
    pub score: ScoreBreakdown,
    pub total_score: f64,
}

Allocation

pub struct Allocation {
    pub demand_id: String,
    pub supply_id: String,
    pub item_id: String,
    pub qty: f64,
    pub allocation_date: i64,
    pub supply_location_id: String,
}

UnfulfilledDemand

pub struct UnfulfilledDemand {
    pub demand_id: String,
    pub item_id: String,
    pub location_id: String,
    pub requested_qty: f64,
    pub allocated_qty: f64,
    pub shortfall_qty: f64,
    pub reasons: Vec<String>,
}

AllocationExplanation

pub struct AllocationExplanation {
    pub demand_id: String,
    pub supply_id: Option<String>,
    pub total_score: f64,
    pub priority_component: f64,
    pub aging_component: f64,
    pub lateness_component: f64,
    pub strategic_component: f64,
    pub revenue_component: f64,
    pub criticality_component: f64,
    pub routing_penalty: f64,
    pub fairness_penalty: f64,
    pub reservation_penalty: f64,
    pub violated_soft_constraints: Vec<String>,
    pub narrative: Vec<String>,
}

AllocationResult

pub struct AllocationResult {
    pub allocations: Vec<Allocation>,
    pub unfulfilled_demands: Vec<UnfulfilledDemand>,
    pub explanations: Vec<AllocationExplanation>,
    pub metrics: AllocationMetrics,
}

AllocationMetrics

pub struct AllocationMetrics {
    pub total_demand_qty: f64,
    pub total_allocated_qty: f64,
    pub total_shortfall_qty: f64,
    pub fill_rate: f64,
    pub demands_fully_filled: usize,
    pub demands_partially_filled: usize,
    pub demands_unfilled: usize,
    pub candidates_generated: usize,
    pub candidates_after_hard_filter: usize,
    pub elapsed_ms: u64,
}

Enums

pub enum PriorityClass { Low = 0, Standard = 1, Premium = 2, Critical = 3, Aog = 4 }
pub enum SourceType { Buy, Make, Repair, Transfer, Inventory }
pub enum AgingMode { None, Linear, Exponential }
pub enum FairnessGroupField { Channel, CustomerClass, Program, Region }

Configuration

AllocationConfig

The root configuration struct (config.rs), stored as JSONB in {schema}.allocation_parameter_set.params:

pub struct AllocationConfig {
    pub current_date: i64,                              // Unix epoch days
    pub pipeline_id: i32,
    pub scenario_id: i32,
    pub priority_scores: PriorityScores,                // per-class score table
    pub strategic_weight_factor: f64,                   // default 1.0
    pub revenue_weight_factor: f64,                     // default 1.0
    pub criticality_weight_factor: f64,                 // default 1.0
    pub starvation: StarvationConfig,
    pub fairness: Option<FairnessConfig>,
    pub reservation: Option<ReservationConfig>,
    pub routing_policies: HashMap<String, RoutingPolicy>,
    pub aging_policies: HashMap<String, AgingPolicy>,
    pub lateness_curves: HashMap<String, LatenessCurve>,
    pub default_aging_policy_id: Option<String>,
    pub default_lateness_curve_id: Option<String>,
}

StarvationConfig

pub struct StarvationConfig {
    pub enabled: bool,
    pub escalation_threshold_days: i32,   // default 14
    pub score_per_extra_day: f64,         // default 50.0
    pub max_starvation_bonus: f64,        // default 3_000.0 (0 = uncapped)
    pub auto_upgrade_priority: bool,      // default false
}

ReservationConfig

pub struct ReservationConfig {
    pub enabled: bool,
    pub lookahead_days: i32,              // default 14
    pub min_protected_priority: PriorityClass, // default Premium
    pub penalty: f64,                     // default 2_000.0
    pub safety_margin: f64,               // default 0.1 (10% extra)
}

FairnessConfig

pub struct FairnessConfig {
    pub enabled: bool,
    pub group_field: FairnessGroupField,  // default Channel
    pub target_shares: HashMap<String, f64>, // empty = equal shares
    pub tolerance: f64,                   // default 0.1 (10%)
    pub penalty_per_over_unit: f64,      // default 50.0
    pub hard_cap: bool,                  // default false
}

Segment-level overrides

Parameters can be overridden per segment via the allocation_parameter_set_segment junction table. The resolver checks segment linkage first, then falls back to the scenario default (is_default = TRUE).

def _resolve_global(self, param_type, segment_id):
    if segment_id is not None:
        for r in rows:
            segs = self._seg_index.get(r["id"], [])
            if segment_id in segs:
                return r["params"], r["id"]
    for r in rows:
        if r["is_default"]:
            return r["params"], r["id"]
    return rows[0]["params"], rows[0]["id"]

Building

Prerequisites

  • Rust toolchain — rustup with stable channel (edition 2021)
  • maturin — Python packaging tool for PyO3 projects

Build command

From the project root, activate the virtual environment and build the native extension:

cd files/supply
maturin develop --release

This compiles the supply_engine crate as a C-compatible shared library (cdylib) and installs it into the active Python environment. The --release flag is essential for production performance — debug builds are 10-100× slower due to lack of optimization and bounds-checking overhead.

The crate manifest (Cargo.toml) declares:

[lib]
name = "supply_engine"
crate-type = ["cdylib", "rlib"]

[dependencies]
pyo3 = { version = "0.21", features = ["extension-module", "abi3-py38"] }
rayon = "1.10"
serde = { version = "1.0.185", features = ["derive"] }
serde_json = "1"

The abi3-py38 feature enables stable ABI — the wheel is forward-compatible with any Python ≥ 3.8 without recompilation.

Verifying the build

import supply_engine
result = supply_engine.run_allocation('[]', '[]', '{}')
# Should return: {"allocations":[], "unfulfilled_demands":[], "explanations":[], "metrics":{...}}

Testing

CLI — full pipeline

python -m files.run_pipeline --only allocation --pipeline-id 1 --scenario-id 1

Standalone runner

python files/allocation_runner.py --pipeline-id 1 --scenario-id 1
python files/allocation_runner.py --pipeline-id 1 --scenario-id 1 --dry-run
python files/allocation_runner.py --pipeline-id 1 --scenario-id 1 --segment-id 5

Verification — scenario 1 baseline

Metric Value
Demands 20,016
Supplies 575
Elapsed ~486 ms
Allocations 1,905
Fill rate varies by supply/demand ratio

Rust unit tests

cd files/supply
cargo test score_alloc

The test suite in engine.rs covers:

  • AOG wins over Standard for shared supply
  • Partial allocation when supply insufficient
  • Starvation escalation (stale Standard beats fresh Premium)
  • Hard routing constraint blocks candidate
  • Future reservation protection preserves supply for AOG
  • Fair share hard cap limits dominant group
  • Deterministic output on repeated runs
  • No supply → unfulfilled
  • Metrics consistency

CH Tables

Results are written to three ClickHouse tables, partitioned by pipeline_id for efficient partition replacement.

PIPE_allocation_results

CREATE TABLE IF NOT EXISTS PIPE_allocation_results
(
    pipeline_id         UInt32,
    scenario_id         UInt32,
    run_id              UInt64,
    demand_id           String,
    supply_id           String,
    item_id             UInt32,
    site_id             UInt32,
    allocated_qty       Float64,
    total_score         Float64,
    score_priority      Float64,
    score_aging         Float64,
    score_lateness      Float64,
    score_strategic     Float64,
    score_revenue       Float64,
    score_criticality   Float64,
    penalty_routing     Float64,
    penalty_fairness    Float64,
    penalty_reservation Float64,
    explanation         String
)
ENGINE = MergeTree()
PARTITION BY pipeline_id
ORDER BY (pipeline_id, scenario_id, demand_id, supply_id);

PIPE_allocation_unfulfilled

CREATE TABLE IF NOT EXISTS PIPE_allocation_unfulfilled
(
    pipeline_id   UInt32,
    scenario_id   UInt32,
    run_id        UInt64,
    demand_id     String,
    item_id       UInt32,
    site_id       UInt32,
    requested_qty Float64,
    allocated_qty Float64,
    shortfall_qty Float64,
    reasons       Array(String)
)
ENGINE = MergeTree()
PARTITION BY pipeline_id
ORDER BY (pipeline_id, scenario_id, demand_id);

PIPE_allocation_explanations

CREATE TABLE IF NOT EXISTS PIPE_allocation_explanations
(
    pipeline_id UInt32,
    scenario_id UInt32,
    run_id      UInt64,
    demand_id   String,
    supply_id   String,
    step        String,    -- e.g. "step_0", "step_1", ...
    narrative   String
)
ENGINE = MergeTree()
PARTITION BY pipeline_id
ORDER BY (pipeline_id, scenario_id, demand_id, step);

The DDL is maintained in files/DDL/allocation_ch_schema.sql.

Write pattern

The Python runner uses an atomic replace pattern:

drop_ch_partition("PIPE_allocation_results",      pipeline_id)
drop_ch_partition("PIPE_allocation_unfulfilled",  pipeline_id)
drop_ch_partition("PIPE_allocation_explanations", pipeline_id)
# ... then bulk_insert_ch() for each table

This ensures no stale data from a previous run coexists with the current results.

Detail Timeline API

The Time-Phased Allocation Timeline panel in the detail view (AllocationTimeline.jsx) is fed by a single endpoint that aggregates the persisted PIPE_allocation_* rows for one (item, site).

Endpoint

GET /api/allocation/detail/{item_id}/{site_id}/timeline
    ?pipeline_id=…&scenario_id=…

Both pipeline_id and scenario_id are required (HTTP 400 if missing — see Pipeline ID guard). The endpoint resolves the latest run_id + archive_date via _alloc_run_info(pipeline_id, scenario_id) and returns { supplies, demands, reservations }.

Supply rows

Top-12 supplies by allocated quantity, grouped by supply_id:

SELECT supply_id, sum(allocated_qty)
FROM PIPE_allocation_results
WHERE pipeline_id = {pid} AND scenario_id = {sid} AND archive_date = {ad} AND run_id = {rid}
  AND item_id = {iid} AND site_id = {site}
GROUP BY supply_id ORDER BY sum(allocated_qty) DESC LIMIT 12

Each row is decorated with source_type (inventory if supply_id starts with OH_, else purchase_order) and available_date = planning_today(pipeline_id).

Demand rows

Top-18 demands by allocated quantity, grouped by demand_id:

SELECT demand_id, sum(allocated_qty), sum(total_score)
FROM PIPE_allocation_results
WHERE pipeline_id = {pid} AND scenario_id = {sid} AND archive_date = {ad} AND run_id = {rid}
  AND item_id = {iid} AND site_id = {site}
GROUP BY demand_id ORDER BY sum(allocated_qty) DESC LIMIT 18

The week_index is parsed from the trailing W{n} of the demand_id and converted to a required_date = planning_today(pipeline_id) + n weeks.

Requested / Allocated reconciliation

The endpoint cross-references PIPE_allocation_unfulfilled to derive requested_qty and the fulfilment status:

Demand in unfulfilled table? requested allocated status
Yes requested_qty + allocated_qty allocated_qty unfulfilled (if allocated ≈ 0) else partial
No sum(allocated_qty) sum(allocated_qty) full

Fallback-branch false positive

When a demand has sum(allocated_qty) ≈ 0 and is absent from PIPE_allocation_unfulfilled, the fallback branch sets requested = allocated = 0 and labels the status "full" — even though nothing was committed. This is the root cause of the "Requested 0, Allocated 0, Score 6000" symptom in the timeline (see User Guide › Troubleshooting).

The Score field

score in the response is sum(total_score) — the unrounded composite score summed across every PIPE_allocation_results row for that demand_id. Because total_score reflects priority and business weights (not committed quantity), it can be non-zero even when allocated_qty rounds to zero.

With default allocation_priority settings (config.rs:PriorityScores::default()):

PriorityClass Default score
AOG 10 000
CRITICAL 5 000
PREMIUM 2 000
STANDARD 1 000
LOW 100

A score of 6 000 on a standard demand = 1 000 × 6 scored supply candidates (or priority + aging components). The score being identical across every week indicates every week shares the same priority class and the same count of scored supplies — i.e. the part was scored but essentially nothing was committed to it.

Frontend rendering

AllocationTimeline.jsx renders a custom ECharts Gantt series:

  • Y-axis categories: supply rows first ([S] {source} · {supply_id}), then demand rows ({PRIORITY} {demand_id})
  • Each bar = one week wide (7 × 24 × 3600 × 1000 ms), colour by priority class (demand) or source type (supply)
  • Status encoding: status=2 (unfulfilled) → dashed red outline + faded fill; status=1 (partial) → right-edge stripe overlay; lateness_days > 0 → red border
  • Tooltip parses the JSON-serialised demand/supply object stored in the bar's value[7]
  • Click → selectDemand(demand_id) + openSupplyPath(demand_id, item_id, site_id), or selectSupply(supply_id) for supply bars

Key files

File Role
files/api/main.py → alloc_detail_timeline Endpoint implementation (lines ~34515)
files/frontend/src/components/allocation/AllocationTimeline.jsx Gantt chart component
files/frontend/src/components/allocation/useAllocationData.js → useAllocationTimeline Data-fetch hook
files/frontend/src/components/allocation/DetailAllocationTab.jsx Hosts the timeline panel (panel key timeline)
files/supply/src/score_alloc/scoring.rs → ScoreEngine::score_one Composite score computation
files/supply/src/score_alloc/config.rs → PriorityScores::default Default priority score table