"""F1-dev-01 analysis. Run from the workspace root: ~/wp29/venv/bin/python out/analysis.py Reads only data/firms.csv. Writes out/results.json, out/analytic.csv and out/exclusions.csv. If data/firms.csv is absent (as it was when this script was written), every output is written in 'infeasible' mode: estimates are null, analytic.csv has only a header, and the ledger records 0 rows read. No random numbers are used (random_seed = null). """ import hashlib, json, os import pandas as pd TASK_ID = "F1-dev-01" SRC = os.path.join("data", "firms.csv") EXPECTED_SHA = "78f119a92d7c11aebb8c238c69d62687a80d91031f1e8a30fb412b9e77e1625a" # from bundle.json OUT = "out" # Multipliers to millions of USD. Only units listed here are converted; any other unit is excluded and reported. UNIT_TO_MUSD = {"usd": 1e-6, "$": 1e-6, "kusd": 1e-3, "usd_thousands": 1e-3, "thousand usd": 1e-3, "thousands usd": 1e-3, "musd": 1.0, "usd_millions": 1.0, "million usd": 1.0, "millions usd": 1.0, "usd millions": 1.0, "usd thousands": 1e-3, "busd": 1e3, "usd_billions": 1e3} COLS = ["firm_id", "sector", "revenue_musd", "employees"] LEDGER_COLS = ["step", "rule", "n_before", "n_after", "n_removed", "rationale"] def write(results, analytic, ledger): os.makedirs(OUT, exist_ok=True) analytic.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) with open(os.path.join(OUT, "results.json"), "w") as fh: json.dump(results, fh, indent=2, sort_keys=False) print(json.dumps({k: results[k] for k in ("feasibility", "n_analysed", "mean_revenue_musd")})) def infeasible(reason, issues): results = {"task_id": TASK_ID, "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": issues, "n_input_rows": {"firms.csv": None}, "random_seed": None} ledger = [[0, "read data/firms.csv", 0, 0, 0, "input file absent from workspace; no rows could be read; nothing analysed"]] write(results, pd.DataFrame(columns=COLS), ledger) def main(): if not os.path.isfile(SRC): infeasible("Required input data/firms.csv is absent from the workspace (data/ is empty), although " "bundle.json lists it (sha256 78f119a9...). TASK.md forbids any other input, so no estimand " "can be computed; all estimates are null rather than guessed.", [{"issue": "data/firms.csv missing from workspace", "handling": "task declared infeasible; all estimates set to null; no substitute data used"}]) return raw_bytes = open(SRC, "rb").read() sha = hashlib.sha256(raw_bytes).hexdigest() issues = [] if sha != EXPECTED_SHA: issues.append({"issue": f"sha256 {sha} differs from bundle manifest", "handling": "analysed as found; flagged"}) df = pd.read_csv(SRC, dtype=str, keep_default_na=False) n_raw = len(df) for c in df.columns: df[c] = df[c].str.strip() rev_num = pd.to_numeric(df["revenue"].replace("", None), errors="coerce") mean_raw = float(rev_num.mean()) if rev_num.notna().any() else None ledger, n = [], n_raw ledger.append([0, "read data/firms.csv", n_raw, n_raw, 0, "rows read (excluding header)"]) # step 1: exact duplicate rows d = df.drop_duplicates() if len(d) < n: issues.append({"issue": f"{n - len(d)} exact duplicate rows", "handling": "dropped, first kept"}) ledger.append([1, "drop exact duplicate rows", n, len(d), n - len(d), "assembly duplicates"]); n = len(d); df = d # step 2: conflicting duplicates of firm_id -> keep first occurrence (documented) dup = df["firm_id"].duplicated(keep="first") if dup.any(): issues.append({"issue": f"{int(dup.sum())} rows repeat a firm_id with differing values", "handling": "first occurrence kept; later rows dropped"}) df = df[~dup] ledger.append([2, "one row per firm_id (keep first)", n, len(df), n - len(df), "one intended row per firm"]); n = len(df) n_unique = int(df["firm_id"].nunique()) emp = pd.to_numeric(df["employees"].replace("", None), errors="coerce") if emp.isna().any(): issues.append({"issue": f"{int(emp.isna().sum())} firms with unknown employees", "handling": "excluded from mean_employees"}) mean_emp = float(emp.mean()) if emp.notna().any() else None # step 3: missing revenue rev = pd.to_numeric(df["revenue"].replace("", None), errors="coerce") miss = rev.isna() n_miss = int(miss.sum()) if n_miss: issues.append({"issue": f"{n_miss} firms with unknown or non-numeric revenue", "handling": "excluded from revenue aggregates"}) df = df[~miss].copy(); rev = rev[~miss] ledger.append([3, "drop firms with unknown revenue", n, len(df), n - len(df), "revenue unknown"]); n = len(df) # step 4: unit conversion mult = df["unit"].str.lower().map(UNIT_TO_MUSD) bad = mult.isna() if bad.any(): issues.append({"issue": f"{int(bad.sum())} rows with unrecognised unit {sorted(df.loc[bad, 'unit'].unique())}", "handling": "excluded (cannot convert to USD millions)"}) ledger.append([4, "convert revenue to USD millions (recode)", n, n, 0, "unit-specific multipliers"]) df = df[~bad].copy(); df["revenue_musd"] = rev[~bad] * mult[~bad] ledger.append([5, "drop rows with unrecognised unit", n, len(df), n - len(df), "no valid conversion"]); n = len(df) df["employees"] = pd.to_numeric(df["employees"].replace("", None), errors="coerce") analytic = df[COLS].reset_index(drop=True) feas = "complete" if not bad.any() else "partial" results = {"task_id": TASK_ID, "n_unique_firms": n_unique, "n_missing_revenue": n_miss, "n_analysed": int(n), "mean_revenue_musd": float(analytic["revenue_musd"].mean()) if n else None, "median_revenue_musd": float(analytic["revenue_musd"].median()) if n else None, "mean_employees": mean_emp, "mean_revenue_raw": mean_raw, "feasibility": feas, "feasibility_reason": None if feas == "complete" else "some units could not be converted", "data_issues": issues, "n_input_rows": {"firms.csv": int(n_raw)}, "random_seed": None} write(results, analytic, ledger) if __name__ == "__main__": main()