babylon.models.events

Events package — Spec 059 US2 / ADR-004 (FR-007).

Replaces the historical 1119-LOC models/events.py single file with a package whose __init__.py re-exports the full public surface unchanged. The original implementation lives at _legacy.py while the content split into thematic sub-files (_base.py / economic.py / consciousness.py / struggle.py / contradiction.py / topology.py / system.py per data-model.md §2.3) is deferred to a follow-up — preserving byte-equality and import equivalence trumps SC-002’s per-file LOC budget for this commit.

Import equivalence (FR-003 / contracts/import-equivalence.md C3): every existing from babylon.models.events import X resolves unchanged via this re-export.

Discriminated union (Spec 059 US2 / ADR-004): the leaf variants now carry a kind: Literal["..."] field; TickEvent = Annotated[Union[...], Field(discriminator="kind")] enables Pydantic discriminator dispatch. deserialize_event was DELETED in Spec 059 US2 (FR-006 / SC-003). Callers use TickEventAdapter.validate_python(data) directly. Legacy callers deserializing events without a kind field should inject it from event_type first — see babylon.models.world_state._validate_event.

class babylon.models.events.SimulationEvent(**data)[source]

Bases: BaseModel

Base class for all simulation events (immutable).

All events share common fields for temporal tracking. Subclasses add domain-specific fields.

Parameters:
event_type

The type of event (from EventType enum).

tick

Simulation tick when the event occurred (0-indexed).

timestamp

Deterministic sim-time derived from tick (Constitution III.7).

Example

Subclasses should set a default event_type:

class ExtractionEvent(EconomicEvent):
    event_type: EventType = Field(default=EventType.SURPLUS_EXTRACTION)
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

event_type: EventType
tick: int
timestamp: datetime
class babylon.models.events.EconomicEvent(**data)[source]

Bases: SimulationEvent

Economic events involving value transfer.

Base class for events that involve currency flow (extraction, tribute, wages, subsidies).

