Memory

History Items

class econsimulacra.memory.memory_items.InvalidActionHistoryItem(action_type, description, time, time_step)[source]

Bases: object

A class representing an invalid action history item in the agent’s memory.

Parameters:

action_type (str)
description (str)
time (int | str)
time_step (int)

action_type

the type of the invalid action.

Type:: str

description

the description of the invalid action.

Type:: str

time

the time of the invalid action.

Type:: int | str

time_step

the time step of the invalid action.

Type:: int

class econsimulacra.memory.memory_items.ConsumptionHistoryItem(item_name, quantity, time, time_step)[source]

Bases: object

A class representing a consumption history item in the agent’s memory.

Parameters:

item_name (str)
quantity (int | float)
time (int | str)
time_step (int)

item_name

the name of the consumed item.

Type:: str

quantity

the quantity of the consumed item.

Type:: int | float

time

the time of the consumption.

Type:: int | str

time_step

the time step of the consumption.

Type:: int

Note

This history item is generated based on the ConsumptionLog. See also: econsimulacra.logs.base.ConsumptionLog, econsimulacra.envs.base.Environment._consume_items(agent_id, consumptions)

class econsimulacra.memory.memory_items.MoveHistoryItem(pos, init_pos, time, time_step)[source]

Bases: object

A class representing a movement history item in the agent’s memory.

Parameters:

pos (tuple[int, ...])
init_pos (tuple[int, ...])
time (str | int | None)
time_step (int)

pos

the position of the agent after the movement.

Type:: tuple[int, …]

init_pos

the initial position of the agent assigned by the environment.

Type:: tuple[int, …]

time

the time of the movement. It can be None for the initial position assigned by the environment, which is based on the SpaceAssignLog.

Type:: int | str, optional

Note

This history item is generated based on the MoveLog and SpaceAssignLog. See also: econsimulacra.logs.base.MoveLog, econsimulacra.logs.base.SpaceAssignLog, econsimulacra.envs.base.Environment._move(agent_id, new_pos), econsimulacra.envs.base.Environment._assign_agent_to_space(agent_id, coords)

class econsimulacra.memory.memory_items.PurchaseHistoryItem(item_name, quantity, price, time, time_step, from_agent_id)[source]

Bases: object

A class representing a purchase history item in the agent’s memory.

Parameters:

item_name (str)
quantity (int | float)
price (int | float)
time (int | str)
time_step (int)
from_agent_id (int)

item_name

the name of the purchased item.

Type:: str

quantity

the quantity of the purchased item.

Type:: int | float

price

the price of the purchased item.

Type:: int | float

time

the time of the purchase.

Type:: int | str

from_agent_id

the id of the agent from whom the item is purchased.

Type:: int

Note

This history item is generated based on the OrderReactionLog where the agent is the purchase agent and the reaction is accept. See also: econsimulacra.logs.base.OrderReactionLog, econsimulacra.envs.base.Environment._process_reactions(agent_id, reactions)

class econsimulacra.memory.memory_items.InnerThoughtHistoryItem(inner_thought, time, time_step)[source]

Bases: object

A class representing an inner thought history item in the agent’s memory.

Parameters:

inner_thought (str)
time (int | str)
time_step (int)

inner_thought

the content of the inner thought.

Type:: str

time

the time of the inner thought.

Type:: int | str

time_step

the time step of the inner thought.

Type:: int

agent_id

the id of the agent generating the inner thought.

Type:: int

Note

This history item is generated based on the InnerThoughtLog. See also: econsimulacra.logs.base.InnerThoughtLog, econsimulacra.envs.base.Environment._process_inner_thought(agent_id, inner_thought)

class econsimulacra.memory.memory_items.SaleHistoryItem(item_name, quantity, price, time, time_step, to_agent_id)[source]

Bases: object

A class representing a sale history item in the agent’s memory.

Parameters:

item_name (str)
quantity (int | float)
price (int | float)
time (int | str)
time_step (int)
to_agent_id (int)

item_name

the name of the sold item.

Type:: str

quantity

the quantity of the sold item.

Type:: int | float

price

the price of the sold item.

Type:: int | float

time

the time of the sale.

Type:: int | str

to_agent_id

the id of the agent to whom the item is sold.

Type:: int

Note

This history item is generated based on the OrderReactionLog where the agent is the sale agent and the reaction is accept. See also: econsimulacra.logs.base.OrderReactionLog, econsimulacra.envs.base.Environment._process_reactions(agent_id, reactions)

