Integration of early Dropped-Jobs, RNG Seeding and Workload generator & alternative "prices" script.

prices_deterministic.py

@@ -0,0 +1,151 @@
+import numpy as np
+import matplotlib.pyplot as plt
+from collections import deque
+
+class Prices:
+    ELECTRICITY_PRICE_BASE = 20
+    PERCENTILE_MIN = 1
+    PERCENTILE_MAX = 99
+    PREDICTION_WINDOW = 24
+    HISTORY_WINDOW = 24
+
+    def __init__(self, external_prices=None):
+        self.original_prices = external_prices
+        self.external_prices = None
+
+        self.price_shift = 0
+        self.price_index = 0
+        self.price_history = deque(maxlen=self.HISTORY_WINDOW)
+        self.predicted_prices = None
+
+        if self.original_prices is not None:
+            prices = np.asarray(self.original_prices, dtype=np.float32)
+            if prices.size == 0:
+                raise ValueError("external_prices must be a non-empty sequence")
+
+            min_price = float(np.min(prices))
+            if min_price < 1:
+                self.price_shift = 1 - min_price
+                prices = prices + self.price_shift
+
+            self.external_prices = prices
+            self.MIN_PRICE = float(np.percentile(self.external_prices, self.PERCENTILE_MIN))
+            self.MAX_PRICE = float(np.percentile(self.external_prices, self.PERCENTILE_MAX))
+        else:
+            self.external_prices = None
+            # Keep your defaults for normalization bounds
+            self.MAX_PRICE = 24
+            self.MIN_PRICE = 16
+
+        # IMPORTANT: initialize state in one place
+        self.reset(start_index=0)
+
+    # ---------- core price model (pure) ----------
+    def _synthetic_price_at(self, t: int) -> float:
+        # deterministic daily sinusoid
+        base = self.ELECTRICITY_PRICE_BASE
+        return float(max(1.0, base * (1 + 0.2 * np.sin((t % 24) / 24 * 2 * np.pi))))
+
+    def get_real_price(self, shifted_price):
+        return shifted_price - self.price_shift
+
+    def reset(self, start_index: int = 0):
+        """Reset internal timeline/state to episode start.
+
+        start_index is the index in external_prices for the *first* element
+        of the 24h prediction window.
+        """
+        self.price_history = deque(maxlen=self.HISTORY_WINDOW)
+
+        if self.external_prices is not None:
+            n = len(self.external_prices)
+            start_index = start_index % n
+
+            # 24-hour prediction window starting at start_index, with wrap
+            idxs = (np.arange(self.PREDICTION_WINDOW, dtype=np.int64) + start_index) % n
+            self.predicted_prices = self.external_prices[idxs].astype(np.float32, copy=True)
+
+            # next unseen price *after* the window
+            self.price_index = (start_index + self.PREDICTION_WINDOW) % n
+        else:
+            # synthetic mode
+            self.price_index = self.PREDICTION_WINDOW
+            self.predicted_prices = np.array(
+                [self._synthetic_price_at(i) for i in range(self.PREDICTION_WINDOW)],
+                dtype=np.float32,
+            )
+
+    # ---------- *stateful* stepping (used by env.step) ----------
+    def get_next_price(self):
+        if self.external_prices is not None:
+            new_price = float(self.external_prices[self.price_index % len(self.external_prices)])
+        else:
+            new_price = self._synthetic_price_at(self.price_index)
+
+        self.price_index += 1
+
+        self.price_history.append(new_price)
+
+        return new_price
+
+    def get_price_context(self):
+        history_avg = float(np.mean(self.price_history)) if self.price_history else None
+        future_avg = float(np.mean(self.predicted_prices))
+        return history_avg, future_avg
+
+    def advance_and_get_predicted_prices(self): # Changed name for readability
+        new_price = self.get_next_price()
+        self.predicted_prices = np.roll(self.predicted_prices, -1)
+        self.predicted_prices[-1] = new_price
+        return self.predicted_prices.copy()
+
+    # ---------- NON-MUTATING utilities ----------
+    def _generated_prices_for_stats(self, n: int):
+        # Do NOT touch price_index/history here.
+        return np.array([self._synthetic_price_at(i) for i in range(n)], dtype=np.float32)
+
+    def get_price_stats(self, use_original=False):
+        if use_original and self.original_prices is not None:
+            prices = np.asarray(self.original_prices, dtype=np.float32)
+        elif self.external_prices is not None:
+            prices = np.asarray(self.external_prices, dtype=np.float32)
+        else:
+            prices = self._generated_prices_for_stats(24 * 7 * 52)
+
+        return {
+            'min': float(np.min(prices)),
+            'max': float(np.max(prices)),
+            'mean': float(np.mean(prices)),
+            'median': float(np.median(prices)),
+            'std': float(np.std(prices)),
+            f'{self.PERCENTILE_MIN}th_percentile': float(np.percentile(prices, self.PERCENTILE_MIN)),
+            f'{self.PERCENTILE_MAX}th_percentile': float(np.percentile(prices, self.PERCENTILE_MAX)),
+            'price_shift': float(self.price_shift),
+        }
+
+    def plot_price_histogram(self, num_bins=50, save_path=None, use_original=False):
+        if use_original and self.original_prices is not None:
+            prices = np.asarray(self.original_prices, dtype=np.float32)
+            price_type = "Original"
+        elif self.external_prices is not None:
+            prices = np.asarray(self.external_prices, dtype=np.float32)
+            price_type = "Shifted" if self.price_shift else "Original"
+        else:
+            prices = self._generated_prices_for_stats(24 * 7 * 52)
+            price_type = "Generated"
+
+        plt.figure(figsize=(10, 6))
+        plt.hist(prices, bins=num_bins, edgecolor='black')
+        plt.title(f'Distribution of Electricity Prices ({price_type})')
+        plt.xlabel('Price ($/MWh)')
+        plt.ylabel('Frequency')
+        plt.axvline(np.percentile(prices, self.PERCENTILE_MIN), color='r', linestyle='dashed', linewidth=2, label=f'{self.PERCENTILE_MIN}th Percentile')
+        plt.axvline(np.percentile(prices, self.PERCENTILE_MAX), color='g', linestyle='dashed', linewidth=2, label=f'{self.PERCENTILE_MAX}th Percentile')
+        plt.axvline(np.mean(prices), color='b', linestyle='dashed', linewidth=2, label='Mean Price')
+        plt.legend()
+
+        if save_path:
+            plt.savefig(save_path)
+        else:
+            plt.show()
+        plt.close()

