CWICR Schedule Integrator
Business Case
Problem Statement
Project planning requires:
- Linking costs to schedule activities
- Generating cost-loaded schedules
- Projecting cash flow requirements
- Tracking earned value
Solution
Integrate CWICR cost data with project schedules to create cost-loaded Gantt charts, cash flow curves, and earned value tracking.
Business Value
- Cost visibility - See when costs occur
- Cash flow - Project funding requirements
- Earned value - Track cost/schedule performance
- Integration - Connect cost and schedule data
Technical Implementation
import pandas as pd
import numpy as np
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime, timedelta, date
from enum import Enum
from collections import defaultdict
class CostDistribution(Enum):
"""Methods for distributing costs over time."""
UNIFORM = "uniform" # Even distribution
FRONT_LOADED = "front_loaded" # More at start
BACK_LOADED = "back_loaded" # More at end
S_CURVE = "s_curve" # S-curve distribution
@dataclass
class ScheduleActivity:
"""Project schedule activity."""
activity_id: str
description: str
start_date: date
end_date: date
duration_days: int
work_items: List[str]
budgeted_cost: float
predecessors: List[str] = field(default_factory=list)
@dataclass
class CostLoadedActivity:
"""Activity with daily cost distribution."""
activity: ScheduleActivity
daily_costs: Dict[date, float]
cumulative_costs: Dict[date, float]
@dataclass
class CashFlowProjection:
"""Cash flow projection."""
project_name: str
start_date: date
end_date: date
total_cost: float
daily_costs: Dict[date, float]
weekly_costs: Dict[str, float]
monthly_costs: Dict[str, float]
cumulative: Dict[date, float]
@dataclass
class EarnedValueMetrics:
"""Earned value metrics at point in time."""
data_date: date
planned_value: float # PV / BCWS
earned_value: float # EV / BCWP
actual_cost: float # AC / ACWP
schedule_variance: float # SV = EV - PV
cost_variance: float # CV = EV - AC
spi: float # Schedule Performance Index
cpi: float # Cost Performance Index
eac: float # Estimate at Completion
etc: float # Estimate to Complete
class CWICRScheduleIntegrator:
"""Integrate CWICR costs with project schedules."""
def __init__(self, cwicr_data: pd.DataFrame):
self.cost_data = cwicr_data
self._index_data()
def _index_data(self):
"""Index cost data."""
if 'work_item_code' in self.cost_data.columns:
self._code_index = self.cost_data.set_index('work_item_code')
else:
self._code_index = None
def get_work_item_cost(self, code: str, quantity: float) -> float:
"""Get total cost for work item."""
if self._code_index is None or code not in self._code_index.index:
return 0
item = self._code_index.loc[code]
labor = float(item.get('labor_cost', 0) or 0)
material = float(item.get('material_cost', 0) or 0)
equipment = float(item.get('equipment_cost', 0) or 0)
return (labor + material + equipment) * quantity
def create_schedule_activity(self,
activity_id: str,
description: str,
start_date: date,
duration_days: int,
work_items: List[Dict[str, Any]],
predecessors: List[str] = None) -> ScheduleActivity:
"""Create schedule activity with linked work items."""
# Calculate budgeted cost
total_cost = 0
codes = []
for item in work_items:
code = item.get('work_item_code', item.get('code'))
qty = item.get('quantity', 0)
total_cost += self.get_work_item_cost(code, qty)
codes.append(code)
end_date = start_date + timedelta(days=duration_days)
return ScheduleActivity(
activity_id=activity_id,
description=description,
start_date=start_date,
end_date=end_date,
duration_days=duration_days,
work_items=codes,
budgeted_cost=round(total_cost, 2),
predecessors=predecessors or []
)
def distribute_cost(self,
activity: ScheduleActivity,
method: CostDistribution = CostDistribution.UNIFORM) -> CostLoadedActivity:
"""Distribute activity cost over duration."""
daily_costs = {}
days = activity.duration_days
if days == 0:
daily_costs[activity.start_date] = activity.budgeted_cost
else:
if method == CostDistribution.UNIFORM:
daily_amount = activity.budgeted_cost / days
for i in range(days):
day = activity.start_date + timedelta(days=i)
daily_costs[day] = daily_amount
elif method == CostDistribution.FRONT_LOADED:
# Higher at start, decreasing
total_weight = sum(range(days, 0, -1))
for i in range(days):
day = activity.start_date + timedelta(days=i)
weight = (days - i) / total_weight
daily_costs[day] = activity.budgeted_cost * weight
elif method == CostDistribution.BACK_LOADED:
# Lower at start, increasing
total_weight = sum(range(1, days + 1))
for i in range(days):
day = activity.start_date + timedelta(days=i)
weight = (i + 1) / total_weight
daily_costs[day] = activity.budgeted_cost * weight
elif method == CostDistribution.S_CURVE:
# S-curve distribution (sigmoid)
for i in range(days):
day = activity.start_date + timedelta(days=i)
# Sigmoid function normalized
x = (i / days - 0.5) * 10
sigmoid = 1 / (1 + np.exp(-x))
daily_costs[day] = activity.budgeted_cost * sigmoid / days * 2
# Calculate cumulative
cumulative = {}
running_total = 0
for day in sorted(daily_costs.keys()):
running_total += daily_costs[day]
cumulative[day] = running_total
return CostLoadedActivity(
activity=activity,
daily_costs=daily_costs,
cumulative_costs=cumulative
)
def generate_cash_flow(self,
activities: List[ScheduleActivity],
project_name: str = "Project",
distribution: CostDistribution = CostDistribution.S_CURVE) -> CashFlowProjection:
"""Generate project cash flow projection."""
# Distribute costs for all activities
loaded_activities = [
self.distribute_cost(a, distribution)
for a in activities
]
# Aggregate daily costs
daily_costs = defaultdict(float)
for loaded in loaded_activities:
for day, cost in loaded.daily_costs.items():
daily_costs[day] += cost
# Sort and calculate cumulative
sorted_days = sorted(daily_costs.keys())
cumulative = {}
running_total = 0
for day in sorted_days:
running_total += daily_costs[day]
cumulative[day] = running_total
# Aggregate to weekly
weekly_costs = defaultdict(float)
for day, cost in daily_costs.items():
week_key = day.strftime('%Y-W%W')
weekly_costs[week_k