class econsimulacra.memory.memory_items.ExchangeHistoryItem(give_item_name, give_item_quantity, get_item_name, get_item_quantity, time, time_step, counterparty_id)[source]

Bases: object

A class representing an exchange history item in the agent’s memory.

Parameters:

give_item_name (str)
give_item_quantity (int | float)
get_item_name (str)
get_item_quantity (int | float)
time (int | str)
time_step (int)
counterparty_id (int)

give_item_name

the name of the item given in the exchange.

Type:: str

give_item_quantity

the quantity of the item given in the exchange.

Type:: int | float

get_item_name

the name of the item received in the exchange.

Type:: str

get_item_quantity

the quantity of the item received in the exchange.

Type:: int | float

time

the time of the exchange.

Type:: int | str

counterparty_id

the id of the agent with whom the exchange is made.

Type:: int

Note

This history item is generated based on the ProposalReactionLog where the reaction is accept. See also: econsimulacra.logs.base.ProposalReactionLog, econsimulacra.envs.base.Environment._process_reactions(agent_id, reactions)

class econsimulacra.memory.memory_items.SleepHistoryItem(start_time, end_time)[source]

Bases: object

A class representing a sleep history item in the agent’s memory.

Parameters:

start_time (int | str)
end_time (str | int | None)

start_time

the time of the sleep action.

Type:: int | str

end_time

the time when the sleep action ends.

Type:: int | str

Note

This history item is generated based on the SleepStartLog and SleepEndLog. See also: econsimulacra.logs.base.SleepStartLog, econsimulacra.logs.base.SleepEndLog, econsimulacra.envs.base.Environment.(agent_id)

class econsimulacra.memory.memory_items.SetPriceHistoryItem(item_name, old_price, new_price, time, time_step)[source]

Bases: object

A class representing a price change history item in the agent’s memory.

Parameters:

item_name (str)
old_price (int | float)
new_price (int | float)
time (int | str)
time_step (int)

item_name

the name of the item whose price is changed.

Type:: str

old_price

the old price of the item.

Type:: int | float

new_price

the new price of the item.

Type:: int | float

time

the time of the price change.

Type:: int | str

Note

This history item is generated based on the ChangePriceLog. See also: econsimulacra.logs.base.ChangePriceLog, econsimulacra.envs.base.Environment._set_price(agent_id, set_prices)

class econsimulacra.memory.memory_items.SocialHistoryItem(action, target_agent_id, time, time_step, num_followers, num_follows)[source]

Bases: object

A class representing a social action history item in the agent’s memory.

Parameters:

action (Literal['follow', 'unfollow'])
target_agent_id (int)
time (int | str)
time_step (int)
num_followers (int)
num_follows (int)

action

the type of the social action.

Type:: Literal[“follow”, “unfollow”]

target_agent_id

the id of the target agent whom the agent follows or unfollows.

Type:: int

time

the time of the social action.

Type:: int | str

num_followers

the number of followers of the agent after the social action.

Type:: int

num_follows

the number of agents that the agent follows after the social action.

Type:: int

Note

This history item is generated based on the FollowLog and UnfollowLog. See also: econsimulacra.logs.base.FollowLog, econsimulacra.logs.base.UnfollowLog, econsimulacra.envs.base.Environment._act_in_social_network(agent_id, tweet, follow_agent_id, unfollow_agent_id)

class econsimulacra.memory.memory_items.StateEvaluationHistoryItem(wealth, relative_wealth, buying_power, inventory_dic, persona_dic, time, time_step)[source]

Bases: object

A class representing a state evaluation item in the agent’s memory.

Parameters:

wealth (float)
relative_wealth (float | None)
buying_power (float | None)
inventory_dic (dict[str, int | float])
persona_dic (dict[str, Any] | None)
time (int | str)
time_step (int)

wealth

the wealth of the agent at the time of evaluation.

Type:: float

relative_wealth

The relative wealth of the agent at the time of evaluation. Only household agents have this value; for other agent types, it is None.

Type:: float, optional

buying_power

The buying power of the agent at the time of evaluation. Only household agents have this value; for other agent types, it is None.

Type:: float, optional

inventory_dic

the inventory of the agent at the time of evaluation.

Type:: dict[str, int | float]

persona_dic

the persona of the agent at the time of evaluation.

