Agents

Agent Base Class

class econsimulacra.agents.base.Agent(agent_id, agent_name, env_service_dic, prng=None, config=None)[source]

Bases: ABC, Generic[ObsT]

Agent class.

Once you define the agent class inheriting this agent ABC class, environment automatically generate agents using the class you define. act(self, obs: ObsT) method is the only method that must be implemented in the agent class you define, and it will be called by environment at each step to get the action of the agent.

Parameters:
get_self_name()[source]

Get the name of the agent.

Return type:

str

self_assign_name(config)[source]

Assign a name to the agent based on the configuration.

If “name” is not provided in the config, the agent name will be the default name assigned by environment.

Parameters:

config (dict[str, Any])

Return type:

None

get_inventory()[source]

Get the inventory of the agent.

Return type:

dict[str, float | int]

get_item_amount(item_name)[source]

Get the amount of a specific item in the agent’s inventory.

Parameters:

item_name (str)

Return type:

float | int

abstractmethod async act(obs)[source]

Perform an action based on the observation (abstract method).

Parameters:

obs (ObsT)

Return type:

dict[str, Any]

exchange_goods(get_item_name=None, get_item_amount=None, give_item_name=None, give_item_amount=None)[source]

Exchange goods in the agent’s inventory.

Parameters:

  • get_item_name (Optional[str]) – the name of the item to get (optional).

  • get_item_amount (Optional[float | int]) – the amount of the item to get (optional, must be provided if get_item_name is provided).

  • give_item_name (Optional[str]) – the name of the item to give (optional).

  • give_item_amount (Optional[float | int]) – the amount of the item to give (optional, must be provided if give_item_name is provided).

Return type:

None

provide_info4all_agents()[source]

provide information for all agents.

Returns:

a list of information keys that the agent can provide for all agents.

Return type:

list[str]

Note

Usually called by observation providers registered in econsimulacra.environment.base._build_observation_registry. Currently, the following built-in observation provider is supported: econsimulacra.environment.base._obs_others_pos If another agent is requesting “others_pos” information (i.e., “others_pos” is in the self.request_obses of the another agent), the agent can provide its position information to them by adding “self_pos” in self.info4all_agents.

provide_info4co_located_agents()[source]

provide information for those agents who are co-located.

Returns:

a list of information keys that the agent can provide for those agents who are co-located.

Return type:

list[str]

Note

Usually called by observation providers registered in econsimulacra.environment.base._build_observation4co_located_agents_registry. Currently, the following built-in observation provider is supported: econsimulacra.environment.base._obs_others_inventory If another agent who is co-located with the agent is requesting “others_inventory” informaation (i.e., “others_inventory” is in the self.request_obses of the another agent), the agent can provide its inventory information to them by adding “inventory” in self.info4co_located_agents.

provide_info4allowed_agents()[source]

provide information for those agents who are allowed.

Returns:

a list of information keys that the agent can provide for those agents who are is_rich_info_allowed.

Return type:

list[str]

Note

Usually called by observation providers registered in econsimulacra.environment.base._build_observation4allowed_agents_registry. Currently, built-in observation provider is not supported, but users can implement their own observation provider and register it in econsimulacra.environment.base._build_observation4allowed_agents_registry to provide rich information for those agents who are is_rich_info_allowed.

request_obs()[source]

Request observations from the environment.

Returns:

a list of observation keys that the agent is requesting from the environment.

Return type:

list[str]

See also

econsimulacra.environment.base.get_observations

LLM Agent

class econsimulacra.agents.llm_agent.LLMAgent(agent_id, agent_name, env_service_dic, prng=None, config=None)[source]

Bases: Agent[dict[str, Any]]

LLM agent that uses a language model to generate actions based on observations.

The LLMAgent relies on the environment services to provide an LLM client, a prompt builder, and optionally a persona builder. The agent constructs prompts based on the current observation and persona (if applicable), sends the prompt to the LLM client, and returns the generated response as its action.

Parameters:
self_assign_name(config)[source]

Assign name to the agent, potentially using the persona builder if available.

Parameters:

config (dict) – Configuration dictionary for the agent. This may include a “personaConfig” key with configuration for the persona builder.

See also

econsimulacra.llm_services.personas.base.PersonaBuilder

async act(obs)[source]

Generate an action based on the current observation.

Parameters:

obs (dict) – The current observation for the agent. The structure of this dictionary depends on the environment.

Returns:

The action generated by the agent based on the current observation.

The generated action is applied in the environment.

Return type:

dict

Note

This method constructs a prompt by combining the persona prompt (if a persona builder is available) and the observation prompt (constructed by the prompt builder). It then sends the prompt to the LLM client to generate a response.

Government Agent

class econsimulacra.agents.government.Government(agent_id, agent_name, env_service_dic, prng=None, config=None)[source]

Bases: Agent[dict[str, Any]]

Rule-based government agent.

This class only posts tweets based on a predefined schedule and does not interact with other agents or the environment in any other way. The policy must be defined separately by the user as “events”.

