104 lines
5.6 KiB
Python
104 lines
5.6 KiB
Python
"""Read-only, independent audit of the three source workbooks."""
|
|
from pathlib import Path
|
|
from collections import Counter
|
|
import json
|
|
import sys
|
|
import math
|
|
import re
|
|
import xlrd
|
|
import pandas as pd
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
ROOT = Path(__file__).resolve().parent
|
|
DATA = ROOT / "data"
|
|
out = {}
|
|
|
|
f = next(DATA.glob("FIN*.xls"))
|
|
w = xlrd.open_workbook(f, formatting_info=True)
|
|
s = w.sheet_by_index(0)
|
|
rows = [s.row_values(i) for i in range(1, s.nrows)]
|
|
cols = ["comcd", "name", "stkcd", "b_stkcd", "h_stkcd", "lstflg", "excel_date", "reporttype", "roa", "roe"]
|
|
fin = pd.DataFrame(rows, columns=cols)
|
|
fin["date"] = fin.excel_date.map(lambda v: xlrd.xldate_as_datetime(v, w.datemode).strftime("%Y-%m-%d"))
|
|
fin["year"] = fin.date.str[:4].astype(int)
|
|
out["finance"] = {
|
|
"file": f.name, "rows": len(fin), "columns": s.row_values(0),
|
|
"year_counts": fin.groupby("year").size().to_dict(),
|
|
"companies": int(fin.stkcd.nunique()),
|
|
"date_suffixes": dict(Counter(fin.date.str[5:])),
|
|
"listing_flags": dict(Counter(fin.lstflg)),
|
|
"report_types": dict(Counter(fin.reporttype)),
|
|
"missing_a": int((fin.stkcd == "").sum()),
|
|
"nonsixdigit_code_rows": [
|
|
{"excel_row":i+2, "stkcd_repr":repr(row["stkcd"]), "cell_type":s.cell(i+1,2).ctype,
|
|
"comcd":row["comcd"], "name":row["name"], "year":int(row["year"])}
|
|
for i,row in fin.iterrows() if not re.fullmatch(r"[0-9]{6}",str(row["stkcd"]).strip())
|
|
],
|
|
"populated_b": int((fin.b_stkcd != "").sum()),
|
|
"populated_h": int((fin.h_stkcd != "").sum()),
|
|
"duplicate_key_rows": int(fin.duplicated(["stkcd", "year"], keep=False).sum()),
|
|
"duplicate_comcd_year_rows": int(fin.duplicated(["comcd", "year"], keep=False).sum()),
|
|
"formats": [w.format_map[w.xf_list[s.cell(1,j).xf_index].format_key].format_str for j in range(s.ncols)],
|
|
}
|
|
for variable in ("roa", "roe"):
|
|
original = fin[variable]
|
|
values = pd.to_numeric(original, errors="coerce")
|
|
fin[variable] = values
|
|
out["finance"][variable] = {
|
|
"missing": int(values.isna().sum()),
|
|
"unexpected_nonnumeric": original[values.isna() & (original != "")].tolist(),
|
|
"summary": values.describe(percentiles=[.01, .25, .5, .75, .99]).to_dict(),
|
|
"min_row": fin.loc[values.idxmin(), ["stkcd", "year", variable]].to_dict(),
|
|
"max_row": fin.loc[values.idxmax(), ["stkcd", "year", variable]].to_dict(),
|
|
"annual_summary": fin.groupby("year")[variable].agg(["count", "mean", "median", "min", "max"]).to_dict(orient="index"),
|
|
}
|
|
out["finance"]["code_prefix3_counts"] = fin.stkcd.str[:3].value_counts().to_dict()
|
|
out["finance"]["year_2014_code_prefix_counts"] = fin.loc[fin.year == 2014,"stkcd"].str[:3].value_counts().to_dict()
|
|
out["finance"]["last_source_rows"] = fin.tail(10)[["comcd", "name", "stkcd", "date", "roa", "roe"]].to_dict(orient="records")
|
|
|
|
govs = []
|
|
out["governance_sources"] = []
|
|
for f in sorted(DATA.glob("CG*.xls")):
|
|
w = xlrd.open_workbook(f)
|
|
s = w.sheet_by_index(0)
|
|
headers, labels, units = s.row_values(0), s.row_values(1), s.row_values(2)
|
|
df = pd.DataFrame([s.row_values(i) for i in range(3,s.nrows)], columns=headers)
|
|
df["year"] = df.Reptdt.str[:4].astype(int)
|
|
df["source"] = f.name
|
|
govs.append(df)
|
|
out["governance_sources"].append({
|
|
"file": f.name, "rows": len(df), "companies": int(df.Stkcd.nunique()),
|
|
"year_counts": df.groupby("year").size().to_dict(),
|
|
"date_suffixes": dict(Counter(df.Reptdt.str[5:])),
|
|
"duplicates": int(df.duplicated(["Stkcd", "year"], keep=False).sum()),
|
|
"metadata": [{"variable":a,"label":b,"unit":c} for a,b,c in zip(headers,labels,units)],
|
|
})
|
|
gov = pd.concat(govs, ignore_index=True).rename(columns={"Stkcd":"stkcd"})
|
|
merged = fin.merge(gov, how="outer", on=["stkcd", "year"], validate="1:1", indicator=True)
|
|
out["merge"] = {
|
|
"governance_rows": len(gov), "governance_companies": int(gov.stkcd.nunique()),
|
|
"governance_duplicate_rows": int(gov.duplicated(["stkcd","year"], keep=False).sum()),
|
|
"counts": {str(k):int(v) for k,v in merged._merge.value_counts().items()},
|
|
"year_counts": merged.groupby(["year","_merge"], observed=True).size().unstack(fill_value=0).to_dict(orient="index"),
|
|
"matched_companies": int(merged.loc[merged._merge == "both", "stkcd"].nunique()),
|
|
"matched_missing_roa": int(merged.loc[merged._merge == "both", "roa"].isna().sum()),
|
|
"matched_missing_roe": int(merged.loc[merged._merge == "both", "roe"].isna().sum()),
|
|
"gov_only_codes": sorted(merged.loc[merged._merge == "right_only", "stkcd"].unique()),
|
|
"gov_only_sample": merged.loc[merged._merge == "right_only", ["stkcd", "year"]].head(20).to_dict(orient="records"),
|
|
}
|
|
out["merge"]["matched_complete_both"] = int(((merged._merge == "both") & merged.roa.notna() & merged.roe.notna()).sum())
|
|
out["merge"]["matched_summary"] = merged.loc[merged._merge == "both", ["roa", "roe"]].describe(percentiles=[.01, .25, .5, .75, .99]).to_dict()
|
|
out["merge"]["governance_code_prefix3_counts"] = gov.stkcd.str[:3].value_counts().to_dict()
|
|
def clean_json(value):
|
|
if isinstance(value, dict):
|
|
return {str(k): clean_json(v) for k,v in value.items()}
|
|
if isinstance(value, list):
|
|
return [clean_json(v) for v in value]
|
|
if isinstance(value, float) and not math.isfinite(value):
|
|
return None
|
|
return value
|
|
|
|
out = clean_json(out)
|
|
(ROOT / "audit_independent.json").write_text(json.dumps(out, ensure_ascii=False, indent=2, default=int, allow_nan=False), encoding="utf-8")
|
|
print(json.dumps({"finance_rows":len(fin), "governance_rows":len(gov), "merge":out["merge"]["counts"], "audit_path":str(ROOT / "audit_independent.json")}, ensure_ascii=False, indent=2))
|