Type:: dict[str, Any], optional

time

the time of the state evaluation.

Type:: int | str

Note

This history item is generated based on the StateEvaluationLog. See also: econsimulacra.logs.base.StateEvaluationLog, econsimulacra.envs.base.Environment.evaluate_agent_state(agent_id)

class econsimulacra.memory.memory_items.ObsHistoryItem(obs_type, time, time_step, obs)[source]

Bases: object

A class representing an observation item in the agent’s memory.

Parameters:

obs_type (str)
time (int | str)
time_step (int)
obs (Any)

obs_type

The type of the observation.

Type:: str

time

The time of the observation.

Type:: int | str

time_step

The time step of the observation.

Type:: int

agent_id

The unique id of the agent making the observation.

Type:: int

obs

The observation details.

Type:: Any

class econsimulacra.memory.memory_items.AgentMemory(invalid_action_history, consumption_history, sleep_history, move_history, purchase_history, sale_history, exchange_history, set_price_history, inner_thought_history, social_history, state_evaluation_history, obs_history)[source]

Bases: object

Agent Memory class.

Store the history of the agent’s actions and observations in a summarized form. The memory is updated based on the logs generated by the environment.

Parameters:

invalid_action_history (Deque[InvalidActionHistoryItem])
consumption_history (Deque[ConsumptionHistoryItem])
sleep_history (Deque[SleepHistoryItem])
move_history (Deque[MoveHistoryItem])
purchase_history (Deque[PurchaseHistoryItem])
sale_history (Deque[SaleHistoryItem])
exchange_history (Deque[ExchangeHistoryItem])
set_price_history (Deque[SetPriceHistoryItem])
inner_thought_history (Deque[InnerThoughtHistoryItem])
social_history (Deque[SocialHistoryItem])
state_evaluation_history (Deque[StateEvaluationHistoryItem])
obs_history (Deque[ObsHistoryItem])

invalid_action_history

the history of the agent’s invalid actions.

Type:: Deque[InvalidActionHistoryItem]

consumption_history

the history of the agent’s consumption.

Type:: Deque[ConsumptionHistoryItem]

move_history

the history of the agent’s movement.

Type:: Deque[MoveHistoryItem]

purchase_history

the history of the agent’s purchase.

Type:: Deque[PurchaseHistoryItem]

sale_history

the history of the agent’s sale.

Type:: Deque[SaleHistoryItem]

exchange_history

the history of the agent’s exchange.

Type:: Deque[ExchangeHistoryItem]

set_price_history

the history of the agent’s price change.

Type:: Deque[SetPriceHistoryItem]

inner_thought_history

the history of the agent’s inner thoughts.

Type:: Deque[InnerThoughtHistoryItem]

social_history

the history of the agent’s social actions.

Type:: Deque[SocialHistoryItem]

state_evaluation_history

the history of the agent’s state evaluations.

Type:: Deque[StateEvaluationHistoryItem]

obs_history

the history of the agent’s observations.

Type:: Deque[ObsHistoryItem]

Note

The history is stored in a deque with a maximum length of memory_length, which is defined in the MemoryHandler. When the history exceeds the maximum length, the oldest history will be removed.

Memory Handler

class econsimulacra.memory.base.MemoryHandler(config, prng=None, registered_classes=[])[source]

Bases: object

Memory Handler class.

Parameters:

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

get_memory(agent_id)[source]

Summarize and return the memory of the agent with the given agent_id.

Parameters:: agent_id (int) – the id of the agent whose memory is to be retrieved.
Returns:: the summarized memory of the agent.
Return type:: dict[str, Any]

Note

Memory is provided as a part of the observation to the agent. econsimulacra.envs.obs_providers.MemoryProvider calls this method to get the memory of the agent. See also: econsimulacra.envs.obs_providers.MemoryProvider

The structure of the summarized memory is:

{
    "memory_length": int,
    "invalid_action_history": "You failed to perform action_type at time (description), ...",
    "sleep_history": "slept from time to time, ...",
    "move_history": "(x0,y0) -> (x1,y1) -> (x2,y2)",
    "consumption_history": "item_name x quantity at time, ...",
    "purchase_history": "item_name x quantity at price from agent_id at time, ...",
    "sale_history": "item_name x quantity at price to agent_id at time, ...",
    "exchange_history": "give item_name x quantity, get item_name x quantity with agent_id at time; ...",
    "set_price_history": "item_name: old_price -> new_price at time, ...",
    "social_history": "follow target_agent_id at time (num_followers: N, num_follows: M); ...",
    "state_evaluation_history": "Wealth: wealth at time; ...",
}