Parameters:
self_assign_name(config)[source]

Assign the agent’s name from the config.

To easily determine the name of the government agent, we require that the config contains a “name” field without adding agent_id.

Parameters:

config (dict[str, Any])

Return type:

None

time_to_tweet(current_time)[source]

Return the scheduled tweet.

Parameters:

current_time (str | int)

Return type:

str

async act(obs)[source]

Post a tweet if there is one scheduled for the current time step.

Parameters:

obs (dict[str, Any])

Return type:

dict[str, Any]

Auto-Reactor Agent

class econsimulacra.agents.auto_reacter.AutoReactLLMAgent(agent_id, agent_name, env_service_dic, prng=None, config=None)[source]

Bases: LLMAgent

An LLMAgent that automatically reacts to all incoming orders and proposals.

This wrapper enforces deterministic acceptance of transactional intents (i.e., incoming_orders and incoming_proposals) posterior to, or independently of, the LLM-generated decision. The primary objective is to eliminate non-responsive or economically irrational behaviors (e.g., ignoring valid orders) that often arise from stochastic LLM outputs in supply-side agents such as retailers and restaurants.

In socio-economic EconSimulacra, supply-side agents are expected to process transactions reliably. However, vanilla LLMAgents may omit reactions due to prompt misalignment or token truncation. This wrapper guarantees that all valid incoming transactional requests are accepted, thereby preserving simulation consistency and preventing deadlocks in the market mechanism.

Parameters:
async act(obs)[source]

Automatically react to all incoming orders and proposals.

Get the LLM-generated response and then overwrite the reactions field to ensure that all incoming transactional intents are accepted.

Parameters:

obs (dict[str, Any]) – The observation dictionary containing the current state of the agent, including any incoming orders and proposals.

Returns:

A dictionary representing the agent’s action,

with the reactions field modified to accept all incoming transactional intents.

Return type:

dict

judge_reaction(incoming_transactional_intent, current_inventory, is_order)[source]

Judge whether to react to an incoming transactional intent based on the current inventory.

Parameters:

  • incoming_transactional_intent (dict[str, Any]) – A dictionary representing the incoming order or proposal that the agent is evaluating.

  • current_inventory (dict[str, float | int]) – A dictionary representing the agent’s current inventory before reacting to the intent.

  • is_order (bool) – A boolean indicating whether the incoming transactional intent is an order (True) or a proposal (False).

Returns:

True if the agent should react to the incoming transactional intent,

False otherwise.

Return type:

bool

Note

Always True as long as the agent has sufficient inventory to fulfill the intent.

Rule-Based Household

class econsimulacra.agents.households.household.RuleBasedHousehold(agent_id, agent_name, env_service_dic, prng=None, config=None, decision_policy=None, supplemental_policies=[])[source]

Bases: Agent[dict[str, Any]]

EconSimulacra adapter for the complete rule-based household.

The final action combines the core fragment and every supplemental fragment:

\[A_t=\Pi_E(a_t^{\rm core}) \oplus\bigoplus_{j=1}^{m}\Pi_E(a_{t,j}^{\rm supplemental}),\]

where \(\Pi_E\) removes disabled keys and \(\oplus\) is implemented by _compose_fragments().

Usage: add a custom action

class WeatherTweetPolicy:
    def decide(self, context, state):
        if context.hour == 8:
            return {"tweet": "Good morning."}
        return {}


class SocialHousehold(RuleBasedHousehold):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.add_supplemental_policy(WeatherTweetPolicy())


simulator.register_classes([SocialHousehold])

Set the agent config type to SocialHousehold. A custom scalar key must not conflict with another fragment, and any key listed in disabledActions is removed before composition.

Parameters:
build_decision_policy()[source]

Build the default core decision policy.

Returns:

Policy wired with default physiology, mobility, shopping, and action capabilities.

Return type:

HouseholdDecisionPolicy

add_supplemental_policy(policy)[source]

Register one custom policy after existing supplemental policies.

Parameters:

policy (SupplementalPolicy) – Object implementing synchronous or asynchronous decide(context, state).

Return type:

None

async act(obs)[source]

Orchestrate one household decision.

Parameters:

obs (dict[str, Any]) – EconSimulacra observation mapping for the current step.

Returns:

Composed, capability-filtered EconSimulacra action mapping.

Return type:

dict[str, Any]

\[O_t\to o_t\to x_t\to (a_t^{\rm core},a_t^{\rm supplemental,*}) \to A_t.\]
property mode: str

Return the current activity mode for diagnostics.

get_time_step(obs)[source]

Extract the canonical environment step with legacy fallbacks.

Parameters:

obs (dict[str, Any]) – Raw observation that may contain time_step or numeric time.

Returns:

Integer simulation step.

Return type:

int

Note

Counting calls is the last resort because sleeping and non-walking agents can legitimately skip calls to act().

get_current_hour(obs, time_step)[source]

Extract wall-clock hour with a step-derived compatibility fallback.

