ergodicity.agents package

Submodules

ergodicity.agents.agent_pool module

agent_pool Submodule Overview

The `agent_pool` submodule focuses on simulating and analyzing the wealth dynamics of a pool of agents interacting with stochastic processes. Agents share wealth dynamically based on a specified sharing rate, and the module provides tools to study how wealth distributions and inequality evolve over time.

Key Features:

1. Agent Pool Simulation:

   • Simulate the evolution of wealth for a group of agents interacting with a stochastic process (e.g., Geometric Brownian Motion, Lévy processes).

   • Each agent starts with a specified initial wealth, and the evolution of their wealth depends on both the process and the sharing rules.

2. Dynamic Sharing Rate:

   • Agents can share wealth dynamically based on their relative wealth or a fixed sharing rate.

   • The simulation can be run with either static or dynamic sharing rates.

3. Wealth History:

   • Track the wealth of each agent over time. The full wealth history is recorded and can be saved for further analysis.

4. Wealth Inequality Measures:

   • The module provides several key metrics to measure wealth inequality, including:

     • Mean Logarithmic Deviation (MLD): A measure of wealth inequality based on the deviation from the mean.

     • Gini Coefficient: A common measure of inequality based on the distribution of wealth.

     • Coefficient of Variation (CV): The ratio of standard deviation to the mean wealth.

     • Palma Ratio: The ratio of wealth held by the richest 10% compared to the poorest 40%.

5. Wealth Distribution Analysis:

   • Save and visualize the final wealth distribution in both normal and log-log scales.

   • Fit a power law distribution to the wealth data to study the tail behavior.

6. 3D Wealth Visualization:

   • Create static and interactive 3D plots of total wealth as a function of the number of agents and the sharing rate.

   • Use Matplotlib for static 3D visualization and Plotly for interactive 3D graphs, which can be saved as HTML.

Example Usage:

### Basic Simulation:

from ergodicity.process.multiplicative import GeometricBrownianMotion

from ergodicity.agent_pool import AgentPool

# Initialize process (e.g., GeometricBrownianMotion)

process = GeometricBrownianMotion(drift=0.02, volatility=0.15)

# Initialize agent pool with 100 agents, starting wealth of 100, sharing rate of 0.1, and time horizon of 10

pool = AgentPool(process, n=100, w=100, s=0.1, time=10)

# Simulate the wealth dynamics

pool.simulate(dynamic_s=False)

# Plot wealth evolution

pool.plot()

# Save and plot final wealth distribution

pool.save_and_plot_wealth_distribution(‘final_wealth’)

class ergodicity.agents.agent_pool.AgentPool(process, n, w, s, time, simulation_timestep=0.01, timestep=1)

Bases: object

AgentPool represents a collection of agents participating in a wealth dynamics simulation. This class simulates the evolution of wealth for a group of agents over time, incorporating a stochastic process for wealth changes and an optional wealth-sharing mechanism. It provides methods for simulation, data analysis, and visualization of wealth dynamics and inequality measures.

process

The stochastic process used for simulating wealth changes.

n

Number of agents in the pool.

Type:

int

initial_wealth

Starting wealth for each agent.

Type:

float

base_s

Base sharing rate for wealth redistribution.

Type:

float

time

Total simulation time.

Type:

float

timestep

Time step for wealth updates and sharing.

Type:

float

simulation_timestep

Time step for the underlying stochastic process simulation.

Type:

float

wealth

Current wealth of each agent.

Type:

numpy.ndarray

history

Historical record of wealth for all agents at each time step.

Type:

list

This class is useful for studying wealth inequality, the effects of different sharing mechanisms, and the impact of various stochastic processes on wealth distribution in a population of agents.

coefficient_of_variation()

Compute the Coefficient of Variation for each time step.

Returns:

Array of CV values for each time step

Return type:

numpy.ndarray

gini_coefficient()

Compute the Gini coefficient for each time step. The Gini coefficient is a measure of statistical dispersion intended to represent the income or wealth distribution of a nation’s residents.

Returns:

Array of Gini coefficient values for each time step

Return type:

numpy.ndarray

mean_logarithmic_deviation()

Compute the Mean Logarithmic Deviation (MLD) for each time step.

Returns:

array of MLD values for each time step

Return type:

numpy.ndarray

palma_ratio()

Compute the Palma ratio for each time step. The Palma ratio is the ratio of the richest 10% of the population’s share of gross national income divided by the poorest 40%’s share.

Returns:

Array of Palma ratio values for each time step

Return type:

numpy.ndarray

plot()

Visualize the wealth dynamics of all agents over time.

Returns:

None

Return type:

None

plot_inequality_measures()

Plot all implemented inequality measures over time.

Returns:

None

Return type:

None

save_and_plot_wealth_distribution(filename_prefix)

Save and plot the final wealth distribution in normal and log-log scales.

Parameters:

filename_prefix (str) – Prefix for saving the plot files

Returns:

None

Return type:

None

save_data(filename)

Save the wealth history data to a file.

Parameters:

filename (str) – Name of the file to save the data

Returns:

None

Return type:

None

simulate(dynamic_s=False)

Run the wealth dynamics simulation for the specified time horizon.

Parameters:

dynamic_s (bool) – If True, agents share wealth dynamically based on their relative wealth

Returns:

List of wealth history for all agents at each time step

Return type:

list

ergodicity.agents.agent_pool.plot_wealth_3d(process, w, time, simulation_timestep, timestep, s_range, n_range, save_html=False)

Plot a 3D graph of average wealth as a function of sharing rate s and number of agents n.

Parameters:

• process (StochasticProcess) – The stochastic process to use (e.g., GeometricBrownianMotion instance)

• w (float) – Initial wealth for each agent

• time (float) – Total simulation time

• simulation_timestep (float) – Timestep for the simulation

• timestep (float) – Timestep for the sharing and update of AgentPool

• s_range (numpy.ndarray) – Range of sharing rates to plot (e.g., np.linspace(0, 0.5, 20))

• n_range (range) – Range of number of agents to plot (e.g., range(5, 105, 5))

• save_html (bool) – If True, save an interactive 3D plot as an HTML file

Returns:

None

Return type:

None

ergodicity.agents.agents module

agents Submodule Overview

The `agents` module focuses on creating and managing economic agents that optimize their utility functions over time in a stochastic environment. It provides various algorithms and tools for simulating agent behavior, evolving utility functions, and running multi-agent evolutionary processes. These agents interact with multiple stochastic processes, making decisions based on expected utility and wealth maximization.

Key Features:

1. Utility Functions:

   • Each agent uses a general utility function parameterized by alpha, beta, gamma, delta, and epsilon.

   • Utility functions are used to calculate the expected utility from various stochastic processes.

2. Expected Utility:

   • The module allows agents to calculate expected utility either numerically (via process simulations) or symbolically (via integral approximations).

3. Evolutionary Algorithms:

   • Agents evolve over time using evolutionary algorithms that simulate wealth accumulation and utility optimization.

   • Agents with higher utility accumulate more wealth and influence the population’s evolution.

   • Includes features like mutation, agent reproduction, and selection of top-performing agents.

4. Processes:

   • Agents interact with multiple stochastic processes, such as Geometric Brownian Motion or Geometric Lévy processes, to maximize their wealth.

   • Agents select and invest in the best process based on expected utility.

5. Multi-Agent Simulations:

   • The module provides tools to simulate the behavior of a population of agents over multiple time steps.

   • Agents are removed or duplicated based on their performance, and new agents are created through mutation.

6. Visualization:

   • Includes tools for visualizing agent evolution, utility function evolution, and other trends over time.

   • Provides methods to generate animations and CSV logs for post-simulation analysis.

7. Support for Numerical and Symbolic Computation:

   • Both symbolic and numerical approaches are supported for computing expected utility, offering flexibility in different simulation scenarios.

8. Evolution with Multiple Processes:

   • Agents can select from multiple stochastic processes, optimizing their utility functions in a diverse environment.

   • The module supports the creation of multiple stochastic processes and allows agents to interact with them over time.

Example Usage:

### Basic Usage of Evolutionary Algorithm:

from ergodicity.process.multiplicative import GeometricBrownianMotion

from ergodicity.agents import Agent_utility

# Define parameters for agents

param_means = np.array([1.0, 1.0, 0.5, 1.0])

param_stds = np.array([0.1, 0.1, 0.05, 0.1])

mutation_rate = 0.01

# Define a set of stochastic processes

processes = [

GeometricBrownianMotion(drift=0.02, volatility=0.15), GeometricBrownianMotion(drift=0.03, volatility=0.18)

]

# Run the evolutionary algorithm

final_agents, history = Agent_utility.evolutionary_algorithm(

n_agents=100, n_steps=1000, save_interval=50, processes=processes, param_means=param_means, param_stds=param_stds, mutation_rate=mutation_rate, stochastic_process_class=GeometricBrownianMotion, keep_top_n=50, removal_interval=10, process_selection_share=0.5

)

# Visualize agent evolution

Agent_utility.visualize_agent_evolution(history)

class ergodicity.agents.agents.Agent_utility(params: ndarray, wealth: float = 1.0, total_accumulated_wealth: float = 1.0)

Bases: object

Represents an agent with utility-based decision making in stochastic processes.

This class implements an agent that can evaluate and interact with various stochastic processes based on a parameterized utility function. It supports both symbolic and numerical methods for calculating expected utilities, and includes evolutionary algorithms for optimizing agent parameters.

params

Parameters of the agent’s utility function [alpha, beta, gamma, delta, epsilon].

Type:

np.ndarray

wealth

Current wealth of the agent.

Type:

float

total_accumulated_wealth

Total accumulated wealth of the agent over time.

Type:

float

The class provides comprehensive tools for modeling agent behavior in complex stochastic environments, including utility calculation, parameter optimization, and visualization of results. It is particularly useful for studying optimal strategies in financial markets, decision making under uncertainty, and evolutionary dynamics in economic systems.

static compare_numerical_and_symbolic_expected_utility(process_dict: Dict[str, Any], params: ndarray, stochastic_process_class: Type, t: float = None)

Compare the numerical and symbolic expected utility calculations for a given process and parameters.

Parameters:

• process_dict (Dict[str, Any]) – Dictionary containing the symbolic function and process parameters

• params (np.ndarray) – Parameters of the utility function [alpha, beta, gamma, delta, epsilon]

• stochastic_process_class (Type) – The class of the stochastic process

• t (float) – Time horizon for the process. If None, compares functions instead of specific values.

Returns:

None (prints the comparison results)

Return type:

None

static evolutionary_algorithm(n_agents: int, n_steps: int, save_interval: int, processes: List[dict | object], param_means: ndarray, param_stds: ndarray, mutation_rate: float, stochastic_process_class: Type = None, keep_top_n: int = 50, removal_interval: int = 10, process_selection_share: float = 0.5, output_dir: str = 'output', process_time=1.0, numeric_utilities: bool = True)

Run an evolutionary algorithm to optimize agent parameters based on expected utility.

Parameters:

• n_agents (int) – Number of agents in the population

• n_steps (int) – Number of steps to run the algorithm

• save_interval (int) – Interval for saving intermediate results