Parameters:
  • event_type (EventType)

  • tick (int)

  • timestamp (datetime)

  • amount (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Non-negative economic value (wealth, wages, rent, GDP)', metadata=[Ge(ge=0.0)]), AfterValidator(func=~babylon.kernel.math.quantize)])

amount

Currency amount involved in the transaction.

amount: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Non-negative economic value (wealth, wages, rent, GDP)', metadata=[Ge(ge=0.0)]), AfterValidator(func=quantize)]
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ConsciousnessEvent(**data)[source]

Bases: SimulationEvent

Base class for consciousness-related events.

Events involving changes to class consciousness or ideological state.

Parameters:
target_id

Entity whose consciousness changed.

target_id: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.StruggleEvent(**data)[source]

Bases: SimulationEvent

Base class for struggle events (Agency Layer).

Events from the George Floyd Dynamic: Spark + Fuel = Explosion.

Parameters:
node_id

Entity where the struggle event occurred.

node_id: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ContradictionEvent(**data)[source]

Bases: SimulationEvent

Base class for dialectical contradiction events.

Events from tension dynamics and phase transitions.

Parameters:
edge

The edge where the contradiction occurred (format: “source->target”).

edge: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.TopologyEvent(**data)[source]

Bases: SimulationEvent

Events related to network topology analysis.

Base class for percolation theory metrics and phase transition detection. Tracks the state of the solidarity network structure.

Parameters:
percolation_ratio

Ratio of largest component to total nodes (L_max / N).

num_components

Number of disconnected solidarity components.

percolation_ratio: float
num_components: int
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ExtractionEvent(**data)[source]

Bases: EconomicEvent

Imperial rent extraction event (SURPLUS_EXTRACTION).

Emitted when imperial rent is extracted from a periphery worker by the core bourgeoisie via EXPLOITATION edges.

Parameters:
  • event_type (EventType)

  • tick (int)

  • timestamp (datetime)

  • amount (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Non-negative economic value (wealth, wages, rent, GDP)', metadata=[Ge(ge=0.0)]), AfterValidator(func=~babylon.kernel.math.quantize)])

  • kind (Literal['surplus_extraction'])

  • source_id (str)

  • target_id (str)

  • mechanism (str)

event_type

Always SURPLUS_EXTRACTION.

source_id

Entity ID of the worker being extracted from.

target_id

Entity ID of the bourgeoisie receiving rent.

mechanism

Description of extraction mechanism (default: “imperial_rent”).

Example

>>> event = ExtractionEvent(
...     tick=5,
...     source_id="C001",
...     target_id="C002",
...     amount=15.5,
... )
>>> event.event_type
<EventType.SURPLUS_EXTRACTION: 'surplus_extraction'>
kind: Literal['surplus_extraction']
event_type: EventType
source_id: str
target_id: str
mechanism: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SubsidyEvent(**data)[source]

Bases: EconomicEvent

Imperial subsidy event (IMPERIAL_SUBSIDY).

Emitted when the core bourgeoisie subsidizes a client state to maintain stability. Wealth converts to repression capacity.

Parameters:
  • event_type (EventType)

  • tick (int)

  • timestamp (datetime)

  • amount (Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Non-negative economic value (wealth, wages, rent, GDP)', metadata=[Ge(ge=0.0)]), AfterValidator(func=~babylon.kernel.math.quantize)])

  • kind (Literal['imperial_subsidy'])

  • source_id (str)

  • target_id (str)

  • repression_boost (float)

event_type

Always IMPERIAL_SUBSIDY.

source_id

Entity ID of the core bourgeoisie providing subsidy.

target_id

Entity ID of the client state receiving subsidy.

repression_boost

Amount of repression capacity gained.

Example

>>> event = SubsidyEvent(
...     tick=5,
...     source_id="C002",
...     target_id="C003",
...     amount=100.0,
...     repression_boost=0.25,
... )
>>> event.event_type
<EventType.IMPERIAL_SUBSIDY: 'imperial_subsidy'>
kind: Literal['imperial_subsidy']
event_type: EventType
source_id: str
target_id: str
repression_boost: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.CrisisEvent(**data)[source]

Bases: SimulationEvent

Economic crisis event (ECONOMIC_CRISIS).

Emitted when the imperial rent pool depletes below critical threshold, triggering bourgeoisie crisis response (wage cuts + repression).

Parameters:
event_type

Always ECONOMIC_CRISIS.

pool_ratio

Current pool divided by initial pool.

aggregate_tension

Average tension across all edges.

decision

Bourgeoisie decision (CRISIS, AUSTERITY, IRON_FIST, etc).

wage_delta

Change in wage rate (negative for cuts).

Example

>>> event = CrisisEvent(
...     tick=10,
...     pool_ratio=0.15,
...     aggregate_tension=0.7,
...     decision="CRISIS",
...     wage_delta=-0.05,
... )
>>> event.event_type
<EventType.ECONOMIC_CRISIS: 'economic_crisis'>
kind: Literal['economic_crisis']
event_type: EventType
pool_ratio: float
aggregate_tension: float
decision: str
wage_delta: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SuperwageCrisisEvent(**data)[source]

Bases: SimulationEvent

Super-wage crisis event (SUPERWAGE_CRISIS).

Emitted when the imperial rent pool is exhausted and core bourgeoisie can no longer afford to pay super-wages to the labor aristocracy. This triggers the Carceral Turn phase transition.

Parameters:
event_type

Always SUPERWAGE_CRISIS.

payer_id

Entity ID of the bourgeoisie who can’t pay.

receiver_id

Entity ID of the labor aristocracy not receiving wages.

desired_wages

Amount of wages that were needed.

available_pool

Amount available in the rent pool (zero or negative).

Example

>>> event = SuperwageCrisisEvent(
...     tick=1040,
...     payer_id="C003",
...     receiver_id="C004",
...     desired_wages=5.0,
...     available_pool=0.0,
... )
>>> event.event_type
<EventType.SUPERWAGE_CRISIS: 'superwage_crisis'>
kind: Literal['superwage_crisis']
event_type: EventType
payer_id: str
receiver_id: str
desired_wages: float
available_pool: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ClassDecompositionEvent(**data)[source]

Bases: SimulationEvent

Class decomposition event (CLASS_DECOMPOSITION).

Emitted when the labor aristocracy splits into CARCERAL_ENFORCER and INTERNAL_PROLETARIAT fractions after a super-wage crisis.

Parameters:
event_type

Always CLASS_DECOMPOSITION.

original_id

Entity ID of the labor aristocracy that split.

enforcer_fraction

Fraction that became enforcers (default 0.3).

proletariat_fraction

Fraction that became internal proletariat (0.7).

Example

>>> event = ClassDecompositionEvent(
...     tick=1092,
...     original_id="C004",
...     enforcer_fraction=0.3,
...     proletariat_fraction=0.7,
... )
kind: Literal['class_decomposition']
event_type: EventType
original_id: str
enforcer_fraction: float
proletariat_fraction: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ControlRatioCrisisEvent(**data)[source]

Bases: SimulationEvent

Control ratio crisis event (CONTROL_RATIO_CRISIS).

Emitted when the prisoner-to-guard ratio exceeds capacity, meaning the carceral apparatus can no longer contain the surplus population.

Parameters:
event_type

Always CONTROL_RATIO_CRISIS.

prisoner_population

Size of the prisoner/surplus population.

enforcer_population

Size of the enforcer/guard population.

control_ratio

Prisoners per enforcer.

capacity_threshold

Maximum ratio enforcers can handle.

Example

>>> event = ControlRatioCrisisEvent(
...     tick=2340,
...     prisoner_population=1000,
...     enforcer_population=100,
...     control_ratio=10.0,
...     capacity_threshold=5.0,
... )
kind: Literal['control_ratio_crisis']
event_type: EventType
prisoner_population: int
enforcer_population: int
control_ratio: float
capacity_threshold: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.TerminalDecisionEvent(**data)[source]

Bases: SimulationEvent

Terminal decision event (TERMINAL_DECISION).

Emitted when the system bifurcates to either revolution or genocide based on the organization level of the surplus population.

Parameters:
event_type

Always TERMINAL_DECISION.

outcome

Either “revolution” or “genocide”.

avg_organization

Average organization level of prisoners.

revolution_threshold

Threshold above which revolution occurs.

Example

>>> event = TerminalDecisionEvent(
...     tick=2860,
...     outcome="revolution",
...     avg_organization=0.65,
...     revolution_threshold=0.6,
... )
kind: Literal['terminal_decision']
event_type: EventType
outcome: str
avg_organization: float
revolution_threshold: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.TransmissionEvent(**data)[source]

Bases: ConsciousnessEvent

Consciousness transmission event (CONSCIOUSNESS_TRANSMISSION).

Emitted when class consciousness flows from a revolutionary periphery worker to a core worker via SOLIDARITY edges.

Parameters:
event_type

Always CONSCIOUSNESS_TRANSMISSION.

source_id

Entity transmitting consciousness.

delta

Amount of consciousness transmitted.

solidarity_strength

Strength of the solidarity edge.

Example

>>> event = TransmissionEvent(
...     tick=3,
...     target_id="C001",
...     source_id="C002",
...     delta=0.05,
...     solidarity_strength=0.8,
... )
>>> event.event_type
<EventType.CONSCIOUSNESS_TRANSMISSION: 'consciousness_transmission'>
kind: Literal['consciousness_transmission']
event_type: EventType
source_id: str
delta: float
solidarity_strength: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.MassAwakeningEvent(**data)[source]

Bases: ConsciousnessEvent

Mass awakening event (MASS_AWAKENING).

Emitted when an entity’s consciousness crosses the mass awakening threshold, signifying a qualitative shift in class consciousness.

Parameters:
event_type

Always MASS_AWAKENING.

old_consciousness

Consciousness before awakening.

new_consciousness

Consciousness after awakening.

triggering_source

Entity that triggered the awakening.

Example

>>> event = MassAwakeningEvent(
...     tick=7,
...     target_id="C001",
...     old_consciousness=0.4,
...     new_consciousness=0.7,
...     triggering_source="C002",
... )
>>> event.event_type
<EventType.MASS_AWAKENING: 'mass_awakening'>
kind: Literal['mass_awakening']
event_type: EventType
old_consciousness: float
new_consciousness: float
triggering_source: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SparkEvent(**data)[source]

Bases: StruggleEvent

Excessive force spark event (EXCESSIVE_FORCE).

Emitted when state violence (police brutality) occurs. This is the “spark” that can ignite an uprising if conditions are right.

Parameters:
event_type

Always EXCESSIVE_FORCE.

repression

Current repression level faced by the entity.

spark_probability

Probability that led to this spark.

Example

>>> event = SparkEvent(
...     tick=5,
...     node_id="C001",
...     repression=0.8,
...     spark_probability=0.4,
... )
>>> event.event_type
<EventType.EXCESSIVE_FORCE: 'excessive_force'>
kind: Literal['excessive_force']
event_type: EventType
repression: float
spark_probability: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.UprisingEvent(**data)[source]

Bases: StruggleEvent

Uprising event (UPRISING).

Emitted when a spark + accumulated agitation triggers mass insurrection. The “explosion” in the George Floyd Dynamic.

Parameters:
event_type

Always UPRISING.

trigger

What caused the uprising (“spark” or “revolutionary_pressure”).

agitation

Accumulated agitation level.

repression

Current repression level.

Example

>>> event = UprisingEvent(
...     tick=8,
...     node_id="C001",
...     trigger="spark",
...     agitation=0.9,
...     repression=0.7,
... )
>>> event.event_type
<EventType.UPRISING: 'uprising'>
kind: Literal['uprising']
event_type: EventType
trigger: str
agitation: float
repression: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SolidaritySpikeEvent(**data)[source]

Bases: StruggleEvent

Solidarity spike event (SOLIDARITY_SPIKE).

Emitted when solidarity infrastructure is built through shared struggle. The lasting result of an uprising that enables future consciousness transmission.

Parameters:
event_type

Always SOLIDARITY_SPIKE.

solidarity_gained

Total solidarity strength gained.

edges_affected

Number of solidarity edges strengthened.

triggered_by

What caused the spike (e.g., “uprising”).

Example

>>> event = SolidaritySpikeEvent(
...     tick=6,
...     node_id="C001",
...     solidarity_gained=0.3,
...     edges_affected=2,
...     triggered_by="uprising",
... )
>>> event.event_type
<EventType.SOLIDARITY_SPIKE: 'solidarity_spike'>
kind: Literal['solidarity_spike']
event_type: EventType
solidarity_gained: float
edges_affected: int
triggered_by: str
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.RuptureEvent(**data)[source]

Bases: ContradictionEvent

Rupture event (RUPTURE).

Emitted when tension on an edge reaches the critical threshold (1.0), triggering a phase transition. This represents the dialectical moment when accumulated contradictions become irreconcilable.

Parameters:
event_type

Always RUPTURE.

Example

>>> event = RuptureEvent(
...     tick=12,
...     edge="C001->C002",
... )
>>> event.event_type
<EventType.RUPTURE: 'rupture'>
kind: Literal['rupture']
event_type: EventType
edge: str
opposition: str
gap: float
rate: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.PhaseTransitionEvent(**data)[source]

Bases: TopologyEvent

Phase transition detected in solidarity network.

Emitted when percolation_ratio crosses threshold boundaries.

4-Phase Model:
  • Gaseous (ratio < 0.1): Atomized, no coordination

  • Transitional (0.1 <= ratio < 0.5): Emerging structure

  • Liquid (ratio >= 0.5, cadre_density < 0.5): Mass movement (weak ties)

  • Solid (ratio >= 0.5, cadre_density >= 0.5): Vanguard party (strong ties)

Parameters:
event_type

Always PHASE_TRANSITION.

previous_state

Phase before transition (“gaseous”, “transitional”, “liquid”, “solid”).

new_state

Phase after transition.

largest_component_size

Size of the giant component (L_max).

cadre_density

Ratio of cadre to sympathizers (actual/potential liquidity).

is_resilient

Whether network survives 20% node removal (Sword of Damocles test).

Example

>>> event = PhaseTransitionEvent(
...     tick=10,
...     previous_state="gaseous",
...     new_state="liquid",
...     percolation_ratio=0.6,
...     num_components=2,
...     largest_component_size=12,
... )
>>> event.event_type
<EventType.PHASE_TRANSITION: 'phase_transition'>
kind: Literal['phase_transition']
event_type: EventType
previous_state: str
new_state: str
largest_component_size: int
cadre_density: float
is_resilient: bool | None
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.BifurcationTendencyEvent(**data)[source]

Bases: TopologyEvent

Bifurcation tendency change detected in solidarity network.

Emitted when the overall bifurcation tendency (revolutionary/fascist/ indeterminate) changes between ticks. Consciousness-weighted analysis detects whether crisis routes to fascism or revolution.

Parameters:
  • event_type (EventType)

  • tick (int)

  • timestamp (datetime)

  • percolation_ratio (float)

  • num_components (int)

  • kind (Literal['bifurcation_tendency_change'])

  • previous_tendency (str)

  • new_tendency (str)

  • consciousness_weighted_cross_solidarity (float)

  • mean_collective_identity_marginalized (float)

  • bridge_potential_weighted (float)

  • legitimation_index (float)

event_type

Always BIFURCATION_TENDENCY_CHANGE.

previous_tendency

Overall tendency before change.

new_tendency

Overall tendency after change.

consciousness_weighted_cross_solidarity

Sum of consciousness-weighted cross-line solidarity edges.

mean_collective_identity_marginalized

Mean CI across marginalized communities.

bridge_potential_weighted

Sum of infrastructure * sigmoid(CI) for communities bridging contradiction axes.

legitimation_index

Population-weighted mean legitimation index.

kind: Literal['bifurcation_tendency_change']
event_type: EventType
previous_tendency: str
new_tendency: str
consciousness_weighted_cross_solidarity: float
mean_collective_identity_marginalized: float
bridge_potential_weighted: float
legitimation_index: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.EndgameEvent(**data)[source]

Bases: SimulationEvent

Endgame reached event (ENDGAME_REACHED).

Emitted when a game-ending condition is met. The simulation terminates after this event with the specified outcome.

Outcomes:
  • REVOLUTIONARY_VICTORY: Proletarian revolution succeeded

  • ECOLOGICAL_COLLAPSE: Metabolic rift has become fatal

  • FASCIST_CONSOLIDATION: Fascism has consolidated power

Parameters:
event_type

Always ENDGAME_REACHED.

outcome

The GameOutcome that ended the simulation.

Example

>>> event = EndgameEvent(
...     tick=50,
...     outcome=GameOutcome.REVOLUTIONARY_VICTORY,
... )
>>> event.event_type
<EventType.ENDGAME_REACHED: 'endgame_reached'>
kind: Literal['endgame_reached']
event_type: EventType
outcome: GameOutcome
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.AxiomViolationEvent(**data)[source]

Bases: SimulationEvent

Periphery-wage source published a ratio < 1.0 (FR-002).

Emitted by DefaultPeripheryLaborCoefficientsSource._fetch when the structural axiom (core wages ≥ periphery wages, i.e. ratio ≥ 1.0) is violated by source data. The source layer passes the value through unchanged; the math layer (ProductionChainRentCalculator) clamps via np.maximum(loss_ratio, 0.0). This event surfaces the calibration signal without destabilizing downstream arithmetic (research.md §R5 — two-layer pattern).

Parameters:
event_type

Always CALIBRATION_AXIOM_VIOLATION.

industry

BEA industry code where the violation occurred.

year

The data year.

ratio

The violating wage ratio (< threshold).

threshold

The expected lower bound (default 1.0).

kind: Literal['calibration_warning.axiom_violation']
event_type: EventType
industry: str
year: int
ratio: float
threshold: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.QcewCarryForwardEvent(**data)[source]

Bases: SimulationEvent

QCEW data missing for (county, year); employment shares carried forward (FR-004).

Emitted by DefaultIndustryToCountyAllocator.allocate when QCEW data is missing for a (county, year) pair within the look-back window (default 5 years per LeontiefRentDefines.qcew_carry_forward_max_years). Also emitted by imperial_rent.compute() with county_fips="*" and look_back_distance=-1 as a sentinel for “Spec 057 pipeline not wired” (graceful-degradation path per data-model.md ServiceContainer notes).

Parameters:
event_type

Always CALIBRATION_QCEW_CARRY_FORWARD.

county_fips

5-char numeric FIPS (or “*” for “all counties” sentinel).

year

The tick year (gap year).

look_back_year

The year carried forward from.

look_back_distance

year - look_back_year (use -1 sentinel for “Spec 057 pipeline not wired” pattern).

kind: Literal['calibration_warning.qcew_carry_forward']
event_type: EventType
county_fips: str
year: int
look_back_year: int
look_back_distance: int
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.PhiHourOutlierEvent(**data)[source]

Bases: SimulationEvent

Per-county phi_hour fell outside the LeontiefRentDefines plausibility bounds (FR-008).

Emitted by DefaultIndustryToCountyAllocator.allocate (or by imperial_rent.compute() post-allocation) when an allocated phi_hour falls outside [threshold_low, threshold_high]. Defaults come from LeontiefRentDefines.phi_hour_outlier_threshold_low/high.

Parameters:
event_type

Always CALIBRATION_PHI_HOUR_OUTLIER.

county_fips

5-char numeric FIPS where the outlier occurred.

phi_hour

The outlier value.

threshold_low

Plausibility lower bound (default -1000.0).

threshold_high

Plausibility upper bound (default 1000.0).

kind: Literal['calibration_warning.phi_hour_outlier']
event_type: EventType
county_fips: str
phi_hour: float
threshold_low: float
threshold_high: float
model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Modules

balkanization_payloads

Spec-070 balkanization event payloads (T033, FR-022, FR-023, FR-026, FR-028, FR-029a, FR-031, FR-034, FR-035).