Parameters:

  • obs (dict[str, Any]) – Raw observation that may contain an ISO display time.

  • time_step (int) – Canonical environment step for fallback calculation.

Returns:

Local hour in the interval [0, 24).

Return type:

float

Note

Display time is authoritative when it is a valid ISO string. This keeps circadian decisions aligned after sleep or automatic travel skips calls to act().

Household Decision Policy

class econsimulacra.agents.households.policy.ActionCapabilities(disabled=frozenset({}))[source]

Bases: object

Represent the action keys enabled for an agent.

EconSimulacra configurations name unavailable actions in disabledActions. Let \(\mathcal{A}\) be the universe of action keys and \(\mathcal{D}\) the configured disabled set. The feasible set is

\[\mathcal{A}^{\mathrm{enabled}}=\mathcal{A}\setminus\mathcal{D}.\]

Core policies query this object before mutating state. filter() is a second boundary check for reactions and third-party supplemental policies. Unknown names are retained in \(\mathcal{D}\) and filtered literally, which permits future EconSimulacra action types without changing this class.

Parameters:

disabled (frozenset[str]) – Immutable set of exact disabled action-dictionary keys.

disabled: frozenset[str] = frozenset({})
classmethod from_config(config)[source]

Construct capabilities from an agent configuration.

Parameters:

config (dict[str, Any]) – Agent configuration that may contain disabledActions.

Returns:

Immutable action capabilities.

Raises:

TypeError – If disabledActions is a string rather than a sequence.

Return type:

ActionCapabilities

is_enabled(action_key)[source]

Return whether an action key may be emitted.

Parameters:

action_key (str) – Exact EconSimulacra action-dictionary key.

Returns:

True when the key is not disabled.

Return type:

bool

filter(action_fragment)[source]

Remove disabled keys from an action fragment.

Parameters:

action_fragment (dict[str, Any]) – Partial action mapping to restrict.

Returns:

New mapping containing only enabled keys.

Return type:

dict[str, Any]

For fragment \(a\), the returned mapping is the restriction \(a|_{\mathcal{A}^{\mathrm{enabled}}}\). The input is not mutated.

class econsimulacra.agents.households.policy.DecisionSignals(can_move, can_sleep, can_consume, can_order, should_sleep, can_eat, should_eat)[source]

Bases: object

Precomputed feasibility and demand signals for one decision.

Parameters:

  • can_move (bool) – Whether move is enabled.

  • can_sleep (bool) – Whether sleep_duration is enabled.

  • can_consume (bool) – Whether consumptions is enabled.

  • can_order (bool) – Whether orders is enabled.

  • should_sleep (bool) – Whether enabled sleep demand has reached onset.

  • can_eat (bool) – Whether consumption is enabled and food is available.

  • should_eat (bool) – Whether enabled scheduled-meal demand is active.

can_move: bool
can_sleep: bool
can_consume: bool
can_order: bool
should_sleep: bool
can_eat: bool
should_eat: bool
class econsimulacra.agents.households.policy.HouseholdDecisionPolicy(physiology, mobility, shopping, capabilities=None)[source]

Bases: object

Apply physiological and activity priorities to produce one core action.

This class is the sole arbiter among sleep, food, shopping, and mobility demands. The priority ordering is an explicit design assumption; it is not claimed as an empirical utility maximization result.

Parameters:

  • physiology (PhysiologyModel) – Sleep, hunger, and meal policy.

  • mobility (MobilityModel) – Movement action adapter.

  • shopping (ShoppingModel) – Replenishment, store-choice, and order policy.

  • capabilities (Optional[ActionCapabilities]) – Enabled-action policy. All actions are enabled by default.

decide(context, state)[source]

Return the highest-priority feasible household action.

The policy evaluates guards in the following strict lexicographic order and executes only the first applicable core branch:

  1. Environment reports that the household is already sleeping: emit no core action.

  2. Sleep and movement are enabled and sleep is due away from home: move home.

  3. Consumption is enabled, hunger is critical, and food exists: eat at the current position.

  4. Sleep is enabled and due at home: sleep.

  5. Movement is enabled and a return-home trip is active and incomplete: continue it.

  6. Consumption is enabled, a meal is due, and food exists: eat at the current position.

  7. Orders are enabled, shopping is due, budget is positive, and no destination is active: choose a store. If movement is disabled, only co-located stores belong to the choice set.

  8. A store trip is active: order on arrival and depart for home in the same action when movement is enabled, move toward the store when necessary, or cancel an infeasible trip.

  9. Movement is enabled and the household remains away from home: move

    home.

  10. Otherwise: normalize the current home/away mode and emit no core action.

Formally, let \(g_k(o_t,x_t)\in\{0,1\}\) denote applicability of branch \(k\) in that list and let \(a_k(o_t,x_t)\) be its state transition and action fragment. The selected branch is

\[k_t^*=\min\{k\in\{1,\ldots,10\}:g_k(o_t,x_t)=1\},\qquad a_t^{\mathrm{core}}=a_{k_t^*}(o_t,x_t).\]

