Airline operations

schedule

Flight schedules: flights, aircraft rotations, crew pairings and passenger connections.

Times are minutes from midnight of the operating day. CSV files may give times as HH:MM (hours may exceed 23 for after-midnight arrivals), as minutes, or as ISO date-times (2026-10-08T06:30), which are converted relative to midnight of the earliest date in the file.

Flight dataclass

Flight(id: str, tail: str, origin: str, dest: str, std: float, sta: float, pax: int = 150, crew: str = '', aircraft_type: str = '', seats: int = 0)

One scheduled flight leg.

Connection dataclass

Connection(inbound: str, outbound: str, pax: int)

pax passengers connecting from flight inbound to flight outbound.

Schedule

Schedule(flights: Iterable[Flight], connections: Iterable[Connection] = ())

A day of flights, with rotations (by tail), crew pairings and connections.

s = Schedule([ ... Flight("A1", "T1", "HUB", "AAA", 360, 450), ... Flight("A2", "T1", "AAA", "HUB", 495, 585), ... ]) ", ".join(f.id for f in s.rotations["T1"]) 'A1, A2'

validate

validate() -> list[str]

Problems that make the schedule unflyable (empty list when consistent).

replace_flights

replace_flights(flights: Iterable[Flight]) -> Schedule

A schedule with flights and the connections that still refer to them.

retime

retime(shifts: Mapping[str, float]) -> Schedule

Move flights by {flight_id: minutes} (positive = later).

from_csv classmethod

from_csv(path: str | Path, connections: str | Path | None = None, *, default_pax: int = 150) -> Schedule

Load a schedule CSV (and optionally a connections CSV: inbound, outbound, pax).

synthetic classmethod

synthetic(*, hub: str = 'HUB', spokes: Sequence[str] = ('AAA', 'BBB', 'CCC', 'DDD', 'EEE', 'FFF', 'GGG', 'HHH'), tails: int = 12, first_departure: float = 360.0, last_departure: float = 1260.0, turn: float = 45.0, banks: bool = True, bank_spacing: float = 120.0, crews_per_day: int = 2, connect_share: float = 0.3, seed: int = 0) -> Schedule

A hub-and-spoke day: each tail flies hub-spoke round trips.

With banks the hub departures cluster in waves bank_spacing apart (good connections, peaky runway load); without, they are spread evenly. Crews fly half-day pairings; a share of passengers on each inbound connects to the next hub departures.

parse_time

parse_time(value: Any, base: datetime | None = None) -> float

Minutes from midnight: "06:30" -> 390, "25:10" -> 1510, 95 -> 95.

format_time

format_time(minutes: float) -> str

390 -> "06:30"; times past midnight keep counting ("25:10").

config

Operating assumptions for the airline network simulation.

:class:OpsConfig holds the rules of the day (turn times, runway rates, gates, curfews, crew duty limits, spares, standby crews, recovery thresholds, costs). :class:DelayModel says how much primary delay (technical, ATC, late passengers - anything not caused by an earlier flight) each departure gets and how variable block times are; it can be fitted to history with :func:simulsi.aviation.fit_delay_model. :class:WeatherEvent cuts an airport's departure rate for a time window, with a probability, so forecasts can carry weather scenarios.

DelayModel dataclass

DelayModel(prob: float = 0.2, mean: float = 25.0, block_cv: float = 0.06, table: dict[str, tuple[float, float]] = dict(), block_bias: dict[str, float] = dict(), block_scale: float = 1.0, shape: list[float] = list(), day_sigma: float = 0.0, airport_days: dict[str, list[float]] = dict(), cancel_rate: float = 0.0, cancel_days: dict[str, list[float]] = dict(), predictor: DelayPredictor | None = None)

Primary departure delays and block-time variability.

Each departure gets a primary delay with probability prob; its size is exponential with mean mean minutes. table overrides both per origin airport ("LHR") or per origin and hour ("LHR@07"), and block_bias scales the scheduled block time per route ("LHR-CDG"). predictor - for example a machine-learning model - returns the expected primary delay of a flight; the simulation then samples around that expectation and propagates it through the network.

