"""Verify Stata outputs against Excel and regenerate the two offline reports.
Requires: numpy, scipy, openpyxl. Run after week7_analysis.do.
"""
from pathlib import Path
import base64
import csv
import hashlib
import html
import json
import re
import xml.etree.ElementTree as ET
import numpy as np
from scipy import stats
from openpyxl import load_workbook
ROOT = Path(__file__).resolve().parent
def read_csv(name):
with (ROOT / name).open(encoding='utf-8-sig', newline='') as f:
return list(csv.DictReader(f))
def numeric(row):
return {k: float(v) for k, v in row.items()}
model = numeric(read_csv('model_statistics.csv')[0])
coeff = {r['term']: {k: float(v) for k, v in r.items() if k != 'term'} for r in read_csv('regression_coefficients.csv')}
robust = {r['term']: {k: float(v) for k, v in r.items() if k != 'term'} for r in read_csv('robust_coefficients.csv')}
groups = [numeric(r) for r in read_csv('income_group_summary.csv')]
audit = [numeric(r) for r in read_csv('source_totals_audit.csv')]
fitted = [numeric(r) for r in read_csv('households_fitted.csv')]
# Independent numerical validation: derive everything directly from source cells.
workbook = load_workbook(ROOT / 'Eg2_1_1/Eg2_1_1.xlsx', data_only=True)
sheet = workbook['Sheet1']
source = []
for col in range(3, 13):
for row in range(4, 17):
value = sheet.cell(row, col).value
if value is not None:
source.append((row, float(sheet.cell(3, col).value), float(value)))
source_array = np.array(source)
assert len(source) == 98
assert all((r['excel_row'], r['income'], r['consume']) == item for r, item in zip(fitted, source))
x, y = source_array[:, 1], source_array[:, 2]
X = np.column_stack([np.ones(len(x)), x])
beta = np.linalg.lstsq(X, y, rcond=None)[0]
yhat = X @ beta
resid = y - yhat
n, k = X.shape
inv = np.linalg.inv(X.T @ X)
rss = resid @ resid
se = np.sqrt(np.diag((rss / (n-k)) * inv))
hc1 = inv @ (X.T @ (X * (resid ** 2)[:, None])) @ inv * n / (n-k)
se_hc1 = np.sqrt(np.diag(hc1))
np.testing.assert_allclose([model['intercept'], model['slope']], beta, rtol=1e-10, atol=1e-9)
np.testing.assert_allclose([r['fitted'] for r in fitted], yhat, rtol=1e-10, atol=1e-8)
np.testing.assert_allclose([r['residual'] for r in fitted], resid, rtol=1e-10, atol=1e-8)
for i, term in enumerate(['_cons', 'income']):
np.testing.assert_allclose(coeff[term]['se'], se[i], rtol=1e-10)
np.testing.assert_allclose(robust[term]['se'], se_hc1[i], rtol=1e-10)
for output, standard_error in [(coeff[term], se[i]), (robust[term], se_hc1[i])]:
p = 2 * stats.t.sf(abs(beta[i] / standard_error), n-k)
np.testing.assert_allclose(output['p'], p, rtol=1e-9, atol=0)
np.testing.assert_allclose(model['r_squared'], 1-rss/np.sum((y-y.mean())**2), rtol=1e-12)
np.testing.assert_allclose(model['root_mse'], np.sqrt(rss/(n-k)), rtol=1e-12)
for g, a in zip(groups, audit):
yy = y[x == g['income']]
assert g['n'] == len(yy) == a['n']
assert a['detail_total'] == yy.sum()
np.testing.assert_allclose(g['mean_consume'], yy.mean(), rtol=1e-12)
assert a['difference'] == a['detail_total'] - a['reported_total']
log = (ROOT / 'week7_stata.log').read_text(encoding='utf-8', errors='replace')
assert 'WEEK7_ANALYSIS_COMPLETED' in log
assert not re.search(r'^r\(\d+\);', log, flags=re.M)
assert (ROOT / 'consumption_fit.png').read_bytes().startswith(b'\x89PNG')
assert (ROOT / 'consumption_fit.pdf').read_bytes().startswith(b'%PDF')
ET.parse(ROOT / 'consumption_fit.svg')
assert '不同可支配收入水平' in (ROOT / 'week7_analysis.do').read_text(encoding='utf-8')
verification = {
'status': 'passed', 'date': '2026-10-23', 'observations': n,
'source_sha256': hashlib.sha256((ROOT/'Eg2_1_1/Eg2_1_1.xlsx').read_bytes()).hexdigest(),
'independent_ols_intercept': float(beta[0]), 'independent_ols_slope': float(beta[1]),
'max_abs_fitted_difference': float(np.max(np.abs(np.array([r['fitted'] for r in fitted])-yhat))),
'checks': ['all 98 Excel source records match Stata data, including worksheet row',
'OLS coefficients, fitted values, residuals, R-squared, RMSE',
'conventional and HC1 standard errors and two-sided t-test p-values',
'all income-group counts, means, and source-total differences',
'Stata log completion and no r() errors', 'PNG, SVG, and PDF file structure']
}
(ROOT / 'verification.json').write_text(json.dumps(verification, ensure_ascii=False, indent=2), encoding='utf-8')
md, body = [], []
sections = []
def inline(text):
text = html.escape(text)
text = re.sub(r'`([^`]+)`', r'\1', text)
text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text)
return text
def heading(title, anchor):
md.append('## ' + title)
body.append(f'
从概念区分到数据核对,用 Stata 估计收入与消费的线性关系。完整呈现原始数据问题、拟合结果与可复现过程。
有效家庭记录98
收入组数10
收入系数{model['slope']:.4f}
拟合优度 R²{model['r_squared']:.4f}
{nav}{''.join(body)}
'''
(ROOT/'Week7_回归分析报告.html').write_text(page, encoding='utf-8')
# Static checks for complete, portable HTML and working local artifact links.
from html.parser import HTMLParser
class ValidateHTML(HTMLParser):
def handle_starttag(self, tag, attrs):
attrs = dict(attrs)
if tag == 'img':
assert attrs['src'].startswith('data:image/png;base64,')
assert attrs.get('alt')
if tag == 'a' and not attrs['href'].startswith('#'):
assert (ROOT / attrs['href']).exists(), attrs['href']
ValidateHTML().feed(page)
assert '\ufffd' not in page
assert len(re.findall(r'