Events

Event Base Class

class econsimulacra.events.base.EventManager(event_names, events_dic, prng=None, registered_classes=[])[source]

Bases: object

Parameters:

event_names (list[str])
events_dic (dict[str, dict[str, Any]])
prng (Optional[random.Random])
registered_classes (list[Type])

trigger_events_after_step(time_step, env)[source]

Checks and triggers events based on the current time step.

Parameters:

time_step (int) – The current time step in the environment.
env (Environment) – The environment instance where the events will be executed.

Return type:

None

Note

See also: econsimulacra.envs.base.Environment.step() where this method is called after each step.

trigger_events_after_log(log, env)[source]

Checks and triggers events based on the provided log.

Parameters:

log (Log) – The log instance that may trigger events.
env (Environment) – The environment instance where the events will be executed.

Return type:

None

class econsimulacra.events.base.Event(trigger, config)[source]

Bases: object

Event class.

To add a custom event:

Define a subclass implementing execute:

class MyEvent(Event):
    def execute(self, env: Environment, log: Optional[Log] = None) -> None:
        # implement event logic here
        pass

Add the event to the simulation config:

{
    "simulation": {
        "events": ["myEvent"]
    },
    "myEvent": {
        "type": "MyEvent",
        "trigger": {"at": [1, 5, 10]}
    }
}

Register the event class with the environment before running:
```
env.register_classes([MyEvent])
```
Run the simulation and observe the event being triggered at the specified steps.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

execute(env, log=None)[source]

Parameters:

env (Environment)
log (Optional[Log])

Return type:

None

class econsimulacra.events.base.EventTrigger(config, registered_classes, prng)[source]

Bases: object

EventTrigger class that defines the conditions for triggering events.

Parameters:

config (dict[str, Any])
registered_classes (list[Type])
prng (random.Random)

check_trigger_after_step(time_step)[source]

Checks whether the event should be triggered based on the current time step.

Parameters:: time_step (int)
Return type:: bool

check_trigger_after_log(log)[source]

Checks whether the event should be triggered based on the provided log.

Parameters:: log (Log)
Return type:: bool

Built-in Events

class econsimulacra.events.constant_salary.ConstantSalary(trigger, config)[source]

Bases: Event

Constant salary class.

This event provides a constant salary to the specified agents at regular intervals. The salary amount is determined by the configuration and does not depend on any other factors.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

execute(env, log=None)[source]

Provide a constant salary to the specified agents at regular intervals.

Parameters:

env (Environment[Any]) – The environment in which the event is executed.
log (Optional[Log]) – The log that triggered the event. It should be an instance of AgentGenerationLog. If provided, the salary for the agent will be stored. If not provided, the salary will be paid to all agents based on the stored salaries.

Return type:

None

class econsimulacra.events.constant_supply.ConstantSupply(trigger, config)[source]

Bases: Event

Constant supply class.

This event provides a constant supply to the specified agents at regular intervals. The supply amount is determined by the configuration and does not depend on any other factors.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

execute(env, log=None)[source]

Provide a constant supply to the specified agents at regular intervals.

Parameters:

env (Environment[Any]) – The environment in which the event is executed.
log (Optional[Log]) – The log that triggered the event. It should be an instance of AgentGenerationLog. If provided, the supply for the agent will be stored. If not provided, the supply will be added to all agents based on the stored supplies.

Return type:

None

class econsimulacra.events.dynamic_supply.DynamicSupply(trigger, config)[source]

Bases: Event

Supply inventories according to time-dependent configuration entries.

The event follows the same two-phase lifecycle as ConstantSupply: AgentGenerationLog calls capture the initial inventories, while step calls apply the supply selected for the environment’s current display time.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

execute(env, log=None)[source]

Capture initial inventory or apply the currently selected supply.

Parameters:

env (Environment[Any]) – Environment whose agent inventories are updated.
log (Optional[Log]) – Agent-generation log during initialization, otherwise None.

Returns:

None.

Return type:

None

Note

A time not covered by supplies performs no inventory mutation.

class econsimulacra.events.keep_out.KeepOut(trigger, config)[source]

Bases: Event

Make configured grid cells non-traversable during scheduled periods.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

execute(env, log=None)[source]

Apply or restore traversal access for the current time.

Parameters:

env (Environment[Any]) – Environment containing the target grid space.
log (Optional[Log]) – Unused log argument required by the event interface.

Returns:

None.

Return type:

None

Note

Agents already inside a newly closed cell are not displaced. They may leave because access checks apply to the destination cell.

class econsimulacra.events.consumption_tax.ConsumptionTax(trigger, config)[source]

Bases: Event

Consumption tax class.

This event levies a consumption tax on all orders. When an order is executed, the buyer pays an additional tax amount to the government, calculated as tax_amount = execution_amount * price * tax_rate.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

get_current_tax_rate(current_time)[source]

Parameters:: current_time (int | str)
Return type:: float

execute(env, log=None)[source]

Impose a consumption tax on the order and transfer the tax amount from the buyer to the government.

Parameters:

env (Environment[Any]) – The environment in which the event is executed.
log (Optional[Log]) – The log that triggered the event. It should be an instance of OrderReactionLog.

Return type:

None

class econsimulacra.events.subsidy4specific_order.Subsidy4SpecificOrder(trigger, config)[source]

Bases: Event

Subsidy for specific orders class.

This event provides subsidies to the agents who purchase specific items during specific periods. The subsidy amount is calculated as subsidy_amount = accept_amount * subsidy_rate.

Parameters:

trigger (EventTrigger)
config (dict[str, Any])

get_current_subsidy_rate(current_time)[source]

Parameters:: current_time (int | str)
Return type:: float

execute(env, log=None)[source]

Provide subsidies to the agents who purchase specific items during specific periods.

Parameters:

env (Environment[Any]) – The environment in which this event is executed.
log (Optional[Log]) – The log that triggers this event. It should be an OrderReactionLog.

Return type:

None