shape replaces the exponential with an empirical distribution: equally spaced quantiles of primary delay divided by their mean (so mean still sets the size). Real delays have heavier tails than an exponential; :func:~simulsi.aviation.fit_delay_model fills it in. day_sigma makes whole days better or worse together (weather, ATC programmes): each simulated day draws a lognormal factor on every delay chance. airport_days does the same per airport from the empirical spread of good and bad days there (quantiles with mean 1; key "*" for any airport). cancel_rate is the chance a departure is cancelled for reasons outside the model (weather, crew, mechanical), scaled up on bad days; cancel_days replaces it with the empirical spread of each airport's daily cancelled share, drawn together with the delay factor.

airport_factors

airport_factors(airports: Sequence[str], rs: RandomStream) -> tuple[dict[str, float], dict[str, float]]

How bad the day is at each airport, drawn from history: (delay factor, cancelled share).

One draw per airport sets both, so a bad delay day is also a bad cancellation day.

day_factor

day_factor(rs: RandomStream) -> float

How bad the whole day is: multiplies every flight's delay chance (mean 1).

WeatherEvent dataclass

WeatherEvent(airport: str, start: float, end: float, capacity: float = 0.4, probability: float = 1.0, name: str = '', cancel: float = 0.0)

Departure capacity at airport falls to capacity x normal from start to end.

probability is the chance the event happens on the day (each replication draws it); capacity=0 closes the airport to departures. cancel is the share of departures in the window the airline cancels in advance (as for a forecast winter storm).

parse classmethod

parse(text: str) -> WeatherEvent

"HUB 15:00-18:00 0.4 p=0.6" (capacity and probability optional).

OpsConfig dataclass

OpsConfig(on_time: float = 15.0, min_turn: float = 35.0, min_turn_by_airport: dict[str, float] = dict(), late_turn_compression: float = 1.0, runway_rate: dict[str, float] = dict(), weather_base_rate: float = 30.0, gates: dict[str, int] = dict(), curfew: dict[str, float] = dict(), min_crew_connect: float = 30.0, duty_limit: float = 13 * 60.0, report_time: float = 60.0, release_time: float = 30.0, standby_callout: float = 60.0, spares: dict[str, int] = dict(), standby_crews: dict[str, int] = dict(), swap_threshold: float = 90.0, cancel_threshold: float = 240.0, mct: float = 35.0, load_factor: float = 0.85, overnight_delay: float = 18 * 60.0, spare_ferry_minutes: float | None = None, delay_cost_per_minute: float = 100.0, cancel_cost: float = 20000.0, misconnect_cost_per_pax: float = 250.0, eu261: bool = False, delays: DelayModel = DelayModel(), otp_recalibration: tuple[float, float] = (0.0, 1.0))

Rules and resources for one day of operations.

Per-airport settings are dictionaries keyed by airport code; airports not listed are unconstrained (no runway queue, unlimited gates, no curfew, no spares or standby crews).

thunderstorm

thunderstorm(airport: str, start: str | float, hours: float = 3.0, probability: float = 1.0) -> WeatherEvent

Convective weather: departure rate cut to 40 %.

morning_fog

morning_fog(airport: str, until: str | float = '10:00', probability: float = 1.0) -> WeatherEvent

Low visibility procedures from 05:00: departure rate cut to 60 %.

snow

snow(airport: str, start: str | float = '06:00', hours: float = 6.0, probability: float = 1.0) -> WeatherEvent

Snow and de-icing: departure rate cut to 50 % for a long window.

closure

closure(airport: str, start: str | float, minutes: float, probability: float = 1.0) -> WeatherEvent

Full closure to departures (storm, runway incident, ATC strike).

weather_from_config

weather_from_config(data: Mapping[str, Any]) -> list[WeatherEvent]

The weather: list of an ops YAML file (strings or mappings).

state

The live operation: what has already happened, and the actions controllers can take.

:class:OpsState captures the day so far - actual departure and arrival times, known delays (estimated departure times), cancellations and aircraft out of service - so a forecast can start from now instead of from the plan. Actions (:class:Cancel, :class:Retime, :class:Swap) change the plan; :func:apply_actions returns the changed schedule so alternatives can be simulated side by side.