Branch 10 is unconditional, so the set is nonempty. Sleep demand, meal demand, food feasibility, shopping demand, and budget are defined by PhysiologyModel.sleep_due(), PhysiologyModel.meal_due(), PhysiologyModel.can_eat(), ShoppingPolicy.shopping_due(), and ShoppingPolicy.budget(). Critical hunger means \(G_t\geq G_{\mathrm{critical}}\). Home equality is exact tuple equality \(p_t=h\). If \(e_a=\mathbf{1}\{a\notin\mathcal{D}\}\) is the capability indicator for action key \(a\), every branch that emits \(a\) also includes \(e_a=1\) in its guard. Thus a disabled high-priority demand cannot block a lower-priority feasible branch.

Parameters:
Returns:

Exactly one core action fragment, possibly empty.

Return type:

dict[str, Any]

Before evaluating the priority branches, any co-located seller offers update this household’s store-specific price and availability beliefs.

get_signals(context, state)[source]

Calculate action feasibility and physiological demand once.

Parameters:
Returns:

Immutable signals shared by all priority rules this step.

Return type:

DecisionSignals

class econsimulacra.agents.households.policy.SupplementalPolicy(*args, **kwargs)[source]

Bases: Protocol

Protocol for optional actions such as social-network behavior.

A supplemental policy observes the same context and mutable state as the core policy and returns an action fragment. It is evaluated after the core household decision, then combined by :class:RuleBasedHousehold._compose_fragments. The default list contains ProposalReactionPolicy; additional policies can add independent behavior such as SNS actions.

Implementations do not need to inherit this Protocol; providing a compatible synchronous or asynchronous decide() method is sufficient.

Usage

class HungerTweetPolicy:
    def decide(self, context, state):
        if state.hunger >= 0.8:
            return {"tweet": "I am getting hungry."}
        return {}


class SocialHousehold(RuleBasedHousehold):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.add_supplemental_policy(HungerTweetPolicy())


simulator.register_classes([SocialHousehold])

The corresponding agent configuration uses "type": "SocialHousehold". Disabled action keys returned by a custom policy are removed before composition. Custom policies should normally read rather than mutate state so that core behavior remains independent.

decide(context, state)[source]

Return an action fragment to merge with the core decision.

Parameters:

  • context (DecisionContext) – Normalized current observation and time.

  • state (HouseholdState) – Mutable household state after the core policy has run.

Returns:

Partial EconSimulacra action mapping. Return an empty mapping when the supplemental policy has no action for this step.

Return type:

dict[str, Any] | Awaitable[dict[str, Any]]

class econsimulacra.agents.households.policy.SocialMediaPolicy(config, prng, capabilities=None, tweet_renderer=None)[source]

Bases: _SocialMediaPolicyRules, SupplementalPolicy

Generate a complete supplemental SNS action fragment.

Parameters:

  • config (dict[str, Any]) – socialRule household configuration.

  • prng (Random) – Shared seeded pseudo-random generator.

  • capabilities (Optional[ActionCapabilities]) – Household action capabilities used to avoid infeasible decisions and unnecessary Tiny LM calls.

  • tweet_renderer (Optional[TweetRenderer]) – Optional asynchronous renderer. It is required only when tweet actions are enabled.

Follow, unfollow, tweet-event occurrence, topic, sentiment, and style are selected by rules. The renderer controls wording only. This policy is asynchronous solely because text generation may take time; existing synchronous supplemental policies remain valid.

async decide(context, state)[source]

Generate one supplemental social-network action fragment.

Parameters:
Returns:

Mapping containing zero or more of follow, unfollow, and tweet.

Return type:

dict[str, Any]

class econsimulacra.agents.households.policy.ProposalReactionPolicy[source]

Bases: SupplementalPolicy

React to proposals from other agents.

This policy is the default supplemental policy for all households. It observes the same context and mutable state as the core policy and returns an action fragment. It is evaluated after the core household decision, then combined by :class:RuleBasedHousehold._compose_fragments.

decide(context, state)[source]

Return a proposal-reaction action fragment.

Parameters:

  • context (DecisionContext) – Normalized current observation and time.

  • state (HouseholdState) – Mutable household state after the core policy has run.

Returns:

Partial EconSimulacra action mapping. Return an empty mapping when no proposal reactions are applicable.

Return type:

dict[str, Any]

Household Social-Media Rules

class econsimulacra.agents.households.social.TweetIntent(topic, sentiment, style, memory_excerpt)[source]

Bases: object

Describe tweet content before a language model realizes its wording.

Parameters:

  • topic (Literal['daily_life', 'consumption', 'shopping', 'finances', 'health', 'mobility', 'social']) – Rule-selected subject of the tweet.

  • sentiment (Literal['positive', 'neutral', 'negative']) – Rule-selected emotional polarity.

  • style (Literal['casual', 'informative', 'reflective', 'enthusiastic', 'terse']) – Rule-selected writing style.

  • memory_excerpt (str) – Relevant summarized memory supplied as grounding.

