PPO Agent refuses to place orders in custom Gymnasium Inventory Environment (stuck in local minimum)
I am building an RL-based Inventory Management System using Python, Gymnasium, and Stable-Baselines3 (PPO). The environment simulates a single SKU (Item-Centric approach).
The agent's goal is to minimize total cost (Holding + Stockout + Fixed Order).
The Problem: With standard PPO hyperparameters (ent_coef=0.01 to 0.05, gamma=0.99), the agent gets stuck in a local minimum: it completely refuses to place any orders. It realizes that paying the fixed order cost hurts the immediate reward, so it just stops ordering entirely and eats the stockout penalties.
When I drastically increase the entropy coefficient (ent_coef=0.3), the agent acts randomly enough to accidentally stumble onto a strategy where it orders massive amounts, eliminates stockouts, and beats the baseline. However, 0.3 is too high for stable long-term training, and when I lower it back to standard values, it reverts to not ordering.
The Math (Why I think it's doing this): For the "Powerbank" SKU:
fixed_order_cost= RM 150.00stockout_cost= RM 34.00 per unitdaily_demand= 3.0 unitsMax stockout penalty per day = 3 units * RM 34 = RM 102.00
Because the daily stockout penalty (102) is lower than the fixed order cost (150), a cautious PPO agent mathematically prefers to do nothing. It doesn't understand the long-term compounding pain of stockouts.
What I have tried:
Tuning
gamma(0.95 and 0.99) to give the agent longer-term vision.Normalizing the reward by dividing by the
max_theoretical_daily_cost.Bumping
ent_coefto0.3(works, but too unstable).
My Question: How do I adjust the environment logic or reward function to force a standard PPO agent to learn that placing orders is necessary, without relying on massive forced exploration (ent_coef=0.3)?
Is there a standard RL technique to penalize "doing nothing" in inventory environments, or do I need to add an artificial "service level / goodwill" penalty if stock hits zero?
import gymnasium as gym
import numpy as np
from gymnasium import spaces
class InventoryEnv(gym.Env):
"""
Item-Centric Inventory Environment for RL.
This environment simulates the inventory dynamics of a SINGLE SKU at a time.
It is designed to be agnostic to the data source (JSON for testing, SQLite for production) by accepting a dictionary configuration upon initialization.
"""
def __init__(self, product_config: dict, warehouse_config: dict):
"""
Args:
product_config (dict): Contains SKU-specific details (demand, costs, volume, lead time).
warehouse_config (dict): Contains global warehouse constraints (max volume, holding cost rate).
"""
super().__init__()
# =====================================================================
# 1. CONFIGURATION INJECTION
# This approach decouples the environment from the data source.
# You can pass a JSON dictionary now, and a SQLite query result later.
# =====================================================================
self.product = product_config
self.warehouse = warehouse_config
# Unpack product variables for cleaner code later
# Note: max_warehouse_vol is the *remaining* volume available to this SKU.
self.daily_demand = self.product["daily_demand"]
self.unit_vol_cbm = self.product["unit_vol_cbm"]
self.base_lead_time = self.product["base_lead_time"]
self.lead_time_var = self.product["lead_time_var"]
self.fixed_order_cost = self.product["fixed_order_cost"]
self.stockout_cost = self.product["stockout_cost"]
self.max_warehouse_vol = self.product["max_warehouse_vol"]
# Unpack warehouse variables
self.holding_cost_per_cbm_per_day = self.warehouse["holding_cost_per_cbm_per_day"]
self.max_days = 365
# =====================================================================
# 2. DERIVED VARIABLES (Calculated from the profile data)
# =====================================================================
# Max physical units of THIS product that can fit in the allocated warehouse space
self.max_stock_units = int(self.max_warehouse_vol / self.unit_vol_cbm)
# Max possible order quantity (cannot exceed physical space)
self.max_order_qty = self.max_stock_units
# Reward Normalization Scale: Theoretical max daily cost if everything goes wrong.
# Max Holding (full warehouse) + Max Stockout (demand exceeds max stock) + Fixed Order Cost
max_daily_holding = self.max_warehouse_vol * self.holding_cost_per_cbm_per_day
max_daily_stockout = self.daily_demand * self.stockout_cost
# Using a slightly inflated denominator to ensure reward stays between -1 and 0
self.max_theoretical_daily_cost = max_daily_holding + max_daily_stockout + self.fixed_order_cost
# =====================================================================
# 3. SPACES DEFINITION
# =====================================================================
# Observation: [stock, in_transit, days_left, demand_rate, available_vol, unit_vol]
self.observation_space = spaces.Box(
low=np.array([0, 0, 0, 0, 0, 0], dtype=np.float32),
high=np.array([
self.max_stock_units, # Max stock
self.max_order_qty, # Max in-transit
self.base_lead_time + self.lead_time_var, # Max possible days left
self.daily_demand * 2, # Demand rate (padded for safety)
self.max_warehouse_vol, # Max available volume
self.unit_vol_cbm # Unit volume
], dtype=np.float32),
dtype=np.float32
)
# Action: Continuous value between 0.0 and 1.0
# Represents the percentage of available warehouse space to order.
self.action_space = spaces.Box(low=0.0, high=1.0, shape=(1,), dtype=np.float32)
def reset(self, seed=None, options=None):
super().reset(seed=seed)
# 14-day "Warm Start" initial stock
self.current_stock = 14.0 * self.daily_demand
self.in_transit = 0.0
self.days_left = 0
self.current_day = 0
# Trackers for evaluation metrics
self.ep_holding_cost = 0.0
self.ep_stockout_cost = 0.0
self.ep_fixed_cost = 0.0
return self._get_obs(), {}
def _get_obs(self):
"""Constructs the observation array for the agent."""
current_vol = (self.current_stock + self.in_transit) * self.unit_vol_cbm
available_vol = max(0.0, self.max_warehouse_vol - current_vol)
return np.array([
self.current_stock,
self.in_transit,
float(self.days_left),
self.daily_demand,
available_vol,
self.unit_vol_cbm
], dtype=np.float32)
def step(self, action):
# =====================================================================
# 1. ACTION TRANSLATION (Continuous to Discrete Order Qty)
# =====================================================================
# The neural network outputs [0.0, 1.0]. We scale this to physical units.
current_vol = (self.current_stock + self.in_transit) * self.unit_vol_cbm
available_vol = max(0.0, self.max_warehouse_vol - current_vol)
max_fit_units = int(available_vol / self.unit_vol_cbm)
# Translate percentage to actual order quantity
order_qty = int(action[0] * max_fit_units)
# =====================================================================
# 2. INCOMING DELIVERIES
# =====================================================================
if self.days_left > 0:
self.days_left -= 1
if self.days_left == 0:
self.current_stock += self.in_transit
self.in_transit = 0.0
# =====================================================================
# 3. PLACE NEW ORDERS
# =====================================================================
daily_fixed_cost = 0.0
# Constraint: Can only order if no order currently in transit
if order_qty > 0 and self.days_left == 0:
self.in_transit = float(order_qty)
# Stochastic Lead Time
lt_min = max(1, self.base_lead_time - self.lead_time_var)
lt_max = self.base_lead_time + self.lead_time_var
actual_lt = np.random.randint(lt_min, lt_max + 1)
self.days_left = actual_lt
daily_fixed_cost = self.fixed_order_cost
# =====================================================================
# 4. SIMULATE DEMAND & STOCKOUTS
# =====================================================================
demand = np.random.poisson(self.daily_demand)
stockout_units = max(0, demand - self.current_stock)
self.current_stock = max(0.0, self.current_stock - demand)
daily_stockout_cost = stockout_units * self.stockout_cost
# =====================================================================
# 5. CALCULATE HOLDING COST
# =====================================================================
end_vol = self.current_stock * self.unit_vol_cbm
daily_holding_cost = end_vol * self.holding_cost_per_cbm_per_day
# Accumulate for logging
self.ep_holding_cost += daily_holding_cost
self.ep_stockout_cost += daily_stockout_cost
self.ep_fixed_cost += daily_fixed_cost
# =====================================================================
# 6. REWARD CALCULATION (Normalized per-product)
# =====================================================================
total_daily_cost = daily_holding_cost + daily_stockout_cost + daily_fixed_cost
reward = -(total_daily_cost / self.max_theoretical_daily_cost)
# =====================================================================
# 7. ADVANCE TIME
# =====================================================================
self.current_day += 1
done = self.current_day >= self.max_days
truncated = False
info = {
"holding_cost": daily_holding_cost,
"stockout_cost": daily_stockout_cost,
"fixed_cost": daily_fixed_cost,
"order_placed": order_qty,
"stockout_units": stockout_units
}
return self._get_obs(), reward, done, truncated, info
# train.py
import os
import numpy as np
import torch
from stable_baselines3 import PPO
from stable_baselines3.common.vec_env import DummyVecEnv, VecNormalize
from stable_baselines3.common.monitor import Monitor
from stable_baselines3.common.env_checker import check_env
# Import your environment and data
from demo import InventoryEnv
from config_data import WAREHOUSE_CONFIG, PRODUCT_PROFILES
# =====================================================================
# 1. THE (s, Q) BASELINE AGENT
# =====================================================================
class BaselineAgent:
def __init__(self, s_point, Q_qty):
self.s = s_point
self.Q = Q_qty
def predict(self, obs, deterministic=True):
# obs is a 1D array: [stock, in_transit, days_left, demand_rate, available_vol, unit_vol_cbm]
current_stock = obs[0]
in_transit = obs[1]
inventory_position = current_stock + in_transit
if inventory_position < self.s:
available_vol = obs[4]
unit_vol = obs[5]
if available_vol <= 0 or unit_vol <= 0:
return np.array([0.0]), None
max_fit_units = available_vol / unit_vol
# The Baseline must also obey the [0.0, 1.0] action space rules
action_percentage = min(1.0, self.Q / max_fit_units)
return np.array([action_percentage]), None
else:
return np.array([0.0]), None
# =====================================================================
# 2. EVALUATION FUNCTION
# =====================================================================
def evaluate_agent(agent, env, agent_name="Agent"):
# Gymnasium API: reset() returns (obs, info)
obs, info = env.reset()
total_reward = 0
done = False
hold_cost, stockout_cost, fixed_cost = 0, 0, 0
while not done:
action, _ = agent.predict(obs, deterministic=True)
# Gymnasium API: step() returns 5 values (obs, reward, terminated, truncated, info)
obs, reward, terminated, truncated, info = env.step(action)
# Episode is done if terminated (e.g., 365 days reached) or truncated (e.g., time limit)
done = terminated or truncated
total_reward += reward
hold_cost += info["holding_cost"]
stockout_cost += info["stockout_cost"]
fixed_cost += info["fixed_cost"]
print(f"{agent_name:15s} | Reward: {total_reward:8.2f} | Hold: RM{hold_cost:7.2f} | "
f"Stockout: RM{stockout_cost:7.2f} | Fixed: RM{fixed_cost:6.2f}")
# =====================================================================
# 3. MAIN EXECUTION BLOCK
# =====================================================================
if __name__ == "__main__":
product_name = "powerbank"
prod_config = PRODUCT_PROFILES[product_name]
# --- A. CHECK ENVIRONMENT ---
print(f"Checking environment consistency for {product_name}...")
check_env(InventoryEnv(prod_config, WAREHOUSE_CONFIG))
print("Environment is valid!\n")
# --- B. SETUP ENVIRONMENT & NORMALIZATION ---
# We wrap the env in Monitor so SB3 tracks episode rewards for Tensorboard
env = DummyVecEnv([lambda: Monitor(InventoryEnv(prod_config, WAREHOUSE_CONFIG))])
# VecNormalize scales observations and rewards for the neural network
env = VecNormalize(env, norm_obs=True, norm_reward=True, clip_obs=10.)
# --- C. INITIALIZE PPO ---
print("Initializing PPO Agent...")
model = PPO(
policy="MlpPolicy",
env=env,
learning_rate=3e-4,
gamma=0.99,
n_steps=2048,
batch_size=64,
n_epochs=10,
ent_coef=0.1,
policy_kwargs=dict(
net_arch=[64, 64],
activation_fn=torch.nn.ReLU
),
verbose=1,
tensorboard_log="./ppo_inventory_tensorboard/"
)
# --- D. TRAIN THE AGENT ---
print("\nStarting training...")
model.learn(total_timesteps=1_000_000)
print("Training finished!\n")
# --- E. EVALUATION ---
print("="*70)
print(f"EVALUATION: 5 Test Runs for {product_name.upper()} (365 Days Each)")
print("="*70)
# Create a raw evaluation environment (no VecNormalize, so we get real RM values)
eval_env = InventoryEnv(prod_config, WAREHOUSE_CONFIG)
# 1. Evaluate Baseline
if product_name == "powerbank":
baseline_agent = BaselineAgent(s_point=20, Q_qty=50)
else:
baseline_agent = BaselineAgent(s_point=80, Q_qty=300)
for run in range(5):
evaluate_agent(baseline_agent, eval_env, agent_name=f"Baseline Run {run+1}")
# 2. Evaluate PPO
for run in range(5):
evaluate_agent(model, eval_env, agent_name=f"PPO Run {run+1}")
print("="*70)