FlightStatus dataclass

FlightStatus(atd: float | None = None, ata: float | None = None, etd: float | None = None, cancelled: bool = False)

What is known about one flight: actual or estimated times, or a cancellation.

OpsState dataclass

OpsState(now: float, flights: dict[str, FlightStatus] = dict(), aircraft_unavailable: dict[str, float] = dict())

The operation at time now (minutes from midnight).

from_records classmethod

from_records(now: float | str, rows: Iterable[Mapping[str, Any]] = (), aircraft_unavailable: Mapping[str, float | str] | None = None) -> OpsState

Rows with flight and any of atd, ata, etd, status (cancelled).

load classmethod

load(source: str, now: float | str | None = None) -> OpsState

Live status from a CSV file, or from an http(s) URL serving CSV or JSON.

JSON may be a list of status rows or {"now": "13:05", "flights": [...]}. now defaults to the feed's now, else the local clock.

to_csv

to_csv(path: str | Path | None = None) -> str

The flight statuses as CSV (flight, atd, ata, etd, status), readable by :meth:from_csv.

Cancel dataclass

Cancel(*flights: str)

Cancel flights (cancel a round trip to keep the aircraft where it is).

Retime dataclass

Retime(flight: str, minutes: float)

Delay a flight by minutes on purpose (later flights of the tail move only if they must).

Swap dataclass

Swap(tail_a: str, tail_b: str, after: float)

Swap the remaining rotations of two aircraft from after (minutes) onwards.

Both aircraft must be at the same airport at the swap point.

apply_actions

apply_actions(schedule: Schedule, actions: Iterable[Action]) -> tuple[Schedule, set[str]]

The schedule after actions, and the set of cancelled flight ids.

parse_action

parse_action(text: str) -> Action

"cancel F101 F102", "retime F101 30", "swap T01 T02 12:00".

engine

Simulate one day of an airline network: aircraft rotations, crews, runways, gates and recovery.

Each aircraft (tail) works through its rotation as a simulsi process. A departure waits for the aircraft (inbound arrival plus minimum turn), the crew (its previous flight plus a connection time) and any known estimate, then picks up a sampled primary delay and queues for a departure slot (only at airports with a runway rate or weather; arrivals there queue for landing slots too, and wait out closures). About an hour before each departure, operations control looks at the projected delay and may

• swap in a spare aircraft at the airport (the late aircraft becomes the spare),
• call a standby crew (late inbound crew, or a crew that would break its duty limit),
• cancel a round trip when the projected delay passes cancel_threshold,
• cancel the rest of a rotation that would break a curfew.

Arrivals may wait for a gate at airports with a gate limit. Passenger connections are checked after the day: a connection is missed when either flight is cancelled or the outbound leaves less than mct minutes after the inbound arrives.

The engine is deterministic for a seed, and every flight draws its random primary delay, block time and turn time from its own position in a fixed order, so two plans (say, with and without an action) see the same disturbances - common random numbers, which makes comparisons sharp.

FlightOutcome dataclass

FlightOutcome(id: str, tail: str, dep: float = math.nan, arr: float = math.nan, cancelled: bool = False, reason: str = '', dep_delay: float = math.nan, arr_delay: float = math.nan, primary: float = 0.0, reactionary_aircraft: float = 0.0, reactionary_crew: float = 0.0, runway: float = 0.0, gate_wait: float = 0.0, spare: bool = False, standby: bool = False, crew_risk: bool = False)

What happened to one flight in one simulated day.

DayOutcome dataclass

DayOutcome(flights: dict[str, FlightOutcome], misconnected: dict[tuple[str, str], int], metrics: dict[str, float], weather: list[str] = list())

One simulated day: per-flight outcomes, missed connections and network metrics.

build_day

build_day(sim: Simulation, schedule: Schedule, config: OpsConfig, *, weather: Sequence[WeatherEvent] = (), state: OpsState | None = None, cancelled: Iterable[str] = (), out: dict[str, Any] | None = None) -> None

