Implementation Guide: Continuous Earned Value Management (EVM) Analysis — Detect Schedule/Cost Variances & Draft Corrective Action Plans

# Daily extraction of EVM metrics from Deltek Costpoint

# evm_data_pipeline.py
# Daily extraction of EVM metrics from Deltek Costpoint

import requests
import pandas as pd
import sqlalchemy
import os
from datetime import datetime

COSTPOINT_BASE = os.environ["COSTPOINT_URL"]
COSTPOINT_AUTH = (os.environ["COSTPOINT_USER"], os.environ["COSTPOINT_PASS"])
SQL_ENGINE = sqlalchemy.create_engine(os.environ["AZURE_SQL_CONNECTION"])

def extract_evm_data(contract_ids: list, period: str = None) -> pd.DataFrame:
    """Extract EVM performance data from Costpoint for all active contracts."""

    period = period or datetime.today().strftime("%Y-%m")
    all_records = []

    for contract_id in contract_ids:
        # Get control account level EVM data
        resp = requests.get(
            f"{COSTPOINT_BASE}/api/v1/contracts/{contract_id}/evm/control-accounts",
            auth=COSTPOINT_AUTH,
            params={"period": period},
            headers={"Accept": "application/json"}
        )

        if resp.status_code != 200:
            print(f"Warning: Could not retrieve EVM data for contract {contract_id}")
            continue

        for ca in resp.json().get("controlAccounts", []):
            bcws = ca.get("budgetedCostWorkScheduled", 0) or 0
            bcwp = ca.get("budgetedCostWorkPerformed", 0) or 0
            acwp = ca.get("actualCostWorkPerformed", 0) or 0
            bac  = ca.get("budgetAtCompletion", 0) or 0
            eac  = ca.get("estimateAtCompletion", 0) or 0
            etc  = ca.get("estimateToComplete", 0) or 0

            # Calculate derived EVM metrics
            cv  = bcwp - acwp                           # Cost Variance
            sv  = bcwp - bcws                           # Schedule Variance
            cpi = bcwp / acwp if acwp else None         # Cost Performance Index
            spi = bcwp / bcws if bcws else None         # Schedule Performance Index
            vac = bac - eac                             # Variance At Completion
            tcpi_bac = (bac - bcwp) / (bac - acwp) if (bac - acwp) else None  # TCPI to BAC
            tcpi_eac = (bac - bcwp) / (eac - acwp) if (eac - acwp) else None  # TCPI to EAC

            record = {
                "extract_date": datetime.utcnow().date(),
                "period": period,
                "contract_id": contract_id,
                "contract_number": ca.get("contractNumber", ""),
                "control_account": ca.get("controlAccountId", ""),
                "wbs_element": ca.get("wbsElement", ""),
                "responsible_manager": ca.get("responsibleManager", ""),
                "bcws": bcws,
                "bcwp": bcwp,
                "acwp": acwp,
                "bac": bac,
                "eac": eac,
                "etc": etc,
                "cv": cv,
                "sv": sv,
                "cpi": round(cpi, 4) if cpi else None,
                "spi": round(spi, 4) if spi else None,
                "vac": vac,
                "tcpi_bac": round(tcpi_bac, 4) if tcpi_bac else None,
                "tcpi_eac": round(tcpi_eac, 4) if tcpi_eac else None,
                "percent_complete": round(bcwp / bac * 100, 1) if bac else 0
            }
            all_records.append(record)

    df = pd.DataFrame(all_records)
    return df


def load_evm_to_sql(df: pd.DataFrame):
    """Load EVM data to Azure SQL for trend analysis."""
    if df.empty:
        return
    df.to_sql("evm_performance_history", SQL_ENGINE,
              if_exists="append", index=False, chunksize=500)
    print(f"Loaded {len(df)} EVM records to Azure SQL.")


def check_thresholds(df: pd.DataFrame, thresholds: dict) -> pd.DataFrame:
    """Flag control accounts exceeding variance thresholds."""
    df = df.copy()

    df["cpi_flag"] = df["cpi"].apply(
        lambda x: "CRITICAL" if x and x < thresholds.get("cpi_critical", 0.80)
        else ("WARNING" if x and x < thresholds.get("cpi_warning", 0.90) else "OK")
    )
    df["spi_flag"] = df["spi"].apply(
        lambda x: "CRITICAL" if x and x < thresholds.get("spi_critical", 0.80)
        else ("WARNING" if x and x < thresholds.get("spi_warning", 0.90) else "OK")
    )
    df["vac_flag"] = df.apply(
        lambda row: "CRITICAL" if row["bac"] and abs(row["vac"] / row["bac"]) > thresholds.get("vac_pct_critical", 0.10)
        else ("WARNING" if row["bac"] and abs(row["vac"] / row["bac"]) > thresholds.get("vac_pct_warning", 0.05) else "OK"),
        axis=1
    )

    df["overall_flag"] = df[["cpi_flag", "spi_flag", "vac_flag"]].apply(
        lambda row: "CRITICAL" if "CRITICAL" in row.values
        else ("WARNING" if "WARNING" in row.values else "OK"),
        axis=1
    )

    return df