• processes (List[Union[dict, object]]) – List of stochastic processes for agents to interact with

• param_means (np.ndarray) – Means of the parameter distributions

• param_stds (np.ndarray) – Standard deviations of the parameter distributions

• mutation_rate (float) – Rate of mutation for agent parameters

• stochastic_process_class (Type) – Class of the stochastic process (if using dict-based processes)

• keep_top_n (int) – Number of top agents to keep after each removal interval

• removal_interval (int) – Interval for removing agents based on performance (in time units)

• process_selection_share (float) – Share of processes to select for each agent (0 to 1)

• output_dir (str) – Directory to save output files

• process_time (float) – Time horizon for the stochastic processes

• numeric_utilities (bool) – Use numerical utility calculation if True, else use symbolic

Raises:

InDevelopmentWarning – If numeric_utilities is False (feature in development)

Returns:

List of final agents and history of the evolutionary process

Return type:

Tuple[List[Agent_utility], List[Dict[str, Any]]]

static evolutionary_algorithm_with_exchange(n_agents: int, n_steps: int, save_interval: int, processes: List[dict | object], param_means: ndarray, param_stds: ndarray, noise_std: float, stochastic_process_class: Type = None, top_k: int = 10, process_selection_share: float = 0.5, output_dir: str = 'output', s: int = 10, process_time=1.0, numeric_utilities: bool = True)

Run an evolutionary algorithm to optimize agent parameters based on expected utility with the exchange of parameters (evolution-like mixing of genetic information).

Parameters:

• n_agents (int) – Number of agents in the population

• n_steps (int) – Number of steps to run the algorithm

• save_interval (int) – Interval for saving intermediate results

• processes (List[Union[dict, object]]) – List of stochastic processes for agents to interact with

• param_means (np.ndarray) – Means of the parameter distributions

• param_stds (np.ndarray) – Standard deviations of the parameter distributions

• noise_std (float) – Standard deviation of the noise added to averaged parameters

• stochastic_process_class (Type) – The class of the stochastic process (if using dict-based processes)

• top_k (int) – Number of top agents to keep after each selection interval

• process_selection_share (float) – Share of processes to select for each agent (0 to 1)

• output_dir (str) – Directory to save output files

• s (int) – Interval for selection, averaging, and multiplication

• process_time (float) – Time horizon for the stochastic processes

• numeric_utilities (bool) – Use numerical utility calculation if True, else use symbolic

Raises:

InDevelopmentWarning – If numeric_utilities is False (feature in development)

Returns:

List of final agents and history of the evolutionary process

Return type:

Tuple[List[Agent_utility], List[Dict[str, Any]]]

static evolutionary_algorithm_with_multiple_processes(n_agents: int, n_steps: int, save_interval: int, processes: List[dict | object], param_means: ndarray, param_stds: ndarray, mutation_rate: float, keep_top_n: int = 50, removal_interval: int = 10, process_selection_share: float = 0.5, output_dir: str = 'output', process_time: float = 1.0, numeric_utilities: bool = True)

Run an evolutionary algorithm to optimize agent parameters based on expected utility with multiple process types.

Parameters:

• n_agents (int) – Number of agents in the population

• n_steps (int) – Number of steps to run the algorithm

• save_interval (int) – Interval for saving intermediate results

• processes (List[Union[dict, object]]) – List of stochastic processes for agents to interact with

• param_means (np.ndarray) – Means of the parameter distributions

• param_stds (np.ndarray) – Standard deviations of the parameter distributions

• mutation_rate (float) – Rate of mutation for agent parameters

• keep_top_n (int) – Number of top agents to keep after each removal interval

• removal_interval (int) – Interval for removing agents based on performance (in time units)

• process_selection_share (float) – Share of processes to select for each agent (0 to 1)

• output_dir (str) – Directory to save output files

• process_time (float) – Time horizon for the stochastic processes

• numeric_utilities (bool) – Use numerical utility calculation if True, else use symbolic

Raises:

InDevelopmentWarning – If numeric_utilities is False (feature in development)

Returns:

List of final agents and history of the evolutionary process

Return type:

Tuple[List[Agent_utility], List[Dict[str, Any]]]

static expected_utility(process_dict: Dict[str, Any], params: ndarray, t: float = None) → Callable[[float], float] | float

Calculate the expected utility of a process using robust adaptive numerical integration.

Parameters:

• process_dict (Dict[str, Any]) – Dictionary containing the symbolic function and process parameters

• params (np.ndarray) – Parameters of the utility function [alpha, beta, gamma, delta, epsilon]

• t (float) – Time horizon for the process. If None, returns a function of t.

Returns:

Either a function that computes expected utility for any t, or the numerical value for a specific t

Return type:

Union[Callable[[float], float], float]

static general_utility_function(x, alpha, beta, gamma, delta, epsilon)

Symbolic representation of the general utility function.

Parameters:

• x (sp.Symbol) – The input value

• epsilon (alpha, beta, gamma, delta,) – Utility function parameters

Returns:

The utility function expression

Return type:

sp.Expr

static general_utility_function_np(x, alpha, beta, gamma, delta, epsilon)

Numpy version of the general utility function with overflow protection

Parameters:

• x (Union[float, np.ndarray]) – The input value(s)

• epsilon (alpha, beta, gamma, delta,) – Utility function parameters

Returns:

The utility function value(s)

Return type:

Union[float, np.ndarray]

static initialize_agents(n: int, param_means: ndarray, param_stds: ndarray) → List[Agent_utility]

Initialize a population of agents with random parameters drawn from normal distributions.

Parameters:

• n (int) – Number of agents to create

• param_means (np.ndarray) – Means of the parameter distributions

• param_stds (np.ndarray) – Standard deviations of the parameter distributions

Returns:

List of initialized agents

static mutate_params(params: ndarray, mutation_rate: float) → ndarray

Mutate agent parameters for evolutionary algorithms.

Parameters:

• params (np.ndarray) – Current parameters of the agent

• mutation_rate (float) – Rate of mutation for the parameters

Returns:

Mutated parameters

Return type:

np.ndarray

static numerical_expected_utility(process: Dict[str, Any] | object, params: ndarray, stochastic_process_class: Type, t: float = None, num_instances: int = 1000) → Callable[[float], float] | float

Generate multiple instances of the process and take their average utility to approximate the expected utility.

Parameters:

• process (Union[Dict[str, Any], object]) – Either a dictionary containing process parameters or an instance of a stochastic process

• params (np.ndarray) – Parameters of the utility function [alpha, beta, gamma, delta, epsilon]

• stochastic_process_class (Type) – The class of the stochastic process

• t (float) – Time horizon for the process. If None, returns a function of t.

Returns:

Either a function that computes expected utility for any t, or the numerical value for a specific t

Return type:

Union[Callable[[float], float], float]

params: ndarray

total_accumulated_wealth: float = 1.0

static visualize_utility_function_evolution(history, output_video_path, output_csv_path)

Create an animation of the evolution of the best agent’s utility function over time.

Parameters:

• history (List[Dict[str, Any]]) – List of dictionaries containing step, best_params, and other data

• output_video_path (str) – Path to save the output video

• output_csv_path (str) – Path to save the output CSV file

Returns:

None

Return type:

None

wealth: float = 1.0

ergodicity.agents.agents.analyze_utility_function_trends(history: List[Dict], num_agents: int = 10)

Analyze and visualize trends in the utility functions’ evolutionary dynamics.

Parameters:

• history (List[Dict]) – List of dictionaries containing historical data

• num_agents (int) – Maximum number of top agents to analyze

Returns:

None (displays plots)

Return type:

None

ergodicity.agents.agents.general_utility_function(x, alpha, beta, gamma, delta, epsilon)

Calculates the utility for a given x and set of parameters. Works with both SymPy symbols and numpy arrays.

Parameters:

• x (Union[float, np.ndarray, sp.Expr, sp.Symbol]) – Input value or array

• alpha (float) – Utility function parameter

• beta (float) – Utility function parameter

• gamma (float) – Utility function parameter

• delta (float) – Utility function parameter

• epsilon (float) – Utility function parameter

Returns:

The calculated utility expression or array

Return type:

Union[float, np.ndarray]

ergodicity.agents.agents.generate_processes(num_processes, process_types, param_ranges)

Generate a list of stochastic processes.

Parameters:

• num_processes (int) – Number of processes to generate

• process_types (list) – List of process classes or functions to use

• param_ranges (dict) – Dictionary of parameter ranges for each process type

Returns:

List of generated processes

Return type:

list

ergodicity.agents.agents.process_to_dict(process: Any) → Dict[str, Any]

Converts a process object with a closed_formula method to a dictionary format compatible with Agent_utility.compare_numerical_and_symbolic_expected_utility.

Parameters:

process (Any) – An object with a closed_formula method

Returns:

A dictionary with ‘symbolic’ key for the formula and additional keys for process parameters

Return type:

Dict[str, Any]

ergodicity.agents.agents.recursive_flatten(data: Any) → List[float]

Recursively flatten any nested structure into a 1D list of floats.

Parameters:

data (Any) – Nested structure to flatten

Returns:

1D list of floats

Return type:

List[float]

ergodicity.agents.agents.visualize_agent_evolution(history: List[Dict], top_n: int = 5)

Visualize the evolution of top agents and their utility functions.

Parameters:

• history (List[Dict]) – List of dictionaries containing historical data

• top_n (int) – Number of top agents to visualize

Returns:

None (displays plots)

Return type:

None

ergodicity.agents.evaluation module

evaluation Submodule Overview

The `evaluation` submodule provides tools for analyzing, comparing, and optimizing utility functions in decision-making environments. It allows users to evaluate how agents which learnt their behaviour using different optimization algorithms interact with stochastic processes and fit utility functions to observed behaviors using various optimization techniques.

Key Features:

1. Utility Function Definition and Evaluation:

   • The submodule allows users to define custom utility functions and optimize them based on stochastic process trajectories.

   • The `UtilityFunction` class is used to define utility functions with initial parameters, and it supports fitting parameters to minimize negative log-likelihood.

2. Utility Function Inference:

   • The `UtilityFunctionInference` class facilitates the fitting of utility functions to agent decisions.

   • Users can use both maximum likelihood estimation (MLE) and Bayesian inference (using Metropolis-Hastings sampling) to fit utility functions.

   • This class also includes methods for generating datasets, fitting models, and visualizing results such as utility functions and parameter distributions.

3. Regression Analysis:

   • A neural network-based regression model can be trained to predict agent preferences based on process parameters.

   • The `regression_fit()` method fits a neural network to the dataset, while `plot_regression_results()` provides visualizations of training results.

4. Utility Function Tester:

   • The `UtilityFunctionTester` class allows users to test and compare multiple utility functions by simulating processes and calculating optimal utility values for each function.

   • Includes methods for generating process parameters, simulating process trajectories, and optimizing utility functions for given trajectories.

5. Inverse Reinforcement Learning (IRL):

   • The `MaxEntIRL` (Maximum Entropy Inverse Reinforcement Learning) class infers reward weights from agent behavior using observed trajectories.

   • The IRL approach fits a reward model that explains observed agent decisions by maximizing the likelihood of the agent’s actions.