Add one day of operations to sim; outcomes land in out when the run finishes.

rebook

rebook(schedule: Schedule, cfg: OpsConfig, outcomes: Mapping[str, FlightOutcome], misconnected: Mapping[tuple[str, str], int]) -> dict[str, float]

Re-accommodate disrupted passengers on later direct flights with free seats.

Misconnected passengers and the local passengers of cancelled flights are placed, earliest need first, on the first operated flight from where they are to where they are going that leaves after they are ready and still has seats (seats, or pax / load_factor when a flight has no seat count). Whoever finds no seat today is stranded until tomorrow (overnight_delay minutes).

simulate_day

simulate_day(schedule: Schedule, config: OpsConfig | None = None, *, seed: int | None = 0, weather: Sequence[WeatherEvent] = (), state: OpsState | None = None, actions: Sequence[Action] = ()) -> DayOutcome

Simulate one day and return per-flight outcomes and network metrics.

from simulsi.aviation import Schedule day = simulate_day(Schedule.synthetic(tails=4), seed=1) 0 <= day.metrics["otp"] <= 1 True

network_model

network_model(schedule: Schedule, config: OpsConfig | None = None, *, weather: Sequence[WeatherEvent] = (), hub: str | None = None) -> Model

The day as a simulsi :class:~simulsi.Model, for experiments, optimisation and Pareto search.

Parameters: spares and standby_crews (at the hub), swap_threshold, cancel_threshold, min_turn, delay_scale (multiplies the chance of a primary delay) and weather (0/1).

forecast

Monte Carlo forecasts of a day of operations: per flight, per rotation, per connection.

:func:forecast simulates the day many times (from the plan, or from the live state at now) and turns the outcomes into probabilities and delay ranges: the chance each flight is on time or cancelled, its departure delay quantiles, the expected passengers who misconnect, which aircraft rotations are fragile, and alerts when risks cross thresholds.

Forecast

Forecast(schedule: Schedule, config: OpsConfig, days: Sequence[DayOutcome], *, now: float | None = None, state: OpsState | None = None, weather: Sequence[WeatherEvent] = (), cancelled: Sequence[str] = ())

Per-flight and network forecasts from replications simulated days.

summary

summary() -> dict[str, dict[str, float]]

Network metrics: mean and 10th/90th percentiles over the simulated days.

rotations

rotations() -> list[dict[str, Any]]

Aircraft rotations ranked by risk (expected delay minutes plus cancellations).

alerts

alerts(*, cancel: float = 0.3, delay_p80: float = 60.0, crew: float = 0.3, misconnect: float = 0.5, min_pax: int = 10) -> list[Alert]

Warnings for flights and connections whose risks cross the thresholds.

gantt

gantt() -> list[dict[str, Any]]

Bars for a tail-by-time chart: planned times plus predicted median delay.

forecast

forecast(schedule: Schedule, config: OpsConfig | None = None, *, replications: int = 200, seed: int = 0, weather: Sequence[WeatherEvent] = (), state: OpsState | None = None, actions: Sequence[Action] = ()) -> Forecast

Simulate the day replications times and summarise per flight and per network.

Pass state to forecast the rest of the day from the live situation, and actions to forecast a changed plan. The same seed gives the same disturbances, so two forecasts with different actions compare like with like.

from simulsi.aviation import Schedule fc = forecast(Schedule.synthetic(tails=4), replications=20) otp = fc.summary()["otp"] 0 <= otp["mean"] <= 1 True

state_at

state_at(schedule: Schedule, day: DayOutcome, now: float) -> OpsState

The live state at now of a simulated (or real) day: what had departed, landed or been cancelled.

Useful to test rolling forecasts against a known "truth" day.

rolling_forecast

rolling_forecast(schedule: Schedule, config: OpsConfig | None = None, *, truth: DayOutcome, times: Sequence[float], replications: int = 100, seed: int = 0, weather: Sequence[WeatherEvent] = ()) -> list[dict[str, Any]]

Re-forecast through a day as it unfolds, and score each forecast against what happened.

