Causal Forecast Engine (SQL-First)¶

Chapter summary

The causal forecast engine predicts spare-part demand from fleet deployments, failure rates, and maintenance schedules. This chapter documents the SQL-first implementation that runs the core calculation as a single PostgreSQL CTE chain, the fallback Python path, how the two maintain parity, and the performance indexes and override handling that make it production-viable.

Demand Formula¶

For each combination of (item, customer, period):

\[ D = \underbrace{\text{effective\_qty} \times \text{fleet\_qty} \times \text{failure\_rate} \times \text{total\_usage}}_{\text{unscheduled}} + \underbrace{\text{scheduled\_demand}}_{\text{maintenance}} \]

Where:

Component	Meaning	Source
`effective_qty`	Parts consumed per asset (from BOM)	`master.causal_bom.qty_per_asset`
`fleet_qty`	Fleet count or 1 (depends on `usage_mode`)	`scen_causal_fleet_plan.asset_qty`
`failure_rate`	Replacements per unit usage (waterfall-resolved)	`master_causal_failure_rate`
`total_usage`	Aggregate usage per (customer, usage_type, period)	`scen_causal_asset_usage` (deployment-filtered)
`scheduled_demand`	Parts from planned maintenance events	`scen_maintenance_scenario_event_date` × `maintenance_bom`

Two-Path Architecture¶

The engine has two implementations that must produce identical results:

Path	When used	Implementation
SQL (default)	No `new_entries` or `by_asset` overrides	Single CTE chain in `_build_sql()` → `sql_engine.py`
Python (fallback)	Complex overrides that SQL cannot express	Row-by-row waterfall + range-join → `demand_generator.py`

The decision is made by _requires_python_engine() which checks for:

fleet_overrides.new_entries — SQL cannot inject new fleet rows
usage_overrides.by_asset — SQL cannot apply per-asset usage patches

All other overrides (global/customer multipliers, deactivation, retired entries, per-scenario usage replacement) are handled inside the SQL engine.

Figure: _requires_python_engine() inspects override complexity; only new-fleet or per-asset usage patches force the Python fallback, otherwise the SQL CTE chain runs. Both paths converge on spread_by_coverage().

flowchart TD
  IN["Causal run request<br/>(pipeline_id, scenario_id)"] --> CHK["_requires_python_engine()"]
  CHK --> C1{"fleet_overrides.new_entries?"}
  C1 -- "yes" --> PY["Python fallback<br/>(demand_generator.py)"]
  C1 -- "no" --> C2{"usage_overrides.by_asset?"}
  C2 -- "yes" --> PY
  C2 -- "no" --> SQL["SQL engine<br/>(sql_engine.py, CTE chain)"]
  SQL --> SP["spread_by_coverage()"]
  PY --> SP
  SP --> OUT["PIPE_causal_forecast"]

Coverage Spreading¶

Both paths delegate coverage spreading (splitting demand by site_id) to the shared Python function spread_by_coverage() because it is trivially fast (~0.02 s) and does not benefit from SQL execution.

SQL Engine CTE Chain¶

The SQL engine produces one large WITH query. The CTEs execute in this order:

effective_bom          Fleet plan × BOM (only asset_types in base fleet)
  ↓
deployment             Asset-to-site deployment intervals
  ↓
base_usage_raw         Raw base usage (scenario_id = 1)
scenario_usage_raw    Raw scenario usage (if has_scenario_usage)
  ↓
base_tracked           Deployment-filtered + prorated base usage
base_untracked        Base usage for assets without deployments
base_filtered         UNION of tracked + untracked base
  ↓
scenario_tracked      Deployment-filtered + prorated scenario usage
scenario_untracked    Scenario usage for assets without deployments
scenario_filtered     UNION of tracked + untracked scenario
  ↓
scenario_usage_keys    DISTINCT (customer_id, usage_type_id) from scenario
raw_usage             scenario_filtered UNION base_filtered (anti-join)
  ↓
total_usage           SUM(factor_qty) per (customer, usage_type, period)
  ↓
best_fr               DISTINCT ON waterfall — highest-specificity FR row
  ↓
usage_in_fr           Overlap-prorated range-join: total_usage × FR periods
  ↓
unscheduled           SUM(effective_qty × fleet_qty × failure_rate × total_usage)
  ↓
scheduled             Maintenance demand from build_maintenance_sql()
  ↓
