Allocation Engine

Score-based supply allocation engine written in Rust, exposed to Python via PyO3.

Table of Contents

1. Purpose
2. Architecture
3. Algorithm
4. Scoring Formula
5. Parameter System
6. Policy Types
7. Database Schema
8. Running the Engine
9. Seeding Parameters (theBicycle example)
10. Testing
11. Benchmarking
12. Extending

---

Purpose

Given a set of demands (customer or production orders requiring supply) and a set of supply buckets (on-hand inventory, purchase orders, manufacturing orders), the engine assigns scarce supply to demands in priority order while minimising total business pain.

Design goals: - Explainable — every allocation includes a decomposed score + narrative - Configurable — all policy parameters live in the database; no code changes needed - Deterministic — sort-based tie-breaking, no random choices - Fast — Rayon parallel scoring; 10 000 demands × 2 000 supply buckets < 2 s on 4 cores

---

Architecture

files/
├── supply/src/score_alloc/
│   ├── mod.rs          — module root, public re-exports
│   ├── models.rs       — domain types (Demand, AllocationSupply, ScoreBreakdown …)
│   ├── config.rs       — AllocationConfig, PriorityScores, StarvationConfig
│   ├── starvation.rs   — anti-starvation bonus + priority escalation
│   ├── scoring.rs      — parallel ScoreEngine (Rayon)
│   ├── engine.rs       — AllocationEngine: 8-pass algorithm + integration tests
│   └── policy/
│       ├── aging.rs        — AgingPolicy      (None / Linear / Exponential)
│       ├── fairness.rs     — FairnessConfig + FairnessTracker
│       ├── lateness.rs     — LatenessCurve    (piecewise-linear)
│       ├── reservation.rs  — ReservationConfig + ReservationEngine
│       └── routing.rs      — RoutingPolicy    (hard / soft violations)
├── allocation_runner.py    — Python CLI: resolver → engine → DB writer
├── DDL/
│   ├── allocation_schema.sql     — PostgreSQL tables (parameter sets, run log)
│   └── allocation_ch_schema.sql  — ClickHouse output tables (results, unfulfilled, explanations)
└── config/examples/
    ├── allocation_priority.json
    ├── allocation_starvation.json
    ├── allocation_reservation.json
    ├── allocation_fairness.json
    ├── allocation_routing_domestic.json
    ├── allocation_aging_linear.json
    └── allocation_lateness_standard.json

---

Algorithm

8 deterministic passes:

Pass	Action
1	Enrich backlog — backlog_days = max(0, current_date − required_date)
2	Seed FairnessTracker — total demand qty per group
3	Build reservation fences — protect supply for future high-priority demands
4	Generate candidates — item-matched (demand × supply) pairs
5	Partition — feasible (routing OK) vs. infeasible
6	Score in parallel (Rayon, read-only state)
7	Sort — score DESC, demand_id ASC, supply_id ASC (deterministic tie-break)
8	Greedy allocation — serial consumption; fairness hard-cap check + starvation reason capture per step

---

Scoring Formula

total_score =
    priority_component        +   PriorityScores lookup
    aging_component           +   AgingPolicy + starvation bonus
    lateness_component        +   LatenessCurve (supply arrival delay)
    strategic_component       +   demand.strategic_weight × factor
    revenue_component         +   demand.revenue_weight   × factor
    criticality_component     +   demand.criticality_weight × factor
  − routing_penalty           -   RoutingPolicy soft violation
  − fairness_penalty          -   FairnessTracker over-share
  − reservation_penalty       -   ReservationEngine protection

Higher total = allocated first.

---

Parameter System

Parameters are stored in scenario.allocation_parameter_set, split into seven distinct types. This mirrors the project's existing parameter pattern (forecast_parameter_set, meio_parameter_set) but goes further by giving each concern its own rows — making segment-level overrides granular and independently versionable.

Parameter types

parameter_type	Controls	One or many per scenario
allocation_priority	PriorityScores + weight multipliers	one (global for the run)
allocation_starvation	StarvationConfig	one
allocation_reservation	ReservationConfig	one
allocation_fairness	FairnessConfig	one
allocation_routing	one RoutingPolicy per row	many — demands ref by routing_policy_id
allocation_aging	one AgingPolicy per row	many — demands ref by aging_policy_id
allocation_lateness	one LatenessCurve per row	many — demands ref by lateness_curve_id

Resolution logic (AllocationParameterResolver)

For global types (priority / starvation / reservation / fairness):

for each global type:
  1. Look for a row linked to the demand's segment via allocation_parameter_set_segment
  2. Fall back to the row where is_default = TRUE for this scenario