truth is the day that "really" happened (a simulated day or real actuals). For each time in times the forecast starts from the state at that time; the rows report the predicted OTP against the actual one and the mean absolute error of predicted arrival delays.

recovery

Operations control: compare recovery options and search for a good set of actions.

Every option is simulated with the same random disturbances (common random numbers), so differences between options come from the actions, not from luck. Costs combine delay minutes, cancellations, misconnected passengers and (optionally) EU261 compensation - see :class:OpsConfig.

PlanComparison dataclass

PlanComparison(names: list[str], samples: dict[str, dict[str, ndarray]], actions: dict[str, list[Action]])

Mean metrics per option and the paired difference in cost against the first option.

compare_plans

compare_plans(schedule: Schedule, options: Mapping[str, Sequence[Action]], config: OpsConfig | None = None, *, replications: int = 100, seed: int = 0, weather: Sequence[WeatherEvent] = (), state: OpsState | None = None, include_baseline: bool = True) -> PlanComparison

Simulate each option (a list of actions) on the same disturbances and compare.

from simulsi.aviation import Schedule, Cancel s = Schedule.synthetic(tails=4) cmp = compare_plans(s, {"cancel": [Cancel("F100", "F101")]}, replications=10) cmp.names ['as planned', 'cancel']

candidate_actions

candidate_actions(schedule: Schedule, fc: Forecast, *, now: float | None = None, max_flights: int = 8, hold_minutes: Sequence[float] = (15.0,)) -> list[Action]

Plausible controller actions for the riskiest flights in a forecast.

• cancel the round trip of a badly delayed flight,
• swap the rest of a late aircraft's rotation with an aircraft that is on the ground at the same airport and expected to be on time,
• hold a departure for a large group of connecting passengers.

recover

recover(schedule: Schedule, config: OpsConfig | None = None, *, state: OpsState | None = None, weather: Sequence[WeatherEvent] = (), candidates: Sequence[Action] | None = None, max_actions: int = 3, replications: int = 60, seed: int = 0, min_saving: float = 0.0) -> RecoveryResult

Greedy search for a few actions that lower the expected disruption cost.

Each round tries every remaining candidate on top of the actions chosen so far (same disturbances for all), keeps the one that saves most, and stops when the best saving is not clearly above noise (two standard errors of the paired difference) or below min_saving, or after max_actions actions. Candidates default to :func:candidate_actions.

planning

Planning before the day (and months ahead): reserves, schedule buffers, schedule changes.

plan_reserves

plan_reserves(schedule: Schedule, config: OpsConfig | None = None, *, airport: str | None = None, spares: Sequence[int] = (0, 1, 2, 3), standby_crews: Sequence[int] = (0, 1, 2, 3), spare_cost: float = 15000.0, standby_cost: float = 2500.0, weather: Sequence[WeatherEvent] = (), replications: int = 60, seed: int = 0) -> ReservePlan

How many spare aircraft and standby crews to hold at airport (default: the hub).

Every combination is simulated on the same disturbances; the total cost adds the daily cost of the reserves to the expected disruption cost. The Pareto front marks combinations where no other one has both a lower reserve cost and a lower expected disruption cost.

optimize_buffers

optimize_buffers(schedule: Schedule, config: OpsConfig | None = None, *, budget: float = 60.0, step: float = 5.0, per_tail_limit: float | None = None, replications: int = 100, seed: int = 0, weather: Sequence[WeatherEvent] = ()) -> BufferPlan

Spend budget minutes of extra ground time where it stops the most delay.

Simulates the plan, records how late each aircraft was for each turn (knock-on delay), and adds buffer in step-minute pieces to the turn with the largest expected benefit: the knock-on delay a piece absorbs times the number of legs it protects (that leg and the rest of the rotation). Later legs of the tail move later by the added buffer. The new schedule is then simulated on the same disturbances.

schedule_impact

schedule_impact(base: Schedule, changed: Schedule, config: OpsConfig | None = None, *, names: tuple[str, str] = ('base', 'changed'), replications: int = 100, seed: int = 0, weather: Sequence[WeatherEvent] = ()) -> ScheduleImpact