final                 LEFT JOIN unscheduled + scheduled

Key Design Decisions¶

Figure: the SQL engine compiles one WITH query whose CTEs flow from the base fleet plan through deployment filtering, usage merge, failure-rate waterfall, and the overlap-prorated range-join into the final unscheduled + scheduled demand.

flowchart TD
  EB["effective_bom<br/>(fleet plan x BOM)"] --> DEP["deployment<br/>(asset-site intervals)"]
  DEP --> BU["base_usage_raw / scenario_usage_raw"]
  BU --> BF["base_filtered / scenario_filtered<br/>(deployment-prorated)"]
  BF --> RU["raw_usage<br/>(scenario UNION base anti-join)"]
  RU --> TU["total_usage<br/>(SUM per customer, type, period)"]
  TU --> FR["best_fr<br/>(DISTINCT ON waterfall)"]
  FR --> UF["usage_in_fr<br/>(overlap-prorated range-join)"]
  UF --> UN["unscheduled demand"]
  UN --> FN["final<br/>LEFT JOIN scheduled (maintenance)"]

1. Base Fleet Source¶

The effective_bom CTE reads from scen_causal_fleet_plan WHERE scenario_id = 1 (base fleet) and applies fleet overrides in the CASE expression. This matches the Python path which does apply_fleet_overrides(base_fleet, overrides). Scenarios seeded from master data may have partial fleet rows; reading the base fleet ensures all customers and asset types are present.

2. Deployment Filtering Before Anti-Join¶

Deployment filtering is applied to each usage source independently before the scenario-usage merge. The Python path calls filter_usage_by_deployment() on base usage and scenario usage separately, then merge_scenario_usage(). The SQL path replicates this by:

Filtering base_usage_raw through deployment → base_filtered
Filtering scenario_usage_raw through deployment → scenario_filtered
Anti-joining base_filtered against scenario_usage_keys

If deployment filtering happened after the merge, scenario rows that are zeroed by deployment would still trigger the anti-join, incorrectly dropping valid base rows.

3. Usage Overrides Excluded from Scenario Rows¶

When has_scenario_usage = True, the raw_usage CTE tags each row with is_scenario (1 for scenario-sourced, 0 for base-sourced). The total_usage CTE applies usage_qty_expr (global/customer multipliers) only to base rows:

CASE WHEN au.is_scenario = 0 THEN {usage_qty_expr} ELSE au.factor_qty END

This matches the Python path: apply_usage_overrides() is called on base usage before merge_scenario_usage(), so scenario rows retain raw factor_qty.

4. Fleet Override Rounding¶

The SQL fleet quantity expression uses ROUND(...)::int and GREATEST(0, ...) to match Python's .round().astype(int).clip(lower=0):

GREATEST(0, ROUND(fp.asset_qty * {multiplier!r})::int)

Without rounding, SQL produces float fleet quantities (e.g., 5 × 1.3 = 6.5) while Python rounds to integer (6), causing drift in per_asset usage mode.

5. DISTINCT ON Instead of ROW_NUMBER¶

The best_fr CTE uses PostgreSQL's DISTINCT ON instead of the previous ROW_NUMBER() OVER (...) WHERE rk = 1 pattern. This avoids a full window sort and reduces query time from ~35 s to <1 s on production data.

6. Zero-Demand Filter¶

The unscheduled CTE includes HAVING SUM(...) > 0 to filter out rows where the prorated overlap between FR periods and usage periods is zero. This occurs when FR data contains daily periods (period_start = period_end) with no usage overlap. The Python path also omits these rows (NaN after merge → dropna).

Failure Rate Waterfall (SQL)¶

The best_fr CTE implements the 4-level specificity hierarchy as a DISTINCT ON with ORDER BY specificity DESC:

Level	Condition	Score
Asset-specific	`fr.asset_id IS NOT NULL AND != '0'`	+8
Site-specific	`fr.site_id IS NOT NULL AND != '0'`	+4
Position-specific	`fr.position IS NOT NULL AND != ''`	+2
Customer-specific	`fr.customer_id IS NOT NULL AND != 0`	+1

The JOIN excludes incompatible FR rows (e.g., a FR row specifying asset_id when the lookup context has no asset_id — effective_bom operates at the asset_type level). DISTINCT ON then picks the single highest-specificity surviving row per context.

