135 lines
8.8 KiB
Python
135 lines
8.8 KiB
Python
"""Independently reconcile Stata datasets to every supplied Excel source cell."""
|
|
from pathlib import Path
|
|
import json
|
|
import sys
|
|
import math
|
|
import hashlib
|
|
import xlrd
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
ROOT = Path(__file__).resolve().parent
|
|
results = {"method":"xlrd reads original source cells; pandas independently reads Stata dta", "checks":[], "passed":True}
|
|
|
|
def check(name, passed, **details):
|
|
results["checks"].append({"name":name,"passed":bool(passed),**details})
|
|
if not passed:
|
|
results["passed"] = False
|
|
print("FAIL", name, details)
|
|
|
|
def compare_frames(name, expected, actual, require_exact_numeric=False):
|
|
check(name+" row_count",len(expected)==len(actual), expected=len(expected), actual=len(actual))
|
|
check(name+" key_alignment",expected.index.equals(actual.index))
|
|
if not expected.index.equals(actual.index):
|
|
actual = actual.reindex(expected.index)
|
|
for col in expected.columns:
|
|
if col not in actual:
|
|
check(name+" column "+col,False,reason="missing column")
|
|
continue
|
|
exp, got = expected[col], actual[col]
|
|
if pd.api.types.is_datetime64_any_dtype(exp.dtype):
|
|
ok = ((exp == got) | (exp.isna() & got.isna()))
|
|
detail = {"kind":"date"}
|
|
elif pd.api.types.is_numeric_dtype(got.dtype):
|
|
exp = pd.to_numeric(exp.mask(exp.eq(""), np.nan),errors="raise").astype(float)
|
|
got = got.astype(float)
|
|
exact = (exp == got) | (exp.isna() & got.isna())
|
|
ok = exact if require_exact_numeric else np.isclose(exp.to_numpy(),got.to_numpy(),rtol=1e-13,atol=1e-13,equal_nan=True)
|
|
delta = np.abs(exp-got)
|
|
detail = {"kind":"numeric","exact_equal_cells":int(exact.sum()),"missing_source":int(exp.isna().sum()),"missing_output":int(got.isna().sum()),"max_absolute_difference":float(delta.max()) if delta.notna().any() else 0}
|
|
else:
|
|
exp = exp.fillna("").astype(str)
|
|
got = got.fillna("").astype(str)
|
|
ok = exp == got
|
|
detail = {"kind":"text","max_source_characters":int(exp.str.len().max()),"max_output_characters":int(got.str.len().max())}
|
|
ok = np.asarray(ok)
|
|
detail["cells_compared"] = int(len(exp))
|
|
detail["mismatches"] = int((~ok).sum())
|
|
if not ok.all():
|
|
bad = np.flatnonzero(~ok)[:5]
|
|
detail["examples"] = [{"key":str(exp.index[i]),"source":str(exp.iloc[i]),"output":str(got.iloc[i])} for i in bad]
|
|
check(name+" column "+col,ok.all(),**detail)
|
|
|
|
financial_file = next((ROOT/"data").glob("FIN*.xls"))
|
|
w=xlrd.open_workbook(financial_file)
|
|
s=w.sheet_by_index(0)
|
|
financial_cols=["Comcd","Lcomnm","A_Stkcd","B_Stkcd","H_Stkcd","Lstflg","Enddt","Reporttype","ROA","ROE"]
|
|
financial=pd.DataFrame([s.row_values(i) for i in range(1,s.nrows)],columns=financial_cols)
|
|
financial["Enddt"]=pd.to_datetime(financial.Enddt.map(lambda x:xlrd.xldate_as_datetime(x,w.datemode)))
|
|
financial["resset_excel_row"]=np.arange(2,s.nrows+1)
|
|
financial=financial.set_index("resset_excel_row").sort_index()
|
|
fin_dta=pd.read_stata(ROOT/"resset_financial_all_years.dta",convert_categoricals=False)
|
|
compare_frames("financial_source_vs_clean",financial,fin_dta.set_index("resset_excel_row").sort_index())
|
|
|
|
govs=[]
|
|
source_hashes={financial_file.name:hashlib.sha256(financial_file.read_bytes()).hexdigest()}
|
|
for f in sorted((ROOT/"data").glob("CG*.xls")):
|
|
w=xlrd.open_workbook(f)
|
|
s=w.sheet_by_index(0)
|
|
df=pd.DataFrame([s.row_values(i) for i in range(3,s.nrows)],columns=s.row_values(0))
|
|
df=df.rename(columns={"Stkcd":"stkcd"})
|
|
df["csmar_source"]=f.name
|
|
df["csmar_excel_row"]=np.arange(4,s.nrows+1)
|
|
govs.append(df)
|
|
source_hashes[f.name]=hashlib.sha256(f.read_bytes()).hexdigest()
|
|
governance=pd.concat(govs,ignore_index=True).set_index(["csmar_source","csmar_excel_row"]).sort_index()
|
|
gov_dta=pd.read_stata(ROOT/"csmar_governance_1999_2014.dta",convert_categoricals=False)
|
|
compare_frames("governance_source_vs_clean",governance,gov_dta.set_index(["csmar_source","csmar_excel_row"]).sort_index())
|
|
|
|
merged=pd.read_stata(ROOT/"financial_governance_all_years.dta",convert_categoricals=False)
|
|
check("merged_key_unique",not merged.duplicated(["stkcd","year"]).any())
|
|
compare_frames("financial_source_vs_merged",financial,merged.loc[merged.in_resset==1].set_index("resset_excel_row").sort_index())
|
|
compare_frames("governance_source_vs_merged",governance,merged.loc[merged.in_csmar==1].set_index(["csmar_source","csmar_excel_row"]).sort_index())
|
|
|
|
expected_fin=financial.reset_index()[["A_Stkcd","Enddt"]].rename(columns={"A_Stkcd":"stkcd"})
|
|
expected_fin["year"]=expected_fin.Enddt.dt.year
|
|
expected_gov=governance.reset_index()[["stkcd","Reptdt"]]
|
|
expected_gov["year"]=expected_gov.Reptdt.str[:4].astype(int)
|
|
expected_merge=expected_fin.merge(expected_gov,on=["stkcd","year"],how="outer",validate="1:1",indicator=True)
|
|
status_map={"left_only":1,"right_only":2,"both":3}
|
|
expected_merge["merge_status"]=expected_merge._merge.astype(str).map(status_map)
|
|
expected_merge=expected_merge.set_index(["stkcd","year"]).sort_index()
|
|
actual_merge=merged.set_index(["stkcd","year"]).sort_index()
|
|
check("merge_keyset",expected_merge.index.equals(actual_merge.index))
|
|
check("merge_status_every_row",np.array_equal(expected_merge.merge_status.to_numpy(),actual_merge._merge.to_numpy()))
|
|
check("in_resset_flag",np.array_equal(merged.in_resset.to_numpy(),merged._merge.isin([1,3]).astype(int).to_numpy()))
|
|
check("in_csmar_flag",np.array_equal(merged.in_csmar.to_numpy(),merged._merge.isin([2,3]).astype(int).to_numpy()))
|
|
check("matched_flag",np.array_equal(merged.matched.to_numpy(),(merged._merge==3).astype(int).to_numpy()))
|
|
check("complete_roa_roe_flag",np.array_equal(merged.complete_roa_roe.to_numpy(),((merged._merge==3)&merged.ROA.notna()&merged.ROE.notna()).astype(int).to_numpy()))
|
|
check("unusual_code_retained",merged.loc[merged.code_format_exception==1,["stkcd","year"]].to_dict(orient="records")==[{"stkcd":"0991","year":1998},{"stkcd":"0991","year":1999}])
|
|
|
|
expected_annual=expected_merge.reset_index().groupby(["year","merge_status"]).size().unstack(fill_value=0)
|
|
actual_annual=merged.groupby(["year","_merge"]).size().unstack(fill_value=0)
|
|
check("annual_merge_counts",expected_annual.equals(actual_annual))
|
|
|
|
def verify_csv_against_dta(stem):
|
|
dta=pd.read_stata(ROOT/(stem+".dta"),convert_categoricals=False)
|
|
numeric=[col for col in dta if pd.api.types.is_numeric_dtype(dta[col].dtype)]
|
|
dates=[col for col in dta if pd.api.types.is_datetime64_any_dtype(dta[col].dtype)]
|
|
types={col:(float if col in numeric else str) for col in dta}
|
|
csv=pd.read_csv(ROOT/(stem+".csv"),dtype=types,keep_default_na=False,
|
|
na_values={col:[""] for col in numeric},float_precision="round_trip",encoding="utf-8")
|
|
check(stem+" CSV column_names_order",list(dta.columns)==list(csv.columns),column_count=len(csv.columns))
|
|
for col in dates:
|
|
date_strings=csv[col]
|
|
valid=date_strings.eq("") | date_strings.str.fullmatch(r"\d{4}-\d{2}-\d{2}")
|
|
check(stem+" CSV ISO date "+col,valid.all())
|
|
csv[col]=pd.to_datetime(date_strings.mask(date_strings.eq("")),format="%Y-%m-%d",errors="raise")
|
|
compare_frames(stem+" DTA_vs_CSV",dta,csv,require_exact_numeric=True)
|
|
return {"rows":len(csv),"columns":len(csv.columns),"numeric_columns":len(numeric),
|
|
"date_columns":dates,"cells_compared":int(csv.size),"numeric_rule":"17-digit CSV parsed with float_precision=round_trip; exact equality including missing values"}
|
|
|
|
success_marker="WEEK8_ANALYSIS_COMPLETED_SUCCESSFULLY"
|
|
check("Stata_success_log",success_marker in (ROOT/"week8_stata.log").read_text(encoding="utf-8"))
|
|
results["csv_validation"]={stem:verify_csv_against_dta(stem) for stem in ["financial_governance_all_years","df_example_metrics"]}
|
|
results["annual_counts"]={str(y):{str(k):int(v) for k,v in row.items()} for y,row in actual_annual.to_dict(orient="index").items()}
|
|
results["summary"]={"financial_rows":len(financial),"governance_rows":len(governance),"merged_rows":len(merged),"merge_counts":{str(k):int(v) for k,v in merged._merge.value_counts().sort_index().items()},"financial_cells_compared_per_dataset":int(financial.size),"governance_cells_compared_per_dataset":int(governance.size),"matched_complete_roa_roe":int(merged.complete_roa_roe.sum()),"matched_missing_roa":int(merged.loc[merged.matched==1,"ROA"].isna().sum()),"matched_missing_roe":int(merged.loc[merged.matched==1,"ROE"].isna().sum())}
|
|
results["source_sha256"]=source_hashes
|
|
results["check_count"]=len(results["checks"])
|
|
results["summary"]["all_cells_compared"]=sum(x.get("cells_compared",0) for x in results["checks"])
|
|
(ROOT/"validation_results.json").write_text(json.dumps(results,ensure_ascii=False,indent=2,allow_nan=False),encoding="utf-8")
|
|
print(json.dumps({"passed":results["passed"],"check_count":results["check_count"],"summary":results["summary"]},ensure_ascii=False,indent=2))
|
|
if not results["passed"]:
|
|
raise SystemExit(1)
|