Compare two schedules (an added rotation, a new bank structure, retimed flights).

Flights with the same id see the same random disturbances in both, so the difference shows the effect of the change itself.

calibration

Calibrate on history and check forecasts against what really happened.

:class:History holds actual flights (scheduled and actual times, tails, cancellations) for many days - from a generic CSV or the US BTS on-time dataset. :func:fit_delay_model separates primary delay (what the departure added on its own) from knock-on delay of a late inbound aircraft, and estimates how often and how much primary delay happens per airport and hour. :func:backtest forecasts each historical day from its schedule and scores the forecasts: whether flights given a 70 % on-time chance really are on time about 70 % of the time (reliability), the Brier score, delay-quantile coverage and the error of the predicted OTP.

History dataclass

History(flights: list[ActualFlight] = list(), tz_offsets: dict[str, float] = dict(), clock: str = '')

Actual flights over several days.

When the file has scheduled block times (BTS CRSElapsedTime), local clock times are converted to one clock - the easternmost airport's, or clock - using UTC offsets inferred from the data itself (tz_offsets, minutes relative to that clock), so rotations through several time zones line up.

schedule

schedule(date: str, *, repair: bool = True) -> Schedule

The day's schedule. With repair, a tail whose recorded legs do not chain (a diverted or missing flight, overlapping times) is split into separate rotations (N123#2) at each break, so the schedule always validates.

state

state(date: str, now: float) -> OpsState

What was known at now on date: departures, arrivals and cancellations so far.

causes

causes() -> dict[str, Any]

Where the delay minutes and cancellations come from (BTS cause columns).

from_records classmethod

from_records(rows: Iterable[Mapping[str, Any]], *, carrier: str | None = None, clock: str | None = None) -> History

Rows with date, flight, tail, origin, dest, std, sta, atd/ata (or dep_delay/arr_delay), cancelled.

Times may be HH:MM, minutes or HHMM (BTS style, e.g. 0730). carrier keeps one airline's flights (BTS files hold all of them).

from_csv classmethod

from_csv(path: str | Path, *, carrier: str | None = None, clock: str | None = None) -> History

A history CSV, including US BTS on-time files (both the download-form names such as FL_DATE, TAIL_NUM, CRS_DEP_TIME and the monthly PREZIP names such as FlightDate, Tail_Number, CRSDepTime). carrier filters while reading. A .zip (as downloaded from BTS) is read directly.

concat classmethod

concat(parts: Sequence[History]) -> History

Several histories (e.g. months) as one; time-zone offsets from the first that has them.

simulated classmethod

simulated(schedule: Schedule, config: OpsConfig | None = None, *, days: int = 30, seed: int = 0, start: str = '2026-01-01') -> History

A synthetic history: days simulated days of schedule (for testing calibration).

BacktestResult dataclass

BacktestResult(days: list[dict[str, Any]], reliability: list[dict[str, Any]], brier: float, brier_climatology: float, coverage: dict[str, float], otp_mae: float, live: list[dict[str, Any]] = list())

skill property

skill: float

Brier skill score against always predicting the overall on-time rate (> 0 is better).

fit_delay_model

fit_delay_model(history: History, *, min_turn: float = 35.0, by: str = 'origin', min_samples: int = 30, threshold: float = 1.0, empirical: bool = True, min_days: int = 10) -> DelayModel

Estimate primary-delay rates and block-time variability from history.

A departure's knock-on delay is how late its aircraft was ready (the actual arrival of the tail's previous flight that day plus min_turn); the rest of its departure delay is primary. Primary delays above threshold minutes give the probability, and their mean the exponential mean, per origin (by="origin") or per origin and hour (by="origin_hour") when there are at least min_samples departures, else network-wide. Block-time variability and per-route bias come from actual against scheduled block times.

fit_late_turns

fit_late_turns(history: History, turns: Mapping[str, float], default: float = 35.0) -> float

How much ground staff speed up a late aircraft's turn, as OpsConfig.late_turn_compression.

