Schedule-Cost Linker
Business Case
Problem Statement
Integrating schedule and cost requires:
-
Linking activities to budget items
-
Creating cost-loaded schedules
-
Generating cash flow forecasts
-
Tracking earned value metrics
Solution
Systematic linkage between schedule activities and cost data to enable integrated project control.
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 date, timedelta from enum import Enum from collections import defaultdict
class LoadingMethod(Enum): UNIFORM = "uniform" # Even distribution FRONT_LOADED = "front_loaded" BACK_LOADED = "back_loaded" BELL_CURVE = "bell_curve"
@dataclass class ScheduleActivity: activity_id: str name: str start_date: date finish_date: date duration: int percent_complete: float = 0
@dataclass class CostItem: cost_code: str description: str budgeted_cost: float labor_cost: float material_cost: float equipment_cost: float
@dataclass class ActivityCostLink: activity_id: str cost_code: str budgeted_cost: float loading_method: LoadingMethod
@dataclass class EarnedValueMetrics: data_date: date bcws: float # Budgeted Cost of Work Scheduled (PV) bcwp: float # Budgeted Cost of Work Performed (EV) acwp: float # Actual Cost of Work Performed (AC) sv: float # Schedule Variance cv: float # Cost Variance spi: float # Schedule Performance Index cpi: float # Cost Performance Index eac: float # Estimate at Completion etc: float # Estimate to Complete vac: float # Variance at Completion
class ScheduleCostLinker: """Link schedule activities to cost items."""
def __init__(self, project_name: str, budget_at_completion: float):
self.project_name = project_name
self.bac = budget_at_completion
self.activities: Dict[str, ScheduleActivity] = {}
self.cost_items: Dict[str, CostItem] = {}
self.links: List[ActivityCostLink] = []
self.actual_costs: Dict[str, float] = {} # activity_id -> actual cost
def add_activity(self,
activity_id: str,
name: str,
start_date: date,
finish_date: date,
percent_complete: float = 0):
"""Add schedule activity."""
duration = (finish_date - start_date).days + 1
self.activities[activity_id] = ScheduleActivity(
activity_id=activity_id,
name=name,
start_date=start_date,
finish_date=finish_date,
duration=duration,
percent_complete=percent_complete
)
def add_cost_item(self,
cost_code: str,
description: str,
budgeted_cost: float,
labor_pct: float = 0.4,
material_pct: float = 0.5,
equipment_pct: float = 0.1):
"""Add cost item."""
self.cost_items[cost_code] = CostItem(
cost_code=cost_code,
description=description,
budgeted_cost=budgeted_cost,
labor_cost=budgeted_cost * labor_pct,
material_cost=budgeted_cost * material_pct,
equipment_cost=budgeted_cost * equipment_pct
)
def link_activity_cost(self,
activity_id: str,
cost_code: str,
loading_method: LoadingMethod = LoadingMethod.UNIFORM):
"""Link activity to cost item."""
if activity_id not in self.activities:
return
cost_item = self.cost_items.get(cost_code)
budgeted = cost_item.budgeted_cost if cost_item else 0
self.links.append(ActivityCostLink(
activity_id=activity_id,
cost_code=cost_code,
budgeted_cost=budgeted,
loading_method=loading_method
))
def record_actual_cost(self, activity_id: str, actual_cost: float):
"""Record actual cost for activity."""
self.actual_costs[activity_id] = actual_cost
def _distribute_cost(self,
cost: float,
start_date: date,
duration: int,
method: LoadingMethod) -> Dict[date, float]:
"""Distribute cost over activity duration."""
daily_costs = {}
if duration <= 0:
return {start_date: cost}
if method == LoadingMethod.UNIFORM:
daily = cost / duration
for i in range(duration):
daily_costs[start_date + timedelta(days=i)] = daily
elif method == LoadingMethod.FRONT_LOADED:
total_weight = sum(range(duration, 0, -1))
for i in range(duration):
weight = (duration - i) / total_weight
daily_costs[start_date + timedelta(days=i)] = cost * weight
elif method == LoadingMethod.BACK_LOADED:
total_weight = sum(range(1, duration + 1))
for i in range(duration):
weight = (i + 1) / total_weight
daily_costs[start_date + timedelta(days=i)] = cost * weight
elif method == LoadingMethod.BELL_CURVE:
# Simplified bell curve
mid = duration / 2
for i in range(duration):
distance = abs(i - mid)
weight = 1 - (distance / mid) * 0.5
daily_costs[start_date + timedelta(days=i)] = cost * weight / duration
return daily_costs
def generate_cost_loaded_schedule(self) -> pd.DataFrame:
"""Generate cost-loaded schedule."""
data = []
for link in self.links:
activity = self.activities.get(link.activity_id)
cost_item = self.cost_items.get(link.cost_code)
if activity and cost_item:
data.append({
'Activity ID': activity.activity_id,
'Activity Name': activity.name,
'Cost Code': link.cost_code,
'Description': cost_item.description,
'Start': activity.start_date,
'Finish': activity.finish_date,
'Duration': activity.duration,
'Budget': link.budgeted_cost,
'% Complete': activity.percent_complete,
'Earned Value': link.budgeted_cost * activity.percent_complete / 100,
'Loading': link.loading_method.value
})
return pd.DataFrame(data)
def generate_cash_flow(self,
project_start: date = None,
project_end: date = None) -> pd.DataFrame:
"""Generate cash flow curve."""
if not self.links:
return pd.DataFrame()
# Get date range
if project_start is None:
project_start = min(self.activities[l.activity_id].start_date for l in self.links)
if project_end is None:
project_end = max(self.activities[l.activity_id].finish_date for l in self.links)
# Aggregate daily costs
daily_totals = defaultdict(float)
for link in self.links:
activity = self.activities.get(link.activity_id)
if not activity:
continue
daily_costs = self._distribute_cost(
link.budgeted_cost,
activity.start_date,
activity.duration,
link.loading_method
)
for day, cost in daily_costs.items():
daily_totals[day] += cost
# Build cash flow data
data = []
cumulative = 0
current = project_start
while current <= project_end:
daily = daily_totals.get(current, 0)
cumulative += daily
data.append({
'Date': current,
'Daily': round(daily, 2),
'Cumulative': round(cumulative, 2),
'Cumulative %': round(cumulative / self.bac * 100, 1) if self.bac > 0 else 0
})
current += timedelta(days=1)
return pd.DataFrame(data)
def calculate_earned_value(self, data_date: date) -> EarnedValueMetrics:
"""Calculate earned value metrics at data date."""
# BCWS - Planned Value through data date
bcws = 0
for link in self.links:
activity = self.activities.get(link.activity_id)
if not activity:
continue
daily_costs = self._distribute_cost(
link.budgeted_cost,
activity.start_date,
activity.duration,
link.loading_method
)
for day, cost in daily_costs.items():
if day <= data_date:
bcws += cost
# BCWP - Earned Value (budget * % complete)
bcwp = 0
for link in self.links:
activity = self.activities.get(link.activity_id)
if activity:
bcwp += link.budgeted_cost * activity.percent_complete / 100
# ACWP - Actual Cost
acwp = sum(self.actual_costs.values())
# Variances
sv = bcwp - bcws
cv = bcwp - acwp
# Indices
spi = bcwp / bcws if bcws > 0 else 0
cpi = bcwp / acwp if acwp > 0 else 0
# Forecasts
eac = self.bac / cpi if cpi > 0 else self.bac
etc = eac - acwp
vac = self.bac - eac
return EarnedValueMetrics(
data_date=data_date,
bcws=round(bcws, 2),
bcwp=round(bcwp, 2),
acwp=round(acwp, 2),
sv=round(sv, 2),
cv=round(cv, 2),
spi=round(spi, 2),
cpi=round(cpi, 2),
eac=round(eac, 2),
etc=round(etc, 2),
vac=round(vac, 2)
)
def get_monthly_cash_flow(self) -> pd.DataFrame:
"""Aggregate cash flow by month."""
daily = self.generate_cash_flow()
if daily.empty:
return pd.DataFrame()
daily['Month'] = pd.to_datetime(daily['Date']).dt.to_period('M')
monthly = daily.groupby('Month').agg({
'Daily': 'sum',
'Cumulative': 'last'
}).reset_index()
monthly.columns = ['Month', 'Monthly Cost', 'Cumulative']
return monthly
def export_to_excel(self, output_path: str) -> str:
"""Export integrated data to Excel."""
with pd.ExcelWriter(output_path, engine='openpyxl') as writer:
# Cost-loaded schedule
schedule = self.generate_cost_loaded_schedule()
schedule.to_excel(writer, sheet_name='Cost-Loaded Schedule', index=False)
# Cash flow
cash_flow = self.generate_cash_flow()
if not cash_flow.empty:
cash_flow.to_excel(writer, sheet_name='Cash Flow', index=False)
# Monthly
monthly = self.get_monthly_cash_flow()
if not monthly.empty:
monthly.to_excel(writer, sheet_name='Monthly', index=False)
# Earned Value
evm = self.calculate_earned_value(date.today())
evm_df = pd.DataFrame([{
'Data Date': evm.data_date,
'BCWS (PV)': evm.bcws,
'BCWP (EV)': evm.bcwp,
'ACWP (AC)': evm.acwp,
'SV': evm.sv,
'CV': evm.cv,
'SPI': evm.spi,
'CPI': evm.cpi,
'EAC': evm.eac,
'ETC': evm.etc,
'VAC': evm.vac
}])
evm_df.to_excel(writer, sheet_name='Earned Value', index=False)
return output_path
Quick Start
from datetime import date, timedelta
Initialize linker
linker = ScheduleCostLinker("Office Building", budget_at_completion=5000000)
Add activities
linker.add_activity("A-001", "Foundation", date(2024, 6, 1), date(2024, 6, 30), percent_complete=100) linker.add_activity("A-002", "Structure", date(2024, 7, 1), date(2024, 9, 30), percent_complete=60) linker.add_activity("A-003", "MEP", date(2024, 8, 1), date(2024, 11, 30), percent_complete=30)
Add cost items
linker.add_cost_item("01-FOUND", "Foundation Work", 500000) linker.add_cost_item("02-STRUCT", "Structural Work", 2000000) linker.add_cost_item("03-MEP", "MEP Systems", 1500000)
Link
linker.link_activity_cost("A-001", "01-FOUND") linker.link_activity_cost("A-002", "02-STRUCT", LoadingMethod.BELL_CURVE) linker.link_activity_cost("A-003", "03-MEP", LoadingMethod.BACK_LOADED)
Record actuals
linker.record_actual_cost("A-001", 520000) linker.record_actual_cost("A-002", 1300000)
Common Use Cases
- Earned Value Analysis
evm = linker.calculate_earned_value(date.today()) print(f"CPI: {evm.cpi}") print(f"SPI: {evm.spi}") print(f"EAC: ${evm.eac:,.2f}")
- Cash Flow Forecast
cash_flow = linker.generate_cash_flow() print(cash_flow.tail(10))
- Monthly Breakdown
monthly = linker.get_monthly_cash_flow() print(monthly)
Resources
- DDC Book: Chapter 4.2 - Schedule-Cost Integration