Figure: the best_fr CTE scores each surviving failure-rate row by specificity (asset +8, site +4, position +2, customer +1) and DISTINCT ON ... ORDER BY specificity DESC keeps the single most-specific match per lookup context.

flowchart TD
  CTX["Lookup context<br/>(item, customer, site, asset, position)"] --> JN["JOIN master_causal_failure_rate<br/>(exclude incompatible dims)"]
  JN --> S1{"asset-specific?"}
  S1 -- "+8" --> B["best_fr<br/>(DISTINCT ON, specificity DESC)"]
  S1 -- "no" --> S2{"site-specific?"}
  S2 -- "+4" --> B
  S2 -- "no" --> S3{"position-specific?"}
  S3 -- "+2" --> B
  S3 -- "no" --> S4{"customer-specific?"}
  S4 -- "+1" --> B
  B --> OUT["single highest-specificity FR row"]

Specificity CASE relies on JOIN exclusion

The SQL specificity CASE awards points for any non-null dimension on the FR row, while Python's _specificity_score awards points only for matching dimensions. They are currently equivalent because the JOIN already excludes mismatching rows. If the JOIN is ever relaxed to allow asset-level FR resolution, the SQL CASE must be updated to also check for matches.

Override Handling in SQL¶

Fleet Overrides¶

Override	SQL Expression	Location
`asset_qty_multiplier`	`ROUND(fp.asset_qty * {mul}::numeric)::int`	`effective_bom` CTE
`asset_type_overrides[id].active = false`	`WHEN fp.asset_type_id = {id} THEN NULL`	`effective_bom` CTE
`asset_type_overrides[id].asset_qty_multiplier`	`WHEN fp.asset_type_id = {id} THEN ROUND(fp.asset_qty * {combined})::int`	`effective_bom` CTE
`retired_entries`	`AND NOT (fp.asset_type_id = {id} AND fp.customer_id = {cid})`	`effective_bom` WHERE clause

Usage Overrides¶

Override	SQL Expression	Location
`global_multiplier`	`factor_qty * {mul}` (base rows only)	`total_usage` CTE
`by_customer[id].multiplier`	`WHEN au.customer_id = {id} THEN factor_qty * {mul}`	`total_usage` CTE

Per-Scenario Usage Merge¶

When a non-base scenario has its own usage rows (has_scenario_usage = True), the SQL engine merges them with the base using an anti-join pattern:

scenario_filtered provides the scenario's deployment-filtered usage
scenario_usage_keys extracts the distinct (customer_id, usage_type_id) keys
base_filtered rows whose (customer_id, usage_type_id) is not in scenario_usage_keys are kept (anti-join via NOT EXISTS)
Scenario rows take priority for any key they cover; base rows fill the gaps

This mirrors Python's merge_scenario_usage() which replaces base usage with scenario usage for matching (customer_id, usage_type_id) combos.

The _has_scenario_usage_rows() function does a lightweight SELECT EXISTS check instead of loading the full usage DataFrame into Python, eliminating a redundant DB round-trip.

Granularity Support¶

Scheduled maintenance demand periods must align with unscheduled demand periods. The period_granularity parameter flows through:

Layer	Parameter	Effect
`scen_causal_scenarios.granularity`	Per-scenario column (default `'weekly'`)	Stored in DB
`run_causal_scenarios(period_granularity=)`	API/pipeline override	Falls back to scenario's own granularity
`build_maintenance_sql(period_granularity=)`	SQL `date_trunc('{unit}', event_date)`	Maps to PG units
`compute_scheduled_demand(period_granularity=)`	Pandas `.dt.to_period('{freq}')`	Maps to Pandas freq

Granularity mapping:

Causal value	PostgreSQL `date_trunc`	Pandas frequency
`daily`	`day`	`D`
`weekly`	`week`	`W`
`monthly`	`month`	`M`
`quarterly`	`quarter`	`Q`
`yearly`	`year`	`Y`

Performance Indexes¶

Four indexes support the SQL engine CTE chain. All use CONCURRENTLY to avoid blocking reads/writes during creation on production tables:

Index	Table	Columns	Serves
`idx_causal_usage_scenario_customer_type_period`	`scen_causal_asset_usage`	`(scenario_id, customer_id, usage_type_id, period_start, period_end)`	Range-join on usage
`idx_causal_fr_scenario_customer_type_period`	`master_causal_failure_rate`	`(scenario_id, customer_id, usage_type_id, period_start, period_end)`	FR customer filter
`idx_causal_fleet_scenario_type_period`	`scen_causal_fleet_plan`	`(scenario_id, asset_type_id, period_start, period_end)`	Fleet plan lookup
`idx_causal_fr_scenario_item_type_period`	`master_causal_failure_rate`	`(scenario_id, item_id, usage_type_id, period_start)`	`best_fr` range-join

The idx_causal_fr_scenario_item_type_period index is critical: the existing idx_causal_fr_waterfall index places period_start after asset_id, site_id, position, preventing a contiguous range scan. The new index places period_start right after the equality columns (scenario_id, item_id, usage_type_id) for efficient range seeks.

DDL: files/DDL/causal_performance_indexes.sql

SQL Injection Safety¶

Override IDs are interpolated into SQL after int()/float() casting. A pre-flight _validate_override_ids() function checks that all ID keys in fleet_overrides and usage_overrides are valid integers before any SQL string is built. This prevents malformed or injectable fragments from reaching the query.

Pydantic Literal types at the API layer constrain granularity to the 5 allowed values (daily, weekly, monthly, quarterly, yearly), preventing arbitrary input from reaching date_trunc().

Parity Testing¶

The test suite (files/tests/causal/test_sql_python_parity.py) verifies that the SQL and Python engines produce identical results for:

Test	Scenarios	Coverage
`test_sql_python_parity_base`	sid=1 (base, no scenario usage)	Base fleet + base usage
`test_sql_python_parity_non_base`	sid=3 (with scenario usage)	Merged usage path
`test_sql_python_parity_all_scenarios`	All 3 scenarios	Full coverage including coverage spreading

Each test compares item_id, customer_id, period_start, demand_mean, scheduled_demand, and unscheduled_demand between both engines using an outer merge with _merge indicator. No row may exist in only one engine, and all shared values must match within floating-point tolerance.

Failure Rate Bulk Save¶

The save_failure_rates_v2() function (in failure_rate_waterfall.py) uses bulk operations instead of the prior N+1 per-row DELETE/INSERT pattern:

Bulk DELETE: execute_values() with an IN %s tuple matching the COALESCE-based unique key. NULL dimension columns must be pre-COALESCE'd to empty strings in the key_rows tuples, because '' = NULL evaluates to UNKNOWN (falsy) in SQL.
Bulk INSERT: execute_values() with VALUES %s for all new rows.

Key Files¶

File	Purpose
`files/causal/sql_engine.py`	SQL-first engine: `_build_sql()`, `generate_demand_sql()`, `_requires_python_engine()`, `_validate_override_ids()`
`files/causal/demand_generator.py`	Python fallback: `generate_demand()`, `spread_by_coverage()`
`files/causal/scenario_runner.py`	Entry point: `run_causal_scenarios()`, `_has_scenario_usage_rows()`, engine decision
`files/causal/maintenance.py`	`_pg_trunc()`, `_PANDAS_FREQ_MAP`, `build_maintenance_sql()`, `compute_scheduled_demand()`
`files/causal/failure_rate_waterfall.py`	`vectorized_resolve()`, `save_failure_rates_v2()`, specificity scoring
`files/causal/fleet.py`	`load_scenarios()` (reads `granularity`), `apply_fleet_overrides()`
`files/DDL/causal_performance_indexes.sql`	4 performance indexes (CONCURRENTLY)
`files/DDL/schema.sql`	`granularity` column on `scen_causal_scenarios` + migration DO block
`files/api/main.py`	`CausalScenarioIn.granularity`, `CausalRunRequest.period_granularity`, CRUD endpoints
`files/tests/causal/test_sql_python_parity.py`	3 parity test functions

Timing Benchmarks (Production Data)¶

Typical performance on 3 scenarios, ~6 600 usage rows, ~180 000 FR rows:

Scenario	SQL Engine	Coverage Spread	Total
sid=1 (base)	1.5 s	0.03 s	1.6 s
sid=2 (scenario usage)	3.1 s	0.03 s	3.2 s
sid=3 (scenario usage, limited fleet)	0.3 s	0.02 s	0.3 s
Full pipeline (3 scenarios + save)			12.2 s

The DISTINCT ON optimization reduced best_fr resolution from ~35 s to <1 s. The _has_scenario_usage_rows() check replaced a full load_asset_usage() call (~0.02 s saved per scenario).