Any thoughts on Fortnite On-policy learning? (Minimize the reward loss)

import torch import torch.nn as nn import torch.optim as optim import numpy as np import matplotlib.pyplot as plt # ========================================== # 1. DEEP MODEL ARCHITECTURE (With Hidden Layers) # ========================================== class FortnitePolicyNet(nn.Module): def __init__(self): super(FortnitePolicyNet, self).__init__() # Added a hidden layer with 64 neurons to allow non-linear strategy mapping self.network = nn.Sequential( nn.Linear(10, 24), nn.ReLU(), nn.Linear(24, 12) ) # Safe initialization to prevent probability collapse for layer in self.network: if isinstance(layer, nn.Linear): torch.nn.init.xavier_uniform_(layer.weight) torch.nn.init.zeros_(layer.bias) def forward(self, s): logits = self.network(s) return torch.softmax(logits, dim=-1) # ========================================== # 2. STATE NORMALIZATION HELPER # ========================================== def normalize_state(state): """ Scales the raw state features between 0 and 1 so no single feature dominates """ norm_s = np.array(state, dtype=np.float32).copy() norm_s[0] /= 100.0 # playerHP norm_s[1] /= 100.0 # playerShield norm_s[2] /= 100.0 # enemyHP norm_s[3] /= 100.0 # playersLeft norm_s[4] /= 100.0 # kills # norm_s[5] is already 0 or 1 (inStorm) norm_s[6] /= 30.0 # ammoCount norm_s[7] /= 30.0 # cooldownTime norm_s[8] /= 100.0 # distToSafezone (assume max 100m) norm_s[9] /= 4.0 return norm_s # ========================================== # 3. REWARD FUNCTION # ========================================== def compute_reward(s, a, fps): r = 0.0 if s[3] < 20: r += 10.0 / fps elif s[3] < 50: r += 5.0 / fps elif s[3] < 80: r += 2.0 / fps if s[2] < 25: r += 6.0 / fps elif s[2] < 50: r += 3.0 / fps else: r += 1.5 / fps r += s[4] / fps if s[5] == 1: r -= (s[9] * 3.0) / fps if a == 7 and s[0] == 100: r -= 5.0 elif a == 8 and s[1] > 50: r -= 5.0 elif a == 11 and s[7] >= 1: r -= 2.5 if a == 10: # fire if s[6] > 0 and s[7] <= 0: s[6] -= 1.0 # Spend ammo s[2] = max(0.0, s[2] - 30.0) # Enemy takes damage! if s[2] <= 0: s[4] += 1 # Kill recorded s[2] = 100.0 # Respawn enemy s[3] = max(1, s[3] - 1) # Reduce lobby count elif a == 11: # reload s[6] = 30.0 # Refill ammo s[7] = 15.0 # Cooldown block animation frame delay if a == 10 and s[6] <= 0: r -= 4.0 / fps if s[0] < 25: r -= 7.0 / fps elif s[0] <= 50: r -= 2.0 / fps elif s[0] > 50: r += 1.0 / fps if s[1] > 50: r += 5.0 / fps done = False if s[0] <= 0: r -= 150.0 done = True elif s[3] == 1: r += 200.0 done = True return r, done # ========================================== # 4. ENVIRONMENT SIMULATOR # ========================================== class MockFortniteEnv: def __init__(self): try: user_input = int(input("FPS [Minimum 30, Default 60]: ")) self.fps = float(user_input) if user_input else 60.0 except ValueError: self.fps = 60.0 if self.fps < 30.0: self.fps = 30.0 self.frame_count = 0 def reset(self, initial_state=None): self.frame_count = 0 if initial_state is not None: self.state = np.array(initial_state, dtype=np.float32) else: self.state = np.array([100.0, 0.0, 100.0, 100.0, 0.0, 0.0, 30.0, 0.0, 150.0, 1.0], dtype=np.float32) return self.state def step(self, action): self.frame_count += 1 self.state[3] = max(1, self.state[3] - np.random.choice([0, 1], p=[0.95, 0.05])) if self.state[7] > 0: self.state[7] = max(0.0, self.state[7] - 1.0) if action == 7: # meds self.state[0] = min(100.0, self.state[0] + 20.0) elif action == 8: # shield potion self.state[1] = min(100.0, self.state[1] + 20.0) elif action == 9: # medkit self.state[0] = 100.0 elif action == 10: # fire if self.state[6] > 0 and self.state[7] <= 0: self.state[6] -= 1.0 self.state[2] = max(0.0, self.state[2] - 30.0) if self.state[2] <= 0: self.state[4] += 1 self.state[2] = 100.0 self.state[3] = max(1, self.state[3] - 1) elif action == 11: # reload self.state[6] = 30.0 self.state[7] = 15.0 if np.random.rand() < 0.05: self.state[0] = max(0.0, self.state[0] - 20.0) reward, done = compute_reward(self.state, action, self.fps) if self.frame_count >= 500: done = True return self.state.copy(), reward, done, self.fps # ========================================== # 5. MAIN TRAINING LOOP # ========================================== def train_reinforce(discount_c, lr, max_epochs): env = MockFortniteEnv() policy = FortnitePolicyNet() optimizer = optim.Adam(policy.parameters(), lr=lr) # FIX 1: Initialize the tracking list before the training loop starts epoch_rewards = [] for epoch in range(1, max_epochs + 1): state = env.reset() saved_log_probs = [] saved_rewards = [] saved_entropies = [] done = False while not done: # FIXED: Normalize the state vector before passing it to the network norm_state = normalize_state(state) state_t = torch.FloatTensor(norm_state) probs = policy(state_t) action_distribution = torch.distributions.Categorical(probs) action = action_distribution.sample() saved_log_probs.append(action_distribution.log_prob(action)) saved_entropies.append(action_distribution.entropy()) state, reward, done, fps = env.step(action.item()) saved_rewards.append(reward) # FIX 2: Capture and save the total raw reward achieved in this epoch total_raw_reward = sum(saved_rewards) epoch_rewards.append(total_raw_reward) T = len(saved_rewards) returns = np.zeros(T) cumulative_return = 0.0 for t in reversed(range(T)): cumulative_return = saved_rewards[t] + (discount_c * cumulative_return) returns[t] = cumulative_return returns = torch.FloatTensor(returns) if len(returns) > 1: returns = (returns - returns.mean()) / (returns.std() + 1e-8) loss = [] for log_prob, G_t, entropy in zip(saved_log_probs, returns, saved_entropies): loss.append(-log_prob * G_t - 0.01 * entropy) loss = torch.stack(loss).sum() optimizer.zero_grad() loss.backward() optimizer.step() if epoch % 25 == 0 or epoch == 1: # FIX 3: Changed T/fps to T/env.fps to eliminate potential scope/unbound issues print(f"Epoch {epoch:4d}/{max_epochs} | Trajectory Length: {round(T/env.fps, 2)} s | Total reward: {total_raw_reward:7.2f}") # ========================================== # 6. DETERMINISTIC TESTING EVALUATION # ========================================== def evaluate_policy(policy, env): print("n==========================================") print(" LAUNCHING EVALUATION TEST EPISODE ") print("==========================================") s_0 = [100.0, 25.0, 95.0, 10.0, 2.0, 0.0, 30.0, 0.0, 0.0, 3.0] state = np.array(env.reset(initial_state=s_0), dtype=np.float32) action_names = ["nothing", "forward", "back", "left", "right", "jump", "crouch", "meds", "shield potion", "medkit", "fire", "reload"] saved_rewards = [] done = False policy.eval() with torch.no_grad(): while not done: norm_state = normalize_state(state) state_t = torch.FloatTensor(norm_state).unsqueeze(0) probs = policy(state_t) # --- FORCE FIRST ACTION TO FIRE --- if env.frame_count == 0: action = 10 # Hardcode index 10 (fire) for the very first frame else: action = torch.argmax(probs, dim=-1).item() # ---------------------------------- t = env.frame_count print(f"{round(t / env.fps, 2):4f} s | HP: {float(state[0]):3.0f} | Shield: {float(state[1]):3.0f} | Enemy HP: {float(state[2]):3.0f} | Ammo: {int(state[6]):2d} -> Action: {action_names[action]}") next_state, reward, done, current_fps = env.step(action) state = np.array(next_state, dtype=np.float32) saved_rewards.append(reward) print("n--- Evaluation Testing Summary ---") print(f"Total Test Frames Processed: {len(saved_rewards)}") print(f"Total Raw Reward Accumulated: {sum(saved_rewards):7.2f}") # ========================================== # 7. EXPLICIT INITIALIZATION BLOCK # ========================================== if __name__ == "__main__": print("--- Fortnite Policy Gradient RL Initialization ---") try: user_epochs = input("Enter max_epochs [Default 1000]: ").strip() max_epochs = int(user_epochs) if user_epochs else 1000 user_lr = input("Enter learning rate (lr) [Default 0.01]: ").strip() lr = float(user_lr) if user_lr else 0.01 user_discount = input("Enter discount factor (c) [Default 0.99]: ").strip() discount_c = float(user_discount) if user_discount else 0.99 if not (0.0 < discount_c <= 1.0): discount_c = 0.99 except ValueError: max_epochs = 1000 lr = 0.01 discount_c = 0.99 env = MockFortniteEnv() policy = FortnitePolicyNet() optimizer = optim.Adam(policy.parameters(), lr=lr) # ------------------------------------------------------ # Run the separate loops cleanly using the globally initialized objects train_reinforce(discount_c, lr, max_epochs) evaluate_policy(policy, env)