update(log)[source]

Update memory based on the log. This method is called by the environment.

Note

See also:: econsimulacra.envs.base.Environment.remember_log(log: Log)

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

Summarizers

class econsimulacra.memory.summarizer.MemorySummarizer(config, prng=None, registered_classes=None)[source]

Bases: object

Memory Summarizer class.

MemorySummarizer is used to summarize the memory of the agent into a form that can be provided as a part of the observation to the agent.

Parameters:

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

sync_time(current_time, current_time_step)[source]

Synchronize the current time and time step in the summarizer with the MemoryHandler.

Parameters:

current_time (int | str)
current_time_step (int)

Return type:

None

Observation Summarization Utilities

econsimulacra.memory.obs_summarization_utils.summarize_observed_price_changes(obs_items, relative_threshold=0.01)[source]

Summarizes significant price changes in the observed inventory over time.

Parameters:

obs_items (list[ObsHistoryItem]) – A list of observation items to summarize.
relative_threshold (float)

Returns:

A summarized string of inventory price changes.

Return type:

str

Note

The components in obs_items must be: ObsHistoryItem with obs_type==’others_inventory’. The function identifies significant price changes and summarizes them in a human-readable format. If no significant changes are found, it returns an empty string.

econsimulacra.memory.obs_summarization_utils.summarize_num_changes(obs_items, is_follow=True)[source]

Summarizes significant changes in the number of follows/followers over time.

Parameters:

obs_items (list[ObsHistoryItem]) – A list of observation items to summarize.
is_follow (bool)

Returns:

A summarized string of follows changes.

Return type:

str

Note

The components in obs_items must be: ObsHistoryItem with obs_type==’num_follows’ or ‘num_followers’.

econsimulacra.memory.obs_summarization_utils.summarize_self_tweet_frequency(obs_items)[source]

Summarizes the frequency of the agent’s own tweets over time.

Parameters:: obs_items (list[ObsHistoryItem]) – A list of observation items to summarize.
Returns:: A summary of the agent’s tweet frequency.
Return type:: str

Note

The components in obs_items must be: ObsHistoryItem with obs_type==’self_tweet’. - obs == None means the agent has never tweeted yet. - obs stores the latest tweet content. - If obs is unchanged from the previous observation,

it means no new tweet was posted.

Stress-Aware Summarizer

class econsimulacra.memory.stress_aware_summarizer.StressCalculator(config, prng=None, registered_classes=[])[source]

Bases: object

Stress Calculator class.

StressCalculator is a class that calculates the stress level of the agent based on the memory of the agent.

Parameters:

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

sync_time(time, time_step)[source]

Sync the current time and time step of the stress calculator.

Parameters:

time (int | str)
time_step (int)

Return type:

None

summarize_stress(field_name, history)[source]

Summarize the stress level based on the history.

Parameters:

field_name (str) – the name of the history field.
history (Deque[...]) – the history corresponding to field_name.

Returns:

the summarized stress level and description.

Return type:

tuple[Optional[int], str]

class econsimulacra.memory.stress_aware_summarizer.StressAwareSummarizer(config, prng=None, registered_classes=[])[source]

Bases: MemorySummarizer

Stress Aware Summarizer class.

StressAwareSummarizer is a MemorySummarizer that summarizes the memory of the agent into a form that can be provided as a part of the observation to the agent, while being aware of the stress level of the agent.

Parameters:

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

sync_time(current_time, current_time_step)[source]

Synchronize the current time and time step in the summarizer with the MemoryHandler.

Parameters:

current_time (int | str)
current_time_step (int)

Return type:

None

Stress Calculation

econsimulacra.memory.stress_utils.calc_stress_from_consumption_history(consumption_history, current_time_step, max_stress, target_quantity, window_size, time_decay, tolerance_threshold, item2weight)[source]

Calculate the stress level based on the consumption history.

Parameters:

consumption_history (Deque[ConsumptionHistoryItem]) – A deque of ConsumptionHistoryItem representing the consumption history.
current_time_step (int) – The current time step in the simulation.
max_stress (int) – The maximum stress level.
target_quantity (int) – The target quantity to consume.
window_size (int) – The size of the time window in time steps to consider for stress calculation.
time_decay (float) – The decay factor for the stress contribution of past consumption events.
tolerance_threshold (float) – The tolerance threshold for stress.
item2weight (dict[str, float]) – A dictionary mapping item names to their corresponding weights for stress calculation.

Returns:

A tuple containing:

stress_level: The calculated stress level.
stress_reason: The reason for the stress level.

Return type:

tuple[int, str]

Note

The stress level is calculated based on the quantity of items consumed within the specified time window, weighted by their respective weights, and decayed over time.

The weighted quantity is defined as:

\[Q(t) = \sum_{k: t_k \in [t-W, t]} \gamma^{(t - t_k)} \, w_{i_k} \, q_k\]

where:

\(t\) is the current time step
\(W\) is the window size
\(\gamma\) is the time decay factor
\(w_{i_k}\) is the weight of item \(i_k\)
\(q_k\) is the consumed quantity at time \(t_k\)

The stress level is then computed as:

\[s(t) = \min\left( s_{\max}, \left\lfloor \frac{|Q(t) - Q^*|}{Q^*} \, s_{\max} \right\rfloor \right)\]

where \(Q^*\) is the target quantity.

Corner cases:

If consumption_history is empty and \(t \geq W\), then \(s(t) = s_{\max}\).
If consumption_history is empty and \(t < W\), then \(s(t) = 0\).

econsimulacra.memory.stress_utils.calc_stress_from_move_history(move_history, current_time_step, max_stress, target_distance, window_size, time_decay, tolerance_threshold, home_comfort)[source]

Calculate the stress level based on the move history.

Parameters:

move_history (Deque[MoveHistoryItem]) – A deque of MoveHistoryItem representing the move history.
current_time_step (int) – The current time step in the simulation.
max_stress (int) – The maximum stress level.
target_distance (float) – The target distance to move from the initial position.
window_size (int) – The size of the time window in time steps to consider for stress calculation.
time_decay (float) – The decay factor for the stress contribution of past move events.
tolerance_threshold (float) – The tolerance threshold for stress.
home_comfort (float) – The comfort level of being at home.

Returns:

A tuple containing:

stress_level: The calculated stress level.
stress_reason: The reason for the stress level.

Return type:

tuple[int, str]

Note

The stress level is calculated based on the distance moved during the time steps within the specified time window, decayed over time, and adjusted by the home comfort level.

The distance moved is defined as:

\[D(t) = \sum_{k: t_k \in [t-W, t]} \gamma^{(t - t_k)} \, \|x_k - x_{k-1}\|\]

where:

\(t\) is the current time step
\(W\) is the window size
\(\gamma\) is the time decay factor
\(\|x_k - x_{k-1}\|\) is the distance moved at time \(t_k\)

The stress level is then computed as:

\[s(t) = \min\left( s_{\max}, \left\lfloor \frac{|D(t) - D^*|}{D^*} \, s_{\max} \, (1 - h)\textbf{1}(x_t=x_0) \right\rfloor \right)\]

where \(D^*\) is the target distance and \(h\) is the home comfort level.

Corner cases: - If move_history is empty and

\(t \geq W\), then \(s(t) = s_{\max}\).

If move_history is empty and \(t < W\), then \(s(t) = 0\).

econsimulacra.memory.stress_utils.calc_stress_from_state_evaluation_history(state_evaluation_history, current_time_step, max_stress, target_buying_power, target_relative_wealth, target_wealth_growth, window_size, tolerance_threshold, buying_power_weight=1.0, relative_wealth_weight=1.0, wealth_drawdown_weight=1.0)[source]

Calculate economic stress based on state evaluation history.

Parameters:

state_evaluation_history (Deque[StateEvaluationHistoryItem]) – A deque of StateEvaluationHistoryItem representing the agent’s economic state history.
current_time_step (int) – The current time step in the simulation.
max_stress (int) – The maximum stress level.
target_buying_power (float) – The target buying power level.
target_relative_wealth (float) – The target relative wealth level.
target_wealth_growth (float) – The target wealth growth over the window.
window_size (int) – The size of the time window in time steps.
tolerance_threshold (float) – The threshold above which stress is reported.
buying_power_weight (float) – Weight for buying-power stress.
relative_wealth_weight (float) – Weight for relative-wealth stress.
wealth_drawdown_weight (float) – Weight for wealth-drawdown stress.