For policy types (routing / aging / lateness):

Load ALL rows for the scenario (filtered to the segment if one is given).
Put them all into the AllocationConfig HashMaps.
Demands reference the right policy by ID at scoring time.

Rows with no segment linkage are visible to all demands (global policy pool). Rows linked to a segment are restricted to demands in that segment.

Assembled config snapshot

The runner records the full assembled AllocationConfig JSONB in allocation_run.resolved_config for auditability and replay. It also records the parameter_set.id used for each global type so you can trace exactly which row drove a given run.

---

Policy Types

allocation_priority

{
  "priority_scores": { "aog": 10000, "critical": 5000, "premium": 2000, "standard": 1000, "low": 100 },
  "strategic_weight_factor":   1.0,
  "revenue_weight_factor":     1.0,
  "criticality_weight_factor": 1.0
}

Segment use-case: give the MRO-critical segment a higher strategic_weight_factor to amplify the strategic importance encoded on each demand.

allocation_starvation

{
  "enabled": true,
  "escalation_threshold_days": 14,
  "score_per_extra_day":       50.0,
  "max_starvation_bonus":    3000.0,
  "auto_upgrade_priority":     false
}

Segment use-case: AOG-prone segments might use escalation_threshold_days: 7 (escalate faster) while standard segments keep 14.

allocation_reservation

{
  "enabled": true,
  "lookahead_days": 14,
  "min_protected_priority": "premium",
  "penalty": 2000.0,
  "safety_margin": 0.10
}

Supply within the lookahead window that is needed by demands at or above min_protected_priority is fenced. Lower-priority demands that try to consume it receive the reservation penalty.

allocation_fairness

{
  "enabled": true,
  "group_field": "channel",
  "target_shares": { "retail": 0.40, "wholesale": 0.35, "direct": 0.25 },
  "tolerance": 0.05,
  "penalty_per_over_unit": 50.0,
  "hard_cap": true
}

group_field selects which field on the Demand struct defines the group: channel, customer_class, program, or region.

allocation_routing (one row per policy)

{
  "policy_id":   "domestic_only",
  "hard_mode":   true,
  "allowed_locations":   ["WH_US_EAST", "WH_US_WEST"],
  "forbidden_pools":     ["quarantine"],
  "routing_violation_penalty": 100.0
}

hard_mode = true → violations make the candidate infeasible. hard_mode = false → violations add routing_violation_penalty.

allocation_aging (one row per policy)

{ "policy_id": "linear_10", "mode": "linear", "multiplier": 10.0, "exponent": 1.0 }

mode	formula
none	0
linear	multiplier × days_late
exponential	multiplier × days_late^exponent

allocation_lateness (one row per curve)

{
  "curve_id": "standard_lateness",
  "breakpoints": [[0,0],[5,10],[10,50],[20,500],[30,2000]]
}

Piecewise-linear; flat extrapolation beyond the last breakpoint.

---

Database Schema

-- Replace {schema} with scenario
\i files/DDL/allocation_schema.sql

Key tables:

Table	Purpose
allocation_parameter_set	One row per parameter set; parameter_type column distinguishes the 7 types
allocation_parameter_set_segment	Links a parameter set to a segment (for overrides)
allocation_run	Run log with resolved param IDs + full config snapshot
ClickHouse output tables
PIPE_allocation_results	Per-allocation record with full score breakdown (partitioned by pipeline_id)
PIPE_allocation_unfulfilled	Demands not fully met, with reasons Array(String) (partitioned by pipeline_id)
PIPE_allocation_explanations	Step-by-step narrative fragments per demand (partitioned by pipeline_id)

---

Running the Engine

Build the Rust extension

cd files/supply
pip install maturin
maturin develop          # debug (fast compile)
# maturin develop --release  # release (optimised)

Apply the schema

psql -U postgres -d thebicycle -c "\set schema scenario" -f files/DDL/allocation_schema.sql

Seed parameters (theBicycle example)

import json, psycopg2
conn = psycopg2.connect(host="localhost", dbname="thebicycle", user="postgres")
cur  = conn.cursor()

SCENARIO_ID = 1   # base supply scenario

def seed(param_type, name, params, is_default=True):
    cur.execute(
        """INSERT INTO scenario.allocation_parameter_set
               (scenario_id, parameter_type, name, is_default, params)
           VALUES (%s,%s,%s,%s,%s)""",
        (SCENARIO_ID, param_type, name, is_default, json.dumps(params))
    )