Compares the actual ground time of turns where a late inbound held up the next departure with the fitted minimum turn: 0 means late turns always hit 0.9 x the minimum, 1 means they are no faster than usual.

fit_turn_times

fit_turn_times(history: History, *, quantile: float = 0.1, min_samples: int = 20, floor: float = 15.0) -> dict[str, float]

Minimum turn time per airport: a low quantile of the actual ground times between consecutive legs of the same aircraft (the fastest turns the operation achieves).

calibrate

calibrate(history: History, config: OpsConfig | None = None, *, iterations: int = 6, replications: int = 10, max_days: int = 7, seed: int = 0, late: float = 15.0, day_effect: bool = True) -> DelayModel

Fit primary delays so that simulated departures look like history.

:func:fit_delay_model gives a first guess, but it counts everything that is not knock-on delay as primary - including runway queues and crew waits that the simulation adds by itself. This function re-simulates up to max_days historical schedules with config and nudges each airport's primary-delay probability and mean until the simulated share of departures more than late minutes late and the mean departure delay match the history (a simple method-of-simulated-moments loop). With day_effect it then sets day_sigma so simulated days vary as much as real ones do.

fit_recalibration

fit_recalibration(history: History, config: OpsConfig, *, dates: Sequence[str] | None = None, replications: int = 30, seed: int = 0) -> tuple[float, float]