Example Usage:

### Fitting Utility Functions to Agent’s Choices:

from ergodicity.evaluation import UtilityFunctionInference, UtilityFunction

from ergodicity.process.basic import BrownianMotion, GeometricBrownianMotion

# Initialize UtilityFunctionInference

ufi = UtilityFunctionInference(‘path/to/your/model.h5’, param_ranges={

‘BrownianMotion’: {‘mu’: (0, 0.5), ‘sigma’: (0.1, 0.5)}, ‘GeometricBrownianMotion’: {‘mu’: (0, 0.5), ‘sigma’: (0.1, 0.5)}

})

# Add utility functions to be fitted

ufi.add_utility_function(UtilityFunction(‘Power’, lambda x, beta: x ** beta, [1.0]))

ufi.add_utility_function(UtilityFunction(‘Exponential’, lambda x, alpha: 1 - np.exp(-alpha * x), [1.0]))

# Generate dataset of stochastic processes

dataset = ufi.generate_dataset(100)

# Get agent’s choices based on the dataset

choices = ufi.get_agent_choices(dataset)

# Fit utility functions based on the agent’s choices

ufi.fit_utility_functions(dataset, choices)

# Plot the fitted utility functions

ufi.plot_utility_functions()

class ergodicity.agents.evaluation.AgentEvaluation(model)

Bases: object

AgentEvaluation Class

select_process(encoded_processes, select_max=True)

Select a process based on the model’s predictions.

Parameters:

• encoded_processes – Encoded processes to choose from

• select_max – Whether to select the process with the maximum score

Returns:

Index of the selected process

class ergodicity.agents.evaluation.MaxEntIRL(n_features, n_actions, learning_rate=0.01, n_iterations=100)

Bases: object

MaxEntIRL (Maximum Entropy Inverse Reinforcement Learning) Class

This class implements the Maximum Entropy Inverse Reinforcement Learning algorithm, which aims to recover the underlying reward function from observed optimal behavior in a Markov Decision Process (MDP).

n_features

Number of features in the state space.

Type:

int

n_actions

Number of possible actions in the MDP.

Type:

int

learning_rate

Learning rate for the optimization process.

Type:

float

n_iterations

Number of iterations for the optimization process.

Type:

int

reward_weights

Weights representing the reward function.

Type:

np.ndarray

The MaxEntIRL class implements the core algorithm of Maximum Entropy Inverse Reinforcement Learning. It aims to find a reward function that makes the observed behavior appear near-optimal. The algorithm works by iteratively updating the reward weights to maximize the likelihood of the observed trajectories under the maximum entropy distribution.

Key aspects of the implementation:

1. It uses feature expectations to characterize the observed behavior.

2. It computes state visitation frequencies to understand the importance of different states.

3. The optimization process uses the L-BFGS-B algorithm to find the optimal reward weights.

4. The resulting policy is computed using a softmax over Q-values.

This implementation is particularly useful in scenarios where we want to understand the underlying motivations or rewards that drive observed behavior, such as in robotics, autonomous systems, or behavioral economics.

Usage:

irl_model = MaxEntIRL(n_features=5, n_actions=3)

trajectories = […] # List of observed state-action trajectories

learned_rewards = irl_model.fit(trajectories)

# Predict reward for a new state

new_state = np.array([…])

predicted_reward = irl_model.predict_reward(new_state)

Note: This implementation assumes discrete state and action spaces and may require modifications for continuous domains or large-scale problems.

compute_expected_svf(trajectories, policy)

Compute expected state visitation frequency. It estimates the expected frequency of visiting each state under the given policy.

Parameters:

• trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

• policy (Callable) – Policy function mapping states to action probabilities

Returns:

Expected state visitation frequency

Return type:

np.ndarray

compute_gradient(feat_exp, exp_svf)

Compute the gradient for the optimization. It represents the difference between empirical feature expectations and expected state visitation frequency.

Parameters:

• feat_exp (np.ndarray) – Empirical feature expectations

• exp_svf (np.ndarray) – Expected state visitation frequency

Returns:

Gradient vector

Return type:

np.ndarray

compute_policy(trajectories)

Compute the policy based on current reward weights. It uses a softmax over Q-values to derive the action probabilities. The result is a policy that explains the observed behavior.

Parameters:

trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

Returns:

Computed policy

Return type:

np.ndarray

compute_state_visitation_freq(trajectories, policy)

Compute state visitation frequencies. It calculates the frequency of visiting each state under a given policy.

Parameters:

• trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

• policy (Callable) – Policy function mapping states to action probabilities

Returns:

State visitation frequencies

Return type:

np.ndarray

feature_expectations(trajectories)

Compute empirical feature expectations. It is the average feature value observed in the given trajectories.

Parameters:

trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

Returns:

Empirical feature expectations

Return type:

np.ndarray

fit(trajectories)

Fit the IRL model to the given trajectories. The method allows to learn the reward function that explains the observed behavior.

Parameters:

trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

Returns:

Inferred reward

Return type:

np.ndarray

optimize_reward(trajectories)

Optimize the reward function. It uses the L-BFGS-B algorithm to find the optimal reward weights. The result is a reward function that explains the observed behavior.

Parameters:

trajectories (List[List[Tuple[np.ndarray, int]]]) – List of observed state-action trajectories

Returns:

None

Return type:

None

predict_reward(state)

Predict the reward for a given state. It uses the learned reward weights to estimate the reward value.

Parameters:

state (np.ndarray) – Input state for reward prediction

Returns:

Predicted reward value

Return type:

float

class ergodicity.agents.evaluation.UtilityFunction(name: str, func: Callable, initial_params: List[float])

Bases: object

UtilityFunction Class

This class represents a utility function used in decision-making models and optimization problems.

name

The name of the utility function.

Type:

str

func

The actual utility function implementation.

Type:

Callable

initial_params

Initial parameters for the utility function.

Type:

List[float]

fitted_params

Fitted parameters after optimization, if any.

Type:

List[float] or None

nll

Negative log-likelihood of the fitted function, if calculated.

Type:

float or None

The UtilityFunction class encapsulates both the definition and the behavior of a utility function. It supports initial parametrization and subsequent fitting, making it suitable for use in optimization processes where the parameters of the utility function are adjusted based on data.

Usage:

def power_utility(x, beta):

return x ** beta

util_func = UtilityFunction(“Power Utility”, power_utility, [1.0])

# Before fitting

value = util_func(2.0) # Uses initial parameter

# After fitting (assuming fitting has been done elsewhere)

util_func.fitted_params = [0.8]

value = util_func(2.0) # Uses fitted parameter

This class is particularly useful in contexts where multiple utility functions need to be defined, compared, and optimized, such as in economic models or decision theory applications.

class ergodicity.agents.evaluation.UtilityFunctionInference(model_path: str, param_ranges: Dict[str, Dict[str, Tuple[float, float]]], model=None)

Bases: object

UtilityFunctionInference Class

This class is designed for inferring and analyzing utility functions based on agent behavior in stochastic processes. It combines machine learning, Bayesian inference, and economic modeling to understand decision-making patterns.

model

A neural network model loaded from a file, used for decision prediction.

Type:

tf.keras.Model

agent

An agent that uses the loaded model for decision-making.

Type:

NeuralNetworkAgent

process_encoder

Encoder for converting stochastic processes into a format suitable for the model.

Type:

ProcessEncoder

utility_functions

Collection of utility functions to be analyzed.

Type:

List[UtilityFunction]

param_ranges

Ranges of parameters for different stochastic processes.

Type:

Dict[str, Dict[str, Tuple[float, float]]]

mcmc_samples

Stores samples from Markov Chain Monte Carlo simulations for Bayesian inference.

Type:

Dict

regression_model

A regression model for preference prediction, if trained.

Type:

tf.keras.Model or None

regression_history

Training history of the regression model, if available.

Type:

tf.keras.callbacks.History or None

This class provides a comprehensive toolkit for analyzing decision-making behavior in the context of stochastic processes. It supports various methods of utility function inference, including maximum likelihood estimation, Bayesian inference, and inverse reinforcement learning. The class also includes functionality for visualizing results and analyzing feature importance in the decision-making process.

Usage:

ufi = UtilityFunctionInference(‘path/to/model.h5’, param_ranges)

ufi.add_utility_function(UtilityFunction(‘Power’, utility_power, [1.0]))

dataset = ufi.generate_dataset(1000)

choices = ufi.get_agent_choices(dataset)

ufi.fit_utility_functions(dataset, choices)

ufi.plot_utility_functions()

This class is particularly useful for researchers and practitioners in fields such as economics, decision theory, and reinforcement learning, where understanding the underlying utility functions driving agent behavior is crucial.

add_utility_function(utility_function)

Add a utility function to the collection for analysis.

Parameters:

utility_function (UtilityFunction) – UtilityFunction instance to add

Returns:

None

Return type:

None

analyze_feature_importance()

Analyze the importance of different features in the regression model. The method calculates the feature importance based on the weights of the first layer of the model. It helps understand which features have the most influence on the agent’s decision-making process.

Returns:

Dictionary of feature names and their importance scores

Return type:

Dict[str, float]

analyze_feature_interactions()

Analyze the interaction strength between features in the regression model. It is done by calculating the dot product of the weights of the first layer. The resulting matrix shows how features interact with each other in the decision-making process.

Returns:

Interaction strength matrix

Return type:

np.ndarray

bayesian_fit_utility_functions(dataset: List[List[ndarray]], choices: List[int], n_samples: int = 10000, burn_in: int = 1000)

Perform Bayesian inference on utility functions using Metropolis-Hastings sampling. It generates samples from the posterior distribution of the utility function parameters. Provides a distribution of parameter values instead of a single point estimate.

Parameters:

• dataset (List[List[np.ndarray]]) – List of datasets, each containing lists of process trajectories

• choices (List[int]) – List of agent choices

• n_samples (int) – Number of MCMC samples to generate

• burn_in (int) – Number of burn-in samples to discard

Returns:

None

Return type:

None

fit_utility_functions(dataset: List[List[ndarray]], choices: List[int])

Fit the utility functions to the observed choices using maximum likelihood estimation (MLE), assuming the agent maximizes expected utility.

Parameters:

• dataset (List[List[np.ndarray]]) – List of datasets, each containing lists of process trajectories

• choices (List[int]) – List of agent choices

Returns:

None

Return type:

None

generate_choices(n_options: int = 2, n_choices: int = 100)

Generate agent choices among stochastic processes for testing utility functions with the corresponding dataset of processes to choose from.

Parameters:

• n_options – Number of process options per decision instance

• n_choices – Number of choices to generate (number of decision instances)

Return dataset:

A list of datasets, each containing process options

Rtype dataset:

List[List[np.ndarray]]

Returns:

A list of choices where each choice is an index of the selected process

Return type:

List[int]

generate_dataset(n_processes: int, n_options: int = 2, simulate_method=True) → List[List[ndarray]]

Generate a dataset of stochastic processes for analysis.

Parameters:

• n_processes (int) – Number of datasets to generate (i.e., number of decision instances)

• n_options (int) – Number of process options per decision instance

Returns:

List of datasets, each containing process options

Return type:

List[List[np.ndarray]]

get_agent_choices(data: List[ndarray]) → int

Get the agent’s choice based on a dataset of process options.

Parameters:

data (List[np.ndarray]) – A list of process options, each containing process data (np.ndarray)

Returns:

The index of the agent’s choice

Return type:

int

get_process_type(type_code) → str

Get the process type name based on the type code.

Parameters:

type_code (int) – Process type code

Returns:

Process type name

Return type:

str

load_model(model_path: str)

Load a Keras model from a file, handling potential compatibility issues.

Parameters:

model_path (str) – Path to the model file

Returns:

Loaded Keras model

Return type:

tf.keras.Model

metropolis_hastings(utility_function, dataset: List[List[ndarray]], choices: List[int], n_samples: int, burn_in: int) → ndarray

Perform Metropolis-Hastings sampling for Bayesian inference on utility functions. Generates samples from the posterior distribution of the utility function parameters.

Parameters:

• utility_function (UtilityFunction) – Utility function to fit

• dataset (List[List[np.ndarray]]) – List of datasets, each containing lists of process trajectories

• choices (List[int]) – List of agent choices

• n_samples (int) – Number of samples to generate

• burn_in (int) – Number of burn-in samples to discard

Returns:

Array of MCMC samples

Return type:

np.ndarray

negative_log_likelihood(params: List[float], utility_func: Callable, dataset: List[List[ndarray]], choices: List[int]) → float

Calculate the negative log-likelihood of the utility function given the dataset and choices, assuming the agent maximizes expected utility.

Parameters:

• params (List[float]) – Utility function parameters

• utility_func (Callable) – Utility function to evaluate

• dataset (List[List[np.ndarray]]) – List of datasets, each containing lists of process trajectories

• choices (List[int]) – List of agent choices

Returns:

Negative log-likelihood value

Return type:

float

perform_irl(n_processes=1000, n_options=2)

Perform Inverse Reinforcement Learning (IRL) to infer the reward function from agent choices. It uses the MaxEntIRL algorithm to learn the reward weights based on agent behavior. It is designed to understand the underlying reward structure that drives the agent’s decisions.

Parameters:

• n_processes (int) – Number of decision instances to generate for IRL

• n_options (int) – Number of process options per decision instance

Returns:

Inferred reward weights

Return type:

np.ndarray

plot_bayesian_results(x_range: Tuple[float, float] = None)

Plot the fitted utility functions based on Bayesian inference.

Parameters:

x_range (Tuple[float, float] or None) – Range of x values to plot. If None, it will be determined based on the data.

Returns:

None

Return type:

None

plot_parameter_distributions()

Plot the distributions of fitted parameters from Bayesian inference.

Returns:

None

Return type:

None

plot_partial_dependence(feature_index, num_points=100)

Plot the partial dependence of the predicted preference on a selected feature. The partial dependence shows how the predicted preference changes with variations in a single feature. It helps understand the relationship between the feature and the agent’s decision-making process. It is calculated by fixing all other features at a reference point and varying the selected feature.

Parameters:

• feature_index (int) – Index of the feature to analyze

• num_points (int) – Number of points to sample for the selected feature

Returns:

None

Return type:

None

plot_preference_heatmap(process_type: str, param1: str, param2: str, n_points: int = 20)

Plot a heatmap of agent preferences for different parameter values of a process type. The heatmap shows how the agent’s preference changes with different parameter combinations. It allows visualizing the utility landscape for the agent.

Parameters:

• process_type (str) – Type of the stochastic process

• param1 (str) – First parameter to vary

• param2 (str) – Second parameter to vary

• n_points (int) – Number of points to sample for each parameter

Returns:

None

Return type:

None

plot_regression_results()

Plot the results of the regression model training. It includes accuracy and loss curves for both training and validation sets.

Returns:

None

Return type:

None

plot_utility_functions(x_range: Tuple[float, float] = (0, 0.5))

Plot the fitted utility functions for visualization.

Parameters:

x_range (Tuple[float, float]) – Range of x values to plot

Returns:

None

Return type:

None

predict_preference(process_type: str, params: Dict[str, float]) → float

Predict the agent’s preference for a given process type and parameters. It is done by encoding the process and passing it through the regression model.

Parameters:

• process_type (str) – Type of the stochastic process

• params (Dict[str, float]) – Parameters of the stochastic process

Returns:

Predicted preference value

Return type:

float

print_bayesian_results()

Print the mean and standard deviation of the fitted parameters from Bayesian inference.

Returns:

None

print_results()

Print the fitted utility functions and their parameters, including negative log-likelihood.

Returns:

None

Return type:

None

regression_fit(n_processes: int = 10000, n_options: int = 2, test_size: float = 0.2, epochs: int = 100, batch_size: int = 32)

Train a regression model to predict agent preferences based on process parameters. The model is trained using a dataset of stochastic processes and agent choices, assuming the agent maximizes expected utility.

Parameters:

• n_processes (int) – Number of decision instances to generate for training

• n_options (int) – Number of process options per decision instance

• test_size (float) – Fraction of data to use for testing

• epochs (int) – Number of training epochs

• batch_size (int) – Batch size for training

Raises:

ValueError – If input parameters are invalid

Returns:

None

Return type:

None

class ergodicity.agents.evaluation.UtilityFunctionTester(process_class, param_ranges: Dict[str, Tuple[float, float]], utility_functions: Dict[str, Callable])

Bases: object

UtilityFunctionTester Class

This class is designed to test and analyze various utility functions against stochastic processes. It provides tools for generating process parameters, simulating processes, optimizing utility functions, and analyzing the results through statistical and visual methods.

process_class

The class of the stochastic process to be tested.

param_ranges

Ranges for each parameter of the process.

Type:

Dict[str, Tuple[float, float]]

utility_functions

Dictionary of utility functions to be tested.

Type:

Dict[str, Callable]

results

Stores the results of the tests.

Type:

List

This class provides a comprehensive framework for evaluating and comparing different utility functions in the context of stochastic processes. It is particularly useful for researchers and practitioners in fields such as economics, finance, and decision theory, where understanding the performance of utility functions under various stochastic conditions is crucial.

Key features:

1. Automatic generation of process parameters for comprehensive testing.

2. Parallel processing capability for efficient large-scale testing.

3. Advanced statistical analysis including correlation, PCA, and clustering.

4. Visualization tools for intuitive interpretation of results.

Usage:

process_class = BrownianMotion

param_ranges = {‘mu’: (0, 0.5), ‘sigma’: (0.1, 0.5)}

utility_functions = {

‘power’: lambda x, beta: x ** beta, ‘exponential’: lambda x, alpha: 1 - np.exp(-alpha * x), ‘logarithmic’: lambda x, gamma: np.log(1 + gamma * x)

}

tester = UtilityFunctionTester(process_class, param_ranges, utility_functions)

tester.run_tests(n_processes=1000, n_steps=1000)

tester.analyze_results()

tester.plot_optimal_utility_vs_process_params()

This class enables researchers to gain insights into how different utility functions perform under various stochastic process conditions, helping in the selection and refinement of utility models for specific applications.

analyze_results()

Perform statistical analysis on the test results for all utility functions. It includes correlation analysis, PCA, and clustering to understand the relationships. The results are aimed to demonstrate the performance and characteristics of each utility function.

Returns:

None

Return type:

None

calculate_utility(utility_func: Callable, trajectory: ndarray, utility_params: List[float]) → float

Calculate the utility of a given stochastic process instance using a utility function.

Parameters:

• utility_func (Callable) – Utility function to evaluate

• trajectory (np.ndarray) – Trajectory of the stochastic process

• utility_params (List[float]) – Parameters of the utility function

Returns:

Utility value

generate_process_parameters(n_samples: int) → List[Dict[str, float]]

Generate random process parameters within specified ranges.

Parameters:

n_samples (int) – Number of parameter sets to generate

Returns:

List of process parameters

Return type:

List[Dict[str, float

optimize_utility_function(utility_func: Callable, trajectory: ndarray) → Tuple[List[float], float]

Optimize the parameters of a utility function for a given trajectory. It is done by maximizing the utility value using numerical optimization.

Parameters:

• utility_func (Callable) – Utility function to optimize

• trajectory (np.ndarray) – Trajectory of the stochastic process

Returns:

Tuple of optimal parameters and maximum utility value

Return type:

Tuple[List[float], float]

plot_optimal_utility_vs_process_params()

Plot the relationship between process parameters and optimal utility values. It visualizes how the utility functions perform under different process conditions.

Returns:

None

Return type:

None

run_tests(n_processes: int, n_steps: int, n_jobs: int = 1)

Run tests for multiple processes in parallel.

Parameters:

• n_processes (int) – Number of processes to test

• n_steps (int) – Number of steps to simulate for each process

• n_jobs (int) – Number of parallel jobs to run (-1 for all cores)

Returns:

None

Return type:

None

simulate_process(params: Dict[str, float], n_steps: int) → ndarray

Simulate the stochastic process with given parameters.

Parameters:

• params (Dict[str, float]) – Parameters of the process

• n_steps (int) – Number of steps to simulate

Returns:

Trajectory of the process

Return type:

np.ndarray

test_utility_functions(process_params: Dict[str, float], n_steps: int) → Dict

Test all utility functions against a simulated process trajectory. It evaluates each utility function’s performance and optimizes its parameters.

Parameters:

• process_params (Dict[str, float])

• n_steps (int)

Returns:

Dictionary of results for each utility function

Return type:

Dict

ergodicity.agents.evaluation.evaluate_utility_functions(utility_functions, agent, processes, param_ranges, n_process_batches=100, n_options=2, num_instances=1000, select_max=True)

Evaluate utility functions based on agent choices. The utility functions are estimated according to the maximum likelihood of the agent’s choices.

Parameters:

• utility_functions – List of utility functions to evaluate

• agent – An instance of the AgentEvaluation class

• processes – A list of process types

• param_ranges – A dictionary of parameter ranges for each process type

• n_process_batches – Number of process batches to evaluate

• n_options – Number of process options per batch

• num_instances – Number of instances used in the expected utility calculation

• select_max – Whether to select the process with the maximum expected utility or with the minimum

Return type:

List[Dict[str, float]]

Rtype agent:

AgentEvaluation

Rtype processes:

List[Dict[str, str]]

Rtype param_ranges:

Dict[str, Dict[str, Tuple[float, float]]]

Rtype n_process_batches:

int

Rtype n_options:

int

Rtype num_instances:

int

Rtype select_max:

bool

Returns:

Dictionary of likelihood scores for each utility function

Return type:

Dict[str, float]

ergodicity.agents.evaluation.utility_arctan(x: float, k: float) → float

Arctan utility function. The expression is: U(x) = arctan(k * x)

Parameters:

• x (float) – Input value

• k (float) – Utility parameter

Returns:

Utility value

Return type:

float

ergodicity.agents.evaluation.utility_cobb_douglas(x: float, alpha: float, beta: float) → float

Cobb-Douglas utility function. The expression is: U(x) = x^alpha * (1 - x)^beta

Parameters:

• x (float) – Input value

• alpha (float) – Utility parameter corresponding to first term

• beta (float) – Utility parameter corresponding to second term

Returns:

Utility value

Return type:

float

ergodicity.agents.evaluation.utility_exp(x)

ergodicity.agents.evaluation.utility_linear_threshold(x: float, a: float, b: float) → float

Linear threshold utility function. The expression is: U(x) = max(0, a * x - b)

Parameters:

• x (float) – Input value

• a (float) – Utility parameter corresponding to linear term

• b (float) – Utility parameter corresponding to threshold

Returns:

Utility value

Return type:

float

ergodicity.agents.evaluation.utility_log(x)

ergodicity.agents.evaluation.utility_power(x)

ergodicity.agents.evaluation.utility_prospect_theory(x: float, alpha: float, lambda_: float) → float

Prospect Theory utility function.

Parameters:

• x (float) – Input value

• alpha (float) – Utility parameter corresponding to loss aversion

• lambda (float) – Utility parameter corresponding to risk-seeking behavior

Returns:

Utility value

Return type:

float

ergodicity.agents.evaluation.utility_quadratic(x: float, a: float, b: float) → float

Quadratic utility function. The expression is: U(x) = a * x - b * x^2

Parameters:

• x (float) – Input value

• a (float) – Utility parameter corresponding to linear term

• b (float) – Utility parameter corresponding to quadratic term

Returns:

Utility value

Return type:

float

ergodicity.agents.evaluation.utility_sigmoid(x: float, k: float, x0: float) → float

Sigmoid utility function. The expression is: U(x) = 1 / (1 + exp(-k * (x - x0)))

Parameters:

• x (float) – Input value

• k (float) – Utility parameter controlling slope

• x0 (float) – Utility parameter controlling inflection point

Returns:

Utility value

Return type:

float

ergodicity.agents.evolutionary_nn module

evolutionary_nn Submodule Overview

The `evolutionary_nn` submodule combines neural networks with evolutionary strategies to solve problems involving stochastic processes, ergodicity, and time-optimal behaviour. This submodule allows users to create neural network-based agents that evolve over time, optimize their behavior, and select processes to maximize their wealth or other objectives. The agents can be trained using evolutionary algorithms, with or without reinforcement learning techniques.

Key Features:

1. Neural Network Creation:

   • Easily create customizable feedforward neural networks with various configurations.

   • Features include batch normalization, dropout, and several activation functions (ReLU, Leaky ReLU, Tanh, etc.).

   • Supports multiple weight initialization methods (Xavier, He, etc.) and optimizers (Adam, SGD, RMSprop).

2. Agent-Based Evolution:

   • The NeuralNetworkAgent class represents agents with a neural network-based decision-making process.

   • Agents accumulate wealth based on their decisions, and their performance (fitness) is measured by their accumulated wealth.

   • Agents can mutate, reproduce, and be cloned, allowing for evolutionary strategies.

3. Evolutionary Neural Network Training:

   • The EvolutionaryNeuralNetworkTrainer class enables training a population of agents using evolutionary algorithms.

   • Agents are evaluated based on their wealth, and top-performing agents are selected to produce offspring for the next generation.

   • The training process supports wealth sharing and mutation of agents to explore new strategies.

4. Reinforcement Learning with Evolution:

   • The ReinforcementEvolutionaryTrainer class combines evolutionary strategies with reinforcement learning.

   • Agents are trained to select processes that maximize their wealth using reinforcement learning principles.

   • Crossover and mutation are applied to create new agents based on elite performers.

5. Process Encoding:

   • The ProcessEncoder class encodes stochastic processes (e.g., Brownian motion, Geometric Brownian motion) into a numeric format for input into neural networks.

   • This allows agents to process different types of stochastic processes and make decisions based on encoded data.

6. Visualization:

   • Visualize the performance of agents, including wealth evolution over time, using Matplotlib and animations.

   • Generate visualizations of neural network evolution and save the parameters of the best-performing agents to files.

Example Usage:

### Creating and Training Neural Network Agents:

from ergodicity.process.multiplicative import GeometricBrownianMotion, BrownianMotion

from ergodicity.evolutionary_nn import NeuralNetwork, NeuralNetworkAgent, EvolutionaryNeuralNetworkTrainer

# Define stochastic processes

process_types = [GeometricBrownianMotion, BrownianMotion]

processes = generate_processes(100, process_types, param_ranges)

# Initialize the ProcessEncoder

encoder = ProcessEncoder()

# Create a neural network for an agent

net = NeuralNetwork(input_size=11, hidden_sizes=[20, 10], output_size=1)

# Create an agent with the neural network

agent = NeuralNetworkAgent(net)

# Train a population of agents using evolutionary strategies

trainer = EvolutionaryNeuralNetworkTrainer(

population_size=10, input_size=11, hidden_sizes=[20, 10], output_size=1, processes=processes, process_encoder=encoder, process_times=[1.0, 5.0, 10.0],

)

population, history = trainer.train(n_steps=100, save_interval=10)

# Get the best agent

best_agent = max(population, key=lambda agent: agent.accumulated_wealth)

print(f”Best agent accumulated wealth: {best_agent.accumulated_wealth}”)

class ergodicity.agents.evolutionary_nn.DynamicBatchNorm1d(num_features)

Bases: Module

DynamicBatchNorm1d Class

This class implements a dynamic version of 1D Batch Normalization, designed to handle both batch and single-sample inputs. It extends PyTorch’s nn.Module.

bn

Standard PyTorch 1D Batch Normalization layer.

Type:

nn.BatchNorm1d

running_mean

Running mean of the features, not updated during training.

Type:

nn.Parameter

running_var

Running variance of the features, not updated during training.

Type:

nn.Parameter

The DynamicBatchNorm1d class addresses a common issue in batch normalization where single-sample inputs (batch size of 1) can cause problems due to the lack of batch statistics. This implementation provides a solution by using running statistics for single samples, ensuring stable behavior regardless of batch size.

Key Features:

1. Seamless handling of both batch and single-sample inputs.

2. Uses standard BatchNorm1d for batches to leverage its optimizations.

3. Fallback to running statistics for single samples to avoid statistical instability.

Usage:

layer = DynamicBatchNorm1d(num_features=64)

output = layer(input_tensor)

This class is particularly useful in scenarios where the model might receive inputs of varying batch sizes, including single samples, such as in online learning or when processing sequential data of varying lengths.

Note: The running mean and variance are not updated during training in this implementation. For applications requiring adaptive statistics, additional logic for updating these values may be necessary.

forward(x)

Forward pass of the dynamic Batch Normalization layer.

Parameters:

x (torch.Tensor) – Input tensor of shape (batch_size, num_features)

Returns:

Normalized tensor

Return type:

torch.Tensor

class ergodicity.agents.evolutionary_nn.EvolutionaryNeuralNetworkTrainer(population_size: int, hidden_sizes: List[int], processes: List[dict | object], process_encoder: ProcessEncoder, process_times: List[float], input_size: int = 11, output_size: int = 1, mutation_rate: float = 0.1, mutation_scale: float = 0.1, with_exchange: bool = False, top_k: int = 10, exchange_interval: int = 10, initial_wealth: float = 1.0, keep_top_n: int = 50, removal_interval: int = 10, process_selection_share: float = 0.5, output_dir: str = 'output_nn')

Bases: object

EvolutionaryNeuralNetworkTrainer Class

This class implements an evolutionary algorithm for training neural networks to make decisions in stochastic process environments. It manages a population of neural network agents, evolves them over time, and provides comprehensive logging and visualization capabilities.

population_size

The number of agents in the population.

Type:

int

input_size

The size of the input layer for the neural networks.

Type:

int

hidden_sizes

The sizes of the hidden layers.

Type:

List[int]

output_size

The size of the output layer.

Type:

int

processes

The stochastic processes used for training.

Type:

List[Union[dict, object]]

process_encoder

Encoder for the stochastic processes.

Type:

ProcessEncoder

process_times

Time horizons for process simulations.

Type:

List[float]

mutation_rate

Rate of mutation for genetic operations.

Type:

float

mutation_scale

Scale of mutations.

Type:

float

with_exchange

Whether to use population-wide information exchange.

Type:

bool

top_k

Number of top agents to consider in exchange.

Type:

int

exchange_interval

Interval for population-wide information exchange.

Type:

int

initial_wealth

Initial wealth of agents.

Type:

float

keep_top_n

Number of top agents to keep after each removal interval.

Type:

int

removal_interval

Interval for removing underperforming agents.

Type:

int

process_selection_share

Proportion of processes to select in each step.

Type:

float

output_dir

Directory for saving outputs.

Type:

str

This class combines evolutionary algorithms with neural networks to tackle decision-making in stochastic environments. It’s particularly suited for financial modeling, economic simulations, and other domains with complex, uncertain dynamics.

Key Features:

1. Flexible neural network architecture for agents.

2. Support for various stochastic processes as the environment.

3. Evolutionary mechanisms including mutation and reproduction.

4. Option for population-wide information exchange.

5. Comprehensive logging and visualization of training progress.

6. Persistence of best models and training statistics.

Usage:

trainer = EvolutionaryNeuralNetworkTrainer(

population_size=100, input_size=10, hidden_sizes=[64, 32], output_size=1, processes=stochastic_processes, process_encoder=encoder, process_times=[1.0, 2.0, 5.0], with_exchange=True

)

final_population, history = trainer.train(n_steps=1000, save_interval=50)

This class is ideal for researchers and practitioners in fields such as quantitative finance, economics, and artificial intelligence who are interested in evolving adaptive agents for decision-making in complex, stochastic environments.

initialize_population() → List[NeuralNetworkAgent]

Initialize the population of neural network agents.

Returns:

List of NeuralNetworkAgent instances

Return type:

List[NeuralNetworkAgent]

load_best_weights(agent: NeuralNetworkAgent)

Load the best weights into an agent’s network.

Parameters:

agent (NeuralNetworkAgent) – Agent instance

Returns:

None

:rtype

log(message: str)

Log a message to the log file.

Parameters:

message (str) – Message to log

Returns:

None

Return type:

None

reproduce_with_exchange()

Reproduce agents with population-wide information exchange.

Returns:

None

Return type:

None

reproduce_without_exchange()

save_best_weights(best_agent: NeuralNetworkAgent)

Save the weights of the best performing agent.

Parameters:

best_agent (NeuralNetworkAgent) – Best performing agent

Returns:

None

Return type:

None

save_performance_metrics()

Save performance metrics to a CSV file.

Returns:

None

Return type:

None

select_top_agents(k: int) → List[NeuralNetworkAgent]

Select the top k agents based on accumulated wealth.

Parameters:

k (int) – Number of top agents to select

Returns:

List of top agents

Return type:

List[NeuralNetworkAgent]

train(n_steps: int, save_interval: int)

Run the main training loop for the specified number of steps to train the evolutionary neural network.

Parameters:

• n_steps (int) – Number of training steps

• save_interval (int) – Interval for saving metrics and weights

Returns:

Tuple of final population and training history

Return type:

Tuple[List[NeuralNetworkAgent], List[Dict]]

visualize_neural_network_evolution(output_video_path='neural_network_evolution.mp4', output_csv_path='best_agent_params.csv')

Create a video visualization of the neural network evolution and save best agent parameters to CSV.

Parameters:

• output_video_path (str) – Path to save the output video

• output_csv_path (str) – Path to save the CSV file with best agent parameters

Returns:

None

Return type:

None

visualize_performance()

Create separate visualizations for average and max wealth during training.

Returns:

None

Return type:

None

class ergodicity.agents.evolutionary_nn.NeuralNetwork(input_size, hidden_sizes, output_size, activation='relu', output_activation=None, dropout_rate=0.0, batch_norm=False, weight_init='xavier_uniform', learning_rate=0.001, optimizer='adam')

Bases: Module

NeuralNetwork Class

This class implements a flexible and customizable neural network using PyTorch. It provides a wide range of options for network architecture, activation functions, regularization, and optimization.

input_size

Size of the input layer.

Type:

int

hidden_sizes

Sizes of the hidden layers.

Type:

List[int]

output_size

Size of the output layer.

Type:

int

dropout_rate

Dropout rate for regularization.

Type:

float

batch_norm

Whether to use batch normalization.

Type:

bool

weight_init

Weight initialization method.

Type:

str

learning_rate

Learning rate for the optimizer.

Type:

float

optimizer_name

Name of the optimizer to use.

Type:

str

model

The PyTorch sequential model containing all layers.

Type:

nn.Sequential

optimizer

The optimizer for training the network.

Type:

torch.optim.Optimizer

This NeuralNetwork class offers a high degree of flexibility and customization:

1. Supports arbitrary numbers and sizes of hidden layers.

2. Offers multiple activation functions (ReLU, LeakyReLU, Tanh, Sigmoid, ELU).

3. Includes options for dropout and batch normalization for regularization.

4. Provides various weight initialization methods (Xavier, He initialization).

5. Supports different optimizers (Adam, SGD, RMSprop).

6. Includes methods for genetic algorithm-style operations (mutation, cloning).

7. Implements save and load functionality for model persistence.

Usage:

model = NeuralNetwork(

input_size=10, hidden_sizes=[64, 32], output_size=1, activation=’relu’, dropout_rate=0.1, batch_norm=True, weight_init=’he_uniform’, optimizer=’adam’

)

output = model(input_tensor)

model.save(‘model.pth’)

loaded_model = NeuralNetwork.load(‘model.pth’)

This class is particularly useful for experiments involving neural architecture search, evolutionary algorithms, or any scenario requiring dynamic creation and modification of neural networks.

clone()

Create a deep copy of the network.

Returns:

Cloned network

Return type:

NeuralNetwork

forward(x)

Forward pass through the neural network.

Parameters:

x (torch.Tensor) – Input tensor

Returns:

Output tensor

Return type:

torch.Tensor

get_num_parameters()

Return the total number of trainable parameters in the network.

Returns:

Number of parameters

Return type:

int

classmethod load(path)

Load a model from a file.

Parameters:

path (str) – Path to the input file

Returns:

NeuralNetwork instance

Return type:

NeuralNetwork

mutate(mutation_rate=0.1, mutation_scale=0.1)

Apply random mutations to the network parameters. The mutations are applied by adding Gaussian noise to the parameters.

Parameters:

• mutation_rate (float) – Probability of mutating each parameter

• mutation_scale (float) – scale of the mutation (the standard deviation of the Gaussian noise)

Returns:

None

Return type:

None

save(path)

Save the model state and hyperparameters to a file.

Parameters:

path (str) – Path to the output file

Returns:

None

Return type:

None

class ergodicity.agents.evolutionary_nn.NeuralNetworkAgent(neural_network: NeuralNetwork, initial_wealth: float = 1.0)

Bases: object

NeuralNetworkAgent Class

This class represents an agent that uses a neural network to make decisions in a stochastic process environment. It encapsulates the neural network along with methods for process selection, wealth management, and evolutionary operations.

network

The neural network used for decision making.

Type:

NeuralNetwork

wealth

The current wealth of the agent.

Type:

float

accumulated_wealth

The total accumulated wealth over time.

Type:

float

fitness

The fitness score of the agent, based on accumulated wealth.

Type:

float

The NeuralNetworkAgent class is designed to work in environments where decisions are made based on encoded representations of stochastic processes. It’s particularly suited for evolutionary algorithms and reinforcement learning scenarios in financial or economic simulations.

Key features:

1. Decision making using a neural network on encoded process representations.

2. Wealth tracking and accumulation based on process returns.

3. Fitness calculation for use in evolutionary algorithms.

4. Support for genetic operations like mutation and cloning.

5. Persistence through save and load functionality.

Usage:

network = NeuralNetwork(input_size=10, hidden_sizes=[64, 32], output_size=1)

agent = NeuralNetworkAgent(network)

selected_process = agent.select_process(encoded_processes)

agent.update_wealth(process_return)

agent.calculate_fitness()

mutated_agent = agent.clone()

mutated_agent.mutate()

agent.save(‘agent.pth’)

loaded_agent = NeuralNetworkAgent.load(‘agent.pth’)

This class is ideal for simulations where agents need to learn and adapt to complex, stochastic environments, particularly in financial modeling or economic simulations involving decision-making under uncertainty.

calculate_fitness()

Calculate the fitness of the agent based on accumulated wealth.

clone()

Create a clone of the agent with the same network structure but newly initialized weights.

classmethod load(path: str)

Load an agent from a file.

Parameters:

path (str) – Path to the input file

Returns:

NeuralNetworkAgent instance

Return type:

NeuralNetworkAgent

mutate(mutation_rate: float = 0.1, mutation_scale: float = 0.1)

Mutate the agent’s neural network.

reset_wealth()

Reset the agent’s wealth to the initial value.

save(path: str)

Save the agent’s state to a file.

Parameters:

path (str) – Path to the output file

Returns:

None

Return type:

None

select_process(encoded_processes: List[List[float]]) → int

Select a process based on the neural network’s output.

Parameters:

encoded_processes (List[List[float]]) – List of encoded processes

Returns:

Index of the selected process

Return type:

int

update_wealth(process_return: float)

Update the agent’s wealth based on the process return.

Parameters:

process_return (float) – The return from the selected process

class ergodicity.agents.evolutionary_nn.ReinforcementEvolutionaryTrainer(population_size: int, input_size: int, hidden_sizes: List[int], output_size: int, processes: List[dict | object], process_encoder: ProcessEncoder, process_times: List[float], learning_rate: float = 0.001, mutation_rate: float = 0.1, mutation_scale: float = 0.1, rl_interval: int = 10, elite_percentage: float = 0.2, output_dir: str = 'output_nn')

Bases: object

ReinforcementEvolutionaryTrainer Class This class implements a hybrid approach combining reinforcement learning and evolutionary algorithms for training neural network agents in stochastic process environments. It manages a population of agents, evolves them over time, and applies reinforcement learning techniques to improve their decision-making capabilities.

population_size

The number of agents in the population.

Type:

int

input_size

The size of the input layer for the neural networks.

Type:

int

hidden_sizes

The sizes of the hidden layers.

Type:

List[int]

output_size

The size of the output layer.

Type:

int

processes

The stochastic processes used for training.

Type:

List[Union[dict, object]]

process_encoder

Encoder for the stochastic processes.

Type:

ProcessEncoder

process_times

Time horizons for process simulations.

Type:

List[float]

learning_rate

Learning rate for the reinforcement learning updates.

Type:

float

mutation_rate

Rate of mutation for genetic operations.

Type:

float

mutation_scale

Scale of mutations.

Type:

float

rl_interval

Interval for applying reinforcement learning updates.

Type:

int

elite_percentage

Percentage of top-performing agents to consider as elite.

Type:

float

This class combines evolutionary algorithms with reinforcement learning to create a powerful hybrid approach for training agents in complex, stochastic environments. It’s particularly well-suited for financial modeling, economic simulations, and other domains with intricate, uncertain dynamics where both long-term evolution and short-term learning are beneficial.

Key Features:

Flexible neural network architecture for agents.

Support for various stochastic processes as the environment.

Reinforcement learning updates to improve agent performance.

Evolutionary mechanisms including mutation, crossover, and elite selection.

Periodic population renewal based on fitness.

Adaptive learning through a combination of exploration and exploitation.

Usage:

trainer = ReinforcementEvolutionaryTrainer( population_size=100, input_size=10, hidden_sizes=[64, 32], output_size=1, processes=stochastic_processes, process_encoder=encoder, process_times=[1.0, 2.0, 5.0], learning_rate=0.001, mutation_rate=0.1, rl_interval=10 )

final_population = trainer.train(n_steps=1000)

This class is ideal for researchers and practitioners in fields such as quantitative finance, economics, and artificial intelligence who are interested in developing sophisticated, adaptive agents capable of making informed decisions in complex, stochastic environments. The combination of evolutionary algorithms and reinforcement learning provides a robust framework for discovering and refining effective strategies in these challenging domains.

calculate_reward(initial_wealth: float, final_wealth: float) → float

Calculate the reward based on the change in wealth.

Parameters:

• initial_wealth (float) – Initial wealth value

• final_wealth (float) – Final wealth value

Returns:

Reward value

Return type:

float

crossover(parent1: NeuralNetworkAgent, parent2: NeuralNetworkAgent) → NeuralNetworkAgent

Perform crossover between two parent agents to create a child agent. It means that the child agent inherits some parameters from each parent.

Parameters:

• parent1 (NeuralNetworkAgent) – First parent agent

• parent2 (NeuralNetworkAgent) – Second parent agent

Returns:

Child agent

Return type:

NeuralNetworkAgent

initialize_population() → List[NeuralNetworkAgent]

Initialize the population of neural network agents.

Returns:

List of NeuralNetworkAgent instances

Return type:

List[NeuralNetworkAgent]

log(message: str)

Log a message to the log file.

Parameters:

message (str) – Message to log

mutate(agent: NeuralNetworkAgent)

Apply mutation to an agent’s neural network. The mutation is applied to the parameters of the network based on the mutation rate and scale.

Parameters:

agent (NeuralNetworkAgent) – NeuralNetworkAgent instance

Returns:

None

Return type:

None

reinforce(agent: NeuralNetworkAgent, optimizer: Optimizer, reward: float)

Apply reinforcement learning update to an agent. This is done by performing a gradient ascent step on the agent’s neural network.

Parameters:

• agent (NeuralNetworkAgent) – NeuralNetworkAgent instance

• optimizer (optim.Optimizer) – Optimizer for the neural network

• reward (float) – Reward value

Returns:

None

save_best_weights(best_agent: NeuralNetworkAgent)

Save the weights of the best-performing agent.

Parameters:

best_agent (NeuralNetworkAgent) – Best-performing agent

save_performance_metrics()

Save performance metrics to a CSV file.

select_elite(population: List[NeuralNetworkAgent]) → List[NeuralNetworkAgent]

Select the elite agents from the population.

Parameters:

population (List[NeuralNetworkAgent]) – List of NeuralNetworkAgent instances

Returns:

List of elite agents

Return type:

List[NeuralNetworkAgent]

select_process(agent: NeuralNetworkAgent) → Tuple[object, float]

Select a process and time horizon for an agent to interact with.

Parameters:

agent (NeuralNetworkAgent) – NeuralNetworkAgent instance

Returns:

Tuple of process and time

Return type:

Tuple[object, float]

simulate_process(process: object, time: float) → float

Simulate the selected process for the given time horizon.

Parameters:

• process (object) – Stochastic process instance

• time (float) – Time horizon for simulation

Returns:

Final value of the process

Return type:

float

train(n_steps: int)

Run the main training loop for the specified number of steps.

Parameters:

n_steps (int) – Number of training steps

Returns:

List of final agents in the population

Return type:

List[NeuralNetworkAgent]

visualize_neural_network_evolution(output_video_path='neural_network_evolution.mp4', output_csv_path='best_agent_params.csv')

Create a video visualization of the neural network evolution and save best agent parameters to CSV.

Parameters:

• output_video_path (str) – Path to save the output video

• output_csv_path (str) – Path to save the CSV file with best agent parameters

visualize_performance()

Create visualizations for average and maximum fitness during training.

ergodicity.agents.portfolio module

portfolio Submodule Overview

The `portfolio` submodule allows for the simulation and analysis of portfolios composed of multiple stochastic processes. This is particularly useful for studying portfolio dynamics, wealth growth, and the evolution of asset weights over time in stochastic investment environments.

Key Features:

1. Portfolio Simulation:

   • Simulates a portfolio consisting of different stochastic processes (e.g., Geometric Brownian Motion).

   • Dynamically adjusts the weights of each process in the portfolio based on the simulated returns of the processes.

2. Wealth and Weight Dynamics:

   • Tracks the evolution of portfolio wealth and the individual weights of each asset in the portfolio over time.

   • Allows visualization of both wealth and weight dynamics to analyze performance and diversification.

3. Process Integration:

   • Works with any stochastic process object that provides a simulate() method (e.g., GeometricBrownianMotion).

Example Usage:

from ergodicity.portfolio import Portfolio

from ergodicity.process.multiplicative import GeometricBrownianMotion

# Number of processes (e.g., 100 assets)

n = 100

# Create stochastic processes (e.g., GBMs for assets)

processes = []

for i in range(n):

gbm = GeometricBrownianMotion(drift=0.016, volatility=0.3)

processes.append(gbm)

# Initialize portfolio with equal weights for all assets

weights = [1/n] * n

portfolio = Portfolio(processes, weights)

# Simulate portfolio over time

wealth_history, weight_history = portfolio.simulate(timestep=0.5, time_period=1, total_time=1000)

# Visualize portfolio wealth and weight dynamics

portfolio.visualize()

class ergodicity.agents.portfolio.Portfolio(processes: List[Any], weights: List[float])

Bases: object

Portfolio Class

This class represents a portfolio of stochastic processes, simulating their combined behavior over time. It’s designed to model and analyze the dynamics of a diversified investment portfolio in a stochastic environment.

processes

A list of stochastic process objects representing different assets.

Type:

List[Any]

initial_weights

The initial allocation weights for each process, normalized to sum to 1.

Type:

np.array

current_weights

The current allocation weights, updated during simulation.

Type:

np.array

initial_wealth

The initial wealth of the portfolio, set to 1.0.

Type:

float

current_wealth

The current wealth of the portfolio, updated during simulation.

Type:

float

_current_portfolio

The current value of each asset in the portfolio.

Type:

np.array

wealth_history

A record of the portfolio’s wealth over time.

Type:

List[float]

weight_history

A record of the asset weights over time.

Type:

List[np.array]

This class is particularly useful for:

1. Simulating the behavior of a diversified portfolio over time.

2. Analyzing how different stochastic processes (assets) interact within a portfolio.

3. Visualizing the evolution of portfolio wealth and asset allocation.

4. Studying the effects of various rebalancing strategies (implicitly implemented through weight updates).

Usage:

processes = [GeometricBrownianMotion(drift=0.05, volatility=0.2) for _ in range(5)]

weights = [0.2, 0.2, 0.2, 0.2, 0.2]

portfolio = Portfolio(processes, weights)

wealth_history, weight_history = portfolio.simulate(timestep=0.01, time_period=1, total_time=100)

portfolio.visualize()

Note: The simulation assumes that each process has a ‘simulate’ method that returns a time series of values. The portfolio is rebalanced at each ‘time_period’ interval, reflecting a dynamic asset allocation strategy.

simulate(timestep: float, time_period: float, total_time: float)

Simulate the portfolio over time.

Parameters:

• timestep (float) – Simulation timestep

• time_period (float) – Period for recalculating weights and wealth

• total_time (float) – Total simulation time

Returns:

Tuple of wealth history and weight history

Return type:

Tuple[List[float], List

visualize()

Visualize the wealth and weight dynamics of the portfolio.

ergodicity.agents.probability_weighting module

probability_weighting Submodule Overview

The `probability_weighting` submodule provides tools for computing and visualizing probability weighting functions in stochastic processes, particularly focusing on how different drifts and diffusions influence stochastic trajectories under changes of measure (e.g., Girsanov’s theorem).

Key Features:

1. Girsanov’s Theorem:

   • Apply Girsanov’s theorem to transform a probability measure in Geometric Brownian Motion (GBM) and other martingale processes.

   • Adjusts the drift of the process based on the probability weighting function.

2. Stochastic Simulation:

   • Simulate weighted stochastic processes based on the drift and volatility parameters using the adjusted probability density function (PDF).

3. Visualization:

   • Provides functions for simulating and plotting stochastic trajectories under different probability weighting schemes.

Example Usage:

from ergodicity.probability_weighting import gbm_weighting, visualize_weighting

# Parameters for Geometric Brownian Motion (GBM)

initial_mu = 0.05 # Initial drift

sigma = 0.2 # Volatility

# Get the weighted PDF using Girsanov’s theorem

weighted_pdf = gbm_weighting(initial_mu, sigma)

# Visualize the weighted process

new_mu = 0.03 # New drift for visualization

X = visualize_weighting(weighted_pdf, new_mu, sigma, timestep=0.01, num_samples=1000, t=1.0)

ergodicity.agents.probability_weighting.gbm_weighting(initial_mu, sigma, time_horizon=1)

Apply Girsanov’s theorem to transform the probability measure of a Geometric Brownian Motion (GBM) process to a new measure. The new measure is defined by adjusting the drift of the GBM process.

Parameters:

• initial_mu (float) – Initial drift of the GBM process. It corresponds to the intended time average of the process. So you insert the time average to get such probability weighting that taking expected value under the new measure gives the time average.

• sigma (float) – Volatility of the GBM process.

Returns:

Weighted probability density function (PDF) after applying Girsanov’s theorem.

Return type:

sympy.core.add.Add

ergodicity.agents.probability_weighting.martingale_weighting(initial_mu, sigma)

Apply Girsanov’s theorem to transform the probability measure of a martingale process to a new measure. The new measure is defined by adjusting the drift of the martingale process.

Parameters:

• initial_mu – Initial drift of the martingale process.

• initial_mu – float

• sigma – Volatility of the martingale process.

• sigma – float

Returns:

Weighted probability density function (PDF) after applying Girsanov’s theorem.

Return type:

sympy.core.add.Add

ergodicity.agents.probability_weighting.visualize_weighting(weighted_pdf, new_mu, sigma, timestep=0.01, num_samples=1000, t=1.0)

Visualize the weighted stochastic process based on the adjusted drift and volatility parameters.

Parameters:

• weighted_pdf (sympy.core.add.Add) – Weighted probability density function.

• new_mu (float) – New drift parameter.

• sigma (float) – Volatility parameter.

• timestep (float) – Time step for simulation.

• num_samples (int) – Number of sample paths to simulate.

• t (float) – Total time for simulation.

Returns:

Simulated paths.

Return type:

numpy.ndarray

ergodicity.agents.sml module

sml Submodule Overview

The `sml` (Stochastic machine Learning) submodule integrates stochastic processes, utility functions, and machine learning techniques to enable the study and optimization of decision-making processes in non-ergodic systems. It allows users to infer optimal behaviors in uncertain environments by simulating processes, fitting utility functions, and applying machine learning models.

Key Features:

1. Utility Functions:

   • Allows users to define utility functions, which model the preferences or satisfaction of an agent.

   • Utility functions can be optimized to find the parameters that best explain observed decisions.

2. Utility Function Inference:

   • UtilityFunctionInference class fits utility functions to agents’ decisions by minimizing negative log-likelihood or using Bayesian inference.

   • Includes Metropolis-Hastings MCMC sampling for fitting utility functions in a Bayesian framework.

   • Allows the user to analyze, visualize, and compare different utility functions using the provided dataset.

3. Agent-based Process Selection:

   • Agents are created with neural network models that can be trained to select optimal stochastic processes based on encoded inputs.

   • Simulates multiple stochastic processes (e.g., Brownian motion, Geometric Brownian motion) to study agent preferences and process performance.

4. Utility Function Testing:

   • The UtilityFunctionTester class provides tools to test different utility functions against stochastic processes.

   • Generates random process parameters and simulates trajectories, calculating and comparing the utilities of different functions for the same process.

5. Maximum Entropy Inverse Reinforcement Learning (MaxEnt IRL):

   • Enables inference of reward weights from agent behavior using MaxEntIRL.

   • Learns reward weights for different features by optimizing the expected state visitation frequencies to match observed behavior.

6. Machine Learning and Regression:

   • The submodule includes a regression-based approach to predict agent preferences and analyze feature importance using neural networks.

   • Includes tools to visualize model training results, such as loss and accuracy plots, and feature importance analysis.

Example Usage:

# Fitting Utility Functions to an Agent’s Choices:

from ergodicity.sml import UtilityFunctionInference, UtilityFunction

from ergodicity.process.basic import BrownianMotion, GeometricBrownianMotion

# Initialize the inference object

ufi = UtilityFunctionInference(‘path/to/your/model.h5’, param_ranges={

‘BrownianMotion’: {‘mu’: (0, 0.5), ‘sigma’: (0.1, 0.5)}, ‘GeometricBrownianMotion’: {‘mu’: (0, 0.5), ‘sigma’: (0.1, 0.5)}

})

# Add utility functions

ufi.add_utility_function(UtilityFunction(‘Power’, lambda x, beta: x ** beta, [1.0]))

ufi.add_utility_function(UtilityFunction(‘Exponential’, lambda x, alpha: 1 - np.exp(-alpha * x), [1.0]))

# Generate dataset

dataset = ufi.generate_dataset(100)

# Get agent’s choices based on the dataset

choices = ufi.get_agent_choices(dataset)

# Fit the utility functions

ufi.fit_utility_functions(dataset, choices)

# Plot results

ufi.plot_utility_functions()

class ergodicity.agents.sml.NeuralNetworkAgent(model: Model)

Bases: object

NeuralNetworkAgent Class

This class represents an agent that uses a neural network to make decisions in a stochastic process environment. It encapsulates the neural network model along with methods for process selection and wealth management.

model

The neural network model used for decision making.

Type:

tf.keras.Model

wealth

The current wealth of the agent, initialized to 1.0.

Type:

float

performance_history

A list tracking the agent’s wealth over time.

Type:

list

This class is designed for use in reinforcement learning scenarios where an agent learns to select optimal processes in a stochastic environment. The neural network model is used to predict the best process based on encoded representations, and the agent’s performance is tracked through its wealth accumulation.

Key Features:

1. Integration with TensorFlow Keras models for decision making.

2. Wealth tracking to measure agent performance over time.

3. Process selection based on minimizing the predicted value from the neural network.

4. Ability to reset wealth for multiple episodes or experiments.

Usage:

model = tf.keras.Sequential([…]) # Define your neural network model

agent = NeuralNetworkAgent(model)

# In each step of your simulation:

selected_process = agent.select_process(encoded_processes)

process_outcome = simulate_process(selected_process)

agent.update_wealth(process_outcome)

# To start a new episode:

agent.reset_wealth()

reset_wealth()

Reset the agent’s wealth to the initial value (1.0) and clear the performance history.

Returns:

None

Return type:

None

select_process(encoded_processes: ndarray) → int

Select the optimal process based on the encoded processes.

Parameters:

encoded_processes (np.ndarray) – Array of encoded processes.

Returns:

Index of the selected process.

Return type:

int

update_wealth(process_value: float)

Update the agent’s wealth based on the outcome of a selected process.

Parameters:

process_value (float) – The return value of the selected process.

Returns:

None

Return type:

None

ergodicity.agents.sml.create_model(hidden_layers, output_shape)

Create a neural network model with the specified hidden layers and output shape.

Parameters:

• hidden_layers (List[int]) – List of integers specifying the number of units in each hidden layer.

• output_shape (int) – Integer specifying the output shape of the model.

Returns:

A compiled Keras model.

Return type:

tf.keras.Model

ergodicity.agents.sml.generate_dataset(processes: List[Dict[str, Any]], param_ranges: Dict[str, Dict[str, tuple]], num_instances: int, n_simulations: int, output_dir: str = 'output_general', save: bool = False, simulate_method=False) → List[ndarray]

Generate a dataset of simulated processes with random parameters.

Parameters:

• processes (List[Dict[str, Any]]) – List of dictionaries specifying the processes to simulate.

• param_ranges (Dict[str, Dict[str, tuple]]) – Dictionary of parameter ranges for each process type.

• num_instances (int) – Number of instances to simulate for each process.

• n_simulations (int) – Number of simulations to run for each process. It sets how many process objects will be created within the specified parameter ranges.

• output_dir (str) – Output directory to save the results.

• save (bool) – Boolean indicating whether to save the results to files.

• simulate_method (bool) – Boolean indicating whether to use the simulation method or the simulate_until method.

Returns:

List of arrays containing the simulated process data.

Return type:

List[np.ndarray]

ergodicity.agents.sml.pad_array(arr, target_shape)

Pad an array to the target shape with NaN values.

Parameters:

• arr (np.ndarray) – Array to pad.

• target_shape (Tuple[int]) – Target shape to pad the array to.

Returns:

Padded array.

Return type:

np.ndarray

ergodicity.agents.sml.ranked_array(arr)

Rank the elements of an array in descending order.

Parameters:

arr (np.ndarray) – Input array.

Returns:

Array with elements and their corresponding ranks.

Return type:

np.ndarray

ergodicity.agents.sml.save_model(model, filepath)

Save the full model (architecture + weights) to a file.

Parameters:

• model (tf.keras.Model) – Keras model to save.

• filepath (str) – Path to save the model file.

Returns:

None

Return type:

None

ergodicity.agents.sml.save_model_weights(model, filepath)

Save the weights of a Keras model to a file.

Parameters:

• model (tf.keras.Model) – Keras model to save.

• filepath (str) – Path to save the model weights.

Returns:

None

Return type:

None

ergodicity.agents.sml.simulate_process(process_type: str, params: Dict[str, float]) → ndarray

Simulate a stochastic process with the specified parameters.

Parameters:

• process_type (str) – Type of stochastic process to simulate.

• params (Dict[str, float]) – Dictionary of parameters for the stochastic process.

Returns:

Array of simulated process values.

Return type:

np.ndarray

ergodicity.agents.sml.train_agent_time(agent: NeuralNetworkAgent, processes: List[Dict[str, Any]], param_ranges: Dict[str, Dict[str, tuple]], n_episodes: int, n_steps: int, num_instances: int = 10, n_simulations: int = 10, output_dir: str = 'output_general')

Train the agent to select optimal processes in a stochastic environment.

Parameters:

• agent (NeuralNetworkAgent) – NeuralNetworkAgent object to train.

• processes (List[Dict[str, Any]]) – List of process dictionaries to simulate.

• param_ranges (Dict[str, Dict[str, tuple]]) – Dictionary of parameter ranges for each process type.

• n_episodes (int) – Number of episodes to train the agent.

• n_steps (int) – Number of steps per episode.

• num_instances (int) – Number of instances to simulate for each process.

• n_simulations (int) – Number of simulations to run for each process.

• output_dir (str) – Output directory to save the results.

Returns:

Tuple of agent, episode wealth history, episode performance history, min times, and actual min times.

Return type:

Tuple[NeuralNetworkAgent, List[List[float]], List[float], List[float], List[float]]

ergodicity.agents.sml.train_agent_value(agent: NeuralNetworkAgent, processes: List[Dict[str, Any]], param_ranges: Dict[str, Dict[str, tuple]], n_episodes: int, n_steps: int, output_dir: str = 'output_general', ergodicity_transform: bool = False, early_stopping_patience: int = 10, early_stopping_min_delta: float = 0.0001)

Train the agent to predict and select stochastic processes based on their values.

Parameters:

• agent – NeuralNetworkAgent object to train.

• processes – List of process dictionaries to simulate.

• param_ranges – Dictionary of parameter ranges for each process type.

• n_episodes – Number of episodes to train the agent.

• n_steps – Number of steps per episode.

• output_dir – Output directory to save the results.

• ergodicity_transform – Whether to use ergodicity transform of process values.

• early_stopping_patience – Number of episodes with no improvement after which training will be stopped.

• early_stopping_min_delta – Minimum change in MSE to qualify as an improvement.

Returns:

Tuple of agent, episode performance history, and best MSE.

ergodicity.agents.sml.visualize_results(episode_wealth_history, episode_performance_history, min_times, actual_min_times, output_dir)

Visualize the results of the agent training.

Parameters:

• episode_wealth_history (List[np.ndarray]) – List of episode wealth histories.

• episode_performance_history (List[float]) – List of episode performance histories.

• min_times (List[float]) – List of predicted min times.

• actual_min_times (List[float]) – List of actual min times.

• output_dir (str) – Output directory to save the visualizations.

Returns:

None

Return type:

None

ergodicity.agents.sml.worker(args)

Worker function for parallel processing. It simulates a stochastic process with random parameters and returns the results.

Parameters:

args (Tuple) – Tuple of arguments for the worker function.

Returns:

List of results from the worker function.

Return type:

List

ergodicity.agents.sml.x_reach_2(X)

Check if the value of X is greater than or equal to 2.

Parameters:

X (float) – Value of X.

Returns:

Boolean indicating if X is greater than or equal to 2.

Return type:

bool

Module contents

agents Module Overview

The `agents` module provides a comprehensive suite of tools for simulating, analyzing, and evaluating agents interacting with stochastic processes. The module is built with a focus on various decision-making frameworks, optimization strategies, evolutionary algorithms, and portfolio management. It integrates methods from utility theory, reinforcement learning, and stochastic modeling, offering flexible ways to simulate agent behaviors and dynamics in complex environments.

Submodules:

1. `agent_pool.py`:

   • Defines a pool of agents that interact with stochastic processes.

   • Agents in the pool can share wealth, update wealth based on individual dynamics, and simulate multi-agent environments.

   • Inequality measures such as Gini coefficient, Palma ratio, and mean logarithmic deviation are tracked to monitor wealth distribution across the agent population.

2. `agents.py`:

   • Contains core agent models, focusing on decision-making in stochastic environments.

   • Includes base classes for agents interacting with processes, updating wealth, and calculating fitness based on wealth accumulation.

   • Supports agent cloning, mutation, and other operations essential for evolutionary algorithms.

3. `evaluation.py`:

   • Focuses on the evaluation of agents’ decision-making using utility functions.

   • Supports utility function inference, Bayesian fitting, and the use of utility functions in reinforcement learning (RL) contexts.

   • Provides visualization and regression tools to understand how agents’ decisions evolve over time and under different stochastic conditions.

4. `evolutionary_nn.py`:

   • Implements evolutionary neural networks for agents to learn optimal strategies.

   • Provides a framework for evolving agent behavior using neural networks, mutation, and crossover.

   • Includes reinforcement-based evolution for optimizing agent performance, with visualization tools for network evolution and performance tracking.

5. `portfolio.py`:

   • Simulates and manages portfolios of assets that follow stochastic processes (e.g., Geometric Brownian Motion).

   • Enables agents to allocate resources across multiple processes, dynamically adjusting weights and tracking portfolio wealth over time.

   • Provides visualization of portfolio wealth and weight dynamics.

6. `probability_weighting.py`:

   • Applies probability weighting to stochastic processes using Girsanov’s theorem.

   • Adjusts drift and volatility based on changes in the probability measure, simulating weighted processes and visualizing their behavior over time.

   • Useful in risk management, derivative pricing, and understanding the effects of probability distortions on process outcomes.

7. `sml.py`:

   • Provides methods for Stochastic Maximum Likelihood (SML) estimation, utility optimization, and process simulation.

   • Agents can optimize their decisions based on estimated likelihood functions, updating their strategies to maximize utility or wealth in stochastic environments.

   • Includes utility function definitions and optimization routines for modeling agent preferences.

Applications: The `agents` module is designed for use in multi-agent simulations, financial modeling, evolutionary algorithms, and stochastic decision-making research. It supports applications in:

• Financial modeling: Simulating portfolios, agent wealth dynamics, and stochastic processes like Geometric Brownian Motion.

• Multi-agent systems: Tracking wealth distribution, inequality, and cooperative behaviors across agents in a shared environment.

• Reinforcement learning: Training agents to optimize their strategies using neural networks, evolutionary algorithms, and utility function-based evaluation.

• Risk management and probability weighting: Using Girsanov’s theorem to model the effect of probability distortions on stochastic process outcomes.

Example Usage:

```python from ergodicity.agents import agent_pool, portfolio, probability_weighting

# Example: Simulating a portfolio of Geometric Brownian Motion processes

from ergodicity.process.multiplicative import GeometricBrownianMotion

processes = [GeometricBrownianMotion(drift=0.01, volatility=0.2) for _ in range(10)]

weights = [0.1] * 10

my_portfolio = portfolio.Portfolio(processes, weights)

wealth_history, weight_history = my_portfolio.simulate(timestep=0.1, time_period=1.0, total_time=100)

my_portfolio.visualize()

# Example: Using Girsanov’s theorem for probability weighting

weighted_pdf = probability_weighting.gbm_weighting(initial_mu=0.05, sigma=0.2)

probability_weighting.visualize_weighting(weighted_pdf, new_mu=0.03, sigma=0.2, timestep=0.01, num_samples=1000, t=1.0)