The intent is deliberately model-independent. A text generator may express it in natural language, but it must not choose whether to tweet or change the topic, sentiment, or style.

topic: Literal['daily_life', 'consumption', 'shopping', 'finances', 'health', 'mobility', 'social']
sentiment: Literal['positive', 'neutral', 'negative']
style: Literal['casual', 'informative', 'reflective', 'enthusiastic', 'terse']
memory_excerpt: str
class econsimulacra.agents.households.social.SocialDecision(follow_agent_id=None, unfollow_agent_id=None, tweet_intent=None)[source]

Bases: object

Represent one rule-based social-network decision.

Parameters:

  • follow_agent_id (int | None) – Recommended agent to follow, if any.

  • unfollow_agent_id (int | None) – Currently followed agent to unfollow, if any.

  • tweet_intent (TweetIntent | None) – Structured tweet specification, if a tweet event occurs.

follow_agent_id: int | None = None
unfollow_agent_id: int | None = None
tweet_intent: TweetIntent | None = None

Household Tweet Renderer

class econsimulacra.agents.households.tweet_renderer.TweetTextGenerator(*args, **kwargs)[source]

Bases: Protocol

Protocol implemented by an asynchronous tweet text generator.

async generate_text(prompt)[source]

Generate plain text for one supplied prompt.

Parameters:

prompt (str) – Fully constructed generation prompt.

Returns:

Generated plain text.

Return type:

str

class econsimulacra.agents.households.tweet_renderer.TweetRenderer(text_generator, language='English', max_characters=140)[source]

Bases: object

Render a rule-selected tweet intent with a small language model.

Parameters:

  • text_generator (TweetTextGenerator) – Service that generates plain text asynchronously.

  • language (str) – Output language named in the prompt.

  • max_characters (int) – Maximum emitted tweet length.

The renderer controls wording only. Follow decisions, tweet timing, topic, sentiment, and style have already been fixed by rules before this class is called.

async generate_tweet(intent, previous_tweet=None)[source]

Generate and sanitize one tweet from a structured intent.

Parameters:

  • intent (TweetIntent) – Rule-selected tweet specification.

  • previous_tweet (str | None) – Most recent tweet used for exact duplicate removal.

Returns:

Sanitized tweet, or None after empty or duplicate generation.

Return type:

str | None

generate_prompt(intent)[source]

Generate a compact grounding prompt for a Tiny LM.

Parameters:

intent (TweetIntent) – Rule-selected content specification.

Returns:

Prompt that asks only for surface realization of the intent.

Return type:

str

Household State and Context

class econsimulacra.agents.households.states.HouseholdState(sleep_pressure, hunger, last_meal_elapsed, home=None, destination=None, mode='HOME', last_step=None, has_been_sleeping=False)[source]

Bases: object

Mutable state shared by the household’s policy components.

Parameters:

  • sleep_pressure (float) – Homeostatic sleep stock \(H_t\).

  • hunger (float) – Bounded hunger stock \(G_t\).

  • last_meal_elapsed (float) – Hours since the last meal \(M_t\).

  • home (tuple[int, ...] | None) – Fixed initial position \(h\), or None before first act.

  • destination (tuple[int, ...] | None) – Active movement destination \(d_t\), if any.

  • mode (Literal['HOME', 'AWAY', 'SLEEPING', 'TRAVEL_STORE', 'WAITING_ORDER', 'RETURN_HOME', 'RETURN_HOME_SLEEP', 'RETURN_HOME_MEAL']) – Finite-state activity label \(z_t\).

  • last_step (int | None) – Last processed simulation step \(\ell_t\), if any.

  • sleeping_last_interval – Whether the preceding interval was asleep.

  • has_been_sleeping (bool)

The private state at decision step \(t\) is

\[x_t=(H_t,G_t,M_t,h,d_t,z_t,\ell_t,\sigma_t).\]
sleep_pressure: float
hunger: float
last_meal_elapsed: float
home: tuple[int, ...] | None = None
destination: tuple[int, ...] | None = None
mode: Literal['HOME', 'AWAY', 'SLEEPING', 'TRAVEL_STORE', 'WAITING_ORDER', 'RETURN_HOME', 'RETURN_HOME_SLEEP', 'RETURN_HOME_MEAL'] = 'HOME'
last_step: int | None = None
has_been_sleeping: bool = False
class econsimulacra.agents.households.states.SocialState(hawkes_excitation=0.0, last_hawkes_step=None, last_memory_snapshot=<factory>, changed_memory_keys=(), empty_timeline_steps=<factory>, last_follow_step=None, last_unfollow_step=None, last_tweet_intent=None)[source]

Bases: object

Store private state used by the household social-media policy.

Parameters:

  • hawkes_excitation (float) – Current excitation above the baseline intensity.

  • last_hawkes_step (Optional[int]) – Last step at which excitation was decayed.

  • last_memory_snapshot (dict[str, Any]) – Previous relevant summarized-memory values.

  • changed_memory_keys (tuple[str, ...]) – Memory categories changed at the current step.

  • empty_timeline_steps (dict[int, int]) – Consecutive empty-tweet observations by followee.

  • last_follow_step (Optional[int]) – Most recent step at which follow was selected.

  • last_unfollow_step (Optional[int]) – Most recent step at which unfollow was selected.

  • last_tweet_intent (Optional[TweetIntent]) – Most recently realized tweet intent.