sampler_hourly.py

@@ -1,7 +1,6 @@
 import re
 import datetime
 import math
-import random
 from collections import defaultdict
 import numpy as np
 
@@ -100,7 +99,7 @@ def parse_jobs(self, filepath):
             zero_pct = (dist["job_count"].count(0) / len(dist["job_count"]) * 100) if dist["job_count"] else 0
             print(f"  Hour {hour:2d}: avg={avg_count:.1f} jobs/hour, {zero_pct:.0f}% zero-job samples, {len(dist['durations'])} total jobs")
 
-    def sample(self, hour_of_day):
+    def sample(self, hour_of_day: int, rng, max_jobs: int | None = None):
         """
         Sample jobs for a given hour of day.
 
@@ -119,17 +118,20 @@ def sample(self, hour_of_day):
         dist = self.hour_distributions[hour_of_day]
 
         # Sample number of jobs for this hour (can be 0)
-        num_jobs = random.choice(dist["job_count"])
+        num_jobs = rng.choice(dist["job_count"])
 
-        if num_jobs == 0:
+        if num_jobs <= 0:
             return []
 
-        # Sample characteristics for each job
+        if max_jobs is not None:
+            num_jobs = min(num_jobs, int(max_jobs))
+            if num_jobs <= 0:
+                return []
         jobs = []
         for _ in range(num_jobs):
-            duration = random.choice(dist["durations"])
-            nodes = random.choice(dist["nodes"])
-            cores = random.choice(dist["cores_per_node"])
+            duration = rng.choice(dist["durations"])
+            nodes = rng.choice(dist["nodes"])
+            cores = rng.choice(dist["cores_per_node"])
 
             jobs.append({
                 "duration": duration,

test_determinism_workloadgen.py

@@ -0,0 +1,76 @@
+# determinism test for workload generator
+import numpy as np
+from workloadgen import WorkloadGenerator, WorkloadGenConfig
+
+def jobs_to_tuples(jobs):
+    return [(j.duration, j.nodes, j.cores_per_node) for j in jobs]
+
+def check_determinism(cfg, seed=12345, hours=50):
+    g1 = WorkloadGenerator(cfg)
+    g2 = WorkloadGenerator(cfg)
+
+    r1 = np.random.default_rng(seed)
+    r2 = np.random.default_rng(seed)
+
+    for h in range(hours):
+        a = jobs_to_tuples(g1.sample(h, r1))
+        b = jobs_to_tuples(g2.sample(h, r2))
+        assert a == b, f"Determinism failed at hour={h}: {a[:3]} vs {b[:3]}"
+    print("[OK] determinism")
+
+def check_bounds(cfg, seed=1, hours=50):
+    gen = WorkloadGenerator(cfg)
+    rng = np.random.default_rng(seed)
+
+    for h in range(hours):
+        jobs = gen.sample(h, rng)
+        assert len(jobs) <= cfg.max_new_jobs_per_hour
+        if cfg.hard_cap_jobs is not None:
+            assert len(jobs) <= cfg.hard_cap_jobs
+
+        for j in jobs:
+            assert cfg.min_duration <= j.duration <= cfg.max_duration
+            assert cfg.min_nodes <= j.nodes <= cfg.max_nodes
+            assert cfg.min_cores <= j.cores_per_node <= cfg.max_cores
+
+    print("[OK] bounds & caps")
+
+def check_distribution_smoke(cfg, seed=7, hours=2000):
+    gen = WorkloadGenerator(cfg)
+    rng = np.random.default_rng(seed)
+
+    counts = np.array([len(gen.sample(h, rng)) for h in range(hours)], dtype=np.int32)
+    mean = float(counts.mean())
+    p0 = float((counts == 0).mean())
+
+    print(f"[INFO] arrivals={cfg.arrivals} mean={mean:.2f} p0={p0:.3f} max={counts.max()}")
+
+    if cfg.arrivals == "flat":
+        # Flat mode: mean should be close to flat_jobs_per_hour (with jitter variation)
+        expected = cfg.flat_jobs_per_hour
+        assert abs(mean - expected) / expected < 0.10, "flat mean looks off (smoke check)"
+    elif cfg.arrivals == "uniform":
+        # Uniform integers 0..M => mean ~ M/2
+        expected = cfg.max_new_jobs_per_hour / 2.0
+        assert abs(mean - expected) / expected < 0.10, "uniform mean looks off (smoke check)"
+    elif cfg.arrivals == "poisson":
+        # Only a loose sanity check: mean should not be wildly off lambda unless capped heavily
+        target = min(cfg.poisson_lambda, cfg.max_new_jobs_per_hour)
+        if target > 0:
+            assert abs(mean - target) / target < 0.20, "poisson mean looks off (smoke check)"
+
+    print("[OK] distribution smoke")
+
+def main():
+    # Example configs to test
+    flat_cfg = WorkloadGenConfig(arrivals="flat", max_new_jobs_per_hour=1000)
+    pois_cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=200.0, max_new_jobs_per_hour=1000)
+    unif_cfg = WorkloadGenConfig(arrivals="uniform", max_new_jobs_per_hour=1000)
+
+    for cfg in (flat_cfg, pois_cfg, unif_cfg):
+        check_determinism(cfg)
+        check_bounds(cfg)
+        check_distribution_smoke(cfg)
+
+if __name__ == "__main__":
+    main()

test_sanity_workloadgen.py

@@ -0,0 +1,43 @@
+# sanity test for workload generator
+import numpy as np
+from workloadgen import WorkloadGenerator, WorkloadGenConfig
+
+def assert_job_valid(j, cfg):
+    assert cfg.min_duration <= j.duration <= cfg.max_duration
+    assert cfg.min_nodes <= j.nodes <= cfg.max_nodes
+    assert cfg.min_cores <= j.cores_per_node <= cfg.max_cores
+
+def test_determinism():
+    cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=100.0, max_new_jobs_per_hour=1500)
+    gen = WorkloadGenerator(cfg)
+
+    rng1 = np.random.default_rng(123)
+    rng2 = np.random.default_rng(123)
+
+    for hour in range(50):
+        a = gen.sample(hour, rng1)
+        b = gen.sample(hour, rng2)
+        assert [(j.duration, j.nodes, j.cores_per_node) for j in a] == [(j.duration, j.nodes, j.cores_per_node) for j in b]
+
+def test_constraints():
+    cfg = WorkloadGenConfig(arrivals="flat", flat_jobs_per_hour=200, max_new_jobs_per_hour=200)
+    gen = WorkloadGenerator(cfg)
+    rng = np.random.default_rng(7)
+    for hour in range(200):
+        jobs = gen.sample(hour, rng)
+        for j in jobs:
+            assert_job_valid(j, cfg)
+
+def test_poisson_mean_sanity():
+    cfg = WorkloadGenConfig(arrivals="poisson", poisson_lambda=50.0, max_new_jobs_per_hour=1500)
+    gen = WorkloadGenerator(cfg)
+    rng = np.random.default_rng(1)
+    counts = [len(gen.sample(h, rng)) for h in range(2000)]
+    mean = float(np.mean(counts))
+    assert 48.0 < mean < 52.0, mean  # tighter band, still reliable for 2000 samples
+
+if __name__ == "__main__":
+    test_determinism()
+    test_constraints()
+    test_poisson_mean_sanity()
+    print("[OK] workloadgen sanity checks passed")