# evm_analysis_generator.py

from openai import AzureOpenAI
import os, json, pandas as pd

client = AzureOpenAI(
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    api_key=os.environ["AZURE_OPENAI_KEY"],
    api_version="2024-08-01-preview"
)

VARIANCE_ANALYSIS_PROMPT = """You are a defense program management EVM specialist.
Generate a variance analysis narrative for the following EVM performance data.
This narrative will be used in the program's Cost Performance Report (CPR)
and reviewed by DCMA surveillance personnel.

CONTRACT: {contract_number}
CONTROL ACCOUNT: {control_account}
PERIOD: {period}
RESPONSIBLE MANAGER: {responsible_manager}

EVM PERFORMANCE DATA:
BAC: ${bac:,.0f}
BCWS: ${bcws:,.0f}  (Planned Value)
BCWP: ${bcwp:,.0f}  (Earned Value)
ACWP: ${acwp:,.0f}  (Actual Cost)
EAC:  ${eac:,.0f}   (Estimate at Completion)
ETC:  ${etc:,.0f}   (Estimate to Complete)

CALCULATED METRICS:
CPI: {cpi:.3f} ({cpi_status})
SPI: {spi:.3f} ({spi_status})
CV:  ${cv:,.0f} ({cv_dir})
SV:  ${sv:,.0f} ({sv_dir})
VAC: ${vac:,.0f} ({vac_dir})
% Complete: {pct_complete:.1f}%

PROGRAM CONTEXT (if provided): {program_context}

Generate the following CPR narrative sections:

## COST VARIANCE ANALYSIS
[Explain the CV in plain language: what caused it, whether it is isolated or systemic,
 and trend direction. Reference specific work activities where possible.
 If context not provided, indicate [PM INPUT NEEDED: specific cause of cost overrun/underrun]]

## SCHEDULE VARIANCE ANALYSIS
[Explain the SV: what schedule events drove performance, which milestones are affected,
 and impact on downstream work.
 If context not provided, indicate [PM INPUT NEEDED: specific cause of schedule variance]]

## EAC ASSESSMENT
[Assess whether the EAC is realistic: does TCPI-to-EAC = {tcpi_eac:.3f} represent
 a achievable performance improvement? If TCPI > 1.10, note that the EAC may be
 optimistic and an EAC revision may be warranted.]

## OUTLOOK AND RISK
[Assess forward-looking risk based on current performance trends.
 Flag if CPI or SPI trends indicate further deterioration is likely.]

[DRAFT — REQUIRES PROGRAM MANAGER REVIEW AND APPROVAL BEFORE CPR SUBMISSION]
[All [PM INPUT NEEDED] items must be filled before submission to DCMA]"""

CAP_PROMPT = """You are a defense program management specialist drafting a Corrective Action Plan
for a control account with adverse EVM performance. This CAP will be reviewed by DCMA.

CONTROL ACCOUNT: {control_account}
CONTRACT: {contract_number}
PM: {responsible_manager}
PERFORMANCE ISSUE: CPI={cpi:.3f}, SPI={spi:.3f}
VAC: ${vac:,.0f}

Generate a structured Corrective Action Plan (CAP):

## CORRECTIVE ACTION PLAN
**Control Account:** {control_account}
**Contract:** {contract_number}
**CAP Prepared By:** [PM NAME — to be filled]
**Date:** [TODAY]
**CAP Number:** CAP-[SEQUENCE]

### 1. ROOT CAUSE OF VARIANCE
[Primary cause (technical / scope / estimation / resource): [PM INPUT REQUIRED]
 Secondary contributing factors: [PM INPUT REQUIRED]]

### 2. CORRECTIVE ACTIONS
| Action | Description | Owner | Due Date | Expected Impact |
|--------|-------------|-------|----------|-----------------|
| CA-01 | [Specific action — PM input] | [Owner] | [Date] | [CPI/SPI impact] |
| CA-02 | [Specific action — PM input] | [Owner] | [Date] | [CPI/SPI impact] |
| CA-03 | [Specific action — PM input] | [Owner] | [Date] | [CPI/SPI impact] |

### 3. REVISED EAC (IF APPLICABLE)
Current EAC: ${eac:,.0f}
Revised EAC: [PM INPUT — if EAC revision is warranted]
EAC Rationale: [PM INPUT]

### 4. PERFORMANCE RECOVERY PLAN
Projected CPI recovery timeline: [PM INPUT]
Projected SPI recovery timeline: [PM INPUT]
Metrics to track recovery: [CPI/SPI monthly, milestone completion rate]

### 5. LESSONS LEARNED
[What process improvement will prevent recurrence: PM INPUT]

[DRAFT — AI GENERATED — PROGRAM MANAGER MUST COMPLETE ALL [PM INPUT] SECTIONS]
[DCMA CAP SUBMISSION REQUIRES PM AND CONTRACT MANAGER SIGN-OFF]"""