Tweet occurrence state is separate from HouseholdState because the Hawkes process and graph-management rules are supplemental behavior and do not alter sleep, hunger, mobility, or shopping state.

hawkes_excitation: float = 0.0
last_hawkes_step: int | None = None
last_memory_snapshot: dict[str, Any]
changed_memory_keys: tuple[str, ...] = ()
empty_timeline_steps: dict[int, int]
last_follow_step: int | None = None
last_unfollow_step: int | None = None
last_tweet_intent: TweetIntent | None = None
class econsimulacra.agents.households.states.DecisionContext(obs, time_step, hour, current_pos, inventory)[source]

Bases: object

Normalized observation supplied to every household policy.

Parameters:

  • obs (dict[str, Any]) – Original EconSimulacra observation \(O_t\).

  • time_step (int) – Nonnegative simulation step \(t\).

  • hour (float) – Local clock time \(\tau_t\in[0,24)\).

  • position – Current grid position \(p_t\).

  • inventory (dict[str, float]) – Numeric self-inventory mapping \(I_t\).

  • current_pos (tuple[int, ...])

The adapter computes

\[\tau_t=(\tau_0+t\Delta)\bmod 24,\]

where \(\tau_0\) is startHour and \(\Delta\) is stepHours.

obs: dict[str, Any]
time_step: int
hour: float
current_pos: tuple[int, ...]
inventory: dict[str, float]

Household Stylized Models

class econsimulacra.agents.households.stylized_models.PhysiologyModel(config, food_items, step_hours)[source]

Bases: object

Update sleep pressure and hunger and create physiological actions.

This is a stylized discrete-time model inspired by the exponential sleep Process S and sinusoidal Process C (Borbély, 1982), plus empirical circadian meal timing (de Castro, 1987). It is not a calibrated reproduction.

References

Parameters:
initialize_state()[source]

Create the initial household state.

Returns:

State with configured initial sleep pressure and hunger, and an infinite elapsed meal interval.

Return type:

HouseholdState

\[H_0=H_{\mathrm{init}},\quad G_0=G_{\mathrm{init}},\quad M_0=\infty.\]
update_state(state, step)[source]

Advance physiological stocks to step.

Parameters:

  • state (HouseholdState) – Mutable household state to update in place.

  • step (int) – Current nonnegative simulation step.

Return type:

None

For elapsed hours \(\delta_t\), lower and upper asymptotes \(L<U\), and time constants \(\theta_s,\theta_w>0\),

\[\begin{split}H_t=\begin{cases} L+(H_{t-1}-L)e^{-\delta_t/\theta_s},&\sigma_{t-1}=1,\\ U-(U-H_{t-1})e^{-\delta_t/\theta_w},&\sigma_{t-1}=0. \end{cases}\end{split}\]

Hunger evolves as

\[G_t=\min\{1,G_{t-1}+\rho_t\delta_t\},\qquad M_t=M_{t-1}+\delta_t,\]

using the configured awake or asleep rate \(\rho_t\).

should_sleep(state, current_hour)[source]

Test the sleep-onset condition.

Parameters:

  • state (HouseholdState) – Household state containing current sleep pressure.

  • current_hour (float) – Current local clock hour in [0, 24).

Returns:

True when homeostatic plus circadian pressure reaches onset.

Return type:

bool

\[C(\tau)=A\cos\left(\frac{2\pi(\tau-\phi)}{24}\right),\qquad D_t^{\mathrm{sleep}}=\mathbf{1}\{H_t+C(\tau_t)\geq\eta_{\rm on}\}.\]
generate_sleep_action(state, current_hour)[source]

Enter sleep mode and construct a sleep action.

Parameters:

  • state (HouseholdState) – Mutable household state to mark as sleeping.

  • current_hour (float) – Current local clock hour used to predict wake time.

Returns:

Action containing the predicted integer sleep_duration.

Return type:

dict[str, Any]

For maximum duration \(N\), the chosen duration is the first \(n\in\{1,\ldots,N\}\) at which projected pressure reaches the wake threshold, or \(N\) if no such step exists.

should_eat(state, current_hour)[source]

Test the scheduled-meal condition.

Parameters:

  • state (HouseholdState) – Household state containing hunger and elapsed meal time.

  • current_hour (float) – Current local clock hour in [0, 24).

Returns:

True when minimum spacing and weighted meal demand both hold.

Return type:

bool

\[S(\tau)=\max_{\mu\in\mathcal K} \exp\left[-\frac{d_{24}(\tau,\mu)^2}{2w^2}\right],\]
\[D_t^{\mathrm{meal}}=\mathbf{1}\{M_t\geq m_{\min}\ \land\ G_t+\omega S(\tau_t)\geq\eta_m\}.\]
can_eat(inventory)[source]