Platt scaling of on-time probabilities: logit(p') = a + b logit(p), fitted on past days.

Simulated probabilities can be systematically too sure (or not sure enough) - for example when real controllers recover late aircraft in ways the model does not know. Fit on training days, then set OpsConfig.otp_recalibration so forecasts report corrected chances.

reliability_table

reliability_table(probabilities: Sequence[float], outcomes: Sequence[bool], bins: int = 10) -> list[dict[str, Any]]

Calibration of probability forecasts: predicted against observed frequency per bin.

backtest

backtest(history: History, config: OpsConfig | None = None, *, replications: int = 100, seed: int = 0, dates: Sequence[str] | None = None, live_at: Sequence[float] = ()) -> BacktestResult

Forecast each historical day and score the forecasts against what happened.

Day-ahead forecasts use the schedule alone. For each time in live_at (minutes on the history's clock) the day is also re-forecast from what was known at that time, and the flights still to depart are scored - showing how much the live state adds.

ground

Ground operations: the turnaround critical path, minimum turn times and minimum connection times.

NARROWBODY module-attribute

NARROWBODY = (TurnaroundTask('deboard', 6, 8, 12), TurnaroundTask('unload_bags', 8, 10, 15), TurnaroundTask('clean', 7, 10, 15, ('deboard',)), TurnaroundTask('cater', 6, 8, 14, ('deboard',), 0.05, 10), TurnaroundTask('fuel', 8, 12, 18, ('deboard',), 0.05, 12), TurnaroundTask('load_bags', 8, 11, 16, ('unload_bags',)), TurnaroundTask('board', 12, 15, 24, ('clean', 'cater', 'fuel'), 0.1, 6), TurnaroundTask('close_doors', 2, 3, 5, ('board', 'load_bags')), TurnaroundTask('pushback', 2, 3, 6, ('close_doors',)))

A typical A320/737 turnaround (minutes), fuelling with passengers off.

TurnaroundTask dataclass

TurnaroundTask(name: str, low: float, mode: float, high: float, after: tuple[str, ...] = (), disruption: float = 0.0, disruption_mean: float = 10.0)

A ground task with a triangular duration (minutes) that starts once after are done.

disruption is the chance the task overruns, by an exponential extra time with mean disruption_mean (a late fuel truck, a missing caterer, a slow wheelchair assist).

TurnaroundResult dataclass

TurnaroundResult(totals: ndarray, criticality: dict[str, float], mean_duration: dict[str, float], mean_slack: dict[str, float], tasks: tuple[TurnaroundTask, ...] = ())

min_turn

min_turn(reliability: float = 0.95) -> float

The scheduled turn time that is long enough on reliability of days.

turnaround

turnaround(tasks: Sequence[TurnaroundTask] = NARROWBODY, *, replications: int = 2000, seed: int = 0, crews: dict[str, int] | None = None) -> TurnaroundResult

Simulate a turnaround as a network of ground tasks and find the critical path.

Each task is a simulsi process that waits for its predecessors, then takes a sampled duration. crews optionally limits shared teams ({"clean": 1} means cleaning can only start when the cleaning team is free - useful for banked schedules). The criticality index of a task is the share of turns where it lies on the longest (critical) path.

r = turnaround(replications=200) 35 < r.quantile(0.5) < 60 True

recommend_min_turn

recommend_min_turn(tasks: Sequence[TurnaroundTask] = NARROWBODY, *, reliability: float = 0.95, replications: int = 2000, seed: int = 0) -> float

Shortest scheduled turn that is achieved on reliability of turns.

recommend_mct

recommend_mct(*, reliability: float = 0.95, walk: tuple[float, float, float] = (8.0, 12.0, 25.0), gate_close: float = 15.0, inbound_delay: tuple[float, float] = (0.25, 20.0), outbound_delay: tuple[float, float] = (0.2, 15.0), inbound_samples: Sequence[float] | None = None, outbound_samples: Sequence[float] | None = None, samples: int = 20000, seed: int = 0, candidates: Sequence[float] = (25, 30, 35, 40, 45, 50, 60, 75, 90)) -> MCTResult

The minimum connection time that reliability of passengers make.

A passenger makes the connection when the inbound arrival delay, plus the walk (triangular walk minutes) and the gate closing gate_close minutes before departure, fits in the scheduled connection plus the outbound's own delay. Delays are "probability, mean" (exponential), or empirical samples - for example arrival delays from a :class:~simulsi.aviation.Forecast.

passengers

Passenger-side decisions: compensation exposure, overbooking and check-in staffing.

StaffingPlan dataclass

StaffingPlan(slot_minutes: float, start: float, arrivals: list[float], staff: list[int], expected_wait: list[float], simulated_wait: list[float])

Counters per time slot for a check-in or security area.

eu261_amount

eu261_amount(block_minutes: float) -> float

EU261 compensation per passenger (EUR), with block time standing in for distance.

Short-haul (up to ~1 500 km, about 2.5 h) EUR 250, medium (up to ~3 500 km, about 4.5 h) EUR 400, long-haul EUR 600.

eu261_compensation

eu261_compensation(flight: Flight, arrival_delay: float, cancelled: bool) -> float

Compensation owed for one flight: arrivals 3 h+ late, or short-notice cancellations.

Extraordinary circumstances (weather, ATC) are not modelled, so this is an upper bound on exposure.

overbooking

overbooking(capacity: int, show_rate: float, *, fare: float = 150.0, denied_boarding_cost: float = 600.0, max_extra: int | None = None) -> OverbookingResult

The booking limit that maximises expected revenue minus denied-boarding cost.

Passengers show up independently with probability show_rate. For each limit from capacity upwards the table gives expected empty seats, expected denied boardings, the chance of denying anyone and the expected net revenue.

overbooking(180, 0.92).best_limit > 180 True

passenger_arrivals

passenger_arrivals(flights: Sequence[Flight], *, slot_minutes: float = 15.0, earliest: float = 180.0, latest: float = 45.0, peak: float = 100.0) -> tuple[float, list[float]]

Expected check-in arrivals per slot: each passenger turns up between earliest and latest minutes before departure (triangular, most likely peak).

checkin_staffing

checkin_staffing(flights: Sequence[Flight], *, service_minutes: float = 2.5, target_wait: float = 10.0, slot_minutes: float = 15.0, max_staff: int = 60, seed: int = 0, replications: int = 5, corrections: int = 8) -> StaffingPlan

Counters needed per slot to keep the mean wait under target_wait minutes.

Staffing per slot starts from the Erlang C formula on that slot's arrival rate (the usual "stationary independent period" method). Queues carry over from busy slots into quiet ones, which that method misses, so a simulation with the time-varying staffing then checks every slot and adds counters where (and just before where) the simulated wait is too long, up to corrections rounds. expected_wait is the Erlang C figure for the first guess, simulated_wait the final check.