Returns:

A tuple containing:

stress_level: The calculated stress level.
stress_reason: The reason for the stress level.

Return type:

tuple[int, str]

Note

The stress is computed from three economic factors:

Buying-power stress: stress from having insufficient purchasing power.
Relative-wealth stress: stress from having less wealth than others.
Wealth-drawdown stress: stress from recent decreases in wealth.

The buying-power stress is defined as:

\[s_{bp}(t) = \max\left( 0, \frac{B^* - B(t)}{B^*} \right)\]

where \(B(t)\) is buying power and \(B^*\) is the target buying power.

The relative-wealth stress is defined as:

\[s_{rw}(t) = \max\left( 0, \frac{R^* - R(t)}{|R^*| + 1 + \epsilon} \right)\]

where \(R(t)\) is relative wealth and \(R^*\) is the target relative wealth.

The wealth-drawdown stress is defined as:

\[s_{dd}(t) = \max\left( 0, \frac{G^* - G(t)}{|G^*| + w_i(0) \epsilon} \right)\]

where \(G(t)\) is the wealth change over the window, \(w_i(0)\) is the initial wealth, and \(G^*\) is the target wealth growth.

The total stress score is:

\[S(t) = \alpha s_{bp}(t) + \beta s_{rw}(t) + \delta s_{dd}(t)\]

and the final stress level is:

\[\min\left( s_{\max}, \left\lfloor S(t) s_{\max} \right\rfloor \right)\]

econsimulacra.memory.stress_utils.calc_stress_from_sleep_history(sleep_history, current_time, current_time_step, max_stress, target_sleep_duration, window_size, tolerance_threshold, duration_weight=0.7, regularity_weight=0.3)[source]

Calculate sleep stress from sleep duration and sleep regularity.

This function evaluates the agent’s sleep condition over a recent time window by combining two stress components:

sleep-duration stress, which increases when the total sleep duration within the window is below the target duration; and
sleep-regularity stress, which increases when sleep onset and wake-up times are irregular across sleep episodes.

Parameters:

sleep_history (Deque[SleepHistoryItem]) – A deque of sleep history items. Each item represents one sleep interval with start_time and end_time.
current_time (int | str) – The current simulation time. If this is an int, all sleep times are interpreted as integer simulation steps. If this is a str, all sleep times are parsed using time_format and interpreted as datetimes.
current_time_step (int) – The current simulation step. This is used only for detecting the initial phase of the simulation. If no sleep is observed and current_time_step < window_size, the function returns zero stress because the agent has not yet had enough time to sleep.
max_stress (int) – The maximum possible stress level.
target_sleep_duration (float) – The target amount of sleep within the time window. The unit must be consistent with current_time and window_size. If current_time is an int, this is measured in simulation steps. If current_time is a datetime string, this is measured in hours.
window_size (float) – The size of the retrospective time window. The function evaluates sleep intervals overlapping with [current_time - window_size, current_time]. If current_time is an int, this is measured in simulation steps. If current_time is a datetime string, this is measured in hours.
time_format (str) – Datetime format used to parse string-valued times. The default is "%Y-%m-%d %H:%M:%S".
tolerance_threshold (float) – Threshold below which the stress level is regarded as acceptable.
duration_weight (float) – Weight assigned to sleep-duration stress.
regularity_weight (float) – Weight assigned to sleep-regularity stress.

Returns:

A tuple containing:

stress_level: the calculated sleep stress level.
stress_reason: a human-readable explanation of the stress.

Return type:

tuple[int, str]

Raises:

ValueError – If max_stress is negative.
ValueError – If target_sleep_duration is not positive.
ValueError – If window_size is not positive.
ValueError – If tolerance_threshold is outside [0, max_stress].
ValueError – If duration_weight or regularity_weight is negative.
ValueError – If duration_weight + regularity_weight is not positive.
ValueError – If a sleep history item has end_time is None.
ValueError – If a sleep interval satisfies end_time < start_time.

Notes

Let \(t\) be the current time and \(W\) be the window size. The retrospective evaluation window is

\[[t - W, t].\]

Each completed sleep episode \(i\) is represented by its start time \(a_i\) and end time \(b_i\). Only the overlap between the sleep interval and the evaluation window contributes to the observed sleep duration:

\[d_i(t) = \max\left( 0, \min(b_i, t) - \max(a_i, t - W) \right).\]

