"""Generate fictional PEO-FS practice datasets (seeded, reproducible).""" import random, zipfile, io from datetime import date, timedelta from pathlib import Path import numpy as np, pandas as pd from openpyxl import Workbook from openpyxl.styles import Font, Alignment SEED = 20261014 rng = np.random.default_rng(SEED); random.seed(SEED) OUT = Path(__file__).resolve().parent OUT.mkdir(parents=True, exist_ok=True) ASOF = date(2026, 9, 30) def month_end(y, m): return (date(y + (m == 12), m % 12 + 1, 1) - timedelta(days=1)) PERIODS = [] # Oct 2023 .. Sep 2026 y, m = 2023, 10 for _ in range(36): PERIODS.append(month_end(y, m)); m += 1 if m == 13: y, m = y + 1, 1 progs = [ ("PRG-001","KESTREL Sensor Upgrade","C4ISR","II","EMD"), ("PRG-002","IRONWOOD Tactical Network","C4ISR","I","EMD"), ("PRG-003","HARBOR Logistics Modernization","Sustainment","III","P&D"), ("PRG-004","SENTINEL Training Simulator","Training Systems","III","EMD"), ("PRG-005","BLUE HERON Unmanned Surface Vessel","C4ISR","II","EMD"), ("PRG-006","ANVIL Depot Tooling Refresh","Sustainment","III","P&D"), ("PRG-007","ORION Mission Planning Software","C4ISR","III","TMRR"), ("PRG-008","GRANITE Shelter Systems","Sustainment","III","P&D"), ("PRG-009","FALCONER Aircrew Trainer","Training Systems","II","EMD"), ("PRG-010","TIDEWATER Port Security Radar","C4ISR","II","TMRR"), ("PRG-011","SUMMIT Spare Parts Analytics","Sustainment","III","O&S"), ("PRG-012","LANTERN Night Vision Retrofit","Training Systems","III","P&D"), ] pms = ["J. Rivera","A. Patel","S. Okafor","L. Nguyen","R. Thompson","K. Alvarez","D. Kim","M. Brooks","T. Haddad","E. Walsh","C. Johnson","P. Lindqvist"] programs = [] for i,(pid,name,port,acat,ph) in enumerate(progs): bs = date(2023, 10, 1) if i % 3 else date(2023, 4, 1) if i % 2 else date(2022, 10, 1) be = date(2028 + (i % 3), 9, 30) programs.append(dict(ProgramID=pid, ProgramName=name, Portfolio=port, ACAT=acat, Phase=ph, ProgramManager=pms[i], BaselineStart=bs, BaselineEnd=be, ActiveFlag=True)) programs = pd.DataFrame(programs) # Program performance personalities (target cumulative CPI/SPI at as-of date) perf = {p: (float(np.clip(rng.normal(0.96, 0.05), 0.84, 1.07)), float(np.clip(rng.normal(0.96, 0.05), 0.85, 1.06))) for p in programs.ProgramID} perf["PRG-002"] = (0.89, 0.92) # IRONWOOD troubled (matches curriculum narrative) perf["PRG-005"] = (0.93, 0.95) # ---------- contracts ---------- vendors = ["Contoso Defense Systems","Fabrikam Federal LLC","Northwind Integrated Solutions","Adventure Works Engineering", "Tailspin Aerospace","Wide World Logistics","Litware Technologies","Proseware Inc.","Woodgrove Analytics","Coho Marine Systems"] cos = ["M. Chen","B. Ortiz","H. Yamada","G. Fischer","N. Diallo"] ctypes = ["CPIF","CPFF","FFP","CPAF","T&M"] contracts = []; seq = 100 for i, p in enumerate(programs.itertuples()): n = 3 if i < 6 else 2 for k in range(n): seq += 1 + int(rng.integers(0, 9)) yy = 24 if k == 0 else int(rng.choice([24, 25])) award = date(2000 + yy - 1, 10, 1) + timedelta(days=int(rng.integers(30, 330))) if k else date(2023, 9, 1) + timedelta(days=int(rng.integers(0, 25))) pops = award + timedelta(days=int(rng.integers(10, 40))) pope = pops + timedelta(days=int(rng.integers(2, 5)) * 365 - 1) contracts.append(dict(ContractNumber=f"XX000{(i%3)+1}-{yy}-{'C' if k==0 else rng.choice(['C','D','F'])}-{seq:04d}", ProgramID=p.ProgramID, Vendor=vendors[(i*3 + k) % len(vendors)], ContractType=("CPIF" if k == 0 and i % 2 == 0 else "CPFF") if k == 0 else str(rng.choice(ctypes)), AwardDate=award, PoPStart=pops, PoPEnd=pope, CeilingValue=0, ObligatedToDate=0, ContractingOfficer=cos[(i + k) % len(cos)], CLINCount=int(rng.integers(2, 12)), Primary=(k == 0))) contracts = pd.DataFrame(contracts) # ---------- EVM ---------- ca_names = ["SYSENG","HW","SW","INTEG","TEST","PM","HULL","TRAIN","LOG"] evm_rows = []; ca_bac_total = {} for i, p in enumerate(programs.itertuples()): cpi_t, spi_t = perf[p.ProgramID] con = contracts[(contracts.ProgramID == p.ProgramID) & contracts.Primary].iloc[0] nca = int(rng.integers(4, 7)) names = list(rng.choice(ca_names, nca, replace=False)) if p.ProgramID == "PRG-005" and "HULL" not in names: names[1] = "HULL" for j, nm in enumerate(names): wbs = f"1.{j+1}.{int(rng.integers(1,4))}" bac = float(round(rng.uniform(4, 25)) * 500_000) dur = int(rng.integers(48, 66)) # planned months for CA (beyond window) start = int(rng.integers(0, 6)) x = np.arange(dur) + 0.5 w = np.exp(-0.5 * ((x - dur * 0.5) / (dur * 0.28)) ** 2); w = w / w.sum() # bell-shaped spend planned = bac * w ca_cpi = cpi_t * rng.uniform(0.96, 1.04); ca_spi = spi_t * rng.uniform(0.97, 1.03) cum_ev = cum_ac = 0.0 for t, pe in enumerate(PERIODS): k = t - start bcws = planned[k] if 0 <= k < dur else 0.0 if k < 0: continue drift = 1 + (ca_spi - 1) * min(1, (k + 1) / 18) bcwp = max(0.0, bcws * drift * rng.normal(1, 0.06)) bcwp = min(bcwp, bac - cum_ev) cdrift = 1 + (ca_cpi - 1) * min(1, (k + 1) / 15) acwp = bcwp / cdrift * rng.normal(1, 0.05) if t == 20 and j == 0: acwp -= 0.15 * acwp # a negative-ish correction month (edge case) cum_ev += bcwp; cum_ac += acwp cpi_c = cum_ev / cum_ac if cum_ac else 1 ieac = bac / cpi_c eac_c = cum_ac + (bac - cum_ev) / min(1.0, cpi_c + 0.04) # contractor is optimistic eac_c = min(eac_c, ieac) evm_rows.append(dict(ProgramID=p.ProgramID, ContractNumber=con.ContractNumber, WBS=wbs, ControlAccount=f"CA-{wbs}-{nm}", PeriodEnd=pe, BCWS=round(bcws, 2), BCWP=round(bcwp, 2), ACWP=round(acwp, 2), BAC=bac, EAC_Contractor=round(eac_c, -2))) ca_bac_total[p.ProgramID] = ca_bac_total.get(p.ProgramID, 0) + bac evm = pd.DataFrame(evm_rows) # ---------- budget execution ---------- apps = ["RDT&E","Procurement","O&M"] be_rows = [] for p in programs.itertuples(): base = ca_bac_total[p.ProgramID] / 36 mix = {"RDT&E": 0.55, "Procurement": 0.3, "O&M": 0.15} if p.Phase in ("EMD","TMRR") else {"RDT&E": 0.15, "Procurement": 0.5, "O&M": 0.35} if p.Phase == "O&S": mix = {"RDT&E": 0.05, "Procurement": 0.2, "O&M": 0.75} lag = rng.uniform(0.75, 1.0) for a in apps: for t, pe in enumerate(PERIODS): cm = date(pe.year, pe.month, 1) fy = cm.year + (cm.month >= 10); fym = (cm.month - 10) % 12 + 1 seasonal = 0.6 if fym in (1, 2, 3) else 1.4 if fym in (11, 12) and False else (1.5 if fym == 12 else 1.0) po = round(base * 1.2 * mix[a] * seasonal * rng.uniform(0.8, 1.2), -3) ao = round(po * lag * rng.uniform(0.6, 1.25), -3) if pe <= ASOF else 0 pex = round(po * 0.75, -3); aex = round(ao * rng.uniform(0.55, 0.85), -3) be_rows.append(dict(ProgramID=p.ProgramID, FiscalYear=fy, FYMonth=fym, CalendarMonth=cm, Appropriation=a, PlannedObligation=po, ActualObligation=ao, PlannedExpenditure=pex, ActualExpenditure=aex)) budget = pd.DataFrame(be_rows) # contract ceilings / obligations consistent with EVM + budget for idx, c in contracts.iterrows(): tot_obl = budget[budget.ProgramID == c.ProgramID].ActualObligation.sum() if c.Primary: ceil = round(ca_bac_total[c.ProgramID] * rng.uniform(1.12, 1.3), -5) cum_ac = evm[evm.ProgramID == c.ProgramID].ACWP.sum() obl = round(min(ceil * 0.95, max(cum_ac * rng.uniform(1.05, 1.2), tot_obl * 0.7)), -4) else: ceil = round(rng.uniform(2, 15) * 1e6, -5) obl = round(ceil * rng.uniform(0.2, 0.85), -4) contracts.at[idx, "CeilingValue"] = ceil; contracts.at[idx, "ObligatedToDate"] = obl contracts = contracts.drop(columns="Primary") # ---------- milestones ---------- ms_catalog = [("Milestone A","Decision Point"),("System Requirements Review","Technical Review"),("Preliminary Design Review","Technical Review"), ("Milestone B","Decision Point"),("Critical Design Review","Technical Review"),("Test Readiness Review","Technical Review"), ("Developmental Test Start","Test Event"),("First Article Delivery","Delivery"),("Operational Test Start","Test Event"), ("Milestone C","Decision Point"),("Initial Operational Capability","Delivery"),("Full-Rate Production Decision","Decision Point"), ("Full Operational Capability","Delivery")] owners = ["IPT Lead – Systems","IPT Lead – Software","IPT Lead – Test","IPT Lead – Logistics","Chief Engineer","Business Financial Mgr"] comments = ["","","","Awaiting vendor drawings","Test range availability","Software build slipped","Pending ECP approval","On track per IMS","Supplier delay on long-lead parts","Rebaselined per APB"] ms_rows = [] for p in programs.itertuples(): _, spi_t = perf[p.ProgramID] n = 12 + (p.ProgramID in ("PRG-002","PRG-005","PRG-009")) span = (p.BaselineEnd - p.BaselineStart).days for k in range(n): name, typ = ms_catalog[k % len(ms_catalog)] bd = p.BaselineStart + timedelta(days=int(span * (k + 0.7) / (n + 0.5))) slip = int(max(-10, rng.normal((1 - spi_t) * 400 * (k / n + 0.3), 20))) fd = bd + timedelta(days=slip) ad = None if fd <= ASOF - timedelta(days=15) and rng.random() < 0.85: ad = fd + timedelta(days=int(rng.integers(-5, 10))) ms_rows.append(dict(MilestoneID=f"MS-{p.ProgramID[-3:]}-{k+1:02d}", ProgramID=p.ProgramID, MilestoneName=name, MilestoneType=typ, BaselineDate=bd, ForecastDate=(ad or fd), ActualDate=ad, Owner=str(rng.choice(owners)), Comments=str(rng.choice(comments)))) milestones = pd.DataFrame(ms_rows) # ---------- schedule tasks ---------- verbs = ["Design","Integrate","Test","Procure","Review","Deliver","Update","Verify","Install","Document"] objs = ["radio firmware v2","sensor mount","hull module","training scenario","spares kit","software build","antenna array","power subsystem","shelter frame","data link","maintenance manual","test procedures"] task_rows = [] for i, p in enumerate(programs.itertuples()): _, spi_t = perf[p.ProgramID] base_id = 10000 + i * 1000 ids = [] for k in range(125): tid = base_id + k * 3 + 1 bstart = date(2024, 1, 8) + timedelta(days=int(k * 7 + rng.integers(0, 10))) dur = int(rng.integers(5, 60)) bfin = bstart + timedelta(days=dur) lag = int(max(0, rng.normal((1 - spi_t) * 150, 10))) fs = bstart + timedelta(days=lag); ff = fs + timedelta(days=dur + int(rng.integers(-3, 8))) if ff <= ASOF: pc, af = 1.0, ff elif fs >= ASOF: pc, af = 0.0, None else: pc, af = round((ASOF - fs).days / max(1, (ff - fs).days), 2), None tf = int(rng.normal(10 - lag / 3, 12)) preds = [] if ids: for pid_ in rng.choice(ids[-6:], size=min(len(ids[-6:]), int(rng.integers(1, 3))), replace=False): preds.append(f"{pid_}" + (str(rng.choice(["", "FS+5d", "SS", "FS+10d"])) if rng.random() < 0.4 else "")) task_rows.append(dict(TaskID=tid, ProgramID=p.ProgramID, WBS=f"1.{k%6+1}.{k%4+1}.{k%3+1}", TaskName=f"{verbs[k % len(verbs)]} {objs[(k*7 + i) % len(objs)]}", BaselineStart=bstart, BaselineFinish=bfin, ForecastStart=fs, ForecastFinish=ff, ActualFinish=af, PercentComplete=pc, TotalFloat_Days=tf, Predecessors=",".join(preds), CriticalFlag="Y" if tf <= 0 else "N")) ids.append(tid) tasks = pd.DataFrame(task_rows) # ---------- risks ---------- risk_titles = {"Cost":["Material cost escalation","Labor rate increase","Rework on failed qualification","Inflation exceeds budget assumptions"], "Schedule":["Key software developer attrition","Long-lead parts delay","Test range availability","Late GFE delivery"], "Performance":["Sensor range below threshold","Cyber accreditation findings","Reliability below KPP","Interoperability defects"]} ipts = ["Software IPT","Systems IPT","Test IPT","Logistics IPT","Business IPT","Hardware IPT"] risk_rows = [] counts = [7,8,6,6,8,6,7,6,7,7,6,6] for i, p in enumerate(programs.itertuples()): for k in range(counts[i]): cat = str(rng.choice(list(risk_titles))) st = str(rng.choice(["Open","Mitigating","Closed","Realized"], p=[0.45,0.3,0.18,0.07])) risk_rows.append(dict(RiskID=f"R-{p.ProgramID[-3:]}-{k+1:02d}", ProgramID=p.ProgramID, Title=str(rng.choice(risk_titles[cat])), Category=cat, Likelihood=int(rng.integers(1, 6)), Consequence=int(rng.integers(1, 6)), Status=st, Owner=str(rng.choice(ipts)), MitigationDueDate=ASOF + timedelta(days=int(rng.integers(-60, 240))))) risks = pd.DataFrame(risk_rows) # ---------- write core datasets ---------- sets = {"programs": programs, "milestones": milestones, "evm_monthly": evm, "contracts": contracts, "budget_execution": budget, "schedule_tasks": tasks, "risks": risks} with pd.ExcelWriter(OUT / "PEO-FS_all_datasets.xlsx", engine="openpyxl") as allw: for name, df in sets.items(): df.to_csv(OUT / f"{name}.csv", index=False) df.to_excel(OUT / f"{name}.xlsx", index=False, sheet_name=name) df.to_excel(allw, index=False, sheet_name=name) # ---------- D8 messy status drops ---------- SD = OUT / "status_drops"; SD.mkdir(exist_ok=True) for f in SD.glob("*.xlsx"): f.unlink() fc_hdr = ["Forecast Date","Fcst Dt","Est. Completion","Forecast Date ","Fcst Date"] rag_hdr = ["RAG","Status (RAG)","Rag ","R/A/G"] cost_hdr = ["Cost","Est Cost ($)","Cost Estimate","EAC ($)"] rag_vars = {"Green":["G","Green","grn","GREEN","green "],"Amber":["A","Amber","amb","Yellow","Y "],"Red":["R","Red","RED","red "]} months = ["Oct","Nov","Dec","Jan","Feb","Mar","Apr","May","Jun","Jul","Aug","Sep"] def money_text(v): s = rng.integers(0, 4) if s == 0: return f"${v/1e6:.1f}M" if s == 1: return f"{v/1e3:.0f}K" if s == 2: return f"{v:,.0f}" return f" {v:,.0f} " def date_text(d): r = rng.random() if r < 0.08: return "TBD" if r < 0.14: fy = d.year + (d.month >= 10); q = ((d.month - 10) % 12) // 3 + 1 return f"Q{q} FY{fy%100}" if r < 0.45: return d.strftime("%m/%d/%y") return d # true date cpi_map = {p: perf[p][0] for p in perf} for i, p in enumerate(programs.itertuples()): pm = milestones[milestones.ProgramID == p.ProgramID] fy26 = budget[(budget.ProgramID == p.ProgramID) & (budget.FiscalYear == 2026)] for mi, ym in enumerate(["2026-07","2026-08","2026-09"]): style = (i + mi) % 5 wb = Workbook(); ws = wb.active; ws.title = "Status" if style != 3 else "Sheet1" ws["A1"] = f"{p.ProgramName.upper()} – MONTHLY STATUS INPUT ({ym})"; ws["A1"].font = Font(bold=True, size=14) ws.merge_cells("A1:H1"); ws["A1"].alignment = Alignment(horizontal="center") ws["A2"] = "FICTIONAL TRAINING DATA – PEO-FS"; ws.merge_cells("A2:H2") hdr = ["ProgramID" if style % 2 == 0 else "Program", "Milestone", fc_hdr[(i + mi) % len(fc_hdr)], rag_hdr[(i + 2 * mi) % len(rag_hdr)], cost_hdr[(i + mi) % len(cost_hdr)], "Notes"] + months ws.append([]) if style in (0, 2, 4) else None ws.append(hdr) report_end = month_end(int(ym[:4]), int(ym[5:])) open_ms = pm[pm.ActualDate.isna()].head(6) if len(open_ms) < 3: open_ms = pm.tail(4) for r in open_ms.itertuples(): slip = (r.ForecastDate - r.BaselineDate).days rag = "Green" if slip < 15 else "Amber" if slip < 60 else "Red" cost = float(round(ca_bac_total[p.ProgramID] / len(pm) * rng.uniform(0.6, 1.4), -3)) monthly = [] for fym in range(1, 13): row = fy26[(fy26.FYMonth == fym) & (fy26.Appropriation == "RDT&E")] cm = date(2025 + (fym <= 3), (fym + 8) % 12 + 1, 1) if cm > date(report_end.year, report_end.month, 1): monthly.append(None); continue v = float(row.ActualObligation.iloc[0]) / len(open_ms) if len(row) else 0 monthly.append(money_text(round(v, -3)) if rng.random() < 0.2 else round(v, -3)) pid_txt = p.ProgramID + (" " if rng.random() < 0.3 else "") ws.append([pid_txt, r.MilestoneName + (" " if rng.random() < 0.3 else ""), date_text(r.ForecastDate), str(rng.choice(rag_vars[rag])), money_text(cost), str(r.Comments)] + monthly) if rng.random() < 0.15: ws.append([]) # stray blank row for c in ws[4 if style in (0,2,4) else 3]: c.font = Font(bold=True) ws.column_dimensions["B"].width = 30; ws.column_dimensions["C"].width = 16 if style == 4: ws2 = wb.create_sheet("Instructions"); ws2["A1"] = "Fill in yellow cells. Do not change headers (people did anyway)." wb.save(SD / f"{p.ProgramID}_Status_{ym}.xlsx") with zipfile.ZipFile(OUT / "status_drops.zip", "w", zipfile.ZIP_DEFLATED) as z: for f in sorted(SD.glob("*.xlsx")): z.write(f, f"status_drops/{f.name}") # ---------- golden answer key: latest cumulative EVM per program ---------- g = evm.groupby(["ProgramID"]).agg(BCWS=("BCWS","sum"), BCWP=("BCWP","sum"), ACWP=("ACWP","sum")).reset_index() last = evm[evm.PeriodEnd == evm.PeriodEnd.max()].groupby("ProgramID").agg(BAC=("BAC","sum"), EAC_Contractor=("EAC_Contractor","sum")).reset_index() g = g.merge(last, on="ProgramID") g["CPI"] = g.BCWP / g.ACWP; g["SPI"] = g.BCWP / g.BCWS; g["CV"] = g.BCWP - g.ACWP; g["SV"] = g.BCWP - g.BCWS g["EAC_CPI"] = g.BAC / g.CPI; g["VAC"] = g.BAC - g.EAC_CPI; g["TCPI_BAC"] = (g.BAC - g.BCWP) / (g.BAC - g.ACWP) g["EAC_Gap"] = g.EAC_CPI - g.EAC_Contractor g = g.round({"CPI":3,"SPI":3,"TCPI_BAC":3,"BCWS":2,"BCWP":2,"ACWP":2,"CV":2,"SV":2,"EAC_CPI":2,"VAC":2,"EAC_Gap":2}) g.to_csv(OUT / "answer_key_evm_cumulative_2026-09.csv", index=False) # sanity checks assert len(programs) == 12 and evm.duplicated(["ControlAccount","ProgramID","PeriodEnd"]).sum() == 0 assert set(evm.ContractNumber) <= set(contracts.ContractNumber) for df in (milestones, evm, budget, tasks, risks, contracts): assert set(df.ProgramID) <= set(programs.ProgramID) assert (contracts.ObligatedToDate <= contracts.CeilingValue).all() assert (g.BCWP <= g.BAC).all() print({k: len(v) for k, v in sets.items()}, "status files:", len(list(SD.glob('*.xlsx')))) print(g[["ProgramID","CPI","SPI","EAC_CPI","EAC_Contractor"]].to_string())