def generate_variance_analysis(ca_data: dict, program_context: str = "") -> str:
    response = client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a DCMA-experienced EVM analyst. Generate technically accurate variance analysis narratives. Flag any assumptions clearly."},
            {"role": "user", "content": VARIANCE_ANALYSIS_PROMPT.format(
                **ca_data,
                cv_dir="unfavorable" if ca_data.get("cv", 0) < 0 else "favorable",
                sv_dir="unfavorable" if ca_data.get("sv", 0) < 0 else "favorable",
                vac_dir="unfavorable" if ca_data.get("vac", 0) < 0 else "favorable",
                cpi_status="below threshold" if ca_data.get("cpi", 1) < 0.90 else "acceptable",
                spi_status="below threshold" if ca_data.get("spi", 1) < 0.90 else "acceptable",
                program_context=program_context or "[No context provided — PM should review all narrative sections]"
            )}
        ],
        temperature=0.1,
        max_tokens=2000
    )
    return response.choices[0].message.content


def generate_cap(ca_data: dict) -> str:
    response = client.chat.completions.create(
        model=os.environ["AZURE_OPENAI_DEPLOYMENT"],
        messages=[
            {"role": "system", "content": "You are a defense program management specialist. Generate structured CAP frameworks that guide program managers to document specific, accountable corrective actions."},
            {"role": "user", "content": CAP_PROMPT.format(**ca_data)}
        ],
        temperature=0.1,
        max_tokens=2000
    )
    return response.choices[0].message.content

## Power BI GCC Dashboard: EVM Program Health Monitor

DATA SOURCE: Azure SQL (Government) — evm_performance_history table

DASHBOARD PAGES:

PAGE 1: PROGRAM HEALTH OVERVIEW
- KPI Cards (traffic light — green/amber/red):
  * CPI (cumulative): [value] — Green(≥0.95) / Amber(0.85-0.95) / Red(<0.85)
  * SPI (cumulative): [value]
  * VAC ($): [value] — Green(favorable) / Red(unfavorable > 5% of BAC)
  * % Complete: [value]
  * EAC vs BAC: [$ variance]

- Waterfall chart: BAC → BCWP → ACWP → EAC (shows cost variance visually)
- Line chart: CPI and SPI trend (last 12 months) with threshold reference lines at 0.85 and 1.0
- Gauge: % Complete vs Time Elapsed (should track together on healthy programs)

PAGE 2: CONTROL ACCOUNT DRILL-DOWN
- Table: All control accounts with CPI, SPI, CV, SV, % Complete, Flag status
  Color-coded rows: Green / Amber / Red per overall_flag
  Sortable by: CPI ascending (worst performers first)
- Bar chart: CV by control account (horizontal, sorted worst to best)
- Filter: By WBS level, by responsible manager, by flag status

PAGE 3: THRESHOLD ALERTS
- Table: All control accounts in WARNING or CRITICAL status
  Columns: CA, Manager, CPI, SPI, VAC, Flag, Days Since First Flag, CAP Status
- KPI: Count of CAs in CRITICAL status
- DCMA Surveillance Risk: CAs below 0.85 that will appear in next CPR

PAGE 4: EAC PROJECTION AND TCPI
- Scatter: EAC vs BAC by control account (diagonal line = EAC=BAC)
  Points above the line = over-budget projections
- TCPI analysis: CAs with TCPI > 1.10 (effectively impossible recovery pace) highlighted
- EAC confidence: Range showing optimistic/realistic/pessimistic EAC scenarios

REFRESH: Daily at 07:00 ET (after overnight Costpoint data extraction)
ALERT INTEGRATION: Power BI data-driven alerts push to Teams when CPI/SPI cross thresholds

SYSTEM PROMPT:
You are a defense program controls specialist assembling the CPR Format 5
(Problem Analysis Report) for DCMA submission.

CONTRACT: {contract_number}
REPORTING PERIOD: {period}
PROGRAM MANAGER: {pm_name}

Compile the following individual control account variance analyses into a
cohesive CPR Format 5 narrative that:
1. Provides a program-level summary of performance and EAC confidence
2. Presents significant variances by WBS element (only those meeting threshold criteria)
3. Describes corrective actions in place or planned
4. Assesses the risk to the program's EAC
5. Meets DCMA's CPR Format 5 content requirements per ANSI/EIA-748

INDIVIDUAL VARIANCE ANALYSES:
{variance_analyses}

PROGRAM CONTEXT:
{program_context}

[DRAFT — REQUIRES PM AND PROGRAM CONTROLS MANAGER REVIEW AND SIGN-OFF BEFORE DCMA SUBMISSION]