The total sleep duration in the window is therefore

\[D(t) = \sum_i d_i(t).\]

Given the target sleep duration \(D^*\), the sleep-duration stress is defined as

\[s_{\mathrm{dur}}(t) = \min\left( s_{\max}, \left\lfloor \frac{\max(0, D^* - D(t))}{D^*} s_{\max} \right\rfloor \right).\]

This definition penalizes sleep deficiency but does not penalize oversleeping. That is, if \(D(t) \geq D^*\), then \(s_{\mathrm{dur}}(t) = 0\).

Sleep regularity is evaluated using the circular variance of sleep onset times and wake-up times. Let \(x_j \in [0, 24)\) denote clock times in hours. Each clock time is mapped to an angle

\[\theta_j = \frac{2\pi x_j}{24}.\]

The mean resultant length is

\[R = \sqrt{ \left( \frac{1}{n} \sum_{j=1}^n \cos \theta_j \right)^2 + \left( \frac{1}{n} \sum_{j=1}^n \sin \theta_j \right)^2 }.\]

The circular variance is then

\[V = 1 - R.\]

Let \(V_{\mathrm{start}}\) and \(V_{\mathrm{end}}\) denote the circular variances of sleep onset times and wake-up times, respectively. The sleep-regularity stress is

\[s_{\mathrm{reg}}(t) = \left\lfloor \frac{ V_{\mathrm{start}} + V_{\mathrm{end}} }{2} s_{\max} \right\rfloor.\]

If fewer than two completed sleep episodes are available in the evaluation window, regularity cannot be estimated and the function sets

\[s_{\mathrm{reg}}(t) = 0.\]

Finally, the total sleep stress is the weighted average of duration stress and regularity stress:

\[s(t) = \min\left( s_{\max}, \left\lfloor \tilde{w}_{\mathrm{dur}} s_{\mathrm{dur}}(t) + \tilde{w}_{\mathrm{reg}} s_{\mathrm{reg}}(t) \right\rfloor \right),\]

where normalized weights are given by

\[\tilde{w}_{\mathrm{dur}} = \frac{w_{\mathrm{dur}}}{ w_{\mathrm{dur}} + w_{\mathrm{reg}} }, \quad \tilde{w}_{\mathrm{reg}} = \frac{w_{\mathrm{reg}}}{ w_{\mathrm{dur}} + w_{\mathrm{reg}} }.\]

Corner cases:

If no sleep overlaps with the evaluation window and current_time_step >= window_size, the function returns max_stress.
If no sleep overlaps with the evaluation window and current_time_step < window_size, the function returns 0.
If only one sleep episode is available in the evaluation window, sleep-duration stress is computed normally, but sleep-regularity stress is set to 0.

Examples

Evaluate sleep stress using integer simulation steps:

stress, reason = calc_stress_from_sleep_history(
    sleep_history=sleep_history,
    current_time=48,
    current_time_step=48,
    max_stress=10,
    target_sleep_duration=8.0,
    window_size=24.0,
)

Evaluate sleep stress using datetime strings:

stress, reason = calc_stress_from_sleep_history(
    sleep_history=sleep_history,
    current_time="2026-05-26 09:00:00",
    current_time_step=48,
    max_stress=10,
    target_sleep_duration=8.0,
    window_size=24.0,
    time_format="%Y-%m-%d %H:%M:%S",
)

econsimulacra.memory.stress_utils.calc_stress_from_obs_history(obs_history, current_time_step, max_stress, target_num_nearby, tolerance_threshold)[source]

Calculate stress from the number of nearby agents in the observation history.

Parameters:

obs_history (Deque[ObsHistoryItem]) – A deque of ObsHistoryItem representing the agent’s observation history.
current_time_step (int) – The current time step in the simulation.
max_stress (int) – The maximum stress level.
target_num_nearby (int) – The target number of nearby agents.
tolerance_threshold (float) – The threshold above which stress is reported.

Returns:

A tuple containing: - The calculated stress level (int). - A message describing the stress reason (str).

Return type:

tuple[int, str]

Note

The stress level is calculated based on the latest number of nearby agents observed in the history. If the number of nearby agents differs from the target, the stress level increases proportionally to the difference, up to the maximum stress level.

\[s(t) = \min\left( s_{\max}, \left\lfloor \frac{|N(t) - N^*|}{N^*} s_{\max} \right\rfloor \right),\]