Test whether at least one configured food is available.

Parameters:

inventory (dict[str, float]) – Current self-inventory keyed by item name.

Returns:

True when any configured food has a positive amount.

Return type:

bool

generate_eat_action(state, inventory)[source]

Allocate one meal and update hunger.

Parameters:

  • state (HouseholdState) – Mutable household state to update after consumption.

  • inventory (dict[str, float]) – Current self-inventory keyed by item name.

Returns:

consumptions records for positive food quantities, or an empty mapping when no positive quantity can be consumed.

Return type:

dict[str, Any]

For food share \(s_i\) and energy per unit \(e_i\),

\[c_{i,t}=\min\left\{I_t(i), \frac{\max(0,G_t)s_i}{\max(e_i,10^{-12})}\right\}.\]
class econsimulacra.agents.households.stylized_models.ShoppingModel(config, food_items, cash_name, prng, shopping_items=None)[source]

Bases: object

Manage replenishment, store choice, budgeting, and order settlement.

The replenishment trigger is an \((s,S)\)-style rule inspired by the classical inventory policy (Scarf, 1960). Here it is a household heuristic, not the solution of Scarf’s dynamic program: reorder points and target stocks are fixed configuration values and the basket is cash constrained.

Store choice uses a multinomial logit over observed stores, following the random-utility form associated with conditional logit (McFadden, 1974). Every household maintains seller-specific expected prices and availability probabilities, updates them after co-located observations, and values coverage according to item-specific replenishment importance. Travel is embedded in an activity sequence (home, shopping, home), a minimal stylization of activity-schedule demand models (Bowman & Ben-Akiva, 2001).

All stochastic choice uses the injected random.Random instance. A fixed seed, observation order, configuration, and initial state therefore reproduce the same choice sequence.

References

Parameters:
should_shop(inventory)[source]

Return whether any stock is at or below its reorder point.

Parameters:

inventory (dict[str, float]) – Current self-inventory keyed by item name.

Returns:

Whether at least one purchase target has reached its reorder point.

Return type:

bool

For purchase targets \(\mathcal{I}\) and configured reorder point \(s_i\),

\[D_t^{\mathrm{shop}}= \mathbf{1}\{\exists i\in\mathcal{I}:I_t(i)\leq s_i\}.\]
get_budget(inventory)[source]

Return cash available after reserve and basket-share constraints.

Parameters:

inventory (dict[str, float]) – Current self-inventory containing the configured cash.

Returns:

Nonnegative maximum basket expenditure.

Return type:

float

Let \(Y_t=I_t(c)\) be holdings of configured cash item \(c\), \(R\geq0\) the cash reserve, and \(\alpha\geq0\) the maximum basket share. The shopping budget is

\[B_t=\max\{0,\min(Y_t-R,\alpha Y_t)\}.\]
get_stores(context)[source]

Return every visible non-household agent as a store candidate.

Parameters:

context (DecisionContext) – Observation context containing optional others_pos.

Returns:

Visible sellers in observation order. Example:

[
    {
        "agent_id": 10,
        "agent_name": "MarketEast",
        "is_household": False,
        "pos": (2, 3),
    },
    {
        "agent_id": 12,
        "agent_name": "MarketWest",
        "is_household": False,
        "pos": (8, 4),
    },
]

Return type:

list[dict[str, Any]]

Seller order is preserved from others_pos because it is also the deterministic tie order used by store choice.

update_beliefs(context)[source]

Update store-specific beliefs from co-located seller observations.

Parameters:

context (DecisionContext) – Observation containing optional others_inventory.

Return type:

None

For seller \(j\), item \(i\), learning rate \(\rho\in[0,1]\), observed posted price \(p_{ij}\) and availability indicator \(a_{ij}\in\{0,1\}\), beliefs follow

\[\hat p_{ij}\leftarrow(1-\rho_p)\hat p_{ij}+\rho_p p_{ij}, \qquad \hat a_{ij}\leftarrow(1-\rho_a)\hat a_{ij}+\rho_a a_{ij}.\]

An item is observed available when the seller reports a positive or masked amount. A missing item is observed unavailable. Price is updated only when the item is offered. Beliefs are held by this model instance and therefore are not shared across households.

choose_store(current_pos, stores, inventory)[source]

Choose a store with an importance-weighted belief logit.

Parameters:

  • current_pos (tuple[int, ...]) – Household’s current grid position.

  • stores (list[dict[str, Any]]) – Ordered candidate-store observation records.

  • inventory (dict[str, float]) – Current household inventory.

Returns:

Sampled store record, or None for an empty choice set.

Return type:

dict[str, Any] | None

Let \(q_i=\max(0,T_i-I_t(i))\) be desired replenishment, \(w_i\geq0\) item importance, \(\bar p_i\) the household price prior, \(\hat a_{ij}\in[0,1]\), \(\hat p_{ij}\) the household’s availability and price beliefs for store \(j\), and \(B_t>0\) the household’s shopping budget. The importance-weighted desired quantity is

