Create analysis_runtime.py

analysis_runtime.py

@@ -0,0 +1,48 @@
+# analysis_runtime.py
+from __future__ import annotations
+from typing import Dict, Any, List, Optional
+import pandas as pd
+
+class ExecContext:
+    def __init__(self, frames: Dict[str,pd.DataFrame]):
+        self.frames = frames  # name -> DataFrame
+
+def _find_frame_with_columns(ctx: ExecContext, cols: List[str]) -> Optional[pd.DataFrame]:
+    for df in ctx.frames.values():
+        if all(c in df.columns for c in cols):
+            return df
+    return None
+
+def op_summary_table(ctx: ExecContext, metrics: List[str], by: List[str], aggs: List[str]=["median"]) -> Optional[pd.DataFrame]:
+    df = _find_frame_with_columns(ctx, metrics + by)
+    if df is None: return None
+    agg_spec = {m: aggs for m in metrics}
+    out = df.groupby(by, dropna=False).agg(agg_spec)
+    out.columns = [f"{m}_{a}" for m,a in out.columns]
+    return out.reset_index()
+
+def op_rank_top_n(ctx: ExecContext, metric: str, by: str, n: int=5, agg: str="median", ascending=False) -> Optional[pd.DataFrame]:
+    df = _find_frame_with_columns(ctx, [metric, by])
+    if df is None: return None
+    g = df.groupby(by, dropna=False)[metric].agg(agg).sort_values(ascending=ascending).head(n)
+    return g.reset_index().rename(columns={metric:f"{metric}_{agg}"})
+
+def op_delta_over_time(ctx: ExecContext, metric: str, by: List[str], time_col: str, t0: str, t1: str, agg: str="median") -> Optional[pd.DataFrame]:
+    df = _find_frame_with_columns(ctx, [metric, time_col] + by)
+    if df is None: return None
+    a = df[df[time_col]==t0].groupby(by, dropna=False)[metric].agg(agg)
+    b = df[df[time_col]==t1].groupby(by, dropna=False)[metric].agg(agg)
+    out = (b - a).reset_index()
+    out = out.rename(columns={0:f"{metric}_delta_{agg}_{t0}_to_{t1}"})
+    return out
+
+def op_capacity_calc(clients_per_day: float, teams: int, days: int) -> pd.DataFrame:
+    total = float(clients_per_day) * float(teams) * float(days)
+    return pd.DataFrame([{"clients_per_day": clients_per_day, "teams": teams, "days": days, "capacity_total": total}])
+
+def op_cost_total(startup_per_client: float, ongoing_per_client: float, volume: int) -> pd.DataFrame:
+    total = float(startup_per_client + ongoing_per_client) * float(volume)
+    return pd.DataFrame([{
+        "startup_per_client": startup_per_client, "ongoing_per_client": ongoing_per_client,
+        "volume": volume, "total_cost": total
+    }])