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:
BaseModelBase class for all simulation events (immutable).
All events share common fields for temporal tracking. Subclasses add domain-specific fields.
- 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].
- class babylon.models.events.EconomicEvent(**data)[source]
Bases:
SimulationEventEconomic events involving value transfer.
Base class for events that involve currency flow (extraction, tribute, wages, subsidies).
- Parameters:
- 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:
SimulationEventBase class for consciousness-related events.
Events involving changes to class consciousness or ideological state.
- target_id
Entity whose consciousness changed.
- 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:
SimulationEventBase class for struggle events (Agency Layer).
Events from the George Floyd Dynamic: Spark + Fuel = Explosion.
- node_id
Entity where the struggle event occurred.
- 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:
SimulationEventBase class for dialectical contradiction events.
Events from tension dynamics and phase transitions.
- edge
The edge where the contradiction occurred (format: “source->target”).
- 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:
SimulationEventEvents 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.
- 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:
EconomicEventImperial 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'>
- 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:
EconomicEventImperial 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'>
- 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:
SimulationEventEconomic 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'>
- 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:
SimulationEventSuper-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'>
- 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:
SimulationEventClass 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, ... )
- 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:
SimulationEventControl 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, ... )
- 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:
SimulationEventTerminal 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, ... )
- 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:
ConsciousnessEventConsciousness 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'>
- 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:
ConsciousnessEventMass 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'>
- 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:
StruggleEventExcessive 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'>
- 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:
StruggleEventUprising 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'>
- 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:
StruggleEventSolidarity 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'>
- 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:
ContradictionEventRupture 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'>
- 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:
TopologyEventPhase 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'>
- 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:
TopologyEventBifurcation 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.
- 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:
SimulationEventEndgame 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 (EventType)
tick (int)
timestamp (datetime)
kind (Literal['endgame_reached'])
outcome (GameOutcome)
- 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'>
- 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:
SimulationEventPeriphery-wage source published a ratio < 1.0 (FR-002).
Emitted by
DefaultPeripheryLaborCoefficientsSource._fetchwhen 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 vianp.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).
- 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:
SimulationEventQCEW data missing for (county, year); employment shares carried forward (FR-004).
Emitted by
DefaultIndustryToCountyAllocator.allocatewhen QCEW data is missing for a (county, year) pair within the look-back window (default 5 years perLeontiefRentDefines.qcew_carry_forward_max_years). Also emitted byimperial_rent.compute()withcounty_fips="*"andlook_back_distance=-1as 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).
- 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:
SimulationEventPer-county phi_hour fell outside the LeontiefRentDefines plausibility bounds (FR-008).
Emitted by
DefaultIndustryToCountyAllocator.allocate(or byimperial_rent.compute()post-allocation) when an allocatedphi_hourfalls outside[threshold_low, threshold_high]. Defaults come fromLeontiefRentDefines.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).
- model_config: ClassVar[ConfigDict] = {'frozen': True}
Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].
Modules
Spec-070 balkanization event payloads (T033, FR-022, FR-023, FR-026, FR-028, FR-029a, FR-031, FR-034, FR-035). |