import pandas as pd
from scipy.optimize import root_scalar
OPERATIONS = [
{"site": "Clean Express", "role": "Data Analyst", "cost_before": 1.00, "cost_target": 0.80, "output_before": 1.00, "output_target": 1.20},
{"site": "Proptverse", "role": "Senior Data Analyst", "cost_before": 1.00, "cost_target": 0.88, "output_before": 1.00, "output_target": 1.25},
{"site": "Turing", "role": "AI Quality Engineer", "cost_before": 1.00, "cost_target": 0.94, "output_before": 1.00, "output_target": 1.30},
]
CLEAN_EXPRESS_RECOVERY_FLOOR = 0.20
PROPTVERSE_RECOVERY_FLOOR = 0.25
TURING_QUALITY_FLOOR = 0.30
def load_operations(rows):
"""Build the operational ledger and derive every efficiency delta."""
frame = pd.DataFrame(rows)
frame["cost_delta_pct"] = (frame["cost_target"] / frame["cost_before"] - 1.0) * 100.0
frame["output_delta_pct"] = (frame["output_target"] / frame["output_before"] - 1.0) * 100.0
frame["cost_saved_pct"] = -frame["cost_delta_pct"]
frame["efficiency_shift_pct"] = frame["output_delta_pct"] + frame["cost_saved_pct"]
return frame
def recovery_curve(floor, ceiling=0.98, steps=6):
"""Model a concave recovery path from the floor to the ceiling."""
curve = pd.DataFrame({
"step": range(steps),
"recovery_rate": [
floor + (ceiling - floor) * (step / (steps - 1)) ** 0.5
for step in range(steps)
],
})
curve["recovery_pct"] = curve["recovery_rate"] * 100.0
return curve
def break_even_volume(fixed_cost, unit_margin, target_margin):
"""Solve fixed_cost / (unit_margin - target_margin) for scaled volume."""
if unit_margin <= target_margin:
raise ValueError("unit_margin must exceed target_margin")
result = root_scalar(
lambda volume: unit_margin - fixed_cost / volume - target_margin,
bracket=[1.0, 1_000_000.0],
)
return float(result.root)
def efficiency_report():
ledger = load_operations(OPERATIONS)
ledger = ledger.sort_values("efficiency_shift_pct", ascending=False)
summary = ledger[
["site", "role", "cost_saved_pct", "output_delta_pct", "efficiency_shift_pct"]
].copy()
summary = summary.round(2)
summary.columns = [
"site",
"role",
"cost_saved_pct",
"output_scale_pct",
"efficiency_shift_pct",
]
return summary.reset_index(drop=True)
def model_quality_protocol(quality_floor=TURING_QUALITY_FLOOR):
"""Enforce the Turing model quality checking protocols."""
protocols = [
{"protocol": "rubric_adherence", "score": 0.94, "owner": "Turing"},
{"protocol": "annotation_agreement", "score": 0.91, "owner": "Turing"},
{"protocol": "safety_gate", "score": 0.97, "owner": "Turing"},
]
frame = pd.DataFrame(protocols)
frame["floor"] = quality_floor
frame['passes_floor'] = frame['score'] >= frame['floor']
frame["margin"] = (frame["score"] - frame["floor"]).round(3)
return frame
if __name__ == "__main__":
report = efficiency_report()
quality = model_quality_protocol()
scale = break_even_volume(
fixed_cost=42_000.0,
unit_margin=18.0,
target_margin=4.5,
)
curve = recovery_curve(PROPTVERSE_RECOVERY_FLOOR)
print(report.to_string(index=False))
print()
print(quality.to_string(index=False))
print()
metadata = {
"clean_express_floor": CLEAN_EXPRESS_RECOVERY_FLOOR,
"proptverse_floor": PROPTVERSE_RECOVERY_FLOOR,
"turing_floor": TURING_QUALITY_FLOOR,
}
print(metadata)
summary = {
"break_even_volume": round(scale, 1),
"recovery_final_pct": round(float(curve["recovery_pct"].iloc[-1]), 2),
}
print(summary)