Catalyst / admin/Apex 14.8 GB / 57.8 GB 40.0 GB free
Help Sign in

admin / Apex

public

Bid Management and Orchestration Tool with AI Capability

Code Issues Pull requests Pipelines Packages Security Insights Wiki Settings
Apex / Synapse-Apexv2 / backend / app / services / costing.py 6795 B · main 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
"""Cost engine (Module 3) and pricing maths (Module 4).

All money maths use Decimal. These functions are pure and unit-tested.
"""
from __future__ import annotations

from decimal import ROUND_HALF_UP, Decimal

TWOPLACES = Decimal("0.01")


def q(value: Decimal) -> Decimal:
    return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP)


def compute_effort_lines(rules: list[dict], inputs: dict) -> list[dict]:
    """Resolve effort rules against endpoint inputs → resource hour lines.

    rules: [{name, trigger, endpoint_type, per_qty, resource_type_id,
             resource_name, hourly_cost_rate, hours_recurring, hours_oneoff}]
    inputs: {endpoints: {servers, workstations, network_devices, users, sites},
             term_months, scope: {...}}
    """
    endpoints = inputs.get("endpoints", {})
    lines: list[dict] = []
    for r in rules:
        trigger = r["trigger"]
        per_qty = Decimal(str(r.get("per_qty") or 1))
        if trigger == "fixed":
            multiplier = Decimal("1")
        elif trigger in ("per_endpoint", "per_site"):
            etype = r.get("endpoint_type") or ("sites" if trigger == "per_site" else "")
            count = Decimal(str(endpoints.get(etype, 0) or 0))
            if per_qty == 0:
                continue
            multiplier = count / per_qty
        else:
            continue
        rec = Decimal(str(r.get("hours_recurring") or 0)) * multiplier
        one = Decimal(str(r.get("hours_oneoff") or 0)) * multiplier
        if rec == 0 and one == 0:
            continue
        rate = Decimal(str(r.get("hourly_cost_rate") or 0))
        if rec > 0:
            lines.append({
                "group": "labour_recurring",
                "description": f"{r.get('resource_name', 'Resource')} — {r['name']}",
                "resource_type_id": r.get("resource_type_id"),
                "qty": 1,
                "hours": q(rec),
                "rate_snapshot": q(rate),
                "monthly": q(rec * rate),
                "oneoff": Decimal("0.00"),
            })
        if one > 0:
            lines.append({
                "group": "labour_oneoff",
                "description": f"{r.get('resource_name', 'Resource')} — {r['name']} (one-off)",
                "resource_type_id": r.get("resource_type_id"),
                "qty": 1,
                "hours": q(one),
                "rate_snapshot": q(rate),
                "monthly": Decimal("0.00"),
                "oneoff": q(one * rate),
            })
    return lines


def compute_licence_lines(licences: list[dict], inputs: dict) -> list[dict]:
    endpoints = inputs.get("endpoints", {})
    total_endpoints = Decimal(str(
        (endpoints.get("servers", 0) or 0)
        + (endpoints.get("workstations", 0) or 0)
        + (endpoints.get("network_devices", 0) or 0)
    ))
    users = Decimal(str(endpoints.get("users", 0) or 0))
    sites = Decimal(str(endpoints.get("sites", 0) or 0))
    lines: list[dict] = []
    for lic in licences:
        unit = lic["unit"]
        unit_cost = Decimal(str(lic["unit_cost"]))
        qty = {
            "per_endpoint": total_endpoints,
            "per_user": users,
            "per_site": sites,
            "flat": Decimal("1"),
        }.get(unit, Decimal("0"))
        amount = qty * unit_cost
        if amount == 0:
            continue
        billing = lic.get("billing", "monthly")
        monthly = amount if billing == "monthly" else (amount / 12 if billing == "annual" else Decimal("0"))
        oneoff = amount if billing == "one_off" else Decimal("0")
        lines.append({
            "group": "licensing",
            "description": f"{lic['name']} ({unit})",
            "resource_type_id": None,
            "qty": q(qty),
            "hours": Decimal("0.00"),
            "rate_snapshot": q(unit_cost),
            "monthly": q(monthly),
            "oneoff": q(oneoff),
        })
    return lines


def compute_non_people_lines(inputs: dict) -> list[dict]:
    """User-entered non-people resource costs (hardware, travel, connectivity, etc.)."""
    lines: list[dict] = []
    for item in inputs.get("non_people", []) or []:
        monthly = Decimal(str(item.get("monthly", 0) or 0))
        oneoff = Decimal(str(item.get("oneoff", 0) or 0))
        if monthly == 0 and oneoff == 0:
            continue
        lines.append({
            "group": "non_people",
            "description": item.get("description", "Non-people cost"),
            "resource_type_id": None,
            "qty": Decimal("1"),
            "hours": Decimal("0.00"),
            "rate_snapshot": Decimal("0.00"),
            "monthly": q(monthly),
            "oneoff": q(oneoff),
        })
    return lines


def sheet_totals(lines: list[dict], term_months: int) -> dict:
    monthly = sum((Decimal(str(l["monthly"])) for l in lines), Decimal("0"))
    oneoff = sum((Decimal(str(l["oneoff"])) for l in lines), Decimal("0"))
    total_contract = monthly * Decimal(str(term_months)) + oneoff
    return {
        "monthly": q(monthly),
        "oneoff": q(oneoff),
        "term_months": term_months,
        "total_contract": q(total_contract),
    }


def price_scenario(base_cost: Decimal, risk_pct: Decimal, mgmt_pct: Decimal,
                   margin_pct: Decimal, term_months: int) -> dict:
    """Waterfall: base → +risk → +mgmt → total cost → ÷(1-margin) → sell price.

    margin_pct is margin ON PRICE (not markup). All inputs Decimal percentages.
    """
    base = Decimal(str(base_cost))
    risk_amt = base * risk_pct / Decimal("100")
    mgmt_amt = base * mgmt_pct / Decimal("100")
    total_cost = base + risk_amt + mgmt_amt
    margin_frac = margin_pct / Decimal("100")
    if margin_frac >= Decimal("1"):
        raise ValueError("margin must be < 100%")
    sell_price = total_cost / (Decimal("1") - margin_frac)
    profit = sell_price - total_cost
    monthly_price = sell_price / Decimal(str(term_months)) if term_months else Decimal("0")
    return {
        "base_cost": q(base),
        "risk_amount": q(risk_amt),
        "mgmt_amount": q(mgmt_amt),
        "total_cost": q(total_cost),
        "margin_pct": q(margin_pct),
        "sell_price": q(sell_price),
        "profit": q(profit),
        "monthly_price": q(monthly_price),
        "term_months": term_months,
        "waterfall": [
            {"label": "Base delivery cost", "value": q(base)},
            {"label": f"Risk contingency ({q(risk_pct)}%)", "value": q(risk_amt)},
            {"label": f"Management contingency ({q(mgmt_pct)}%)", "value": q(mgmt_amt)},
            {"label": "Total cost", "value": q(total_cost)},
            {"label": f"Margin ({q(margin_pct)}% of price)", "value": q(profit)},
            {"label": "Sell price (total contract)", "value": q(sell_price)},
        ],
    }