\[W_t=\sum_i w_iq_i.\]

For \(W_t>0\), expected coverage and price savings relative to the household prior are

\[C_{tj}=\frac{\sum_iw_iq_i\hat a_{ij}}{W_t},\qquad R_{tj}=\frac{\sum_iw_iq_i\hat a_{ij} (\bar p_i-\hat p_{ij})/\bar p_i}{W_t}.\]

Expected basket expenditure and budget pressure are

\[E_{tj}=\sum_iq_i\hat a_{ij}\max(0,\hat p_{ij}),\qquad L_{tj}=E_{tj}/B_t.\]
\[D_{tj}=\max_k|p_{t,k}-d_{j,k}|,\qquad V_{tj}=\beta_P R_{tj}+\beta_A C_{tj} -\beta_B L_{tj}-\beta_D D_{tj}.\]

A store is sampled with

\[P(J_t=j\mid o_t,x_t)= \frac{\exp(V_{tj})}{\sum_{r\in\mathcal{J}_t}\exp(V_{tr})}.\]

Because \(L_{tj}\) divides by the household-specific budget, the same expensive basket deters a budget-constrained household more than a wealthy one. Unavailable goods contribute no artificial price saving and reduce coverage, so a store cannot look cheap merely because it is expected to carry few needed goods. Larger itemImportance makes shortages of necessities contribute more strongly than shortages of preference goods. Numerically, max(V) is subtracted before exponentiation. An empty choice set, nonpositive budget, or basket with zero weighted need returns None.

generate_order_action(context)[source]

Build a budget-feasible basket from a co-located seller.

Parameters:

context (DecisionContext) – Observation containing current inventory and co-located seller inventories.

Returns:

Order records, or an empty mapping when no order is feasible.

Return type:

dict[str, Any]

The first matching seller in others_inventory is used. Items are processed by descending itemImportance with configured order as the stable tie break, so a tight budget fills necessities first. Let \(p_i>0\) be observed price (zero or negative price removes the affordability bound), \(A_i\) observed seller stock when numeric and \(+\infty\) otherwise, and \(B_1=B_t\). Then

\[q_i^*=\min\{\max(0,T_i-I_t(i)),A_i\},\]
\[\begin{split}q_i=\begin{cases} \min\{q_i^*,B_i/p_i\},&p_i>0,\\ q_i^*,&p_i\leq0, \end{cases} \qquad B_{i+1}=B_i-p_iq_i.\end{split}\]

Positive \(q_i\) values become EconSimulacra orders with TTL 2. EconSimulacra processes orders before move in the same action dictionary and exposes resulting inventory in the next observation; therefore no settlement-wait state is retained. An empty seller set or basket produces no action.

class econsimulacra.agents.households.stylized_models.MobilityModel[source]

Bases: object

Translate an activity destination into EconSimulacra movement.

Destination choice belongs to the activity policy; this component only adapts a chosen position to the environment action schema. Reissuing the destination also makes movement intent explicit if environment-side motion persistence changes.

select_mobility(context)[source]

Select a mobility mode from the current observation.

Parameters:

context (DecisionContext) – Normalized current observation.

Returns:

Name of the mobility mode to use.

Return type:

str

Note

An active journey retains its current mode when that mode remains available. A new journey uses the mode with the highest effective velocity. Walking preserves compatibility when mobility observations are absent.

generate_move_action(context, state, destination, mode)[source]

Persist intent and reissue the destination for this step.

Parameters:

  • context (DecisionContext) – Normalized current observation used to select mobility.

  • state (HouseholdState) – Mutable household state in which to persist movement intent.

  • destination (tuple[int, ...]) – Target grid position.

  • mode (Literal['HOME', 'AWAY', 'SLEEPING', 'TRAVEL_STORE', 'WAITING_ORDER', 'RETURN_HOME', 'RETURN_HOME_SLEEP', 'RETURN_HOME_MEAL']) – Activity-mode label associated with the trip.

Returns:

move action targeting destination with its mobility mode.

Return type:

dict[str, Any]

Note

Availability and effective velocity are supplied by the environment; ownership and fuel requirements are not recalculated here.

For selected destination \(d\) and activity label \(z\), the state transition and action are

\[(d_t,z_t,\sigma_t)\leftarrow(d,z,0),\qquad a_t=\{\mathtt{move}:d,\mathtt{mobility}:m\}.\]
class econsimulacra.agents.households.stylized_models.ProposalReactionModel[source]

Bases: object

Default model for unsolicited proposals.

This is a boundary policy rather than a behavioral model: households do not initiate swaps and deterministically reject every observed proposal.

generate_reactions(context, state)[source]

Reject every incoming swap proposal.

Parameters:

  • context (DecisionContext) – Observation containing optional incoming_proposals.

  • state (HouseholdState) – Household state; unused because rejection is stateless.

Returns:

Proposal-rejection reactions, or an empty mapping.

Return type:

dict[str, Any]