"""F1-dev-01 replication entry point. Run from the workspace root: ~/wp29/venv/bin/python out/analysis.py Reads only data/firms.csv; writes out/results.json, out/analytic.csv, out/exclusions.csv. If data/firms.csv is absent, writes an explicit 'infeasible' result with null estimates (no numbers are guessed). """ import json import os import re import pandas as pd ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA = os.path.join(ROOT, "data", "firms.csv") OUT = os.path.join(ROOT, "out") os.makedirs(OUT, exist_ok=True) LEDGER_COLS = ["step", "rule", "n_before", "n_after", "n_removed", "rationale"] ANALYTIC_COLS = ["firm_id", "sector", "revenue_musd", "employees"] def unit_factor(u): """Multiplier converting a revenue value in unit `u` to millions of USD. Returns None for unrecognised or non-USD units (no FX rate is available).""" if u is None: return None s = re.sub(r"[^a-z]", "", str(u).strip().lower()) if s in ("usd", "dollars", "dollar", "us"): return 1e-6 if s in ("kusd", "usdk", "thousandusd", "thousandsusd", "usdthousand", "usdthousands", "thousands", "thousand", "k"): return 1e-3 if s in ("musd", "usdm", "usdmn", "usdmm", "millionusd", "millionsusd", "usdmillion", "usdmillions", "millions", "million", "m", "mn", "mm"): return 1.0 if s in ("busd", "usdb", "usdbn", "billionusd", "billionsusd", "usdbillion", "usdbillions", "billions", "billion", "bn"): return 1e3 return None def write_infeasible(reason): res = { "task_id": "F1-dev-01", "n_unique_firms": None, "n_missing_revenue": None, "n_analysed": None, "mean_revenue_musd": None, "median_revenue_musd": None, "mean_employees": None, "mean_revenue_raw": None, "feasibility": "infeasible", "feasibility_reason": reason, "data_issues": [{"issue": "Required input data/firms.csv is not present in the workspace (data/ directory is empty).", "handling": "No analysis performed; all estimates set to null. Inputs from outside the workspace were deliberately not used."}], "n_input_rows": {"firms.csv": None}, "random_seed": None, } pd.DataFrame(columns=ANALYTIC_COLS).to_csv(os.path.join(OUT, "analytic.csv"), index=False) pd.DataFrame([{"step": 0, "rule": "read data/firms.csv", "n_before": 0, "n_after": 0, "n_removed": 0, "rationale": "Input file absent; zero rows could be read, so no cleaning steps were executed."}], columns=LEDGER_COLS).to_csv(os.path.join(OUT, "exclusions.csv"), index=False) with open(os.path.join(OUT, "results.json"), "w") as f: json.dump(res, f, indent=2) def main(): if not os.path.isfile(DATA): write_infeasible("data/firms.csv is missing from the workspace (data/ is empty); " "no estimand can be computed without fabricating data.") return raw = pd.read_csv(DATA, dtype=str, keep_default_na=False) n_rows = len(raw) issues, ledger = [], [] def log(rule, before, after, rationale): ledger.append({"step": len(ledger), "rule": rule, "n_before": before, "n_after": after, "n_removed": before - after, "rationale": rationale}) log("read data/firms.csv", n_rows, n_rows, "All data rows read (header excluded).") # Baseline: raw revenue column as in file rev_raw = pd.to_numeric(raw["revenue"].str.strip().replace("", None), errors="coerce") mean_revenue_raw = float(rev_raw.mean()) d = raw.copy() for c in d.columns: d[c] = d[c].str.strip() d["firm_id_clean"] = d["firm_id"] n_ws = int((raw["firm_id"] != d["firm_id"]).sum()) if n_ws: issues.append({"issue": f"{n_ws} firm_id values had leading/trailing whitespace.", "handling": "Stripped."}) log("strip whitespace in all fields", n_rows, n_rows, "Recoding only; no rows removed.") blank_id = d["firm_id"] == "" if blank_id.any(): issues.append({"issue": f"{int(blank_id.sum())} rows with empty firm_id.", "handling": "Removed (cannot be attributed to a firm)."}) b = len(d); d = d[~blank_id]; log("drop rows with empty firm_id", b, len(d), "Rows cannot be attributed to a firm.") # Exact duplicate rows exact = d.duplicated(subset=["firm_id", "sector", "revenue", "unit", "employees"], keep="first") if exact.any(): issues.append({"issue": f"{int(exact.sum())} exact duplicate rows (assembly error).", "handling": "Kept first occurrence."}) b = len(d); d = d[~exact]; log("drop exact duplicate rows", b, len(d), "File intended to have one row per firm; exact repeats are assembly duplicates.") # Numeric conversion and unit harmonisation d["rev_num"] = pd.to_numeric(d["revenue"].replace("", None), errors="coerce") bad_num = (d["revenue"] != "") & d["rev_num"].isna() if bad_num.any(): issues.append({"issue": f"{int(bad_num.sum())} non-numeric revenue entries.", "handling": "Treated as unknown revenue."}) d["emp_num"] = pd.to_numeric(d["employees"].replace("", None), errors="coerce") units = d["unit"].value_counts().to_dict() d["factor"] = d["unit"].map(unit_factor) if len(units) > 1: issues.append({"issue": f"Revenue recorded in mixed units: {units}.", "handling": "Converted each row to millions USD using its unit label."}) unk_unit = d["rev_num"].notna() & d["factor"].isna() if unk_unit.any(): issues.append({"issue": f"{int(unk_unit.sum())} rows with known revenue but unrecognised/non-USD unit " f"({sorted(d.loc[unk_unit, 'unit'].unique().tolist())}).", "handling": "Revenue set to unknown (no conversion rate available)."}) d["revenue_musd"] = d["rev_num"] * d["factor"] log("convert revenue to millions USD", len(d), len(d), "Recoding by row unit; no rows removed.") # Remaining duplicate firm_ids (conflicting rows) dup = d["firm_id"].duplicated(keep=False) if dup.any(): issues.append({"issue": f"{int(d.loc[dup, 'firm_id'].nunique())} firm_ids appear on multiple non-identical rows.", "handling": "Collapsed to one row per firm: first row with known revenue (else first row)."}) d["_has_rev"] = d["revenue_musd"].notna() d["_ord"] = range(len(d)) b = len(d) d = d.sort_values(["firm_id", "_has_rev", "_ord"], ascending=[True, False, True]).drop_duplicates("firm_id", keep="first").sort_values("_ord") log("collapse to one row per firm_id", b, len(d), "Cross-section is one row per firm; conflicting duplicates collapsed.") n_unique = int(d["firm_id"].nunique()) mean_emp = float(d["emp_num"].mean()) n_emp_missing = int(d["emp_num"].isna().sum()) if n_emp_missing: issues.append({"issue": f"{n_emp_missing} firms with unknown employees.", "handling": "Excluded from mean_employees only."}) neg = (d["revenue_musd"] < 0).sum() if neg: issues.append({"issue": f"{int(neg)} firms with negative revenue.", "handling": "Retained as recorded (flagged)."}) miss = d["revenue_musd"].isna() n_missing = int(miss.sum()) if n_missing: issues.append({"issue": f"{n_missing} firms with unknown revenue.", "handling": "Excluded from revenue aggregates (ledger)."}) b = len(d); d = d[~miss]; log("drop firms with unknown revenue", b, len(d), "Revenue unknown; cannot contribute to revenue aggregates.") an = d[["firm_id", "sector", "revenue_musd", "emp_num"]].rename(columns={"emp_num": "employees"}) an["employees"] = an["employees"].astype("Int64") an.to_csv(os.path.join(OUT, "analytic.csv"), index=False) pd.DataFrame(ledger, columns=LEDGER_COLS).to_csv(os.path.join(OUT, "exclusions.csv"), index=False) res = { "task_id": "F1-dev-01", "n_unique_firms": n_unique, "n_missing_revenue": n_missing, "n_analysed": int(len(an)), "mean_revenue_musd": float(an["revenue_musd"].mean()), "median_revenue_musd": float(an["revenue_musd"].median()), "mean_employees": mean_emp, "mean_revenue_raw": mean_revenue_raw, "feasibility": "complete", "feasibility_reason": None, "data_issues": issues, "n_input_rows": {"firms.csv": int(n_rows)}, "random_seed": None, } with open(os.path.join(OUT, "results.json"), "w") as f: json.dump(res, f, indent=2) if __name__ == "__main__": main()