# Global types — one per scenario
seed("allocation_priority", "Default priority",
     json.load(open("files/config/examples/allocation_priority.json")))

seed("allocation_starvation", "Default starvation",
     json.load(open("files/config/examples/allocation_starvation.json")))

seed("allocation_reservation", "Default reservation",
     json.load(open("files/config/examples/allocation_reservation.json")))

seed("allocation_fairness", "Default fairness",
     json.load(open("files/config/examples/allocation_fairness.json")))

# Policy types — as many as needed; each demand refs by ID
seed("allocation_routing", "Domestic only",
     json.load(open("files/config/examples/allocation_routing_domestic.json")),
     is_default=False)   # policies are never "default"; demands ref by policy_id

seed("allocation_aging", "Linear ×10",
     json.load(open("files/config/examples/allocation_aging_linear.json")),
     is_default=False)

seed("allocation_lateness", "Standard lateness",
     json.load(open("files/config/examples/allocation_lateness_standard.json")),
     is_default=False)

conn.commit()
print("Parameter sets seeded.")

Add a segment-specific starvation override

# Aggressive starvation for the AOG-critical segment (segment_id = 3)
cur.execute(
    """INSERT INTO scenario.allocation_parameter_set
           (scenario_id, parameter_type, name, is_default, params)
       VALUES (%s,%s,%s,%s,%s) RETURNING id""",
    (SCENARIO_ID, "allocation_starvation", "Aggressive starvation (AOG segment)", False,
     json.dumps({
       "enabled": True,
       "escalation_threshold_days": 7,   # escalate after 7 days, not 14
       "score_per_extra_day": 100.0,
       "max_starvation_bonus": 5000.0,
       "auto_upgrade_priority": True
     }))
)
agg_id = cur.fetchone()[0]

cur.execute(
    "INSERT INTO scenario.allocation_parameter_set_segment (parameter_set_id, segment_id) VALUES (%s,%s)",
    (agg_id, 3)
)
conn.commit()

Run via CLI

# Dry run — engine runs, nothing written to DB
python files/allocation_runner.py --pipeline-id 1 --scenario-id 1 --dry-run

# Full run using scenario defaults
python files/allocation_runner.py --pipeline-id 1 --scenario-id 1

# Run preferring parameters for segment 3
python files/allocation_runner.py --pipeline-id 1 --scenario-id 1 --segment-id 3

Call the Rust engine directly from Python

import supply_engine, json

demands = [{
    "demand_id": "D1", "item_id": "10", "location_id": "301",
    "qty": 5.0, "required_date": 0,     "priority_class": "AOG",
    "strategic_weight": 0.0, "revenue_weight": 0.0, "criticality_weight": 0.0,
    "aging_policy_id": "linear_10"
}]
supplies = [{
    "supply_id": "S1", "item_id": "10", "location_id": "301",
    "qty_available": 20.0, "available_date": 0, "source_type": "inventory",
    "reliability_score": 1.0
}]
config = {
    "current_date": 0, "pipeline_id": 1, "scenario_id": 1,
    "aging_policies": {
        "linear_10": {"policy_id": "linear_10", "mode": "linear", "multiplier": 10.0, "exponent": 1.0}
    }
}

result = json.loads(supply_engine.run_allocation(
    json.dumps(demands), json.dumps(supplies), json.dumps(config)
))
print(result["metrics"])
# {'total_demand_qty': 5.0, 'total_allocated_qty': 5.0, 'fill_rate': 1.0, ...}

---

Testing

cd files/supply
cargo test --lib allocation            # all 38 allocation tests
cargo test --lib allocation -- --nocapture   # with output

---

Benchmarking

cd files/supply
cargo bench --bench allocation_bench
# HTML report: target/criterion/

Benchmark	Demands	Supplies	Typical
allocation_100d_20s	100	20	~0.1 ms
allocation_1000d_200s	1 000	200	~5 ms
allocation_10k_demands	10 000	2 000	~1.5 s

---

Extending

Add a new global parameter type

1. Add the field to AllocationConfig in config.rs
2. Compute the component in ScoreEngine::score_one in scoring.rs
3. Add the type constant to allocation_runner.py and wire it through resolve_config()
4. Create an example JSON in config/examples/

Add a new per-demand policy type

1. Create policy/<name>.rs, add pub mod <name>; to policy/mod.rs
2. Add a HashMap<String, NewPolicy> to AllocationConfig
3. Add a field <name>_policy_id: Option<String> to Demand
4. Reference the policy in scoring.rs
5. Use PT_<NAME> = "allocation_<name>" in the runner; add to POLICY_TYPES